amadan
host: two-phase move paging, never suppressed by a journal failure
moves-phase1-6 Tasks 4+5 fix round 1, ruling 4. pageOnce journals pending:<rule> before the page and paged:<rule> only once the pager took it; a call that finds pending without paged sends it again. A journal that cannot record pending pages anyway, and one that cannot record paged leaves the rule to be sent again, so no journal failure, a directory sync failing after the replace included, suppresses a rule (at least once). moves.SetJournalSyncForTest injects a failure of the directory sync after the replace (journal.go), for the test. The bad/host subtest, which failed name validation before any write, is replaced by an injected sync failure, a refused delivery retried, and an unreadable journal that still pages. No censused unit changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+125
−23
internal/host/movespaging.go+36 −17internal/host/movespaging_test.go+74 −5internal/moves/journal.go+15 −1
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go| index c3c3b11c8..cdfdcdf9f 100644 |
| --- a/internal/host/movespaging.go |
| +++ b/internal/host/movespaging.go |
| @@ -39,14 +39,17 @@ import ( |
| // an overdue close and an overdue abort. A move converted to relocating with |
| // no live box in its region (M-C22) is Plan 5 Task 7a's page. |
| // |
| -// A rule's rate state is the op journal on this box (moves.Journal, step |
| -// paged:<rule>, the time it was marked), so a restarted agent does not page |
| -// it again. It is read and written under the journal's one lock |
| -// (moves.UpdateJournal), which every other writer of the journal shares. The step is written before the page is sent: a crash between the |
| -// two, or a send that fails (logged), pages nothing more, as the console's |
| -// move_pages records an attempt whether or not delivery worked (pages are |
| -// best effort). A journal that cannot be read or written sends nothing, so a |
| -// rule that fires every pass can never flood. |
| +// A rule's rate state is the op journal on this box (moves.Journal), read |
| +// and written under the journal's one lock (moves.UpdateJournal), which every |
| +// other writer of the journal shares, so a restarted agent does not page it |
| +// again. Paging is two-phase (fix round 1, ruling 4): pending:<rule> is |
| +// journalled before the page, and paged:<rule> only once the pager took it. |
| +// A call that finds paged sends nothing; one that finds pending without paged |
| +// sends it again. No journal failure suppresses a rule, a directory sync that |
| +// fails after the replace included: a call that cannot record pending pages |
| +// anyway, and one that cannot record paged leaves the rule to be sent again. |
| +// At least once, then: a rare second page beats a lost one. A page the pager |
| +// refuses stays pending, and is sent again when its rule fires again. |
| // |
| // Unwired (nil Page, as production is until Plan 5 Task 9b wires it), a rule |
| // is never silent: the agent's log says so loudly, once per rule per move |
| @@ -62,6 +65,9 @@ var pagingLogf = log.Printf |
| const ( |
| pagedStep = "paged:" |
| unwiredStep = "unwired:" |
| + // pendingStep is journalled before a rule's page is sent (fix round 1, |
| + // ruling 4); paged:<rule> follows once the pager took it. |
| + pendingStep = "pending:" |
| ) |
| // pageOutcome is what became of one rule's page. |
| @@ -99,32 +105,45 @@ func pageRuleStuck(m *provision.MoveState) string { |
| return "stuck-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10) |
| } |
| -// pageOnce sends one page for rule unless the journal of (host, op) under |
| -// dataDir says it was sent: it marks the journal, then pages. Unwired, it |
| -// warns instead (unwiredOnce). |
| +// pageOnce sends rule's page unless the journal of (host, op) under dataDir |
| +// says it was sent (paged:<rule>), in two phases: pending:<rule> before the |
| +// page, paged:<rule> after it. A journal that cannot be read or written never |
| +// stops the page, so the rule is never suppressed; it may come again. |
| +// Unwired, it warns instead (unwiredOnce). |
| func pageOnce(ctx context.Context, page func(ctx context.Context, level, subject, body string) error, dataDir, host, op, rule string, now time.Time, level, subject, body string) pageOutcome { |
| if page == nil { |
| unwiredOnce(dataDir, host, op, rule, now) |
| return pageUnwired |
| } |
| + stamp := now.UTC().Format(time.RFC3339) |
| + paged := false |
| err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error { |
| if j.Done(pagedStep + rule) { |
| + paged = true |
| return errJournalUnchanged |
| } |
| - j.Mark(pagedStep+rule, now.UTC().Format(time.RFC3339)) |
| + if j.Done(pendingStep + rule) { |
| + return errJournalUnchanged // pending and never sent: send it again |
| + } |
| + j.Mark(pendingStep+rule, stamp) |
| return nil |
| }) |
| - if errors.Is(err, errJournalUnchanged) { |
| + if paged { |
| return pagePagedBefore |
| } |
| - if err != nil { |
| - pagingLogf("moves: %s: the %s page of %s is not sent: its journal: %v", host, rule, op, err) |
| - return pageUndelivered |
| + if err != nil && !errors.Is(err, errJournalUnchanged) { |
| + pagingLogf("moves: %s: the %s page of %s is sent without its pending mark: its journal: %v", host, rule, op, err) |
| } |
| if err := page(ctx, level, subject, body); err != nil { |
| - pagingLogf("moves: %s: the %s page of %s: %v", host, rule, op, err) |
| + pagingLogf("moves: %s: the %s page of %s was not delivered, and is sent again when the rule fires again: %v", host, rule, op, err) |
| return pageUndelivered |
| } |
| + if err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error { |
| + j.Mark(pagedStep+rule, stamp) |
| + return nil |
| + }); err != nil { |
| + pagingLogf("moves: %s: the %s page of %s was sent, and its journal may not say so, so it may be sent again: %v", host, rule, op, err) |
| + } |
| return pageSent |
| } |
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go| index 67ab4b747..74872e5c5 100644 |
| --- a/internal/host/movespaging_test.go |
| +++ b/internal/host/movespaging_test.go |
| @@ -4,6 +4,8 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "os" |
| + "path/filepath" |
| "strings" |
| "testing" |
| "time" |
| @@ -311,13 +313,80 @@ func TestPagingIsRateLimited(t *testing.T) { |
| } |
| }) |
| - t.Run("a journal that cannot be written sends nothing", func(t *testing.T) { |
| + // Fix round 1, ruling 4: paging is two-phase. pending:<rule> is |
| + // journalled before the page and paged:<rule> only once the pager took |
| + // it; a later call that finds pending without paged sends it again. No |
| + // journal failure suppresses a rule, a directory sync that fails after |
| + // the replace included: a page may come twice, never not at all. |
| + t.Run("two-phase: a sync failure after the replace never suppresses a page", func(t *testing.T) { |
| + syncErr := errors.New("injected directory sync failure") |
| + restore := moves.SetJournalSyncForTest(func(string) error { return syncErr }) |
| + defer restore() |
| + dd := t.TempDir() |
| + var pages pagesSeen |
| + now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC) |
| + // Every journal write replaces the file and then fails its sync. |
| + if o := pageOnce(ctx, pages.hook(), dd, p1Host, p1ID, pageRuleStray, now, "incident", "s", "b"); o != pageSent { |
| + t.Fatalf("the first call: %s, want the page sent", o) |
| + } |
| + if o := pageOnce(ctx, pages.hook(), dd, p1Host, p1ID, pageRuleStray, now, "incident", "s", "b"); o != pagePagedBefore { |
| + t.Fatalf("the second call: %s, want paged before (the replace landed)", o) |
| + } |
| + if len(pages) != 1 { |
| + t.Fatalf("pages %+v, want one", pages) |
| + } |
| + j, err := moves.ReadJournal(dd, p1Host, p1ID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !j.Done(pendingStep+pageRuleStray) || !j.Done(pagedStep+pageRuleStray) { |
| + t.Fatalf("the journal: %+v, want pending and paged", j.Steps) |
| + } |
| + |
| + // A page the pager refuses stays pending, and the next call sends it. |
| + failing := true |
| + flaky := func(_ context.Context, level, subject, body string) error { |
| + if failing { |
| + return errors.New("the webhook is down") |
| + } |
| + pages = append(pages, struct{ level, subject, body string }{level, subject, body}) |
| + return nil |
| + } |
| + if o := pageOnce(ctx, flaky, dd, p1Host, p1ID, pageRuleStartCheck, now, "incident", "s2", "b2"); o != pageUndelivered { |
| + t.Fatalf("a refused delivery: %s", o) |
| + } |
| + if j, _ := moves.ReadJournal(dd, p1Host, p1ID); !j.Done(pendingStep+pageRuleStartCheck) || j.Done(pagedStep+pageRuleStartCheck) { |
| + t.Fatalf("after a refused delivery: %+v, want pending without paged", j.Steps) |
| + } |
| + failing = false |
| + if o := pageOnce(ctx, flaky, dd, p1Host, p1ID, pageRuleStartCheck, now, "incident", "s2", "b2"); o != pageSent { |
| + t.Fatalf("the retry: %s, want sent", o) |
| + } |
| + if o := pageOnce(ctx, flaky, dd, p1Host, p1ID, pageRuleStartCheck, now, "incident", "s2", "b2"); o != pagePagedBefore { |
| + t.Fatalf("after the retry: %s, want paged before", o) |
| + } |
| + if n, _ := pagesWith(pages, "s2"); n != 1 { |
| + t.Fatalf("the retried page delivered %d times, want once: %+v", n, pages) |
| + } |
| + }) |
| + |
| + t.Run("two-phase: a journal that cannot be read still pages", func(t *testing.T) { |
| + dd := t.TempDir() |
| + dir := moves.JournalDir(dd, p1Host, p1ID) |
| + if err := os.MkdirAll(dir, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(dir, "journal.json"), []byte("not json"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| var pages pagesSeen |
| - if pageOnce(ctx, pages.hook(), t.TempDir(), "bad/host", p1ID, pageRuleStray, time.Now(), "incident", "s", "b") == pageSent || len(pages) != 0 { |
| - t.Fatalf("paged without a journal: %+v", pages) |
| + for i := 0; i < 2; i++ { |
| + if o := pageOnce(ctx, pages.hook(), dd, p1Host, p1ID, pageRuleStray, time.Now(), "incident", "s", "b"); o != pageSent { |
| + t.Fatalf("call %d: %s, want sent (at least once)", i, o) |
| + } |
| } |
| - if pageOnce(ctx, nil, t.TempDir(), p1Host, p1ID, pageRuleStray, time.Now(), "incident", "s", "b") != pageUnwired { |
| - t.Fatal("an unwired pager reported a page") |
| + if len(pages) != 2 { |
| + t.Fatalf("pages %+v, want one per call while the journal cannot remember", pages) |
| } |
| }) |
| } |
diff --git a/internal/moves/journal.go b/internal/moves/journal.go| index d0a3410a3..1d24ae40b 100644 |
| --- a/internal/moves/journal.go |
| +++ b/internal/moves/journal.go |
| @@ -225,7 +225,21 @@ func (j *Journal) write(dataDir string) error { |
| return err |
| } |
| j.read = j.readSteps() |
| - return syncDir(dir) |
| + return journalReplacedSync(dir) |
| +} |
| + |
| +// journalReplacedSync is the directory sync after the replace: the journal |
| +// on disk is already the new one when it fails. A variable only so a test can |
| +// fail it (SetJournalSyncForTest). |
| +var journalReplacedSync = syncDir |
| + |
| +// SetJournalSyncForTest makes f the directory sync after a journal's |
| +// replace, for a test of what a sync failure after the rename leaves (Plan 6 |
| +// fix round 1, ruling 4), and returns the restore. Tests only. |
| +func SetJournalSyncForTest(f func(dir string) error) (restore func()) { |
| + saved := journalReplacedSync |
| + journalReplacedSync = f |
| + return func() { journalReplacedSync = saved } |
| } |
| // missingDirs lists the directories at and above dir that do not exist yet, |