amadan

carlos / platform Public

host: the moves pass sends refused pages again, with a journal backoff

moves-phase1-6 Tasks 4+5 fix round 1, follow-up. A pending:<rule> step
now carries the page itself and its backoff (tries, next; one minute,
doubling, up to 30). Every moves pass's journal sweep
(sweepFinalJournals, which already reads each journal on the box) sends
a page pending and not paged again once its backoff is over, for as
long as the journal is there, so stuck and rolled_back, which fire at
one site each, are retried too. The rule firing again and the sweep
share one claim under the journal's lock, so one window makes one
attempt; the backoff is in the journal, so a restart keeps it.

SetJournalSyncForTest's comment says it is for tests only and why it
is exported; TestJournalSyncSeamIsTestOnly fails if any production file
but its definition names it.

No censused unit changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b092b10f3bb439433983d7bcf1b1d14ecc165ec4 parent 9445cce
8 files changed, +282 −26
  • internal/host/movesfinal.go +12 −4
  • internal/host/movespaging.go +129 −18
  • internal/host/movespaging_test.go +79 −0
  • internal/host/movespass.go +1 −1
  • internal/host/movesrevert_test.go +1 −1
  • internal/host/movesrollback_test.go +1 −1
  • internal/moves/journal.go +7 −1
  • internal/moves/journal_lock_test.go +52 −0
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 6c9c328b5..bb29c883f 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(recs []provision.Record) {
+func (w *movesWorker) sweepFinalJournals(ctx context.Context, recs []provision.Record) {
dd := w.cfg.DataDir
if dd == "" {
return
@@ -810,19 +810,27 @@ func (w *movesWorker) sweepFinalJournals(recs []provision.Record) {
if !od.IsDir() || !journalComponent(op) {
continue
}
- if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil {
+ if err := w.sweepFinalJournal(ctx, 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.
-func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error {
+// 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 {
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/movespaging.go b/internal/host/movespaging.go
index cdfdcdf9f..ef697ef6f 100644
--- a/internal/host/movespaging.go
+++ b/internal/host/movespaging.go
@@ -2,10 +2,13 @@ package host
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"log"
+ "sort"
"strconv"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/alerting"
@@ -48,8 +51,15 @@ import (
// 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.
+// At least once, then: a rare second page beats a lost one.
+//
+// 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.
//
// 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
@@ -78,6 +88,7 @@ const (
pagePagedBefore // the journal says it was sent 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
)
// String is the outcome as the agent's log says it.
@@ -89,6 +100,8 @@ func (o pageOutcome) String() string {
return "paged before"
case pageUnwired:
return "not paged: no pager is wired"
+ case pagePending:
+ return "not paged yet: it is sent again after a backoff"
}
return "not paged: the page was not delivered"
}
@@ -105,41 +118,112 @@ 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"`
+}
+
+// A refused page is tried again pageRetryFirst after its first attempt, and
+// after each further attempt twice as long again, up to pageRetryMax.
+const (
+ pageRetryFirst = time.Minute
+ pageRetryMax = 30 * time.Minute
+)
+
+// pageBackoff is the wait after the tries-th attempt.
+func pageBackoff(tries int) time.Duration {
+ if tries < 1 {
+ tries = 1
+ }
+ if tries > 6 {
+ return pageRetryMax
+ }
+ if d := pageRetryFirst << (tries - 1); d < pageRetryMax {
+ return d
+ }
+ 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 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).
+// 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
}
- stamp := now.UTC().Format(time.RFC3339)
- paged := false
+ fresh := pendingPage{Level: level, Subject: subject, Body: body}
+ p, send, out := claimPage(dataDir, host, op, rule, now, &fresh)
+ if !send {
+ return out
+ }
+ return deliverPage(ctx, page, dataDir, host, op, rule, now, p)
+}
+
+// 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) {
- paged = true
+ out = pagePagedBefore
return errJournalUnchanged
}
- if j.Done(pendingStep + rule) {
- return errJournalUnchanged // pending and never sent: send it again
+ 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
+ }
+ p.Tries++
+ p.Next = now.Add(pageBackoff(p.Tries))
+ b, err := json.Marshal(p)
+ if err != nil {
+ return err
}
- j.Mark(pendingStep+rule, stamp)
+ j.Mark(pendingStep+rule, string(b))
+ send = true
return nil
})
- if paged {
- return pagePagedBefore
- }
- if err != nil && !errors.Is(err, errJournalUnchanged) {
+ 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
}
- if err := page(ctx, level, subject, body); err != nil {
- 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)
+ pagingLogf("moves: %s: the pending %s page of %s waits: its journal: %v", host, rule, op, err)
+ return p, false, out
+}
+
+// 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)
return pageUndelivered
}
if err := moves.UpdateJournal(dataDir, host, op, func(j *moves.Journal) error {
- j.Mark(pagedStep+rule, stamp)
+ 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)
@@ -147,6 +231,33 @@ func pageOnce(ctx context.Context, page func(ctx context.Context, level, subject
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
+ }
+ var rules []string
+ for step := range j.Steps {
+ if rule, ok := strings.CutPrefix(step, pendingStep); ok && !j.Done(pagedStep+rule) {
+ rules = append(rules, rule)
+ }
+ }
+ sort.Strings(rules)
+ now := w.now()
+ for _, rule := range rules {
+ p, send, _ := claimPage(dataDir, host, op, rule, now, nil)
+ if !send {
+ continue
+ }
+ wrote = true
+ deliverPage(ctx, w.cfg.Page, dataDir, host, op, rule, now, p)
+ }
+ return wrote
+}
+
// 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
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go
index 74872e5c5..21f62e695 100644
--- a/internal/host/movespaging_test.go
+++ b/internal/host/movespaging_test.go
@@ -359,6 +359,12 @@ func TestPagingIsRateLimited(t *testing.T) {
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)
}
@@ -370,6 +376,79 @@ func TestPagingIsRateLimited(t *testing.T) {
}
})
+ // 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) {
+ 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)
+ }
+ if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(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") {
+ t.Fatalf("a refused page counted as sent: %+v", pages)
+ }
+ refuse = false
+ h.pass(t) // inside the backoff: not yet
+ if len(pages) != 0 {
+ 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
+ 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")
+ }
+ h.clock.Advance(time.Hour)
+ h.pass(t)
+ h.freshWorker() // a restarted agent
+ h.pass(t)
+ pagedOnce(t, pages, "ie-box", "source", provision.StuckBaselineMismatch)
+ })
+
+ t.Run("sweep: a journal that is gone sends nothing", func(t *testing.T) {
+ dd := t.TempDir()
+ 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)
+ }
+ 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)
+ }
+ 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)
+ }
+ })
+
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)
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index f516507bb..733fe9524 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -877,7 +877,7 @@ 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(recs)
+ w.sweepFinalJournals(ctx, recs)
}
w.liftFences(recs)
return nil
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go
index 2a0219c19..01709111e 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([]provision.Record{h.read(t, p1Host)})
+ h.w().sweepFinalJournals(context.Background(), []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 63260f601..dd108a91a 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([]provision.Record{h.read(t, p1Host)})
+ w.sweepFinalJournals(context.Background(), []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 1d24ae40b..0f7d6dd22 100644
--- a/internal/moves/journal.go
+++ b/internal/moves/journal.go
@@ -235,7 +235,13 @@ 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.
+// 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
diff --git a/internal/moves/journal_lock_test.go b/internal/moves/journal_lock_test.go
index dd51e629e..d7d1075a4 100644
--- a/internal/moves/journal_lock_test.go
+++ b/internal/moves/journal_lock_test.go
@@ -1,6 +1,10 @@
package moves
import (
+ "io/fs"
+ "os"
+ "path/filepath"
+ "strings"
"testing"
)
@@ -56,3 +60,51 @@ 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)
+ }
+ }
+}