Add recovery for orphaned jobs

This commit is contained in:
2026-08-22 13:42:42 +01:00
parent 3ccefc254a
commit ec448157d9
9 changed files with 409 additions and 5 deletions
+20 -4
View File
@@ -357,10 +357,26 @@ func dispatchAgentMessage(ctx context.Context, c *Conn, hostID string, env api.E
}
case api.MsgCommandResult:
// TODO(P2): persist command.result acks for "did the agent
// accept the dispatch?" forensics. Currently the job lifecycle
// (job.started → job.finished) is sufficient signal.
slog.Debug("ws msg not yet handled", "type", env.Type, "host_id", hostID)
var p api.CommandResultPayload
if err := env.UnmarshalPayload(&p); err != nil {
slog.Warn("ws: decode command result", "host_id", hostID, "err", err)
break
}
// A cancel for a job unknown to the current agent process can never
// produce job.finished. Terminalise the orphan server-side.
if !p.Accepted && p.Error == "job_not_found" && p.JobID != "" {
job, err := deps.Store.GetJob(ctx, p.JobID)
if err != nil || job.HostID != hostID {
slog.Warn("ws: reject orphan result for foreign or missing job", "job_id", p.JobID, "host_id", hostID)
break
}
if job.Status == string(api.JobQueued) || job.Status == string(api.JobRunning) {
if err := deps.Store.MarkJobFinished(ctx, p.JobID, string(api.JobCancelled), -1, nil,
"agent reported job not found during cancellation", time.Now().UTC()); err != nil {
slog.Warn("ws: terminalise orphaned job", "job_id", p.JobID, "err", err)
}
}
}
case api.MsgTreeListResult:
// Reply to a synchronous tree.list RPC. Route to the waiter