Add option to expire subjob keys

README.md

@@ -136,6 +136,16 @@ When a superjob is queued, records for all of its subjobs are created. By defaul
 Sidekiq::Superworker.options[:delete_subjobs_after_superjob_completes] = false
 ```
 
+#### Expire superworkers
+
+When a subjob dies due to too many retries depending jobs will never run and the superjob will never be completed. Therefore the sujob redis keys will never be removed.
+When setting `superjob_expiration` to *x* the subjobs keys will expire in *x* seconds. Default value is `nil` (the keys will never expire).
+
+```ruby
+# config/initializers/superworker.rb
+Sidekiq::Superworker.options[:superjob_expiration] = 2592000 # 1 Month
+```
+
 ### Logging
 
 To make debugging easier, Sidekiq Superworker provides detailed log messages when its logger is set to the DEBUG level:

lib/sidekiq/superworker.rb

@@ -9,7 +9,8 @@ module Sidekiq
   module Superworker
     DEFAULTS = {
       delete_subjobs_after_superjob_completes: true,
-      subjob_redis_prefix: 'subjob'
+      subjob_redis_prefix: 'subjob',
+      superjob_expiration: nil
     }
 
     def self.options

lib/sidekiq/superworker/subjob.rb

@@ -74,7 +74,7 @@ def transaction(&block)
           result = nil
           Sidekiq.redis do |conn|
             conn.multi do
-              result = yield
+              result = yield(conn)
             end
           end
           result
@@ -102,8 +102,10 @@ def initialize(params={})
 
       def save
         return false unless self.valid?
-        Sidekiq.redis do |conn|
+
+        self.class.transaction do |conn|
           conn.mapped_hmset(key, to_param)
+          conn.expire(key,Superworker.options[:superjob_expiration]) if Superworker.options[:superjob_expiration]
         end
         true
       end

spec/subjob_spec.rb

@@ -72,6 +72,38 @@ def create_subjob(custom_attributes={})
     end
   end
 
+  describe '#save' do
+    it 'creates a hashmap in redis' do
+      subjob = described_class.new(attributes)
+      subjob.save
+      Sidekiq.redis do |conn|
+        expect(conn.hkeys(subjob.key)).to eq(subjob.to_param.keys.map(&:to_s))
+      end
+    end
+
+    context 'superjob_expiration is set' do
+      it "sets the subjobs expiry accordingly" do
+        allow(Sidekiq::Superworker).to receive("options") {{subjob_redis_prefix: 'subjob',  superjob_expiration: 123}}
+
+        subjob = described_class.new(attributes)
+        subjob.save
+        Sidekiq.redis do |conn|
+          expect(conn.ttl(subjob.key)).to be 123
+        end
+      end
+    end
+
+    context 'superjob_expiration is not set' do
+      it "sets the subjobs expiry accordingly" do
+        subjob = described_class.new(attributes)
+        subjob.save
+        Sidekiq.redis do |conn|
+          expect(conn.ttl(subjob.key)).to be -1
+        end
+      end
+    end
+  end
+
   describe '#to_param' do
     it 'returns a hash including all given attributes' do
       subjob = described_class.new(attributes)