Fix PeriodicJob starvation by preventing duplicate enqueues

When executors are busy with long-running jobs, the coordinator would
enqueue a new copy of each periodic job every interval, even though
previous copies hadn't been executed yet. This caused periodic jobs to
pile up in the queue, starving other work.

Track the pending job run ID in the periodic job's metadata. Before
enqueuing, check whether the previously enqueued run has finished. If
it hasn't, skip the enqueue. The reference is cleared automatically
when the job finishes or its config expires.

https://claude.ai/code/session_01CGgGg8k3tV5QKYMZ5cFJxT

Author: claude

spec/mosquito/periodic_job_run_spec.cr

@@ -37,4 +37,93 @@ describe Mosquito::PeriodicJobRun do
       end
     end
   end
+
+  it "does not enqueue a second job run when one is already pending" do
+    clean_slate do
+      now = Time.utc.at_beginning_of_second
+      periodic = PeriodicJobRun.new PeriodicTestJob, interval
+
+      # First execution should enqueue.
+      Timecop.freeze(now) do
+        periodic.last_executed_at = now - interval
+        assert periodic.try_to_execute
+      end
+
+      queue = PeriodicTestJob.queue
+      first_size = queue.size(include_dead: false)
+      assert first_size > 0, "Expected at least one job in the queue"
+
+      # Second execution after another interval should be skipped
+      # because the first job run hasn't finished yet.
+      Timecop.freeze(now + interval) do
+        assert periodic.try_to_execute
+      end
+
+      second_size = queue.size(include_dead: false)
+      assert_equal first_size, second_size
+    end
+  end
+
+  it "enqueues again after the pending job run finishes" do
+    clean_slate do
+      now = Time.utc.at_beginning_of_second
+      periodic = PeriodicJobRun.new PeriodicTestJob, interval
+
+      # Enqueue the first job run.
+      Timecop.freeze(now) do
+        periodic.last_executed_at = now - interval
+        periodic.try_to_execute
+      end
+
+      # Simulate the job finishing by writing finished_at to the backend.
+      pending_id = periodic.metadata["pending_run_id"]?
+      refute_nil pending_id
+      Mosquito.backend.set(
+        Mosquito::JobRun.config_key(pending_id.not_nil!),
+        "finished_at",
+        Time.utc.to_unix_ms.to_s
+      )
+
+      queue = PeriodicTestJob.queue
+      size_after_first = queue.size(include_dead: false)
+
+      # Now a new interval passes — should enqueue since the previous one finished.
+      Timecop.freeze(now + interval) do
+        assert periodic.try_to_execute
+      end
+
+      size_after_second = queue.size(include_dead: false)
+      assert size_after_second > size_after_first
+    end
+  end
+
+  it "enqueues again when the pending job run config has been cleaned up" do
+    clean_slate do
+      now = Time.utc.at_beginning_of_second
+      periodic = PeriodicJobRun.new PeriodicTestJob, interval
+
+      # Enqueue the first job run.
+      Timecop.freeze(now) do
+        periodic.last_executed_at = now - interval
+        periodic.try_to_execute
+      end
+
+      pending_id = periodic.metadata["pending_run_id"]?
+      refute_nil pending_id
+
+      # Simulate the job run config being deleted (e.g. TTL expiry).
+      Mosquito.backend.delete Mosquito::JobRun.config_key(pending_id.not_nil!)
+
+      queue = PeriodicTestJob.queue
+      size_before = queue.size(include_dead: false)
+
+      # Next interval should enqueue because the old run is gone.
+      Timecop.freeze(now + interval) do
+        assert periodic.try_to_execute
+      end
+
+      size_after = queue.size(include_dead: false)
+      assert size_after > size_before
+    end
+  end
 end

src/mosquito/periodic_job_run.cr

@@ -1,5 +1,7 @@
 module Mosquito
   class PeriodicJobRun
+    Log = ::Log.for self
+
     property class : Mosquito::PeriodicJob.class
     property interval : Time::Span | Time::MonthSpan
     getter metadata : Metadata { Metadata.new(Mosquito.backend.build_key("periodic_jobs", @class.name)) }
@@ -47,24 +49,43 @@ module Mosquito
       now = Time.utc
 
       if last_executed_at + interval <= now
-        execute
+        if pending_job_run?
+          Log.info { "Skipping enqueue for #{@class.name}: a job run is already pending" }
+        else
+          execute
+        end
 
-        # Weaknesses:
-        # - If something interferes with the job run, it won't be correct that it was executed.
-        # - if the worker is backlogged, the start time will be different from the last executed time.
         self.last_executed_at = now
         true
       else
         false
       end
     end
 
-    # Enqueues the job for execution
+    # Returns true if a previously enqueued job run has not yet finished.
+    # This prevents duplicate enqueues when executors are busy and the
+    # periodic interval elapses multiple times before the job is run.
+    def pending_job_run? : Bool
+      if pending_id = metadata["pending_run_id"]?
+        if job_run = JobRun.retrieve(pending_id)
+          return true if job_run.finished_at.nil?
+        end
+
+        # Job run has finished or was cleaned up; clear the stale reference.
+        metadata["pending_run_id"] = nil
+      end
+
+      false
+    end
+
+    # Enqueues the job for execution and records the job run id so that
+    # subsequent intervals can detect that a run is already pending.
     def execute
       job = @class.new
       job_run = job.build_job_run
       job_run.store
       @class.queue.enqueue job_run
+      metadata["pending_run_id"] = job_run.id
     end
   end
 end