workaround: lock ttl/expiration

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    fleiss (0.4.4)
+    fleiss (0.4.5)
       activejob (>= 6.0)
       activerecord (>= 6.0)
       concurrent-ruby
@@ -53,16 +53,16 @@ GEM
       diff-lcs (>= 1.2.0, < 2.0)
       rspec-support (~> 3.10.0)
     rspec-support (3.10.2)
-    rubocop (1.21.0)
+    rubocop (1.22.0)
       parallel (~> 1.10)
       parser (>= 3.0.0.0)
       rainbow (>= 2.2.2, < 4.0)
       regexp_parser (>= 1.8, < 3.0)
       rexml
-      rubocop-ast (>= 1.9.1, < 2.0)
+      rubocop-ast (>= 1.12.0, < 2.0)
       ruby-progressbar (~> 1.7)
       unicode-display_width (>= 1.4.0, < 3.0)
-    rubocop-ast (1.11.0)
+    rubocop-ast (1.12.0)
       parser (>= 3.0.1.1)
     rubocop-bsm (0.6.0)
       rubocop (~> 1.0)
@@ -74,9 +74,8 @@ GEM
       rubocop-ast (>= 0.4.0)
     rubocop-rake (0.6.0)
       rubocop (~> 1.0)
-    rubocop-rspec (2.4.0)
-      rubocop (~> 1.0)
-      rubocop-ast (>= 1.1.0)
+    rubocop-rspec (2.5.0)
+      rubocop (~> 1.19)
     ruby-progressbar (1.11.0)
     sqlite3 (1.4.2)
     tzinfo (2.0.4)

README.md

@@ -60,3 +60,11 @@ Run the worker:
 ```ruby
 bundle exec fleiss -I . -r config/environment
 ```
+
+## Tweaks
+
+Worker heartbeat is not implemented, so in rare cases jobs can be stuck in "running" state if worker was KILL-ed (and not shut down gracefully). Or if worker is killed after it acquired the job and then DB connection was lost.
+
+To work around that, `FLEISS_LOCK_TTL` (seconds) ENV variable can be set. This should be larger than maximum expected job perform time.
+
+Jobs that are started but not finished in `FLEISS_LOCK_TTL` seconds can be picked up by other workers.

fleiss.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name          = 'fleiss'
-  s.version       = '0.4.4'
+  s.version       = '0.4.5'
   s.authors       = ['Black Square Media Ltd']
   s.email         = ['[email protected]']
   s.summary       = %(Minimialist background jobs backed by ActiveJob and ActiveRecord.)

lib/fleiss/backend/active_record/concern.rb

@@ -4,6 +4,8 @@ class ActiveRecord
       module Concern
         extend ActiveSupport::Concern
 
+        DEFAULT_LOCK_TTL = ENV['FLEISS_LOCK_TTL']&.to_i # seconds
+
         included do
           scope :in_queue, ->(qs) { where(queue_name: Array.wrap(qs)) }
           scope :finished, -> { where.not(finished_at: nil) }
@@ -14,6 +16,15 @@ module Concern
           scope :started,      -> { where(arel_table[:started_at].not_eq(nil)) }
           scope :not_started,  -> { where(arel_table[:started_at].eq(nil)) }
           scope :scheduled,    ->(now = Time.zone.now) { where(arel_table[:scheduled_at].gt(now)) }
+
+          scope :lock_expired, lambda {|now = Time.zone.now, lock_ttl = DEFAULT_LOCK_TTL|
+            if lock_ttl
+              min_started_at = now - (1.1 * lock_ttl) # 10% threshold
+              where(finished_at: nil).where(arel_table[:started_at].lt(min_started_at)) # not yet finished and started at quite a while ago
+            else
+              none # do nothing unless explicitly configured
+            end
+          }
         end
 
         module ClassMethods
@@ -26,12 +37,14 @@ def wrap_perform(&block)
 
           # @return [ActiveRecord::Relation] pending scope
           def pending(now = Time.zone.now)
-            not_finished
-              .not_expired(now)
-              .not_started
-              .where(arel_table[:scheduled_at].lteq(now))
-              .order(priority: :desc)
-              .order(scheduled_at: :asc)
+            pending = not_finished
+                      .not_expired(now)
+                      .not_started
+                      .where(arel_table[:scheduled_at].lteq(now))
+                      .order(priority: :desc)
+                      .order(scheduled_at: :asc)
+
+            pending.or(lock_expired(now, DEFAULT_LOCK_TTL))
           end
 
           # @return [ActiveRecord::Relation] in-progress scope

spec/fleiss/backend/active_record_spec.rb

@@ -47,15 +47,28 @@ def retrieve(job)
   end
 
   it 'scopes pending' do
-    j1 = TestJob.perform_later
-    expect(retrieve(j1).start('owner')).to be_truthy
-    expect(retrieve(j1).finish('owner')).to be_truthy
+    finished = TestJob.perform_later
+    expect(retrieve(finished).start('owner')).to be_truthy
+    expect(retrieve(finished).finish('owner')).to be_truthy
 
-    j2 = TestJob.perform_later
-    _j3 = TestJob.set(wait: 1.hour).perform_later
-    j4 = TestJob.set(priority: 2).perform_later
+    # jobs with expired locks are seen as pending:
+    lock_expired = travel_to(2.days.ago) do
+      stub_const('Fleiss::Backend::ActiveRecord::Concern::DEFAULT_LOCK_TTL', 1.day.seconds)
+
+      TestJob.perform_later.tap do |job|
+        expect(retrieve(job).start('owner')).to be_truthy
+      end
+    end
+
+    pending = TestJob.perform_later
+    pending_high_prio = TestJob.set(priority: 2).perform_later
+    _future = TestJob.set(wait: 1.hour).perform_later # not visible yet
 
-    expect(described_class.pending.ids).to eq [j4.provider_job_id, j2.provider_job_id]
+    expect(described_class.pending.ids).to eq [
+      pending_high_prio.provider_job_id,
+      lock_expired.provider_job_id,
+      pending.provider_job_id,
+    ]
   end
 
   it 'scopes in_progress' do
@@ -132,4 +145,34 @@ def retrieve(job)
     end
       .to raise_error(::ActiveRecord::StatementInvalid) # re-raised anyway
   end
+
+  context 'with internal helpers' do
+    it 'scopes lock_expired' do
+      # one not finished, but "recent" job:
+      travel_to(1.days.ago) { expect(retrieve(TestJob.perform_later).start('owner')).to be_truthy }
+
+      # not finished, and "old":
+      old = travel_to(2.days.ago) do
+        retrieve(TestJob.perform_later).tap do |rec|
+          expect(rec.start('owner')).to be_truthy
+        end
+      end
+
+      # one "old", but finished:
+      travel_to(2.days.ago) do
+        job = TestJob.perform_later
+        expect(retrieve(job).start('owner')).to be_truthy
+        expect(retrieve(job).finish('owner')).to be_truthy
+      end
+
+      # one "old", but not-started job (so not eligible for lock check)
+      travel_to(2.days.ago) { TestJob.perform_later }
+
+      # don't do anything unless TTL configured:
+      expect(described_class.lock_expired(Time.zone.now, nil)).not_to exist
+
+      expect(described_class.lock_expired(Time.zone.now, 1.day)).to contain_exactly(old)
+      expect(described_class.lock_expired(Time.zone.now, 2.day)).not_to exist
+    end
+  end
 end

spec/spec_helper.rb

@@ -5,6 +5,7 @@
 require 'fleiss/backend/active_record/migration'
 require 'active_job'
 require 'active_job/queue_adapters/fleiss_adapter'
+require 'active_support/testing/time_helpers'
 require 'fileutils'
 
 ActiveJob::Base.queue_adapter = :fleiss
@@ -44,6 +45,8 @@ def perform(msg = nil)
 end
 
 RSpec.configure do |c|
+  c.include ActiveSupport::Testing::TimeHelpers
+
   c.after do
     TestJob.performed.clear
     Fleiss.backend.delete_all