Stop between applies and save the node's state as one writer, so a stop never cuts a save off and loses its records (issue 1585)

On novox the node-engine's state had lost the records of nats's two
archives, which only a temporary save file of 2026-10-24 still held, and
about 260 such files (~80 MB) lay beside it. The cause: an apply runs
detached from the engine's stop (applyContext, WithoutCancel), and
runLink returned, and the process ended, without waiting for it; a save
cut between its write and its rename left its temporary file and the
state before it, and whatever the apply did was never recorded.

Quick fix: remove the leftover files by hand and restore the records.
Thorough fix, chosen: the engine ends between applies whichever way it
ends (a stop, standing aside, a move to another bus): it waits up to 60 s
for the apply in flight, then lets no save start (store.Quiesce). Save is
one writer per process, flushes the directory after its rename, and says
loudly each record the state saved before held that this save drops and
nothing forgot (Forget, Release mark what is dropped on purpose, and no
longer write into an array an earlier copy reads). The engine's start
removes leftover temporary saves under the apply lock and says how many.

Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn

cmd/mesh-host/main.go

@@ -936,6 +936,11 @@ func runLink(ctx context.Context, opts options) error {
 	// which the launcher already reads as "run whatever is on disk now".
 	aside, standAside := context.WithCancel(ctx)
 	defer standAside()
+	// **And it ends between applies too** (novox/hq issue 1585): whichever way the link ends, the apply in flight ends and
+	// saves before the process does. Deferred after standAside, so it runs before it: nothing is cancelled under it.
+	defer func() { _ = stopBetweenApplies(say, stopBound) }()
+	// What an earlier stop cut off before its rename is removed, and said, before anything saves.
+	removeLeftoverSaves(opts.state, say)
 	stoodAside := false
 	membership := link.Membership{
 		Node:        mine.Node,
@@ -2019,7 +2024,7 @@ func adoptDeliveredMembership(identityPath string, mine *identity.Identity, say
 		return
 	}
 	say(fmt.Sprintf("moving to the %s bus at %s — restarting to dial it", next.Transport, next.Broker))
-	os.Exit(0)
+	exitBetweenApplies(say)
 }
 
 // profileAsReported is the profile as the mesh reads it — the same bytes enrolment sends, so a

cmd/mesh-host/main_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -132,6 +133,43 @@ func TestAFlagAfterAPositionalIsRead(t *testing.T) {
 // state, acts, and writes it back — so they must not run at the same time, or the last save loses
 // what the other recorded.
 func TestOnlyOneApplyRunsAtATime(t *testing.T) {
+	// Issue 1585: the engine's stop waits for the apply in flight to end and save, and then lets no save start; before,
+	// the process ended under it and a save cut before its rename left its temporary file and the state before it.
+	t.Run("the stop waits for the apply in flight and its save", func(t *testing.T) {
+		path := filepath.Join(t.TempDir(), "state.json")
+		applying.Lock()
+		var said []string
+		var mu sync.Mutex
+		say := func(line string) { mu.Lock(); said = append(said, line); mu.Unlock() }
+		stopped := make(chan func(), 1)
+		go func() { stopped <- stopBetweenApplies(say, time.Hour) }()
+		select {
+		case release := <-stopped:
+			release()
+			applying.Unlock()
+			t.Fatal("the stop did not wait for the apply in flight")
+		case <-time.After(50 * time.Millisecond):
+		}
+		if err := store.Save(path, store.State{Resources: []store.Applied{{ID: "nats.program-snapshot", Type: "archive"}}}); err != nil {
+			t.Fatal(err)
+		}
+		applying.Unlock()
+		release := <-stopped
+		saved := make(chan error, 1)
+		go func() { saved <- store.Save(path, store.State{}) }()
+		select {
+		case <-saved:
+			t.Error("a save started after the stop")
+		case <-time.After(50 * time.Millisecond):
+		}
+		release()
+		<-saved
+		mu.Lock()
+		defer mu.Unlock()
+		if len(said) != 2 || !strings.Contains(said[0], "waiting up to 1h0m0s") || !strings.Contains(said[1], "its state is saved") {
+			t.Errorf("said %q", said)
+		}
+	})
 	// A host is built for one system at link time, and a test binary has no link time: this asks
 	// the machine it runs on, and stands aside where the answer is no.
 	built, err := system.For("arch")

cmd/mesh-host/stopping.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/novox/mesh-host/internal/store"
+)
+
+// stopBound is how long a stopping engine waits for the apply in flight to end and save what it did: under the service
+// manager's own stop bound (90 s by default, then every process of the unit is killed), so the wait ends with a line in
+// the journal rather than with a kill.
+const stopBound = 60 * time.Second
+
+// stopBetweenApplies is how this engine ends, whichever way (a stop of the service, standing aside for a successor, a
+// move to another bus): it waits, up to bound, for the apply in flight to end, then lets no save start and waits for
+// the one writing (novox/hq issue 1585). An apply runs detached from the engine's own stop (applyContext), so without
+// this the process ended under it: what it unpacked was never recorded, and a save cut between its write and its
+// rename left a temporary file and the state before it. A stopping engine never gives back what it takes (the process
+// ends); the release it returns is for a test, and gives back the apply only when this took it.
+func stopBetweenApplies(say func(string), bound time.Duration) (release func()) {
+	took := true
+	if !applying.TryLock() {
+		say(fmt.Sprintf("stopping: waiting up to %s for the apply in flight to end and save what it did", bound))
+		got := make(chan struct{})
+		go func() {
+			applying.Lock()
+			close(got)
+		}()
+		select {
+		case <-got:
+			say("stopping: the apply in flight ended and its state is saved")
+		case <-time.After(bound):
+			took = false
+			say(fmt.Sprintf("stopping while an apply still runs after %s: what it did since its last save is not "+
+				"recorded, and the next start applies its module again", bound))
+		}
+	}
+	saves := store.Quiesce()
+	return func() {
+		saves()
+		if took {
+			applying.Unlock()
+		}
+	}
+}
+
+// removeLeftoverSaves removes, at the engine's start, the temporary files of saves of the node's state that never
+// reached their rename, and says how many (issue 1585: about 260 on novox, ~80 MB).
+func removeLeftoverSaves(statePath string, say func(string)) {
+	removed, err := store.RemoveLeftovers(statePath)
+	if len(removed) > 0 {
+		shown := removed
+		if len(shown) > 5 {
+			shown = append(append([]string(nil), shown[:5]...), "…")
+		}
+		say(fmt.Sprintf("removed %d leftover save(s) of this node's state, cut off before their rename by an earlier "+
+			"stop: %s", len(removed), strings.Join(shown, ", ")))
+	}
+	if err != nil {
+		say(err.Error())
+	}
+}
+
+// exitBetweenApplies ends the process cleanly once no apply runs and no save writes: an exit that is not runLink's
+// return (a move to another bus). A variable, so a test sees it called rather than the test process ending.
+var exitBetweenApplies = func(say func(string)) {
+	_ = stopBetweenApplies(say, stopBound)
+	os.Exit(0)
+}