feat(jobs): refuse to mutate alongside another operation's job container

internal/engine/deploy.go

@@ -75,7 +75,19 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri
 	if err != nil {
 		return err
 	}
-	defer e.ReleaseLock(ctx)
+	// A non-empty reason keeps the lock and says why. The check below refuses
+	// both when a job container is running and when the host could not be
+	// asked, and the lock is kept for the same reason either way: releasing it
+	// would hand the host to the next mutator over a state this deploy declined
+	// to proceed against.
+	holdLockReason := ""
+	defer func() {
+		if holdLockReason != "" {
+			e.warnf("%s", holdLockReason)
+			return
+		}
+		e.ReleaseLock(ctx)
+	}()
 	if err := e.WriteFence(ctx, releaseID, epoch); err != nil {
 		return err
 	}
@@ -105,6 +117,16 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri
 	if err := e.requireServingApplicationManifest(ctx, prev); err != nil {
 		return err
 	}
+	// After preflight, so an unreachable daemon is reported by the check that
+	// exists for it rather than by a raw `docker ps` failure — and still before
+	// any workload is rolled or any gate job runs.
+	if err := e.refuseForeignJobContainers(ctx, releaseID, epoch); err != nil {
+		holdLockReason = fmt.Sprintf(
+			"nothing was deployed: %v. The application lock is being kept until this is "+
+				"resolved, so nothing else mutates meanwhile; it expires on its own after %s",
+			err, e.lockTTL())
+		return err
+	}
 	rollbackDebt := false
 	if done == nil {
 		rollbackDebt, err = e.rollbackEffectDebt(ctx, prev)

internal/engine/deploy_test.go

@@ -563,3 +563,52 @@ func TestRollbackReplaysPreviousRelease(t *testing.T) {
 		t.Fatalf("rollback must re-activate previous:\n%s", seq)
 	}
 }
+
+// A deploy rolls workloads and runs its own gate jobs. An orphaned job
+// container still changing data underneath it is the overlap the lock exists to
+// prevent, and the lock does not catch it once its holder is gone.
+func TestDeployRefusesWhileAForeignJobContainerRuns(t *testing.T) {
+	f := happyFake()
+	inner := f.Dynamic
+	f.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)
+	}
+	e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
+	err := e.Deploy(context.Background(), "20260101-000000-aaa111", t.TempDir())
+	if err == nil || !strings.Contains(err.Error(), "other-op") {
+		t.Fatalf("deploy = %v, want a refusal naming the foreign operation", err)
+	}
+	// Refused before anything is rolled or any gate job runs.
+	if seq := strings.Join(f.Commands, "\n"); strings.Contains(seq, "--scale web=") {
+		t.Fatalf("the deploy rolled anyway:\n%s", seq)
+	}
+}
+
+// A refused deploy keeps the application lock, and has to say so: an operator
+// who is told only that the deploy stopped will not know the host is still held.
+func TestDeployKeepsAndExplainsTheLockWhenItRefuses(t *testing.T) {
+	f := happyFake()
+	inner := f.Dynamic
+	f.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)
+	}
+	var out bytes.Buffer
+	e := New(testConfig(), testProject(t), f, Options{Out: &out, Sleep: noSleep})
+	if err := e.Deploy(context.Background(), "20260101-000000-aaa111", t.TempDir()); err == nil {
+		t.Fatal("expected a refusal")
+	}
+	for _, c := range f.Commands {
+		if strings.Contains(c, "rm -f") && strings.Contains(c, "/lock") {
+			t.Fatalf("the lock was released over a live container:\n%s", c)
+		}
+	}
+	if s := out.String(); !strings.Contains(s, "lock is being kept") {
+		t.Fatalf("the operator was not told the lock is held:\n%s", s)
+	}
+}

internal/engine/job.go

@@ -49,13 +49,13 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	// 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
+	// A non-empty reason keeps the lock and says why. Two different situations
+	// hold it, and telling an operator the wrong one sends them looking for a
+	// run that never started.
+	holdLockReason := ""
 	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())
+		if holdLockReason != "" {
+			e.warnf("%s", holdLockReason)
 			return
 		}
 		e.ReleaseLock(ctx)
@@ -86,6 +86,25 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 		return operationID, nil, errors.New("job plan is stale: current release runtime changed — re-plan")
 	}
 
+	// After the staleness checks, so a stale plan is told it is stale rather
+	// than told about a container, and before this run creates one of its own.
+	if err := e.refuseForeignJobContainers(ctx, operationID, epoch); err != nil {
+		// Keep the lock. Releasing it here would hand the host to the next
+		// mutator over a container this check has just established is alive —
+		// the opposite of what refusing is for, and worse than not refusing,
+		// because the lock reclaimed from the interrupted run would be gone too.
+		// The sentence around the error asserts nothing about what was found:
+		// this refuses both when a job container is running and when the host
+		// could not be asked, and the lock is kept for the same reason either
+		// way — an unanswered question is not an answer of no. What was
+		// actually determined travels in the error itself.
+		holdLockReason = fmt.Sprintf(
+			"nothing was run: %v. The application lock is being kept until this is "+
+				"resolved, so nothing else mutates meanwhile; it expires on its own after %s",
+			err, e.lockTTL())
+		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,
@@ -175,7 +194,13 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	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)
+		if e.jobContainerRunning(operationID) {
+			holdLockReason = fmt.Sprintf(
+				"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())
+		}
 	}
 	var result *journal.JobResultEvidence
 	if evidence, ok := e.jobResults[job]; ok {

internal/engine/job_containers.go

@@ -0,0 +1,110 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// 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, currentEpoch int) error {
+	containers, err := e.jobContainers(ctx)
+	if err != nil {
+		return err
+	}
+	currentEpochLabel := strconv.Itoa(currentEpoch)
+	for _, c := range containers {
+		// 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 {
+			// A differing epoch says the container belongs to some other
+			// invocation of this operation, not which one or when — epochs are
+			// not ordered against each other here — and a missing epoch says
+			// only that it cannot be placed at all. Neither supports calling it
+			// an earlier run.
+			if c.epoch == "" {
+				return fmt.Errorf(
+					"a job container of operation %s is running on this host (%.12s) carrying no %s label, "+
+						"so it cannot be placed against this run; establish what it did and stop it with "+
+						"`docker rm -f %s`",
+					c.operation, c.id, JobEpochLabel, c.id)
+			}
+			return fmt.Errorf(
+				"another invocation of operation %s (epoch %s, this run is 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 %s`",
+				c.operation, c.epoch, currentEpochLabel, 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
+			// 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 %s`",
+				c.id, JobOperationLabel, c.id)
+		}
+		return fmt.Errorf(
+			"a job container from operation %s is still running on this host (%.12s); "+
+				"if that operation is still in progress, wait for it — otherwise establish what it did "+
+				"and stop it with `docker rm -f %s`",
+			c.operation, c.id, c.id)
+	}
+	return nil
+}
+
+type jobContainer struct {
+	id        string
+	operation string
+	epoch     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+"\"}} {{.Label \""+JobEpochLabel+"\"}}"))
+	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(res.Stdout, "\n") {
+		// Each field is cut and trimmed on its own, and Cut yields empty for a
+		// separator that is not there — so a label docker could not resolve
+		// parses as empty whether or not its separator survives. What matters is
+		// that the id parses: a line that yields none is dropped below, and a
+		// dropped line is a container nobody can see.
+		id, rest, _ := strings.Cut(strings.TrimSpace(line), " ")
+		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, epoch: epoch})
+	}
+	return out, nil
+}