amadan

carlos / platform Public

moves outboxes: acknowledge an entry only once it is durable (final review I2)

The shared outbox helper now returns directory open and sync failures
instead of ignoring them. It syncs every directory it creates into its
parent, and after a link or rename it syncs the entry's directory and each
one above it up to the data directory. An entry it finds already on disk,
which may be an earlier try whose sync failed, is synced (file and
directories) before it counts, and the page outbox's delivered mark does
the same. A caller that gets an error treats the entry as not handed off,
so the next sighting tries again.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 14c7100a0e38a5be203cd563e75c7bb8ae17aa85 parent 3eeb1ed
4 files changed, +260 −20
  • internal/host/movesevents_test.go +98 −0
  • internal/host/moveslogoutbox.go +103 −16
  • internal/host/movespaging.go +6 −4
  • internal/host/movespaging_test.go +53 −0
diff --git a/internal/host/movesevents_test.go b/internal/host/movesevents_test.go
index 5310b5df8..a0ead14b4 100644
--- a/internal/host/movesevents_test.go
+++ b/internal/host/movesevents_test.go
@@ -683,3 +683,101 @@ func TestLogOutboxKeepsTheFirstEntry(t *testing.T) {
t.Fatalf("log %+v %v, want the first entry's time", got, err)
}
}
+
+// syncSpy stands in for outboxSync: it records every path synced, and
+// fails the paths in fail.
+type syncSpy struct {
+ fail map[string]bool
+ synced map[string]int
+}
+
+func spyOutboxSync(t *testing.T) *syncSpy {
+ t.Helper()
+ spy := &syncSpy{fail: map[string]bool{}, synced: map[string]int{}}
+ saved := outboxSync
+ outboxSync = func(f *os.File) error {
+ if spy.fail[f.Name()] {
+ return errors.New("injected sync failure")
+ }
+ spy.synced[f.Name()]++
+ return f.Sync()
+ }
+ t.Cleanup(func() { outboxSync = saved })
+ return spy
+}
+
+// reset forgets what was synced and fails nothing.
+func (s *syncSpy) reset() {
+ s.fail, s.synced = map[string]bool{}, map[string]int{}
+}
+
+// Final review I2: the log outbox acknowledges an entry only once it is
+// durable. Every directory the write created is synced into its parent, a
+// failed directory sync is an error, and an entry already there (an earlier
+// try whose sync failed) is synced, file and directory, before it counts.
+// The caller does not remember an entry the outbox could not make durable,
+// so the next sighting tries again.
+func TestLogOutboxIsDurableBeforeItAcknowledges(t *testing.T) {
+ ctx := context.Background()
+ spy := spyOutboxSync(t)
+ rec := doneFlow()[0]
+ ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"}
+
+ t.Run("new directories are synced into their parents", func(t *testing.T) {
+ spy.reset()
+ dd := t.TempDir()
+ if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
+ t.Fatal(err)
+ }
+ root := moveLogOutboxDir(dd)
+ for _, d := range []string{dd, filepath.Join(dd, ".moves"), root, filepath.Join(root, rec.Host), filepath.Join(root, rec.Host, mvEvMoveID)} {
+ if spy.synced[d] == 0 {
+ t.Errorf("%s was not synced; synced %v", d, spy.synced)
+ }
+ }
+ })
+
+ t.Run("a failed directory sync is an error, and the retry syncs the existing entry", func(t *testing.T) {
+ spy.reset()
+ dd := t.TempDir()
+ dir := filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)
+ spy.fail[dir] = true
+ if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
+ t.Fatal("an entry whose directory sync failed was acknowledged")
+ }
+ file := filepath.Join(dir, "requested.json")
+ if _, err := os.Lstat(file); err != nil {
+ t.Fatalf("the fixture wants the entry linked before the failed sync: %v", err)
+ }
+ spy.reset()
+ spy.fail[file] = true
+ if err := outboxWorker(dd).queueLog(rec, ev); err == nil {
+ t.Fatal("an existing entry whose file sync failed was acknowledged")
+ }
+ spy.reset()
+ if err := outboxWorker(dd).queueLog(rec, ev); err != nil {
+ t.Fatal(err)
+ }
+ if spy.synced[file] == 0 || spy.synced[dir] == 0 {
+ t.Fatalf("an existing entry was acknowledged without syncing it and its directory: %v", spy.synced)
+ }
+ })
+
+ t.Run("the caller retries an entry the outbox could not make durable", func(t *testing.T) {
+ spy.reset()
+ dd := t.TempDir()
+ s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true}
+ w := outboxWorker(dd)
+ spy.fail[filepath.Join(moveLogOutboxDir(dd), rec.Host, mvEvMoveID)] = true
+ w.observeMove(ctx, s, rec)
+ key := rec.Host + "/" + mvEvMoveID + "/requested"
+ if w.logged.seen(key) {
+ t.Fatal("an entry the outbox could not make durable counts as handed off")
+ }
+ spy.reset()
+ w.observeMove(ctx, s, rec)
+ if !w.logged.seen(key) {
+ t.Fatal("the next sighting did not hand the entry off")
+ }
+ })
+}
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go
index 0b42056a0..b49f572e7 100644
--- a/internal/host/moveslogoutbox.go
+++ b/internal/host/moveslogoutbox.go
@@ -65,7 +65,7 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error {
if err != nil {
return err
}
- _, err = outboxCreate(filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op), ev.Identity+".json", body)
+ _, err = outboxCreate(w.cfg.DataDir, filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op), ev.Identity+".json", body)
return err
}
@@ -76,13 +76,31 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error {
// anything is written, so storage that has turned full or read-only still
// answers for it (Plan 6 Tasks 4+5 fix round 3). The move log's outbox and
// the page outbox (movespaging.go) share it.
-func outboxCreate(dir, name string, body []byte) (created bool, err error) {
- if _, err := os.Lstat(filepath.Join(dir, name)); err == nil {
- return false, nil
+//
+// base is the data directory the outbox lives under. Nothing is
+// acknowledged until it is durable (final review I2): every directory the
+// call creates is synced into its parent, and the entry's directory and
+// each one above it up to base are synced after the link, any failure
+// returned. An entry found already there may be an earlier try's whose sync
+// failed after the link, so it counts only once it, its directory and those
+// above it are synced now.
+func outboxCreate(base, dir, name string, body []byte) (created bool, err error) {
+ path := filepath.Join(dir, name)
+ if _, err := os.Lstat(path); err == nil {
+ return false, outboxSettle(base, dir, path)
+ }
+ fresh, err := outboxMissingDirs(dir)
+ if err != nil {
+ return false, err
}
if err := os.MkdirAll(dir, 0o700); err != nil {
return false, err
}
+ for _, c := range fresh {
+ if err := outboxSyncPath(filepath.Dir(c)); err != nil {
+ return false, err
+ }
+ }
tmp, err := os.CreateTemp(dir, ".pending-*")
if err != nil {
return false, err
@@ -99,22 +117,95 @@ func outboxCreate(dir, name string, body []byte) (created bool, err error) {
if err := tmp.Close(); err != nil {
return false, err
}
- if err := os.Link(tmp.Name(), filepath.Join(dir, name)); err != nil {
+ if err := os.Link(tmp.Name(), path); err != nil {
if errors.Is(err, fs.ErrExist) {
- return false, nil
+ return false, outboxSettle(base, dir, path)
}
return false, err
}
- if d, err := os.Open(dir); err == nil {
- _ = d.Sync()
- d.Close()
+ if err := outboxSyncChain(base, dir); err != nil {
+ return true, err
}
return true, nil
}
+// outboxSync syncs one open file or directory of an outbox. It is a
+// variable only so the host's tests can fail it; nothing in production
+// configuration reaches it.
+var outboxSync = func(f *os.File) error { return f.Sync() }
+
+// outboxSyncPath syncs the file or directory at p.
+func outboxSyncPath(p string) error {
+ f, err := os.Open(p)
+ if err != nil {
+ return err
+ }
+ serr := outboxSync(f)
+ if cerr := f.Close(); serr == nil {
+ serr = cerr
+ }
+ return serr
+}
+
+// outboxSyncChain syncs dir and each directory above it up to and
+// including base, so the entries naming them are all on disk. A dir outside
+// base is synced alone.
+func outboxSyncChain(base, dir string) error {
+ base, dir = filepath.Clean(base), filepath.Clean(dir)
+ for p := dir; ; {
+ if err := outboxSyncPath(p); err != nil {
+ return err
+ }
+ q := filepath.Dir(p)
+ if p == base || q == p || !outboxWithin(base, q) {
+ return nil
+ }
+ p = q
+ }
+}
+
+// outboxWithin reports whether p is base or under it.
+func outboxWithin(base, p string) bool {
+ if base == "" || base == "." {
+ return false
+ }
+ rel, err := filepath.Rel(base, p)
+ return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))
+}
+
+// outboxSettle makes an entry found on disk durable before it counts: the
+// file, its directory and those above it up to base.
+func outboxSettle(base, dir, path string) error {
+ if err := outboxSyncPath(path); err != nil {
+ return err
+ }
+ return outboxSyncChain(base, dir)
+}
+
+// outboxMissingDirs lists the directories at and above dir that do not
+// exist yet, deepest first: the ones MkdirAll is about to create.
+func outboxMissingDirs(dir string) ([]string, error) {
+ var out []string
+ for p := filepath.Clean(dir); ; p = filepath.Dir(p) {
+ _, err := os.Lstat(p)
+ if err == nil {
+ return out, nil
+ }
+ if !errors.Is(err, fs.ErrNotExist) {
+ return nil, err
+ }
+ out = append(out, p)
+ if filepath.Dir(p) == p {
+ return out, nil
+ }
+ }
+}
+
// outboxReplace replaces dir/name with body: a temporary file, synced and
-// renamed over it, so a crash leaves the old file or the new one.
-func outboxReplace(dir, name string, body []byte) error {
+// renamed over it, so a crash leaves the old file or the new one. Like
+// outboxCreate, it succeeds only once the rename is durable: dir and each
+// directory above it up to base synced (final review I2).
+func outboxReplace(base, dir, name string, body []byte) error {
tmp, err := os.CreateTemp(dir, ".pending-*")
if err != nil {
return err
@@ -134,11 +225,7 @@ func outboxReplace(dir, name string, body []byte) error {
if err := os.Rename(tmp.Name(), filepath.Join(dir, name)); err != nil {
return err
}
- if d, err := os.Open(dir); err == nil {
- _ = d.Sync()
- d.Close()
- }
- return nil
+ return outboxSyncChain(base, dir)
}
// flushLogOutbox retries every kept entry through the LogEvent hook, into s.
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go
index 30e0cb631..10e7353fa 100644
--- a/internal/host/movespaging.go
+++ b/internal/host/movespaging.go
@@ -214,7 +214,7 @@ func writePageEntry(dataDir string, e pageEntry) error {
if err != nil {
return err
}
- return outboxReplace(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
+ return outboxReplace(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
}
// pageEntryPath is an entry's file.
@@ -271,7 +271,7 @@ func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, hos
b, err := json.Marshal(e)
created := false
if err == nil {
- created, err = outboxCreate(pageEntryDir(dataDir, host, op), rule+".json", b)
+ created, err = outboxCreate(dataDir, pageEntryDir(dataDir, host, op), rule+".json", b)
}
if err == nil && !created {
pageOutboxMu.Unlock()
@@ -324,7 +324,9 @@ func markPageDelivered(dataDir string, e pageEntry, now time.Time) error {
switch {
case err == nil:
if kept.Delivered != nil {
- return nil
+ // Marked before, perhaps by a write whose sync failed: it
+ // counts once it is durable (final review I2).
+ return outboxSettle(dataDir, pageEntryDir(dataDir, e.Host, e.Op), pageEntryPath(dataDir, e.Host, e.Op, e.Rule))
}
kept.Delivered = &at
return writePageEntry(dataDir, kept)
@@ -335,7 +337,7 @@ func markPageDelivered(dataDir string, e pageEntry, now time.Time) error {
if err != nil {
return err
}
- _, err = outboxCreate(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
+ _, err = outboxCreate(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
return err
}
return err
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go
index f647e9769..6f4899a7f 100644
--- a/internal/host/movespaging_test.go
+++ b/internal/host/movespaging_test.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"io/fs"
@@ -758,3 +759,55 @@ func TestSwitchOffPassLeavesTheOutboxes(t *testing.T) {
t.Fatalf("with the switch on again: pages %+v, log entries %+v", pages, logged)
}
}
+
+// Final review I2: the page outbox's replace and its delivered mark are
+// durable before they count. A failed directory sync after the replace is
+// an error, and a delivered entry already on disk is synced before the mark
+// is acknowledged.
+func TestPageOutboxIsDurableBeforeItAcknowledges(t *testing.T) {
+ spy := spyOutboxSync(t)
+ dd := t.TempDir()
+ now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
+ e := pageEntry{V: 1, Host: p1Host, Op: p1ID, Rule: "stuck-0", Level: "incident", Subject: "s", Body: "b", At: now, Tries: 1, Next: now.Add(time.Minute)}
+ if err := writePageEntryFixture(dd, e); err != nil {
+ t.Fatal(err)
+ }
+ dir := pageEntryDir(dd, p1Host, p1ID)
+ file := pageEntryPath(dd, p1Host, p1ID, "stuck-0")
+
+ spy.fail[dir] = true
+ e.Tries = 2
+ if err := writePageEntry(dd, e); err == nil {
+ t.Fatal("a replace whose directory sync failed was acknowledged")
+ }
+
+ spy.reset()
+ if err := markPageDelivered(dd, e, now); err != nil {
+ t.Fatal(err)
+ }
+ spy.reset()
+ spy.fail[file] = true
+ if err := markPageDelivered(dd, e, now); err == nil {
+ t.Fatal("a delivered entry whose sync failed was acknowledged")
+ }
+ spy.reset()
+ if err := markPageDelivered(dd, e, now); err != nil {
+ t.Fatal(err)
+ }
+ if spy.synced[file] == 0 || spy.synced[dir] == 0 {
+ t.Fatalf("a delivered entry was acknowledged without syncing it and its directory: %v", spy.synced)
+ }
+}
+
+// writePageEntryFixture writes a page entry with no sync spy in the way.
+func writePageEntryFixture(dataDir string, e pageEntry) error {
+ b, err := json.Marshal(e)
+ if err != nil {
+ return err
+ }
+ dir := pageEntryDir(dataDir, e.Host, e.Op)
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ return err
+ }
+ return os.WriteFile(filepath.Join(dir, e.Rule+".json"), b, 0o600)
+}