From 9bdf6eed5b1878779c249e75793b4b242ec22e75 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 09:57:33 -0700 Subject: [PATCH 1/7] feat(jobs): tie a one-off container to its operation, and record an interruption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects that compose into one incident: a sealed manual job run whose client goes away leaves a container the host cannot match to anything, an operation that is INCOMPLETE in `ob audit` forever, and an application lock released out from under a container that is still changing data. `compose run` names nothing and inherits no operation identity, so the host had no way to tell which run a one-off container belonged to. Every refusal and every reconciliation has to key on that, so it comes first: generated and hook-supplied compose runs both now carry ob.operation and ob.epoch. The terminal record was appended on the caller's context, which is precisely the context that is gone when it matters. `ob exec` already writes its own on a bounded background context; the job path now does the same and marks the record `interrupted`, because Append redacts Detail on a failure and the reason has to ride on ErrorCode. `ob audit` reports that as its own outcome, ahead of the failure arm the same record matches: the client went away and the outcome is unknown, which is a different thing to tell an operator than "it failed". ReleaseLock runs on its own background context, so on Ctrl-C it succeeded while the journal append on the cancelled context did not — ownership dropped immediately and silently, over a live container, before the TTL could matter. The lock is now held when an interrupted run leaves this operation's container running, and the operator is told what holds it and how to look. An unreadable answer counts as running: keeping the lock over a container that has exited costs one `--break-lock`, releasing it over one that has not costs concurrent writers. This does not make execution durable. A job whose client is killed outright still records nothing, and reconciling what a container did after the fact is the next step. It makes every new run reconcilable, which nothing was before. Refs #179. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/audit.go | 6 ++ internal/engine/audit_test.go | 26 +++++++++ internal/engine/gate.go | 52 +++++++++++++++-- internal/engine/gate_test.go | 43 ++++++++++++-- internal/engine/job.go | 66 +++++++++++++++++++++- internal/engine/schedule_execution_test.go | 2 +- 6 files changed, 182 insertions(+), 13 deletions(-) diff --git a/internal/engine/audit.go b/internal/engine/audit.go index 104ec005..8dbef947 100644 --- a/internal/engine/audit.go +++ b/internal/engine/audit.go @@ -170,6 +170,12 @@ func auditRows(recs []journal.Record) []auditRow { row.action = auditAction(r.Phase) } switch { + // Ahead of the failure arm, which this record also matches: an + // interrupted run is not a failure of the job. The client went away + // and the outcome is unknown, which is a different thing to tell an + // operator than "it failed" or "it never finished". + case r.ErrorCode == "interrupted": + row.outcome = "interrupted" case r.Event == "abort": row.outcome = "aborted" case r.Status == "fail": diff --git a/internal/engine/audit_test.go b/internal/engine/audit_test.go index 2925fa0a..5a61ea54 100644 --- a/internal/engine/audit_test.go +++ b/internal/engine/audit_test.go @@ -129,3 +129,29 @@ func TestAuditReportsFailedJobRun(t *testing.T) { t.Fatalf("failed job run audit = %+v", records) } } + +// An interrupted run is neither a job failure nor incomplete-forever: the +// client went away and the outcome is unknown. The record carries Status fail, +// so it also matches the failure arm and the ordering is what distinguishes it. +func TestAuditDistinguishesAnInterruptedJobRun(t *testing.T) { + f := &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) { + switch { + case strings.Contains(cmd, "ls -1"): + return transport.Result{Stdout: "job-3.jsonl\n"}, true + case strings.Contains(cmd, "job-3.jsonl"): + return transport.Result{Stdout: journalLines( + journal.Record{DeployID: "job-3", Phase: "job", Event: "start", Status: "ok", OperationKind: "job_run", Service: "catalog-refresh", Operator: "v@mac", TS: "t1"}, + journal.Record{DeployID: "job-3", Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", OperationKind: "job_run", Service: "catalog-refresh"}, + )}, true + } + return transport.Result{}, false + }} + e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) + records, err := e.AuditSnapshot(context.Background(), 10) + if err != nil { + t.Fatal(err) + } + if len(records) != 1 || records[0].Outcome != "interrupted" || records[0].Service != "catalog-refresh" { + t.Fatalf("interrupted job audit = %+v", records) + } +} diff --git a/internal/engine/gate.go b/internal/engine/gate.go index 124c26b5..0e163612 100644 --- a/internal/engine/gate.go +++ b/internal/engine/gate.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strconv" "strings" "github.com/labstack/onebox/internal/app" @@ -73,7 +74,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s return fmt.Errorf("journal %s intent: %w", key, err) } st := e.ui.Step("job "+job, true) - safe, detail, err := e.runOneJob(ctx, job, remoteDir, remoteCompose) + safe, detail, err := e.runOneJob(ctx, jw.DeployID, jw.Epoch, job, remoteDir, remoteCompose) if err == nil { e.logf("job %s: %s", job, detail) } @@ -113,7 +114,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s // runOneJob runs a single gate step and reports whether it declared itself // rollback-safe (changed=false). Returns (safe, detail, err). -func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose string) (bool, string, error) { +func (e *Engine) runOneJob(ctx context.Context, operationID string, epoch int, job, remoteDir, remoteCompose string) (bool, string, error) { safeByDeclaration := e.jobDataEffect(job) == app.DataEffectNone if !safeByDeclaration { res, err := e.mutate(ctx, invalidateExecutionCommand(e.names().AppDir())) @@ -128,7 +129,7 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st resultFile := resultDir + "/result" const containerResultFile = "/run/onebox/job-result" containerized := true - runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" + + runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" + jobRunLabels(operationID, epoch) + " -e ONEBOX_RESULT_FILE=" + containerResultFile + " -v " + q(resultFile+":"+containerResultFile+":rw") + " " + job if h, ok := e.Spec.Hooks[job]; ok && h.Run != "" { @@ -147,6 +148,11 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st var injected bool runCmd, injected = injectComposeJobResult(runCmd, resultFile, containerResultFile) containerized = injected + // A hook that runs its own compose command produces a container this + // operation owns just as much as the generated one, so it carries the + // same identity. A hook that is not a compose run has no container to + // label, and is already reported as unresolvable above. + runCmd, _ = injectComposeJobLabels(runCmd, operationID, epoch) } e.ui.Cmd("job", runCmd) // verbose only — the plan lists it resultMode := "600" @@ -200,7 +206,43 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st return !evidence.Changed, jobResultDetail(evidence), nil } +// jobRunLabels ties a one-off container back to the operation that started it. +// `compose run` names nothing and inherits no operation identity, so without +// these an interrupted run leaves a container on the host that nothing can +// match to a journal — every refusal and every reconciliation keys on them. +func jobRunLabels(operationID string, epoch int) string { + if operationID == "" { + return "" + } + return " --label " + q(JobOperationLabel+"="+operationID) + + " --label " + q(JobEpochLabel+"="+strconv.Itoa(epoch)) +} + +const ( + // JobOperationLabel carries the operation id of the run that created a + // one-off job container. + JobOperationLabel = "ob.operation" + // JobEpochLabel carries the lock epoch that run held. + JobEpochLabel = "ob.epoch" +) + +func injectComposeJobLabels(command, operationID string, epoch int) (string, bool) { + labels := jobRunLabels(operationID, epoch) + if labels == "" { + return command, false + } + return injectComposeRunFlags(command, labels+" ") +} + func injectComposeJobResult(command, hostResultFile, containerResultFile string) (string, bool) { + return injectComposeRunFlags(command, + " -e "+"ONEBOX_RESULT_FILE="+containerResultFile+ + " -v "+q(hostResultFile+":"+containerResultFile+":rw")+" ") +} + +// injectComposeRunFlags splices flags into a hook's own `docker compose run`. +// Anything that is not a compose run is left alone and reported as such. +func injectComposeRunFlags(command, flags string) (string, bool) { runIndex := strings.Index(command, " run ") if runIndex < 0 { return command, false @@ -209,9 +251,7 @@ func injectComposeJobResult(command, hostResultFile, containerResultFile string) if !strings.Contains(prefix, "docker compose") && !strings.Contains(prefix, "docker-compose") { return command, false } - flags := " run -e ONEBOX_RESULT_FILE=" + containerResultFile + - " -v " + q(hostResultFile+":"+containerResultFile+":rw") + " " - return prefix + flags + command[runIndex+len(" run "):], true + return prefix + " run" + flags + command[runIndex+len(" run "):], true } func (e *Engine) unknownJobResult(job, reason string) (bool, string, error) { diff --git a/internal/engine/gate_test.go b/internal/engine/gate_test.go index 0dc78a08..b295d342 100644 --- a/internal/engine/gate_test.go +++ b/internal/engine/gate_test.go @@ -142,7 +142,7 @@ func TestJobAutoRunsWithoutHook(t *testing.T) { t.Fatalf("deploy: %v", err) } seq := strings.Join(f.Commands, "\n") - if !strings.Contains(seq, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") { + if !strings.Contains(seq, "run --rm --no-deps") || !strings.Contains(seq, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") { t.Fatalf("a job without a hook must auto-run compose run:\n%s", seq) } // gate protocol still applies to the auto-run job. @@ -266,7 +266,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) { t.Run("no result declaration", func(t *testing.T) { f := happyFake() e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) - safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml") + safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml") if err != nil { t.Fatalf("run job: %v", err) } @@ -282,7 +282,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) { e := New(cfg, testProject(t), happyFake(), Options{ Out: &bytes.Buffer{}, Sleep: noSleep, LocalDir: t.TempDir(), }) - safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml") + safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml") if err != nil { t.Fatalf("run local job: %v", err) } @@ -419,7 +419,7 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) { mount := strings.Index(c, "-v '"+resultFile+":/run/onebox/job-result:rw'") sealedFile := strings.Index(c, "chmod 600 '"+resultFile+"'") if strings.Contains(c, "rm -rf '"+resultDir+"'") && - strings.Contains(c, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") && + strings.Contains(c, "run --rm --no-deps") && strings.Contains(c, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") && privateDir >= 0 && privateDir < writableFile && writableFile < mount && mount < sealedFile { found = true } @@ -428,3 +428,38 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) { t.Fatalf("migrate container must receive a privately staged, writable, subsequently sealed result file:\n%s", strings.Join(f.Commands, "\n")) } } + +// Without an operation label nothing on the host ties a running one-off +// container back to the journal that started it, so an interrupted run cannot +// be refused or reconciled — only guessed at. +func TestJobContainerCarriesItsOperationIdentity(t *testing.T) { + f := happyFake() + e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) + if _, _, err := e.runOneJob(context.Background(), "20260909-053225-abc-job_run-deadbeef", 7, "migrate", "/remote", "/remote/compose.yaml"); err != nil { + t.Fatal(err) + } + seq := strings.Join(f.Commands, "\n") + for _, want := range []string{ + "--label 'ob.operation=20260909-053225-abc-job_run-deadbeef'", + "--label 'ob.epoch=7'", + } { + if !strings.Contains(seq, want) { + t.Fatalf("job container missing %s:\n%s", want, seq) + } + } +} + +func TestInjectComposeJobLabelsOnlyTouchesAComposeRun(t *testing.T) { + got, ok := injectComposeJobLabels("docker compose -f x.yml run --rm migrate", "op-1", 2) + if !ok || !strings.Contains(got, "--label 'ob.operation=op-1'") || !strings.Contains(got, "--label 'ob.epoch=2'") { + t.Fatalf("compose run = %q ok=%v", got, ok) + } + // A hook that is not a compose run has no container to label. + if got, ok := injectComposeJobLabels("/usr/local/bin/migrate.sh", "op-1", 2); ok || got != "/usr/local/bin/migrate.sh" { + t.Fatalf("non-compose hook = %q ok=%v", got, ok) + } + // No operation identity, nothing to add. + if got, ok := injectComposeJobLabels("docker compose run --rm migrate", "", 0); ok || got != "docker compose run --rm migrate" { + t.Fatalf("empty operation = %q ok=%v", got, ok) + } +} diff --git a/internal/engine/job.go b/internal/engine/job.go index 0458a1af..fb6b860c 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "strings" + "time" "github.com/labstack/onebox/internal/app" "github.com/labstack/onebox/internal/journal" @@ -43,7 +44,22 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) if err != nil { return operationID, nil, err } - defer e.ReleaseLock(ctx) + // Released unless an interrupted run left this operation's container alive. + // ReleaseLock runs on its own background context, so on Ctrl-C it succeeds + // while the terminal journal append — which uses the cancelled one — does + // not: ownership would be dropped, immediately and silently, over a + // container still changing data. + holdLockForLiveContainer := false + defer func() { + if holdLockForLiveContainer { + e.warnf("operation %s was interrupted while its container is still running; "+ + "keeping the application lock so nothing else mutates alongside it. "+ + "Inspect with `docker ps --filter label=%s=%s`; the lock expires on its own after %s", + operationID, JobOperationLabel, operationID, e.lockTTL()) + return + } + e.ReleaseLock(ctx) + }() if err := e.WriteFence(ctx, operationID, epoch); err != nil { return operationID, nil, err } @@ -93,7 +109,20 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) record.Status = "fail" record.Detail = runErr.Error() } - if journalErr := writer.Append(ctx, record); journalErr != nil { + // A cancelled context is exactly when the terminal record matters most, + // and exactly when appending on that context cannot work. `ob exec` + // already writes its own on a bounded background context; without the + // same here an interrupted job stays INCOMPLETE in `ob audit` forever, + // with no record that it was ever interrupted. Append redacts Detail on + // a failure, so the reason has to ride on ErrorCode. + journalContext := ctx + if interruptedRun(ctx, runErr) { + record.Status, record.ErrorCode = "fail", "interrupted" + var cancel context.CancelFunc + journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout) + defer cancel() + } + if journalErr := writer.Append(journalContext, record); journalErr != nil { return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr)) } return runErr @@ -121,6 +150,11 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) e.gateOpen = true e.rollbackCovered = true runErr := e.runJobPhase(ctx, writer, nil, remoteDir, remoteCompose, "job", []string{job}) + if interruptedRun(ctx, runErr) { + // Cancelling the client kills at most the wrapper shell; the container + // belongs to the daemon and keeps running. + holdLockForLiveContainer = e.jobContainerRunning(operationID) + } var result *journal.JobResultEvidence if evidence, ok := e.jobResults[job]; ok { resultCopy := evidence @@ -128,3 +162,31 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) } return operationID, result, finish(runErr) } + +const journalCleanupTimeout = 5 * time.Second + +// interruptedRun reports a run that ended because the client went away rather +// than because the job finished. +func interruptedRun(ctx context.Context, runErr error) bool { + return ctx.Err() != nil || + errors.Is(runErr, context.Canceled) || + errors.Is(runErr, context.DeadlineExceeded) +} + +// jobContainerRunning answers whether this operation's one-off container is +// still alive, on a context of its own because the caller's is already gone. +// An unreadable answer is reported as running: keeping the lock over a +// container that has in fact exited costs an operator one `--break-lock`, while +// releasing it over one that has not costs them concurrent writers. +func (e *Engine) jobContainerRunning(operationID string) bool { + ctx, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout) + defer cancel() + res, err := e.T.Run(ctx, "docker ps -q --filter label="+q(JobOperationLabel+"="+operationID)) + if err != nil { + return true + } + if res.ExitCode != 0 { + return true + } + return strings.TrimSpace(res.Stdout) != "" +} diff --git a/internal/engine/schedule_execution_test.go b/internal/engine/schedule_execution_test.go index c4df1f9f..33bbc4cf 100644 --- a/internal/engine/schedule_execution_test.go +++ b/internal/engine/schedule_execution_test.go @@ -150,7 +150,7 @@ func TestDurableCompatibilityInvalidationMustSucceedBeforeDataChangingJob(t *tes return base(command) } e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) - _, _, err := e.runOneJob(context.Background(), "change", "/release", "/release/compose.yaml") + _, _, err := e.runOneJob(context.Background(), "op-1", 1, "change", "/release", "/release/compose.yaml") if !invalidated || err == nil || !strings.Contains(err.Error(), "invalidate durable execution compatibility") { t.Fatalf("failed invalidation did not stop data-changing job: invalidated=%t err=%v", invalidated, err) } From 47a3f56ae898007e0ae2995eb4d345ffb2932aa9 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:01:46 -0700 Subject: [PATCH 2/7] fix(jobs): enumerate the interrupted error code The code reaches an operator through `ob audit` but was not in the registry, so the generated error reference could not document it. The enumeration test missed it because it matches the struct-literal form and the assignment was a tuple; written as separate statements it is now covered by that test too. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job.go | 5 ++++- internal/onebox/operation_errors.go | 7 +++++++ site/src/content/docs/reference/errors.mdx | 1 + 3 files changed, 12 insertions(+), 1 deletion(-) diff --git a/internal/engine/job.go b/internal/engine/job.go index fb6b860c..d81e962a 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -117,7 +117,10 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) // a failure, so the reason has to ride on ErrorCode. journalContext := ctx if interruptedRun(ctx, runErr) { - record.Status, record.ErrorCode = "fail", "interrupted" + record = journal.Record{ + Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", + OperationKind: "job_run", Service: job, + } var cancel context.CancelFunc journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout) defer cancel() diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go index 68b45614..8ea77335 100644 --- a/internal/onebox/operation_errors.go +++ b/internal/onebox/operation_errors.go @@ -119,6 +119,13 @@ var operationFailureDefinitions = map[string]OperationFailure{ Message: "this host is owned by a different Onebox application, and one host has one owner", Command: "ob preflight --output json", }, + "interrupted": { + // Not a failure of the job and not an incomplete operation: the client + // went away, and what the job did is unknown unless it journaled a + // result first. Reconciliation on the next operation writes this. + Message: "the operation's client went away before its outcome could be recorded", + Command: "ob audit --output json", + }, "job_plan_failed": { Message: "the one-shot job plan could not be produced", Command: "ob job plan --output json", diff --git a/site/src/content/docs/reference/errors.mdx b/site/src/content/docs/reference/errors.mdx index a63e5d6c..03012bc2 100644 --- a/site/src/content/docs/reference/errors.mdx +++ b/site/src/content/docs/reference/errors.mdx @@ -136,6 +136,7 @@ step to complete rather than a line to run verbatim. | `finalize_refused` | the release cannot be finalized because the recorded activation evidence disagrees with the live host | diagnostic | `ob status --output json` | | `host_environment_mismatch` | this host is claimed by a different environment of the same application, which would share its container and volume names | diagnostic | `ob preflight --output json` | | `host_owner_mismatch` | this host is owned by a different Onebox application, and one host has one owner | diagnostic | `ob preflight --output json` | +| `interrupted` | the operation's client went away before its outcome could be recorded | diagnostic | `ob audit --output json` | | `job_plan_failed` | the one-shot job plan could not be produced | next | `ob job plan --output json` | | `logs_failed` | log retrieval failed | diagnostic | `ob status --output json` | | `manifest_invalid` | a release manifest is not valid closed JSON for its schema | diagnostic | `ob status --output json` | From c2427803380056d8b2d1188523126fed2ff680e4 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:30:22 -0700 Subject: [PATCH 3/7] fix(jobs): record a job that finished just before the client went away MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The terminal append switched to a background context only when the run itself was classified interrupted, which conflated two independent questions. WHERE to append is decided by the caller's context: once it is cancelled it cannot carry the write, whatever the run did. WHAT to record is decided by what the run did: a job that completed cleanly a moment before Ctrl-C produced its own outcome and is not interrupted. Conflating them meant a successful job, cancelled in the window between its result and its finish record, was written down as a failure — or, if the classification did not catch the cancellation, was not written down at all. Also corrects the registry comment for the code, which described reconciliation as its only writer. The interrupted run writes it itself whenever it still can. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job.go | 13 ++++++++++--- internal/onebox/operation_errors.go | 3 ++- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/internal/engine/job.go b/internal/engine/job.go index d81e962a..92be67f5 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -115,15 +115,22 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) // same here an interrupted job stays INCOMPLETE in `ob audit` forever, // with no record that it was ever interrupted. Append redacts Detail on // a failure, so the reason has to ride on ErrorCode. + // Two independent questions. WHERE to append: a cancelled caller context + // cannot carry the write, whatever the run did, so a job that finished + // cleanly a moment before Ctrl-C still records its success. WHAT to + // record: only a run that ended because the client went away is + // interrupted — an outcome the job itself produced is its own. journalContext := ctx + if ctx.Err() != nil { + var cancel context.CancelFunc + journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout) + defer cancel() + } if interruptedRun(ctx, runErr) { record = journal.Record{ Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", OperationKind: "job_run", Service: job, } - var cancel context.CancelFunc - journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout) - defer cancel() } if journalErr := writer.Append(journalContext, record); journalErr != nil { return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr)) diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go index 8ea77335..b3e7d416 100644 --- a/internal/onebox/operation_errors.go +++ b/internal/onebox/operation_errors.go @@ -122,7 +122,8 @@ var operationFailureDefinitions = map[string]OperationFailure{ "interrupted": { // Not a failure of the job and not an incomplete operation: the client // went away, and what the job did is unknown unless it journaled a - // result first. Reconciliation on the next operation writes this. + // result first. Written by the interrupted run itself when it still can, + // and otherwise by the next operation once it finds the run unfinished. Message: "the operation's client went away before its outcome could be recorded", Command: "ob audit --output json", }, From dce66d307912847f6f08600e7759371bffc15675 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:39:36 -0700 Subject: [PATCH 4/7] fix(jobs): classify the run, not the client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit split where the terminal record is appended from what it says, but left the classification itself keyed on the caller's context. A job that finished cleanly and was then cancelled in the window before its record was still written down as interrupted — the very case that split was meant to fix. A run that produced no error finished, whatever became of the client afterwards. Only once the run failed does a gone context mean the client is why, and that still counts even when the transport reports something other than a cancelled context, which it usually does. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job.go | 9 +++++++++ internal/engine/job_test.go | 21 +++++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/internal/engine/job.go b/internal/engine/job.go index 92be67f5..e2e065d6 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -177,7 +177,16 @@ const journalCleanupTimeout = 5 * time.Second // interruptedRun reports a run that ended because the client went away rather // than because the job finished. +// +// A run that produced no error finished, whatever became of the client +// afterwards — its outcome is its own and must be recorded as such. Only once +// the run failed does a gone context mean the client is why. The transport does +// not always surface a cancelled context as context.Canceled, so a failure +// under a cancelled context counts even when the error says something else. func interruptedRun(ctx context.Context, runErr error) bool { + if runErr == nil { + return false + } return ctx.Err() != nil || errors.Is(runErr, context.Canceled) || errors.Is(runErr, context.DeadlineExceeded) diff --git a/internal/engine/job_test.go b/internal/engine/job_test.go index 67366f89..88cfd4bc 100644 --- a/internal/engine/job_test.go +++ b/internal/engine/job_test.go @@ -3,6 +3,7 @@ package engine import ( "bytes" "context" + "errors" "strings" "testing" @@ -151,3 +152,23 @@ func TestRunJobUsesPlanIdentityAndJournalsAuthorization(t *testing.T) { } } } + +// A job that finished cleanly is not interrupted, whatever became of the client +// in the window before its terminal record. Classification follows the run; +// only the choice of append context follows the caller's context. +func TestInterruptedRunClassifiesTheRunNotTheClient(t *testing.T) { + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + if interruptedRun(cancelled, nil) { + t.Fatal("a run that produced no error must never be recorded interrupted") + } + if !interruptedRun(cancelled, errors.New("ssh: session closed")) { + t.Fatal("a failed run under a cancelled context is interrupted, whatever the transport called it") + } + if !interruptedRun(context.Background(), context.Canceled) { + t.Fatal("a cancellation surfaced by the run itself is interrupted") + } + if interruptedRun(context.Background(), errors.New("migrate: exit 1")) { + t.Fatal("a job that failed on its own terms is not interrupted") + } +} From fa6804a5550353a2c5ea3ba1bfc01727ab48c289 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:55:16 -0700 Subject: [PATCH 5/7] fix(jobs): retry the terminal append when cancellation lands mid-write The append switched to a context of its own only when the caller's was already gone. Cancellation during the write was left with a failed append and no second attempt, which is the same permanently incomplete operation by a narrower path. One retry closes it. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job.go | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/internal/engine/job.go b/internal/engine/job.go index e2e065d6..95ad7c54 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -132,7 +132,18 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) OperationKind: "job_run", Service: job, } } - if journalErr := writer.Append(journalContext, record); journalErr != nil { + journalErr := writer.Append(journalContext, record) + if journalErr != nil && journalContext == ctx { + // Cancellation can land during the write as easily as before it, and + // the check above only sees the context that was already gone. One + // retry on a context of our own is the difference between an + // operation that records its outcome and one that is incomplete + // forever. + retryContext, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout) + defer cancel() + journalErr = writer.Append(retryContext, record) + } + if journalErr != nil { return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr)) } return runErr From de72b60d0d2aaa8279ddaa373034f0f1df82cad5 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 11:04:49 -0700 Subject: [PATCH 6/7] docs(jobs): describe only what this change does for the interrupted code The registry comment named a reconciler that does not exist here, so a reader of this change alone would look for behaviour it does not contain. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/onebox/operation_errors.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go index b3e7d416..ce7ba7eb 100644 --- a/internal/onebox/operation_errors.go +++ b/internal/onebox/operation_errors.go @@ -122,8 +122,8 @@ var operationFailureDefinitions = map[string]OperationFailure{ "interrupted": { // Not a failure of the job and not an incomplete operation: the client // went away, and what the job did is unknown unless it journaled a - // result first. Written by the interrupted run itself when it still can, - // and otherwise by the next operation once it finds the run unfinished. + // result first. Written by the interrupted run itself, on a context of + // its own, whenever it can still reach the host. Message: "the operation's client went away before its outcome could be recorded", Command: "ob audit --output json", }, From 2e3f9ef01ff8bd982343adb32411528e0f73a07b Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 11:31:52 -0700 Subject: [PATCH 7/7] fix(jobs): retry the terminal append only when the context died mid-write MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retry fired on any append failure. A write that failed for another reason — a full disk, a refused write — may still have landed on the host, and appending a second terminal record over it is worse than reporting the first failure. It now retries only when the caller's context died during the write, which is the case it was added for. Also corrects the label comment for a hook that is not a compose run: it gets no label because there is no container to put one on, and its result is unresolvable for the same reason rather than by a check above. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/gate.go | 6 ++++-- internal/engine/job.go | 11 ++++++----- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/internal/engine/gate.go b/internal/engine/gate.go index 0e163612..a310a2df 100644 --- a/internal/engine/gate.go +++ b/internal/engine/gate.go @@ -150,8 +150,10 @@ func (e *Engine) runOneJob(ctx context.Context, operationID string, epoch int, j containerized = injected // A hook that runs its own compose command produces a container this // operation owns just as much as the generated one, so it carries the - // same identity. A hook that is not a compose run has no container to - // label, and is already reported as unresolvable above. + // same identity. A hook that is not a compose run gets no label because + // there is no container to put one on; that hook is also not + // containerized, so its result is unresolvable and reported as such + // below. runCmd, _ = injectComposeJobLabels(runCmd, operationID, epoch) } e.ui.Cmd("job", runCmd) // verbose only — the plan lists it diff --git a/internal/engine/job.go b/internal/engine/job.go index 95ad7c54..8ffd5dee 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -133,12 +133,13 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) } } journalErr := writer.Append(journalContext, record) - if journalErr != nil && journalContext == ctx { + if journalErr != nil && journalContext == ctx && ctx.Err() != nil { // Cancellation can land during the write as easily as before it, and - // the check above only sees the context that was already gone. One - // retry on a context of our own is the difference between an - // operation that records its outcome and one that is incomplete - // forever. + // the check above only sees a context that was already gone. Retried + // only when the context died in the meantime: any other failure — + // a full disk, a refused write — may have landed on the host after + // reporting an error, and appending a second terminal record is + // worse than reporting the first failure. retryContext, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout) defer cancel() journalErr = writer.Append(retryContext, record)