fix(queue): 回收终态任务的孤儿 River Job

原因：Worker 滚动切换后，执行租约恢复会终态化提交结果不确定的任务，但失活 Worker 持有的 River Job 仍需等待一小时通用救援窗口。\n\n影响：仅在任务已终态、执行租约清空、River owner 无活跃 Worker 且超过 45 秒时，将孤儿 Job 幂等标记为 completed；活跃 owner 不受影响。\n\n验证：PostgreSQL 集成测试覆盖失活 owner、活跃 owner 与重复回收；API 全量 go test 和 go vet 通过。
2026-08-04 18:51:47 +08:00
parent 9bcfec890b
commit 484b05d005
3 changed files with 133 additions and 3 deletions
@@ -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
+}