amadan

carlos / platform Public

host: the page index keeps its attempt count, and reads it under its lock

moves-phase1-6 Tasks 4+5, re-review 3's two Minor findings.
- A disk entry's attempt count and deadline that the disk could not
  record stay in the index past the deadline; the next attempt counts
  from the larger of the disk's and the index's tries, and the index
  item goes only once a write of the entry lands (or the disk says it
  was delivered). A stale disk count of one no longer restarts the
  backoff at two minutes. The backoff test checks the following
  deadline too (four minutes).
- firePage reads an index item's delivered flag while holding
  pageOutboxMu.

No censused unit changes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude dd363b1c50bffb65e1cf591055b0dbd56c508313 parent 1defacc
2 files changed, +31 −7
  • internal/host/movespaging.go +18 −7
  • internal/host/movespaging_test.go +13 −0
diff --git a/internal/host/movespaging.go b/internal/host/movespaging.go
index 285d74f17..30e0cb631 100644
--- a/internal/host/movespaging.go
+++ b/internal/host/movespaging.go
@@ -261,8 +261,9 @@ func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, hos
}
}
if item := pageIndex[key]; item != nil {
+ delivered := item.delivered
pageOutboxMu.Unlock()
- if item.delivered {
+ if delivered {
return pagePagedBefore
}
return pagePending
@@ -348,7 +349,8 @@ func markPageDelivered(dataDir string, e pageEntry, now time.Time) error {
// 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.
+// index (fix round 3, ruling 2), with its attempt count, until a write of
+// the entry lands (re-review 3). Unwired, the disk's pages wait.
func (w *movesWorker) flushPageOutbox(ctx context.Context) {
page, dd := w.cfg.Page, w.cfg.DataDir
now := w.now()
@@ -394,24 +396,35 @@ func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p s
pagingLogf("moves: the page outbox: %s is not an entry, and is left for an operator: %v", p, err)
return
}
+ key := pageKeyOf(dd, e)
+ item := pageIndex[key]
if e.Delivered != nil {
+ if item != nil && !item.memory {
+ delete(pageIndex, key) // the disk says it all now
+ }
pageOutboxMu.Unlock()
return
}
- key := pageKeyOf(dd, e)
- if item := pageIndex[key]; item != nil && (item.delivered || now.Before(item.next)) {
+ if 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) {
+ if item == nil && now.Before(e.Next) {
pageOutboxMu.Unlock()
return
}
+ // The index's attempts and deadline, when the disk could not record
+ // them, stand over the disk's until a write lands (re-review 3).
+ if item != nil && item.e.Tries > e.Tries {
+ e.Tries = item.e.Tries
+ }
e.Tries++
e.Next = now.UTC().Add(pageBackoff(e.Tries))
if err := writePageEntry(dd, e); err != nil {
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)
+ } else if item != nil {
+ delete(pageIndex, key)
}
pageOutboxMu.Unlock()
deliverPageEntry(ctx, page, dd, e, now)
@@ -434,8 +447,6 @@ func flushPageIndex(ctx context.Context, now time.Time) {
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()
diff --git a/internal/host/movespaging_test.go b/internal/host/movespaging_test.go
index 7adf7a458..fe3b8a7d4 100644
--- a/internal/host/movespaging_test.go
+++ b/internal/host/movespaging_test.go
@@ -592,6 +592,19 @@ func TestPagingIsRateLimited(t *testing.T) {
if attempts != 3 {
t.Fatalf("attempts after the backoff %d, want 3", attempts)
}
+ // The count the disk could not record is kept too (re-review 3): the
+ // third attempt's backoff is 4 minutes, not the 2 a stale disk count
+ // of one would give again.
+ now = now.Add(2*time.Minute + 30*time.Second)
+ tFlush(ctx, w)
+ if attempts != 3 {
+ t.Fatalf("the third attempt's backoff restarted from the disk's stale count: %d attempts at 2m30s", attempts)
+ }
+ now = now.Add(2 * time.Minute) // 4m30s after the third attempt
+ tFlush(ctx, w)
+ if attempts != 4 {
+ t.Fatalf("attempts after the 4-minute backoff %d, want 4", 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) {