Merge pull request 'Never lose the bus to a replace of its own container (issue 1945)' (#138) from fix/r6-bus-replace into main

cmd/mesh-host/apply_context_test.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// An apply runs in a context the bus's loss cannot end: the link's context ends when the bus goes, and the bus may go
+// because of the apply itself (it replaces the bus's own container), which then must finish (issue 1945, live on
+// 2026-10-25 at 11:23: novox's node-engine removed the bus container, its link's context ended, and the docker run that
+// would have made the new one was never made).
+func TestAnApplyOutlivesTheLinksContext(t *testing.T) {
+	type key struct{}
+	link, cancel := context.WithCancel(context.WithValue(context.Background(), key{}, "v"))
+	ctx, done := applyContext(link)
+	defer done()
+	cancel()
+	if ctx.Err() != nil {
+		t.Fatalf("the apply's context ended with the link's: %v", ctx.Err())
+	}
+	if ctx.Value(key{}) != "v" {
+		t.Fatal("the apply's context lost the link's values")
+	}
+}
+
+// An apply is bounded all the same: a step that hangs (a docker call that never returns) cannot hold the node agent for
+// ever (the review of issue 1945).
+func TestAnApplyIsBounded(t *testing.T) {
+	ctx, done := applyContext(context.Background())
+	defer done()
+	at, ok := ctx.Deadline()
+	if !ok {
+		t.Fatal("an apply has no bound")
+	}
+	if left := time.Until(at); left > applyBound || left < applyBound-time.Minute {
+		t.Fatalf("the bound is %s away, want about %s", left, applyBound)
+	}
+}

cmd/mesh-host/nodeagent.go

@@ -298,13 +298,17 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
 			Backup:   backupNode(session),
 			Say:      say,
 			Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
-				res := applyComposition(session.WithLink(ctx), opts, c, others, sched, say)
+				actx, done := applyContext(session.WithLink(ctx))
+				res := applyComposition(actx, opts, c, others, sched, say)
+				done()
 				afterApply()
 				sayFacts(ctx, ag)
 				return res
 			},
 			Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result {
-				res := removeComposition(session.WithLink(ctx), opts, module, others, sched, say)
+				actx, done := applyContext(session.WithLink(ctx))
+				res := removeComposition(actx, opts, module, others, sched, say)
+				done()
 				afterApply()
 				sayFacts(ctx, ag)
 				return res
@@ -567,3 +571,15 @@ func declaresNothingToBackUp(said json.RawMessage, module string) (nothing bool,
 	}
 	return true, true
 }
+
+// applyBound is how long one apply or removal may take: long enough for the slowest image pull and backup step, short
+// enough that a step that hangs cannot hold the node agent for ever.
+const applyBound = 30 * time.Minute
+
+// applyContext is the context an apply or a removal runs in: the link's values, never its end, and a bound of its own.
+// The link's context ends when the bus goes, and the bus may go because of the apply itself (it replaces the bus's own
+// container), which must then finish: a container removed and never made again took the mesh down (issue 1945, live on
+// 2026-10-25 at 11:23). applyBound keeps a hung step from blocking the agent for ever.
+func applyContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.WithoutCancel(ctx), applyBound)
+}

internal/apply/apply.go

@@ -2464,12 +2464,18 @@ func ExecRunner(ctx context.Context, name string, args ...string) (string, error
 	if err != nil {
 		var exit *exec.ExitError
 		if errors.As(err, &exit) {
-			// The words as they always were, and the exit underneath them, so a caller asks the
+			words := fmt.Sprintf("%s exited %d: %s", name, exit.ExitCode(), strings.TrimSpace(string(exit.Stderr)))
-			// code (system.ExitCode) rather than matching text that this line is free to reword.
+			if exit.ExitCode() == -1 {
-			return string(out), &exited{
+				// Killed, not exited: say by what, and the context's end when that is what killed it (issue 1945:
-				words: fmt.Sprintf("%s exited %d: %s", name, exit.ExitCode(), strings.TrimSpace(string(exit.Stderr))),
+				// "docker exited -1:" said nothing while a bus container was half replaced).
-				exit:  exit,
+				words = fmt.Sprintf("%s was %s", name, exit.String())
+				if ctx.Err() != nil {
+					words += " because its context ended: " + ctx.Err().Error()
+				}
 			}
+			// The words, and the exit underneath them, so a caller asks the code (system.ExitCode) rather than
+			// matching text that this line is free to reword.
+			return string(out), &exited{words: words, exit: exit}
 		}
 		return string(out), fmt.Errorf("%s: %w", name, err)
 	}
@@ -2958,6 +2964,14 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner,
 		return out, nil
 	}
 
+	aside := ""
+	if !existed {
+		// A container an engine set aside and then stopped before it ran the replacement (it died between the two):
+		// the set-aside one of this apply, put back if the replacement fails and removed once it runs (issue 1945).
+		if _, err := containerState(ctx, r.Name+asideSuffix, run); err == nil {
+			aside = r.Name + asideSuffix
+		}
+	}
 	switch {
 	case existed && (before.Spec == want || legacy) && before.Running && len(reasons) == 0:
 		out.Action = "unchanged"
@@ -2980,15 +2994,40 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner,
 		if err := ensureImage(ctx, cri, r.Image, run); err != nil {
 			return out, fmt.Errorf("not replacing container %s, which keeps running: %w", r.Name, err)
 		}
-		if _, err := run(ctx, cri, "rm", "-f", r.Name); err != nil {
+		// **The running container is set aside, never removed, until its replacement runs** (issue 1945): stopped and
-			return out, fmt.Errorf("replacing container %s: %w", r.Name, err)
+		// renamed, so its name and ports are free, and started again under its own name when the replacement does not
+		// come up. Removed first, the bus's own container left a node without a bus when its replacement was never
+		// made (2026-10-25 at 11:23). A container set aside by an earlier apply that stopped midway goes first.
+		aside = r.Name + asideSuffix
+		_, _ = run(ctx, cri, "rm", "-f", aside)
+		if _, err := run(ctx, cri, "rename", r.Name, aside); err != nil {
+			return out, fmt.Errorf("replacing container %s: setting the running one aside: %w", r.Name, err)
 		}
+		if _, err := run(ctx, cri, "stop", aside); err != nil {
+			_, _ = run(ctx, cri, "rename", aside, r.Name)
+			return out, fmt.Errorf("replacing container %s: stopping the running one: %w", r.Name, err)
+		}
+	}
+
+	// putBack starts the container set aside again under its own name, after a replacement that did not come up.
+	putBack := func(cause error) error {
+		if aside == "" {
+			return cause
+		}
+		_, _ = run(ctx, cri, "rm", "-f", r.Name)
+		if _, err := run(ctx, cri, "rename", aside, r.Name); err != nil {
+			return fmt.Errorf("%w; and the container it replaced could not be put back: %v", cause, err)
+		}
+		if _, err := run(ctx, cri, "start", r.Name); err != nil {
+			return fmt.Errorf("%w; and the container it replaced could not be started again: %v", cause, err)
+		}
+		return fmt.Errorf("%w; the container it replaced was started again", cause)
 	}
 
 	args := detachedRunArgs(r, want)
 
 	if _, err := run(ctx, cri, args...); err != nil {
-		return out, fmt.Errorf("starting container %s: %w", r.Name, err)
+		return out, putBack(fmt.Errorf("starting container %s: %w", r.Name, err))
 	}
 
 	// Read back. `docker run --detach` returning an id says the container was created, not
@@ -2996,15 +3035,21 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner,
 	// command exactly as one that came up does.
 	after, err := containerState(ctx, r.Name, run)
 	if err != nil {
-		return out, fmt.Errorf("started container %s and it is not there: %w", r.Name, err)
+		return out, putBack(fmt.Errorf("started container %s and it is not there: %w", r.Name, err))
 	}
 	if !after.Running {
-		return out, fmt.Errorf(
+		return out, putBack(fmt.Errorf(
 			"container %s was started and is not running. It exited; ask the runtime for its "+
-				"logs", r.Name)
+				"logs", r.Name))
 	}
 	if after.Spec != want {
-		return out, fmt.Errorf("container %s is not the one that was declared after creating it", r.Name)
+		return out, putBack(fmt.Errorf("container %s is not the one that was declared after creating it", r.Name))
+	}
+	// Only now, with the replacement running, is the container set aside removed.
+	if aside != "" {
+		if _, err := run(ctx, cri, "rm", "-f", aside); err != nil {
+			out.Detail = fmt.Sprintf("the container it replaced, set aside as %s, could not be removed: %v", aside, err)
+		}
 	}
 	// The found networks a per-machine setting keeps for it (novox/hq ADR 0163, rule 4), joined
 	// once it runs: a runtime starts a container on one network, and the others are connected.
@@ -3040,6 +3085,9 @@ func applyContainer(ctx context.Context, r *declaration.Container, run Runner,
 	return out, nil
 }
 
+// asideSuffix names a running container set aside while its replacement is started (issue 1945).
+const asideSuffix = ".replaced"
+
 // detachedRunArgs is the `docker run` that starts a service container and leaves it running.
 func detachedRunArgs(r *declaration.Container, want string) []string {
 	args := []string{"run", "--detach", "--name", r.Name, "--restart", "unless-stopped"}

internal/apply/busy_test.go

@@ -33,6 +33,7 @@ func newBusyMachine() *aBusyMachine {
 }
 
 func (m *aBusyMachine) run(_ context.Context, _ string, args ...string) (string, error) {
+	args = asideShape(args)
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	switch args[0] {

internal/apply/containers_kept_test.go

@@ -20,6 +20,9 @@ func TestAContainerJoinsTheNetworksItKeeps(t *testing.T) {
 		if name != "docker" {
 			return "", errors.New("not installed")
 		}
+		if len(args) > 0 && args[0] == "container" && strings.HasSuffix(args[len(args)-1], asideSuffix) {
+			return "", errors.New("no such container") // nothing was left set aside
+		}
 		ran = append(ran, strings.Join(args, " "))
 		switch args[0] {
 		case "info":

internal/apply/exec_killed_test.go

@@ -0,0 +1,19 @@
+package apply
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// A command killed because its context ended says so, and why: "docker exited -1:" said nothing while novox's bus
+// container was half replaced (issue 1945, live on 2026-10-25 at 11:23).
+func TestACommandKilledByItsContextSaysWhy(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() { time.Sleep(50 * time.Millisecond); cancel() }()
+	_, err := ExecRunner(ctx, "sleep", "5")
+	if err == nil || !strings.Contains(err.Error(), "killed") || !strings.Contains(err.Error(), "context canceled") {
+		t.Fatalf("a command its context killed said: %v", err)
+	}
+}

internal/apply/hold_test.go

@@ -111,7 +111,10 @@ type fakeContainer struct {
 // options. The host passes `--` before every unit (third review of mesh-catalog #147), which
 // internal/system's own test holds; the fakes read the unit where they always did.
 func oldShape(name string, args []string) []string {
-	if name != "systemctl" || len(args) < 2 || args[len(args)-2] != "--" {
+	if name != "systemctl" {
+		return asideShape(args)
+	}
+	if len(args) < 2 || args[len(args)-2] != "--" {
 		return args
 	}
 	v := 0
@@ -1128,3 +1131,16 @@ func TestAPathTheMeshOnlySetAccessOnIsStillFound(t *testing.T) {
 		t.Errorf("it was re-moded to %o", info.Mode().Perm())
 	}
 }
+
+// asideShape reads the steps of a container set aside while its replacement starts (issue 1945) as the removal the
+// tests were written against: setting it aside is its removal, and what is done to the set-aside container afterwards
+// is no step of theirs.
+func asideShape(args []string) []string {
+	if len(args) == 3 && args[0] == "rename" && args[2] == args[1]+asideSuffix {
+		return []string{"rm", "-f", args[1]}
+	}
+	if len(args) > 0 && strings.HasSuffix(args[len(args)-1], asideSuffix) && (args[0] == "rm" || args[0] == "stop") {
+		return []string{"aside"}
+	}
+	return args
+}

internal/apply/maintenance_window_test.go

@@ -270,6 +270,7 @@ func newRegistryMachine(spec string) *aRegistryMachine {
 }
 
 func (m *aRegistryMachine) run(_ context.Context, _ string, args ...string) (string, error) {
+	args = asideShape(args)
 	m.mu.Lock()
 	switch args[0] {
 	case "info", "image":

internal/apply/replace_image_test.go

@@ -91,3 +91,129 @@ func TestAReplacementImageIsPresentBeforeTheRunningContainerIsRemoved(t *testing
 		t.Errorf("order of runtime calls = %s, want the image pulled before rm", got)
 	}
 }
+
+// asideRun is a runtime whose container "store" runs from an older spec, recording the steps a replacement takes; run
+// fails when failRun, and the new container reads back as declared otherwise.
+func asideRun(t *testing.T, want string, failRun bool, order *[]string) Runner {
+	created := false
+	return func(ctx context.Context, name string, args ...string) (string, error) {
+		switch args[0] {
+		case "info":
+			return "27.0\n", nil
+		case "container":
+			if created {
+				return "true\t" + want, nil
+			}
+			return "true\tsome-older-spec", nil
+		case "image":
+			return "", nil
+		case "rm", "rename", "stop", "start":
+			*order = append(*order, strings.Join(args, " "))
+			return "", nil
+		case "run":
+			*order = append(*order, "run")
+			if failRun {
+				return "", errors.New("port is already allocated")
+			}
+			created = true
+			return "deadbeef\n", nil
+		}
+		return "", nil
+	}
+}
+
+// A container whose replacement does not come up is put back: the running one is set aside (renamed and stopped), never
+// removed before the new one runs, and started again under its name when the new one fails, the apply failing loudly.
+// Removed first, the bus's own container left novox without a bus when its replacement was never made (issue 1945,
+// live on 2026-10-25 at 11:23).
+func TestAContainerWhoseReplacementFailsIsPutBack(t *testing.T) {
+	d := parseTrusted(t, `{"declaration":1,"resources":[
+	  {"id":"store","type":"container","name":"store","image":"`+pinned+`","env":{"PGDATA":"/data"}}
+	]}`)
+	want := containerSpec(d.Resources[0].(*declaration.Container), inputs{})
+	var order []string
+	_, _, err := Apply(context.Background(), archHost(t), d, store.State{}, store.OriginCarried, asideRun(t, want, true, &order), nil, nil)
+	if err == nil || !strings.Contains(err.Error(), "started again") {
+		t.Fatalf("a failed replacement: %v", err)
+	}
+	got := strings.Join(order, "; ")
+	if !strings.Contains(got, "rename store store.replaced; stop store.replaced; run") ||
+		!strings.Contains(got, "rename store.replaced store; start store") {
+		t.Fatalf("the steps were %q", got)
+	}
+	if strings.Contains(got, "rm -f store;") && strings.Index(got, "rm -f store;") < strings.Index(got, "run") {
+		t.Fatalf("the running container was removed before its replacement ran: %q", got)
+	}
+}
+
+// A replacement that runs removes the container set aside, and only after it runs.
+func TestAReplacementThatRunsRemovesTheContainerSetAside(t *testing.T) {
+	d := parseTrusted(t, `{"declaration":1,"resources":[
+	  {"id":"store","type":"container","name":"store","image":"`+pinned+`","env":{"PGDATA":"/data"}}
+	]}`)
+	want := containerSpec(d.Resources[0].(*declaration.Container), inputs{})
+	var order []string
+	if _, _, err := Apply(context.Background(), archHost(t), d, store.State{}, store.OriginCarried, asideRun(t, want, false, &order), nil, nil); err != nil {
+		t.Fatal(err)
+	}
+	got := strings.Join(order, "; ")
+	run, removed := strings.Index(got, "run"), strings.LastIndex(got, "rm -f store.replaced")
+	if run < 0 || removed < run || strings.Contains(got, "start store") {
+		t.Fatalf("the steps were %q: the set-aside container is removed after the replacement runs", got)
+	}
+}
+
+// A container left set aside by an engine that stopped between setting it aside and running its replacement is the
+// set-aside one of the next apply: put back when the replacement fails, removed once it runs (the review of issue
+// 1945).
+func TestAContainerLeftAsideIsPutBackOrRemovedByTheNextApply(t *testing.T) {
+	d := parseTrusted(t, `{"declaration":1,"resources":[
+	  {"id":"store","type":"container","name":"store","image":"`+pinned+`","env":{"PGDATA":"/data"}}
+	]}`)
+	want := containerSpec(d.Resources[0].(*declaration.Container), inputs{})
+	for _, failRun := range []bool{true, false} {
+		var order []string
+		created := false
+		run := func(ctx context.Context, name string, args ...string) (string, error) {
+			switch args[0] {
+			case "info":
+				return "27.0\n", nil
+			case "container":
+				switch args[len(args)-1] {
+				case "store":
+					if created {
+						return "true\t" + want, nil
+					}
+					return "", errors.New("no such container")
+				case "store.replaced":
+					return "false\tsome-older-spec", nil
+				}
+				return "", errors.New("no such container")
+			case "image":
+				return "", nil
+			case "rm", "rename", "stop", "start":
+				order = append(order, strings.Join(args, " "))
+				return "", nil
+			case "run":
+				order = append(order, "run")
+				if failRun {
+					return "", errors.New("port is already allocated")
+				}
+				created = true
+				return "deadbeef\n", nil
+			}
+			return "", nil
+		}
+		_, _, err := Apply(context.Background(), archHost(t), d, store.State{}, store.OriginCarried, run, nil, nil)
+		got := strings.Join(order, "; ")
+		if failRun {
+			if err == nil || !strings.Contains(got, "rename store.replaced store; start store") {
+				t.Fatalf("a failed run beside a container left aside: %v; steps %q", err, got)
+			}
+			continue
+		}
+		if err != nil || !strings.Contains(got, "run; rm -f store.replaced") {
+			t.Fatalf("a run beside a container left aside: %v; steps %q", err, got)
+		}
+	}
+}