amadan

carlos / platform Public

moves: the lost-box gate covers the whole pass, an unreadable claim included (Task 4 fix round 1, item 3)

hostBlocked decides once per pass per host: the claim lists this box in
Lost, or the claim does not read. The record's round, the target's
report pass and the op journal sweep all skip a blocked host, so nothing
is reported, committed or retired for it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 79e817dd72b248b5c67e533eff55353a6d7f06e7 parent c339c8c
5 files changed, +128 −30
  • internal/host/movesfinal.go +4 −1
  • internal/host/movespass.go +44 −27
  • internal/host/movespass_lost_test.go +78 −0
  • internal/host/movesrevert_test.go +1 −1
  • internal/host/movesrollback_test.go +1 −1
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 1df6fc778..f07e71d33 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -781,7 +781,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, s release.Store, recs []provision.Record) {
dd := w.cfg.DataDir
if dd == "" {
return
@@ -802,6 +802,9 @@ func (w *movesWorker) sweepFinalJournals(recs []provision.Record) {
if !hd.IsDir() || !journalComponent(host) {
continue
}
+ if w.hostBlocked(ctx, s, host) {
+ continue // the lost-box gate (movespass.go hostBlocked)
+ }
ops, err := os.ReadDir(filepath.Join(dd, ".moves", host))
if err != nil {
w.notice(host+"#final-sweep", "moves: %s: the op journal sweep cannot list its journals: %v", host, err)
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 4be316274..d83535a37 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -173,32 +173,51 @@ func movesNamesBox(cfg moveCfg, rec provision.Record) bool {
return rec.Move != nil && !rec.Move.Legacy() && (rec.Region == cfg.Region || rec.Move.To == cfg.Region || rec.Move.From == cfg.Region)
}
-// lostHere is the lost-box gate (moves Q9, Plan 4 Task 10, delta M15): a
-// box the host's claim lists in Lost runs no move step for the host, in any
-// role and at any state, until the claim readmits it, and logs one standing
-// notice. The quarantine's local retirement is the claim's (readmission
-// requires its retirement receipt); moves adds only this gate. The answer
-// holds for the pass (w.lost), so the target's report pass skips the host
-// too. A record that names neither this box's region nor a move is not read.
-func (w *movesWorker) lostHere(ctx context.Context, s release.Store, rec provision.Record) (bool, error) {
- if w.cfg.Box == "" || !movesNamesBox(w.cfg, rec) {
- return false, nil
+// lostHere is the lost-box gate (moves Q9, Plan 4 Task 10, delta M15) for
+// a record's round: a record that names this box's region is gated by
+// hostBlocked. A record that names neither this box's region nor a move is
+// not read.
+func (w *movesWorker) lostHere(ctx context.Context, s release.Store, rec provision.Record) bool {
+ if !movesNamesBox(w.cfg, rec) {
+ return false
}
- obs, err := hostclaim.Read(ctx, s, rec.Host)
- if err != nil {
- return false, fmt.Errorf("moves: read the host claim: %w", err)
+ return w.hostBlocked(ctx, s, rec.Host)
+}
+
+// hostBlocked decides, once per pass, whether this box runs no move step
+// for host: the host's claim lists this box in Lost (until the claim
+// readmits it), or the claim does not read. The quarantine's local
+// retirement is the claim's (readmission requires its retirement receipt);
+// moves adds only this gate. The decision holds for the whole pass (w.lost):
+// the record's round, the target's report pass and the op journal sweep
+// all skip a blocked host, and one standing notice is logged.
+func (w *movesWorker) hostBlocked(ctx context.Context, s release.Store, host string) bool {
+ if w.cfg.Box == "" {
+ return false
+ }
+ if w.lost == nil {
+ w.lost = map[string]bool{}
}
- if obs.Absent || obs.Claim == nil {
- return false, nil
+ if blocked, decided := w.lost[host]; decided {
+ return blocked
}
- for _, box := range obs.Claim.Lost {
- if box == w.cfg.Box {
- w.lost[rec.Host] = true
- w.notice(rec.Host+"#lost", "moves: %s: this box (%s) is lost on this host's claim: it runs no move step until the claim readmits it", rec.Host, box)
- return true, nil
+ blocked := false
+ obs, err := hostclaim.Read(ctx, s, host)
+ switch {
+ case err != nil:
+ blocked = true
+ w.notice(host+"#lost", "moves: %s: the host claim does not read: this box runs no move step for it this pass: %v", host, err)
+ case !obs.Absent && obs.Claim != nil:
+ for _, box := range obs.Claim.Lost {
+ if box == w.cfg.Box {
+ blocked = true
+ w.notice(host+"#lost", "moves: %s: this box (%s) is lost on this host's claim: it runs no move step until the claim readmits it", host, box)
+ break
+ }
}
}
- return false, nil
+ w.lost[host] = blocked
+ return blocked
}
var armedMovesWorker atomic.Pointer[movesWorker]
@@ -864,11 +883,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
continue
}
// The lost-box gate (Q9, Plan 4 Task 10): a box the host's claim
- // lists in Lost runs no move step for the host.
- if lost, err := w.lostHere(ctx, s, rec); err != nil {
- w.logTrouble(rec.Host, err)
- continue
- } else if lost {
+ // lists in Lost, or whose claim does not read, runs no move step
+ // for the host this pass.
+ if w.lostHere(ctx, s, rec) {
continue
}
if rec.Move.State == provision.MoveRollingBack && movesParticipantRecord(cfg, rec) == roleSource {
@@ -983,7 +1000,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, s, recs)
}
if cfg.Enabled {
// Delivered pages of moves retired on this box are removed, now that
diff --git a/internal/host/movespass_lost_test.go b/internal/host/movespass_lost_test.go
index 4f9483b10..533f4d53e 100644
--- a/internal/host/movespass_lost_test.go
+++ b/internal/host/movespass_lost_test.go
@@ -7,7 +7,9 @@ import (
"testing"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
)
// count is how many times s was logged (housekeeper_test.go's buffer).
@@ -107,3 +109,79 @@ func TestReturningLostBoxRefusesEveryMoveStep(t *testing.T) {
}
})
}
+
+// rcCorruptClaim makes the host claim unreadable: its object no longer
+// decodes.
+func rcCorruptClaim(t *testing.T, s release.Store) {
+ t.Helper()
+ key, err := hostclaim.Key(p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, etag, err := s.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(t.Context(), key, []byte("{not a claim"), etag); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// Fix round 1, item 3: the gate covers the whole pass. A host the gate
+// blocks, confirmed lost or with a claim that does not read, gets no target
+// report and no journal sweep: nothing is reported, committed or retired.
+func TestLostBoxGateCoversTheWholePass(t *testing.T) {
+ t.Run("an unreadable claim blocks the target's report", func(t *testing.T) {
+ buf := lostLog(t)
+ h := newPhase1Harness(t)
+ h.start(t, "auto")
+ h.pass(t)
+ rcCorruptClaim(t, h.s)
+ h.targetPass(t, h.targetCfg())
+ if r, err := readTargetReport(t.Context(), h.s, h.read(t, p1Host)); err != nil || r != nil {
+ t.Fatalf("a target with an unreadable claim reported: %+v %v", r, err)
+ }
+ if buf.count("does not read") == 0 {
+ t.Fatalf("no notice:\n%s", buf.String())
+ }
+ })
+ for name, block := range map[string]func(t *testing.T, h *revertHarness){
+ "a confirmed lost host": func(t *testing.T, h *revertHarness) {
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
+ c.Lost, c.LostRevisions = append(c.Lost, "ie-box"), map[string]uint64{"ie-box": 1}
+ })
+ },
+ "an unreadable claim": func(t *testing.T, h *revertHarness) { rcCorruptClaim(t, h.s) },
+ } {
+ t.Run(name+" blocks the journal sweep", func(t *testing.T) {
+ lostLog(t)
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.pass(t)
+ h.reopened(t)
+ h.rvAt = "after-close"
+ o := h.laneChore(t)
+ h.tPass(t)
+ h.pastHorizon(t, o)
+ h.tPass(t)
+ h.pass(t)
+ if !h.rvDone || h.read(t, p1Host).Move != nil || !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
+ t.Fatalf("the setup did not leave the closed revert's journal: %v", h.rvHits)
+ }
+ journal, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ block(t, h)
+ h.freshWorker()
+ h.pass(t)
+ if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
+ t.Fatal("the sweep retired a blocked host's journal")
+ }
+ if after, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID); err != nil || !reflect.DeepEqual(after.Steps, journal.Steps) {
+ t.Fatalf("the sweep committed a blocked host's journal: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go
index 2a0219c19..9731834d4 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(), h.s, []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..6c3e9c380 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(), h.s, []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)
}