amadan

carlos / platform Public

host: one page index over the outbox answers for what the disk cannot

moves-phase1-6 Tasks 4+5 fix round 3. An in-memory index keyed by
(outbox, host, move, rule) sits over the disk page outbox.
- A firing stats the disk entry first (outboxCreate now looks for the
  file before writing anything), then the index; only an identity in
  neither is created, and only a failed create is sent directly.
- A backoff the disk cannot record is kept in the index, so the next
  pass waits it out.
- A page the outbox cannot keep is in the index before its send, so
  re-fires add nothing; delivered, it is marked so there, and every
  pass writes its tombstone again until it lands. The same holds for a
  disk entry whose delivered mark cannot be written.
- The flush runs first in every pass, before any record or fleet read;
  tombstone retirement (retirePageTombstones) stays after a successful
  record listing.

No censused unit changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1defaccf4063aa1a239cb3f777deb0f8c95df8fa parent a7a22a6
4 files changed, +435 −150
  • internal/host/moveslogoutbox.go +7 −2
  • internal/host/movespaging.go +231 −134
  • internal/host/movespaging_test.go +190 −10
  • internal/host/movespass.go +7 −4
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go
index 0177ea650..0b42056a0 100644
--- a/internal/host/moveslogoutbox.go
+++ b/internal/host/moveslogoutbox.go
@@ -72,9 +72,14 @@ func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error {
// 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.
+// there: the first one kept stays, and an existing file is found before
+// anything is written, so storage that has turned full or read-only still
+// answers for it (Plan 6 Tasks 4+5 fix round 3). 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.Lstat(filepath.Join(dir, name)); err == nil {
+ return false, nil
+ }
if err := os.MkdirAll(dir, 0o700); err != nil {
return false, err
}
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go
index 956e202bf..285d74f17 100644
--- a/internal/host/movespaging.go
+++ b/internal/host/movespaging.go
@@ -51,17 +51,23 @@ import (
// 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.
+// again by every moves pass (flushPageOutbox, first thing in the pass, before
+// any record or fleet read) 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
+// (retirePageTombstones, only after a pass listed every record). The outbox
+// depends on nothing in the move's journal.
//
-// 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.
+// An index in memory (pageIndex, fix round 3) answers for what the disk
+// cannot: a firing looks at the disk entry, then the index, and only a page
+// in neither is created; only when that create fails is the page sent at
+// once, kept in the index, and, if that fails too, sent again by every pass
+// with a loud line in the agent's log. A delivery or a backoff the disk
+// cannot record is kept in the index too, and a delivered page's tombstone
+// is written again every pass until it lands. Only a restart while the index
+// holds what the disk does not 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
@@ -160,11 +166,35 @@ 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.
+// pageOutboxMu serializes every change to an outbox entry, and to the page
+// index, 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
+// The page index (fix round 3): what this process knows of a page that the
+// disk does not say, keyed by (outbox, host, move, rule). It holds a page
+// the outbox could not keep (memory), a backoff deadline the disk could not
+// record (next), and a delivery the disk does not record yet (delivered,
+// whose tombstone every pass writes again until it lands). A firing looks at
+// the disk and then the index, so one (host, move, rule) is one page however
+// the storage fails; only a restart while the index holds a page the disk
+// does not loses it.
+type pageKey struct{ dataDir, host, op, rule string }
+
+type pageIndexItem struct {
+ e pageEntry
+ page pageFunc // the pager that fired a memory page
+ memory bool // the outbox could not keep it
+ delivered bool
+ next time.Time
+}
+
+var pageIndex = map[pageKey]*pageIndexItem{}
+
+func pageKeyOf(dataDir string, e pageEntry) pageKey {
+ return pageKey{dataDir, e.Host, e.Op, e.Rule}
+}
+
// readPageEntry reads one entry.
func readPageEntry(path string) (pageEntry, error) {
var e pageEntry
@@ -187,11 +217,18 @@ func writePageEntry(dataDir string, e pageEntry) error {
return outboxReplace(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
}
-// 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.
+// pageEntryPath is an entry's file.
+func pageEntryPath(dataDir, host, op, rule string) string {
+ return filepath.Join(pageEntryDir(dataDir, host, op), rule+".json")
+}
+
+// firePage is a rule firing (fix round 3, ruling 1). The disk entry is
+// looked at first: there, delivered or not, the rule sends nothing (the
+// passes retry a refused one). Then the index: there, it sends nothing.
+// Only a (host, move, rule) in neither is created, its attempt counted, and
+// sent; and only when that create fails is the page sent directly, and kept
+// in the index if that fails too. 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 {
@@ -200,49 +237,69 @@ func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, hos
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)
+ if !outboxSegment(host) || !outboxSegment(op) || !outboxSegment(rule) || dataDir == "" {
+ pagingLogf("moves: %s: WARNING: the %s page of move %s has no usable outbox name, so it is sent now", host, rule, op)
+ if err := page(ctx, level, subject, body); err != nil {
+ pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered: %v", host, rule, op, err)
+ return pageUndelivered
}
- return pageUndelivered
+ return pageSent
}
- if !created {
- pageOutboxMu.Lock()
- kept, rerr := readPageEntry(filepath.Join(pageEntryDir(dataDir, host, op), rule+".json"))
+ key := pageKeyOf(dataDir, e)
+ pageOutboxMu.Lock()
+ if kept, err := readPageEntry(pageEntryPath(dataDir, host, op, rule)); err == nil {
pageOutboxMu.Unlock()
- if rerr == nil && kept.Delivered != nil {
+ if kept.Delivered != nil {
return pagePagedBefore
}
return pagePending
+ } else if !errors.Is(err, fs.ErrNotExist) {
+ if _, serr := os.Lstat(pageEntryPath(dataDir, host, op, rule)); serr == nil {
+ // There, but not readable now: it is the rule's entry all the same.
+ pageOutboxMu.Unlock()
+ return pagePending
+ }
}
- return deliverPageEntry(ctx, page, dataDir, e, now)
-}
-
-// 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)
+ if item := pageIndex[key]; item != nil {
+ pageOutboxMu.Unlock()
+ if item.delivered {
+ return pagePagedBefore
+ }
+ return pagePending
}
b, err := json.Marshal(e)
+ created := false
+ if err == nil {
+ created, err = outboxCreate(pageEntryDir(dataDir, host, op), rule+".json", b)
+ }
+ if err == nil && !created {
+ pageOutboxMu.Unlock()
+ return pagePending
+ }
if err != nil {
- return false, err
+ // The outbox cannot keep it: the index does, before the send, so no
+ // re-fire sends it again.
+ item := &pageIndexItem{e: e, page: page, memory: true, next: e.Next}
+ pageIndex[key] = item
+ pageOutboxMu.Unlock()
+ 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 {
+ 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)
+ return pageUndelivered
+ }
+ pageOutboxMu.Lock()
+ item.delivered = true
+ pageOutboxMu.Unlock()
+ return pageSent
}
- pageOutboxMu.Lock()
- defer pageOutboxMu.Unlock()
- return outboxCreate(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
+ pageOutboxMu.Unlock()
+ return deliverPageEntry(ctx, page, dataDir, e, now)
}
// 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.
+// it delivered once the pager took it: on disk, or, when the disk cannot
+// record it, in the index, whose tombstone the passes write again until it
+// lands. A refused page stays for the passes.
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)
@@ -250,40 +307,71 @@ func deliverPageEntry(ctx context.Context, page pageFunc, dataDir string, e page
}
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)
+ if err := markPageDelivered(dataDir, e, now); err != nil {
+ pageIndex[pageKeyOf(dataDir, e)] = &pageIndexItem{e: e, delivered: true}
+ pagingLogf("moves: %s: the %s page of move %s was sent; the outbox cannot say so yet, so this agent remembers it and writes it again every pass: %v", e.Host, e.Rule, e.Op, err)
}
return pageSent
}
-// 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) {
+// markPageDelivered writes e's entry as delivered: the entry on disk marked,
+// or, when there is none, a tombstone created. The caller holds
+// pageOutboxMu.
+func markPageDelivered(dataDir string, e pageEntry, now time.Time) error {
+ at := now.UTC()
+ kept, err := readPageEntry(pageEntryPath(dataDir, e.Host, e.Op, e.Rule))
+ switch {
+ case err == nil:
+ if kept.Delivered != nil {
+ return nil
+ }
+ kept.Delivered = &at
+ return writePageEntry(dataDir, kept)
+ case errors.Is(err, fs.ErrNotExist):
+ t := e
+ t.Delivered = &at
+ b, err := json.Marshal(t)
+ if err != nil {
+ return err
+ }
+ _, err = outboxCreate(pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b)
+ return err
+ }
+ return err
+}
+
+// flushPageOutbox is every moves pass's delivery half of paging (fix round 3,
+// ruling 4: it runs first, before any record or fleet read, so a bucket that
+// cannot be read never holds a page back). The index's pages are sent again
+// once their backoff is over and its delivered pages' tombstones written
+// again; then each disk entry not yet delivered is sent again once its
+// backoff (the later of the disk's and the index's) is over, its attempt
+// counted and the next put off before the send, so two senders in one
+// window send it once. A deadline the disk cannot record is kept in the
+// index (fix round 3, ruling 2). Unwired, the disk's pages wait.
+func (w *movesWorker) flushPageOutbox(ctx context.Context) {
page, dd := w.cfg.Page, w.cfg.DataDir
now := w.now()
- pageMemory.flush(ctx, now)
+ flushPageIndex(ctx, now)
if page == nil || dd == "" {
return
}
- live := map[string]bool{}
- for _, rec := range recs {
- if rec.Move != nil && rec.Move.ID != "" {
- live[rec.Host+"/"+rec.Move.ID] = true
+ root := pageOutboxDir(dd)
+ err := walkPageOutbox(root, func(p string) error {
+ if err := ctx.Err(); err != nil {
+ return err
}
+ w.flushPageEntry(ctx, page, dd, p, now)
+ return nil
+ })
+ if err != nil {
+ pagingLogf("moves: the page outbox: %v", err)
}
- root := pageOutboxDir(dd)
- err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+}
+
+// walkPageOutbox calls fn with each entry file under root.
+func walkPageOutbox(root string, fn func(p string) error) error {
+ return filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
if errors.Is(err, fs.ErrNotExist) && p == root {
return fs.SkipAll
}
@@ -293,20 +381,12 @@ func (w *movesWorker) flushPageOutbox(ctx context.Context, recs []provision.Reco
if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") {
return nil
}
- if err := ctx.Err(); err != nil {
- return err
- }
- w.flushPageEntry(ctx, page, dd, p, now, live)
- return nil
+ return fn(p)
})
- if err != nil {
- pagingLogf("moves: the page outbox: %v", err)
- }
- pruneEmptyDirs(root)
}
-// 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) {
+// flushPageEntry is one disk entry of flushPageOutbox.
+func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p string, now time.Time) {
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))) {
@@ -315,15 +395,14 @@ func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p s
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)
- }
- }
pageOutboxMu.Unlock()
return
}
+ key := pageKeyOf(dd, e)
+ if item := pageIndex[key]; item != nil && (item.delivered || now.Before(item.next)) {
+ pageOutboxMu.Unlock()
+ return // delivered (its mark is the index's to write) or backing off
+ }
if now.Before(e.Next) {
pageOutboxMu.Unlock()
return
@@ -331,64 +410,82 @@ func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p s
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)
+ pageIndex[key] = &pageIndexItem{e: e, next: e.Next}
+ pagingLogf("moves: %s: the %s page of move %s is sent with its backoff kept in memory: the outbox cannot record it: %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)
+// flushPageIndex sends each memory page whose backoff is over (a delivered
+// one is marked delivered, a refused one waits for its next attempt), and
+// writes each delivered page's tombstone, dropping it from the index once
+// it lands (fix round 3, ruling 3).
+func flushPageIndex(ctx context.Context, now time.Time) {
+ pageOutboxMu.Lock()
+ var due []*pageIndexItem
+ for key, item := range pageIndex {
+ switch {
+ case item.delivered:
+ if err := markPageDelivered(key.dataDir, item.e, now); err == nil {
+ delete(pageIndex, key)
+ }
+ case item.memory && !now.Before(item.next):
+ item.e.Tries++
+ item.next = now.UTC().Add(pageBackoff(item.e.Tries))
+ due = append(due, item)
+ case !item.memory && !now.Before(item.next):
+ delete(pageIndex, key) // the disk's own deadline governs again
+ }
+ }
+ pageOutboxMu.Unlock()
+ for _, item := range due {
+ e := item.e
+ if err := item.page(ctx, e.Level, e.Subject, e.Body); err != nil {
+ pagingLogf("moves: %s: WARNING: the %s page of move %s, kept in memory only, was not delivered (try %d): %v", e.Host, e.Rule, e.Op, e.Tries, err)
+ continue
+ }
+ pagingLogf("moves: %s: the %s page of move %s, kept in memory, is delivered", e.Host, e.Rule, e.Op)
+ pageOutboxMu.Lock()
+ item.delivered = true
+ pageOutboxMu.Unlock()
+ }
}
-// 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
+// retirePageTombstones removes each delivered disk entry whose move is
+// retired on this box: no record in recs carries it and its journal is gone.
+// Only a pass that listed every record calls it.
+func (w *movesWorker) retirePageTombstones(recs []provision.Record) {
+ dd := w.cfg.DataDir
+ if dd == "" {
+ return
+ }
+ live := map[string]bool{}
+ for _, rec := range recs {
+ if rec.Move != nil && rec.Move.ID != "" {
+ live[rec.Host+"/"+rec.Move.ID] = true
}
- 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)
+ root := pageOutboxDir(dd)
+ err := walkPageOutbox(root, func(p string) error {
+ pageOutboxMu.Lock()
+ defer pageOutboxMu.Unlock()
+ e, err := readPageEntry(p)
+ if err != nil || e.Delivered == nil || live[e.Host+"/"+e.Op] || moveJournalPresent(dd, e.Host, e.Op) {
+ return nil
+ }
+ if _, held := pageIndex[pageKeyOf(dd, e)]; held {
+ return nil
+ }
+ if err := os.Remove(p); err != nil {
+ pagingLogf("moves: the page outbox: %v", err)
}
- m.keep(kp.page, kp.e)
+ return nil
+ })
+ if err != nil {
+ pagingLogf("moves: the page outbox: %v", err)
}
+ pruneEmptyDirs(root)
}
// firePage is the worker's firePage: its pager, outbox and clock.
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go
index 3825ba6a8..7adf7a458 100644
--- a/internal/host/movespaging_test.go
+++ b/internal/host/movespaging_test.go
@@ -4,14 +4,17 @@ import (
"context"
"errors"
"fmt"
+ "io/fs"
"os"
"path/filepath"
"strings"
"testing"
"time"
+ "github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
)
// pagesWith counts the pages whose subject starts with prefix.
@@ -49,6 +52,62 @@ func pageDelivered(t *testing.T, dataDir, host, op, rule string) bool {
return e.Delivered != nil
}
+// memoryPending counts the pages the index holds for dataDir that the
+// outbox could not keep and that are not delivered yet.
+func memoryPending(dataDir string) int {
+ pageOutboxMu.Lock()
+ defer pageOutboxMu.Unlock()
+ n := 0
+ for key, item := range pageIndex {
+ if key.dataDir == dataDir && item.memory && !item.delivered {
+ n++
+ }
+ }
+ return n
+}
+
+// tFlush is one pass's page flush, with no record listing.
+func tFlush(ctx context.Context, w *movesWorker) { w.flushPageOutbox(ctx) }
+
+// readOnlyDir makes dir and everything under it unwritable until the test
+// ends (or restore is called).
+func readOnlyDir(t *testing.T, dir string) (restore func()) {
+ t.Helper()
+ var dirs []string
+ filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
+ if err == nil && d.IsDir() {
+ dirs = append(dirs, p)
+ }
+ return nil
+ })
+ for _, d := range dirs {
+ if err := os.Chmod(d, 0o500); err != nil {
+ t.Fatal(err)
+ }
+ }
+ restore = func() {
+ for _, d := range dirs {
+ os.Chmod(d, 0o700)
+ }
+ }
+ t.Cleanup(restore)
+ return restore
+}
+
+// unreachableStore answers every read and listing with an error, as a bucket
+// that cannot be reached.
+type unreachableStore struct{ release.Store }
+
+var errBucketDown = errors.New("the bucket cannot be reached")
+
+func (unreachableStore) Get(context.Context, string) ([]byte, string, error) {
+ return nil, "", errBucketDown
+}
+func (unreachableStore) List(context.Context, string) ([]string, error) { return nil, errBucketDown }
+func (unreachableStore) ListDirs(context.Context, string) ([]string, error) {
+ return nil, errBucketDown
+}
+
// refusingPager is a pager that refuses while *refuse is set, recording
// what it takes into pages.
func refusingPager(pages *pagesSeen, refuse *bool) pageFunc {
@@ -370,7 +429,6 @@ func TestPagingIsRateLimited(t *testing.T) {
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 {
@@ -386,21 +444,23 @@ func TestPagingIsRateLimited(t *testing.T) {
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)
+ if memoryPending(dd) != 1 || !strings.Contains(strings.Join(*logged, "\n"), "kept in memory only") {
+ t.Fatalf("the page is not kept in memory, loudly: %d %q", memoryPending(dd), *logged)
}
refuse = false
- w.flushPageOutbox(ctx, nil) // inside the backoff
+ tFlush(ctx, w) // inside the backoff
if len(pages) != 0 {
t.Fatalf("retried inside its backoff: %+v", 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())
+ tFlush(ctx, w)
+ w.retirePageTombstones(nil)
+ if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || memoryPending(dd) != 0 {
+ t.Fatalf("the memory's page: %d delivered, %d kept; want 1 and 0", n, memoryPending(dd))
}
now = now.Add(time.Hour)
- w.flushPageOutbox(ctx, nil)
+ tFlush(ctx, w)
+ w.retirePageTombstones(nil)
if len(pages) != 1 {
t.Fatalf("a delivered memory page was sent again: %+v", pages)
}
@@ -424,7 +484,8 @@ func TestPagingIsRateLimited(t *testing.T) {
live := []provision.Record{rec}
for i := 0; i < 3; i++ {
now = now.Add(time.Hour)
- w.flushPageOutbox(ctx, live)
+ tFlush(ctx, w)
+ w.retirePageTombstones(live)
w.pageRolledBack(ctx, rec)
worker().pageRolledBack(ctx, rec)
}
@@ -440,7 +501,8 @@ func TestPagingIsRateLimited(t *testing.T) {
if err := j.Commit(dd); err != nil {
t.Fatal(err)
}
- w.flushPageOutbox(ctx, nil)
+ tFlush(ctx, w)
+ w.retirePageTombstones(nil)
if !pageDelivered(t, dd, p1Host, p1ID, pageRuleRolledBack) {
t.Fatal("the tombstone went while the move's journal is on the box")
}
@@ -477,4 +539,122 @@ func TestPagingIsRateLimited(t *testing.T) {
t.Fatalf("M-C20 in the agent's log %d times, want once: %q", n, *logged)
}
})
+
+ // Fix round 3: the outbox's failure paths. An in-memory index over the
+ // disk outbox keeps each (host, move, rule) once, through any storage
+ // failure.
+ t.Run("index: a delivered tombstone on read-only storage stops every re-fire", func(t *testing.T) {
+ dd := t.TempDir()
+ now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
+ var pages pagesSeen
+ w := &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.MoveStuck}}
+ w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
+ if !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") {
+ t.Fatal("not delivered")
+ }
+ readOnlyDir(t, pageOutboxDir(dd))
+ for i := 0; i < 3; i++ {
+ w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
+ (&movesWorker{cfg: w.cfg}).pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
+ }
+ if len(pages) != 1 {
+ t.Fatalf("re-fires on read-only storage paged again: %d pages", len(pages))
+ }
+ })
+
+ t.Run("index: a backoff the disk cannot record still holds", func(t *testing.T) {
+ dd := t.TempDir()
+ now := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
+ attempts := 0
+ refuse := func(context.Context, string, string, string) error {
+ attempts++
+ return errors.New("the webhook is down")
+ }
+ w := &movesWorker{cfg: moveCfg{DataDir: dd, Box: "ie-box", Now: func() time.Time { return now }, Page: refuse}}
+ rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
+ w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch) // attempt 1, next in 1 minute
+ readOnlyDir(t, pageOutboxDir(dd))
+ now = now.Add(2 * time.Minute)
+ tFlush(ctx, w) // attempt 2: the disk cannot record its backoff (2 minutes)
+ if attempts != 2 {
+ t.Fatalf("attempts %d, want 2", attempts)
+ }
+ for i := 0; i < 3; i++ {
+ now = now.Add(30 * time.Second)
+ tFlush(ctx, w)
+ }
+ if attempts != 2 {
+ t.Fatalf("retried inside a backoff the disk could not record: %d attempts", attempts)
+ }
+ now = now.Add(time.Minute) // past the 2-minute backoff
+ tFlush(ctx, w)
+ if attempts != 3 {
+ t.Fatalf("attempts after the backoff %d, want 3", attempts)
+ }
+ })
+
+ t.Run("index: stray fires in a double fault page once, and once the storage is back the tombstone is written", func(t *testing.T) {
+ dd := t.TempDir()
+ blocker := pageOutboxDir(dd)
+ if err := os.MkdirAll(filepath.Dir(blocker), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(blocker, []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: "ie-box", Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)}}
+ rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveCutting}}
+ for i := 0; i < 3; i++ {
+ w.pageStray(ctx, rec)
+ }
+ refuse = false
+ for i := 0; i < 3; i++ {
+ w.pageStray(ctx, rec) // re-fires while the page waits in memory
+ }
+ now = now.Add(2 * time.Minute)
+ tFlush(ctx, w)
+ tFlush(ctx, w)
+ w.pageStray(ctx, rec)
+ if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 {
+ t.Fatalf("stray pages during the double fault %d, want one: %+v", n, pages)
+ }
+ // The storage is back: the next pass writes the tombstone, and a
+ // re-fire, here or after a restart, does not page.
+ if err := os.Remove(blocker); err != nil {
+ t.Fatal(err)
+ }
+ tFlush(ctx, w)
+ if !pageDelivered(t, dd, p1Host, p1ID, pageRuleStray) {
+ t.Fatal("the delivered page's tombstone was not written once the storage came back")
+ }
+ w.pageStray(ctx, rec)
+ (&movesWorker{cfg: w.cfg}).pageStray(ctx, rec)
+ tFlush(ctx, w)
+ if n, _ := pagesWith(pages, "carlos move stray process: "); n != 1 {
+ t.Fatalf("stray pages after the storage came back %d, want still one: %+v", n, pages)
+ }
+ })
+
+ t.Run("index: a pending page is delivered while records and the fleet cannot be read", 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: "ie-box", Enabled: true, Now: func() time.Time { return now }, Page: refusingPager(&pages, &refuse)},
+ mirrors: map[string]lease.MirrorStore{}}
+ rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, State: provision.MoveStuck}}
+ w.pageStuck(ctx, rec, "source", provision.StuckBaselineMismatch)
+ refuse = false
+ now = now.Add(2 * time.Minute)
+ if err := movesPass(ctx, unreachableStore{}, w); err == nil {
+ t.Fatal("a pass over an unreachable bucket succeeded")
+ }
+ if n, _ := pagesWith(pages, "carlos move stuck: "); n != 1 || !pageDelivered(t, dd, p1Host, p1ID, "stuck-0") {
+ t.Fatalf("the pending page with the bucket down: %+v", pages)
+ }
+ })
}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index d3418e97f..5dca41750 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -735,7 +735,10 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
if cfg.claimRecords {
ctx = provision.WithRecordSequence(ctx)
}
- // The move log's kept entries are retried first, whatever their moves
+ // The pages the outbox and the index keep undelivered are sent again
+ // first, before any record or fleet read (movespaging.go, fix round 3).
+ w.flushPageOutbox(ctx)
+ // The move log's kept entries are retried next, whatever their moves
// now are (moveslogoutbox.go).
w.flushLogOutbox(ctx, s)
recs, err := provision.ListAllRecords(ctx, s)
@@ -882,9 +885,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
// no longer routes here (movesfinal.go, fix round 1).
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)
+ // Delivered pages of moves retired on this box are removed, now that
+ // every record is listed (movespaging.go).
+ w.retirePageTombstones(recs)
w.liftFences(recs)
return nil
}