fix(batch): release terminal markers after startup discovery

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/batch-discovery_test.go

@@ -209,6 +209,9 @@ func TestBatchDiscoveryKeepsNewlyFinishedJobsTerminal(t *testing.T) {
 					return errErasureReadQuorum
 				}
 				defer close(results)
+				if owner := pool.jobOwners["new-job"]; owner == nil || owner.state != "terminal" || owner.req != nil || owner.cancel != nil {
+					t.Fatalf("terminal marker lost or retained the job payload: %+v", owner)
+				}
 				results <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Name: "new-job"}}
 				return nil
 			},
@@ -229,8 +232,17 @@ func TestBatchDiscoveryKeepsNewlyFinishedJobsTerminal(t *testing.T) {
 		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 {
+		if !pool.discoveryDone || len(pool.jobOwners) != 0 {
-			t.Fatalf("terminal marker lost or retained the job payload: %+v", owner)
+			t.Fatal("completed discovery retained terminal job markers")
+		}
+		job := &BatchJobRequest{ID: "after-discovery"}
+		mustIAM(t, pool.queueJob(job))
+		if !pool.beginJob(<-pool.jobCh) {
+			t.Fatal("new submission after discovery has no owner")
+		}
+		pool.finishJob(job, nil)
+		if len(pool.jobOwners) != 0 {
+			t.Fatal("normal completion after discovery leaked an owner")
 		}
 	})
 }

cmd/batch-handlers.go

@@ -2242,14 +2242,15 @@ func (a adminAPIHandlers) CancelBatchJob(w http.ResponseWriter, r *http.Request)
 
 // BatchJobPool batch job pool
 type BatchJobPool struct {
-	ctx          context.Context
+	ctx           context.Context
-	objLayer     ObjectLayer
+	objLayer      ObjectLayer
-	mu           sync.Mutex
+	mu            sync.Mutex
-	jobCh        chan *BatchJobRequest
+	jobCh         chan *BatchJobRequest
-	jmu          sync.Mutex // protects jobOwners
+	jmu           sync.Mutex // protects jobOwners and discoveryDone
-	jobOwners    map[string]*batchJobOwner
+	jobOwners     map[string]*batchJobOwner
-	workerKillCh chan struct{}
+	discoveryDone bool
-	workerSize   int
+	workerKillCh  chan struct{}
+	workerSize    int
 }
 
 var globalBatchJobPool *BatchJobPool
@@ -2326,6 +2327,7 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) {
 }
 
 func (j *BatchJobPool) resume(randomWait func() time.Duration) {
+	defer j.finishDiscovery()
 	timer := time.NewTimer(randomWait())
 	defer timer.Stop()
 	// A retry scans the manifest prefix again, but only unadmitted paths may

cmd/batch-retry.go

@@ -105,9 +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 and terminal records are small
+// pointer is the generation token. Canceled records fence stale requests until
-// tombstones until process shutdown, so a delayed discovery cannot requeue a
+// shutdown. Terminal records fence discovery until the startup scan finishes,
-// job after its owner finished, including a terminal partial failure.
+// including jobs submitted and terminated while the scan was unavailable.
 type batchJobOwner struct {
 	req     *BatchJobRequest
 	cancel  context.CancelFunc
@@ -141,7 +141,25 @@ func (j *BatchJobPool) finishJob(job *BatchJobRequest, err error) {
 		return
 	}
 	o.cancel()
-	j.jobOwners[job.ID] = &batchJobOwner{state: "terminal"}
+	if j.discoveryDone {
+		delete(j.jobOwners, job.ID)
+	} else {
+		j.jobOwners[job.ID] = &batchJobOwner{state: "terminal"}
+	}
+}
+
+// After the only startup discovery has returned, no loaded manifest remains
+// to re-admit a terminal job. Release its marker, but keep queued, running,
+// paused and canceled ownership intact. New submissions use new job IDs.
+func (j *BatchJobPool) finishDiscovery() {
+	j.jmu.Lock()
+	defer j.jmu.Unlock()
+	j.discoveryDone = true
+	for id, owner := range j.jobOwners {
+		if owner.state == "terminal" {
+			delete(j.jobOwners, id)
+		}
+	}
 }
 
 func batchJobRetryDelay(retries uint) time.Duration {