From 40afd7b72d7d22217bf1d09a2e8d9844b33ecd19 Mon Sep 17 00:00:00 2001 From: Michael Lovitt Date: Thu, 9 Jul 2026 13:54:18 -0500 Subject: [PATCH 1/3] Gate batch completion on the parent finishing enqueuing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `schedule_pending_jobs` dispatches each child to Cloud Tasks and only then registers it in the batch state, one child at a time. `complete?` considers a batch done as soon as every *currently-registered* child is complete — it has no notion of the parent having finished enqueuing. On a large fan-out with uneven child durations, fast early children can complete while the parent is still scheduling slower siblings. At that point every registered child is complete, so `on_child_complete` fires `on_batch_complete` prematurely — before the slow children have even been scheduled. The one-shot completion guard makes this worse: it fires exactly once, so the early (wrong) result is never corrected. Flag the batch as fully enqueued at the end of `setup`, and require that flag before a child may trigger completion. The parent's own post-`setup` `complete` call remains the backstop for the case where every child finishes before the flag is set, so no batch can hang. Children still register after dispatch, so a partial setup cannot orphan state. --- lib/cloudtasker/batch/job.rb | 37 ++++++++++++++++++++++++++++++ spec/cloudtasker/batch/job_spec.rb | 28 +++++++++++++++++++++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/lib/cloudtasker/batch/job.rb b/lib/cloudtasker/batch/job.rb index ee005ab..8c00395 100644 --- a/lib/cloudtasker/batch/job.rb +++ b/lib/cloudtasker/batch/job.rb @@ -206,6 +206,16 @@ def batch_completion_gid "#{batch_state_gid}/completed" end + # + # Return the key used to flag that the parent has finished enqueuing all + # children of the batch. + # + # @return [String] The batch setup-complete key. + # + def batch_setup_gid + "#{batch_state_gid}/setup_complete" + end + # # Return the number of jobs in a given state # @@ -371,6 +381,22 @@ def complete? redis.hvals(batch_state_gid).all? { |e| COMPLETION_STATUSES.include?(e) } end + # + # Return true once the parent has finished enqueuing all children of the + # batch (end of #setup). + # + # Completion must not be evaluated before this point: the parent enqueues + # children one at a time, so a fast child can complete while later siblings + # have not yet been scheduled or registered. Without this gate, `complete?` + # would see only the children registered so far and report the batch done, + # firing `on_batch_complete` prematurely. + # + # @return [Boolean] True if the batch has been fully enqueued. + # + def setup_complete? + redis.exists?(batch_setup_gid) + end + # # Run worker callback. The error and dead callbacks get # silenced should they raise an error. @@ -436,6 +462,11 @@ def on_child_complete(child_batch, status = :completed) return if status == :errored return unless complete? + # Do not let a child complete the batch until the parent has finished + # enqueuing every child. Otherwise a fast child can fire on_complete + # before its slower siblings have even been scheduled. See #setup_complete?. + return unless setup_complete? + # Notify the parent batch that we are done with this batch. Use SETNX to ensure # only the first concurrent child to complete triggers on_complete. return unless redis.set(batch_completion_gid, true, nx: true, ex: BATCH_COMPLETION_TTL) @@ -480,6 +511,7 @@ def cleanup m.del(batch_gid) m.del(batch_state_gid) m.del(batch_completion_gid) + m.del(batch_setup_gid) BATCH_STATUSES.each { |e| m.del(batch_state_count_gid(e)) } end end @@ -553,6 +585,11 @@ def setup # Schedule all child workers schedule_pending_jobs + + # Flag the batch as fully enqueued. Until this is set, a child that + # completes must not be allowed to trigger batch completion — see + # #setup_complete?. + redis.set(batch_setup_gid, true) end # diff --git a/spec/cloudtasker/batch/job_spec.rb b/spec/cloudtasker/batch/job_spec.rb index b35f861..bac8c91 100644 --- a/spec/cloudtasker/batch/job_spec.rb +++ b/spec/cloudtasker/batch/job_spec.rb @@ -388,6 +388,7 @@ expect(child_worker).not_to receive(:schedule) end + after { expect(batch).not_to be_setup_complete } it { is_expected.to be_truthy } end @@ -399,10 +400,25 @@ expect(batch).to receive(:schedule_pending_jobs).and_return(true) end + after { expect(batch).to be_setup_complete } it { is_expected.to be_truthy } end end + describe '#setup_complete?' do + subject { batch } + + context 'without the batch having been enqueued' do + it { is_expected.not_to be_setup_complete } + end + + context 'with the batch flagged as fully enqueued' do + before { redis.set(batch.batch_setup_gid, true) } + + it { is_expected.to be_setup_complete } + end + end + describe '#update_state' do subject { batch.batch_state&.dig(child_id) } @@ -592,7 +608,7 @@ let(:complete) { true } before do - allow(batch).to receive_messages(complete?: complete, on_complete: true) + allow(batch).to receive_messages(complete?: complete, on_complete: true, setup_complete?: true) allow(batch).to receive(:update_state).with(child_batch.batch_id, status) allow(batch).to receive(:run_worker_callback) batch.pending_jobs.push(child_worker) @@ -612,6 +628,15 @@ it { is_expected.to be_truthy } end + context 'with batch complete but the parent still enqueuing children' do + before { allow(batch).to receive(:setup_complete?).and_return(false) } + + after { expect(batch).to have_received(:update_state) } + after { expect(batch).to have_received(:run_worker_callback).with(:on_child_complete, child_batch.worker) } + after { expect(batch).not_to have_received(:on_complete) } + it { is_expected.to be_falsey } + end + context 'with batch complete but completion already claimed by another child' do before do expect(redis).to receive(:set) @@ -708,6 +733,7 @@ [ side_batch.batch_gid, side_batch.batch_state_gid, + side_batch.batch_setup_gid, side_batch.batch_state_count_gid('all'), side_batch.batch_state_count_gid('scheduled') ].sort From 85f7dfb087ec8ac8daee9a50116eb0dfd9279649 Mon Sep 17 00:00:00 2001 From: Michael Lovitt Date: Thu, 9 Jul 2026 14:24:26 -0500 Subject: [PATCH 2/3] Add coverage for batch expansion under the setup seal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Progressive expansion (a running child adding more jobs to an already-enqueued batch) is a supported workflow. It is unaffected by the premature-completion race — and by the seal — because the expanding child sits in the parent's batch state as `processing` until after it has registered its additions, so the batch cannot be judged complete mid- expansion. Pin that down so the seal can't regress it. --- spec/cloudtasker/batch/job_spec.rb | 43 ++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/spec/cloudtasker/batch/job_spec.rb b/spec/cloudtasker/batch/job_spec.rb index bac8c91..3a34d55 100644 --- a/spec/cloudtasker/batch/job_spec.rb +++ b/spec/cloudtasker/batch/job_spec.rb @@ -419,6 +419,49 @@ end end + describe 'progressive batch expansion' do + # A running child may add more jobs to a batch that has already been set up + # (see docs/BATCH_JOBS.md "Expanding the parent batch"). The setup seal is + # set once at the initial setup and must not interfere: the batch stays open + # while the expanding child runs, then completes normally. + let(:initial_child) { worker.new_instance } + let(:added_child) { worker.new_instance } + + before do + allow_any_instance_of(Cloudtasker::Worker).to receive(:schedule) + .and_return(instance_double(Cloudtasker::CloudTask)) + + # Parent sets up with one initial child and seals the batch + batch.pending_jobs.push(initial_child) + batch.setup + + # The initial child starts running, then expands the batch before completing + batch.update_state(initial_child.job_id, 'processing') + batch.pending_jobs.push(added_child) + batch.schedule_pending_jobs + end + + it { expect(batch).to be_setup_complete } + + context 'with the expanding child still running' do + before { batch.update_state(added_child.job_id, 'completed') } + + # The expanding child's own 'processing' entry holds the batch open, so + # the freshly added child completing does not complete the batch. + it { expect(batch).not_to be_complete } + end + + context 'with the expanding child and the added child both completed' do + before do + batch.update_state(added_child.job_id, 'completed') + batch.update_state(initial_child.job_id, 'completed') + end + + it { expect(batch).to be_complete } + it { expect(batch).to be_setup_complete } + end + end + describe '#update_state' do subject { batch.batch_state&.dig(child_id) } From e964835ae822b20d66699a01d07a9dddd050d802 Mon Sep 17 00:00:00 2001 From: Michael Lovitt Date: Fri, 10 Jul 2026 11:53:17 -0500 Subject: [PATCH 3/3] Make the enqueuing gate backward-compatible across upgrades Track setup as in_progress/done instead of a single end-of-setup flag, and treat a marker-less batch (enqueued before this change) as fully enqueued so batches in flight during an upgrade are not stranded. --- lib/cloudtasker/batch/job.rb | 34 +++++++++++----------- spec/cloudtasker/batch/job_spec.rb | 45 ++++++++++++++++++++++++------ 2 files changed, 54 insertions(+), 25 deletions(-) diff --git a/lib/cloudtasker/batch/job.rb b/lib/cloudtasker/batch/job.rb index 8c00395..fea0f9e 100644 --- a/lib/cloudtasker/batch/job.rb +++ b/lib/cloudtasker/batch/job.rb @@ -207,10 +207,10 @@ def batch_completion_gid end # - # Return the key used to flag that the parent has finished enqueuing all - # children of the batch. + # Return the key tracking the batch enqueuing lifecycle: 'in_progress' + # while the parent enqueues children, 'done' once finished. # - # @return [String] The batch setup-complete key. + # @return [String] The batch setup marker key. # def batch_setup_gid "#{batch_state_gid}/setup_complete" @@ -382,19 +382,19 @@ def complete? end # - # Return true once the parent has finished enqueuing all children of the - # batch (end of #setup). + # Return true unless the parent is still enqueuing children. The parent + # enqueues one at a time, so a fast child could otherwise complete off a + # partially-registered state and fire `on_batch_complete` prematurely + # (see #on_child_complete). # - # Completion must not be evaluated before this point: the parent enqueues - # children one at a time, so a fast child can complete while later siblings - # have not yet been scheduled or registered. Without this gate, `complete?` - # would see only the children registered so far and report the batch done, - # firing `on_batch_complete` prematurely. + # A batch with no marker predates this feature (enqueued by an older + # #setup); treat it as fully enqueued so batches in flight during an + # upgrade complete normally instead of being stranded. # - # @return [Boolean] True if the batch has been fully enqueued. + # @return [Boolean] True unless the batch is still enqueuing children. # def setup_complete? - redis.exists?(batch_setup_gid) + redis.get(batch_setup_gid) != 'in_progress' end # @@ -583,13 +583,15 @@ def setup # Save batch save + # Mark as enqueuing before scheduling any child: while set, a completing + # child must not trigger batch completion (see #setup_complete?). + redis.set(batch_setup_gid, 'in_progress') + # Schedule all child workers schedule_pending_jobs - # Flag the batch as fully enqueued. Until this is set, a child that - # completes must not be allowed to trigger batch completion — see - # #setup_complete?. - redis.set(batch_setup_gid, true) + # Mark as fully enqueued; children may now trigger completion. + redis.set(batch_setup_gid, 'done') end # diff --git a/spec/cloudtasker/batch/job_spec.rb b/spec/cloudtasker/batch/job_spec.rb index 3a34d55..87d09ed 100644 --- a/spec/cloudtasker/batch/job_spec.rb +++ b/spec/cloudtasker/batch/job_spec.rb @@ -388,7 +388,7 @@ expect(child_worker).not_to receive(:schedule) end - after { expect(batch).not_to be_setup_complete } + after { expect(redis.get(batch.batch_setup_gid)).to be_nil } it { is_expected.to be_truthy } end @@ -400,7 +400,7 @@ expect(batch).to receive(:schedule_pending_jobs).and_return(true) end - after { expect(batch).to be_setup_complete } + after { expect(redis.get(batch.batch_setup_gid)).to eq('done') } it { is_expected.to be_truthy } end end @@ -408,22 +408,30 @@ describe '#setup_complete?' do subject { batch } - context 'without the batch having been enqueued' do + context 'when the batch is still enqueuing children' do + before { redis.set(batch.batch_setup_gid, 'in_progress') } + it { is_expected.not_to be_setup_complete } end - context 'with the batch flagged as fully enqueued' do - before { redis.set(batch.batch_setup_gid, true) } + context 'when the batch is fully enqueued' do + before { redis.set(batch.batch_setup_gid, 'done') } + + it { is_expected.to be_setup_complete } + end + context 'without a marker (batch enqueued before this feature)' do + # Backward compatibility: batches in flight during an upgrade have no + # marker and must be treated as fully enqueued so they are not stranded. it { is_expected.to be_setup_complete } end end describe 'progressive batch expansion' do # A running child may add more jobs to a batch that has already been set up - # (see docs/BATCH_JOBS.md "Expanding the parent batch"). The setup seal is - # set once at the initial setup and must not interfere: the batch stays open - # while the expanding child runs, then completes normally. + # (see docs/BATCH_JOBS.md "Expanding the parent batch"). The setup marker is + # written during the initial enqueue and must not interfere: the batch stays + # open while the expanding child runs, then completes normally. let(:initial_child) { worker.new_instance } let(:added_child) { worker.new_instance } @@ -431,7 +439,7 @@ allow_any_instance_of(Cloudtasker::Worker).to receive(:schedule) .and_return(instance_double(Cloudtasker::CloudTask)) - # Parent sets up with one initial child and seals the batch + # Parent sets up with one initial child and marks the batch enqueued batch.pending_jobs.push(initial_child) batch.setup @@ -462,6 +470,25 @@ end end + describe 'batch enqueued before the setup marker (upgrade in flight)' do + # A batch enqueued by a previous version has no setup marker. After an + # upgrade its remaining children must still be able to complete it - the + # parent already ran under the old code and cannot act as the fallback - + # otherwise the batch would be stranded and on_batch_complete never fire. + before do + batch.save + redis.hset(batch.batch_state_gid, child_batch.batch_id, 'processing') + end + + it { expect(redis.get(batch.batch_setup_gid)).to be_nil } + it { expect(batch).to be_setup_complete } + + it 'lets the final child complete the batch' do + expect(batch).to receive(:on_complete) + batch.on_child_complete(child_batch, :completed) + end + end + describe '#update_state' do subject { batch.batch_state&.dig(child_id) }