From fdb46fa9554af64c3e29daf5cf70f7f27f7964b8 Mon Sep 17 00:00:00 2001 From: "Raphael (manual office deploy after cloud-state fix)" Date: Sat, 25 Jul 2026 20:21:14 -0700 Subject: [PATCH] pool: treat an occupied scheduled key as a satisfied transition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A scheduler transition whose planned key is already running had its dispatch rejected with ErrJobExists, which released the scheduler's ownership record and aborted the rest of the plan. The next transition then proposed a fresh dispatch ID and failed identically, so a scheduled job that outlived one interval — or a key held by a foreign job — produced a permanent re-dispatch loop and starved every job planned after it. An occupied key already satisfies the transition. A job this scheduler dispatched keeps its ownership so later transitions retry the same exact dispatch instead of minting a new identity; a foreign occupant drops the speculative claim and is retried next transition, preserving the rule that a scheduler never owns or stops a job it did not dispatch. Either way the remaining planned jobs still run. --- pool/scheduler.go | 42 ++++++++++++++++++++------ pool/scheduler_test.go | 68 ++++++++++++++++++++++++++++++++++++++---- 2 files changed, 96 insertions(+), 14 deletions(-) diff --git a/pool/scheduler.go b/pool/scheduler.go index 56a4519..2d9320d 100644 --- a/pool/scheduler.go +++ b/pool/scheduler.go @@ -275,7 +275,10 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job if err != nil { return fmt.Errorf("read job %q ownership: %w", job.Key, err) } - if dispatchID == "" { + // A pre-existing record proves this scheduler already dispatched the + // key; a fresh proposal means any occupant of the key is foreign. + owned := dispatchID != "" + if !owned { proposed := "scheduler-" + ulid.Make().String() previous, err := sched.claimJobOwnership(ctx, fence, field, proposed) if err != nil { @@ -284,6 +287,7 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job dispatchID = proposed if previous != "" { dispatchID = previous + owned = true } } dispatched := &Job{ @@ -294,16 +298,36 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job dispatchID: dispatchID, } if _, err := sched.dispatchJob(ctx, fence, dispatchID, dispatched); err != nil { - release := errors.Is(err, ErrJobExists) - if !release { - identity, identityErr := dispatchIdentity(job.Key, job.Payload) - if identityErr != nil { - return fmt.Errorf("encode job %q identity: %w", job.Key, identityErr) + if errors.Is(err, ErrJobExists) { + // The scheduled key is already running, which is the state + // this transition wanted. An occupant this scheduler + // dispatched keeps its ownership so later transitions retry + // the same exact dispatch instead of proposing a new one; a + // foreign occupant must never become scheduler-owned, so the + // speculative claim is dropped and the key is retried on the + // next transition. Either way the remaining planned jobs + // still run. + if owned { + continue + } + if releaseErr := sched.releaseOwnership(ctx, fence, field, dispatchID); releaseErr != nil { + return fmt.Errorf("release foreign job %q ownership: %w", job.Key, releaseErr) } - record, readErr := sched.node.readDispatchRecord(ctx, dispatchID, identity) - release = readErr == nil && record.status == dispatchTerminal + sched.logger.Debug( + "scheduled job key held by a foreign job", + "job", job.Key, + "scheduler", sched.name, + ) + continue + } + identity, identityErr := dispatchIdentity(job.Key, job.Payload) + if identityErr != nil { + return fmt.Errorf("encode job %q identity: %w", job.Key, identityErr) } - if release { + record, readErr := sched.node.readDispatchRecord(ctx, dispatchID, identity) + if readErr == nil && record.status == dispatchTerminal { + // The run this scheduler owns already settled: drop ownership + // so the next transition dispatches a fresh run. if releaseErr := sched.releaseOwnership(ctx, fence, field, dispatchID); releaseErr != nil { return errors.Join( fmt.Errorf("dispatch job %q as %q: %w", job.Key, dispatchID, err), diff --git a/pool/scheduler_test.go b/pool/scheduler_test.go index 7b7ad7c..ded34d4 100644 --- a/pool/scheduler_test.go +++ b/pool/scheduler_test.go @@ -372,14 +372,72 @@ func TestSchedulerCollisionNeverOwnsForeignJob(t *testing.T) { } fence := claimTestSchedulerTransition(t, ctx, sched) - err := sched.startJobs(ctx, fence, []*JobParam{{Key: "collision", Payload: []byte("scheduled")}}) - require.ErrorIs(t, err, ErrJobExists) - ownership, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"collision") + // A foreign job already holds the key: the transition is satisfied, so it + // reports no error and plans the remaining jobs, but it must never take + // ownership of a job it did not dispatch. Repeated transitions converge on + // the same outcome instead of erroring every interval. + plan := []*JobParam{ + {Key: "collision", Payload: []byte("scheduled")}, + {Key: "independent", Payload: []byte("independent")}, + } + for range 2 { + require.NoError(t, sched.startJobs(ctx, fence, plan)) + ownership, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"collision") + require.NoError(t, err) + require.Empty(t, ownership, "a foreign job must never become scheduler-owned") + } + // The job planned after the collision still ran and is scheduler-owned. + independent, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"independent") require.NoError(t, err) - require.Empty(t, ownership) + require.NotEmpty(t, independent, "a collision must not abort the remaining plan") + require.NoError(t, sched.clearJobs(ctx, fence)) + require.Eventually(t, func() bool { + jobs := worker.Jobs() + return len(jobs) == 1 && string(jobs[0].Payload) == "foreign" + }, time.Second, time.Millisecond, "clearing schedules must not stop the foreign job") + require.NoError(t, node.Shutdown(ctx)) +} + +func TestSchedulerKeepsOwnershipWhenItsOwnJobStillRuns(t *testing.T) { + rdb := ptesting.NewRedisClient(t) + defer ptesting.CleanupRedis(t, rdb, false, "") + ctx := ptesting.NewTestContext(t) + node := newTestNode(t, ctx, rdb, ulid.Make().String()) + worker := newTestWorker(t, ctx, node) + producer := newTestProducer("schedule", func() (*JobPlan, error) { return &JobPlan{}, nil }) + encodedName := hex.EncodeToString([]byte(producer.Name())) + sched := &scheduler{ + name: node.PoolName + ":schedule", + interval: 20 * time.Millisecond, + producer: producer, + node: node, + keyPrefix: encodedName + ":", + transitionPrefix: "=transition:" + encodedName + ":", + owner: "test-" + ulid.Make().String(), + lease: node.workerTTL, + logger: node.logger, + } + fence := claimTestSchedulerTransition(t, ctx, sched) + plan := []*JobParam{{Key: "recurring", Payload: []byte("scheduled")}} + + require.NoError(t, sched.startJobs(ctx, fence, plan)) + require.Eventually(t, func() bool { return len(worker.Jobs()) == 1 }, time.Second, time.Millisecond) + first, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"recurring") + require.NoError(t, err) + require.NotEmpty(t, first) + + // Later transitions observe their own job still running: the exact + // dispatch identity is preserved instead of proposing a new one, so the + // scheduler converges silently rather than re-dispatching every interval. + for range 3 { + require.NoError(t, sched.startJobs(ctx, fence, plan)) + again, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"recurring") + require.NoError(t, err) + require.Equal(t, first, again) + } require.Len(t, worker.Jobs(), 1) - require.Equal(t, []byte("foreign"), worker.Jobs()[0].Payload) + require.NoError(t, sched.clearJobs(ctx, fence)) require.NoError(t, node.Shutdown(ctx)) }