mirror of
https://github.com/pgsty/minio.git
synced 2026-11-11 01:55:30 +03:00
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 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
+10
-4
@@ -2520,15 +2520,21 @@ func (j *BatchJobPool) canceler(jobID string, cancel bool) error {
|
|||||||
if !cancel {
|
if !cancel {
|
||||||
return nil
|
return nil
|
||||||
} // Worker completion uses its generation token.
|
} // 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 := j.jobOwners[jobID]; owner != nil {
|
||||||
if owner.cancel != nil {
|
if owner.cancel != nil {
|
||||||
owner.cancel()
|
owner.cancel()
|
||||||
}
|
}
|
||||||
j.jobOwners[jobID] = &batchJobOwner{state: "canceled"}
|
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"}
|
j.jobOwners[jobID] = &batchJobOwner{state: "canceled"}
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"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.
|
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 {
|
type batchPolicyOutage struct {
|
||||||
IAMStorageAPI
|
IAMStorageAPI
|
||||||
failed string
|
failed string
|
||||||
|
|||||||
Reference in New Issue
Block a user