amadan
moves outboxes: an absolute data directory, so the ancestor sync reaches it (final re-review 1, I2)
With CARLOS_DATA_DIR=".", the ancestor walk stopped at the entry's own directory. A retry after a failed ancestor sync found the new directories already there and acknowledged the entry without syncing them. armMovesPass now resolves the data directory once, with filepath.Abs (movesDataDir), so every path the worker derives is under an absolute root and the walk ends there. The outbox helpers refuse a base that is not absolute and clean, so a path that could not be resolved fails closed instead of acknowledging. The regression runs from a temporary working directory with CARLOS_DATA_DIR=".", fails an ancestor sync, and checks that the retry syncs every new directory up to the data directory.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+74
−2
internal/host/movesevents_test.go+40 −2internal/host/moveslogoutbox.go+16 −0internal/host/movespass.go+18 −0
diff --git a/internal/host/movesevents_test.go b/internal/host/movesevents_test.go| index 9b7bb9ea5..17c941e8b 100644 |
| --- a/internal/host/movesevents_test.go |
| +++ b/internal/host/movesevents_test.go |
| @@ -697,10 +697,16 @@ func spyOutboxSync(t *testing.T) *syncSpy { |
| spy := &syncSpy{fail: map[string]bool{}, synced: map[string]int{}} |
| saved := outboxSync |
| outboxSync = func(f *os.File) error { |
| - if spy.fail[f.Name()] { |
| + // Keyed by the absolute path, so a relative data directory's |
| + // syncs are named the same way. |
| + name := f.Name() |
| + if abs, err := filepath.Abs(name); err == nil { |
| + name = abs |
| + } |
| + if spy.fail[name] { |
| return errors.New("injected sync failure") |
| } |
| - spy.synced[f.Name()]++ |
| + spy.synced[name]++ |
| return f.Sync() |
| } |
| t.Cleanup(func() { outboxSync = saved }) |
| @@ -782,3 +788,35 @@ func TestLogOutboxIsDurableBeforeItAcknowledges(t *testing.T) { |
| } |
| }) |
| } |
| + |
| +// Final re-review 1, I2: the worker is built with its data directory made |
| +// absolute (movesDataDir), so the outbox's ancestor walk always ends at that |
| +// root. With CARLOS_DATA_DIR=".", an ancestor sync that fails leaves the new |
| +// directories behind; the retry finds them there and still syncs every one |
| +// of them, up to the data directory, before it acknowledges the entry. |
| +func TestLogOutboxUnderARelativeDataDir(t *testing.T) { |
| + spy := spyOutboxSync(t) |
| + t.Chdir(t.TempDir()) |
| + t.Setenv("CARLOS_DATA_DIR", ".") |
| + root, err := filepath.Abs(".") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + w := outboxWorker(movesDataDir(dataDir())) |
| + rec := doneFlow()[0] |
| + ev := moves.LogEvent{V: 1, Op: mvEvMoveID, Host: rec.Host, Event: "requested", Identity: "requested"} |
| + spy.fail[filepath.Join(root, ".moves")] = true |
| + if err := w.queueLog(rec, ev); err == nil { |
| + t.Fatal("an entry whose new directory's sync failed was acknowledged") |
| + } |
| + spy.reset() |
| + if err := w.queueLog(rec, ev); err != nil { |
| + t.Fatal(err) |
| + } |
| + out := moveLogOutboxDir(root) |
| + for _, d := range []string{root, filepath.Join(root, ".moves"), out, filepath.Join(out, rec.Host), filepath.Join(out, rec.Host, mvEvMoveID)} { |
| + if spy.synced[d] == 0 { |
| + t.Errorf("the retry acknowledged the entry without syncing %s; synced %v", d, spy.synced) |
| + } |
| + } |
| +} |
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go| index b49f572e7..7091e5e5e 100644 |
| --- a/internal/host/moveslogoutbox.go |
| +++ b/internal/host/moveslogoutbox.go |
| @@ -85,6 +85,9 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| // 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) { |
| + if err := outboxBase(base); err != nil { |
| + return false, err |
| + } |
| path := filepath.Join(dir, name) |
| if _, err := os.Lstat(path); err == nil { |
| return false, outboxSettle(base, dir, path) |
| @@ -129,6 +132,16 @@ func outboxCreate(base, dir, name string, body []byte) (created bool, err error) |
| return true, nil |
| } |
| +// outboxBase refuses a data directory that is not absolute and clean: the |
| +// ancestor walk (outboxSyncChain) ends only at such a root. The worker |
| +// resolves its own once, at construction (movesDataDir). |
| +func outboxBase(base string) error { |
| + if !filepath.IsAbs(base) || filepath.Clean(base) != base { |
| + return fmt.Errorf("the outbox's data directory %q is not an absolute, clean path", base) |
| + } |
| + return 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. |
| @@ -206,6 +219,9 @@ func outboxMissingDirs(dir string) ([]string, error) { |
| // 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 { |
| + if err := outboxBase(base); err != nil { |
| + return err |
| + } |
| tmp, err := os.CreateTemp(dir, ".pending-*") |
| if err != nil { |
| return err |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index fd9a0703c..ad6e3589f 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -197,6 +197,7 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro |
| // rollout sink, which the move log tells the app's members through (Plan 6 |
| // Task 3). |
| func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, providers ClaimProviders, events activator.EventSink) { |
| + dataDir = movesDataDir(dataDir) |
| readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv) |
| if !ok { |
| log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta) |
| @@ -265,6 +266,23 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| armedMovesWorker.Store(w) |
| } |
| +// movesDataDir is the data directory the moves worker is built with, |
| +// resolved once to an absolute, cleaned path (final re-review 1, I2): the |
| +// outboxes under it sync every directory from an entry's own up to it, and |
| +// that walk ends only at an absolute root. A CARLOS_DATA_DIR of "." or any |
| +// relative path names the same directory as before. If the working |
| +// directory cannot be read the path is kept as given and logged; the |
| +// outboxes then refuse every write (outboxCreate), so nothing is |
| +// acknowledged that is not durable. |
| +func movesDataDir(dd string) string { |
| + abs, err := filepath.Abs(dd) |
| + if err != nil { |
| + log.Printf("agent: moves: the data directory %q cannot be made absolute, so the moves outboxes keep nothing: %v", dd, err) |
| + return dd |
| + } |
| + return abs |
| +} |
| + |
| // armPhase1 wires the Phase 1 hooks (moves design §7.3). Only with the |
| // switch on: off, none is set and eb.Strict stays nil. writes is the |
| // agent's host-write configuration (the claim wiring's): the landing's |