amadan
moves-phase1-6 Task 3 fix round 3 (4): the box outbox keeps the first entry per identity
queueLog links the kept file into place create-exclusive, so a transition re-observed after a restart during an outage never replaces the first entry's bytes or time. TestLogOutboxKeepsTheFirstEntry.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+43
−2
internal/host/movesevents_test.go+35 −0internal/host/moveslogoutbox.go+8 −2
diff --git a/internal/host/movesevents_test.go b/internal/host/movesevents_test.go| index 549cb2663..5310b5df8 100644 |
| --- a/internal/host/movesevents_test.go |
| +++ b/internal/host/movesevents_test.go |
| @@ -648,3 +648,38 @@ func TestLogMemoIsBounded(t *testing.T) { |
| t.Fatalf("%d entries after forget", m.len()) |
| } |
| } |
| + |
| +// Fix round 3, ruling 4: the outbox keeps the first entry queued for an |
| +// identity. A restart during the outage forgets what was handed off, so the |
| +// next pass sees the transition again and builds it again, later; that never |
| +// replaces the kept bytes or their time. |
| +func TestLogOutboxKeepsTheFirstEntry(t *testing.T) { |
| + ctx := context.Background() |
| + dataDir := t.TempDir() |
| + s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true} |
| + rec := doneFlow()[0] |
| + w := outboxWorker(dataDir) |
| + w.observeMove(ctx, s, rec) |
| + kept := filepath.Join(moveLogOutboxDir(dataDir), rec.Host, mvEvMoveID, "requested.json") |
| + first, err := os.ReadFile(kept) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + restarted := outboxWorker(dataDir) |
| + restarted.cfg.Now = func() time.Time { return time.Unix(1_800_000_900, 0).UTC() } |
| + restarted.flushLogOutbox(ctx, s) // still failing |
| + restarted.observeMove(ctx, s, rec) |
| + again, err := os.ReadFile(kept) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if string(again) != string(first) { |
| + t.Fatalf("the kept entry was replaced:\n%s\n%s", first, again) |
| + } |
| + s.fail = false |
| + restarted.flushLogOutbox(ctx, s) |
| + got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, mvEvMoveID)) |
| + if err != nil || len(got) != 1 || !got[0].At.Equal(time.Unix(1_800_000_100, 0)) { |
| + t.Fatalf("log %+v %v, want the first entry's time", got, err) |
| + } |
| +} |
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go| index ef969b67a..5c0f1af1e 100644 |
| --- a/internal/host/moveslogoutbox.go |
| +++ b/internal/host/moveslogoutbox.go |
| @@ -46,7 +46,10 @@ func outboxSegment(s string) bool { |
| } |
| // queueLog keeps ev, for rec's app and host, in the outbox: written to a |
| -// temporary file, synced, and renamed into place. |
| +// temporary file, synced, and linked into place create-exclusive. The first |
| +// entry kept for an identity stays: a transition seen again after a restart |
| +// (built again, later) never replaces its bytes or its time (fix round 3, |
| +// ruling 4). |
| func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| if w.cfg.DataDir == "" { |
| return errors.New("no data directory for the move log's outbox") |
| @@ -82,7 +85,10 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| if err := tmp.Close(); err != nil { |
| return err |
| } |
| - if err := os.Rename(tmp.Name(), filepath.Join(dir, ev.Identity+".json")); err != nil { |
| + if err := os.Link(tmp.Name(), filepath.Join(dir, ev.Identity+".json")); err != nil { |
| + if errors.Is(err, fs.ErrExist) { |
| + return nil |
| + } |
| return err |
| } |
| if d, err := os.Open(dir); err == nil { |