From ad6900e23d89afb44d1b950934a32dc8ec303b3e Mon Sep 17 00:00:00 2001 From: Hampton Lintorn-Catlin Date: Tue, 1 Sep 2026 16:07:31 -0500 Subject: [PATCH 1/2] Add CoPlan::DebouncedJob, a reusable event-coalescing primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Downstream hosts wanting to coalesce a burst of comment_created events into one delayed notification (e.g. a Slack digest) keep hand-rolling the same debounce logic, and getting the same three bugs wrong: using call time instead of the triggering event's own timestamp as the batch boundary, a racy read-then-write claim that lets two callers both enqueue, and cache TTLs shorter than the job's own retry backoff that quietly drop part of a retried batch. CoPlan::DebouncedJob is a mixin any host ActiveJob can include to get this once, correctly: an atomic write-if-absent claim, a boundary that's required to be the caller-supplied event timestamp, and a state TTL sized off the window plus a configurable retry horizon rather than the window alone. The including job implements perform_batch(key:, batch_start:) instead of perform, and triggers a batch with MyJob.debounce(key:, event_at:). Not wired into Comment/comment_created or any host callback yet — this is the standalone primitive, tested against each of the three failure modes above plus a 25-thread concurrent-claim test. --- .../app/jobs/concerns/coplan/debounced_job.rb | 190 ++++++++++++ spec/jobs/coplan/debounced_job_spec.rb | 286 ++++++++++++++++++ 2 files changed, 476 insertions(+) create mode 100644 engine/app/jobs/concerns/coplan/debounced_job.rb create mode 100644 spec/jobs/coplan/debounced_job_spec.rb diff --git a/engine/app/jobs/concerns/coplan/debounced_job.rb b/engine/app/jobs/concerns/coplan/debounced_job.rb new file mode 100644 index 00000000..d8b72f72 --- /dev/null +++ b/engine/app/jobs/concerns/coplan/debounced_job.rb @@ -0,0 +1,190 @@ +module CoPlan + # Coalesces a burst of same-key events into a single delayed job. + # + # The shape of the problem: something fires an event per occurrence — a + # comment created, a plan edited — and a host wants one notification for + # the whole burst instead of one per event. Include this in the host's own + # ActiveJob class, call `debounce` from the event callback, and implement + # `perform_batch`; the first event in a window schedules one delayed job + # and later events inside the window are no-ops. When the job finally runs + # it is handed the timestamp the batch started at so it can query for + # everything since. + # + # class DigestJob < ApplicationJob + # include CoPlan::DebouncedJob + # + # retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5 + # debounces_with window: 2.minutes, retry_horizon: 30.minutes + # + # def perform_batch(key:, batch_start:) + # comments = CommentThread.find(key).comments.where(created_at: batch_start..) + # deliver(comments) + # end + # end + # + # # in a notification_handler, or a model callback: + # DigestJob.debounce(key: comment.comment_thread_id, event_at: comment.created_at) + # + # This deliberately knows nothing about comments, threads or delivery + # channels. It owns exactly three things — the atomic claim, the batch + # boundary, and when the claim is released — because those are the three + # things hand-rolled debouncers keep getting wrong. + # + # Requires a `Rails.cache` store that honours `unless_exist:` on write + # (MemoryStore, RedisCacheStore, MemCacheStore, SolidCache all do). A + # store that ignores it — notably `:null_store`, the Rails default in the + # test environment — degrades to "every event enqueues its own job". + module DebouncedJob + extend ActiveSupport::Concern + + # How long the first event waits for the rest of its burst. + DEFAULT_WINDOW = 1.minute + + # How far past the scheduled run the claim and the batch boundary must + # still be readable. This is not the debounce window: it exists to + # outlive the *job's own retry backoff*, so a job on its last retry + # still finds the boundary it was enqueued with instead of quietly + # falling back to a narrower one and dropping the front of its batch. + # 30 minutes comfortably clears five attempts of `:polynomially_longer` + # (~6 minutes); jobs with a longer retry policy should raise it. + DEFAULT_RETRY_HORIZON = 30.minutes + + included do + class_attribute :debounce_window, instance_writer: false, default: DEFAULT_WINDOW + class_attribute :debounce_retry_horizon, instance_writer: false, default: DEFAULT_RETRY_HORIZON + end + + class_methods do + # Configures the two durations. They are separate on purpose: the + # window is a product decision ("how long do we wait for the rest of + # the burst"), the retry horizon is an operational one ("how long can + # this job still be in flight"). Tying cache TTLs to the window alone + # is what expires batch state out from under a retrying job. + def debounces_with(window: nil, retry_horizon: nil) + self.debounce_window = validate_duration!(window, :window) if window + self.debounce_retry_horizon = validate_duration!(retry_horizon, :retry_horizon) if retry_horizon + end + + # TTL for both the claim and the batch boundary. Always longer than + # the window by the full retry horizon. + def debounce_state_ttl + debounce_window + debounce_retry_horizon + end + + # Claims a pending slot for `key` and, if this caller won it, + # schedules the one delayed job for the burst. Returns true if this + # call started a batch, false if a batch was already pending. + # + # `event_at` is the triggering event's own timestamp — the record's + # `created_at`, not `Time.current`. It is required and unforgiving on + # purpose. `debounce` runs after the event is already persisted (often + # from a callback or another job), so call time is always *later* than + # the event; using it as the boundary excludes the very event that + # started the batch from the batch's own query. + def debounce(key:, event_at:, **arguments) + boundary = coerce_event_at(event_at) + + # One atomic write-if-absent is the whole claim. A read-then-write + # would let two callers both see "no batch pending" and both + # enqueue, delivering the burst twice. + claimed = Rails.cache.write( + debounce_pending_cache_key(key), true, + expires_in: debounce_state_ttl, unless_exist: true + ) + return false unless claimed + + Rails.cache.write(debounce_batch_start_cache_key(key), boundary.iso8601(9), expires_in: debounce_state_ttl) + set(wait: debounce_window).perform_later(key: key, **arguments) + true + end + + # The timestamp the batch for `key` started at, or nil if no batch + # state exists. + def batch_start_for(key) + raw = Rails.cache.read(debounce_batch_start_cache_key(key)) + raw && Time.zone.parse(raw) + end + + def debounce_pending?(key) + Rails.cache.exist?(debounce_pending_cache_key(key)) + end + + # Ends the batch for `key`, so the next event starts a new one. + # Called for you after `perform_batch` returns; the ordering matters, + # see the comment inside. + def release_debounce(key) + # Boundary first, claim second. The other order leaves a moment + # where a new event can win the claim and write its own boundary, + # only for this call to delete it a line later. + Rails.cache.delete(debounce_batch_start_cache_key(key)) + Rails.cache.delete(debounce_pending_cache_key(key)) + end + + def debounce_pending_cache_key(key) + "coplan:debounced_job:#{debounce_namespace}:pending:#{key}" + end + + def debounce_batch_start_cache_key(key) + "coplan:debounced_job:#{debounce_namespace}:batch_start:#{key}" + end + + # Namespaces cache keys per job class, so two debounced jobs keyed on + # the same record don't fight over one claim. + def debounce_namespace + name.presence || to_s + end + + private + + def coerce_event_at(event_at) + unless event_at.respond_to?(:to_time) + raise ArgumentError, + "#{self}.debounce requires event_at: the triggering event's own timestamp " \ + "(e.g. record.created_at), got #{event_at.inspect}" + end + + event_at.to_time.utc + end + + def validate_duration!(value, label) + unless value.respond_to?(:to_i) && value.to_i.positive? + raise ArgumentError, "#{self}.debounces_with #{label}: must be a positive duration, got #{value.inspect}" + end + + value + end + end + + # Including jobs implement `perform_batch`, not `perform` — this owns + # `perform` so the claim is always released in the right place. + def perform(key:, **arguments) + perform_batch(key: key, batch_start: debounce_batch_start(key), **arguments) + + # Only after the work is done. Releasing on enqueue, or in an + # `ensure`, would let an event arriving mid-send (or mid-retry, since + # a raise skips this line and leaves the claim held) open a second + # batch that overlaps and clobbers this one. + self.class.release_debounce(key) + end + + private + + def debounce_batch_start(key) + self.class.batch_start_for(key) || missing_batch_start(key) + end + + # Should not happen: the boundary outlives the window by the whole + # retry horizon. If it does, the state was lost rather than the batch + # being empty, so fall back to the widest window we could have retained + # and say so loudly. Erring wide risks a duplicate; erring narrow drops + # events silently, which is the failure nobody notices. + def missing_batch_start(key) + Rails.logger.warn( + "#{self.class}: batch start for #{key.inspect} expired before the job ran; " \ + "falling back to #{self.class.debounce_state_ttl.inspect} ago. " \ + "Raise retry_horizon above this job's retry backoff." + ) + self.class.debounce_state_ttl.ago + end + end +end diff --git a/spec/jobs/coplan/debounced_job_spec.rb b/spec/jobs/coplan/debounced_job_spec.rb new file mode 100644 index 00000000..94d05e85 --- /dev/null +++ b/spec/jobs/coplan/debounced_job_spec.rb @@ -0,0 +1,286 @@ +require "rails_helper" + +# A transient failure, so the test job can exercise its own retry policy. +class DebouncedJobSpecError < StandardError; end + +# Stand-in for a host's job. Deliberately touches nothing but the concern: +# the primitive is supposed to know nothing about comments or delivery. +class DebouncedJobSpecJob < CoPlan::ApplicationJob + include CoPlan::DebouncedJob + + retry_on DebouncedJobSpecError, wait: 5.minutes, attempts: 3 + + debounces_with window: 2.minutes, retry_horizon: 30.minutes + + cattr_accessor :batches, default: [] + cattr_accessor :failures_remaining, default: 0 + + def perform_batch(key:, batch_start:) + if self.class.failures_remaining.positive? + self.class.failures_remaining -= 1 + raise DebouncedJobSpecError, "transient" + end + + self.class.batches << { key: key, batch_start: batch_start, performed_at: Time.current } + end +end + +RSpec.describe CoPlan::DebouncedJob, type: :job do + include ActiveJob::TestHelper + + let(:job) { DebouncedJobSpecJob } + let(:key) { "thread-abc" } + + # The test environment runs :null_store, which silently ignores + # `unless_exist:` — swap in a real store so claims actually claim. + around do |example| + original_cache = Rails.cache + Rails.cache = ActiveSupport::Cache::MemoryStore.new + example.run + Rails.cache = original_cache + end + + before do + DebouncedJobSpecJob.batches = [] + DebouncedJobSpecJob.failures_remaining = 0 + end + + describe "the batch boundary" do + # Failure mode 1: using claim time as the boundary. `debounce` always + # runs after the triggering event is persisted, so Time.current is + # later than the event's created_at and the batch's own query excludes + # the event that started it. + it "records the triggering event's timestamp, not the time debounce was called" do + freeze_time do + event_at = 90.seconds.ago + + expect(job.debounce(key: key, event_at: event_at)).to be(true) + + expect(job.batch_start_for(key)).to eq(event_at) + expect(job.batch_start_for(key)).to be < Time.current + end + end + + it "hands that same timestamp to perform_batch, so the triggering event is inside its own batch" do + freeze_time do + event_at = 90.seconds.ago + job.debounce(key: key, event_at: event_at) + + travel 2.minutes + job.perform_now(key: key) + + batch = DebouncedJobSpecJob.batches.sole + expect(batch[:batch_start]).to eq(event_at) + expect(batch[:batch_start]).to be < batch[:performed_at] + end + end + + it "refuses to guess a boundary when event_at is missing" do + expect { job.debounce(key: key, event_at: nil) } + .to raise_error(ArgumentError, /triggering event's own timestamp/) + expect(enqueued_jobs).to be_empty + end + + it "keeps the first event's boundary when later events land inside the window" do + freeze_time do + first_at = 10.seconds.ago + job.debounce(key: key, event_at: first_at) + + travel 30.seconds + expect(job.debounce(key: key, event_at: Time.current)).to be(false) + + expect(job.batch_start_for(key)).to eq(first_at) + end + end + end + + # Failure mode 2: a read-then-write claim. Two workers both read "no batch + # pending", both write, both enqueue — the burst is delivered twice. The + # threaded example below is the end-to-end assertion; the two after it pin + # the structure, since MRI rarely interleaves a read-then-write on demand. + describe "claiming the batch" do + it "lets exactly one of many concurrent callers win the claim" do + event_at = 1.minute.ago + results = [] + mutex = Mutex.new + start_gate = Queue.new + + threads = 25.times.map do + Thread.new do + start_gate.pop # all 25 launch together, into the same claim + won = job.debounce(key: key, event_at: event_at) + mutex.synchronize { results << won } + end + end + 25.times { start_gate << :go } + threads.each(&:join) + + expect(results.count(true)).to eq(1) + expect(results.size).to eq(25) + expect(enqueued_jobs.count { |j| j["job_class"] == "DebouncedJobSpecJob" }).to eq(1) + end + + it "claims with a write-if-absent and never reads the cache first" do + allow(Rails.cache).to receive(:read).and_call_original + allow(Rails.cache).to receive(:write).and_call_original + + job.debounce(key: key, event_at: 1.minute.ago) + + expect(Rails.cache).not_to have_received(:read) + expect(Rails.cache).to have_received(:write).with( + job.debounce_pending_cache_key(key), true, hash_including(unless_exist: true) + ).once + end + + # The atomic write's return value is the only thing that decides who + # owns the batch. A read-then-write claim would look at the empty cache + # here, conclude it had won, and enqueue a duplicate. + it "loses the claim purely on the write's return value, not on cache contents" do + pending_key = job.debounce_pending_cache_key(key) + allow(Rails.cache).to receive(:write).and_call_original + allow(Rails.cache).to receive(:write) + .with(pending_key, true, hash_including(unless_exist: true)).and_return(false) + + expect(Rails.cache.read(pending_key)).to be_nil + expect(job.debounce(key: key, event_at: 1.minute.ago)).to be(false) + + expect(enqueued_jobs).to be_empty + expect(job.batch_start_for(key)).to be_nil + end + + it "enqueues one delayed job for the burst and no-ops for the rest" do + freeze_time do + expect { + job.debounce(key: key, event_at: Time.current) + }.to have_enqueued_job(DebouncedJobSpecJob).with(key: key).at(2.minutes.from_now) + + expect { + 3.times { job.debounce(key: key, event_at: Time.current) } + }.not_to have_enqueued_job(DebouncedJobSpecJob) + end + end + + it "keys claims per job class and per key" do + job.debounce(key: "one", event_at: 1.minute.ago) + + expect(job.debounce_pending?("one")).to be(true) + expect(job.debounce_pending?("two")).to be(false) + expect(job.debounce_pending_cache_key("one")).to include("DebouncedJobSpecJob") + end + end + + describe "state lifetime" do + # Failure mode 3: TTLs tied to the debounce window. A job that retries + # for minutes finds its batch state already gone and silently narrows + # its query, dropping the front of the batch. + it "sizes state TTL as the window plus the full retry horizon" do + expect(job.debounce_state_ttl).to eq(32.minutes) + expect(job.debounce_state_ttl).to be > job.debounce_window + expect(job.debounce_state_ttl).to be > job.debounce_retry_horizon + end + + it "keeps the boundary and the claim readable long past the debounce window" do + freeze_time do + event_at = Time.current + job.debounce(key: key, event_at: event_at) + + travel 20.minutes # far past the 2-minute window, inside the retry horizon + + expect(job.batch_start_for(key)).to eq(event_at) + expect(job.debounce_pending?(key)).to be(true) + end + end + + it "holds the claim across a retry and gives the retry the original boundary" do + freeze_time do + event_at = 30.seconds.ago + job.debounce(key: key, event_at: event_at) + DebouncedJobSpecJob.failures_remaining = 1 + + travel 2.minutes + job.perform_now(key: key) # raises inside perform_batch, retry_on re-enqueues + + expect(DebouncedJobSpecJob.batches).to be_empty + expect(job.debounce_pending?(key)).to be(true) + + travel 20.minutes # the retry finally runs, way past the window + job.perform_now(key: key) + + expect(DebouncedJobSpecJob.batches.sole[:batch_start]).to eq(event_at) + end + end + + it "warns and falls back to the widest retained window if state is somehow lost" do + freeze_time do + job.debounce(key: key, event_at: Time.current) + + travel 33.minutes # past state TTL entirely + expect(job.batch_start_for(key)).to be_nil + + allow(Rails.logger).to receive(:warn) + job.perform_now(key: key) + + expect(Rails.logger).to have_received(:warn).with(/expired before the job ran/) + expect(DebouncedJobSpecJob.batches.sole[:batch_start]).to eq(32.minutes.ago) + end + end + end + + describe "releasing the claim" do + it "releases only after perform_batch completes, so a mid-flight event cannot open a second batch" do + freeze_time do + job.debounce(key: key, event_at: Time.current) + + observed = nil + allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do + observed = job.debounce_pending?(key) + end + + job.perform_now(key: key) + + expect(observed).to be(true) + expect(job.debounce_pending?(key)).to be(false) + end + end + + it "leaves the claim held when the batch raises" do + allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch).and_raise("delivery is down") + + job.debounce(key: key, event_at: 1.minute.ago) + expect { job.perform_now(key: key) }.to raise_error("delivery is down") + + expect(job.debounce_pending?(key)).to be(true) + expect(job.batch_start_for(key)).to be_present + end + + it "lets the next event start a fresh batch once the previous one finished" do + freeze_time do + first_at = Time.current + job.debounce(key: key, event_at: first_at) + travel 2.minutes + job.perform_now(key: key) + + travel 1.minute + later_at = Time.current + expect(job.debounce(key: key, event_at: later_at)).to be(true) + expect(job.batch_start_for(key)).to eq(later_at) + end + end + end + + describe "configuration" do + it "defaults to the concern's window and retry horizon" do + default_job = Class.new(CoPlan::ApplicationJob) { include CoPlan::DebouncedJob } + + expect(default_job.debounce_window).to eq(CoPlan::DebouncedJob::DEFAULT_WINDOW) + expect(default_job.debounce_retry_horizon).to eq(CoPlan::DebouncedJob::DEFAULT_RETRY_HORIZON) + end + + it "rejects a non-positive window or retry horizon" do + klass = Class.new(CoPlan::ApplicationJob) { include CoPlan::DebouncedJob } + + expect { klass.debounces_with(window: -1.minute) }.to raise_error(ArgumentError, /positive duration/) + expect { klass.debounces_with(retry_horizon: "soon") }.to raise_error(ArgumentError, /positive duration/) + end + end +end From bee0583a514865fc82dfeb1bc58789df93211954 Mon Sep 17 00:00:00 2001 From: Hampton Lintorn-Catlin Date: Wed, 2 Sep 2026 11:59:52 -0500 Subject: [PATCH 2/2] Rearm on mid-run arrivals, release the claim on enqueue failure Two P1/P2 issues from Codex review on block/coplan#208: - An event landing after perform_batch has already queried but before release_debounce runs was silently dropped: the held claim makes debounce a no-op, and the completed query can't retroactively include it. Adds a narrower `running?` flag (distinct from the claim, which also spans the wait) so debounce can tell "still waiting, will be picked up naturally" apart from "already querying, might be missed", and marks the event dirty in the latter case. perform now rearms the same claim for a follow-up run instead of releasing when something came in dirty, keeping the earliest such event so multiple mid-run arrivals don't shadow each other. release_debounce deliberately leaves a lingering dirty marker alone (rather than clearing it) so a later claim can still fold it in as a fallback. - If perform_later raised (queue adapter error) or returned a job that wasn't actually enqueued (a halted enqueue callback), debounce still left the claim in place, so every event for that key would silently no-op until debounce_state_ttl expired with no job ever scheduled. schedule_run now releases the claim in both cases and re-raises on the exception path rather than swallowing it. --- .../app/jobs/concerns/coplan/debounced_job.rb | 126 ++++++++++++++++-- spec/jobs/coplan/debounced_job_spec.rb | 109 ++++++++++++++- 2 files changed, 221 insertions(+), 14 deletions(-) diff --git a/engine/app/jobs/concerns/coplan/debounced_job.rb b/engine/app/jobs/concerns/coplan/debounced_job.rb index d8b72f72..a4c95aa0 100644 --- a/engine/app/jobs/concerns/coplan/debounced_job.rb +++ b/engine/app/jobs/concerns/coplan/debounced_job.rb @@ -91,11 +91,27 @@ def debounce(key:, event_at:, **arguments) debounce_pending_cache_key(key), true, expires_in: debounce_state_ttl, unless_exist: true ) - return false unless claimed - Rails.cache.write(debounce_batch_start_cache_key(key), boundary.iso8601(9), expires_in: debounce_state_ttl) - set(wait: debounce_window).perform_later(key: key, **arguments) - true + unless claimed + # Still waiting out its window: the eventual perform_batch query + # is "since batch_start" and naturally picks this event up, no + # action needed. Already running perform_batch, though, means the + # query may have already executed before this event landed — + # record it so perform schedules a catch-up run instead of + # silently dropping it (see the rearm branch in #perform). + mark_dirty(key, boundary) if running?(key) + return false + end + + # Fold in anything a previous run marked dirty but didn't live long + # enough to rearm for itself (see #perform) — release_debounce + # deliberately leaves the dirty marker alone so a later claim can + # still pick it up instead of the event being lost between the two. + leftover = consume_dirty(key) + effective_boundary = leftover && leftover < boundary ? leftover : boundary + + Rails.cache.write(debounce_batch_start_cache_key(key), effective_boundary.iso8601(9), expires_in: debounce_state_ttl) + schedule_run(key, arguments) end # The timestamp the batch for `key` started at, or nil if no batch @@ -110,8 +126,9 @@ def debounce_pending?(key) end # Ends the batch for `key`, so the next event starts a new one. - # Called for you after `perform_batch` returns; the ordering matters, - # see the comment inside. + # Called for you after `perform_batch` returns with nothing left + # dirty; the ordering matters, see the comment inside. Deliberately + # does not touch the dirty marker — see `debounce`'s leftover fold-in. def release_debounce(key) # Boundary first, claim second. The other order leaves a moment # where a new event can win the claim and write its own boundary, @@ -120,6 +137,54 @@ def release_debounce(key) Rails.cache.delete(debounce_pending_cache_key(key)) end + # True while a job for `key` is actively inside `perform_batch` — a + # narrower window than the claim itself, which also spans the wait. + # `debounce` uses this to tell "still waiting, will naturally be + # picked up" apart from "query may have already run, could be missed". + def running?(key) + Rails.cache.exist?(debounce_running_cache_key(key)) + end + + def mark_running(key) + Rails.cache.write(debounce_running_cache_key(key), true, expires_in: debounce_state_ttl) + end + + def clear_running(key) + Rails.cache.delete(debounce_running_cache_key(key)) + end + + # Records that an event arrived while perform_batch was running and + # may not have made it into that run's query. Keeps the earliest such + # event so a second and third arrival during the same run don't + # shadow the first. + def mark_dirty(key, event_at) + cache_key = debounce_dirty_cache_key(key) + current = Rails.cache.read(cache_key) + earliest = current ? [ Time.zone.parse(current), event_at ].min : event_at + Rails.cache.write(cache_key, earliest.iso8601(9), expires_in: debounce_state_ttl) + end + + # Returns and clears the dirty boundary, or nil if nothing arrived + # mid-run. + def consume_dirty(key) + cache_key = debounce_dirty_cache_key(key) + raw = Rails.cache.read(cache_key) + return nil unless raw + + Rails.cache.delete(cache_key) + Time.zone.parse(raw) + end + + # Re-arms the claim this job already holds for a follow-up run, + # instead of releasing and re-claiming. Release-then-reclaim would + # open a gap where an unrelated concurrent `debounce` call could win + # the fresh claim with its own (later) boundary and skip over the + # event that's still unhandled here. + def rearm(key, event_at, arguments) + Rails.cache.write(debounce_batch_start_cache_key(key), event_at.iso8601(9), expires_in: debounce_state_ttl) + schedule_run(key, arguments) + end + def debounce_pending_cache_key(key) "coplan:debounced_job:#{debounce_namespace}:pending:#{key}" end @@ -128,6 +193,14 @@ def debounce_batch_start_cache_key(key) "coplan:debounced_job:#{debounce_namespace}:batch_start:#{key}" end + def debounce_running_cache_key(key) + "coplan:debounced_job:#{debounce_namespace}:running:#{key}" + end + + def debounce_dirty_cache_key(key) + "coplan:debounced_job:#{debounce_namespace}:dirty:#{key}" + end + # Namespaces cache keys per job class, so two debounced jobs keyed on # the same record don't fight over one claim. def debounce_namespace @@ -153,18 +226,47 @@ def validate_duration!(value, label) value end + + # Releases the claim if the job never actually made it onto the + # queue — a raised adapter error, or a halted enqueue callback + # returning an unsuccessfully-enqueued job — so a stuck claim doesn't + # block every `debounce` call for this key until debounce_state_ttl + # expires with no job to show for it. + def schedule_run(key, arguments) + job = set(wait: debounce_window).perform_later(key: key, **arguments) + return true if job&.successfully_enqueued? + + release_debounce(key) + false + rescue StandardError + release_debounce(key) + raise + end end # Including jobs implement `perform_batch`, not `perform` — this owns - # `perform` so the claim is always released in the right place. + # `perform` so the claim is always released (or rearmed) in the right + # place. def perform(key:, **arguments) + self.class.mark_running(key) perform_batch(key: key, batch_start: debounce_batch_start(key), **arguments) - # Only after the work is done. Releasing on enqueue, or in an - # `ensure`, would let an event arriving mid-send (or mid-retry, since - # a raise skips this line and leaves the claim held) open a second - # batch that overlaps and clobbers this one. - self.class.release_debounce(key) + # An event that arrived while perform_batch was running may already + # be too late for the query it just ran — rearm the same claim for a + # follow-up instead of releasing, so it isn't silently dropped. + if (dirty_at = self.class.consume_dirty(key)) + self.class.rearm(key, dirty_at, arguments) + else + self.class.release_debounce(key) + end + ensure + # Cleared unconditionally, including on a raise from perform_batch + # (retry_on leaves the claim itself held — see release_debounce not + # being called above on that path). A retry re-queries batch_start + # fresh, so it naturally covers anything that arrived in the gap; + # `running?` only needs to be accurate for the window it's actually + # protecting. + self.class.clear_running(key) end private diff --git a/spec/jobs/coplan/debounced_job_spec.rb b/spec/jobs/coplan/debounced_job_spec.rb index 94d05e85..7b8f5374 100644 --- a/spec/jobs/coplan/debounced_job_spec.rb +++ b/spec/jobs/coplan/debounced_job_spec.rb @@ -120,13 +120,17 @@ def perform_batch(key:, batch_start:) expect(enqueued_jobs.count { |j| j["job_class"] == "DebouncedJobSpecJob" }).to eq(1) end - it "claims with a write-if-absent and never reads the cache first" do + it "claims with a write-if-absent and never reads the pending key to decide" do allow(Rails.cache).to receive(:read).and_call_original allow(Rails.cache).to receive(:write).and_call_original job.debounce(key: key, event_at: 1.minute.ago) - expect(Rails.cache).not_to have_received(:read) + # The claim itself never reads: the atomic write's return value is the + # only input. (A read of the *dirty* key does happen here, to fold in + # anything left over from a previous run — see "leftover dirty state" + # below — but that's unrelated to who wins the claim.) + expect(Rails.cache).not_to have_received(:read).with(job.debounce_pending_cache_key(key)) expect(Rails.cache).to have_received(:write).with( job.debounce_pending_cache_key(key), true, hash_including(unless_exist: true) ).once @@ -268,6 +272,107 @@ def perform_batch(key:, batch_start:) end end + # Failure mode 4: an event lands after perform_batch has already queried + # but before the claim is released. The claim being held makes a plain + # `debounce` call at that moment a no-op, and the query that already ran + # can't retroactively include it — without a rearm, it's gone for good. + describe "events arriving mid-run" do + it "rearms a follow-up run for an event that arrives while perform_batch is executing" do + freeze_time do + job.debounce(key: key, event_at: 30.seconds.ago) + travel 2.minutes + + late_arrival = Time.current + allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do + # Simulates a comment landing after this job's own query already ran. + expect(job.debounce(key: key, event_at: late_arrival)).to be(false) + end + + expect { + job.perform_now(key: key) + }.to have_enqueued_job(DebouncedJobSpecJob).with(key: key) + + expect(job.debounce_pending?(key)).to be(true) + expect(job.batch_start_for(key)).to eq(late_arrival) + end + end + + it "does not rearm when nothing arrives while perform_batch is executing" do + freeze_time do + job.debounce(key: key, event_at: Time.current) + travel 2.minutes + + job.perform_now(key: key) + + expect(job.debounce_pending?(key)).to be(false) + # Only the original debounce's own enqueue — no follow-up rearm. + expect(enqueued_jobs.count { |j| j["job_class"] == "DebouncedJobSpecJob" }).to eq(1) + end + end + + it "keeps the earliest of several mid-run arrivals as the follow-up boundary" do + freeze_time do + job.debounce(key: key, event_at: Time.current) + travel 2.minutes + + earlier = Time.current + later = Time.current + 1.second + allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do + job.debounce(key: key, event_at: later) + job.debounce(key: key, event_at: earlier) + end + + job.perform_now(key: key) + + expect(job.batch_start_for(key)).to eq(earlier) + end + end + + it "does not treat an event still waiting out the window as mid-run" do + freeze_time do + job.debounce(key: key, event_at: Time.current) + + travel 10.seconds + job.debounce(key: key, event_at: Time.current) # still waiting, perform hasn't started + + travel 2.minutes + job.perform_now(key: key) + + expect(job.debounce_pending?(key)).to be(false) # released normally, no rearm needed + end + end + end + + # Failure mode 5: `debounce` writes the claim, then perform_later fails + # (adapter error, or a halted enqueue callback) before actually scheduling + # anything. Without cleanup, every event for this key silently no-ops + # until debounce_state_ttl expires, even though no job exists to run them. + describe "when enqueueing fails" do + it "releases the claim and re-raises when the queue adapter raises" do + configured_job = double("configured_job") + allow(job).to receive(:set).and_return(configured_job) + allow(configured_job).to receive(:perform_later).and_raise(StandardError, "queue unavailable") + + expect { + job.debounce(key: key, event_at: 1.minute.ago) + }.to raise_error(StandardError, "queue unavailable") + + expect(job.debounce_pending?(key)).to be(false) + expect(job.batch_start_for(key)).to be_nil + end + + it "releases the claim when perform_later returns a job that wasn't actually enqueued" do + configured_job = double("configured_job") + allow(job).to receive(:set).and_return(configured_job) + allow(configured_job).to receive(:perform_later).and_return(instance_double(DebouncedJobSpecJob, successfully_enqueued?: false)) + + expect(job.debounce(key: key, event_at: 1.minute.ago)).to be(false) + + expect(job.debounce_pending?(key)).to be(false) + expect(job.batch_start_for(key)).to be_nil + end + end + describe "configuration" do it "defaults to the concern's window and retry horizon" do default_job = Class.new(CoPlan::ApplicationJob) { include CoPlan::DebouncedJob }