store a separate key for keeping object`s created_time

Author: prog-supdex

lib/graphql/anycable/cleaner.rb

@@ -29,10 +29,12 @@ def clean_subscriptions
         return unless config.use_redis_object_on_cleanup
 
         redis.scan_each(match: "#{redis_key(adapter::SUBSCRIPTION_PREFIX)}*") do |key|
-          idle = redis.object("IDLETIME", key)
-          next if idle&.<= config.subscription_expiration_seconds
+          next unless object_created_time_expired?(key)
 
-          redis.del(key)
+          redis.multi do |pipeline|
+            pipeline.del(key)
+            pipeline.hdel(redis_key(adapter::CREATED_AT_KEY), key)
+          end
         end
       end
 
@@ -74,6 +76,16 @@ def config
       def redis_key(prefix)
         "#{config.redis_prefix}-#{prefix}"
       end
+
+      def object_created_time_expired?(key)
+        last_created_time = redis.hget(redis_key(adapter::CREATED_AT_KEY), key)
+
+        return false unless last_created_time
+
+        expire_date = Time.parse(last_created_time) + config.subscription_expiration_seconds
+
+        Time.now >= expire_date
+      end
     end
   end
 end

lib/graphql/subscriptions/anycable_subscriptions.rb

@@ -3,7 +3,6 @@
 require "anycable"
 require "graphql/subscriptions"
 require "graphql/anycable/errors"
-
 # rubocop: disable Metrics/AbcSize, Metrics/LineLength, Metrics/MethodLength
 
 # A subscriptions implementation that sends data as AnyCable broadcastings.
@@ -56,10 +55,11 @@ class AnyCableSubscriptions < GraphQL::Subscriptions
 
       def_delegators :"GraphQL::AnyCable", :redis, :config
 
-      SUBSCRIPTION_PREFIX  = "subscription:"  # HASH: Stores subscription data: query, context, …
-      FINGERPRINTS_PREFIX  = "fingerprints:"  # ZSET: To get fingerprints by topic
-      SUBSCRIPTIONS_PREFIX = "subscriptions:" # SET:  To get subscriptions by fingerprint
-      CHANNEL_PREFIX       = "channel:"       # SET:  Auxiliary structure for whole channel's subscriptions cleanup
+      SUBSCRIPTION_PREFIX  = "subscription:"                # HASH: Stores subscription data: query, context, …
+      FINGERPRINTS_PREFIX  = "fingerprints:"                # ZSET: To get fingerprints by topic
+      SUBSCRIPTIONS_PREFIX = "subscriptions:"               # SET:  To get subscriptions by fingerprint
+      CHANNEL_PREFIX       = "channel:"                     # SET:  Auxiliary structure for whole channel's subscriptions cleanup
+      CREATED_AT_KEY       = "objects:list-created-times"   # HASH: Stores name and created_time of object
 
       # @param serializer [<#dump(obj), #load(string)] Used for serializing messages before handing them to `.broadcast(msg)`
       def initialize(serializer: Serialize, **rest)
@@ -131,7 +131,6 @@ def write_subscription(query, events)
         # Store subscription_id in the channel state to cleanup on disconnect
         write_subscription_id(channel, channel_uniq_id)
 
-
         events.each do |event|
           channel.stream_from(redis_key(SUBSCRIPTIONS_PREFIX) + event.fingerprint)
         end
@@ -145,8 +144,13 @@ def write_subscription(query, events)
         }
 
         redis.multi do |pipeline|
+          full_subscription_id = "#{redis_key(SUBSCRIPTION_PREFIX)}#{subscription_id}"
+
           pipeline.sadd(redis_key(CHANNEL_PREFIX) + channel_uniq_id, [subscription_id])
-          pipeline.mapped_hmset(redis_key(SUBSCRIPTION_PREFIX) + subscription_id, data)
+          pipeline.mapped_hmset(full_subscription_id, data)
+
+          pipeline.hset(redis_key(CREATED_AT_KEY), full_subscription_id, Time.now.to_s)
+
           events.each do |event|
             pipeline.zincrby(redis_key(FINGERPRINTS_PREFIX) + event.topic, 1, event.fingerprint)
             pipeline.sadd(redis_key(SUBSCRIPTIONS_PREFIX) + event.fingerprint, [subscription_id])
@@ -182,7 +186,10 @@ def delete_subscription(subscription_id)
             fingerprint_subscriptions[redis_key(FINGERPRINTS_PREFIX) + topic] = score
           end
           # Delete subscription itself
-          pipeline.del(redis_key(SUBSCRIPTION_PREFIX) + subscription_id)
+          full_subscription_id = "#{redis_key(SUBSCRIPTION_PREFIX)}#{subscription_id}"
+
+          pipeline.del(full_subscription_id)
+          pipeline.hdel(redis_key(CREATED_AT_KEY), full_subscription_id)
         end
         # Clean up fingerprints that doesn't have any subscriptions left
         redis.pipelined do |pipeline|

spec/graphql/anycable_spec.rb

@@ -138,10 +138,12 @@
         expect(redis.exists?("graphql-subscription:some-truly-random-number")).to be true
         expect(redis.exists?("graphql-channel:some-truly-random-number")).to be true
         expect(redis.exists?("graphql-fingerprints::productUpdated:")).to be true
+        expect(redis.hexists("graphql-objects:list-created-times", "graphql-subscription:some-truly-random-number")).to be true
         subject
         expect(redis.exists?("graphql-channel:some-truly-random-number")).to be false
         expect(redis.exists?("graphql-fingerprints::productUpdated:")).to be false
         expect(redis.exists?("graphql-subscription:some-truly-random-number")).to be false
+        expect(redis.hexists("graphql-objects:list-created-times", "graphql-subscription:some-truly-random-number")).to be false
       end
     end