feat(jobs): reconcile job runs whose client never recorded an outcome

internal/engine/deploy.go

@@ -81,6 +81,13 @@ 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; 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, epoch); 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.
@@ -89,6 +96,16 @@ 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. 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)
 	if err := e.preflight(ctx, false); err != nil {
 		pf(err)
@@ -107,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)
 		}
@@ -214,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])

internal/engine/job.go

@@ -66,6 +66,13 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	stopHeartbeat := e.StartHeartbeat(ctx)
 	defer stopHeartbeat()
 
+	// 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, epoch); err != nil {
+		return operationID, nil, err
+	}
+
 	current, err := release.Current(ctx, e.T, e.names())
 	if err != nil {
 		return operationID, nil, err
@@ -86,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,

internal/engine/job_reconcile.go

@@ -0,0 +1,211 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/labstack/onebox/internal/journal"
+)
+
+// 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 {
+			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
+			// 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`",
+			c.operation, c.id, c.id)
+	}
+	return nil
+}
+
+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
+// 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") {
+		// Not TrimSpace before the cut: a container whose label carries no value
+		// prints "<id> " 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, 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, epoch: epoch})
+	}
+	return out, nil
+}
+
+// closeInterruptedJobRuns writes the terminal record an interrupted client
+// could not, so an operation stops being incomplete forever.
+//
+// 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.
+// 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 {
+				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
+		}
+		s, seen := byEpoch[r.Epoch]
+		if !seen {
+			s = &state{run: unfinishedJobRun{Epoch: r.Epoch}}
+			byEpoch[r.Epoch], order = s, append(order, r.Epoch)
+		}
+		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"
+		}
+	}
+	var out []unfinishedJobRun
+	for _, epoch := range order {
+		if s := byEpoch[epoch]; s.started && !s.finished {
+			out = append(out, s.run)
+		}
+	}
+	return out
+}
+
+// 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: run.Job,
+	}
+	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.
+	// 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,
+	}
+	if err := writer.Append(ctx, record); err != nil {
+		return fmt.Errorf("close interrupted job run %s: %w", operationID, err)
+	}
+	e.logf("closed interrupted job run %s (%s) as %s", operationID, run.Job, record.Status)
+	return nil
+}