From 5a1f1ab8092f62f8f6f02a782427dae78c58ebd5 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:04:45 -0700 Subject: [PATCH 1/7] feat(jobs): reconcile job runs whose client never recorded an outcome MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A sealed job run that lost its client left a journal with a start and no terminal record. Nothing ever went back for it: `ob resume` and `ob abort` skip non-deploy journals, `ob status` reports the application in sync, and only `ob audit` showed the operation — as INCOMPLETE, permanently. Meanwhile the container it started could still be changing data, and once the lock's TTL expired the next operation took ownership on the strength of a dead heartbeat. Both mutating entry points now reconcile under the lock, before anything mutates. A still-running container refuses the operation outright and names it: the host is executing a data-changing job that no process owns, and the TTL says nothing about whether that is finished. A container that is gone gets the terminal record its client could not write. What it does not do is guess. `--rm` means nothing about the container survives its exit, so success is only recorded where the client itself journaled the job's result — the one window where the outcome was observed before the client died. Everything else is recorded interrupted, which keeps the rollback debt an unresolved data-changing job carries rather than erasing it with a fabricated success. Deploy journals and completed runs are not orphans, and cost no container lookup. Refs #179. Execution is still attached; a client killed outright still records nothing at the time, but is now closed honestly by the next operation instead of staying incomplete forever. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/deploy.go | 6 ++ internal/engine/job.go | 6 ++ internal/engine/job_reconcile.go | 127 ++++++++++++++++++++++++++ internal/engine/job_reconcile_test.go | 100 ++++++++++++++++++++ 4 files changed, 239 insertions(+) create mode 100644 internal/engine/job_reconcile.go create mode 100644 internal/engine/job_reconcile_test.go diff --git a/internal/engine/deploy.go b/internal/engine/deploy.go index 7274aa4..53af5bb 100644 --- a/internal/engine/deploy.go +++ b/internal/engine/deploy.go @@ -81,6 +81,12 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri } stopHB := e.StartHeartbeat(ctx) defer stopHB() + // A deploy rolls workloads and runs its own gate jobs; an orphaned job run + // still changing data underneath it is exactly the overlap the lock exists + // to prevent, and the lock alone does not catch it once its holder is gone. + if err := e.reconcileOrphanedJobRuns(ctx); err != nil { + return err + } // The plan binding is the mutation boundary. Check it under the application // lock before converging even host-scoped support components; a stale plan // must leave both the application and proxy untouched. diff --git a/internal/engine/job.go b/internal/engine/job.go index 8ffd5de..4d261a8 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -66,6 +66,12 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) stopHeartbeat := e.StartHeartbeat(ctx) defer stopHeartbeat() + // Under the lock, before anything mutates: a previous run of this or any + // job may still be on the host with no process owning it. + if err := e.reconcileOrphanedJobRuns(ctx); err != nil { + return operationID, nil, err + } + current, err := release.Current(ctx, e.T, e.names()) if err != nil { return operationID, nil, err diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go new file mode 100644 index 0000000..59556cb --- /dev/null +++ b/internal/engine/job_reconcile.go @@ -0,0 +1,127 @@ +package engine + +import ( + "context" + "fmt" + "strings" + + "github.com/labstack/onebox/internal/journal" +) + +// orphanedJobRun is a sealed job run whose journal has a start and no terminal +// record. Either the client went away before it could write one, or the job is +// still running and this operation is the one that owns it. +type orphanedJobRun struct { + OperationID string + Job string + // ResultRecorded is true when the client got far enough to journal the + // job's result. That is the only after-the-fact proof of the outcome: the + // container is `--rm`, so once it exits nothing about it survives. + ResultRecorded bool + ResultOK bool +} + +// findOrphanedJobRuns reads every journal in one round trip and returns the job +// runs that never reached a terminal record. +func (e *Engine) findOrphanedJobRuns(ctx context.Context) ([]orphanedJobRun, error) { + ids, byID, err := journal.Journals(ctx, e.T, e.names()) + if err != nil { + return nil, err + } + var out []orphanedJobRun + for _, id := range ids { + if orphan, ok := orphanedJobRunOf(byID[id]); ok { + orphan.OperationID = id + out = append(out, orphan) + } + } + return out, nil +} + +// orphanedJobRunOf reduces one journal. A job run is orphaned when its start +// record has no matching finish — deploy journals and completed runs are not. +func orphanedJobRunOf(records []journal.Record) (orphanedJobRun, bool) { + orphan, started := orphanedJobRun{}, false + for _, r := range records { + if r.OperationKind != "job_run" || r.Phase != "job" { + continue + } + switch r.Event { + case "start": + started, orphan.Job = true, r.Service + case "finish": + return orphanedJobRun{}, false + case "result": + if r.SubStep == "job:"+orphan.Job { + orphan.ResultRecorded = true + orphan.ResultOK = r.Status == "ok" + } + } + } + return orphan, started +} + +// reconcileOrphanedJobRuns closes what it can prove and refuses what it cannot. +// +// It runs under the application lock, before any mutation. A still-running +// container means this host is already executing a data-changing job that no +// process owns: the only safe answer is to refuse, whatever the lock's TTL says +// about the client that started it. +func (e *Engine) reconcileOrphanedJobRuns(ctx context.Context) error { + orphans, err := e.findOrphanedJobRuns(ctx) + if err != nil { + return err + } + for _, orphan := range orphans { + running, err := e.jobContainerIDs(ctx, orphan.OperationID) + if err != nil { + return err + } + if len(running) > 0 { + return fmt.Errorf( + "job %s from operation %s is still running on the host (container %.12s) with no process owning it; "+ + "wait for it, or stop it with `docker rm -f %.12s` once you have established what it did", + orphan.Job, orphan.OperationID, running[0], running[0]) + } + if err := e.closeOrphanedJobRun(ctx, orphan); err != nil { + return err + } + } + return nil +} + +// closeOrphanedJobRun writes the terminal record the interrupted client could +// not. It never invents success: only a result the client itself journaled +// proves the job's outcome, and anything else is recorded as interrupted so the +// rollback debt an unresolved data-changing job carries is preserved. +func (e *Engine) closeOrphanedJobRun(ctx context.Context, orphan orphanedJobRun) error { + record := journal.Record{ + Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", + OperationKind: "job_run", Service: orphan.Job, + } + if orphan.ResultRecorded && orphan.ResultOK { + record.Status, record.ErrorCode = "ok", "" + } + writer := &journal.Writer{ + T: e.T, Names: e.names(), DeployID: orphan.OperationID, + Operator: journal.DefaultOperator(), + } + if err := writer.Append(ctx, record); err != nil { + return fmt.Errorf("close interrupted job run %s: %w", orphan.OperationID, err) + } + e.logf("closed interrupted job run %s (%s) as %s", orphan.OperationID, orphan.Job, record.Status) + return nil +} + +// jobContainerIDs lists containers still running for one operation. +func (e *Engine) jobContainerIDs(ctx context.Context, operationID string) ([]string, error) { + res, err := e.T.Run(ctx, "docker ps -q --filter label="+q(JobOperationLabel+"="+operationID)) + if err != nil { + return nil, err + } + if res.ExitCode != 0 { + return nil, fmt.Errorf("list containers of operation %s (exit %d): %s", + operationID, res.ExitCode, strings.TrimSpace(res.Stderr)) + } + return splitIDs(res.Stdout) +} diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go new file mode 100644 index 0000000..08f03e4 --- /dev/null +++ b/internal/engine/job_reconcile_test.go @@ -0,0 +1,100 @@ +package engine + +import ( + "bytes" + "context" + "strings" + "testing" + + "github.com/labstack/onebox/internal/transport" +) + +// reconcileFake serves one journal listing and answers the operation-label +// container lookup with whatever `running` names. +func reconcileFake(journals string, running []string) *transport.Fake { + return &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) { + switch { + case strings.Contains(cmd, "for f in"): + return transport.Result{Stdout: journals}, true + case strings.Contains(cmd, "docker ps -q") && strings.Contains(cmd, "ob.operation="): + return transport.Result{Stdout: strings.Join(running, "\n") + "\n"}, true + } + return transport.Result{}, false + }} +} + +func reconcileEngine(t *testing.T, f *transport.Fake) *Engine { + t.Helper() + return New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) +} + +const startedJobJournal = journalMarkerLine + "J1.jsonl\n" + + `{"deploy_id":"J1","phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t"}` + "\n" + +// The safety crux: a container still changing data with no process owning it +// must stop the next operation, whatever the lock's TTL says about the client +// that started it. +func TestReconcileRefusesWhileAnOrphanedJobRuns(t *testing.T) { + f := reconcileFake(startedJobJournal, []string{"CID0123456789ab"}) + err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()) + if err == nil { + t.Fatal("a live orphaned job must refuse the operation") + } + for _, want := range []string{"catalog-refresh", "J1", "still running", "CID012345678"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("refusal missing %q: %v", want, err) + } + } + if strings.Contains(strings.Join(f.Commands, "\n"), `"event":"finish"`) { + t.Fatalf("a running job must not be closed:\n%s", strings.Join(f.Commands, "\n")) + } +} + +// Gone, and the client never journaled a result: the outcome is unknown and +// must be recorded as such. Writing `ok` here would erase the rollback debt an +// unresolved data-changing job carries. +func TestReconcileClosesAGoneOrphanAsInterrupted(t *testing.T) { + f := reconcileFake(startedJobJournal, nil) + if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { + t.Fatalf("reconcile: %v", err) + } + appended := strings.Join(f.Commands, "\n") + if !strings.Contains(appended, `"error_code":"interrupted"`) || !strings.Contains(appended, `"status":"fail"`) { + t.Fatalf("interrupted run was not recorded honestly:\n%s", appended) + } +} + +// The one window where success is provable after the fact: the client observed +// the exit and journaled the result, then died before the finish record. +func TestReconcileClosesAProvenSuccessAsSucceeded(t *testing.T) { + journals := startedJobJournal + + `{"deploy_id":"J1","phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"ok","operation_kind":"job_run","ts":"t"}` + "\n" + f := reconcileFake(journals, nil) + if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { + t.Fatalf("reconcile: %v", err) + } + appended := strings.Join(f.Commands, "\n") + if !strings.Contains(appended, `"event":"finish"`) || !strings.Contains(appended, `"status":"ok"`) { + t.Fatalf("a proven success was not closed as one:\n%s", appended) + } + if strings.Contains(appended, "interrupted") { + t.Fatalf("a proven success must not be recorded as interrupted:\n%s", appended) + } +} + +// Deploy journals and completed job runs are not orphans. +func TestReconcileIgnoresDeploysAndFinishedRuns(t *testing.T) { + journals := journalMarkerLine + "R1.jsonl\n" + + `{"deploy_id":"R1","phase":"deploy","event":"start","ts":"t"}` + "\n" + + journalMarkerLine + "J2.jsonl\n" + + `{"deploy_id":"J2","phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + + `{"deploy_id":"J2","phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + f := reconcileFake(journals, nil) + if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { + t.Fatalf("reconcile: %v", err) + } + joined := strings.Join(f.Commands, "\n") + if strings.Contains(joined, "ob.operation=") { + t.Fatalf("nothing was orphaned, so no container lookup should happen:\n%s", joined) + } +} From d9d9083756f82281884072a045de848392651613 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:31:02 -0700 Subject: [PATCH 2/7] feat(jobs): ask Docker what is running, and close what the client could not MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconciliation derived "is a job still running" from a journal reduction. A journal says what a client managed to write, and the failure this exists for is a client that did not write — so the reduction was wrong in three ways at once. An interrupted run that DID record its interruption looked finished on paper while its container kept changing data, which is the Ctrl-C case the previous commit made more common. A plan re-run appends a second invocation to the same journal, and a finish in an earlier epoch was read as closing a later one. And a deploy-phase gate container carries the same labels but never appears in a job_run journal at all. All three are one `docker ps` away, so the refusal now asks Docker: any running one-off container whose operation is not this one stops the operation. The journal reduction stays only for the closing half, where being wrong is cheap. The closing half was also not doing what it claimed. It appended without an epoch, so the record landed in an invocation of its own and left the one it meant to close still open — `ob audit` went on reporting INCOMPLETE and grew a phantom row. It now groups a journal by invocation and writes into the one it closes. It also read the result record by an operation kind that record does not carry, so the branch that recognises a proven outcome could never fire and every reconciled run was recorded interrupted. Matched on its own shape now. A recorded failure is evidence exactly as much as a recorded success: only an absent result is an unknown outcome. The refusal is read-only and runs before the plan boundary; the append runs after it, so a plan that will not execute writes nothing. Refs #179. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/deploy.go | 13 +- internal/engine/job.go | 12 +- internal/engine/job_reconcile.go | 213 ++++++++++++++++---------- internal/engine/job_reconcile_test.go | 135 +++++++++++----- internal/onebox/service_test.go | 5 + 5 files changed, 247 insertions(+), 131 deletions(-) diff --git a/internal/engine/deploy.go b/internal/engine/deploy.go index 53af5bb..2eff111 100644 --- a/internal/engine/deploy.go +++ b/internal/engine/deploy.go @@ -81,10 +81,11 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri } stopHB := e.StartHeartbeat(ctx) defer stopHB() - // A deploy rolls workloads and runs its own gate jobs; an orphaned job run - // still changing data underneath it is exactly the overlap the lock exists - // to prevent, and the lock alone does not catch it once its holder is gone. - if err := e.reconcileOrphanedJobRuns(ctx); err != nil { + // A deploy rolls workloads and runs its own gate jobs; a job container still + // changing data underneath it is exactly the overlap the lock exists to + // prevent, and the lock alone does not catch it once its holder is gone. + // Read-only, so it runs before the plan-binding boundary below. + if err := e.refuseForeignJobContainers(ctx, releaseID); err != nil { return err } // The plan binding is the mutation boundary. Check it under the application @@ -95,6 +96,10 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri return fmt.Errorf("deploy precondition under lock: %w", err) } } + // Past the plan boundary, so a stale plan leaves the host untouched. + if err := e.closeInterruptedJobRuns(ctx); err != nil { + return err + } pf := e.ui.Step("preflight", false) if err := e.preflight(ctx, false); err != nil { pf(err) diff --git a/internal/engine/job.go b/internal/engine/job.go index 4d261a8..563c6c6 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -66,9 +66,10 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) stopHeartbeat := e.StartHeartbeat(ctx) defer stopHeartbeat() - // Under the lock, before anything mutates: a previous run of this or any - // job may still be on the host with no process owning it. - if err := e.reconcileOrphanedJobRuns(ctx); err != nil { + // Under the lock, before anything mutates and before any host write: a job + // container from an earlier operation may still be running with no process + // owning it. Read-only, so it is safe on this side of the plan boundary. + if err := e.refuseForeignJobContainers(ctx, operationID); err != nil { return operationID, nil, err } @@ -175,6 +176,11 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) } } + // Past the staleness checks, so a plan that will not execute writes nothing. + if err := e.closeInterruptedJobRuns(ctx); err != nil { + return operationID, nil, err + } + e.gateOpen = true e.rollbackCovered = true runErr := e.runJobPhase(ctx, writer, nil, remoteDir, remoteCompose, "job", []string{job}) diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go index 59556cb..a2889a9 100644 --- a/internal/engine/job_reconcile.go +++ b/internal/engine/job_reconcile.go @@ -8,120 +8,165 @@ import ( "github.com/labstack/onebox/internal/journal" ) -// orphanedJobRun is a sealed job run whose journal has a start and no terminal -// record. Either the client went away before it could write one, or the job is -// still running and this operation is the one that owns it. -type orphanedJobRun struct { - OperationID string - Job string - // ResultRecorded is true when the client got far enough to journal the - // job's result. That is the only after-the-fact proof of the outcome: the - // container is `--rm`, so once it exits nothing about it survives. - ResultRecorded bool - ResultOK bool -} - -// findOrphanedJobRuns reads every journal in one round trip and returns the job -// runs that never reached a terminal record. -func (e *Engine) findOrphanedJobRuns(ctx context.Context) ([]orphanedJobRun, error) { - ids, byID, err := journal.Journals(ctx, e.T, e.names()) +// refuseForeignJobContainers refuses when a one-off job container from another +// operation is still running on the host. +// +// The question "is a job still running" is asked of Docker, not of the journal. +// A journal says what a client managed to write, and the whole failure mode +// here is a client that did not write. An interrupted run that DID record its +// interruption looks finished on paper while its container keeps changing data, +// and a plan re-run appends a second invocation to the same journal — both are +// invisible to any reduction over records, and both are one `docker ps` away. +func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperationID string) error { + containers, err := e.jobContainers(ctx) if err != nil { - return nil, err + return err } - var out []orphanedJobRun - for _, id := range ids { - if orphan, ok := orphanedJobRunOf(byID[id]); ok { - orphan.OperationID = id - out = append(out, orphan) + for _, c := range containers { + if c.operation == currentOperationID { + continue } + return fmt.Errorf( + "a job container from operation %s is still running on this host (%.12s) with no process owning it; "+ + "wait for it to finish, or establish what it did and stop it with `docker rm -f %.12s`", + c.operation, c.id, c.id) } - return out, nil + return nil } -// orphanedJobRunOf reduces one journal. A job run is orphaned when its start -// record has no matching finish — deploy journals and completed runs are not. -func orphanedJobRunOf(records []journal.Record) (orphanedJobRun, bool) { - orphan, started := orphanedJobRun{}, false - for _, r := range records { - if r.OperationKind != "job_run" || r.Phase != "job" { +type jobContainer struct { + id string + operation string +} + +// jobContainers lists every running one-off job container, whichever operation +// created it. The label is unvalued in the filter so this finds containers of +// operations this process knows nothing about, which is the point. +func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) { + res, err := e.T.Run(ctx, + "docker ps --filter label="+q(JobOperationLabel)+ + " --format "+q("{{.ID}} {{.Label \""+JobOperationLabel+"\"}}")) + if err != nil { + return nil, err + } + if res.ExitCode != 0 { + return nil, fmt.Errorf("list running job containers (exit %d): %s", res.ExitCode, strings.TrimSpace(res.Stderr)) + } + var out []jobContainer + for _, line := range strings.Split(strings.TrimSpace(res.Stdout), "\n") { + id, operation, found := strings.Cut(strings.TrimSpace(line), " ") + if !found || id == "" { continue } - switch r.Event { - case "start": - started, orphan.Job = true, r.Service - case "finish": - return orphanedJobRun{}, false - case "result": - if r.SubStep == "job:"+orphan.Job { - orphan.ResultRecorded = true - orphan.ResultOK = r.Status == "ok" - } + if !validID.MatchString(id) { + return nil, fmt.Errorf("suspicious container id %q from docker ps — refusing to reuse in a command", id) } + out = append(out, jobContainer{id: id, operation: operation}) } - return orphan, started + return out, nil } -// reconcileOrphanedJobRuns closes what it can prove and refuses what it cannot. +// closeInterruptedJobRuns writes the terminal record an interrupted client +// could not, so an operation stops being incomplete forever. // -// It runs under the application lock, before any mutation. A still-running -// container means this host is already executing a data-changing job that no -// process owns: the only safe answer is to refuse, whatever the lock's TTL says -// about the client that started it. -func (e *Engine) reconcileOrphanedJobRuns(ctx context.Context) error { - orphans, err := e.findOrphanedJobRuns(ctx) +// Housekeeping, not safety: the refusal above is what protects a live +// container, and this runs only once nothing of the sort is running. Being +// wrong here is therefore cheap, which is why a journal reduction is good +// enough for it and is not good enough for the refusal. +func (e *Engine) closeInterruptedJobRuns(ctx context.Context) error { + ids, byID, err := journal.Journals(ctx, e.T, e.names()) if err != nil { return err } - for _, orphan := range orphans { - running, err := e.jobContainerIDs(ctx, orphan.OperationID) - if err != nil { - return err + for _, id := range ids { + for _, run := range unfinishedJobRuns(byID[id]) { + if err := e.closeJobRun(ctx, id, run); err != nil { + return err + } + } + } + return nil +} + +// unfinishedJobRun is one invocation that never reached a terminal record. +type unfinishedJobRun struct { + Epoch int + Job string + // ResultRecorded is true when the client journaled the job's own result + // before it went away. That is the only proof of the outcome that survives: + // the container is `--rm`, so nothing about it outlives its exit. A + // recorded failure is evidence exactly as much as a recorded success — only + // an absent result means the outcome is unknown. + ResultRecorded bool + ResultOK bool +} + +// unfinishedJobRuns groups a journal by epoch, because a journal holds one +// invocation per epoch and a plan may legitimately be run more than once. A +// finish in an earlier epoch says nothing about a later one. +func unfinishedJobRuns(records []journal.Record) []unfinishedJobRun { + type state struct { + started, finished bool + run unfinishedJobRun + } + byEpoch, order := map[int]*state{}, []int{} + for _, r := range records { + if r.Phase != "job" { + continue } - if len(running) > 0 { - return fmt.Errorf( - "job %s from operation %s is still running on the host (container %.12s) with no process owning it; "+ - "wait for it, or stop it with `docker rm -f %.12s` once you have established what it did", - orphan.Job, orphan.OperationID, running[0], running[0]) + s, seen := byEpoch[r.Epoch] + if !seen { + s = &state{run: unfinishedJobRun{Epoch: r.Epoch}} + byEpoch[r.Epoch], order = s, append(order, r.Epoch) } - if err := e.closeOrphanedJobRun(ctx, orphan); err != nil { - return err + switch { + case r.Event == "start" && r.OperationKind == "job_run": + s.started, s.run.Job = true, r.Service + case r.Event == "finish": + s.finished = true + case r.Event == "result" && strings.HasPrefix(r.SubStep, "job:"): + // The result record is written by the shared job phase and carries + // no operation kind, so it is matched on its own shape. + s.run.ResultRecorded = true + s.run.ResultOK = r.Status == "ok" } } - return nil + var out []unfinishedJobRun + for _, epoch := range order { + if s := byEpoch[epoch]; s.started && !s.finished { + out = append(out, s.run) + } + } + return out } -// closeOrphanedJobRun writes the terminal record the interrupted client could -// not. It never invents success: only a result the client itself journaled -// proves the job's outcome, and anything else is recorded as interrupted so the -// rollback debt an unresolved data-changing job carries is preserved. -func (e *Engine) closeOrphanedJobRun(ctx context.Context, orphan orphanedJobRun) error { +// closeJobRun records the outcome, never inventing one. Only a result the +// client itself journaled proves the job succeeded; anything else is recorded +// interrupted, which is what an unknown outcome actually is. +func (e *Engine) closeJobRun(ctx context.Context, operationID string, run unfinishedJobRun) error { record := journal.Record{ Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", - OperationKind: "job_run", Service: orphan.Job, + OperationKind: "job_run", Service: run.Job, } - if orphan.ResultRecorded && orphan.ResultOK { - record.Status, record.ErrorCode = "ok", "" + if run.ResultRecorded { + // The outcome was observed and written down. Interrupted describes an + // unknown outcome, and saying it of a known failure hides that the job + // ran and failed on its own terms. + record.ErrorCode = "" + if run.ResultOK { + record.Status = "ok" + } } + // The epoch is what groups a journal into invocations, so a terminal record + // written without it lands in an invocation of its own and leaves the one it + // was meant to close still open. writer := &journal.Writer{ - T: e.T, Names: e.names(), DeployID: orphan.OperationID, + T: e.T, Names: e.names(), DeployID: operationID, Epoch: run.Epoch, Operator: journal.DefaultOperator(), } if err := writer.Append(ctx, record); err != nil { - return fmt.Errorf("close interrupted job run %s: %w", orphan.OperationID, err) + return fmt.Errorf("close interrupted job run %s: %w", operationID, err) } - e.logf("closed interrupted job run %s (%s) as %s", orphan.OperationID, orphan.Job, record.Status) + e.logf("closed interrupted job run %s (%s) as %s", operationID, run.Job, record.Status) return nil } - -// jobContainerIDs lists containers still running for one operation. -func (e *Engine) jobContainerIDs(ctx context.Context, operationID string) ([]string, error) { - res, err := e.T.Run(ctx, "docker ps -q --filter label="+q(JobOperationLabel+"="+operationID)) - if err != nil { - return nil, err - } - if res.ExitCode != 0 { - return nil, fmt.Errorf("list containers of operation %s (exit %d): %s", - operationID, res.ExitCode, strings.TrimSpace(res.Stderr)) - } - return splitIDs(res.Stdout) -} diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go index 08f03e4..70a4570 100644 --- a/internal/engine/job_reconcile_test.go +++ b/internal/engine/job_reconcile_test.go @@ -9,14 +9,14 @@ import ( "github.com/labstack/onebox/internal/transport" ) -// reconcileFake serves one journal listing and answers the operation-label -// container lookup with whatever `running` names. +// reconcileFake serves one journal listing and one running-container listing. +// `running` is ` ` lines, exactly as the label probe formats. func reconcileFake(journals string, running []string) *transport.Fake { return &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) { switch { case strings.Contains(cmd, "for f in"): return transport.Result{Stdout: journals}, true - case strings.Contains(cmd, "docker ps -q") && strings.Contains(cmd, "ob.operation="): + case strings.Contains(cmd, "label='ob.operation'"): return transport.Result{Stdout: strings.Join(running, "\n") + "\n"}, true } return transport.Result{}, false @@ -29,72 +29,127 @@ func reconcileEngine(t *testing.T, f *transport.Fake) *Engine { } const startedJobJournal = journalMarkerLine + "J1.jsonl\n" + - `{"deploy_id":"J1","phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t"}` + "\n" + `{"deploy_id":"J1","epoch":4,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t1"}` + "\n" -// The safety crux: a container still changing data with no process owning it -// must stop the next operation, whatever the lock's TTL says about the client -// that started it. -func TestReconcileRefusesWhileAnOrphanedJobRuns(t *testing.T) { - f := reconcileFake(startedJobJournal, []string{"CID0123456789ab"}) - err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()) +// The safety crux, and the reason this asks Docker rather than the journal: a +// container still changing data with no process owning it must stop the next +// operation. +func TestRefuseWhileAnotherOperationsJobContainerRuns(t *testing.T) { + f := reconcileFake("", []string{"abc123def456 J1"}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2") if err == nil { - t.Fatal("a live orphaned job must refuse the operation") + t.Fatal("a live job container from another operation must refuse") } - for _, want := range []string{"catalog-refresh", "J1", "still running", "CID012345678"} { + for _, want := range []string{"J1", "abc123def456", "still running"} { if !strings.Contains(err.Error(), want) { t.Fatalf("refusal missing %q: %v", want, err) } } - if strings.Contains(strings.Join(f.Commands, "\n"), `"event":"finish"`) { - t.Fatalf("a running job must not be closed:\n%s", strings.Join(f.Commands, "\n")) +} + +// This operation's own container is not a reason to refuse itself — a deploy +// runs gate jobs under its own id. +func TestRefuseAllowsThisOperationsOwnContainer(t *testing.T) { + f := reconcileFake("", []string{"abc123def456 J1"}) + if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1"); err != nil { + t.Fatalf("own container refused: %v", err) + } +} + +// A run that recorded its own interruption still has a live container, and a +// journal reduction would call it finished. That is the Ctrl-C case. +func TestRefuseCatchesAnInterruptedRunThatRecordedItself(t *testing.T) { + journals := startedJobJournal + + `{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"fail","error_code":"interrupted","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n" + f := reconcileFake(journals, []string{"abc123def456 J1"}) + if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2"); err == nil { + t.Fatal("a recorded interruption must not hide a live container") } } -// Gone, and the client never journaled a result: the outcome is unknown and -// must be recorded as such. Writing `ok` here would erase the rollback debt an -// unresolved data-changing job carries. -func TestReconcileClosesAGoneOrphanAsInterrupted(t *testing.T) { +// Gone, and the client never journaled a result: the outcome is unknown. +func TestCloseRecordsAnUnknownOutcomeAsInterrupted(t *testing.T) { f := reconcileFake(startedJobJournal, nil) - if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { - t.Fatalf("reconcile: %v", err) + if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") - if !strings.Contains(appended, `"error_code":"interrupted"`) || !strings.Contains(appended, `"status":"fail"`) { + if !strings.Contains(appended, `"error_code":"interrupted"`) || !strings.Contains(appended, `"event":"finish"`) { t.Fatalf("interrupted run was not recorded honestly:\n%s", appended) } + // The epoch groups a journal into invocations. Written without it, the + // record lands in an invocation of its own and leaves this one open. + if !strings.Contains(appended, `"epoch":4`) { + t.Fatalf("terminal record did not join the invocation it closes:\n%s", appended) + } } -// The one window where success is provable after the fact: the client observed -// the exit and journaled the result, then died before the finish record. -func TestReconcileClosesAProvenSuccessAsSucceeded(t *testing.T) { +// The one window where success is provable after the fact. The result record is +// written by the shared job phase and carries no operation kind, so matching it +// on that would silently never fire. +func TestCloseRecordsAJournaledResultAsSuccess(t *testing.T) { journals := startedJobJournal + - `{"deploy_id":"J1","phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"ok","operation_kind":"job_run","ts":"t"}` + "\n" + `{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"ok","ts":"t2"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { - t.Fatalf("reconcile: %v", err) + if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") - if !strings.Contains(appended, `"event":"finish"`) || !strings.Contains(appended, `"status":"ok"`) { + if !strings.Contains(appended, `"status":"ok"`) || strings.Contains(appended, "interrupted") { t.Fatalf("a proven success was not closed as one:\n%s", appended) } - if strings.Contains(appended, "interrupted") { - t.Fatalf("a proven success must not be recorded as interrupted:\n%s", appended) +} + +// A plan may be run more than once, appending a second invocation to the same +// journal. A finish in an earlier epoch says nothing about a later one. +func TestCloseGroupsAJournalByInvocation(t *testing.T) { + journals := startedJobJournal + + `{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n" + + `{"deploy_id":"J1","epoch":5,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t3"}` + "\n" + f := reconcileFake(journals, nil) + if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + t.Fatalf("close: %v", err) + } + appended := strings.Join(f.Commands, "\n") + if !strings.Contains(appended, `"epoch":5`) { + t.Fatalf("the unfinished second invocation was not closed:\n%s", appended) + } + if strings.Count(appended, `"event":"finish"`) != 1 { + t.Fatalf("the finished invocation must not be closed again:\n%s", appended) } } -// Deploy journals and completed job runs are not orphans. -func TestReconcileIgnoresDeploysAndFinishedRuns(t *testing.T) { +// Deploy journals and completed runs are not orphans. +func TestCloseIgnoresDeploysAndFinishedRuns(t *testing.T) { journals := journalMarkerLine + "R1.jsonl\n" + - `{"deploy_id":"R1","phase":"deploy","event":"start","ts":"t"}` + "\n" + + `{"deploy_id":"R1","epoch":1,"phase":"deploy","event":"start","ts":"t"}` + "\n" + journalMarkerLine + "J2.jsonl\n" + - `{"deploy_id":"J2","phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + - `{"deploy_id":"J2","phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + `{"deploy_id":"J2","epoch":1,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + + `{"deploy_id":"J2","epoch":1,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).reconcileOrphanedJobRuns(context.Background()); err != nil { - t.Fatalf("reconcile: %v", err) + if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + t.Fatalf("close: %v", err) } - joined := strings.Join(f.Commands, "\n") - if strings.Contains(joined, "ob.operation=") { - t.Fatalf("nothing was orphaned, so no container lookup should happen:\n%s", joined) + if strings.Contains(strings.Join(f.Commands, "\n"), `"event":"finish"`) { + t.Fatalf("nothing was unfinished, so nothing should be written:\n%s", strings.Join(f.Commands, "\n")) + } +} + +// A recorded failure is evidence of the outcome exactly as much as a recorded +// success. Calling it interrupted would hide that the job ran and failed on its +// own terms. +func TestCloseKeepsARecordedFailureAsAFailure(t *testing.T) { + journals := startedJobJournal + + `{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"fail","ts":"t2"}` + "\n" + f := reconcileFake(journals, nil) + if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + t.Fatalf("close: %v", err) + } + appended := strings.Join(f.Commands, "\n") + if !strings.Contains(appended, `"status":"fail"`) { + t.Fatalf("a recorded failure was not closed as one:\n%s", appended) + } + if strings.Contains(appended, "interrupted") { + t.Fatalf("a known failure must not be reported as an unknown outcome:\n%s", appended) } } diff --git a/internal/onebox/service_test.go b/internal/onebox/service_test.go index 113c25b..12aeecd 100644 --- a/internal/onebox/service_test.go +++ b/internal/onebox/service_test.go @@ -81,6 +81,9 @@ func serviceFake() *transport.Fake { return transport.Result{Stdout: "demo\n"}, true case strings.Contains(cmd, "readlink"): return transport.Result{Stdout: "releases/R0\n"}, true + // Ahead of the project-container probe, which also uses --format. + case strings.Contains(cmd, "label='ob.operation'"): + return transport.Result{Stdout: "\n"}, true case strings.Contains(cmd, "docker ps") && strings.Contains(cmd, "--format"): return transport.Result{Stdout: "S1|web|R0|Up (healthy)\nPG1|database|R0|Up (healthy)\n"}, true case strings.Contains(cmd, "for f in"): @@ -267,6 +270,8 @@ deployment: {order: [web, worker]} return transport.Result{Stdout: project(false)}, true case strings.Contains(command, "cat ") && strings.Contains(command, "compose.yaml"): return transport.Result{Stdout: liveCompose}, true + case strings.Contains(command, "label='ob.operation'"): + return transport.Result{Stdout: "\n"}, true case strings.Contains(command, "docker ps") && strings.Contains(command, "--format"): return transport.Result{Stdout: "S1|web|R0|Up\nW1|worker|R0|Up\n"}, true default: From 01938e097987fb1107caf042cead6010a1a8b0dd Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:39:58 -0700 Subject: [PATCH 3/7] perf(jobs): read the journals once per deploy, and refuse an unattributable container MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A deploy scanned every journal twice — once to close interrupted job runs and again for rollback debt — which on a high-latency host is the same bytes fetched and parsed for a second time. One read now serves both. A container carrying the operation label with no value cannot be attributed to an operation, and the refusal printed a blank id for it. It is refused on its own terms now: an unattributable job container is exactly as dangerous as an attributable one, and the message says which problem the operator has. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/deploy.go | 18 ++++++++++-------- internal/engine/job.go | 6 +++++- internal/engine/job_reconcile.go | 17 ++++++++++++----- internal/engine/job_reconcile_test.go | 21 ++++++++++++++++----- 4 files changed, 43 insertions(+), 19 deletions(-) diff --git a/internal/engine/deploy.go b/internal/engine/deploy.go index 2eff111..992c06c 100644 --- a/internal/engine/deploy.go +++ b/internal/engine/deploy.go @@ -96,8 +96,14 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri return fmt.Errorf("deploy precondition under lock: %w", err) } } - // Past the plan boundary, so a stale plan leaves the host untouched. - if err := e.closeInterruptedJobRuns(ctx); err != nil { + // Past the plan boundary, so a stale plan leaves the host untouched. One + // read of the journals serves both this and the rollback-debt scan below; + // they are the same bytes off a possibly high-latency host. + journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names()) + if err != nil { + return err + } + if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil { return err } pf := e.ui.Step("preflight", false) @@ -118,7 +124,7 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri } rollbackDebt := false if done == nil { - rollbackDebt, err = e.rollbackEffectDebt(ctx, prev) + rollbackDebt, err = e.rollbackEffectDebt(prev, journalIDs, journalsByID) if err != nil { return fmt.Errorf("rollback effect history: %w", err) } @@ -225,11 +231,7 @@ func (e *Engine) pinnedScheduleDeployConflict() string { // failed deploy can mutate data even though its runner exits cleanly and writes // finish:fail; a later successful activation/current release or an explicit // abort clears that historical debt. -func (e *Engine) rollbackEffectDebt(ctx context.Context, current string) (bool, error) { - ids, byID, err := journal.Journals(ctx, e.T, e.names()) - if err != nil { - return false, err - } +func (e *Engine) rollbackEffectDebt(current string, ids []string, byID map[string][]journal.Record) (bool, error) { debt := false for _, id := range ids { summary := journal.Summarize(byID[id]) diff --git a/internal/engine/job.go b/internal/engine/job.go index 563c6c6..4c283b1 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -177,7 +177,11 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) } // Past the staleness checks, so a plan that will not execute writes nothing. - if err := e.closeInterruptedJobRuns(ctx); err != nil { + journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names()) + if err != nil { + return operationID, nil, err + } + if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil { return operationID, nil, err } diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go index a2889a9..b2a80fa 100644 --- a/internal/engine/job_reconcile.go +++ b/internal/engine/job_reconcile.go @@ -26,6 +26,15 @@ func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperatio if c.operation == currentOperationID { continue } + if c.operation == "" { + // The label is present but carries no value, so the container + // cannot be attributed. Refuse anyway: an unattributable job + // container is exactly as dangerous as an attributable one. + return fmt.Errorf( + "a job container is running on this host (%.12s) with an empty %s label, so the operation that "+ + "started it cannot be identified; establish what it did and stop it with `docker rm -f %.12s`", + c.id, JobOperationLabel, c.id) + } return fmt.Errorf( "a job container from operation %s is still running on this host (%.12s) with no process owning it; "+ "wait for it to finish, or establish what it did and stop it with `docker rm -f %.12s`", @@ -73,11 +82,9 @@ func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) { // container, and this runs only once nothing of the sort is running. Being // wrong here is therefore cheap, which is why a journal reduction is good // enough for it and is not good enough for the refusal. -func (e *Engine) closeInterruptedJobRuns(ctx context.Context) error { - ids, byID, err := journal.Journals(ctx, e.T, e.names()) - if err != nil { - return err - } +// The caller supplies the journals so a deploy, which also scans them for +// rollback debt, reads them once. +func (e *Engine) closeInterruptedJobRuns(ctx context.Context, ids []string, byID map[string][]journal.Record) error { for _, id := range ids { for _, run := range unfinishedJobRuns(byID[id]) { if err := e.closeJobRun(ctx, id, run); err != nil { diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go index 70a4570..221d4a5 100644 --- a/internal/engine/job_reconcile_test.go +++ b/internal/engine/job_reconcile_test.go @@ -6,6 +6,7 @@ import ( "strings" "testing" + "github.com/labstack/onebox/internal/journal" "github.com/labstack/onebox/internal/transport" ) @@ -28,6 +29,16 @@ func reconcileEngine(t *testing.T, f *transport.Fake) *Engine { return New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep}) } +// closeAll reads the journals the way a caller does, then closes what it finds. +func closeAll(t *testing.T, e *Engine) error { + t.Helper() + ids, byID, err := journal.Journals(context.Background(), e.T, e.names()) + if err != nil { + t.Fatalf("read journals: %v", err) + } + return e.closeInterruptedJobRuns(context.Background(), ids, byID) +} + const startedJobJournal = journalMarkerLine + "J1.jsonl\n" + `{"deploy_id":"J1","epoch":4,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t1"}` + "\n" @@ -70,7 +81,7 @@ func TestRefuseCatchesAnInterruptedRunThatRecordedItself(t *testing.T) { // Gone, and the client never journaled a result: the outcome is unknown. func TestCloseRecordsAnUnknownOutcomeAsInterrupted(t *testing.T) { f := reconcileFake(startedJobJournal, nil) - if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + if err := closeAll(t, reconcileEngine(t, f)); err != nil { t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") @@ -91,7 +102,7 @@ func TestCloseRecordsAJournaledResultAsSuccess(t *testing.T) { journals := startedJobJournal + `{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"ok","ts":"t2"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + if err := closeAll(t, reconcileEngine(t, f)); err != nil { t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") @@ -107,7 +118,7 @@ func TestCloseGroupsAJournalByInvocation(t *testing.T) { `{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n" + `{"deploy_id":"J1","epoch":5,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t3"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + if err := closeAll(t, reconcileEngine(t, f)); err != nil { t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") @@ -127,7 +138,7 @@ func TestCloseIgnoresDeploysAndFinishedRuns(t *testing.T) { `{"deploy_id":"J2","epoch":1,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" + `{"deploy_id":"J2","epoch":1,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + if err := closeAll(t, reconcileEngine(t, f)); err != nil { t.Fatalf("close: %v", err) } if strings.Contains(strings.Join(f.Commands, "\n"), `"event":"finish"`) { @@ -142,7 +153,7 @@ func TestCloseKeepsARecordedFailureAsAFailure(t *testing.T) { journals := startedJobJournal + `{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"fail","ts":"t2"}` + "\n" f := reconcileFake(journals, nil) - if err := reconcileEngine(t, f).closeInterruptedJobRuns(context.Background()); err != nil { + if err := closeAll(t, reconcileEngine(t, f)); err != nil { t.Fatalf("close: %v", err) } appended := strings.Join(f.Commands, "\n") From 49b307efefa4042dd89aaca79532b671808a942c Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 10:56:02 -0700 Subject: [PATCH 4/7] fix(jobs): do not skip a container whose operation label is empty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Trimming the docker ps line before splitting it removed the separator on a container whose label carries no value, so the line parsed as unsplittable and the container was skipped — the refusal added for exactly that case could never fire. Split first, trim the fields after. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job_reconcile.go | 12 +++++++++--- internal/engine/job_reconcile_test.go | 14 ++++++++++++++ 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go index b2a80fa..91f1bdb 100644 --- a/internal/engine/job_reconcile.go +++ b/internal/engine/job_reconcile.go @@ -62,9 +62,15 @@ func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) { return nil, fmt.Errorf("list running job containers (exit %d): %s", res.ExitCode, strings.TrimSpace(res.Stderr)) } var out []jobContainer - for _, line := range strings.Split(strings.TrimSpace(res.Stdout), "\n") { - id, operation, found := strings.Cut(strings.TrimSpace(line), " ") - if !found || id == "" { + for _, line := range strings.Split(res.Stdout, "\n") { + // Not TrimSpace before the cut: a container whose label carries no value + // prints " " with nothing after the separator, and trimming the line + // first removes the separator itself — the container would then be + // skipped as unparseable, which is precisely the one that most needs + // refusing. + id, operation, _ := strings.Cut(strings.TrimRight(line, "\r\n"), " ") + id, operation = strings.TrimSpace(id), strings.TrimSpace(operation) + if id == "" { continue } if !validID.MatchString(id) { diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go index 221d4a5..f87861f 100644 --- a/internal/engine/job_reconcile_test.go +++ b/internal/engine/job_reconcile_test.go @@ -164,3 +164,17 @@ func TestCloseKeepsARecordedFailureAsAFailure(t *testing.T) { t.Fatalf("a known failure must not be reported as an unknown outcome:\n%s", appended) } } + +// A container whose label carries no value prints " " with nothing after +// the separator. Trimming the line before the cut removes the separator, and +// the container is skipped as unparseable — the one that most needs refusing. +func TestRefuseCatchesAContainerWithAnEmptyOperationLabel(t *testing.T) { + f := reconcileFake("", []string{"abc123def456 "}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2") + if err == nil { + t.Fatal("an unattributable job container must refuse") + } + if !strings.Contains(err.Error(), "empty") || !strings.Contains(err.Error(), "abc123def456") { + t.Fatalf("refusal did not name the problem: %v", err) + } +} From 011ac48be7d05f79ca2e9a34333e2779371d6f72 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 11:04:14 -0700 Subject: [PATCH 5/7] fix(jobs): reconcile before this run writes its own start record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot was taken after the start record was appended, so it contained a start with no finish yet — this very run — and the reconciler closed the job it was about to execute, writing a spurious interrupted record ahead of the real one. It now runs between the staleness checks and the start append: late enough that a plan which will not execute writes nothing, early enough that this run is not yet in the journal it is reading. The regression test drives the whole run against a journal listing that reflects what the run has appended so far, so the snapshot's contents depend on when it is taken — without that the fake returned nothing and the ordering could not be told apart. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/job.go | 21 ++++++++++------- internal/engine/job_test.go | 47 +++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/internal/engine/job.go b/internal/engine/job.go index 4c283b1..c6d4799 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -93,6 +93,18 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) return operationID, nil, errors.New("job plan is stale: current release runtime changed — re-plan") } + // Before this operation writes its own start record, and after the + // staleness checks above. Reconciling later would find this run's start + // with no finish yet and close the very run about to execute; reconciling + // earlier would let a plan that will not execute write to the host. + journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names()) + if err != nil { + return operationID, nil, err + } + if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil { + return operationID, nil, err + } + writer := &journal.Writer{ T: e.T, Names: e.names(), DeployID: operationID, Epoch: epoch, Operator: journal.DefaultOperator(), GitSHA: e.Opts.GitSHA, ConfigHash: e.Opts.ConfigHash, @@ -176,15 +188,6 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) } } - // Past the staleness checks, so a plan that will not execute writes nothing. - journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names()) - if err != nil { - return operationID, nil, err - } - if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil { - return operationID, nil, err - } - e.gateOpen = true e.rollbackCovered = true runErr := e.runJobPhase(ctx, writer, nil, remoteDir, remoteCompose, "job", []string{job}) diff --git a/internal/engine/job_test.go b/internal/engine/job_test.go index 88cfd4b..acaf402 100644 --- a/internal/engine/job_test.go +++ b/internal/engine/job_test.go @@ -172,3 +172,50 @@ func TestInterruptedRunClassifiesTheRunNotTheClient(t *testing.T) { t.Fatal("a job that failed on its own terms is not interrupted") } } + +// Reconciliation runs before this operation writes its own start record. Run it +// after, and the snapshot contains a start with no finish yet — this very run — +// which the reconciler would close as interrupted before the job executes. +func TestRunJobDoesNotCloseTheRunItIsAboutToStart(t *testing.T) { + const runtime = "services:\n migrate:\n image: ghcr.io/x/app@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n" + target := currentJobFake(runtime) + // The journal listing reflects what this run has appended so far, so the + // snapshot's contents depend on when it is taken — which is the whole + // question. + inner := target.Dynamic + target.Dynamic = func(cmd string) (transport.Result, bool) { + if strings.Contains(cmd, "for f in") { + var lines []string + for _, c := range target.Commands { + start := strings.Index(c, `{"deploy_id":"op-job-run"`) + if start < 0 { + continue + } + if end := strings.LastIndex(c, "}"); end > start { + lines = append(lines, c[start:end+1]) + } + } + if len(lines) == 0 { + return transport.Result{}, true + } + return transport.Result{Stdout: "@@ob-journal@@op-job-run.jsonl\n" + strings.Join(lines, "\n") + "\n"}, true + } + return inner(cmd) + } + engine := manualJobEngine(t, target) + request := JobRunRequest{ + OperationID: "op-job-run", Job: "migrate", ExpectedRelease: engineTestPreviousReleaseID, + ExpectedRuntimeDigest: HashBytes([]byte(runtime)), ExpectedDataEffect: "none", + } + if _, _, err := engine.RunJobWithJournalID(context.Background(), request); err != nil { + t.Fatalf("run job: %v", err) + } + commands := strings.Join(target.Commands, "\n") + if strings.Contains(commands, `"error_code":"interrupted"`) { + t.Fatalf("the run closed itself as interrupted before executing:\n%s", commands) + } + // Exactly one terminal record: the real one, written after the job ran. + if got := strings.Count(commands, `"phase":"job","event":"finish"`); got != 1 { + t.Fatalf("terminal record count = %d, want 1:\n%s", got, commands) + } +} From a2668ef13bf859c8c0bc238c97d9b5b24b612278 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 11:05:10 -0700 Subject: [PATCH 6/7] docs(jobs): note the reconciler as the interrupted code's second writer Now that a later operation closes a run the client could not, the registry comment can say so. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/onebox/operation_errors.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go index ce7ba7e..9370af4 100644 --- a/internal/onebox/operation_errors.go +++ b/internal/onebox/operation_errors.go @@ -123,7 +123,8 @@ var operationFailureDefinitions = map[string]OperationFailure{ // 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, on a context of - // its own, whenever it can still reach the host. + // its own, whenever it can still reach the host — and otherwise by the + // next operation, which finds the run unfinished and closes it. Message: "the operation's client went away before its outcome could be recorded", Command: "ob audit --output json", }, From 11142b01d5d36b6e6c3a8a3dc9d3823de5abb673 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Wed, 9 Sep 2026 12:12:36 -0700 Subject: [PATCH 7/7] fix(jobs): refuse an earlier invocation of the same operation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The refusal exempted any container sharing the current operation id. A sealed job plan carries one operation id for its whole life and is re-runnable, and AcquireLock hands the lock straight back to a caller presenting the id already written in it — so a second run of one plan reclaimed the lock from a live first run and then exempted that run's container as its own. Two concurrent data-changing containers, which is the single thing this exists to prevent. `ob resume` reaches the same hole, since it carries the interrupted attempt's id. Matching now requires the epoch as well, which is what the label added alongside the operation was for and which nothing had read. A container that carries no epoch cannot be shown to belong to this invocation and is not exempt from it. The terminal record no longer stamps the reconciling operator. Audit takes the last non-empty operator in an epoch group, so it rewrote the interrupted run's row to name whoever deployed next; the start record already carries who ran it. Adds the first test that pins a call site rather than the function behind it: every refusal call site could be deleted with the suite green. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 --- internal/engine/deploy.go | 2 +- internal/engine/job.go | 2 +- internal/engine/job_reconcile.go | 40 ++++++++++++++---- internal/engine/job_reconcile_test.go | 58 +++++++++++++++++++++++---- internal/engine/job_test.go | 26 ++++++++++++ 5 files changed, 111 insertions(+), 17 deletions(-) diff --git a/internal/engine/deploy.go b/internal/engine/deploy.go index 992c06c..e67078c 100644 --- a/internal/engine/deploy.go +++ b/internal/engine/deploy.go @@ -85,7 +85,7 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri // changing data underneath it is exactly the overlap the lock exists to // prevent, and the lock alone does not catch it once its holder is gone. // Read-only, so it runs before the plan-binding boundary below. - if err := e.refuseForeignJobContainers(ctx, releaseID); err != nil { + if err := e.refuseForeignJobContainers(ctx, releaseID, epoch); err != nil { return err } // The plan binding is the mutation boundary. Check it under the application diff --git a/internal/engine/job.go b/internal/engine/job.go index c6d4799..3f99758 100644 --- a/internal/engine/job.go +++ b/internal/engine/job.go @@ -69,7 +69,7 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest) // Under the lock, before anything mutates and before any host write: a job // container from an earlier operation may still be running with no process // owning it. Read-only, so it is safe on this side of the plan boundary. - if err := e.refuseForeignJobContainers(ctx, operationID); err != nil { + if err := e.refuseForeignJobContainers(ctx, operationID, epoch); err != nil { return operationID, nil, err } diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go index 91f1bdb..e865dce 100644 --- a/internal/engine/job_reconcile.go +++ b/internal/engine/job_reconcile.go @@ -3,6 +3,7 @@ package engine import ( "context" "fmt" + "strconv" "strings" "github.com/labstack/onebox/internal/journal" @@ -17,15 +18,28 @@ import ( // interruption looks finished on paper while its container keeps changing data, // and a plan re-run appends a second invocation to the same journal — both are // invisible to any reduction over records, and both are one `docker ps` away. -func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperationID string) error { +func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperationID string, currentEpoch int) error { containers, err := e.jobContainers(ctx) if err != nil { return err } + currentEpochLabel := strconv.Itoa(currentEpoch) for _, c := range containers { - if c.operation == currentOperationID { + // Operation AND epoch. A sealed job plan is re-runnable and carries one + // operation id for its whole life, and AcquireLock hands the lock + // straight back to a caller presenting the id already written in it. So + // a second run of one plan would reclaim the lock from a live first run + // and then exempt that run's container as its own — two concurrent + // data-changing containers, which is the single thing this prevents. + if c.operation == currentOperationID && c.epoch == currentEpochLabel { continue } + if c.operation == currentOperationID { + return fmt.Errorf( + "an earlier run of operation %s (epoch %s) left a job container running on this host (%.12s); "+ + "wait for it to finish, or establish what it did and stop it with `docker rm -f %.12s`", + c.operation, labelOrUnknown(c.epoch), c.id, c.id) + } if c.operation == "" { // The label is present but carries no value, so the container // cannot be attributed. Refuse anyway: an unattributable job @@ -46,6 +60,14 @@ func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperatio type jobContainer struct { id string operation string + epoch string +} + +func labelOrUnknown(value string) string { + if value == "" { + return "unknown" + } + return value } // jobContainers lists every running one-off job container, whichever operation @@ -54,7 +76,7 @@ type jobContainer struct { func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) { res, err := e.T.Run(ctx, "docker ps --filter label="+q(JobOperationLabel)+ - " --format "+q("{{.ID}} {{.Label \""+JobOperationLabel+"\"}}")) + " --format "+q("{{.ID}} {{.Label \""+JobOperationLabel+"\"}} {{.Label \""+JobEpochLabel+"\"}}")) if err != nil { return nil, err } @@ -68,15 +90,16 @@ func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) { // first removes the separator itself — the container would then be // skipped as unparseable, which is precisely the one that most needs // refusing. - id, operation, _ := strings.Cut(strings.TrimRight(line, "\r\n"), " ") - id, operation = strings.TrimSpace(id), strings.TrimSpace(operation) + id, rest, _ := strings.Cut(strings.TrimRight(line, "\r\n"), " ") + operation, epoch, _ := strings.Cut(rest, " ") + id, operation, epoch = strings.TrimSpace(id), strings.TrimSpace(operation), strings.TrimSpace(epoch) if id == "" { continue } if !validID.MatchString(id) { return nil, fmt.Errorf("suspicious container id %q from docker ps — refusing to reuse in a command", id) } - out = append(out, jobContainer{id: id, operation: operation}) + out = append(out, jobContainer{id: id, operation: operation, epoch: epoch}) } return out, nil } @@ -173,9 +196,12 @@ func (e *Engine) closeJobRun(ctx context.Context, operationID string, run unfini // The epoch is what groups a journal into invocations, so a terminal record // written without it lands in an invocation of its own and leaves the one it // was meant to close still open. + // No Operator: audit takes the last non-empty operator in an epoch group, so + // stamping the reconciling operator here would rewrite the interrupted run's + // row to name whoever happened to deploy next. The start record already + // carries who ran it. writer := &journal.Writer{ T: e.T, Names: e.names(), DeployID: operationID, Epoch: run.Epoch, - Operator: journal.DefaultOperator(), } if err := writer.Append(ctx, record); err != nil { return fmt.Errorf("close interrupted job run %s: %w", operationID, err) diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go index f87861f..02104f4 100644 --- a/internal/engine/job_reconcile_test.go +++ b/internal/engine/job_reconcile_test.go @@ -46,8 +46,8 @@ const startedJobJournal = journalMarkerLine + "J1.jsonl\n" + // container still changing data with no process owning it must stop the next // operation. func TestRefuseWhileAnotherOperationsJobContainerRuns(t *testing.T) { - f := reconcileFake("", []string{"abc123def456 J1"}) - err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2") + f := reconcileFake("", []string{"abc123def456 J1 4"}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4) if err == nil { t.Fatal("a live job container from another operation must refuse") } @@ -61,8 +61,8 @@ func TestRefuseWhileAnotherOperationsJobContainerRuns(t *testing.T) { // This operation's own container is not a reason to refuse itself — a deploy // runs gate jobs under its own id. func TestRefuseAllowsThisOperationsOwnContainer(t *testing.T) { - f := reconcileFake("", []string{"abc123def456 J1"}) - if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1"); err != nil { + f := reconcileFake("", []string{"abc123def456 J1 4"}) + if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 4); err != nil { t.Fatalf("own container refused: %v", err) } } @@ -72,8 +72,8 @@ func TestRefuseAllowsThisOperationsOwnContainer(t *testing.T) { func TestRefuseCatchesAnInterruptedRunThatRecordedItself(t *testing.T) { journals := startedJobJournal + `{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"fail","error_code":"interrupted","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n" - f := reconcileFake(journals, []string{"abc123def456 J1"}) - if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2"); err == nil { + f := reconcileFake(journals, []string{"abc123def456 J1 4"}) + if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4); err == nil { t.Fatal("a recorded interruption must not hide a live container") } } @@ -169,8 +169,8 @@ func TestCloseKeepsARecordedFailureAsAFailure(t *testing.T) { // the separator. Trimming the line before the cut removes the separator, and // the container is skipped as unparseable — the one that most needs refusing. func TestRefuseCatchesAContainerWithAnEmptyOperationLabel(t *testing.T) { - f := reconcileFake("", []string{"abc123def456 "}) - err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2") + f := reconcileFake("", []string{"abc123def456 "}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4) if err == nil { t.Fatal("an unattributable job container must refuse") } @@ -178,3 +178,45 @@ func TestRefuseCatchesAContainerWithAnEmptyOperationLabel(t *testing.T) { t.Fatalf("refusal did not name the problem: %v", err) } } + +// A sealed job plan carries one operation id for its whole life and is +// re-runnable, and AcquireLock hands the lock straight back to a caller +// presenting the id already written in it. Matching the operation alone would +// let a second run exempt the container its own earlier run left behind. +func TestRefuseCatchesAnEarlierRunOfTheSameOperation(t *testing.T) { + f := reconcileFake("", []string{"abc123def456 J1 4"}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 5) + if err == nil { + t.Fatal("an earlier invocation of the same plan must refuse") + } + for _, want := range []string{"earlier run", "J1", "epoch 4", "abc123def456"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("refusal missing %q: %v", want, err) + } + } +} + +// A container carrying no epoch label cannot be shown to belong to this +// invocation, so it is not exempt from it either. +func TestRefuseDoesNotExemptAContainerWithNoEpoch(t *testing.T) { + f := reconcileFake("", []string{"abc123def456 J1 "}) + err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 4) + if err == nil || !strings.Contains(err.Error(), "epoch unknown") { + t.Fatalf("unlabelled epoch = %v, want a refusal naming it", err) + } +} + +// The reconciling operator must not be recorded as the interrupted run's. +// Audit takes the last non-empty operator in an epoch group, so stamping it +// here rewrites the row to name whoever deployed next. +func TestCloseDoesNotAttributeTheRunToTheReconciler(t *testing.T) { + f := reconcileFake(startedJobJournal, nil) + if err := closeAll(t, reconcileEngine(t, f)); err != nil { + t.Fatalf("close: %v", err) + } + for _, c := range f.Commands { + if strings.Contains(c, `"event":"finish"`) && strings.Contains(c, `"operator"`) { + t.Fatalf("terminal record claimed an operator:\n%s", c) + } + } +} diff --git a/internal/engine/job_test.go b/internal/engine/job_test.go index acaf402..f2056de 100644 --- a/internal/engine/job_test.go +++ b/internal/engine/job_test.go @@ -219,3 +219,29 @@ func TestRunJobDoesNotCloseTheRunItIsAboutToStart(t *testing.T) { t.Fatalf("terminal record count = %d, want 1:\n%s", got, commands) } } + +// The refusal is only worth having if it is actually called. Deleting the call +// site left every test green, so this drives the whole run against a host +// reporting a foreign job container and requires it to stop. +func TestRunJobRefusesWhileAForeignJobContainerRuns(t *testing.T) { + const runtime = "services:\n migrate:\n image: ghcr.io/x/app@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n" + target := currentJobFake(runtime) + inner := target.Dynamic + target.Dynamic = func(cmd string) (transport.Result, bool) { + if strings.Contains(cmd, "label='ob.operation'") { + return transport.Result{Stdout: "abc123def456 other-op 2\n"}, true + } + return inner(cmd) + } + engine := manualJobEngine(t, target) + _, _, err := engine.RunJobWithJournalID(context.Background(), JobRunRequest{ + OperationID: "op-job-run", Job: "migrate", ExpectedRelease: engineTestPreviousReleaseID, + ExpectedRuntimeDigest: HashBytes([]byte(runtime)), ExpectedDataEffect: "none", + }) + if err == nil || !strings.Contains(err.Error(), "other-op") { + t.Fatalf("run job = %v, want a refusal naming the foreign operation", err) + } + if strings.Contains(strings.Join(target.Commands, "\n"), "ONEBOX_RESULT_FILE=") { + t.Fatalf("the job ran anyway:\n%s", strings.Join(target.Commands, "\n")) + } +}