Skip to content

Commit 2128338

Browse files
authored
feat(jobs): tie a one-off container to its operation, and record an interruption (#180)
* feat(jobs): tie a one-off container to its operation, and record an interruption Three defects that compose into one incident: a sealed manual job run whose client goes away leaves a container the host cannot match to anything, an operation that is INCOMPLETE in `ob audit` forever, and an application lock released out from under a container that is still changing data. `compose run` names nothing and inherits no operation identity, so the host had no way to tell which run a one-off container belonged to. Every refusal and every reconciliation has to key on that, so it comes first: generated and hook-supplied compose runs both now carry ob.operation and ob.epoch. The terminal record was appended on the caller's context, which is precisely the context that is gone when it matters. `ob exec` already writes its own on a bounded background context; the job path now does the same and marks the record `interrupted`, because Append redacts Detail on a failure and the reason has to ride on ErrorCode. `ob audit` reports that as its own outcome, ahead of the failure arm the same record matches: the client went away and the outcome is unknown, which is a different thing to tell an operator than "it failed". ReleaseLock runs on its own background context, so on Ctrl-C it succeeded while the journal append on the cancelled context did not — ownership dropped immediately and silently, over a live container, before the TTL could matter. The lock is now held when an interrupted run leaves this operation's container running, and the operator is told what holds it and how to look. An unreadable answer counts as running: keeping the lock over a container that has exited costs one `--break-lock`, releasing it over one that has not costs concurrent writers. This does not make execution durable. A job whose client is killed outright still records nothing, and reconciling what a container did after the fact is the next step. It makes every new run reconcilable, which nothing was before. Refs #179. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * fix(jobs): enumerate the interrupted error code The code reaches an operator through `ob audit` but was not in the registry, so the generated error reference could not document it. The enumeration test missed it because it matches the struct-literal form and the assignment was a tuple; written as separate statements it is now covered by that test too. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * fix(jobs): record a job that finished just before the client went away The terminal append switched to a background context only when the run itself was classified interrupted, which conflated two independent questions. WHERE to append is decided by the caller's context: once it is cancelled it cannot carry the write, whatever the run did. WHAT to record is decided by what the run did: a job that completed cleanly a moment before Ctrl-C produced its own outcome and is not interrupted. Conflating them meant a successful job, cancelled in the window between its result and its finish record, was written down as a failure — or, if the classification did not catch the cancellation, was not written down at all. Also corrects the registry comment for the code, which described reconciliation as its only writer. The interrupted run writes it itself whenever it still can. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * fix(jobs): classify the run, not the client The previous commit split where the terminal record is appended from what it says, but left the classification itself keyed on the caller's context. A job that finished cleanly and was then cancelled in the window before its record was still written down as interrupted — the very case that split was meant to fix. A run that produced no error finished, whatever became of the client afterwards. Only once the run failed does a gone context mean the client is why, and that still counts even when the transport reports something other than a cancelled context, which it usually does. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * fix(jobs): retry the terminal append when cancellation lands mid-write The append switched to a context of its own only when the caller's was already gone. Cancellation during the write was left with a failed append and no second attempt, which is the same permanently incomplete operation by a narrower path. One retry closes it. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * docs(jobs): describe only what this change does for the interrupted code The registry comment named a reconciler that does not exist here, so a reader of this change alone would look for behaviour it does not contain. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2 * fix(jobs): retry the terminal append only when the context died mid-write The retry fired on any append failure. A write that failed for another reason — a full disk, a refused write — may still have landed on the host, and appending a second terminal record over it is worse than reporting the first failure. It now retries only when the caller's context died during the write, which is the case it was added for. Also corrects the label comment for a hook that is not a compose run: it gets no label because there is no container to put one on, and its result is unresolvable for the same reason rather than by a check above. Claude-Session: https://claude.ai/code/session_01JaxHfqFZk8GdrBNbtQZ6c2
1 parent 7a0aab8 commit 2128338

9 files changed

Lines changed: 245 additions & 13 deletions

File tree

‎internal/engine/audit.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,12 @@ func auditRows(recs []journal.Record) []auditRow {
170170
row.action = auditAction(r.Phase)
171171
}
172172
switch {
173+
// Ahead of the failure arm, which this record also matches: an
174+
// interrupted run is not a failure of the job. The client went away
175+
// and the outcome is unknown, which is a different thing to tell an
176+
// operator than "it failed" or "it never finished".
177+
case r.ErrorCode == "interrupted":
178+
row.outcome = "interrupted"
173179
case r.Event == "abort":
174180
row.outcome = "aborted"
175181
case r.Status == "fail":

‎internal/engine/audit_test.go‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,3 +129,29 @@ func TestAuditReportsFailedJobRun(t *testing.T) {
129129
t.Fatalf("failed job run audit = %+v", records)
130130
}
131131
}
132+
133+
// An interrupted run is neither a job failure nor incomplete-forever: the
134+
// client went away and the outcome is unknown. The record carries Status fail,
135+
// so it also matches the failure arm and the ordering is what distinguishes it.
136+
func TestAuditDistinguishesAnInterruptedJobRun(t *testing.T) {
137+
f := &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) {
138+
switch {
139+
case strings.Contains(cmd, "ls -1"):
140+
return transport.Result{Stdout: "job-3.jsonl\n"}, true
141+
case strings.Contains(cmd, "job-3.jsonl"):
142+
return transport.Result{Stdout: journalLines(
143+
journal.Record{DeployID: "job-3", Phase: "job", Event: "start", Status: "ok", OperationKind: "job_run", Service: "catalog-refresh", Operator: "v@mac", TS: "t1"},
144+
journal.Record{DeployID: "job-3", Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", OperationKind: "job_run", Service: "catalog-refresh"},
145+
)}, true
146+
}
147+
return transport.Result{}, false
148+
}}
149+
e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
150+
records, err := e.AuditSnapshot(context.Background(), 10)
151+
if err != nil {
152+
t.Fatal(err)
153+
}
154+
if len(records) != 1 || records[0].Outcome != "interrupted" || records[0].Service != "catalog-refresh" {
155+
t.Fatalf("interrupted job audit = %+v", records)
156+
}
157+
}

‎internal/engine/gate.go‎

Lines changed: 48 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"errors"
66
"fmt"
7+
"strconv"
78
"strings"
89

910
"github.com/labstack/onebox/internal/app"
@@ -73,7 +74,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s
7374
return fmt.Errorf("journal %s intent: %w", key, err)
7475
}
7576
st := e.ui.Step("job "+job, true)
76-
safe, detail, err := e.runOneJob(ctx, job, remoteDir, remoteCompose)
77+
safe, detail, err := e.runOneJob(ctx, jw.DeployID, jw.Epoch, job, remoteDir, remoteCompose)
7778
if err == nil {
7879
e.logf("job %s: %s", job, detail)
7980
}
@@ -113,7 +114,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s
113114

114115
// runOneJob runs a single gate step and reports whether it declared itself
115116
// rollback-safe (changed=false). Returns (safe, detail, err).
116-
func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose string) (bool, string, error) {
117+
func (e *Engine) runOneJob(ctx context.Context, operationID string, epoch int, job, remoteDir, remoteCompose string) (bool, string, error) {
117118
safeByDeclaration := e.jobDataEffect(job) == app.DataEffectNone
118119
if !safeByDeclaration {
119120
res, err := e.mutate(ctx, invalidateExecutionCommand(e.names().AppDir()))
@@ -128,7 +129,7 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st
128129
resultFile := resultDir + "/result"
129130
const containerResultFile = "/run/onebox/job-result"
130131
containerized := true
131-
runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" +
132+
runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" + jobRunLabels(operationID, epoch) +
132133
" -e ONEBOX_RESULT_FILE=" + containerResultFile +
133134
" -v " + q(resultFile+":"+containerResultFile+":rw") + " " + job
134135
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
147148
var injected bool
148149
runCmd, injected = injectComposeJobResult(runCmd, resultFile, containerResultFile)
149150
containerized = injected
151+
// A hook that runs its own compose command produces a container this
152+
// operation owns just as much as the generated one, so it carries the
153+
// same identity. A hook that is not a compose run gets no label because
154+
// there is no container to put one on; that hook is also not
155+
// containerized, so its result is unresolvable and reported as such
156+
// below.
157+
runCmd, _ = injectComposeJobLabels(runCmd, operationID, epoch)
150158
}
151159
e.ui.Cmd("job", runCmd) // verbose only — the plan lists it
152160
resultMode := "600"
@@ -200,7 +208,43 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st
200208
return !evidence.Changed, jobResultDetail(evidence), nil
201209
}
202210

211+
// jobRunLabels ties a one-off container back to the operation that started it.
212+
// `compose run` names nothing and inherits no operation identity, so without
213+
// these an interrupted run leaves a container on the host that nothing can
214+
// match to a journal — every refusal and every reconciliation keys on them.
215+
func jobRunLabels(operationID string, epoch int) string {
216+
if operationID == "" {
217+
return ""
218+
}
219+
return " --label " + q(JobOperationLabel+"="+operationID) +
220+
" --label " + q(JobEpochLabel+"="+strconv.Itoa(epoch))
221+
}
222+
223+
const (
224+
// JobOperationLabel carries the operation id of the run that created a
225+
// one-off job container.
226+
JobOperationLabel = "ob.operation"
227+
// JobEpochLabel carries the lock epoch that run held.
228+
JobEpochLabel = "ob.epoch"
229+
)
230+
231+
func injectComposeJobLabels(command, operationID string, epoch int) (string, bool) {
232+
labels := jobRunLabels(operationID, epoch)
233+
if labels == "" {
234+
return command, false
235+
}
236+
return injectComposeRunFlags(command, labels+" ")
237+
}
238+
203239
func injectComposeJobResult(command, hostResultFile, containerResultFile string) (string, bool) {
240+
return injectComposeRunFlags(command,
241+
" -e "+"ONEBOX_RESULT_FILE="+containerResultFile+
242+
" -v "+q(hostResultFile+":"+containerResultFile+":rw")+" ")
243+
}
244+
245+
// injectComposeRunFlags splices flags into a hook's own `docker compose run`.
246+
// Anything that is not a compose run is left alone and reported as such.
247+
func injectComposeRunFlags(command, flags string) (string, bool) {
204248
runIndex := strings.Index(command, " run ")
205249
if runIndex < 0 {
206250
return command, false
@@ -209,9 +253,7 @@ func injectComposeJobResult(command, hostResultFile, containerResultFile string)
209253
if !strings.Contains(prefix, "docker compose") && !strings.Contains(prefix, "docker-compose") {
210254
return command, false
211255
}
212-
flags := " run -e ONEBOX_RESULT_FILE=" + containerResultFile +
213-
" -v " + q(hostResultFile+":"+containerResultFile+":rw") + " "
214-
return prefix + flags + command[runIndex+len(" run "):], true
256+
return prefix + " run" + flags + command[runIndex+len(" run "):], true
215257
}
216258

217259
func (e *Engine) unknownJobResult(job, reason string) (bool, string, error) {

‎internal/engine/gate_test.go‎

Lines changed: 39 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ func TestJobAutoRunsWithoutHook(t *testing.T) {
142142
t.Fatalf("deploy: %v", err)
143143
}
144144
seq := strings.Join(f.Commands, "\n")
145-
if !strings.Contains(seq, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") {
145+
if !strings.Contains(seq, "run --rm --no-deps") || !strings.Contains(seq, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") {
146146
t.Fatalf("a job without a hook must auto-run compose run:\n%s", seq)
147147
}
148148
// gate protocol still applies to the auto-run job.
@@ -266,7 +266,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) {
266266
t.Run("no result declaration", func(t *testing.T) {
267267
f := happyFake()
268268
e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
269-
safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml")
269+
safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml")
270270
if err != nil {
271271
t.Fatalf("run job: %v", err)
272272
}
@@ -282,7 +282,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) {
282282
e := New(cfg, testProject(t), happyFake(), Options{
283283
Out: &bytes.Buffer{}, Sleep: noSleep, LocalDir: t.TempDir(),
284284
})
285-
safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml")
285+
safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml")
286286
if err != nil {
287287
t.Fatalf("run local job: %v", err)
288288
}
@@ -419,7 +419,7 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) {
419419
mount := strings.Index(c, "-v '"+resultFile+":/run/onebox/job-result:rw'")
420420
sealedFile := strings.Index(c, "chmod 600 '"+resultFile+"'")
421421
if strings.Contains(c, "rm -rf '"+resultDir+"'") &&
422-
strings.Contains(c, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") &&
422+
strings.Contains(c, "run --rm --no-deps") && strings.Contains(c, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") &&
423423
privateDir >= 0 && privateDir < writableFile && writableFile < mount && mount < sealedFile {
424424
found = true
425425
}
@@ -428,3 +428,38 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) {
428428
t.Fatalf("migrate container must receive a privately staged, writable, subsequently sealed result file:\n%s", strings.Join(f.Commands, "\n"))
429429
}
430430
}
431+
432+
// Without an operation label nothing on the host ties a running one-off
433+
// container back to the journal that started it, so an interrupted run cannot
434+
// be refused or reconciled — only guessed at.
435+
func TestJobContainerCarriesItsOperationIdentity(t *testing.T) {
436+
f := happyFake()
437+
e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
438+
if _, _, err := e.runOneJob(context.Background(), "20260909-053225-abc-job_run-deadbeef", 7, "migrate", "/remote", "/remote/compose.yaml"); err != nil {
439+
t.Fatal(err)
440+
}
441+
seq := strings.Join(f.Commands, "\n")
442+
for _, want := range []string{
443+
"--label 'ob.operation=20260909-053225-abc-job_run-deadbeef'",
444+
"--label 'ob.epoch=7'",
445+
} {
446+
if !strings.Contains(seq, want) {
447+
t.Fatalf("job container missing %s:\n%s", want, seq)
448+
}
449+
}
450+
}
451+
452+
func TestInjectComposeJobLabelsOnlyTouchesAComposeRun(t *testing.T) {
453+
got, ok := injectComposeJobLabels("docker compose -f x.yml run --rm migrate", "op-1", 2)
454+
if !ok || !strings.Contains(got, "--label 'ob.operation=op-1'") || !strings.Contains(got, "--label 'ob.epoch=2'") {
455+
t.Fatalf("compose run = %q ok=%v", got, ok)
456+
}
457+
// A hook that is not a compose run has no container to label.
458+
if got, ok := injectComposeJobLabels("/usr/local/bin/migrate.sh", "op-1", 2); ok || got != "/usr/local/bin/migrate.sh" {
459+
t.Fatalf("non-compose hook = %q ok=%v", got, ok)
460+
}
461+
// No operation identity, nothing to add.
462+
if got, ok := injectComposeJobLabels("docker compose run --rm migrate", "", 0); ok || got != "docker compose run --rm migrate" {
463+
t.Fatalf("empty operation = %q ok=%v", got, ok)
464+
}
465+
}

‎internal/engine/job.go‎

Lines changed: 95 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"errors"
66
"fmt"
77
"strings"
8+
"time"
89

910
"github.com/labstack/onebox/internal/app"
1011
"github.com/labstack/onebox/internal/journal"
@@ -43,7 +44,22 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
4344
if err != nil {
4445
return operationID, nil, err
4546
}
46-
defer e.ReleaseLock(ctx)
47+
// Released unless an interrupted run left this operation's container alive.
48+
// ReleaseLock runs on its own background context, so on Ctrl-C it succeeds
49+
// while the terminal journal append — which uses the cancelled one — does
50+
// not: ownership would be dropped, immediately and silently, over a
51+
// container still changing data.
52+
holdLockForLiveContainer := false
53+
defer func() {
54+
if holdLockForLiveContainer {
55+
e.warnf("operation %s was interrupted while its container is still running; "+
56+
"keeping the application lock so nothing else mutates alongside it. "+
57+
"Inspect with `docker ps --filter label=%s=%s`; the lock expires on its own after %s",
58+
operationID, JobOperationLabel, operationID, e.lockTTL())
59+
return
60+
}
61+
e.ReleaseLock(ctx)
62+
}()
4763
if err := e.WriteFence(ctx, operationID, epoch); err != nil {
4864
return operationID, nil, err
4965
}
@@ -93,7 +109,42 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
93109
record.Status = "fail"
94110
record.Detail = runErr.Error()
95111
}
96-
if journalErr := writer.Append(ctx, record); journalErr != nil {
112+
// A cancelled context is exactly when the terminal record matters most,
113+
// and exactly when appending on that context cannot work. `ob exec`
114+
// already writes its own on a bounded background context; without the
115+
// same here an interrupted job stays INCOMPLETE in `ob audit` forever,
116+
// with no record that it was ever interrupted. Append redacts Detail on
117+
// a failure, so the reason has to ride on ErrorCode.
118+
// Two independent questions. WHERE to append: a cancelled caller context
119+
// cannot carry the write, whatever the run did, so a job that finished
120+
// cleanly a moment before Ctrl-C still records its success. WHAT to
121+
// record: only a run that ended because the client went away is
122+
// interrupted — an outcome the job itself produced is its own.
123+
journalContext := ctx
124+
if ctx.Err() != nil {
125+
var cancel context.CancelFunc
126+
journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout)
127+
defer cancel()
128+
}
129+
if interruptedRun(ctx, runErr) {
130+
record = journal.Record{
131+
Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted",
132+
OperationKind: "job_run", Service: job,
133+
}
134+
}
135+
journalErr := writer.Append(journalContext, record)
136+
if journalErr != nil && journalContext == ctx && ctx.Err() != nil {
137+
// Cancellation can land during the write as easily as before it, and
138+
// the check above only sees a context that was already gone. Retried
139+
// only when the context died in the meantime: any other failure —
140+
// a full disk, a refused write — may have landed on the host after
141+
// reporting an error, and appending a second terminal record is
142+
// worse than reporting the first failure.
143+
retryContext, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout)
144+
defer cancel()
145+
journalErr = writer.Append(retryContext, record)
146+
}
147+
if journalErr != nil {
97148
return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr))
98149
}
99150
return runErr
@@ -121,10 +172,52 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
121172
e.gateOpen = true
122173
e.rollbackCovered = true
123174
runErr := e.runJobPhase(ctx, writer, nil, remoteDir, remoteCompose, "job", []string{job})
175+
if interruptedRun(ctx, runErr) {
176+
// Cancelling the client kills at most the wrapper shell; the container
177+
// belongs to the daemon and keeps running.
178+
holdLockForLiveContainer = e.jobContainerRunning(operationID)
179+
}
124180
var result *journal.JobResultEvidence
125181
if evidence, ok := e.jobResults[job]; ok {
126182
resultCopy := evidence
127183
result = &resultCopy
128184
}
129185
return operationID, result, finish(runErr)
130186
}
187+
188+
const journalCleanupTimeout = 5 * time.Second
189+
190+
// interruptedRun reports a run that ended because the client went away rather
191+
// than because the job finished.
192+
//
193+
// A run that produced no error finished, whatever became of the client
194+
// afterwards — its outcome is its own and must be recorded as such. Only once
195+
// the run failed does a gone context mean the client is why. The transport does
196+
// not always surface a cancelled context as context.Canceled, so a failure
197+
// under a cancelled context counts even when the error says something else.
198+
func interruptedRun(ctx context.Context, runErr error) bool {
199+
if runErr == nil {
200+
return false
201+
}
202+
return ctx.Err() != nil ||
203+
errors.Is(runErr, context.Canceled) ||
204+
errors.Is(runErr, context.DeadlineExceeded)
205+
}
206+
207+
// jobContainerRunning answers whether this operation's one-off container is
208+
// still alive, on a context of its own because the caller's is already gone.
209+
// An unreadable answer is reported as running: keeping the lock over a
210+
// container that has in fact exited costs an operator one `--break-lock`, while
211+
// releasing it over one that has not costs them concurrent writers.
212+
func (e *Engine) jobContainerRunning(operationID string) bool {
213+
ctx, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout)
214+
defer cancel()
215+
res, err := e.T.Run(ctx, "docker ps -q --filter label="+q(JobOperationLabel+"="+operationID))
216+
if err != nil {
217+
return true
218+
}
219+
if res.ExitCode != 0 {
220+
return true
221+
}
222+
return strings.TrimSpace(res.Stdout) != ""
223+
}

‎internal/engine/job_test.go‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package engine
33
import (
44
"bytes"
55
"context"
6+
"errors"
67
"strings"
78
"testing"
89

@@ -151,3 +152,23 @@ func TestRunJobUsesPlanIdentityAndJournalsAuthorization(t *testing.T) {
151152
}
152153
}
153154
}
155+
156+
// A job that finished cleanly is not interrupted, whatever became of the client
157+
// in the window before its terminal record. Classification follows the run;
158+
// only the choice of append context follows the caller's context.
159+
func TestInterruptedRunClassifiesTheRunNotTheClient(t *testing.T) {
160+
cancelled, cancel := context.WithCancel(context.Background())
161+
cancel()
162+
if interruptedRun(cancelled, nil) {
163+
t.Fatal("a run that produced no error must never be recorded interrupted")
164+
}
165+
if !interruptedRun(cancelled, errors.New("ssh: session closed")) {
166+
t.Fatal("a failed run under a cancelled context is interrupted, whatever the transport called it")
167+
}
168+
if !interruptedRun(context.Background(), context.Canceled) {
169+
t.Fatal("a cancellation surfaced by the run itself is interrupted")
170+
}
171+
if interruptedRun(context.Background(), errors.New("migrate: exit 1")) {
172+
t.Fatal("a job that failed on its own terms is not interrupted")
173+
}
174+
}

‎internal/engine/schedule_execution_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ func TestDurableCompatibilityInvalidationMustSucceedBeforeDataChangingJob(t *tes
150150
return base(command)
151151
}
152152
e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
153-
_, _, err := e.runOneJob(context.Background(), "change", "/release", "/release/compose.yaml")
153+
_, _, err := e.runOneJob(context.Background(), "op-1", 1, "change", "/release", "/release/compose.yaml")
154154
if !invalidated || err == nil || !strings.Contains(err.Error(), "invalidate durable execution compatibility") {
155155
t.Fatalf("failed invalidation did not stop data-changing job: invalidated=%t err=%v", invalidated, err)
156156
}

0 commit comments

Comments
 (0)