From 378f281bef9819d91100fc3c388f9637789cbd60 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 7 Nov 2026 16:08:52 +0800 Subject: [PATCH] fix(batch): leave no record for a canceled job nobody owns after discovery Canceling a job ID this node does not own recorded a canceled owner so that startup discovery, which may still admit the manifest of that ID, would not run a job that was canceled before the node had loaded it. The record was kept for any ID, at any time, and never removed: a caller with admin:CancelBatchJob could grow the ownership table by one entry per cancellation of an unknown ID, and the retry scheduler walks that table every second. Once discovery has finished, nothing admits a job this node does not own: new submissions use new IDs, so the record has nothing left to fence. canceler now records a canceled owner for an unknown ID only while discovery is still running. Canceling an owned job, and the fencing of a cancellation during discovery, are unchanged, and the cancel handler still removes the manifest. TestBatchCancelUnknownJobAfterDiscovery cancels an unknown ID during discovery and requires its admission to be refused, then cancels a thousand unknown IDs after discovery and requires the ownership table not to grow. Before this change it held a thousand records. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Feng Ruohang --- cmd/batch-handlers.go | 14 ++++++++++---- cmd/batch-retry_test.go | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go index 14467a6bf..e86af95a8 100644 --- a/cmd/batch-handlers.go +++ b/cmd/batch-handlers.go @@ -2520,15 +2520,21 @@ func (j *BatchJobPool) canceler(jobID string, cancel bool) error { if !cancel { return nil } // Worker completion uses its generation token. - if j.jobOwners == nil { - j.jobOwners = make(map[string]*batchJobOwner) - } if owner := j.jobOwners[jobID]; owner != nil { if owner.cancel != nil { owner.cancel() } j.jobOwners[jobID] = &batchJobOwner{state: "canceled"} - } else { + return nil + } + // Only startup discovery admits a manifest this node does not own yet. + // A cancellation before that admission leaves a record that fences it; + // once discovery has finished, nothing admits the ID again (submissions + // use new IDs), so an unknown ID leaves no record to accumulate. + if !j.discoveryDone { + if j.jobOwners == nil { + j.jobOwners = make(map[string]*batchJobOwner) + } j.jobOwners[jobID] = &batchJobOwner{state: "canceled"} } return nil diff --git a/cmd/batch-retry_test.go b/cmd/batch-retry_test.go index 987353c23..35a32eb2c 100644 --- a/cmd/batch-retry_test.go +++ b/cmd/batch-retry_test.go @@ -8,6 +8,7 @@ import ( "encoding/binary" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -191,6 +192,38 @@ func TestBatchPoolOwnershipAndCancellation(t *testing.T) { pool.retryDue(time.Now().Add(time.Hour)) // channel remains open; no send/close race. } +// A cancellation of a job this node does not own leaves a record only while +// startup discovery may still admit its manifest. Once discovery has +// finished, canceling unknown IDs must not grow the ownership table. +func TestBatchCancelUnknownJobAfterDiscovery(t *testing.T) { + pool := &BatchJobPool{ctx: t.Context(), jobCh: make(chan *BatchJobRequest, 1)} + // Before discovery finished, the record fences the manifest it may admit. + mustIAM(t, pool.canceler("expire-before-discovery", true)) + if err := pool.queueJob(&BatchJobRequest{ID: "expire-before-discovery"}); !errors.Is(err, context.Canceled) { + t.Fatalf("admission of a job canceled during discovery: %v", err) + } + pool.finishDiscovery() + for i := range 1000 { + mustIAM(t, pool.canceler(fmt.Sprintf("expire-unknown-%d", i), true)) + } + pool.jmu.Lock() + owners := len(pool.jobOwners) + pool.jmu.Unlock() + if owners != 1 { + t.Fatalf("%d owner records after canceling unknown jobs, want only the record from discovery", owners) + } + // An owned job is still canceled and fenced. + job := &BatchJobRequest{ID: "expire-owned"} + mustIAM(t, pool.queueJob(job)) + mustIAM(t, pool.canceler(job.ID, true)) + if job.ctx.Err() == nil { + t.Fatal("cancellation did not stop the owned job") + } + if err := pool.queueJob(&BatchJobRequest{ID: job.ID}); !errors.Is(err, context.Canceled) { + t.Fatalf("stale admission of a canceled job: %v", err) + } +} + type batchPolicyOutage struct { IAMStorageAPI failed string