diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index bff6c87..1f056e0 100644 --- a/cmd/mesh-host/main.go +++ b/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 diff --git a/cmd/mesh-host/main_test.go b/cmd/mesh-host/main_test.go index dbe477b..72a43f6 100644 --- a/cmd/mesh-host/main_test.go +++ b/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") diff --git a/cmd/mesh-host/stopping.go b/cmd/mesh-host/stopping.go new file mode 100644 index 0000000..e9b4e0c --- /dev/null +++ b/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) +} diff --git a/internal/store/save.go b/internal/store/save.go new file mode 100644 index 0000000..36bec2d --- /dev/null +++ b/internal/store/save.go @@ -0,0 +1,176 @@ +package store + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" +) + +// What a node knows is written by one writer at a time, completely or not at all, and never loses a record without +// saying so (novox/hq issue 1585). On novox, 2026-11-04, the node-engine's state had lost the records of two archives +// (nats's program snapshot and its tools bundle) that a save of 2026-10-24 had written into its temporary file and +// never renamed into place, and about 260 such temporary files (~80 MB) lay beside it: the process ended between +// writing a save and renaming it, again and again, because the engine stopped without waiting for the apply in flight +// (cmd/mesh-host, stopBetweenApplies). The temporary files and the lost records are the same fault. + +// tempPattern is the name a save writes before it renames the file into place. +const tempPattern = ".state-*.json" + +// saving is the one writer of this process: a save holds it from its read of what was saved before to its rename and +// the directory's flush, and Quiesce holds it for good. +var saving sync.Mutex + +// Said is where a save says loudly what must not pass in silence: a record the state held before and this save would +// drop although nothing forgot it. The journal when the engine runs as a service. +var Said = func(line string) { fmt.Fprintln(os.Stderr, line) } + +// Save writes the state, atomically and durably: the whole file to a temporary one beside it, flushed, renamed over +// the old one, and the directory flushed so the rename itself survives a power cut. +// +// Atomic because the alternative has a failure mode with no floor: a host interrupted while +// writing loses the record of everything it owns, and then owns nothing it can clean up. +// +// Before it writes, it reads what was saved before and says, through Said, each resource or hold that state recorded +// and this one would drop although this copy never forgot it (Forget, Release): a save from a copy read before +// another writer saved, or a copy that lost a record on the way, is said at once rather than found weeks later when a +// file the host put there reads as foreign. The save still goes ahead: it is what the apply found on the machine, and +// refusing it would lose that too. +func Save(path string, s State) error { + saving.Lock() + defer saving.Unlock() + return save(path, s) +} + +func save(path string, s State) error { + s.UpdatedAt = time.Now().UTC() + + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("making room for the node's state: %w", err) + } + + if before, err := Load(path); err == nil { + for _, lost := range Dropped(before, s) { + Said(fmt.Sprintf("the node's state is saved without the record of %s, which the state saved before held "+ + "and nothing forgot: what it put on this machine is no longer known as the mesh's (issue 1585)", lost)) + } + } + + raw, err := json.MarshalIndent(s, "", " ") + if err != nil { + return fmt.Errorf("encoding the node's state: %w", err) + } + raw = append(raw, '\n') + + tmp, err := os.CreateTemp(dir, tempPattern) + if err != nil { + return fmt.Errorf("writing the node's state: %w", err) + } + defer os.Remove(tmp.Name()) + + if _, err := tmp.Write(raw); err != nil { + tmp.Close() + return fmt.Errorf("writing the node's state: %w", err) + } + // Flushed before the rename: a rename is atomic, and a rename of a file whose contents are + // still in the page cache is atomically the wrong thing. + if err := tmp.Sync(); err != nil { + tmp.Close() + return fmt.Errorf("flushing the node's state: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("closing the node's state: %w", err) + } + if err := os.Chmod(tmp.Name(), 0o600); err != nil { + return fmt.Errorf("securing the node's state: %w", err) + } + if err := os.Rename(tmp.Name(), path); err != nil { + return fmt.Errorf("replacing the node's state: %w", err) + } + // And the directory, so the new name is on disk and not only the new contents. + if d, err := os.Open(dir); err == nil { + syncErr := d.Sync() + d.Close() + if syncErr != nil { + return fmt.Errorf("flushing the directory of the node's state: %w", syncErr) + } + } + return nil +} + +// Dropped is each resource and hold before recorded that after does not, and that after never forgot: "resource " +// or "hold ", sorted. +func Dropped(before, after State) []string { + kept := map[string]bool{} + for _, r := range after.Resources { + kept["resource "+r.ID] = true + } + for _, h := range after.Held { + kept["hold "+h.ID] = true + } + var out []string + for _, r := range before.Resources { + if k := "resource " + r.ID; !kept[k] && !after.forgotten[k] { + out = append(out, k) + } + } + for _, h := range before.Held { + if k := "hold " + h.ID; !kept[k] && !after.forgotten[k] { + out = append(out, k) + } + } + sort.Strings(out) + return out +} + +// forget notes that this copy dropped a record on purpose. +func (s *State) forget(what string) { + if s.forgotten == nil { + s.forgotten = map[string]bool{} + } + s.forgotten[what] = true +} + +// Quiesce waits for a save in flight to end and lets no other start in this process, for the engine's stop: a process +// that ends while a save writes leaves a temporary file and the state before it (issue 1585). It returns what lets saves +// start again, for a test; a stopping engine never calls it. +func Quiesce() (release func()) { + saving.Lock() + return saving.Unlock +} + +// RemoveLeftovers removes the temporary files of saves that never reached their rename, beside the state at path, and +// names each it removed. Under the machine's apply lock, so no save of another process is in flight: every writer of +// the state takes it (Lock). +func RemoveLeftovers(path string) ([]string, error) { + unlock, err := Lock(path, nil) + if err != nil { + return nil, err + } + defer unlock() + saving.Lock() + defer saving.Unlock() + found, err := filepath.Glob(filepath.Join(filepath.Dir(path), tempPattern)) + if err != nil { + return nil, err + } + var removed []string + var failed []string + for _, f := range found { + if err := os.Remove(f); err != nil { + failed = append(failed, fmt.Sprintf("%s: %v", filepath.Base(f), err)) + continue + } + removed = append(removed, filepath.Base(f)) + } + sort.Strings(removed) + if len(failed) > 0 { + return removed, fmt.Errorf("leftover saves of the node's state could not be removed: %s", strings.Join(failed, "; ")) + } + return removed, nil +} diff --git a/internal/store/store.go b/internal/store/store.go index 5755b81..892f5c0 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,7 +16,6 @@ import ( "fmt" "io/fs" "os" - "path/filepath" "sort" "strings" "time" @@ -263,6 +262,10 @@ type State struct { // Members is every library group the engine put a tool runner's account in that it was not in before (ADR 0328 // §11, as ADR 0252 records an account's groups): taken back when no longer declared. Members []Membership `json:"members,omitempty"` + + // forgotten is what this copy dropped on purpose since it was loaded (Forget, Release): never written, read by Save + // to tell a record given up from one lost (issue 1585). + forgotten map[string]bool } // MadeGroup is one library group the engine made. @@ -484,7 +487,7 @@ func (s *State) RecordHeld(h Held) { // Release drops a hold, once its module is taken and the host has converged what was held. func (s *State) Release(id string) { - kept := s.Held[:0] + kept := make([]Held, 0, len(s.Held)) for _, h := range s.Held { if h.ID != id { kept = append(kept, h) @@ -494,6 +497,7 @@ func (s *State) Release(id string) { if len(s.Held) == 0 { s.Held = nil } + s.forget("hold " + id) } // RetiredAt returns the retirement of the found configuration at a path, if the mesh retired one. @@ -563,51 +567,6 @@ func Load(path string) (State, error) { return s, nil } -// Save writes the state, atomically. -// -// Atomic because the alternative has a failure mode with no floor: a host interrupted while -// writing loses the record of everything it owns, and then owns nothing it can clean up. -func Save(path string, s State) error { - s.UpdatedAt = time.Now().UTC() - - if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { - return fmt.Errorf("making room for the node's state: %w", err) - } - - raw, err := json.MarshalIndent(s, "", " ") - if err != nil { - return fmt.Errorf("encoding the node's state: %w", err) - } - raw = append(raw, '\n') - - tmp, err := os.CreateTemp(filepath.Dir(path), ".state-*.json") - if err != nil { - return fmt.Errorf("writing the node's state: %w", err) - } - defer os.Remove(tmp.Name()) - - if _, err := tmp.Write(raw); err != nil { - tmp.Close() - return fmt.Errorf("writing the node's state: %w", err) - } - // Flushed before the rename: a rename is atomic, and a rename of a file whose contents are - // still in the page cache is atomically the wrong thing. - if err := tmp.Sync(); err != nil { - tmp.Close() - return fmt.Errorf("flushing the node's state: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("closing the node's state: %w", err) - } - if err := os.Chmod(tmp.Name(), 0o600); err != nil { - return fmt.Errorf("securing the node's state: %w", err) - } - if err := os.Rename(tmp.Name(), path); err != nil { - return fmt.Errorf("replacing the node's state: %w", err) - } - return nil -} - // Record adds or replaces what is known about one resource, preserving order. func (s *State) Record(a Applied) { for i, existing := range s.Resources { @@ -640,15 +599,18 @@ func FormerID(id, target string) string { return id + "@former:" + target } // IsFormer says whether a record names a former target. func IsFormer(id string) bool { return strings.Contains(id, "@former:") } -// Forget drops a resource from what the node owns. +// Forget drops a resource from what the node owns, and notes that it was dropped on purpose, so the save that follows +// does not say it lost the record (Save). A new slice, never the old one's array: a copy of the state taken before keeps +// what it held. func (s *State) Forget(id string) { - kept := s.Resources[:0] + kept := make([]Applied, 0, len(s.Resources)) for _, r := range s.Resources { if r.ID != id { kept = append(kept, r) } } s.Resources = kept + s.forget("resource " + id) } // Orphans returns what the host applied and the declaration no longer names, newest first. diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 712c3bd..d650171 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -93,6 +93,66 @@ func TestSavingLeavesNoDebrisBehind(t *testing.T) { } t.Errorf("expected only the state file, found: %v", names) } + + // Issue 1585: a save cut off between its write and its rename (the engine ended under it) left its temporary file; + // the engine's start removes them under the apply lock and names each, and leaves the state and every other file. + t.Run("leftovers of cut-off saves are removed", func(t *testing.T) { + for _, name := range []string{".state-3758449435.json", ".state-1403853615.json", ".declared-1.json"} { + if err := os.WriteFile(filepath.Join(dir, name), []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + } + removed, err := RemoveLeftovers(path) + if err != nil { + t.Fatal(err) + } + if strings.Join(removed, ",") != ".state-1403853615.json,.state-3758449435.json" { + t.Errorf("removed %v", removed) + } + for _, keep := range []string{"state.json", ".declared-1.json"} { + if _, err := os.Stat(filepath.Join(dir, keep)); err != nil { + t.Errorf("%s was not left: %v", keep, err) + } + } + }) + + // Issue 1585: a save that drops a record the state saved before held, which nothing forgot, says so loudly; one + // forgotten or released on purpose is not said. Forget never writes into the array a copy taken before still reads. + t.Run("a dropped record is said, a forgotten one is not", func(t *testing.T) { + var said []string + was := Said + Said = func(line string) { said = append(said, line) } + defer func() { Said = was }() + before := State{Resources: []Applied{{ID: "c", Type: "file", Target: "/c"}, + {ID: "nats.program-snapshot", Type: "archive", Target: "/s"}, {ID: "nats.bundle-tools", Type: "archive", Target: "/t"}}, + Held: []Held{{ID: "h"}}} + if err := Save(path, before); err != nil { + t.Fatal(err) + } + if len(said) != 1 || !strings.Contains(said[0], "resource a,") { + t.Errorf("the record of a, saved before and dropped, was not said: %v", said) + } + said = nil + copied := before + copied.Forget("c") + copied.Release("h") + if before.Resources[0].ID != "c" { + t.Errorf("Forget wrote into the copy taken before: %+v", before.Resources) + } + if err := Save(path, copied); err != nil { + t.Fatal(err) + } + if len(said) != 0 { + t.Errorf("a record forgotten on purpose was said lost: %v", said) + } + stale := State{Resources: []Applied{{ID: "nats.bundle-tools", Type: "archive", Target: "/t"}}} + if err := Save(path, stale); err != nil { + t.Fatal(err) + } + if len(said) != 1 || !strings.Contains(said[0], "resource nats.program-snapshot") { + t.Errorf("a lost record was not said, or said wrong: %v", said) + } + }) } func TestRecordReplacesRatherThanDuplicating(t *testing.T) {