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