amadan
host: a box page outbox replaces the journal marks for move pages
moves-phase1-6 Tasks 4+5 fix round 2, ruling 1. Paging leaves the move journal. A page is one file, <DataDir>/.moves/.page-outbox/<host>/ <move>/<rule>.json, created exclusively (outboxCreate, the helper the move log's outbox now shares) when its rule first fires: the file is the once-per-entry limit. It carries the page and its backoff (one minute, doubling, to 30). Every moves pass (flushPageOutbox) sends an undelivered entry again once its backoff is over, whatever its move and journals have done since, until delivered; a delivered entry stays as a tombstone until the move is retired on the box (no record carries it and its journal is gone). If the outbox cannot keep a page it is sent at once, and if that fails it is kept in memory for every pass, loudly. M-C20 goes to the agent's log when the box detects the stray process. The pending:/paged:/unwired: journal marks, the journal sweep and SetJournalSyncForTest are gone; an unwired pager still warns once per rule per move (the worker's memory). Census: recoverRun.finalize's B8 digest (pageRolledBack takes no journal now): 34fd48b2... to f2bab5b8....
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
14 files changed,
+539
−455
internal/host/movesfinal.go+4 −12internal/host/moveslogoutbox.go+44 −6internal/host/movespaging.go+294 −190internal/host/movespaging_test.go+171 −162internal/host/movespass.go+8 −2internal/host/movesprobeowners.go+2 −0internal/host/movesrecover.go+1 −1internal/host/movesrecover_test.go+9 −4internal/host/movesrevert.go+1 −1internal/host/movesrevert_test.go+1 −1internal/host/movesrollback_test.go+1 −1internal/moves/journal.go+1 −21internal/moves/journal_lock_test.go+0 −52internal/provision/plan9_b8_census_test.go+2 −2
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go| index bb29c883f..6c9c328b5 100644 |
| --- a/internal/host/movesfinal.go |
| +++ b/internal/host/movesfinal.go |
| @@ -779,7 +779,7 @@ func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step |
| // A journal that has not reached the stamp step (no source-parked step: a |
| // landing's, a cut's, a finalization still under way) is never touched, and |
| // nothing is ever deleted: a journal is only renamed into its retirement. |
| -func (w *movesWorker) sweepFinalJournals(ctx context.Context, recs []provision.Record) { |
| +func (w *movesWorker) sweepFinalJournals(recs []provision.Record) { |
| dd := w.cfg.DataDir |
| if dd == "" { |
| return |
| @@ -810,27 +810,19 @@ func (w *movesWorker) sweepFinalJournals(ctx context.Context, recs []provision.R |
| if !od.IsDir() || !journalComponent(op) { |
| continue |
| } |
| - if err := w.sweepFinalJournal(ctx, dd, host, op, byHost[host]); err != nil { |
| + if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil { |
| w.logTrouble(host, fmt.Errorf("the op journal sweep, %s: %w", op, err)) |
| } |
| } |
| } |
| } |
| -// sweepFinalJournal is one journal of sweepFinalJournals. It first sends |
| -// again the pages the journal holds pending (retryPendingPages, |
| -// movespaging.go), so a refused page is retried for as long as its journal is |
| -// on the box. |
| -func (w *movesWorker) sweepFinalJournal(ctx context.Context, dataDir, host, op string, recs []provision.Record) error { |
| +// sweepFinalJournal is one journal of sweepFinalJournals. |
| +func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error { |
| j, err := moves.ReadJournal(dataDir, host, op) |
| if err != nil { |
| return err |
| } |
| - if w.retryPendingPages(ctx, dataDir, host, op, j) { |
| - if j, err = moves.ReadJournal(dataDir, host, op); err != nil { |
| - return err |
| - } |
| - } |
| stamp, reached := j.Steps[finalStepParked] |
| if !reached { |
| // Not a finalization's: a rollback source's recovery journal |
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go| index 5c0f1af1e..0177ea650 100644 |
| --- a/internal/host/moveslogoutbox.go |
| +++ b/internal/host/moveslogoutbox.go |
| @@ -65,10 +65,51 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| if err != nil { |
| return err |
| } |
| - dir := filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op) |
| + _, err = outboxCreate(filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op), ev.Identity+".json", body) |
| + return err |
| +} |
| + |
| +// outboxCreate keeps body as dir/name, create-exclusive: written to a |
| +// temporary file, synced, and linked into place, so a crash leaves the whole |
| +// file or none. created is false, with no error, when dir/name was already |
| +// there: the first one kept stays. 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.MkdirAll(dir, 0o700); err != nil { |
| - return err |
| + return false, err |
| } |
| + tmp, err := os.CreateTemp(dir, ".pending-*") |
| + if err != nil { |
| + return false, err |
| + } |
| + defer os.Remove(tmp.Name()) |
| + if _, err := tmp.Write(body); err != nil { |
| + tmp.Close() |
| + return false, err |
| + } |
| + if err := tmp.Sync(); err != nil { |
| + tmp.Close() |
| + return false, err |
| + } |
| + if err := tmp.Close(); err != nil { |
| + return false, err |
| + } |
| + if err := os.Link(tmp.Name(), filepath.Join(dir, name)); err != nil { |
| + if errors.Is(err, fs.ErrExist) { |
| + return false, nil |
| + } |
| + return false, err |
| + } |
| + if d, err := os.Open(dir); err == nil { |
| + _ = d.Sync() |
| + d.Close() |
| + } |
| + return true, 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 { |
| tmp, err := os.CreateTemp(dir, ".pending-*") |
| if err != nil { |
| return err |
| @@ -85,10 +126,7 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| if err := tmp.Close(); err != nil { |
| return err |
| } |
| - if err := os.Link(tmp.Name(), filepath.Join(dir, ev.Identity+".json")); err != nil { |
| - if errors.Is(err, fs.ErrExist) { |
| - return nil |
| - } |
| + if err := os.Rename(tmp.Name(), filepath.Join(dir, name)); err != nil { |
| return err |
| } |
| if d, err := os.Open(dir); err == nil { |
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go| index ef697ef6f..956e202bf 100644 |
| --- a/internal/host/movespaging.go |
| +++ b/internal/host/movespaging.go |
| @@ -5,10 +5,13 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "io/fs" |
| "log" |
| - "sort" |
| + "os" |
| + "path/filepath" |
| "strconv" |
| "strings" |
| + "sync" |
| "time" |
| "github.com/carlosframework/platform/internal/alerting" |
| @@ -24,14 +27,14 @@ import ( |
| // stuck-<epoch>, the record's recovery retry epoch, which only an |
| // operator's retry advances, so a retried move that sticks again pages |
| // again and a replay of the same mark does not; |
| -// - rolled_back, a warning, once per op, where this box closes a rollback |
| -// (recoverRun.finalize) or a revert (revertRun.close); |
| +// - rolled_back, a warning, once per move, where this box closes a |
| +// rollback (recoverRun.finalize) or a revert (revertRun.close); |
| // - a stop evidence refused for a process the move cannot account for |
| -// (StepStray), an incident, once per op, from the cut's stopping: the |
| -// page's body is M-C20, which the agent's log says too. M-C20 is |
| -// operator copy: the app's members are never told (fix round 1, |
| -// ruling 1); |
| -// - a refused start check (moves.StartCheck), an incident, once per op, |
| +// (StepStray), an incident, once per move, from the cut's stopping: the |
| +// page's body is M-C20, and the agent's log says M-C20 when the box |
| +// detects it. M-C20 is operator copy: the app's members are never told |
| +// (fix round 1, ruling 1); |
| +// - a refused start check (moves.StartCheck), an incident, once per move, |
| // from the probe owners (movesProbeOwners.startCheck). |
| // |
| // The console's move pass (internal/console/movesrecovery.go, Plan 5 Task 7) |
| @@ -42,53 +45,57 @@ 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), 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. |
| +// The page outbox (fix round 2, ruling 1). A page is one file, |
| +// <DataDir>/.moves/.page-outbox/<host>/<move>/<rule>.json, created |
| +// exclusively (outboxCreate, the move log's outbox's own helper) when its |
| +// rule first fires: the file being there is the rule's once-per-entry limit, |
| +// so a rule that fires again never pages again. The entry carries the page |
| +// and its backoff. It is sent when created; one the pager refused is sent |
| +// again by every moves pass (flushPageOutbox) once its backoff is over |
| +// (pageRetryFirst, doubling, up to pageRetryMax), until it is delivered, |
| +// whatever the move and its journals have done since. A delivered entry |
| +// stays as a tombstone, so a rule that fires again still finds it, until the |
| +// move is retired on this box: no listed record carries it and its journal |
| +// is gone. The outbox depends on nothing in the move's journal. |
| // |
| -// The pending step carries the page itself and its backoff (pendingPage). A |
| -// page the pager refuses stays pending, and is sent again, after a doubling |
| -// backoff (pageRetryFirst to pageRetryMax), by whichever comes first: its rule |
| -// firing again, or the moves pass's journal sweep (sweepFinalJournals, which |
| -// reads every journal on the box each pass), for as long as the journal is |
| -// there. So stuck and rolled_back, which fire at one site each, are retried |
| -// too. The backoff is in the journal, so a restart keeps it. |
| +// If the outbox cannot keep a page, it is sent at once; if that fails too, |
| +// it is kept in memory and every pass sends it again (pageMemory), with a |
| +// loud line in the agent's log: only a restart during that double fault |
| +// loses it. |
| // |
| // 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 |
| -// (the journal's unwired:<rule> step remembers it across a restart), and |
| -// nothing is refused (fix round 1, ruling 2). |
| +// for as long as the agent runs, and nothing is refused or kept (fix round 1, |
| +// ruling 2). |
| // pagingLogf is the agent's log for the paging code (a variable so tests |
| // can read it). |
| var pagingLogf = log.Printf |
| -// pagedStep is the journal step that records a rule's page; unwiredStep |
| -// records that its warning about an unwired pager was logged. |
| +// pageFunc is the pager: moveCfg.Page's shape. |
| +type pageFunc = func(ctx context.Context, level, subject, body string) error |
| + |
| +// The rules' names in the outbox. |
| 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:" |
| + pageRuleRolledBack = "rolled_back" |
| + pageRuleStray = "stray" |
| + pageRuleStartCheck = "start_check" |
| ) |
| -// pageOutcome is what became of one rule's page. |
| +// pageRuleStuck is the stuck rule of one entry into stuck: its retry epoch. |
| +func pageRuleStuck(m *provision.MoveState) string { |
| + return "stuck-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10) |
| +} |
| + |
| +// pageOutcome is what became of one rule's firing. |
| type pageOutcome int |
| const ( |
| pageSent pageOutcome = iota // handed to the pager without error |
| - pagePagedBefore // the journal says it was sent before |
| + pagePagedBefore // the outbox says it was delivered before |
| pageUnwired // no pager is wired (logged loudly once) |
| - pageUndelivered // the journal or the pager failed (logged) |
| - pagePending // refused before, and waiting out its backoff |
| + pageUndelivered // first firing, refused: kept to send again |
| + pagePending // fired before, not delivered yet: the passes send it again |
| ) |
| // String is the outcome as the agent's log says it. |
| @@ -103,30 +110,7 @@ func (o pageOutcome) String() string { |
| case pagePending: |
| return "not paged yet: it is sent again after a backoff" |
| } |
| - return "not paged: the page was not delivered" |
| -} |
| - |
| -// The rules' names in the journal. |
| -const ( |
| - pageRuleRolledBack = "rolled_back" |
| - pageRuleStray = "stray" |
| - pageRuleStartCheck = "start_check" |
| -) |
| - |
| -// pageRuleStuck is the stuck rule of one entry into stuck: its retry epoch. |
| -func pageRuleStuck(m *provision.MoveState) string { |
| - return "stuck-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10) |
| -} |
| - |
| -// pendingPage is a pending:<rule> step's value: the page, so a later pass |
| -// can send it again, and its backoff: the attempts reserved so far and when |
| -// the next may start. |
| -type pendingPage struct { |
| - Level string `json:"level"` |
| - Subject string `json:"subject"` |
| - Body string `json:"body"` |
| - Tries int `json:"tries"` |
| - Next time.Time `json:"next"` |
| + return "not paged: the page was not delivered, and is sent again after a backoff" |
| } |
| // A refused page is tried again pageRetryFirst after its first attempt, and |
| @@ -150,172 +134,290 @@ func pageBackoff(tries int) time.Duration { |
| return pageRetryMax |
| } |
| -// 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 page refused before waits out its backoff |
| -// (pagePending). 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 |
| +// pageEntry is one page in the outbox: the page, the attempts so far, when |
| +// the next may start, and when it was delivered. |
| +type pageEntry struct { |
| + V int `json:"v"` |
| + Host string `json:"host"` |
| + Op string `json:"op"` |
| + Rule string `json:"rule"` |
| + Level string `json:"level"` |
| + Subject string `json:"subject"` |
| + Body string `json:"body"` |
| + At time.Time `json:"at"` |
| + Tries int `json:"tries"` |
| + Next time.Time `json:"next"` |
| + Delivered *time.Time `json:"delivered,omitempty"` |
| +} |
| + |
| +// pageOutboxDir is the page outbox's root. |
| +func pageOutboxDir(dataDir string) string { |
| + return filepath.Join(dataDir, ".moves", ".page-outbox") |
| +} |
| + |
| +// pageEntryDir is the directory of one move's pages. |
| +func pageEntryDir(dataDir, host, op string) string { |
| + return filepath.Join(pageOutboxDir(dataDir), host, op) |
| +} |
| + |
| +// pageOutboxMu serializes every change to an outbox entry in this process |
| +// (the moves pass and the probe owners both fire pages); delivery itself is |
| +// outside it, after the attempt is counted. |
| +var pageOutboxMu sync.Mutex |
| + |
| +// readPageEntry reads one entry. |
| +func readPageEntry(path string) (pageEntry, error) { |
| + var e pageEntry |
| + b, err := os.ReadFile(path) |
| + if err != nil { |
| + return e, err |
| + } |
| + if err := json.Unmarshal(b, &e); err != nil || e.V != 1 { |
| + return e, fmt.Errorf("%s is not a page outbox entry", path) |
| } |
| - fresh := pendingPage{Level: level, Subject: subject, Body: body} |
| - p, send, out := claimPage(dataDir, host, op, rule, now, &fresh) |
| - if !send { |
| - return out |
| + return e, nil |
| +} |
| + |
| +// writePageEntry replaces an entry. |
| +func writePageEntry(dataDir string, e pageEntry) error { |
| + b, err := json.Marshal(e) |
| + if err != nil { |
| + return err |
| } |
| - return deliverPage(ctx, page, dataDir, host, op, rule, now, p) |
| + return outboxReplace(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b) |
| } |
| -// claimPage decides, under the journal's lock, whether this call sends rule's |
| -// page, and reserves the attempt: its try counted and the next one put off by |
| -// its backoff, before the page is sent, so two callers in one window send it |
| -// once. fresh is the page to record when none is pending (a rule firing); nil |
| -// sends only one already pending (the sweep). A journal that cannot be read or |
| -// written sends fresh anyway, never the sweep's (it found nothing to read). |
| -func claimPage(dataDir, host, op, rule string, now time.Time, fresh *pendingPage) (p pendingPage, send bool, out pageOutcome) { |
| - out = pageUndelivered |
| - err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error { |
| - if j.Done(pagedStep + rule) { |
| - out = pagePagedBefore |
| - return errJournalUnchanged |
| +// firePage is a rule firing: the first firing of (host, op, rule) creates its |
| +// outbox entry, counting the attempt it is about to make, and sends the page; |
| +// a later firing finds the entry and sends nothing (the passes retry a |
| +// refused one). Unwired, it warns once per rule per move (warned is the |
| +// memory of that) and keeps nothing. |
| +func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, host, op, rule string, now time.Time, level, subject, body string) pageOutcome { |
| + if page == nil { |
| + if _, seen := warned.LoadOrStore(host+"/"+op+"/"+rule, true); !seen { |
| + pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op) |
| } |
| - raw, pending := j.Steps[pendingStep+rule] |
| - switch { |
| - case pending && json.Unmarshal([]byte(raw), &p) == nil && p.Subject != "": |
| - if now.Before(p.Next) { |
| - out = pagePending |
| - return errJournalUnchanged |
| - } |
| - case fresh != nil: |
| - p = *fresh |
| - default: |
| - return errJournalUnchanged // nothing the sweep can send |
| + return pageUnwired |
| + } |
| + e := pageEntry{V: 1, Host: host, Op: op, Rule: rule, Level: level, Subject: subject, Body: body, At: now.UTC(), Tries: 1, Next: now.UTC().Add(pageBackoff(1))} |
| + created, err := createPageEntry(dataDir, e) |
| + if err != nil { |
| + pagingLogf("moves: %s: WARNING: the page outbox cannot keep the %s page of move %s, so it is sent now: %v", host, rule, op, err) |
| + if perr := page(ctx, level, subject, body); perr == nil { |
| + return pageSent |
| + } else { |
| + pageMemory.keep(page, e) |
| + pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered either (%v); it is kept in memory only and sent again every pass, and is lost if the agent restarts first", host, rule, op, perr) |
| } |
| - p.Tries++ |
| - p.Next = now.Add(pageBackoff(p.Tries)) |
| - b, err := json.Marshal(p) |
| - if err != nil { |
| - return err |
| + return pageUndelivered |
| + } |
| + if !created { |
| + pageOutboxMu.Lock() |
| + kept, rerr := readPageEntry(filepath.Join(pageEntryDir(dataDir, host, op), rule+".json")) |
| + pageOutboxMu.Unlock() |
| + if rerr == nil && kept.Delivered != nil { |
| + return pagePagedBefore |
| } |
| - j.Mark(pendingStep+rule, string(b)) |
| - send = true |
| - return nil |
| - }) |
| - switch { |
| - case err == nil, errors.Is(err, errJournalUnchanged): |
| - return p, send, out |
| - case fresh != nil: |
| - pagingLogf("moves: %s: the %s page of %s is sent without its pending mark: its journal: %v", host, rule, op, err) |
| - return *fresh, true, out |
| + return pagePending |
| } |
| - pagingLogf("moves: %s: the pending %s page of %s waits: its journal: %v", host, rule, op, err) |
| - return p, false, out |
| + return deliverPageEntry(ctx, page, dataDir, e, now) |
| } |
| -// deliverPage sends a claimed page and, once the pager took it, journals |
| -// paged:<rule>. A refused page stays pending for its next attempt. |
| -func deliverPage(ctx context.Context, page func(ctx context.Context, level, subject, body string) error, dataDir, host, op, rule string, now time.Time, p pendingPage) pageOutcome { |
| - if err := page(ctx, p.Level, p.Subject, p.Body); err != nil { |
| - pagingLogf("moves: %s: the %s page of %s was not delivered (try %d), and is sent again after %s: %v", host, rule, op, p.Tries, pageBackoff(p.Tries), err) |
| +// createPageEntry creates e's file, create-exclusive. |
| +func createPageEntry(dataDir string, e pageEntry) (bool, error) { |
| + if dataDir == "" { |
| + return false, errors.New("no data directory for the page outbox") |
| + } |
| + if !outboxSegment(e.Host) || !outboxSegment(e.Op) || !outboxSegment(e.Rule) { |
| + return false, fmt.Errorf("an unusable page outbox name: %s/%s/%s", e.Host, e.Op, e.Rule) |
| + } |
| + b, err := json.Marshal(e) |
| + if err != nil { |
| + return false, err |
| + } |
| + pageOutboxMu.Lock() |
| + defer pageOutboxMu.Unlock() |
| + return outboxCreate(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b) |
| +} |
| + |
| +// deliverPageEntry sends an entry whose attempt is already counted, and marks |
| +// it delivered once the pager took it. A refused page stays for the passes; |
| +// a delivered one whose mark cannot be written may be sent again. |
| +func deliverPageEntry(ctx context.Context, page pageFunc, dataDir string, e pageEntry, now time.Time) pageOutcome { |
| + if err := page(ctx, e.Level, e.Subject, e.Body); err != nil { |
| + pagingLogf("moves: %s: the %s page of move %s was not delivered (try %d), and is sent again after %s: %v", e.Host, e.Rule, e.Op, e.Tries, pageBackoff(e.Tries), err) |
| return pageUndelivered |
| } |
| - if err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error { |
| - j.Mark(pagedStep+rule, now.UTC().Format(time.RFC3339)) |
| - 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) |
| + pageOutboxMu.Lock() |
| + defer pageOutboxMu.Unlock() |
| + path := filepath.Join(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json") |
| + kept, err := readPageEntry(path) |
| + if err == nil { |
| + at := now.UTC() |
| + kept.Delivered = &at |
| + err = writePageEntry(dataDir, kept) |
| + } |
| + if err != nil { |
| + pagingLogf("moves: %s: the %s page of move %s was sent, and the outbox may not say so, so it may be sent again: %v", e.Host, e.Rule, e.Op, err) |
| } |
| return pageSent |
| } |
| -// retryPendingPages is the journal sweep's half of paging: each page j holds |
| -// pending and not paged is sent again once its backoff is over (claimPage, |
| -// deliverPage). wrote is true when it wrote the journal, so the caller reads |
| -// it again. Unwired, it sends nothing: the rule warned when it fired. |
| -func (w *movesWorker) retryPendingPages(ctx context.Context, dataDir, host, op string, j *moves.Journal) (wrote bool) { |
| - if w.cfg.Page == nil { |
| - return false |
| +// flushPageOutbox is every moves pass's half of paging: each entry not yet |
| +// delivered is sent again once its backoff is over (its attempt counted and |
| +// the next put off before the send, so two senders in one window send it |
| +// once); each delivered entry whose move is retired on this box (no record |
| +// in recs carries it and its journal is gone) is removed; and the pages kept |
| +// in memory are sent again. Unwired, nothing is sent. |
| +func (w *movesWorker) flushPageOutbox(ctx context.Context, recs []provision.Record) { |
| + page, dd := w.cfg.Page, w.cfg.DataDir |
| + now := w.now() |
| + pageMemory.flush(ctx, now) |
| + if page == nil || dd == "" { |
| + return |
| } |
| - var rules []string |
| - for step := range j.Steps { |
| - if rule, ok := strings.CutPrefix(step, pendingStep); ok && !j.Done(pagedStep+rule) { |
| - rules = append(rules, rule) |
| + live := map[string]bool{} |
| + for _, rec := range recs { |
| + if rec.Move != nil && rec.Move.ID != "" { |
| + live[rec.Host+"/"+rec.Move.ID] = true |
| } |
| } |
| - sort.Strings(rules) |
| - now := w.now() |
| - for _, rule := range rules { |
| - p, send, _ := claimPage(dataDir, host, op, rule, now, nil) |
| - if !send { |
| - continue |
| + root := pageOutboxDir(dd) |
| + err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| + if errors.Is(err, fs.ErrNotExist) && p == root { |
| + return fs.SkipAll |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") { |
| + return nil |
| } |
| - wrote = true |
| - deliverPage(ctx, w.cfg.Page, dataDir, host, op, rule, now, p) |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + w.flushPageEntry(ctx, page, dd, p, now, live) |
| + return nil |
| + }) |
| + if err != nil { |
| + pagingLogf("moves: the page outbox: %v", err) |
| } |
| - return wrote |
| + pruneEmptyDirs(root) |
| } |
| -// unwiredOnce logs, loudly, that rule's page for op is not sent because no |
| -// pager is wired: once per rule per move, which the journal's unwired:<rule> |
| -// step remembers. A journal that cannot be read or written does not quiet |
| -// it: the warning is logged again the next time. |
| -func unwiredOnce(dataDir, host, op, rule string, now time.Time) { |
| - warned := false |
| - err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error { |
| - if j.Done(unwiredStep + rule) { |
| - return errJournalUnchanged |
| +// flushPageEntry is one entry of flushPageOutbox. |
| +func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p string, now time.Time, live map[string]bool) { |
| + pageOutboxMu.Lock() |
| + e, err := readPageEntry(p) |
| + if err != nil || e.Rule+".json" != filepath.Base(p) || e.Op != filepath.Base(filepath.Dir(p)) || e.Host != filepath.Base(filepath.Dir(filepath.Dir(p))) { |
| + pageOutboxMu.Unlock() |
| + pagingLogf("moves: the page outbox: %s is not an entry, and is left for an operator: %v", p, err) |
| + return |
| + } |
| + if e.Delivered != nil { |
| + retired := !live[e.Host+"/"+e.Op] && !moveJournalPresent(dd, e.Host, e.Op) |
| + if retired { |
| + if err := os.Remove(p); err != nil { |
| + pagingLogf("moves: the page outbox: %v", err) |
| + } |
| } |
| - warned = true |
| - pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op) |
| - j.Mark(unwiredStep+rule, now.UTC().Format(time.RFC3339)) |
| - return nil |
| - }) |
| - if err == nil || errors.Is(err, errJournalUnchanged) { |
| + pageOutboxMu.Unlock() |
| + return |
| + } |
| + if now.Before(e.Next) { |
| + pageOutboxMu.Unlock() |
| return |
| } |
| - if !warned { |
| - pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op) |
| + e.Tries++ |
| + e.Next = now.UTC().Add(pageBackoff(e.Tries)) |
| + if err := writePageEntry(dd, e); err != nil { |
| + pagingLogf("moves: %s: the %s page of move %s is sent without its attempt counted: %v", e.Host, e.Rule, e.Op, err) |
| + } |
| + pageOutboxMu.Unlock() |
| + deliverPageEntry(ctx, page, dd, e, now) |
| +} |
| + |
| +// pageMemory is the pages neither the outbox could keep nor the pager could |
| +// take: the passes send them again, until delivered or the agent restarts. |
| +var pageMemory memoryPages |
| + |
| +type memoryPages struct { |
| + mu sync.Mutex |
| + pages []memoryPage |
| +} |
| + |
| +type memoryPage struct { |
| + page pageFunc |
| + e pageEntry |
| +} |
| + |
| +func (m *memoryPages) keep(page pageFunc, e pageEntry) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + m.pages = append(m.pages, memoryPage{page: page, e: e}) |
| +} |
| + |
| +func (m *memoryPages) len() int { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + return len(m.pages) |
| +} |
| + |
| +// flush sends each kept page whose backoff is over; a delivered one is |
| +// dropped, a refused one waits for its next attempt. |
| +func (m *memoryPages) flush(ctx context.Context, now time.Time) { |
| + m.mu.Lock() |
| + due := m.pages[:0:0] |
| + var keep []memoryPage |
| + for _, kp := range m.pages { |
| + if now.Before(kp.e.Next) { |
| + keep = append(keep, kp) |
| + continue |
| + } |
| + due = append(due, kp) |
| + } |
| + m.pages = keep |
| + m.mu.Unlock() |
| + for _, kp := range due { |
| + if err := kp.page(ctx, kp.e.Level, kp.e.Subject, kp.e.Body); err == nil { |
| + pagingLogf("moves: %s: the %s page of move %s, kept in memory, is delivered", kp.e.Host, kp.e.Rule, kp.e.Op) |
| + continue |
| + } else { |
| + kp.e.Tries++ |
| + kp.e.Next = now.UTC().Add(pageBackoff(kp.e.Tries)) |
| + pagingLogf("moves: %s: WARNING: the %s page of move %s, kept in memory only, was not delivered (try %d): %v", kp.e.Host, kp.e.Rule, kp.e.Op, kp.e.Tries, err) |
| + } |
| + m.keep(kp.page, kp.e) |
| } |
| - pagingLogf("moves: %s: the warning about the unwired %s page of %s is not recorded, so it is logged again: %v", host, rule, op, err) |
| } |
| -// pageOnce is the worker's pageOnce: its pager, journals and clock. |
| -func (w *movesWorker) pageOnce(ctx context.Context, host, op, rule, level, subject, body string) pageOutcome { |
| - return pageOnce(ctx, w.cfg.Page, w.cfg.DataDir, host, op, rule, w.now(), level, subject, body) |
| +// firePage is the worker's firePage: its pager, outbox and clock. |
| +func (w *movesWorker) firePage(ctx context.Context, host, op, rule, level, subject, body string) pageOutcome { |
| + return firePage(ctx, w.cfg.Page, &w.pageWarned, w.cfg.DataDir, host, op, rule, w.now(), level, subject, body) |
| } |
| // pageStuck pages an operator once a MoveMarkStuck has landed (moves spec |
| // §13: stuck is an incident; K11), once per entry into stuck. role is this |
| // box's part in the move, "source" or "target", and reason the recorded |
| // stuck sentence. The page names the host, the move, the box and the |
| -// reason, and nothing else: no path, no error text, no secret. Its journal |
| -// is the one the stopped half keeps: a rollback's source keeps its |
| -// recovery's (the rollback id), every other half the move id's. |
| +// reason, and nothing else: no path, no error text, no secret. |
| func (w *movesWorker) pageStuck(ctx context.Context, rec provision.Record, role, reason string) { |
| m := rec.Move |
| if m == nil || m.ID == "" { |
| return |
| } |
| - op := m.ID |
| - if role == "source" && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.ID != "" { |
| - op = m.Rollback.ID |
| - } |
| subject, body := stuckPage(rec.Host, m.ID, w.cfg.Box, role, reason) |
| - w.pageOnce(ctx, rec.Host, op, pageRuleStuck(m), alerting.LevelIncident, subject, body) |
| + w.firePage(ctx, rec.Host, m.ID, pageRuleStuck(m), alerting.LevelIncident, subject, body) |
| } |
| // pageRolledBack pages a warning once this box has closed a move rolled |
| -// back (§13): rec is the record as the close found it, op the journal the |
| -// closing half keeps (the recovery's rollback id, or the revert's move id). |
| -func (w *movesWorker) pageRolledBack(ctx context.Context, rec provision.Record, op string) { |
| +// back (§13): rec is the record as the close found it. |
| +func (w *movesWorker) pageRolledBack(ctx context.Context, rec provision.Record) { |
| if rec.Move == nil || rec.Move.ID == "" { |
| return |
| } |
| subject, body := rolledBackPage(rec.Host, rec.Move.ID, w.cfg.Box) |
| - w.pageOnce(ctx, rec.Host, op, pageRuleRolledBack, alerting.LevelWarning, subject, body) |
| + w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleRolledBack, alerting.LevelWarning, subject, body) |
| } |
| // strayRefused reports whether a stop evidence report was refused for a |
| @@ -326,30 +428,32 @@ func strayRefused(rep moves.StoppedReport) bool { |
| return ok && !c.OK |
| } |
| -// pageStray pages an incident once per op for a stop evidence refused by a |
| -// stray process (§13). The page's body is M-C20, and once it is sent the |
| -// agent's log says M-C20 as well, with the move. rec is the cut's record. |
| +// pageStray pages an incident once per move for a stop evidence refused by a |
| +// stray process (§13). The page's body is M-C20, and the agent's log says |
| +// M-C20, with the move, when the box first detects it (the firing that keeps |
| +// the page), whether or not the page is delivered then (fix round 2, |
| +// ruling 1). rec is the cut's record. |
| func (w *movesWorker) pageStray(ctx context.Context, rec provision.Record) { |
| if rec.Move == nil || rec.Move.ID == "" { |
| return |
| } |
| subject, body := strayPage(rec.Host, w.cfg.Box) |
| - if w.pageOnce(ctx, rec.Host, rec.Move.ID, pageRuleStray, alerting.LevelIncident, subject, body) != pageSent { |
| - return |
| + switch w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleStray, alerting.LevelIncident, subject, body) { |
| + case pageSent, pageUndelivered: |
| + pagingLogf("moves: %s: move %s: %s", rec.Host, rec.Move.ID, strayLine(w.cfg.Box, rec.Host)) |
| } |
| - pagingLogf("moves: %s: move %s: %s", rec.Host, rec.Move.ID, strayLine(w.cfg.Box, rec.Host)) |
| } |
| -// pageStartRefused pages an incident once per op for a refused §5.1 start |
| -// check on this box (§13), and returns what became of the page. Its journal |
| -// write takes the journal's one lock, which the landing's pass shares. |
| +// pageStartRefused pages an incident once per move for a refused §5.1 start |
| +// check on this box (§13), and returns what became of the page. Its outbox |
| +// is the box's, which the moves pass retries. |
| func (m *movesProbeOwners) pageStartRefused(ctx context.Context, host, op string) pageOutcome { |
| now := time.Now() |
| if m.Now != nil { |
| now = m.Now() |
| } |
| subject, body := startRefusedPage(host, op, m.Box) |
| - return pageOnce(ctx, m.Page, m.DataDir, host, op, pageRuleStartCheck, now, alerting.LevelIncident, subject, body) |
| + return firePage(ctx, m.Page, &m.pageWarned, m.DataDir, host, op, pageRuleStartCheck, now, alerting.LevelIncident, subject, body) |
| } |
| // rolledBackPage is the rolled_back page: the subject (MV-N385) and the |
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go| index 21f62e695..756ac5fbc 100644 |
| --- a/internal/host/movespaging_test.go |
| +++ b/internal/host/movespaging_test.go |
| @@ -35,14 +35,29 @@ func capturePagingLog(t *testing.T) *[]string { |
| return &lines |
| } |
| -// pagedInJournal reports whether op's journal on dataDir records rule's page. |
| -func pagedInJournal(t *testing.T, dataDir, host, op, rule string) bool { |
| +// pageDelivered reports whether the page outbox under dataDir holds rule's |
| +// entry for (host, op), delivered. |
| +func pageDelivered(t *testing.T, dataDir, host, op, rule string) bool { |
| t.Helper() |
| - j, err := moves.ReadJournal(dataDir, host, op) |
| + e, err := readPageEntry(filepath.Join(pageEntryDir(dataDir, host, op), rule+".json")) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return false |
| + } |
| if err != nil { |
| t.Fatal(err) |
| } |
| - return j.Done(pagedStep + rule) |
| + return e.Delivered != nil |
| +} |
| + |
| +// refusingPager is a pager that refuses while *refuse is set, recording |
| +// what it takes into pages. |
| +func refusingPager(pages *pagesSeen, refuse *bool) pageFunc { |
| + return func(ctx context.Context, level, subject, body string) error { |
| + if *refuse { |
| + return errors.New("the webhook is down") |
| + } |
| + return pages.hook()(ctx, level, subject, body) |
| + } |
| } |
| // TestPagingIsRateLimited (Review Focus 4; spec §13; preflight L8): a rule |
| @@ -74,8 +89,8 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if n, level := pagesWith(pages, "carlos move stuck: "); n != 1 || level != "incident" { |
| t.Fatalf("stuck pages %d (%s), want one incident: %+v", n, level, pages) |
| } |
| - if !pagedInJournal(t, dd, p1Host, p1ID, "stuck-0") { |
| - t.Fatal("the target's journal of the move does not record the stuck page") |
| + if !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") { |
| + t.Fatal("the outbox does not hold the stuck page delivered") |
| } |
| // An operator's retry advances the epoch; stuck again is a new entry. |
| rec.Move.RecoveryRetryEpoch = 1 |
| @@ -84,11 +99,10 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if n, _ := pagesWith(pages, "carlos move stuck: "); n != 2 { |
| t.Fatalf("stuck pages after a retry %d, want two: %+v", n, pages) |
| } |
| - // The source of a rollback keeps its recovery's journal. |
| + // One entry into stuck is one page, whichever half fires it. |
| w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch) |
| - w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch) |
| - if n, _ := pagesWith(pages, "carlos move stuck: "); n != 3 || !pagedInJournal(t, dd, p1Host, "rb_abcdefghijklmnopqrstuvwxyz", "stuck-1") { |
| - t.Fatalf("the source's stuck page %d, want one more, in the recovery's journal: %+v", n, pages) |
| + if n, _ := pagesWith(pages, "carlos move stuck: "); n != 2 || !pageDelivered(t, dd, p1Host, p1ID, "stuck-1") { |
| + t.Fatalf("stuck pages %d, want still two, the second delivered: %+v", n, pages) |
| } |
| }) |
| @@ -172,7 +186,7 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 { |
| t.Fatalf("the agent's log says M-C20 %d times, want once: %q", n, *logged) |
| } |
| - if !pagedInJournal(t, h.cfg.DataDir, p1Host, p1ID, pageRuleStray) { |
| + if !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, pageRuleStray) { |
| t.Fatal("the cut's journal does not record the stray page") |
| } |
| // Evidence that fails for any other reason pages nothing. |
| @@ -217,15 +231,15 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if n, level := pagesWith(pages, "carlos move start refused: "); n != 1 || level != "incident" { |
| t.Fatalf("start check pages %d (%s), want one incident: %+v", n, level, pages) |
| } |
| - if !pagedInJournal(t, f.dataDir, poHost, poOp, pageRuleStartCheck) { |
| + if !pageDelivered(t, f.dataDir, poHost, poOp, pageRuleStartCheck) { |
| t.Fatal("the landing's journal does not record the start check page") |
| } |
| }) |
| // Fix round 1, ruling 2: production stays unwired until Plan 5 9b, and |
| // an unwired pager is never silent. Each rule says so in the agent's |
| - // log, loudly, once per move (its journal remembers, across a restart); |
| - // nothing is refused. The probe owners' refused start check says what |
| + // log, loudly, once per move while the agent runs (fix round 2: the |
| + // worker's memory, not the journal); nothing is refused or kept. The probe owners' refused start check says what |
| // became of its page, so the result is never dropped. |
| t.Run("an unwired pager is loud, once per rule per move", func(t *testing.T) { |
| logged := capturePagingLog(t) |
| @@ -236,12 +250,12 @@ func TestPagingIsRateLimited(t *testing.T) { |
| w := worker() |
| w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase) |
| w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase) |
| - worker().pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase) // a restarted agent |
| + w.pageStuck(ctx, rec, "target", provision.StuckReturnTargetDatabase) |
| if loud() != 1 { |
| t.Fatalf("the unwired stuck page warned %d times, want once: %q", loud(), *logged) |
| } |
| - w.pageRolledBack(ctx, rec, p1ID) |
| - w.pageRolledBack(ctx, rec, p1ID) |
| + w.pageRolledBack(ctx, rec) |
| + w.pageRolledBack(ctx, rec) |
| if loud() != 2 { |
| t.Fatalf("a second rule warned %d times in all, want 2: %q", loud(), *logged) |
| } |
| @@ -273,125 +287,22 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if n := strings.Count(strings.Join(*logged, "\n"), "refused to start "+poHost+" for "+poOp+" (not paged: no pager is wired)"); n != 2 { |
| t.Fatalf("the refusals' lines %d, want 2 saying the page was not sent: %q", n, *logged) |
| } |
| - }) |
| - |
| - // Fix round 1, ruling 3: one lock per journal. A refused start check's |
| - // page that comes while the landing holds its journal (moveslandprobe's |
| - // admitting mark, through moves.UpdateJournal) waits for it, and both |
| - // the landing's mark and paged:start_check are in the journal after. |
| - t.Run("one journal lock: the landing's mark and the page both survive", func(t *testing.T) { |
| - f := newPOFixture(t) |
| - var pages pagesSeen |
| - f.owners.Page = pages.hook() |
| - done := make(chan pageOutcome, 1) |
| - err := moves.UpdateJournal(f.dataDir, poHost, poOp, func(j *moves.Journal) error { |
| - go func() { done <- f.owners.pageStartRefused(ctx, poHost, poOp) }() |
| - select { |
| - case o := <-done: |
| - t.Errorf("the page wrote the journal (%s) while the landing held it", o) |
| - case <-time.After(100 * time.Millisecond): |
| - } |
| - j.Phase, j.Attempt = provision.LandProbing, 1 |
| - j.Mark(probeStepAdmitting+"1", `{"session":7}`) |
| - return nil |
| - }) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if o := <-done; o != pageSent { |
| - t.Fatalf("the page after the landing's write: %s", o) |
| - } |
| - j, err := moves.ReadJournal(f.dataDir, poHost, poOp) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if !j.Done(probeStepAdmitting+"1") || !j.Done(pagedStep+pageRuleStartCheck) || j.Phase != provision.LandProbing { |
| - t.Fatalf("the journal lost a writer's step: %+v", j) |
| - } |
| - if n, _ := pagesWith(pages, "carlos move start refused: "); n != 1 { |
| - t.Fatalf("pages %+v, want one", pages) |
| - } |
| - }) |
| - |
| - // 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 |
| - // Inside the refused page's backoff the rule firing again waits; after |
| - // it, it sends. |
| - if o := pageOnce(ctx, flaky, dd, p1Host, p1ID, pageRuleStartCheck, now, "incident", "s2", "b2"); o != pagePending { |
| - t.Fatalf("a retry inside the backoff: %s, want pending", o) |
| - } |
| - now = now.Add(pageRetryFirst) |
| - 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) |
| + if _, err := os.Stat(pageOutboxDir(dd)); !os.IsNotExist(err) { |
| + t.Fatalf("an unwired worker kept pages: %v", err) |
| } |
| }) |
| - // Fix round 1, follow-up: a page the pager refused stays pending, and |
| - // every moves pass's journal sweep (sweepFinalJournals, which already |
| - // reads each journal) sends it again after a backoff, so stuck and |
| - // rolled_back, which fire at one site each, are retried too. |
| - t.Run("sweep: a refused stuck page is sent by a later pass", func(t *testing.T) { |
| + // Fix round 2, ruling 1: the page outbox. A refused page is sent again |
| + // by a later pass, whatever its move has done since; it is never lost |
| + // to the journal being parked; a double fault keeps it in memory; a |
| + // rule that fires again after delivery never pages again. |
| + t.Run("outbox: a refused stuck page is sent by a later pass", func(t *testing.T) { |
| h := newRevertHarness(t) |
| h.fromStopped(t) |
| h.arm(t) |
| var pages pagesSeen |
| refuse := true |
| - h.cfg.Page = func(ctx context.Context, level, subject, body string) error { |
| - if refuse { |
| - return errors.New("the webhook is down") |
| - } |
| - return pages.hook()(ctx, level, subject, body) |
| - } |
| + h.cfg.Page = refusingPager(&pages, &refuse) |
| if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil { |
| t.Fatal(err) |
| } |
| @@ -399,7 +310,7 @@ func TestPagingIsRateLimited(t *testing.T) { |
| if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch { |
| t.Fatalf("not stuck: %+v", m) |
| } |
| - if len(pages) != 0 || pagedInJournal(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") { |
| + if len(pages) != 0 || pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") { |
| t.Fatalf("a refused page counted as sent: %+v", pages) |
| } |
| refuse = false |
| @@ -408,10 +319,10 @@ func TestPagingIsRateLimited(t *testing.T) { |
| t.Fatalf("retried inside its backoff: %+v", pages) |
| } |
| h.clock.Advance(2 * time.Minute) |
| - h.pass(t) // no new stuck event: the sweep sends it |
| + h.pass(t) // no new stuck event: the pass's flush sends it |
| pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch) |
| - if !pagedInJournal(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") { |
| - t.Fatal("the swept page is not marked paged") |
| + if !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, "stuck-0") { |
| + t.Fatal("the flushed page is not marked delivered") |
| } |
| h.clock.Advance(time.Hour) |
| h.pass(t) |
| @@ -420,52 +331,150 @@ func TestPagingIsRateLimited(t *testing.T) { |
| pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch) |
| }) |
| - t.Run("sweep: a journal that is gone sends nothing", func(t *testing.T) { |
| + t.Run("outbox: a refused rolled_back page is delivered after the journal is parked", func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + refuse := true |
| + var pages pagesSeen |
| + h.srcPage = refusingPager(&pages, &refuse) // the source's pager |
| + h.released(t) |
| + h.srcPass(t) |
| + h.opened(t) |
| + h.closeOut(t) |
| + if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != "rolled_back" { |
| + t.Fatalf("not closed: %+v %+v", rec.Move, rec.LastMove) |
| + } |
| + if live, parked := h.parkedJournal(t, p1ID); live || !parked { |
| + t.Fatalf("the source's journal of the move: live %v, parked %v; want parked", live, parked) |
| + } |
| + if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 0 { |
| + t.Fatalf("a refused page was delivered: %+v", pages) |
| + } |
| + refuse = false |
| + h.clock.Advance(2 * time.Minute) |
| + h.srcPass(t) |
| + if n, level := pagesWith(pages, "carlos move rolled back: "); n != 1 || level != "warning" { |
| + t.Fatalf("rolled_back pages after the park %d (%s), want one warning: %+v", n, level, pages) |
| + } |
| + // Delivered, and the move retired on this box (no record carries it, |
| + // its journals are parked): the tombstone goes, at the next pass. |
| + h.srcPass(t) |
| + if _, err := os.Stat(pageEntryDir(h.sdir, p1Host, p1ID)); !os.IsNotExist(err) { |
| + t.Fatalf("the retired move's tombstone is still kept: %v", err) |
| + } |
| + h.clock.Advance(time.Hour) |
| + h.srcPass(t) |
| + if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 1 { |
| + t.Fatalf("rolled_back paged again: %+v", pages) |
| + } |
| + }) |
| + |
| + t.Run("outbox: the outbox write and the delivery fail, a later pass delivers from memory", func(t *testing.T) { |
| + logged := capturePagingLog(t) |
| + t.Cleanup(func() { pageMemory = memoryPages{} }) |
| dd := t.TempDir() |
| + // The outbox cannot be made: a file stands where its directory goes. |
| + if err := os.MkdirAll(filepath.Join(dd, ".moves"), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(pageOutboxDir(dd), []byte("in the way"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC) |
| var pages pagesSeen |
| refuse := true |
| - w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "uk-box", Now: func() time.Time { return now }, |
| - Page: func(ctx context.Context, level, subject, body string) error { |
| - if refuse { |
| - return errors.New("the webhook is down") |
| - } |
| - return pages.hook()(ctx, level, subject, body) |
| - }}} |
| - kept := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}} |
| - gone := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: "mv_zyxwvutsrqponmlkjihgfedcba", State: provision.MoveStuck}} |
| - w.pageStuck(ctx, kept, "source", provision.StuckBaselineMismatch) |
| - w.pageStuck(ctx, gone, "source", provision.StuckBaselineMismatch) |
| - if err := os.RemoveAll(moves.JournalDir(dd, p1Host, gone.Move.ID)); err != nil { |
| - t.Fatal(err) |
| + w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "uk-box", Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)}} |
| + if o := w.firePage(ctx, p1Host, p1ID, "stuck-0", "incident", "carlos move stuck: "+p1Host, "b"); o != pageUndelivered { |
| + t.Fatalf("the double fault: %s", o) |
| + } |
| + if pageMemory.len() != 1 || !strings.Contains(strings.Join(*logged, "\n"), "kept in memory only") { |
| + t.Fatalf("the page is not kept in memory, loudly: %d %q", pageMemory.len(), *logged) |
| + } |
| + refuse = false |
| + w.flushPageOutbox(ctx, nil) // inside the backoff |
| + if len(pages) != 0 { |
| + t.Fatalf("retried inside its backoff: %+v", pages) |
| } |
| - refuse, now = false, now.Add(time.Hour) |
| - w.sweepFinalJournals(ctx, nil) |
| - if len(pages) != 1 || !strings.Contains(pages[0].body, p1ID) { |
| - t.Fatalf("the sweep sent %+v, want only the kept journal's page", pages) |
| + now = now.Add(2 * time.Minute) |
| + w.flushPageOutbox(ctx, nil) |
| + if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || pageMemory.len() != 0 { |
| + t.Fatalf("the memory's page: %d delivered, %d kept; want 1 and 0", n, pageMemory.len()) |
| } |
| - if _, err := os.Stat(moves.JournalDir(dd, p1Host, gone.Move.ID)); !os.IsNotExist(err) { |
| - t.Fatalf("the sweep brought a gone journal back: %v", err) |
| + now = now.Add(time.Hour) |
| + w.flushPageOutbox(ctx, nil) |
| + if len(pages) != 1 { |
| + t.Fatalf("a delivered memory page was sent again: %+v", pages) |
| } |
| }) |
| - t.Run("two-phase: a journal that cannot be read still pages", func(t *testing.T) { |
| + t.Run("outbox: a rule that fires again after delivery does not page again", func(t *testing.T) { |
| dd := t.TempDir() |
| - dir := moves.JournalDir(dd, p1Host, p1ID) |
| - if err := os.MkdirAll(dir, 0o700); err != nil { |
| + now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC) |
| + var pages pagesSeen |
| + worker := func() *movesWorker { |
| + return &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: pages.hook()}} |
| + } |
| + rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveRollingBack}} |
| + w := worker() |
| + w.pageRolledBack(ctx, rec) |
| + if !pageDelivered(t, dd, p1Host, p1ID, pageRuleRolledBack) { |
| + t.Fatal("not delivered") |
| + } |
| + // The record still carries the move: the tombstone stays through |
| + // passes, and a firing again, here or after a restart, finds it. |
| + live := []provision.Record{rec} |
| + for i := 0; i < 3; i++ { |
| + now = now.Add(time.Hour) |
| + w.flushPageOutbox(ctx, live) |
| + w.pageRolledBack(ctx, rec) |
| + worker().pageRolledBack(ctx, rec) |
| + } |
| + if n, _ := pagesWith(pages, "carlos move rolled back: "); n != 1 { |
| + t.Fatalf("rolled_back pages %d, want one: %+v", n, pages) |
| + } |
| + // A journal of the move on this box keeps it too. |
| + j, err := moves.ReadJournal(dd, p1Host, p1ID) |
| + if err != nil { |
| t.Fatal(err) |
| } |
| - if err := os.WriteFile(filepath.Join(dir, "journal.json"), []byte("not json"), 0o600); err != nil { |
| + j.Mark("x", "y") |
| + if err := j.Write(dd); err != nil { |
| t.Fatal(err) |
| } |
| + w.flushPageOutbox(ctx, nil) |
| + if !pageDelivered(t, dd, p1Host, p1ID, pageRuleRolledBack) { |
| + t.Fatal("the tombstone went while the move's journal is on the box") |
| + } |
| + }) |
| + |
| + t.Run("outbox: a flush-delivered M-C20 still has its box log line", func(t *testing.T) { |
| + logged := capturePagingLog(t) |
| + h := newCutHarness(t) |
| + h.ready(t) |
| var pages pagesSeen |
| - 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) |
| - } |
| + refuse := true |
| + h.cfg.Page = refusingPager(&pages, &refuse) |
| + evidence := h.cfg.Evidence |
| + h.cfg.Evidence = func(ctx context.Context, host, location string, dbs []string) (moves.StoppedReport, error) { |
| + r, err := evidence(ctx, host, location, dbs) |
| + r.Steps[moves.StepStray] = moves.Check{Reason: "1 process(es) besides the agent in its leaf"} |
| + r.Stray = []int{4242} |
| + return r, err |
| + } |
| + h.cut(t) |
| + h.pass(t) |
| + mc20 := "ie-box is running a process for " + p1Host + " that this move cannot account for. An operator has been paged." |
| + if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 { |
| + t.Fatalf("M-C20 in the agent's log %d times at the detection, want once: %q", n, *logged) |
| } |
| - if len(pages) != 2 { |
| - t.Fatalf("pages %+v, want one per call while the journal cannot remember", pages) |
| + refuse = false |
| + h.clock.Advance(2 * time.Minute) |
| + h.pass(t) |
| + h.pass(t) |
| + if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 || !pageDelivered(t, h.cfg.DataDir, p1Host, p1ID, pageRuleStray) { |
| + t.Fatalf("the stray page after the flush: %+v", pages) |
| + } |
| + if n := strings.Count(strings.Join(*logged, "\n"), mc20); n != 1 { |
| + t.Fatalf("M-C20 in the agent's log %d times, want once: %q", n, *logged) |
| } |
| }) |
| } |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 733fe9524..d3418e97f 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -93,7 +93,10 @@ type movesWorker struct { |
| // logged is the move log identities this worker has handed off |
| // (movesevents.go): host/op/identity, bounded. |
| logged moveLogMemo |
| - strict atomic.Pointer[map[string]bool] |
| + // pageWarned is the host/op/rule pages this worker has warned are not |
| + // sent because no pager is wired (movespaging.go). |
| + pageWarned sync.Map |
| + strict atomic.Pointer[map[string]bool] |
| // holds is what reconcile and the schedule halves ask about this |
| // box's running moves (moveshold.go), rebuilt every pass. Nil with the |
| // switch off. |
| @@ -877,8 +880,11 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| if cfg.Enabled { |
| // The source's op journals that reached the stamp, whoever's record |
| // no longer routes here (movesfinal.go, fix round 1). |
| - w.sweepFinalJournals(ctx, recs) |
| + w.sweepFinalJournals(recs) |
| } |
| + // The pages the outbox keeps undelivered are sent again, and delivered |
| + // ones of retired moves removed (movespaging.go). |
| + w.flushPageOutbox(ctx, recs) |
| w.liftFences(recs) |
| return nil |
| } |
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go| index fb84a9a41..137d32f57 100644 |
| --- a/internal/host/movesprobeowners.go |
| +++ b/internal/host/movesprobeowners.go |
| @@ -131,6 +131,8 @@ type movesProbeOwners struct { |
| // (movespaging.go, §13): PageOperator once claim Plan 10 registers this |
| // owner set; nil, nothing pages. |
| Page func(ctx context.Context, level, subject, body string) error |
| + // pageWarned is the pages warned unsent for want of a pager. |
| + pageWarned sync.Map |
| mu sync.Mutex |
| published map[string]ClaimPublicationReceipt |
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go| index 5b518ad59..886fc7a09 100644 |
| --- a/internal/host/movesrecover.go |
| +++ b/internal/host/movesrecover.go |
| @@ -820,7 +820,7 @@ func (r *recoverRun) finalize(ctx context.Context) (bool, error) { |
| } |
| r.w.logTransition(ctx, r.s, pre, *closed) |
| log.Printf("moves: %s: move %s is rolled back and closed", host, id) |
| - r.w.pageRolledBack(ctx, r.rec, r.rb) |
| + r.w.pageRolledBack(ctx, r.rec) |
| nudgeSerial(ctx, r.s) |
| if !released { |
| if err := r.fault("after-close-record"); err != nil { |
diff --git a/internal/host/movesrecover_test.go b/internal/host/movesrecover_test.go| index 33fb17910..b00ea422e 100644 |
| --- a/internal/host/movesrecover_test.go |
| +++ b/internal/host/movesrecover_test.go |
| @@ -46,10 +46,12 @@ type recoverHarness struct { |
| // and answers. |
| probe func(req recoveryProbeRequest) (recoveryProbeOutcome, error) |
| probes []recoveryProbeRequest |
| - rcAt string |
| - rcDone bool |
| - rcHits []string |
| - rcHook func(point string) |
| + // srcPage, when set, is the source's pager instead of pages.hook(). |
| + srcPage pageFunc |
| + rcAt string |
| + rcDone bool |
| + rcHits []string |
| + rcHook func(point string) |
| } |
| // newRecoverHarness builds the fixture up to the rollback decision's |
| @@ -159,6 +161,9 @@ func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg { |
| return nil |
| } |
| c.Page = h.pages.hook() |
| + if h.srcPage != nil { |
| + c.Page = h.srcPage |
| + } |
| c.recoveryProbe = func(ctx context.Context, _ release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| h.probes = append(h.probes, req) |
| if h.probe != nil { |
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index 3f21b694c..f995e655d 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -726,7 +726,7 @@ func (r *revertRun) close(ctx context.Context, releaseClaim bool) (bool, error) |
| return false, err |
| } |
| log.Printf("moves: %s: move %s is reverted and closed", host, id) |
| - r.w.pageRolledBack(ctx, r.rec, id) |
| + r.w.pageRolledBack(ctx, r.rec) |
| nudgeSerial(ctx, r.s) |
| if releaseClaim { |
| if err := r.fault("after-close-record"); err != nil { |
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go| index 01709111e..2a0219c19 100644 |
| --- a/internal/host/movesrevert_test.go |
| +++ b/internal/host/movesrevert_test.go |
| @@ -726,7 +726,7 @@ func TestRevertJournalSweptAfterTheClose(t *testing.T) { |
| h.pass(t) |
| h.reopened(t) |
| // In flight: the sweep leaves the marked journal. |
| - h.w().sweepFinalJournals(context.Background(), []provision.Record{h.read(t, p1Host)}) |
| + h.w().sweepFinalJournals([]provision.Record{h.read(t, p1Host)}) |
| if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) { |
| t.Fatal("the sweep parked a journal whose revert is in flight") |
| } |
diff --git a/internal/host/movesrollback_test.go b/internal/host/movesrollback_test.go| index dd108a91a..63260f601 100644 |
| --- a/internal/host/movesrollback_test.go |
| +++ b/internal/host/movesrollback_test.go |
| @@ -1194,7 +1194,7 @@ func TestJournalSweepWaitsForTheTargetPark(t *testing.T) { |
| t.Fatalf("never reached %s: %v", h.rbAt, h.rbHits) |
| } |
| w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}} |
| - w.sweepFinalJournals(context.Background(), []provision.Record{h.read(t, p1Host)}) |
| + w.sweepFinalJournals([]provision.Record{h.read(t, p1Host)}) |
| if _, err := os.Stat(moves.JournalDir(h.tdir, p1Host, p1ID)); err != nil { |
| t.Fatalf("the sweep parked a journal whose cleaning is under way: %v", err) |
| } |
diff --git a/internal/moves/journal.go b/internal/moves/journal.go| index 0f7d6dd22..d0a3410a3 100644 |
| --- a/internal/moves/journal.go |
| +++ b/internal/moves/journal.go |
| @@ -225,27 +225,7 @@ func (j *Journal) write(dataDir string) error { |
| return err |
| } |
| j.read = j.readSteps() |
| - 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. |
| -// |
| -// It is for tests only. No production code calls it, and no environment |
| -// variable, flag or configuration reaches it |
| -// (TestJournalSyncSeamIsTestOnly). It is exported because the tests that |
| -// need it are another package's (internal/host): an export_test.go here is |
| -// compiled only into this package's own tests. |
| -func SetJournalSyncForTest(f func(dir string) error) (restore func()) { |
| - saved := journalReplacedSync |
| - journalReplacedSync = f |
| - return func() { journalReplacedSync = saved } |
| + return syncDir(dir) |
| } |
| // missingDirs lists the directories at and above dir that do not exist yet, |
diff --git a/internal/moves/journal_lock_test.go b/internal/moves/journal_lock_test.go| index d7d1075a4..dd51e629e 100644 |
| --- a/internal/moves/journal_lock_test.go |
| +++ b/internal/moves/journal_lock_test.go |
| @@ -1,10 +1,6 @@ |
| package moves |
| import ( |
| - "io/fs" |
| - "os" |
| - "path/filepath" |
| - "strings" |
| "testing" |
| ) |
| @@ -60,51 +56,3 @@ func TestJournalWritersKeepEachOthersSteps(t *testing.T) { |
| t.Fatalf("a deleted step came back, or another went: %+v", got.Steps) |
| } |
| } |
| - |
| -// Fix round 1, follow-up: SetJournalSyncForTest is a test seam. Nothing but |
| -// its own definition names it in a production (non-test) Go file, so no |
| -// production path, environment variable or flag can reach it. |
| -func TestJournalSyncSeamIsTestOnly(t *testing.T) { |
| - root := filepath.Join("..", "..") |
| - if _, err := os.Stat(filepath.Join(root, "go.mod")); err != nil { |
| - t.Fatalf("the repository root is not two levels up: %v", err) |
| - } |
| - var found []string |
| - err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { |
| - if err != nil { |
| - return err |
| - } |
| - if d.IsDir() { |
| - if n := d.Name(); path != root && (strings.HasPrefix(n, ".") || n == "vendor" || n == "node_modules" || n == "testdata") { |
| - return filepath.SkipDir |
| - } |
| - return nil |
| - } |
| - if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { |
| - return nil |
| - } |
| - b, err := os.ReadFile(path) |
| - if err != nil { |
| - return err |
| - } |
| - if n := strings.Count(string(b), "SetJournalSyncForTest"); n > 0 { |
| - found = append(found, filepath.ToSlash(path)) |
| - } |
| - return nil |
| - }) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if len(found) != 1 || !strings.HasSuffix(found[0], "internal/moves/journal.go") { |
| - t.Fatalf("production files naming SetJournalSyncForTest: %v, want only its definition in internal/moves/journal.go", found) |
| - } |
| - b, err := os.ReadFile("journal.go") |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - for _, line := range strings.Split(string(b), "\n") { |
| - if strings.Contains(line, "SetJournalSyncForTest") && !strings.HasPrefix(strings.TrimSpace(line), "//") && !strings.HasPrefix(line, "func SetJournalSyncForTest(") { |
| - t.Fatalf("journal.go uses the seam outside its definition: %q", line) |
| - } |
| - } |
| -} |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index 96cfdbf27..0decb99be 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -543,8 +543,8 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "febac996d3bb9caf4021cbc4bcffe1c900bddba4287ff95f083bd5e0751e9bc2"}, |
| // Closes a committed move through SetRecordMove (MoveClose, LastMove done) once nothing is owed under the op, then releases the claim (ReleaseMove via releaseAfterClose): record first, ruling M4. A claim already released by this move gets only the record's close. |
| {"internal/host/movesparks.go", "movesWorker.closeMove", "record-writer", "key:SetRecordMove", "380a2a0f8cc450573256f359a77da5e451ad506fef84aaeceb9696e3d2904211"}, |
| - // Closes the rollback record through SetRecordMove once nothing is owed under the op, then releases the claim (record first, ruling M4); local journal parking uses separate storage. Plan 6 Task 5: after the close it pages rolled_back once (pageRolledBack, the recovery's journal), which writes no record. |
| - {"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "34fd48b2d014c5b1cf9480ae93f9e2c0fa6ff1f1ed7600cc34ee495e29c49315"}, |
| + // Closes the rollback record through SetRecordMove once nothing is owed under the op, then releases the claim (record first, ruling M4); local journal parking uses separate storage. Plan 6 Task 5: after the close it pages rolled_back once (pageRolledBack, through the box's page outbox; fix round 2), which writes no record. |
| + {"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "f2bab5b8b4deb1e1698370f1d7ecd52c226978579e3682746759cab9698f39c0"}, |
| // Advances the source recovery phase through SetRecordMove and its sequenced writer. |
| {"internal/host/movesrecover.go", "recoverRun.phase", "record-writer", "key:SetRecordMove", "be641eaba77b940ce471696720b637b9dfb9454de0893aca1fcc40750135719d"}, |
| // Records probe inputs through SetRecordMove; phase and stuck helpers use the same writer. |