Merge pull request 'Node-engine: a stop never cuts a save off, an archive's unpack is recorded at once, and an answer the want carries is acted on at start (issue 1585)' (#207) from fix/archive-record-at-unpack into main
This commit was merged in pull request #207.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -132,6 +133,57 @@ 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
|
||||
waiting := make(chan struct{})
|
||||
say := func(line string) {
|
||||
mu.Lock()
|
||||
said = append(said, line)
|
||||
mu.Unlock()
|
||||
if strings.Contains(line, "waiting up to") {
|
||||
close(waiting)
|
||||
}
|
||||
}
|
||||
stopped := make(chan func(), 1)
|
||||
go func() { stopped <- stopBetweenApplies(say, time.Hour) }()
|
||||
// Said before it blocks, and it cannot end before the apply lets go: so it has not ended now.
|
||||
<-waiting
|
||||
select {
|
||||
case release := <-stopped:
|
||||
release()
|
||||
applying.Unlock()
|
||||
t.Fatal("the stop did not wait for the apply in flight")
|
||||
default:
|
||||
}
|
||||
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{}) }()
|
||||
// A save after the stop blocks; nothing says it blocks, so this one wait is a bounded look.
|
||||
select {
|
||||
case <-saved:
|
||||
release()
|
||||
t.Fatal("a save started after the stop")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
release()
|
||||
if err := <-saved; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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")
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -1375,6 +1375,19 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
a.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
if h.Phase == string(delivery.PhaseHeld) && (w.Want.Backup == delivery.BackupAgain || w.Want.Backup == delivery.BackupSkip) {
|
||||
// Held for a failed backup, and the want already carries the operator's answer, written before this engine
|
||||
// started (nats on novox, 2026-11-04: "Back up again" answered, then the engine restarted and took the
|
||||
// answer for the hold's own revision, so it was never acted on; issue 1585). The answer is acted on now:
|
||||
// again backs it up and applies (a.prior makes the apply back it up first), skip applies without one.
|
||||
// An answer the engine already acted on before its stop, and whose backup failed once more, is so acted
|
||||
// on once more at this start: one backup more, never one skipped.
|
||||
a.mu.Lock()
|
||||
a.sayOnce("answered:"+module, fmt.Sprintf("%s is held for a failed backup, and its want carries the operator's "+
|
||||
"answer %q: acted on at this start", module, w.Want.Backup))
|
||||
a.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
if h.Phase == string(delivery.PhaseHeld) {
|
||||
// Held for a failed backup, and the operator's answer is still to come: the hold is kept, nothing is
|
||||
// backed up again or applied, and the next revision of the want is the answer (ADR 0568 §4).
|
||||
|
||||
@@ -595,7 +595,8 @@ func TestADataModuleIsBackedUpFirstAndHeldWhenItFails(t *testing.T) {
|
||||
}
|
||||
|
||||
// At a start: a backup the node agent's own stop interrupted (postgres on novox, 2026-10-31) is no failure, so it is
|
||||
// backed up again and applied while the want asks a backup; a backup that failed stays held for the operator.
|
||||
// backed up again and applied while the want asks a backup; a backup that failed stays held for the operator's
|
||||
// answer, and an answer the want already carries is acted on.
|
||||
for _, row := range []struct {
|
||||
detail, backup string
|
||||
backups, applied int
|
||||
@@ -605,7 +606,11 @@ func TestADataModuleIsBackedUpFirstAndHeldWhenItFails(t *testing.T) {
|
||||
{BackupInterrupted, delivery.BackupAgain, 1, 1, delivery.PhaseRunning, false},
|
||||
// novox's postgres: its held have, written before mesh-host #189, names nothing it runs.
|
||||
{BackupInterrupted, "", 1, 1, delivery.PhaseRunning, true},
|
||||
{"restic: the repository is locked", delivery.BackupAgain, 0, 0, delivery.PhaseHeld, false},
|
||||
// A backup that failed, and the want carries the operator's answer, written before this start (nats on novox,
|
||||
// 2026-11-04, issue 1585): again backs it up and applies, skip applies without one; no answer keeps the hold.
|
||||
{"restic: the repository is locked", delivery.BackupAgain, 1, 1, delivery.PhaseRunning, false},
|
||||
{"restic: the repository is locked", delivery.BackupSkip, 0, 1, delivery.PhaseRunning, false},
|
||||
{"restic: the repository is locked", "", 0, 0, delivery.PhaseHeld, false},
|
||||
} {
|
||||
r := newBed(t)
|
||||
r.comps.before(ck1, craw1)
|
||||
|
||||
@@ -610,7 +610,7 @@ func applyScoped(
|
||||
err = fmt.Errorf("%w: %s", errRemovalWaits, said)
|
||||
}
|
||||
default:
|
||||
action, detail, err = remove(ctx, sys, orphan, run, made)
|
||||
action, detail, err = remove(ctx, sys, orphan, run, made, journalDir(windows))
|
||||
}
|
||||
if errors.Is(err, errNoRemoval) && store.IsFormer(orphan.ID) {
|
||||
// **A former target of a kind the host cannot remove is left in place and forgotten,
|
||||
@@ -1572,7 +1572,7 @@ func applyOne(ctx context.Context, sys system.System, r declaration.Resource, ru
|
||||
case *declaration.User:
|
||||
return applyUser(ctx, sys, res, run, previous, in.groups, in.handed)
|
||||
case *declaration.Archive:
|
||||
return applyArchive(ctx, res, previous, in.away)
|
||||
return applyArchive(ctx, res, previous, in.away, journalDir(in.windows))
|
||||
case *declaration.Process:
|
||||
return applyProcess(ctx, res, run, changed, previous, in.away)
|
||||
case *declaration.Action:
|
||||
@@ -2483,7 +2483,7 @@ func reflectOnly(ctx context.Context, sys system.System, r *declaration.Service,
|
||||
//
|
||||
// made is the units whose unit file this host wrote where there was none (meshMadeUnits).
|
||||
func remove(ctx context.Context, sys system.System, a store.Applied, run Runner,
|
||||
made map[string]bool) (string, string, error) {
|
||||
made map[string]bool, journals string) (string, string, error) {
|
||||
switch declaration.Type(a.Type) {
|
||||
case declaration.TypeDirectory, declaration.TypeFile, declaration.TypeArchive:
|
||||
// Removal below a home follows no link an account put there either (novox/hq ADR 0266).
|
||||
@@ -2609,7 +2609,7 @@ func remove(ctx context.Context, sys system.System, a store.Applied, run Runner,
|
||||
case declaration.TypeArchive:
|
||||
// Exactly what it unpacked, and the directories the host made for it once they are empty
|
||||
// (novox/hq issue 162).
|
||||
return removeArchive(a)
|
||||
return removeArchive(a, journals)
|
||||
|
||||
default:
|
||||
return "", "", fmt.Errorf("%w: a %q", errNoRemoval, a.Type)
|
||||
|
||||
@@ -35,9 +35,14 @@ import (
|
||||
// enough for a desktop theme and small enough to notice.
|
||||
const maxArchive = 512 << 20
|
||||
|
||||
func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Applied, away *storeAway) (Outcome, error) {
|
||||
func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Applied, away *storeAway, journals string) (Outcome, error) {
|
||||
out := begin(r)
|
||||
out.Action = "unchanged"
|
||||
recorded := previous.Wrote != ""
|
||||
previous, err := withJournal(journals, previous, r.Path)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
|
||||
body, err := fetchMinding(ctx, r.Source, away)
|
||||
if err != nil {
|
||||
@@ -78,8 +83,12 @@ func applyArchive(ctx context.Context, r *declaration.Archive, previous store.Ap
|
||||
return out, err
|
||||
}
|
||||
out.unpacked = &unpacked
|
||||
// At once, before the rest of the apply can fail or the engine stop (novox/hq issue 1585).
|
||||
if err := keepJournal(journals, r.Path, got, &unpacked); err != nil {
|
||||
return out, fmt.Errorf("%s was unpacked, and what it unpacked could not be written down: %w", r.Path, err)
|
||||
}
|
||||
out.Action = "updated"
|
||||
if previous.Wrote == "" {
|
||||
if !recorded {
|
||||
out.Action = "created"
|
||||
}
|
||||
out.Detail = fmt.Sprintf("%d file(s)", written)
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
package apply
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
)
|
||||
|
||||
// What an archive unpacked is written down the moment it is unpacked (novox/hq issue 1585).
|
||||
//
|
||||
// The node's state is saved at the end of a module's apply, so an engine that stopped, was replaced or lost its
|
||||
// state between the unpack and that save held files on the machine it had no record of. The next apply then found
|
||||
// its own files and refused them as ones "the mesh did not put there" (issue 162), and a module could not move to
|
||||
// its new build until a person removed them. So, at once and atomically, the host keeps the digest it unpacked and
|
||||
// the files and directories it placed; an apply that holds no record of the archive reads this instead of treating
|
||||
// everything there as somebody's. The record in the state stays the authority: the journal is read only when there
|
||||
// is none.
|
||||
//
|
||||
// **In the node-engine's own state directory, readable and writable by root alone, never beside the archive.** The
|
||||
// archive's directory may be writable by the module's account (a home, a data directory): a journal there could be
|
||||
// forged, and a root engine would then remove, or take as its own, any path the forger named. And an entry is read
|
||||
// only when it names a path inside the archive's directory (filepath.IsLocal), and a parent only when it is above it.
|
||||
|
||||
// JournalDirName is the directory under the node's state directory that keeps the journals.
|
||||
const JournalDirName = "unpacked"
|
||||
|
||||
type unpackJournal struct {
|
||||
Target string `json:"target"`
|
||||
Wrote string `json:"wrote"`
|
||||
Unpacked *store.Unpacked `json:"unpacked"`
|
||||
}
|
||||
|
||||
// journalDir is where the journals of this apply are kept: under the state directory every applier on the machine
|
||||
// shares (the one its windows are kept in). Empty for a caller with no state directory (a test), which keeps none.
|
||||
func journalDir(w *Windows) string {
|
||||
if w == nil || w.dir == "" {
|
||||
return ""
|
||||
}
|
||||
return filepath.Join(filepath.Dir(w.dir), JournalDirName)
|
||||
}
|
||||
|
||||
// journalPath names the journal of the archive at target: by the digest of its path, so no path names a file
|
||||
// outside the journals' directory.
|
||||
func journalPath(dir, target string) string {
|
||||
sum := sha256.Sum256([]byte(filepath.Clean(target)))
|
||||
return filepath.Join(dir, hex.EncodeToString(sum[:])+".json")
|
||||
}
|
||||
|
||||
// keepJournal writes down what an archive unpacked at path.
|
||||
func keepJournal(dir, path, wrote string, u *store.Unpacked) error {
|
||||
if dir == "" {
|
||||
return nil
|
||||
}
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
raw, err := json.Marshal(unpackJournal{Target: filepath.Clean(path), Wrote: wrote, Unpacked: u})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return writeAtomicallyByPath(journalPath(dir, path), raw, 0o600)
|
||||
}
|
||||
|
||||
// journalOf is what the journal says of path: false when there is none, and an error when there is one that cannot
|
||||
// be read or names a path outside the archive's directory, which is never believed.
|
||||
func journalOf(dir, path string) (unpackJournal, bool, error) {
|
||||
if dir == "" {
|
||||
return unpackJournal{}, false, nil
|
||||
}
|
||||
raw, err := os.ReadFile(journalPath(dir, path))
|
||||
if err != nil {
|
||||
return unpackJournal{}, false, nil
|
||||
}
|
||||
var j unpackJournal
|
||||
if err := json.Unmarshal(raw, &j); err != nil {
|
||||
return unpackJournal{}, false, fmt.Errorf("the journal of what was unpacked at %s is unreadable: %w", path, err)
|
||||
}
|
||||
if j.Wrote == "" || j.Unpacked == nil {
|
||||
return unpackJournal{}, false, fmt.Errorf("the journal of what was unpacked at %s names no digest or files", path)
|
||||
}
|
||||
if err := checkJournal(j, filepath.Clean(path)); err != nil {
|
||||
return unpackJournal{}, false, err
|
||||
}
|
||||
return j, true, nil
|
||||
}
|
||||
|
||||
// checkJournal refuses a journal of another target, an entry that is not a path inside the archive's directory, and
|
||||
// a parent that is not above it.
|
||||
func checkJournal(j unpackJournal, target string) error {
|
||||
if j.Target != target {
|
||||
return fmt.Errorf("the journal kept for %s names %q", target, j.Target)
|
||||
}
|
||||
for _, rel := range append(append([]string(nil), j.Unpacked.Files...), j.Unpacked.Dirs...) {
|
||||
if !filepath.IsLocal(filepath.FromSlash(rel)) {
|
||||
return fmt.Errorf("the journal of what was unpacked at %s names %q, which is not inside it: not believed", target, rel)
|
||||
}
|
||||
}
|
||||
for _, p := range j.Unpacked.Parents {
|
||||
if p = filepath.Clean(p); !filepath.IsAbs(p) || p == target || !strings.HasPrefix(target, p+string(filepath.Separator)) {
|
||||
return fmt.Errorf("the journal of what was unpacked at %s names the parent %q, which is not above it: not believed", target, p)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// withJournal is previous, or, when the state holds no record of the archive, what the journal says the host
|
||||
// unpacked at path. A journal not believed is an error: the apply stops and says so rather than guess.
|
||||
func withJournal(dir string, previous store.Applied, path string) (store.Applied, error) {
|
||||
if previous.Wrote != "" {
|
||||
return previous, nil
|
||||
}
|
||||
j, ok, err := journalOf(dir, path)
|
||||
if err != nil || !ok {
|
||||
return previous, err
|
||||
}
|
||||
previous.Wrote, previous.Target, previous.Unpacked = j.Wrote, path, j.Unpacked
|
||||
return previous, nil
|
||||
}
|
||||
|
||||
// dropJournal removes the journal of the archive at target, once the archive is taken off; whether the state's
|
||||
// record of it still said what it unpacked or not.
|
||||
func dropJournal(dir, target string) {
|
||||
if dir == "" || target == "" {
|
||||
return
|
||||
}
|
||||
_ = os.Remove(journalPath(dir, target))
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package apply
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -152,6 +153,98 @@ func TestAnArchiveDoesNotWriteOverAFileItFound(t *testing.T) {
|
||||
if entries, _ := os.ReadDir(dir); len(entries) != 1 {
|
||||
t.Fatalf("%d entries beside the directory", len(entries))
|
||||
}
|
||||
|
||||
// Issue 1585: what the host unpacked is written down at once, in its own state directory, so an archive whose
|
||||
// record in the state was lost is still the host's: the next version replaces its files, drops the ones it no
|
||||
// longer has and leaves a file somebody added. A file the host never unpacked still refuses it, and so does a
|
||||
// journal naming a path outside the archive's directory. The journal goes with the archive, record lost or not.
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
journaled bool // the first version was unpacked by the host
|
||||
foreign string
|
||||
forged func(j *unpackJournal)
|
||||
refused string
|
||||
}{
|
||||
{"unpacked by the host, record lost", true, "", nil, ""},
|
||||
{"unpacked by the host, somebody's file beside", true, "mine.txt", nil, ""},
|
||||
{"never unpacked by the host", false, "", nil, "did not put there"},
|
||||
{"a journal naming a path outside", true, "", func(j *unpackJournal) {
|
||||
j.Unpacked.Files = append(j.Unpacked.Files, "../../etc/passwd")
|
||||
}, "not inside it"},
|
||||
{"a journal naming a parent not above", true, "", func(j *unpackJournal) {
|
||||
j.Unpacked.Parents = []string{"/etc"}
|
||||
}, "not above it"},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
dir, state := t.TempDir(), t.TempDir()
|
||||
windows, journals := WindowsIn(state), filepath.Join(state, JournalDirName)
|
||||
applyIt := func(decl string, known store.State) (store.State, error) {
|
||||
_, after, err := ApplyMindingWindows(context.Background(), archHost(t), declare(t, decl), known,
|
||||
store.OriginDeclared, noServices, nil, nil, nil, windows)
|
||||
return after, err
|
||||
}
|
||||
target := filepath.Join(dir, "tools")
|
||||
first, _ := archiveDecl(t, "m.tools", target, map[string]string{"bin": "one", "old": "o"}, "")
|
||||
if row.journaled {
|
||||
if _, err := applyIt(first, store.State{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if entries, _ := os.ReadDir(dir); len(entries) != 1 {
|
||||
t.Fatalf("%d entries beside the archive: the journal is the state directory's", len(entries))
|
||||
}
|
||||
info, err := os.Stat(journalPath(journals, target))
|
||||
if err != nil || info.Mode().Perm() != 0o600 {
|
||||
t.Fatalf("no journal of root's alone after the unpack: %v %v", info, err)
|
||||
}
|
||||
} else {
|
||||
os.MkdirAll(target, 0o755)
|
||||
os.WriteFile(filepath.Join(target, "bin"), []byte("one"), 0o644)
|
||||
}
|
||||
if row.forged != nil {
|
||||
j, _, err := journalOf(journals, target)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
row.forged(&j)
|
||||
raw, _ := json.Marshal(j)
|
||||
os.WriteFile(journalPath(journals, target), raw, 0o600)
|
||||
}
|
||||
if row.foreign != "" {
|
||||
os.WriteFile(filepath.Join(target, row.foreign), []byte("x"), 0o644)
|
||||
}
|
||||
second, _ := archiveDecl(t, "m.tools", target, map[string]string{"bin": "two"}, "")
|
||||
// The state is empty: the record of the first unpack was lost.
|
||||
after, err := applyIt(second, store.State{})
|
||||
if row.refused != "" {
|
||||
if err == nil || !strings.Contains(err.Error(), row.refused) {
|
||||
t.Fatalf("refused? %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, _ := os.ReadFile(filepath.Join(target, "bin")); string(got) != "two" {
|
||||
t.Fatalf("bin is %q", got)
|
||||
}
|
||||
if exists(filepath.Join(target, "old")) {
|
||||
t.Fatal("a file the new version no longer has is still there")
|
||||
}
|
||||
if row.foreign != "" && !exists(filepath.Join(target, row.foreign)) {
|
||||
t.Fatal("somebody's file went")
|
||||
}
|
||||
// Taken off, with a record that no longer says what it unpacked: the journal goes too.
|
||||
for i := range after.Resources {
|
||||
after.Resources[i].Unpacked = nil
|
||||
}
|
||||
if _, err := applyIt(`{"id":"other","type":"file","path":"`+dir+`/other","content":"o"}`, after); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if exists(journalPath(journals, target)) {
|
||||
t.Fatal("the journal outlived the archive")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestADirectoryTheHostMadeGoesOnlyWhenEmpty(t *testing.T) {
|
||||
|
||||
@@ -364,7 +364,16 @@ func mergeInto(fresh, path, owner string, files, dirs []string, o ours) (store.U
|
||||
// module's tools from its bundle — sees the whole tree or none of it, never half, and a file it has
|
||||
// open stays readable until it closes it. The runtime is told the module went by its own membership,
|
||||
// not by the files disappearing.
|
||||
func removeArchive(a store.Applied) (string, string, error) {
|
||||
func removeArchive(a store.Applied, journals string) (string, string, error) {
|
||||
action, detail, err := removeArchiveFiles(a)
|
||||
// And its journal (issue 1585), once the record goes: whether the record said what it unpacked or not.
|
||||
if err == nil {
|
||||
dropJournal(journals, a.Target)
|
||||
}
|
||||
return action, detail, err
|
||||
}
|
||||
|
||||
func removeArchiveFiles(a store.Applied) (string, string, error) {
|
||||
if store.IsFormer(a.ID) {
|
||||
// The version before is what a rollback starts (ADR 0141) and what a reader may still have
|
||||
// open; the launcher retires the host's own versions, not the apply (issue 194).
|
||||
|
||||
@@ -159,7 +159,7 @@ func TestRemovalThroughALinkUnderAHomeIsRefused(t *testing.T) {
|
||||
home, outside := aLinkedHome(t)
|
||||
link(t, outside, filepath.Join(home, ".claude"))
|
||||
_, _, err := remove(context.Background(), archHost(t), store.Applied{ID: "m.f", Type: "file",
|
||||
Target: filepath.Join(home, ".claude", "shadow")}, nil, nil)
|
||||
Target: filepath.Join(home, ".claude", "shadow")}, nil, nil, "")
|
||||
mustBeLinkRefusal(t, err)
|
||||
if _, err := os.Stat(filepath.Join(outside, "shadow")); err != nil {
|
||||
t.Fatal("removed through the link")
|
||||
|
||||
@@ -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 <id>"
|
||||
// or "hold <id>", 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
|
||||
}
|
||||
+11
-49
@@ -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.
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+1
-1
File diff suppressed because one or more lines are too long
@@ -38,7 +38,9 @@ BACKOFF="${MESH_HOST_BACKOFF:-5}"
|
||||
# How long a host on trial has to report, in seconds: to-be 45 §8's ten minutes from the apply.
|
||||
BOUND="${MESH_HOST_TRIAL_BOUND:-600}"
|
||||
TICK="${MESH_HOST_TRIAL_TICK:-5}"
|
||||
GRACE="${MESH_HOST_STOP_GRACE:-20}"
|
||||
# How long a host told to stop is waited for before it is killed, in seconds: past the 60 s the host itself waits for
|
||||
# the apply in flight to end and save (novox/hq issue 1585), so a stop never cuts a save off.
|
||||
GRACE="${MESH_HOST_STOP_GRACE:-75}"
|
||||
ONCE="${MESH_HOST_RUN_ONCE:-}" # tests run one iteration; nothing else sets this
|
||||
|
||||
ATTEMPTS="$STATE_DIR/start-attempts"
|
||||
|
||||
@@ -5,4 +5,6 @@
|
||||
name="nox-mesh-host"
|
||||
command="/usr/lib/nox-mesh-host/launch"
|
||||
supervisor="supervise-daemon"
|
||||
# The same stop bound as the systemd unit's TimeoutStopSec (issue 1585): TERM, 120 s, then KILL.
|
||||
retry="TERM/120/KILL/5"
|
||||
depend() { need net; }
|
||||
|
||||
@@ -12,6 +12,9 @@ Restart=always
|
||||
RestartSec=5s
|
||||
StateDirectory=mesh-host
|
||||
KillMode=mixed
|
||||
# Past the launcher's own stop grace (75 s), which is past the 60 s the host waits for the apply in
|
||||
# flight to end and save (novox/hq issue 1585): the service manager's kill is the last resort.
|
||||
TimeoutStopSec=120
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
Reference in New Issue
Block a user