diff --git a/cmd/batch-discovery_test.go b/cmd/batch-discovery_test.go new file mode 100644 index 000000000..74229fa24 --- /dev/null +++ b/cmd/batch-discovery_test.go @@ -0,0 +1,236 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "net/http" + "testing" + "testing/synctest" + "time" +) + +type batchDiscoveryLayer struct { + ObjectLayer + walk func(context.Context, chan<- itemOrErr[ObjectInfo]) error + read func(context.Context, string) ([]byte, error) +} + +func (a batchDiscoveryLayer) Walk(ctx context.Context, _, _ string, results chan<- itemOrErr[ObjectInfo], _ WalkOptions) error { + return a.walk(ctx, results) +} + +func (a batchDiscoveryLayer) GetObjectNInfo(ctx context.Context, _, object string, _ *HTTPRangeSpec, _ http.Header, opts ObjectOptions) (*GetObjectReader, error) { + data, err := a.read(ctx, object) + if err != nil { + return nil, err + } + return NewGetObjectReaderFromReader(bytes.NewReader(data), ObjectInfo{Name: object}, opts) +} + +// A partially successful startup scan must retry the missing work without +// re-admitting work that finished while the scan was waiting to retry. +func TestBatchDiscoveryRecoversInProcess(t *testing.T) { + for _, failure := range []string{"walk-start", "walk-stream", "manifest-read"} { + t.Run(failure, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + pool := &BatchJobPool{ctx: ctx, jobCh: make(chan *BatchJobRequest, 10)} + attempts := 0 + pool.objLayer = batchDiscoveryLayer{ + walk: func(_ context.Context, results chan<- itemOrErr[ObjectInfo]) error { + attempts++ + if attempts == 1 && failure == "walk-start" { + return errErasureReadQuorum + } + defer close(results) + results <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Name: "finished"}} + if attempts == 1 && failure == "walk-stream" { + results <- itemOrErr[ObjectInfo]{Err: errErasureReadQuorum} + } else { + results <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Name: "pending"}} + } + return nil + }, + read: func(_ context.Context, name string) ([]byte, error) { + if attempts == 1 && failure == "manifest-read" && name == "pending" { + return nil, errErasureReadQuorum + } + job := BatchJobRequest{ID: name} + return job.MarshalMsg(nil) + }, + } + done := make(chan struct{}) + go func() { + defer close(done) + pool.resume(func() time.Duration { return 0 }) + }() + synctest.Wait() + if failure != "walk-start" { + select { + case job := <-pool.jobCh: + if job.ID != "finished" || !pool.beginJob(job) { + t.Fatal("first scan did not admit the healthy manifest") + } + pool.finishJob(job, nil) + default: + t.Fatal("first scan lost the healthy manifest") + } + } + select { + case <-done: + case <-time.After(20 * time.Minute): + t.Fatal("discovery failed to finish after storage recovered") + } + if attempts != 2 { + t.Fatalf("discovery attempts = %d, want 2", attempts) + } + want := 1 + if failure == "walk-start" { + want = 2 + } + if len(pool.jobCh) != want { + t.Fatalf("queued %d jobs, want %d; completed work was re-admitted or pending work was lost", len(pool.jobCh), want) + } + for len(pool.jobCh) > 0 { + job := <-pool.jobCh + if failure != "walk-start" && job.ID != "pending" { + t.Fatalf("re-admitted %q", job.ID) + } + if !pool.beginJob(job) { + t.Fatal("recovered job has no owner") + } + pool.finishJob(job, nil) + } + }) + }) + } +} + +func TestBatchDiscoverySkipsUnusableManifests(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + pool := &BatchJobPool{ctx: t.Context(), jobCh: make(chan *BatchJobRequest, 1)} + attempts := 0 + pool.objLayer = batchDiscoveryLayer{ + walk: func(_ context.Context, results chan<- itemOrErr[ObjectInfo]) error { + attempts++ + defer close(results) + for _, name := range []string{"canceled", "deleted", "malformed", "future", "empty-id"} { + results <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Name: name}} + } + return nil + }, + read: func(_ context.Context, name string) ([]byte, error) { + switch name { + case "deleted": + return nil, errConfigNotFound + case "malformed": + return []byte("bad manifest"), nil + case "future": + return []byte("SILO-BATCH-V999\n"), nil + case "empty-id": + return (&BatchJobRequest{}).MarshalMsg(nil) + } + // The administrator cancels after Walk, during manifest load. + mustIAM(t, pool.canceler(name, true)) + job := BatchJobRequest{ID: name} + return job.MarshalMsg(nil) + }, + } + done := make(chan struct{}) + go func() { + defer close(done) + pool.resume(func() time.Duration { return 0 }) + }() + select { + case <-done: + case <-time.After(time.Hour): + t.Fatal("unusable manifests kept discovery retrying") + } + if len(pool.jobCh) != 0 || attempts != 1 { + t.Fatalf("canceled work was queued or retried: queue=%d scans=%d", len(pool.jobCh), attempts) + } + }) +} + +func TestBatchDiscoveryShutdownDuringBackoff(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + attempts := 0 + pool := &BatchJobPool{ctx: ctx, objLayer: batchDiscoveryLayer{ + walk: func(context.Context, chan<- itemOrErr[ObjectInfo]) error { + attempts++ + return errErasureReadQuorum + }, + }} + done := make(chan struct{}) + go func() { + defer close(done) + pool.resume(func() time.Duration { return 0 }) + }() + synctest.Wait() + select { + case <-done: + t.Fatal("discovery abandoned a transient startup error") + default: + } + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("discovery did not stop on shutdown") + } + if attempts != 1 { + t.Fatalf("discovery scanned after shutdown: %d", attempts) + } + }) +} + +func TestBatchDiscoveryKeepsNewlyFinishedJobsTerminal(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + pool := &BatchJobPool{ctx: t.Context(), jobCh: make(chan *BatchJobRequest, 1)} + attempts := 0 + pool.objLayer = batchDiscoveryLayer{ + walk: func(_ context.Context, results chan<- itemOrErr[ObjectInfo]) error { + attempts++ + if attempts == 1 { + // This job was submitted and failed while startup discovery + // had not yet read any manifests. The retry must not rerun it. + job := &BatchJobRequest{ID: "new-job"} + mustIAM(t, pool.queueJob(job)) + if !pool.beginJob(<-pool.jobCh) { + t.Fatal("submitted job has no owner") + } + pool.finishJob(job, errErasureWriteQuorum) + return errErasureReadQuorum + } + defer close(results) + results <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Name: "new-job"}} + return nil + }, + read: func(_ context.Context, name string) ([]byte, error) { + return (&BatchJobRequest{ID: name}).MarshalMsg(nil) + }, + } + done := make(chan struct{}) + go func() { + defer close(done) + pool.resume(func() time.Duration { return 0 }) + }() + select { + case <-done: + case <-time.After(time.Hour): + t.Fatal("discovery did not recover") + } + if attempts != 2 || len(pool.jobCh) != 0 { + t.Fatalf("discovery re-admitted a failed job: scans=%d queue=%d", attempts, len(pool.jobCh)) + } + if owner := pool.jobOwners["new-job"]; owner.state != "terminal" || owner.req != nil || owner.cancel != nil { + t.Fatalf("terminal marker lost or retained the job payload: %+v", owner) + } + }) +} diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go index 0e0150b10..29f1398cc 100644 --- a/cmd/batch-handlers.go +++ b/cmd/batch-handlers.go @@ -1916,12 +1916,18 @@ func (j *BatchJobRequest) load(ctx context.Context, api ObjectLayer, name string if bytes.HasPrefix(job, []byte("SILO-BATCH")) { if !bytes.HasPrefix(job, []byte(batchJobEnvelope)) { - return errors.New("unsupported batch job envelope version") + return fmt.Errorf("%w: unsupported batch job envelope version", errInvalidBatchJobManifest) } job = job[len(batchJobEnvelope):] } _, err = j.UnmarshalMsg(job) - return err + if err != nil { + return fmt.Errorf("%w: %v", errInvalidBatchJobManifest, err) + } + if j.ID == "" { + return fmt.Errorf("%w: missing job ID", errInvalidBatchJobManifest) + } + return nil } func batchReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (putOpts minio.PutObjectOptions, isMP bool, err error) { @@ -2043,6 +2049,8 @@ func (a adminAPIHandlers) BatchJobStatus(w http.ResponseWriter, r *http.Request) var errNoSuchJob = errors.New("no such job") +var errInvalidBatchJobManifest = errors.New("invalid batch job manifest") + // DescribeBatchJob returns the currently active batch job definition func (a adminAPIHandlers) DescribeBatchJob(w http.ResponseWriter, r *http.Request) { ctx := r.Context() @@ -2320,22 +2328,48 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) { func (j *BatchJobPool) resume(randomWait func() time.Duration) { timer := time.NewTimer(randomWait()) defer timer.Stop() - select { - case <-j.ctx.Done(): - return - case <-timer.C: + // A retry scans the manifest prefix again, but only unadmitted paths may + // create owners. A worker can finish (and release ownership) before a + // later scan; re-admitting that manifest could rerun a failed partial job. + admitted := make(map[string]struct{}) + for retries := uint(0); ; retries++ { + select { + case <-j.ctx.Done(): + return + case <-timer.C: + } + if !j.resumeOnce(admitted) { + return + } + timer.Reset(batchJobRetryDelay(retries)) } +} +// resumeOnce returns whether discovery was incomplete. Already removed or +// canceled manifests are settled; uncertain reads remain eligible for retry. +func (j *BatchJobPool) resumeOnce(admitted map[string]struct{}) bool { results := make(chan itemOrErr[ObjectInfo], 100) ctx, cancel := context.WithCancel(j.ctx) defer cancel() if err := j.objLayer.Walk(ctx, minioMetaBucket, batchJobPrefix, results, WalkOptions{}); err != nil { batchLogIf(j.ctx, err) - return + return true } - for result := range results { + incomplete := false + for { + var result itemOrErr[ObjectInfo] + select { + case <-ctx.Done(): + return false + case next, ok := <-results: + if !ok { + return incomplete + } + result = next + } if result.Err != nil { batchLogIf(j.ctx, result.Err) + incomplete = true continue } if strings.HasPrefix(result.Item.Name, batchJobReportsPrefix+slashSeparator) { @@ -2345,9 +2379,20 @@ func (j *BatchJobPool) resume(randomWait func() time.Duration) { if strings.HasSuffix(result.Item.Name, slashSeparator) { continue } + if _, ok := admitted[result.Item.Name]; ok { + continue + } req := &BatchJobRequest{} if err := req.load(ctx, j.objLayer, result.Item.Name); err != nil { - batchLogIf(ctx, err) + if !errors.Is(err, errNoSuchJob) { + batchLogIf(ctx, err) + // Unsupported or malformed definitions cannot recover through + // another storage read. Keep them on disk for inspection, but + // do not let one prevent discovery and report cleanup finishing. + if !errors.Is(err, errInvalidBatchJobManifest) { + incomplete = true + } + } continue } _, nodeIdx := parseRequestToken(req.ID) @@ -2356,9 +2401,16 @@ func (j *BatchJobPool) resume(randomWait func() time.Duration) { continue } if err := j.queueJob(req); err != nil { + if errors.Is(err, context.Canceled) { + // A tombstone is authoritative even if Walk saw the manifest + // before cancellation removed it from storage. + continue + } batchLogIf(ctx, err) + incomplete = true continue } + admitted[result.Item.Name] = struct{}{} } } diff --git a/cmd/batch-retry.go b/cmd/batch-retry.go index 16595cbcd..9a5080813 100644 --- a/cmd/batch-retry.go +++ b/cmd/batch-retry.go @@ -105,8 +105,9 @@ func validateBatchJobAdminReq(ctx context.Context, w http.ResponseWriter, r *htt } // One owner record spans queued, running and paused attempts. The request -// pointer is the generation token. Canceled records are tombstones until -// process shutdown, so a manifest loaded before cancellation cannot requeue. +// pointer is the generation token. Canceled and terminal records are small +// tombstones until process shutdown, so a delayed discovery cannot requeue a +// job after its owner finished, including a terminal partial failure. type batchJobOwner struct { req *BatchJobRequest cancel context.CancelFunc @@ -134,14 +135,18 @@ func (j *BatchJobPool) finishJob(job *BatchJobRequest, err error) { return } if isBatchJobUnavailable(err) && job.ctx.Err() == nil && j.ctx.Err() == nil { - delay := 30 * time.Second * time.Duration(1<