diff --git a/apps/api/internal/runner/queue_client_test.go b/apps/api/internal/runner/queue_client_test.go index b427fae..9d82c9d 100644 --- a/apps/api/internal/runner/queue_client_test.go +++ b/apps/api/internal/runner/queue_client_test.go @@ -254,14 +254,16 @@ func TestRecoverOrphanedAsyncRiverJobAndYieldStaleAdmission(t *testing.T) { orphaned := createQueuedTask("orphaned") protected := createQueuedTask("protected") yielded := createQueuedTask("yielded") + terminalOrphaned := createQueuedTask("terminal-orphaned") + terminalProtected := createQueuedTask("terminal-protected") staleWorkerID := "orphan-recovery-stale-" + suffix activeWorkerID := "orphan-recovery-active-" + suffix t.Cleanup(func() { cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second) defer cleanupCancel() _, _ = db.Pool().Exec(cleanupCtx, `DELETE FROM gateway_worker_instances WHERE instance_id = ANY($1::text[])`, []string{staleWorkerID, activeWorkerID}) - _, _ = db.Pool().Exec(cleanupCtx, `DELETE FROM river_job WHERE id = ANY($1::bigint[])`, []int64{orphaned.RiverJobID, protected.RiverJobID, yielded.RiverJobID}) - _, _ = db.Pool().Exec(cleanupCtx, `DELETE FROM gateway_tasks WHERE id = ANY($1::uuid[])`, []string{orphaned.ID, protected.ID, yielded.ID}) + _, _ = db.Pool().Exec(cleanupCtx, `DELETE FROM river_job WHERE id = ANY($1::bigint[])`, []int64{orphaned.RiverJobID, protected.RiverJobID, yielded.RiverJobID, terminalOrphaned.RiverJobID, terminalProtected.RiverJobID}) + _, _ = db.Pool().Exec(cleanupCtx, `DELETE FROM gateway_tasks WHERE id = ANY($1::uuid[])`, []string{orphaned.ID, protected.ID, yielded.ID, terminalOrphaned.ID, terminalProtected.ID}) }) if _, err := db.Pool().Exec(ctx, ` @@ -286,6 +288,8 @@ SET state = 'running', WHEN $1::bigint THEN $3 WHEN $2::bigint THEN $4 WHEN $5::bigint THEN $4 + WHEN $7::bigint THEN $3 + WHEN $8::bigint THEN $4 END ]::text[] WHERE id = ANY($6::bigint[])`, @@ -294,10 +298,43 @@ WHERE id = ANY($6::bigint[])`, staleWorkerID+"-exec-1-test", activeWorkerID+"-exec-1-test", yielded.RiverJobID, - []int64{orphaned.RiverJobID, protected.RiverJobID, yielded.RiverJobID}, + []int64{orphaned.RiverJobID, protected.RiverJobID, yielded.RiverJobID, terminalOrphaned.RiverJobID, terminalProtected.RiverJobID}, + terminalOrphaned.RiverJobID, + terminalProtected.RiverJobID, ); err != nil { t.Fatalf("mark River jobs running: %v", err) } + if _, err := db.Pool().Exec(ctx, ` +UPDATE gateway_tasks +SET status = 'failed', + error_code = 'test_terminal', + finished_at = now(), + execution_token = NULL, + execution_lease_expires_at = NULL +WHERE id = ANY($1::uuid[])`, []string{terminalOrphaned.ID, terminalProtected.ID}); err != nil { + t.Fatalf("mark terminal River tasks failed: %v", err) + } + finalized, err := db.FinalizeOrphanedTerminalAsyncRiverJobs(ctx, 30*time.Second, 10) + if err != nil { + t.Fatalf("finalize orphaned terminal River jobs: %v", err) + } + if finalized != 1 { + t.Fatalf("finalized terminal jobs=%d, want 1", finalized) + } + var terminalOrphanedState, terminalProtectedState string + if err := db.Pool().QueryRow(ctx, `SELECT state::text FROM river_job WHERE id = $1`, terminalOrphaned.RiverJobID).Scan(&terminalOrphanedState); err != nil { + t.Fatalf("read terminal orphaned River state: %v", err) + } + if err := db.Pool().QueryRow(ctx, `SELECT state::text FROM river_job WHERE id = $1`, terminalProtected.RiverJobID).Scan(&terminalProtectedState); err != nil { + t.Fatalf("read terminal protected River state: %v", err) + } + if terminalOrphanedState != "completed" || terminalProtectedState != "running" { + t.Fatalf("terminal River states orphaned=%s protected=%s, want completed/running", terminalOrphanedState, terminalProtectedState) + } + finalized, err = db.FinalizeOrphanedTerminalAsyncRiverJobs(ctx, 30*time.Second, 10) + if err != nil || finalized != 0 { + t.Fatalf("second terminal River finalization count=%d err=%v, want idempotent zero", finalized, err) + } var platformID, platformModelID string if err := db.Pool().QueryRow(ctx, ` SELECT platform.id::text, model.id::text diff --git a/apps/api/internal/runner/queue_worker.go b/apps/api/internal/runner/queue_worker.go index 400fff0..05b048b 100644 --- a/apps/api/internal/runner/queue_worker.go +++ b/apps/api/internal/runner/queue_worker.go @@ -694,6 +694,20 @@ func (s *Service) recoverOrphanedAsyncRiverJobs(ctx context.Context) { "cleanedTaskAdmissions", runtimeRecovery.CleanedTaskAdmissions, ) } + finalizedTerminalJobs, err := s.coordinationStore.FinalizeOrphanedTerminalAsyncRiverJobs( + ctx, + orphanedRiverJobWorkerStaleAfter, + orphanedRiverJobRecoveryBatchSize, + ) + if err != nil { + if ctx.Err() == nil { + s.logger.Warn("finalize orphaned terminal river jobs failed", "error", err) + } + return + } + if finalizedTerminalJobs > 0 { + s.logger.Warn("orphaned terminal river jobs finalized", "count", finalizedTerminalJobs) + } yieldedAdmissions, err := s.coordinationStore.YieldStaleAsyncTaskAdmissions( ctx, orphanedRiverJobWorkerStaleAfter, diff --git a/apps/api/internal/store/worker_registry.go b/apps/api/internal/store/worker_registry.go index fc399d2..0de79ef 100644 --- a/apps/api/internal/store/worker_registry.go +++ b/apps/api/internal/store/worker_registry.go @@ -629,3 +629,82 @@ FROM recovered_jobs`, workerStaleAfter.String(), limit).Scan(&recovered); err != } return recovered, nil } + +// FinalizeOrphanedTerminalAsyncRiverJobs closes River jobs whose gateway task +// is already terminal after the owning Worker disappeared. River deliberately +// uses a one-hour rescue horizon for long provider polls, so relying on its +// generic rescue loop would leave rollout-interrupted terminal jobs reported +// as running long after their execution lease and business resources were +// released. +func (s *Store) FinalizeOrphanedTerminalAsyncRiverJobs( + ctx context.Context, + workerStaleAfter time.Duration, + limit int, +) (int64, error) { + if workerStaleAfter < workerHeartbeatStaleAfter { + workerStaleAfter = workerHeartbeatStaleAfter + } + if limit <= 0 || limit > 1000 { + limit = 100 + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, err + } + defer rollbackTransaction(tx) + var finalized int64 + if err := tx.QueryRow(ctx, ` +WITH orphaned AS MATERIALIZED ( + SELECT job.id AS job_id, + task.id AS task_id, + job.attempt + FROM river_job job + JOIN gateway_tasks task ON task.river_job_id = job.id + WHERE job.queue = 'gateway_tasks' + AND job.kind = 'gateway_task_run' + AND job.state = 'running' + AND job.attempted_at <= now() - $1::interval + AND task.async_mode = true + AND task.status IN ('succeeded', 'failed', 'cancelled') + AND task.execution_token IS NULL + AND (task.execution_lease_expires_at IS NULL OR task.execution_lease_expires_at <= now()) + AND NOT EXISTS ( + SELECT 1 + FROM gateway_worker_instances worker + WHERE worker.status = 'active' + AND worker.heartbeat_at > now() - $2::interval + AND EXISTS ( + SELECT 1 + FROM unnest(job.attempted_by) attempted_owner + WHERE attempted_owner LIKE worker.instance_id || '-exec-%' + ) + ) + ORDER BY job.attempted_at ASC, job.id ASC + LIMIT $3 + FOR UPDATE OF job SKIP LOCKED +), finalized_jobs AS ( + UPDATE river_job job + SET errors = array_append( + COALESCE(job.errors, ARRAY[]::jsonb[]), + jsonb_build_object( + 'at', now(), + 'attempt', orphaned.attempt, + 'error', 'Terminal gateway job finalized after Worker owner disappeared', + 'trace', '' + ) + ), + finalized_at = now(), + state = 'completed' + FROM orphaned + WHERE job.id = orphaned.job_id + RETURNING job.id +) +SELECT count(*)::bigint +FROM finalized_jobs`, workerStaleAfter.String(), workerHeartbeatStaleAfter.String(), limit).Scan(&finalized); err != nil { + return 0, err + } + if err := tx.Commit(ctx); err != nil { + return 0, err + } + return finalized, nil +}