fix: remove SIGTERM re-enqueue and make heartbeat interval configurable (llm-d#536)

Remove the processor's SIGTERM re-enqueue logic (errShutdown handler)
and delegate crash recovery entirely to the orphan reconciler. This
resolves the TODO in the codebase and simplifies the shutdown path.

Make heartbeat_interval configurable so dev-deploy can set it shorter
than reconciler.interval, preventing the reconciler from falsely marking
active jobs as orphaned during e2e tests.

Update ProcessorGracefulShutdown and OrphanRecovery e2e tests to expect
reconciler-driven failure instead of re-enqueue completion. Increase
e2e test timeout to 20m for reconciler wait time.

Fixes llm-d#532

Signed-off-by: Jooyeon Mok <jmok@redhat.com>

Author: j-mok-dev

Makefile

@@ -385,7 +385,7 @@ test-e2e:
 	@echo "Running E2E tests..."
 	@OUT=$$(mktemp); \
 	echo "Processor observability endpoint: auto-resolved by the e2e test helpers"; \
-	cd test/e2e && $(GO) test -v -count=1 $(if $(TEST_RUN),-run $(TEST_RUN)) ./... 2>&1 | tee $$OUT; \
+	cd test/e2e && $(GO) test -v -count=1 -timeout=20m $(if $(TEST_RUN),-run $(TEST_RUN)) ./... 2>&1 | tee $$OUT; \
 	TEST_EXIT=$${PIPESTATUS[0]}; \
 	PASS_COUNT=$$(grep -- '--- PASS:' $$OUT 2>/dev/null | wc -l | tr -d ' '); \
 	FAIL_COUNT=$$(grep -- '--- FAIL:' $$OUT 2>/dev/null | wc -l | tr -d ' '); \

charts/batch-gateway/templates/processor-configmap.yaml

@@ -25,6 +25,7 @@ data:
 
     terminate_on_observability_failure: {{ .Values.processor.config.terminateOnObservabilityFailure }}
     shutdown_timeout: {{ .Values.processor.config.shutdownTimeout | quote }}
+    heartbeat_interval: {{ .Values.processor.config.heartbeatInterval | quote }}
     work_dir: {{ .Values.processor.config.workDir | quote }}
 
     db_client:

charts/batch-gateway/values.yaml

@@ -344,6 +344,7 @@ processor:
       count: 12
     terminateOnObservabilityFailure: false
     shutdownTimeout: "30s"
+    heartbeatInterval: "5m"
     workDir: "/var/lib/batch-gateway/processor"
     # REQUIRED: Configure exactly one of globalInferenceGateway or modelGateways.
     # The processor will fail to start if neither is set.

cmd/batch-processor/config.yaml

@@ -34,6 +34,11 @@ terminate_on_observability_failure: false
 # Shutdown timeout for processor (should be less than k8s terminationGracePeriodSeconds)
 shutdown_timeout: "30s"
 
+# Heartbeat interval for in-flight entry refresh.
+# Must be shorter than the GC reconciler's interval so active jobs are not
+# mistaken for orphans. Default: 5m (matches GC default of 60m).
+heartbeat_interval: "5m"
+
 # Work directory for processor
 work_dir: "/var/lib/batch-gateway/processor"
 

internal/processor/config/config.go

@@ -167,6 +167,14 @@ type ProcessorConfig struct {
 	// FileClient holds configuration for the shared file storage client (fs or s3).
 	FileClientCfg sharedcfg.FileClientConfig `yaml:"file_client"`
 
+	// HeartbeatInterval controls how often the processor refreshes its in-flight
+	// entry for a running job. The orphan reconciler uses staleness (no heartbeat
+	// for > reconciler interval) to detect abandoned jobs.
+	// Must be shorter than the reconciler's interval so that active jobs are
+	// never mistaken for orphans.
+	// Zero means use the default (5 minutes).
+	HeartbeatInterval time.Duration `yaml:"heartbeat_interval"`
+
 	// DispatchMode selects the inference dispatch backend.
 	// "sync" (default): direct HTTP via InferenceClient.
 	// "async": submit via llm-d-async producer, collect from result queue.
@@ -321,7 +329,8 @@ func NewConfig() *ProcessorConfig {
 		DefaultOutputExpirationSeconds: 90 * 24 * 60 * 60, // 90 days
 		ProgressTTLSeconds:             24 * 60 * 60,      // 24 hours
 
-		DispatchMode: DispatchModeSync,
+		HeartbeatInterval: 5 * time.Minute,
+		DispatchMode:      DispatchModeSync,
 		AsyncDispatchConfig: AsyncDispatchConfig{
 			ResultPollTimeout: 5 * time.Second,
 		},
@@ -349,6 +358,9 @@ func (c *ProcessorConfig) Validate() error {
 	if c.ShutdownTimeout <= 0 {
 		return fmt.Errorf("shutdown_timeout must be > 0")
 	}
+	if c.HeartbeatInterval <= 0 {
+		return fmt.Errorf("heartbeat_interval must be > 0")
+	}
 	if c.Addr == "" {
 		return fmt.Errorf("addr cannot be empty")
 	}

internal/processor/worker/errors.go

@@ -35,7 +35,8 @@ var (
 	errRequestInputRead = errors.New("failed to read request from input file")
 
 	// errShutdown signals that the processor is shutting down (SIGTERM).
-	// The job is re-enqueued so another worker can pick it up.
+	// The job is left in its current non-terminal state for the orphan
+	// reconciler to detect and transition to a terminal state.
 	errShutdown = errors.New("processor shutting down")
 
 	// errFinalizeFailedOver signals that a terminal status transition (completed,

internal/processor/worker/executor.go

@@ -168,8 +168,8 @@ func (ep *executionProgress) counts() *openai.BatchRequestCounts {
 //   - SLO expired:    (counts, errExpired)   — undispatched drained as batch_expired
 //   - User cancel:    (counts, errCancelled) — undispatched drained as batch_cancelled
 //   - System error:   (counts, firstErr)     — undispatched drained as batch_failed
-//   - Pod shutdown:   (counts, errShutdown)  — caller re-enqueues; counts reflect work done
-//     before SIGTERM, flush preserves partial output for startup recovery
+//   - Pod shutdown:   (counts, errShutdown)  — job left for orphan reconciler; counts reflect
+//     work done before SIGTERM
 //
 // requestAbortCtx controls the dispatch loop and all in-flight inference calls: cancelling it
 // stops dispatch and aborts in-flight requests. It is derived from sloCtx in runJob, so SLO
@@ -600,10 +600,10 @@ dispatch:
 	default:
 		if mainCtx.Err() != nil && (len(undispatched) > 0 || shutdownCancelled.Load() > 0) {
 			// Pod shutdown (SIGTERM): main processor context is cancelled.
-			// Do not drain here — the re-enqueued worker will process the
-			// entire job from scratch. The undispatched check catches SIGTERM
-			// arriving during dispatch; shutdownCancelled catches SIGTERM
-			// cancelling already-dispatched in-flight requests.
+			// Do not drain here — the job will be left for the orphan
+			// reconciler to transition to a terminal state. The undispatched
+			// check catches SIGTERM arriving during dispatch; shutdownCancelled
+			// catches SIGTERM cancelling already-dispatched in-flight requests.
 			returnErr = errShutdown
 		} else if requestAbortCtx.Err() != nil && len(undispatched) > 0 {
 			// Sibling model abort: requestAbortCtx was cancelled by another

internal/processor/worker/executor_test.go

@@ -1089,8 +1089,8 @@ func TestProcessModel_ContextCancelledDuringDispatch(t *testing.T) {
 
 // TestProcessModel_SIGTERMCancelsAllDispatched verifies that when SIGTERM arrives after all
 // requests have been dispatched as goroutines, processModel returns errShutdown so the job
-// is re-enqueued. This covers the case where undispatched is empty but in-flight requests
-// were cancelled by context propagation.
+// is left for the orphan reconciler. This covers the case where undispatched is empty but
+// in-flight requests were cancelled by context propagation.
 func TestProcessModel_SIGTERMCancelsAllDispatched(t *testing.T) {
 	cfg := config.NewConfig()
 	cfg.WorkDir = t.TempDir()
@@ -1142,8 +1142,8 @@ func TestProcessModel_SIGTERMCancelsAllDispatched(t *testing.T) {
 //
 // P1c regression: Before the fix, the drain-switch default checked ctx.Err() (which is
 // requestAbortCtx), so a sibling abort looked like a pod shutdown and returned errShutdown.
-// executeJob would then route the job to re-enqueue instead of failed-with-partial, breaking
-// retry safety for batches with partial results. After the fix, the default checks mainCtx
+// executeJob would then route the job to the errShutdown path instead of failed-with-partial,
+// breaking retry safety for batches with partial results. After the fix, the default checks mainCtx
 // (the main processor context), which is only cancelled on SIGTERM.
 func TestProcessModel_SiblingAbort_ReturnsNil(t *testing.T) {
 	cfg := config.NewConfig()
@@ -1422,7 +1422,7 @@ func TestExecuteJob_CancelAfterAllRequestsComplete(t *testing.T) {
 // TestExecuteJob_SIGTERMAfterAllComplete verifies that when all requests finish successfully
 // and SIGTERM arrives before executeJob returns, the function returns nil (not errShutdown).
 // This ensures the caller proceeds to finalizeJob (which uses a detached context) rather than
-// re-enqueueing a fully-complete job.
+// taking the errShutdown path (which leaves the job for the orphan reconciler).
 func TestExecuteJob_SIGTERMAfterAllComplete(t *testing.T) {
 	cfg := config.NewConfig()
 	cfg.WorkDir = t.TempDir()
@@ -2105,7 +2105,7 @@ func TestHandleJobError_errCancelled(t *testing.T) {
 	}
 }
 
-func TestHandleJobError_Shutdown_ReEnqueues(t *testing.T) {
+func TestHandleJobError_Shutdown_LeavesJobForReconciler(t *testing.T) {
 	cfg := config.NewConfig()
 	cfg.WorkDir = t.TempDir()
 
@@ -2118,8 +2118,6 @@ func TestHandleJobError_Shutdown_ReEnqueues(t *testing.T) {
 		BatchJob: &openai.Batch{BatchSpec: openai.BatchSpec{CreatedAt: time.Now().Add(-10 * time.Second).Unix()}},
 	}
 
-	beforeFailed := gatherHistogramSampleCount(t, "batch_job_e2e_latency_seconds", map[string]string{"status": "failed"})
-
 	ctx := testLoggerCtx(t)
 	env.p.handleJobError(ctx, &jobExecutionParams{
 		updater: env.updater,
@@ -2128,17 +2126,26 @@ func TestHandleJobError_Shutdown_ReEnqueues(t *testing.T) {
 		jobInfo: ji,
 	}, errShutdown)
 
-	afterFailed := gatherHistogramSampleCount(t, "batch_job_e2e_latency_seconds", map[string]string{"status": "failed"})
-	if delta := afterFailed - beforeFailed; delta != 0 {
-		t.Fatalf("E2E latency failed: delta=%d, want 0 (re-enqueue succeeded, not terminal)", delta)
-	}
-
+	// Job must NOT be re-enqueued — reconciler handles recovery.
 	tasks, err := env.pqClient.PQDequeue(ctx, 0, 10)
 	if err != nil {
 		t.Fatalf("PQDequeue: %v", err)
 	}
-	if len(tasks) == 0 {
-		t.Fatalf("expected re-enqueued task, got none")
+	if len(tasks) != 0 {
+		t.Fatalf("expected no re-enqueued tasks, got %d", len(tasks))
+	}
+
+	// Job status must remain in_progress (not transitioned by the processor).
+	items, _, _, err := env.dbClient.DBGet(ctx, &db.BatchQuery{BaseQuery: db.BaseQuery{IDs: []string{"job-ctx"}}}, true, 0, 1)
+	if err != nil || len(items) != 1 {
+		t.Fatalf("DBGet: err=%v len=%d", err, len(items))
+	}
+	var got openai.BatchStatusInfo
+	if err := json.Unmarshal(items[0].Status, &got); err != nil {
+		t.Fatalf("unmarshal status: %v", err)
+	}
+	if got.Status != openai.BatchStatusInProgress {
+		t.Fatalf("status = %s, want %s (unchanged, left for reconciler)", got.Status, openai.BatchStatusInProgress)
 	}
 }
 

internal/processor/worker/job_runner.go

@@ -47,11 +47,9 @@ import (
 // Declared as var (not const) so tests can shorten it.
 var panicRecoveryTimeout = time.Minute
 
-// heartbeatInterval controls how often the processor refreshes its in-flight
-// entry for a running job. The orphan reconciler uses staleness (no heartbeat
-// for > reconciler interval) to detect abandoned jobs.
-// Declared as var (not const) so tests can shorten it.
-var heartbeatInterval = 5 * time.Minute
+// defaultHeartbeatInterval is the fallback heartbeat interval used when
+// the config value is zero. Matches ProcessorConfig.HeartbeatInterval default.
+const defaultHeartbeatInterval = 5 * time.Minute
 
 func (p *Processor) runJob(ctx context.Context, params *jobExecutionParams) {
 	// Clean up in-flight entry on exit (first defer = last to run via LIFO),
@@ -323,35 +321,15 @@ func (p *Processor) handleJobError(ctx context.Context, params *jobExecutionPara
 		}
 
 	case errors.Is(err, errShutdown):
-		// SIGTERM received — re-enqueue so another worker can pick up the job.
-		// Use a detached context because ctx is already cancelled.
+		// SIGTERM received — leave the job in its current state for the orphan
+		// reconciler to handle. The reconciler detects non-terminal jobs that
+		// are not in the queue and have a stale (or missing) in-flight entry,
+		// then transitions them to a terminal state (failed, cancelled, etc.).
 		//
-		// Known limitation: there is no way at SIGTERM time to distinguish a container
-		// restart (emptyDir survives, startup recovery can upload partial output) from a
-		// pod deletion (emptyDir destroyed, startup recovery cannot help). Re-enqueueing
-		// is therefore unconditional, which introduces a known race: if this was a
-		// container restart, startup recovery and the worker that picks up the re-enqueued
-		// job compete — startup recovery may mark the job failed while another worker
-		// runs it fresh.
-		// This race is accepted as a known limit until orphan reconciliation is
-		// implemented. Once it is, re-enqueue should be removed here and pod-deletion
-		// recovery delegated to the reconciler. (TODO: orphan reconciliation task)
-		if params.task != nil {
-			bgCtx, bgSpan := uotel.DetachedContext(ctx, "re-enqueue")
-			defer bgSpan.End()
-			if enqErr := p.poller.enqueueOne(bgCtx, params.task); enqErr != nil {
-				logger.Error(enqErr, "Failed to re-enqueue the job to the queue")
-				// executeJob flushed partial output/error files to disk before returning
-				// errShutdown. Upload them so the user can retrieve whatever completed
-				// before SIGTERM, rather than losing those results silently.
-				if failErr := p.handleFailed(bgCtx, params.updater, params.jobItem, params.requestCounts, params.jobInfo); failErr != nil {
-					logger.Error(failErr, "Failed to mark job as failed after re-enqueue failure")
-				}
-			} else {
-				metrics.RecordJobProcessed(metrics.ResultReEnqueued, metrics.ReasonSystemError)
-				logger.V(logging.INFO).Info("Re-enqueued the job to the queue")
-			}
-		}
+		// The in-flight entry is cleaned up by the defer at the top of runJob.
+		// If the process is killed (SIGKILL) before that defer runs, the entry
+		// remains and becomes stale, which the reconciler also handles.
+		logger.V(logging.INFO).Info("SIGTERM received, leaving job for orphan reconciler")
 
 	default:
 		if failErr := p.handleFailed(ctx, params.updater, params.jobItem, params.requestCounts, params.jobInfo); failErr != nil {

internal/processor/worker/job_runner_test.go

@@ -732,11 +732,10 @@ func TestRunJob_FinalizeFailedOver_PreservesFileIDsAndDoesNotCallHandleFailed(t
 	}
 }
 
-// TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput verifies that when
-// errShutdown triggers a re-enqueue that fails, the fallback uploads partial output files
-// and preserves their file IDs in the DB. Before this fix, the fallback called handleFailed
-// with nil counts and empty file IDs, losing already-flushed results.
-func TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput(t *testing.T) {
+// TestHandleJobError_Shutdown_LeavesJobInProgress verifies that errShutdown
+// does NOT re-enqueue or call handleFailed. The job stays in_progress for
+// the orphan reconciler to detect and transition to a terminal state.
+func TestHandleJobError_Shutdown_LeavesJobInProgress(t *testing.T) {
 	ctx := testLoggerCtx(t)
 
 	cfg := config.NewConfig()
@@ -745,7 +744,7 @@ func TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput(t *testing.
 
 	dbClient := newMockBatchDBClient()
 	statusClient := mockdb.NewMockBatchStatusClient()
-	pqClient := &errPQClient{err: errors.New("queue unavailable")}
+	pqClient := mockdb.NewMockBatchPriorityQueueClient()
 
 	p := mustNewProcessor(t, cfg, &clientset.Clientset{
 		BatchDB:   dbClient,
@@ -758,7 +757,7 @@ func TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput(t *testing.
 	})
 	p.poller = NewPoller(pqClient, dbClient)
 
-	jobID := "job-shutdown-enqueue-fail"
+	jobID := "job-shutdown-no-reenqueue"
 	tenantID := "tenant__tenantA"
 
 	jobItem := &db.BatchItem{
@@ -774,19 +773,6 @@ func TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput(t *testing.
 	jobInfo := &batch_types.JobInfo{JobID: jobID, TenantID: tenantID}
 	counts := &openai.BatchRequestCounts{Total: 5, Completed: 3, Failed: 2}
 
-	jobDir, _ := p.jobRootDir(jobID, tenantID)
-	if err := os.MkdirAll(jobDir, 0o755); err != nil {
-		t.Fatalf("MkdirAll: %v", err)
-	}
-	outputPath := filepath.Join(jobDir, "output.jsonl")
-	if err := os.WriteFile(outputPath, []byte(`{"custom_id":"r1"}`+"\n"), 0o644); err != nil {
-		t.Fatalf("WriteFile output: %v", err)
-	}
-	errorPath := filepath.Join(jobDir, "error.jsonl")
-	if err := os.WriteFile(errorPath, []byte(`{"custom_id":"e1"}`+"\n"), 0o644); err != nil {
-		t.Fatalf("WriteFile error: %v", err)
-	}
-
 	updater := NewStatusUpdater(dbClient, statusClient, 86400)
 	params := &jobExecutionParams{
 		updater:       updater,
@@ -806,14 +792,7 @@ func TestHandleJobError_Shutdown_ReEnqueueFails_UploadsPartialOutput(t *testing.
 	if err := json.Unmarshal(items[0].Status, &got); err != nil {
 		t.Fatalf("unmarshal: %v", err)
 	}
-	if got.Status != openai.BatchStatusFailed {
-		t.Fatalf("status = %s, want failed", got.Status)
-	}
-	if got.RequestCounts.Total != 5 || got.RequestCounts.Completed != 3 || got.RequestCounts.Failed != 2 {
-		t.Fatalf("request_counts = %+v, want {5,3,2}", got.RequestCounts)
-	}
-	// handleFailed uploads partial results when jobInfo is non-nil; at least one file ID should be present.
-	if got.OutputFileID == nil && got.ErrorFileID == nil {
-		t.Fatal("expected at least one file ID to be preserved from partial upload")
+	if got.Status != openai.BatchStatusInProgress {
+		t.Fatalf("status = %s, want in_progress (left for reconciler)", got.Status)
 	}
 }