diff --git a/internal/engine/audit.go b/internal/engine/audit.go index 104ec00..8dbef94 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 2925fa0..5a61ea5 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 124c26b..a310a2d 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,13 @@ 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 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 resultMode := "600" @@ -200,7 +208,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 +253,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 0dc78a0..b295d34 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 0458a1a..8ffd5de 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,42 @@ 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. + // 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, + } + } + journalErr := writer.Append(journalContext, record) + 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 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) + } + if journalErr != nil { return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr)) } return runErr @@ -121,6 +172,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 +184,40 @@ 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. +// +// 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) +} + +// 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/job_test.go b/internal/engine/job_test.go index 67366f8..88cfd4b 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") + } +} diff --git a/internal/engine/schedule_execution_test.go b/internal/engine/schedule_execution_test.go index c4df1f9..33bbc4c 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) } diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go index 68b4561..ce7ba7e 100644 --- a/internal/onebox/operation_errors.go +++ b/internal/onebox/operation_errors.go @@ -119,6 +119,14 @@ 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. 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", + }, "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 a63e5d6..03012bc 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` |