Add expired job reaper and related tests for job expiration handling

Author: eliasjpr

spec/expired_job_reaper_spec.cr

@@ -0,0 +1,178 @@
+require "./spec_helper"
+
+module JoobQ
+  describe ExpiredJobReaper do
+    before_each do
+      JoobQ.reset
+    end
+
+    describe "#reap_once" do
+      it "reaps expired jobs from main queue" do
+        store = JoobQ.store.as(RedisStore)
+
+        job = ExampleJob.new(1)
+        job.expires = (Time.local - 1.hour).to_unix_ms
+        store.enqueue(job)
+
+        store.queue_size("example").should eq(1)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        store.queue_size("example").should eq(0)
+        store.redis.zcard("joobq:dead_letter").should be >= 1
+      end
+
+      it "does not reap non-expired jobs" do
+        store = JoobQ.store.as(RedisStore)
+
+        job = ExampleJob.new(1)
+        job.expires = (Time.local + 1.hour).to_unix_ms
+        store.enqueue(job)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        store.queue_size("example").should eq(1)
+      end
+
+      it "reaps only expired jobs from a mixed queue" do
+        store = JoobQ.store.as(RedisStore)
+
+        expired_job = ExampleJob.new(1)
+        expired_job.expires = (Time.local - 1.hour).to_unix_ms
+        store.enqueue(expired_job)
+
+        valid_job = ExampleJob.new(2)
+        valid_job.expires = (Time.local + 1.hour).to_unix_ms
+        store.enqueue(valid_job)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        store.queue_size("example").should eq(1)
+      end
+
+      it "reaps expired jobs from processing queue" do
+        store = JoobQ.store.as(RedisStore)
+
+        job = ExampleJob.new(1)
+        job.expires = (Time.local - 1.hour).to_unix_ms
+
+        processing_key = "joobq:processing:example"
+        store.redis.rpush(processing_key, job.to_json)
+
+        store.redis.llen(processing_key).should eq(1)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        store.redis.llen(processing_key).should eq(0)
+        store.redis.zcard("joobq:dead_letter").should be >= 1
+      end
+
+      it "handles pagination with many expired jobs" do
+        store = JoobQ.store.as(RedisStore)
+
+        # Enqueue more than PAGE_SIZE (100) expired jobs
+        150.times do |i|
+          job = ExampleJob.new(i)
+          job.expires = (Time.local - 1.hour).to_unix_ms
+          store.enqueue(job)
+        end
+
+        store.queue_size("example").should eq(150)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        store.queue_size("example").should eq(0)
+      end
+
+      it "increments stats counters" do
+        store = JoobQ.store.as(RedisStore)
+
+        job = ExampleJob.new(1)
+        job.expires = (Time.local - 1.hour).to_unix_ms
+        store.enqueue(job)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        expired_count = store.redis.hget("joobq:stats:expired", "example")
+        expired_count.should_not be_nil
+        expired_count.should_not be_nil
+        expired_count.as(String).to_i.should be >= 1
+      end
+
+      it "continues scanning other queues after error in one" do
+        store = JoobQ.store.as(RedisStore)
+
+        # Enqueue an expired job in the "single" queue
+        job = Job1.new
+        job.expires = (Time.local - 1.hour).to_unix_ms
+        store.enqueue(job)
+
+        store.queue_size("single").should eq(1)
+
+        reaper = ExpiredJobReaper.new(store: store)
+        reaper.reap_once
+
+        # The reaper should still process other queues even if one has issues
+        store.queue_size("single").should eq(0)
+      end
+    end
+
+    describe "#start and #stop" do
+      it "starts and stops the reaper fiber" do
+        store = JoobQ.store.as(RedisStore)
+        reaper = ExpiredJobReaper.new(store: store, interval: 1.second)
+
+        reaper.start
+        sleep 0.1.seconds
+
+        reaper.stop
+        # Should not raise
+      end
+
+      it "is idempotent on double start" do
+        store = JoobQ.store.as(RedisStore)
+        reaper = ExpiredJobReaper.new(store: store, interval: 1.second)
+
+        reaper.start
+        reaper.start # should be no-op
+
+        reaper.stop
+      end
+
+      it "is idempotent on double stop" do
+        store = JoobQ.store.as(RedisStore)
+        reaper = ExpiredJobReaper.new(store: store, interval: 1.second)
+
+        reaper.start
+        reaper.stop
+        reaper.stop # should be no-op
+      end
+    end
+
+    describe "scan_limit" do
+      it "respects the scan limit" do
+        store = JoobQ.store.as(RedisStore)
+
+        # Enqueue 200 expired jobs
+        200.times do |i|
+          job = ExampleJob.new(i)
+          job.expires = (Time.local - 1.hour).to_unix_ms
+          store.enqueue(job)
+        end
+
+        # Set scan limit to 100 — should only scan first 100 entries
+        reaper = ExpiredJobReaper.new(store: store, scan_limit: 100)
+        reaper.reap_once
+
+        # Some jobs should remain since scan limit was hit
+        store.queue_size("example").should eq(100)
+      end
+    end
+  end
+end

spec/helpers/all_test_jobs.cr spec/helpers/all_test_jobs_spec.crspec/helpers/all_test_jobs.cr renamed to spec/helpers/all_test_jobs_spec.cr

@@ -58,7 +58,7 @@ module JoobQ
           sleep 0.1.seconds
         end
 
-        queue.size.should eq(0)  # All jobs should be processed
+        queue.size.should eq(0) # All jobs should be processed
         queue.stop!
       end
 

spec/queue_spec.cr

@@ -76,7 +76,7 @@ module JoobQ
   end
 
   # Initialize configuration with hybrid loading method (requires block)
-  def self.initialize_config_with(loading_method : Symbol, *args, **kwargs, &block) : Nil
+  def self.initialize_config_with(loading_method : Symbol, *args, **kwargs, &) : Nil
     return if @@config_initialized
 
     case loading_method
@@ -180,6 +180,9 @@ module JoobQ
     # Start delayed job scheduler (processes retrying jobs from DELAYED_SET)
     config.delayed_job_scheduler.start
 
+    # Start expired job reaper (scans queues for expired waiting/processing jobs)
+    config.expired_job_reaper.start
+
     queues.each do |key, queue|
       Log.info { "JoobQ starting #{key} queue..." }
       queue.start

src/joobq.cr

@@ -157,6 +157,19 @@ module JoobQ
     # Delayed job scheduler (processes retrying jobs)
     property delayed_job_scheduler : DelayedJobScheduler = DelayedJobScheduler.new
 
+    # Expired job reaper configuration
+    property reaper_interval : Time::Span = 30.seconds
+    property reaper_scan_limit : Int32 = 1000
+
+    # Expired job reaper (scans queues for expired jobs)
+    getter expired_job_reaper : ExpiredJobReaper do
+      ExpiredJobReaper.new(
+        store: store,
+        interval: reaper_interval,
+        scan_limit: reaper_scan_limit
+      )
+    end
+
     # DSL: Add custom middlewares
     def use(& : ->)
       yield middlewares

src/joobq/configure.cr

@@ -0,0 +1,139 @@
+module JoobQ
+  class ExpiredJobReaper
+    Log = ::Log.for("EXPIRED_JOB_REAPER")
+
+    PAGE_SIZE   = 100
+    DEAD_LETTER = "joobq:dead_letter"
+    PROCESSING  = "joobq:processing"
+
+    @running = Atomic(Bool).new(false)
+    @reaper_fiber : Fiber?
+
+    def initialize(
+      @store : Store = RedisStore.instance,
+      @interval : Time::Span = 30.seconds,
+      @scan_limit : Int32 = 1000,
+    )
+    end
+
+    def start
+      return if @running.get
+      @running.set(true)
+
+      @reaper_fiber = spawn do
+        Log.info &.emit("Expired job reaper started",
+          interval_seconds: @interval.total_seconds,
+          scan_limit: @scan_limit
+        )
+
+        while @running.get
+          begin
+            reap_expired_jobs
+            sleep @interval
+          rescue ex
+            Log.error &.emit("Expired job reaper error", error: ex.message)
+            sleep 5.seconds
+          end
+        end
+      end
+    end
+
+    def stop
+      return unless @running.get
+      @running.set(false)
+      sleep 0.1.seconds
+      Log.info &.emit("Expired job reaper stopped")
+    end
+
+    # Public method for testability — runs a single reap cycle
+    def reap_once
+      reap_expired_jobs
+    end
+
+    private def reap_expired_jobs
+      return unless @store.is_a?(RedisStore)
+
+      redis_store = @store.as(RedisStore)
+
+      JoobQ.config.queues.each do |queue_name, _|
+        begin
+          reaped_waiting = scan_and_reap_queue(redis_store, queue_name)
+
+          processing_key = "#{PROCESSING}:#{queue_name}"
+          reaped_processing = scan_and_reap_queue(redis_store, processing_key, stats_queue: queue_name)
+
+          total = reaped_waiting + reaped_processing
+          if total > 0
+            Log.info &.emit("Reaped expired jobs",
+              queue: queue_name,
+              from_waiting: reaped_waiting,
+              from_processing: reaped_processing
+            )
+          end
+        rescue ex
+          Log.error &.emit("Error reaping expired jobs for queue",
+            queue: queue_name,
+            error: ex.message
+          )
+        end
+      end
+    end
+
+    private def scan_and_reap_queue(
+      redis_store : RedisStore,
+      list_key : String,
+      stats_queue : String? = nil,
+    ) : Int32
+      stats_key = stats_queue || list_key
+      current_time_ms = Time.local.to_unix_ms
+      total_reaped = 0
+      total_scanned = 0
+      offset = 0
+
+      loop do
+        break if total_scanned >= @scan_limit
+
+        batch_size = Math.min(PAGE_SIZE, @scan_limit - total_scanned)
+        jobs = redis_store.redis.lrange(list_key, offset, offset + batch_size - 1).map(&.as(String))
+        break if jobs.empty?
+
+        total_scanned += jobs.size
+        expired_jobs = [] of String
+
+        jobs.each do |job_json|
+          begin
+            parsed = JSON.parse(job_json)
+            expires = parsed["expires"]?.try(&.as_i64)
+            next unless expires
+            expired_jobs << job_json if current_time_ms > expires
+          rescue ex
+            Log.warn &.emit("Failed to parse job during reaping",
+              list_key: list_key,
+              error: ex.message
+            )
+          end
+        end
+
+        if !expired_jobs.empty?
+          redis_store.redis.pipelined do |pipe|
+            expired_jobs.each do |job_json|
+              pipe.lrem(list_key, 0, job_json)
+              pipe.zadd(DEAD_LETTER, current_time_ms, job_json)
+            end
+            pipe.hincrby("joobq:stats:expired", stats_key, expired_jobs.size)
+            pipe.hincrby("joobq:stats:dead_letter", stats_key, expired_jobs.size)
+          end
+          total_reaped += expired_jobs.size
+          # Removed items shift list down; advance past non-expired items only
+          offset += (jobs.size - expired_jobs.size)
+        else
+          offset += jobs.size
+        end
+
+        break if jobs.size < batch_size
+      end
+
+      total_reaped
+    end
+  end
+end