amadan

carlos / platform Public

hostclaim: Plan 10 fix round 1 (revert branch in the recovery adapter, durable nine-start budget, quiet unwired passes, release outcome by kind, census counts, sweep keeps non-moves candidates, loss store bucket and progression checks, console policy precondition test, real loss fixtures, moves copy census rows)

Answers Opus P2-1/P2-2 and P3-1..6 and Astra's P3 on 9a32deaf. The
recovery attempt count lives in the box-local moves journal keyed by
move, rollback and claim incarnation; an unbound provider waits without
recording inputs or waking. HC5-N2 and HC5-N3 are pending copy rows, so
moves.DefaultEnabled stays false until Paul approves them.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude d41c4127cf6c80c0cfd6d941f6a3da5e34839919 parent 9a32dea
25 files changed, +986 −84
  • approved_moves_copy_test.go +7 −1
  • internal/activator/claimwake.go +8 −7
  • internal/activator/recoveryprobe.go +1 −1
  • internal/activator/recoveryprobe_test.go +10 −1
  • internal/console/claim_authority_census_test.go +1 −0
  • internal/console/claim_move_recovery_wiring_test.go +160 −0
  • internal/console/testdata/claim-authority-sites.txt +10 −0
  • internal/host/claimoperations.go +11 −0
  • internal/host/claimwakeassembly.go +1 −1
  • internal/host/movesprobeattempts.go +101 −11
  • internal/host/movesprobeattempts_test.go +133 −0
  • internal/host/movesrecoveryprobe.go +17 −3
  • internal/host/movesrevert.go +9 −0
  • internal/host/plan10_j2_adapter_test.go +179 −7
  • internal/hostclaim/lossrecord.go +10 −1
  • internal/hostops/move_loss_grant.go +25 −5
  • internal/hostops/move_loss_scope_fix_test.go +181 −0
  • internal/hostops/move_loss_test.go +0 −11
  • internal/hostops/move_recovery_grant.go +8 −11
  • internal/hostops/move_recovery_grant_test.go +39 −1
  • internal/hostops/recovery.go +3 −0
  • internal/park/move_sweep_test.go +46 −1
  • internal/park/sweep.go +19 −17
  • internal/provision/plan9_b8_census_test.go +4 −4
  • testdata/claim-moves-copy.md +3 −1

CI passed — run details

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index ccd011d44..ae295e66b 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -39,7 +39,7 @@ const movesCopyRecordPath = "docs/superpowers/plans/2026-10-06-moves-phase1-copy
// "Gate and copy" task appends its ids here as it appends its rows to the
// record. The record and this literal must agree both ways.
var movesCopyIDs = []string{
- "HC5-N1",
+ "HC5-N1", "HC5-N2", "HC5-N3",
"M-C1", "M-C2", "M-C3", "M-C4", "M-C5", "M-C6", "M-C7", "M-C8", "M-C9",
"M-C10", "M-C11", "M-C12", "M-C13", "M-C14", "M-C15", "M-C16", "M-C17",
"M-C18", "M-C19", "M-C20", "M-C21", "M-C22", "MV-N1", "MV-N2", "MV-N3",
@@ -543,6 +543,12 @@ func TestMovesTextAppliedPlaceholders(t *testing.T) {
// copy" task appends the files it touched; TestMovesScanFilesCoverTheDiff
// fails when the list falls behind.
var movesScanFiles = []string{
+ "internal/hostops/move_recovery_grant.go",
+ "internal/hostops/move_loss_owner.go",
+ "internal/host/movesrecoveryprobe.go",
+ "internal/host/claimproduction.go",
+ "internal/host/claimowners.go",
+ "internal/host/claimmoveaccess.go",
"cmd/console/main.go",
"internal/activator/backend_exec.go",
"internal/activator/claimwake.go",
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index bc5abd0bd..68534ff90 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -66,13 +66,14 @@ type startPolicy = StartPolicy
// Admit reserves the first serving attempt and binds its resulting cold session.
// Prepared probes instead consume the moves owner's single-use admission.
type ClaimRuntime struct {
- FailedAdmission func(context.Context, hostclaim.Observed, error) error
- Box string
- Read func(context.Context, string) (hostclaim.Observed, error)
- StopForeign func(context.Context, string) error
- Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
- Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
- Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
+ RecoveryAttemptsReady func() bool
+ FailedAdmission func(context.Context, hostclaim.Observed, error) error
+ Box string
+ Read func(context.Context, string) (hostclaim.Observed, error)
+ StopForeign func(context.Context, string) error
+ Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
+ Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
+ Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
// Pins reports that Prepare honours StartPolicy.Pin. A hold swap is
// refused on a claims-on box whose runtime does not, before anything
// stops.
diff --git a/internal/activator/recoveryprobe.go b/internal/activator/recoveryprobe.go
index 0df1801e1..3783923bf 100644
--- a/internal/activator/recoveryprobe.go
+++ b/internal/activator/recoveryprobe.go
@@ -105,7 +105,7 @@ func (a *Service) RecoveryProbeReady() bool {
return false
}
r := a.cfg.Claims
- return r != nil && r.Box != "" && r.Read != nil && r.StopForeign != nil && r.Prerequisites != nil && r.Admit != nil && r.Prepare != nil && r.FailedAdmission != nil
+ return r != nil && r.RecoveryAttemptsReady != nil && r.RecoveryAttemptsReady() && r.Box != "" && r.Read != nil && r.StopForeign != nil && r.Prerequisites != nil && r.Admit != nil && r.Prepare != nil && r.FailedAdmission != nil
}
// recoveryProbeLimit bounds one probe call, as wake bounds a wake.
diff --git a/internal/activator/recoveryprobe_test.go b/internal/activator/recoveryprobe_test.go
index 5610a5b26..db34ba5c6 100644
--- a/internal/activator/recoveryprobe_test.go
+++ b/internal/activator/recoveryprobe_test.go
@@ -88,7 +88,8 @@ func probeService(t *testing.T, obs hostclaim.Observed) (*Service, *ClaimRuntime
work, _ := (hostwrite.Config{}).Select(t.Context(), probeHost, "spawn")
reads := 0
r := &ClaimRuntime{
- Box: "box",
+ RecoveryAttemptsReady: func() bool { return true },
+ Box: "box",
// The wake's first read sees obs; every later one (startAdmitted's
// fresh read) sees the admission.
Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
@@ -829,3 +830,11 @@ func TestRecoveryHoldSurvivesAdoptionAtProbed(t *testing.T) {
}
}
}
+
+func TestRecoveryProbeUnboundAttemptsAreNotReady(t *testing.T) {
+ a, r, _ := probeService(t, probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")))
+ r.RecoveryAttemptsReady = nil
+ if a.RecoveryProbeReady() {
+ t.Fatal("recovery probe ready without recovery attempts provider")
+ }
+}
diff --git a/internal/console/claim_authority_census_test.go b/internal/console/claim_authority_census_test.go
index 725ce6f57..1a702ccfc 100644
--- a/internal/console/claim_authority_census_test.go
+++ b/internal/console/claim_authority_census_test.go
@@ -90,6 +90,7 @@ func TestClaimAuthoritySourceCensus(t *testing.T) {
slices.Sort(want)
got := claimAuthoritySites(t)
if !slices.Equal(got, want) {
+ t.Errorf("authority census differs: got %d calls, want %d", len(got), len(want))
for _, s := range got {
if !slices.Contains(want, s) {
t.Errorf("unreviewed authority use: %s", s)
diff --git a/internal/console/claim_move_recovery_wiring_test.go b/internal/console/claim_move_recovery_wiring_test.go
new file mode 100644
index 000000000..8e75a20a3
--- /dev/null
+++ b/internal/console/claim_move_recovery_wiring_test.go
@@ -0,0 +1,160 @@
+package console
+
+import (
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "io/fs"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func moveRecoveryPolicyBound(f *ast.File) bool {
+ bound := false
+ named := func(e ast.Expr) bool {
+ switch e := e.(type) {
+ case *ast.Ident:
+ return e.Name == "ClaimMoveRecovery"
+ case *ast.SelectorExpr:
+ return e.Sel.Name == "ClaimMoveRecovery"
+ }
+ return false
+ }
+ nonnil := func(e ast.Expr) bool { id, ok := e.(*ast.Ident); return !ok || id.Name != "nil" }
+ ast.Inspect(f, func(n ast.Node) bool {
+ switch n := n.(type) {
+ case *ast.KeyValueExpr:
+ if named(n.Key) && nonnil(n.Value) {
+ bound = true
+ }
+ case *ast.AssignStmt:
+ for i, lhs := range n.Lhs {
+ if named(lhs) && (i >= len(n.Rhs) || nonnil(n.Rhs[i])) {
+ bound = true
+ }
+ }
+ }
+ return true
+ })
+ return bound
+}
+
+func releaseBeforeClose(f *ast.File, fn, release, close string) bool {
+ for _, d := range f.Decls {
+ f, ok := d.(*ast.FuncDecl)
+ if !ok || f.Name.Name != fn {
+ continue
+ }
+ var released, closed token.Pos
+ ast.Inspect(f.Body, func(n ast.Node) bool {
+ call, ok := n.(*ast.CallExpr)
+ if !ok {
+ return true
+ }
+ sel, ok := call.Fun.(*ast.SelectorExpr)
+ if !ok {
+ return true
+ }
+ if sel.Sel.Name == release && released == 0 {
+ released = call.Pos()
+ }
+ if sel.Sel.Name == close && closed == 0 {
+ closed = call.Pos()
+ }
+ return true
+ })
+ return released != 0 && (closed == 0 || released < closed)
+ }
+ return false
+}
+
+func TestClaimMoveRecoveryWiringWaitsForRecordFirstOwners(t *testing.T) {
+ root := "../.."
+ var owners []string
+ for _, owner := range []struct{ file, fn, release, close string }{
+ {"movesparks.go", "closeMove", "ReleaseMove", "MoveClose"},
+ {"movesrecover.go", "finalize", "ReleaseMove", "MoveRolledBackClose"},
+ {"movesrevert.go", "opened", "ReleaseMove", "close"},
+ {"movesabort.go", "abortSource", "releaseAbort", "MoveGuardedAbortComplete"},
+ } {
+ path := filepath.Join(root, "internal/host", owner.file)
+ f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if releaseBeforeClose(f, owner.fn, owner.release, owner.close) {
+ owners = append(owners, owner.file+":"+owner.fn)
+ }
+ }
+ if len(owners) == 0 {
+ return
+ }
+ err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ if d.IsDir() {
+ if path != root && (strings.HasPrefix(d.Name(), ".") || d.Name() == "testdata" || d.Name() == "vendor") {
+ return filepath.SkipDir
+ }
+ return nil
+ }
+ if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+ return nil
+ }
+ f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
+ if err != nil {
+ return err
+ }
+ if moveRecoveryPolicyBound(f) {
+ t.Errorf("%s wires ClaimMoveRecovery while owners release first: %v", path, owners)
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestClaimMoveRecoveryWiringGuardDetectsBindings(t *testing.T) {
+ for _, src := range []string{
+ `package p; func f(){ c.ClaimMoveRecovery = policy }`,
+ `package p; var c = Config{ClaimMoveRecovery: policy}`,
+ } {
+ f, err := parser.ParseFile(token.NewFileSet(), "fixture.go", src, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !moveRecoveryPolicyBound(f) {
+ t.Fatal("binding escaped census")
+ }
+ }
+ for _, src := range []string{
+ `package p; func f(){ c.ClaimMoveRecovery = nil }`,
+ `package p; var c = Config{ClaimMoveRecovery: nil}`,
+ } {
+ f, err := parser.ParseFile(token.NewFileSet(), "fixture.go", src, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if moveRecoveryPolicyBound(f) {
+ t.Fatal("nil policy counted as binding")
+ }
+ }
+ for _, tc := range []struct {
+ src string
+ want bool
+ }{
+ {`package p; func owner(){hostclaim.ReleaseMove();provision.MoveClose()}`, true},
+ {`package p; func owner(){provision.MoveClose();hostclaim.ReleaseMove()}`, false},
+ } {
+ f, err := parser.ParseFile(token.NewFileSet(), "fixture.go", tc.src, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := releaseBeforeClose(f, "owner", "ReleaseMove", "MoveClose"); got != tc.want {
+ t.Fatalf("release ordering=%v", got)
+ }
+ }
+}
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index 916e5cc68..96aaeeb23 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -21,10 +21,16 @@
../hostops/mutation_boundary.go|NewMutationBoundary|NewObservedMutationBoundary(enabled, raw, check, nil)
../hostops/production_held_start.go|startAppLifecycle|BeginPurge(id, by, a.Now(), regions, boxes)
../hostops/purge_control.go|DrainPurgeRegistrations|w.SavePurgeJournal(ctx, j, tag)
+../hostops/purge_control.go|DrainPurgeRegistrations|w.SavePurgeJournal(ctx, j, tag)
+../hostops/purge_control.go|DrainPurgeRegistrations|w.SavePurgeJournal(ctx, j, tag)
../hostops/purge_fences.go|CompletePurgeFences|c.Control.RemovePurgeJournal(ctx, account, app, id)
../hostops/purge_fences.go|CompletePurgeFences|c.Control.SavePurgeJournal(ctx, fresh, tag)
+../hostops/purge_fences.go|CompletePurgeFences|c.Control.SavePurgeJournal(ctx, fresh, tag)
../hostops/purge_hosts.go|CensusPurge|c.Control.SavePurgeJournal(ctx, j, tag)
../hostops/purge_hosts.go|ClaimPurgeHosts|c.Control.SavePurgeJournal(ctx, fresh, tag)
+../hostops/purge_hosts.go|ClaimPurgeHosts|c.Control.SavePurgeJournal(ctx, fresh, tag)
+../hostops/purge_hosts.go|RetirePurgeHosts|c.Control.SavePurgeJournal(ctx, fresh, tag)
+../hostops/purge_hosts.go|RetirePurgeHosts|c.Control.SavePurgeJournal(ctx, fresh, tag)
../hostops/purge_hosts.go|RetirePurgeHosts|c.Control.SavePurgeJournal(ctx, fresh, tag)
../hostops/purge_journal.go|DeletePurgeSweep|w.SavePurgeJournal(ctx, fresh, tag)
../hostops/purge_journal.go|PreparePurgeSweep|w.SavePurgeJournal(ctx, j, tag)
@@ -66,8 +72,12 @@ claim_purge.go|Resume|s.store.WithAppMutation(ctx, []sqlGrant{permitGrant(ctx, s
claim_purge.go|Resume|s.store.WithAppMutation(ctx, []sqlGrant{permitGrant(ctx, s.cfg.Releases, account, scope)}, func(st *Store) error { return st.BindClaimAppOperation(acct.ID, app, appID, id, revision) })
claim_purge.go|Resume|w.BeginPurge(ctx, account, app, appID, id, p.Owner)
claim_purge.go|Resume|w.RemovePurgeJournal(ctx, account, app, id)
+claim_purge.go|Resume|w.RemovePurgeJournal(ctx, account, app, id)
+claim_purge.go|Resume|w.SavePurgeJournal(ctx, fresh, tag)
claim_purge.go|Resume|w.SavePurgeJournal(ctx, fresh, tag)
claim_purge.go|Resume|w.SavePurgeJournal(ctx, j, tag)
+claim_purge.go|Resume|w.SavePurgeJournal(ctx, j, tag)
+claim_purge.go|Resume|w.SavePurgeJournal(ctx, j, tag)
claim_purge.go|captureDomains|w.SavePurgeJournal(ctx, j, tag)
claim_purge.go|teardown|w.SavePurgeJournal(ctx, j, tag)
claim_relay.go|relayAppMutation|release.WithAppMutation(ctx, s.cfg.Releases, acct, app, action, fn)
diff --git a/internal/host/claimoperations.go b/internal/host/claimoperations.go
index 3b31b9d96..de67e118a 100644
--- a/internal/host/claimoperations.go
+++ b/internal/host/claimoperations.go
@@ -193,3 +193,14 @@ func claimSpawnAction(c *hostclaim.Claim) string {
}
return "spawn"
}
+
+func (w *ClaimWakeAssembly) recoveryAttemptsReady() bool {
+ for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
+ ops, ready := w.operationOwner(hostclaim.Observed{Claim: &hostclaim.Claim{Op: hostclaim.Op{Kind: kind}}})
+ provider, ok := ops.(interface{ RecoveryProbeAttemptsReady() bool })
+ if !ok || !provider.RecoveryProbeAttemptsReady() || ready == nil {
+ return false
+ }
+ }
+ return true
+}
diff --git a/internal/host/claimwakeassembly.go b/internal/host/claimwakeassembly.go
index 5b468ab17..6c64291ce 100644
--- a/internal/host/claimwakeassembly.go
+++ b/internal/host/claimwakeassembly.go
@@ -201,7 +201,7 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
read := func(ctx context.Context, h string) (hostclaim.Observed, error) {
return hostclaim.Read(ctx, writer.Claims, h)
}
- runtime := &activator.ClaimRuntime{Box: writer.Box, Read: read}
+ runtime := &activator.ClaimRuntime{Box: writer.Box, Read: read, RecoveryAttemptsReady: w.recoveryAttemptsReady}
runtime.StopForeign = func(ctx context.Context, h string) error {
o, err := read(ctx, h)
if err != nil {
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go
index 29239567b..98dc5ae7f 100644
--- a/internal/host/movesprobeattempts.go
+++ b/internal/host/movesprobeattempts.go
@@ -19,7 +19,7 @@ import (
)
// movesProbeAttempts is the moves OperationAttempts implementation for a
-// landing probe (moves Phase 1 Plan 3 Task 5b; claim Plan 9 Group D, D2;
+// landing or recovery probe (moves Phase 1 Plan 3 Task 5b; claim Plan 9 Group D, D2;
// rulings H2, H10). The claim binds its methods into the admitted probe's
// activator.ClaimStart; this type journals, counts and settles the probe's
// starts in the box-local op journal (moves.Journal, <DataDir>/.moves/<host>/
@@ -76,6 +76,7 @@ type movesProbeAttempts struct {
// session is bound to.
type boundAttempt struct {
k int
+ key string
attempt hostclaim.WakeRetryAttempt
}
@@ -104,14 +105,16 @@ func (p *movesProbeAttempts) valid() bool {
return p != nil && p.Claims != nil && p.Records != nil && p.DataDir != "" && p.Admitted != nil && p.NewID != nil && p.Now != nil
}
-// bound is every method's check: it returns the record's attempt index k for
-// the admitted probe s, or 0 for a committed wake (Plan 3 Task 7), or a
-// refusal.
+// bound returns the landing record attempt, zero for a committed wake,
+// or -1 for a recovery probe. Every case validates the admitted writer.
func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, error) {
if !p.valid() {
return 0, hostwrite.ErrUnavailable
}
a := s.Admitted
+ if a.Claim != nil && claimRecoveryKind(a.Claim.Op.Kind) {
+ return -1, p.boundRecovery(ctx, s)
+ }
if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding {
return 0, hostbox.ErrUnsupported
}
@@ -178,8 +181,12 @@ func (p *movesProbeAttempts) boundCommitted(ctx context.Context, s OperationStar
}
// startKey names a start's journal step: the record attempt k for a probe,
-// the admitted session for a committed wake (k 0).
+// the admitted session for a committed wake (k 0), or the recovery and
+// admitted session for a recovery probe (k -1).
func startKey(s OperationStart, k int) string {
+ if k < 0 {
+ return recoveryAttemptKey(s) + "-" + strconv.FormatUint(s.Owner.Session, 10)
+ }
if k == 0 {
return "committed-" + strconv.FormatUint(s.Owner.Session, 10)
}
@@ -198,8 +205,8 @@ func probeKey(s OperationStart) string {
// Absent, it is written into j when write is set (the caller writes j) and
// refused otherwise.
func bindSession(j *moves.Journal, s OperationStart, k int, write bool) error {
- if k == 0 {
- return nil // a committed wake has no record attempt
+ if k <= 0 {
+ return nil // committed wakes and recoveries have no landing record attempt
}
step := probeStepSession + strconv.FormatUint(s.Owner.Session, 10)
raw, ok := j.Steps[step]
@@ -236,7 +243,7 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr
if err := bindSession(j, s, k, true); err != nil {
return time.Time{}, err
}
- step := probeStepOpened + strconv.Itoa(k)
+ step := probeStepOpened + startKey(s, k)
if raw, ok := j.Steps[step]; ok {
if err := j.Write(p.DataDir); err != nil {
return time.Time{}, err
@@ -335,6 +342,16 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
}
}
if st.Attempt.ID == "" {
+ if k < 0 {
+ count, err := recoveryAttemptCount(j, s)
+ if err != nil {
+ return err
+ }
+ if count >= recoveryProbeAttempts {
+ return ErrRecoveryProbeExhausted
+ }
+ j.Mark(recoveryAttemptKey(s), strconv.Itoa(count+1))
+ }
o := s.Owner
st.Attempt = hostclaim.WakeRetryAttempt{ID: p.NewID(), Box: o.Box, Enrollment: o.Enrollment, BootID: o.BootID, ControllerID: o.ControllerID, Session: o.Session, InvocationID: p.NewID(), LaunchInputs: digest}
if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
@@ -347,7 +364,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
if p.current == nil {
p.current = map[string]boundAttempt{}
}
- p.current[probeKey(s)] = boundAttempt{k: k, attempt: st.Attempt}
+ p.current[probeKey(s)] = boundAttempt{k: k, key: startKey(s, k), attempt: st.Attempt}
return nil
}
@@ -368,7 +385,7 @@ func (p *movesProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetr
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
- if k, err := p.bound(ctx, s); err != nil || k != cur.k {
+ if k, err := p.bound(ctx, s); err != nil || k != cur.k || startKey(s, k) != cur.key {
return hostclaim.WakeRetryAttempt{}
}
return cur.attempt
@@ -404,7 +421,19 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
return hostbox.ErrChanged
}
st.State = state
- return writeStart(p.DataDir, j, startKey(s, k), st)
+ if err := writeStart(p.DataDir, j, startKey(s, k), st); err != nil {
+ return err
+ }
+ if k < 0 && state == probeFailed {
+ count, err := recoveryAttemptCount(j, s)
+ if err != nil {
+ return err
+ }
+ if count >= recoveryProbeAttempts {
+ return ErrRecoveryProbeExhausted
+ }
+ }
+ return nil
}
// Healthy accepts start n after readiness, replication and publication:
@@ -468,3 +497,64 @@ func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n in
func probeAlive(cmd *exec.Cmd) bool {
return cmd != nil && cmd.Process != nil && cmd.Process.Signal(syscall.Signal(0)) == nil
}
+
+// RecoveryProbeAttemptsReady is checked before the adapter records inputs.
+func (p *movesProbeAttempts) RecoveryProbeAttemptsReady() bool { return p.valid() }
+
+func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) error {
+ a := s.Admitted
+ if a.Claim.Holder == nil || !s.Policy.Probe {
+ return hostbox.ErrChanged
+ }
+ h := a.Claim.Holder
+ if s.Owner.Host != a.Host || s.Owner.Incarnation != h.Incarnation || s.Owner.Stamp != h.Stamp || s.Owner.Session != h.Session || s.Owner.Box != a.Claim.Op.Exec {
+ return hostbox.ErrChanged
+ }
+ fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
+ if err != nil {
+ return err
+ }
+ if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseProbing || fresh.Claim.Op.Exec != s.Owner.Box {
+ return hostbox.ErrChanged
+ }
+ rec, err := p.Records(ctx, h.Account, h.App, a.Host)
+ if err != nil {
+ return err
+ }
+ if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation {
+ return hostbox.ErrChanged
+ }
+ m := rec.Move
+ if a.Claim.Op.Kind == hostclaim.KindReverting {
+ if m.State != provision.MoveReverting || m.ReopenedAt != nil || fresh.Claim.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
+ return hostbox.ErrChanged
+ }
+ } else if m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID == "" || m.Rollback.ID != a.Claim.Op.Intent["rollback"] || m.Rollback.ID != fresh.Claim.Op.Intent["rollback"] || m.Rollback.SourcePhase != provision.RbSourceProbing {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// The move has its own journal directory. JSON preserves the rollback and
+// incarnation boundaries; sessions share this counter across controller restarts.
+func recoveryAttemptKey(s OperationStart) string {
+ c := s.Admitted.Claim
+ rollback := c.Op.ID
+ if c.Op.Kind == hostclaim.KindRestoring {
+ rollback = c.Op.Intent["rollback"]
+ }
+ raw, _ := json.Marshal([]string{c.Op.ID, rollback, c.Holder.Incarnation})
+ return "recovery_attempts_" + string(raw)
+}
+
+func recoveryAttemptCount(j *moves.Journal, s OperationStart) (int, error) {
+ raw, ok := j.Steps[recoveryAttemptKey(s)]
+ if !ok {
+ return 0, nil
+ }
+ n, err := strconv.Atoi(raw)
+ if err != nil || n < 0 || n > recoveryProbeAttempts {
+ return 0, hostbox.ErrChanged
+ }
+ return n, nil
+}
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go
index 885421b8f..179cefc81 100644
--- a/internal/host/movesprobeattempts_test.go
+++ b/internal/host/movesprobeattempts_test.go
@@ -16,6 +16,7 @@ import (
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -372,3 +373,135 @@ func TestMovesProbeAttemptsCommittedWake(t *testing.T) {
t.Fatal("a committed wake's start over a landing record")
}
}
+
+func recoveryPAFixture(t *testing.T, kind string) *paFixture {
+ t.Helper()
+ f := newPAFixture(t)
+ c := f.read().Claim
+ c.Op.Kind, c.Op.Phase = kind, hostclaim.PhaseProbing
+ c.Op.Exec, c.Holder.WriterBox = "ie-box", "ie-box"
+ c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
+ f.rec.Move.State = provision.MoveReverting
+ if kind == hostclaim.KindRestoring {
+ c.Op.Intent["rollback"] = plan10J2Rollback
+ f.rec.Move.State = provision.MoveRollingBack
+ f.rec.Move.Rollback = &provision.MoveRollback{ID: plan10J2Rollback, SourcePhase: provision.RbSourceProbing}
+ }
+ f.putClaim(c)
+ f.start.Admitted = f.read()
+ f.start.Owner.Box = "ie-box"
+ f.start.Owner.Phase = hostclaim.PhaseProbing
+ return f
+}
+
+func TestRecoveryProbeAttemptsDurableBudget(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
+ t.Run(kind, func(t *testing.T) {
+ f := recoveryPAFixture(t, kind)
+ for n := 0; n < 9; n++ {
+ // Three starts per wake, then a new admitted session and provider.
+ local := n % 3
+ if local == 0 && n > 0 {
+ c := f.read().Claim
+ c.Gen++
+ c.Holder.Session = c.Gen
+ f.putClaim(c)
+ f.start.Admitted = f.read()
+ f.start.Owner.Session = c.Gen
+ f.ops = f.attempts()
+ }
+ if _, err := f.ops.OpenProbe(t.Context(), f.start, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.ops.Reserve(t.Context(), f.start, local); err != nil {
+ t.Fatalf("start %d: %v", n+1, err)
+ }
+ if f.ops.CurrentAttempt(f.start).ID == "" {
+ t.Fatal("no launch reservation")
+ }
+ if local == 0 {
+ if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
+ t.Fatal("idempotent reservation", err)
+ }
+ }
+ err := f.ops.Failed(t.Context(), f.start, local)
+ if n == 8 {
+ if !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("ninth failure: %v", err)
+ }
+ } else if err != nil {
+ t.Fatal(err)
+ }
+ }
+ f.ops = f.attempts()
+ if err := f.ops.Reserve(t.Context(), f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) {
+ t.Fatalf("restart lost count: %v", err)
+ }
+ })
+ }
+}
+
+func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
+ f := recoveryPAFixture(t, hostclaim.KindRestoring)
+ w := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }}
+ if w.recoveryAttemptsReady() {
+ t.Fatal("landing-only provider admitted recovery")
+ }
+ w.Operations = f.ops
+ if !w.recoveryAttemptsReady() {
+ t.Fatal("recovery provider not ready")
+ }
+ w.Operations = claimOperationRouter{hostclaim.KindRestoring: {Attempts: f.ops, ProbeReady: w.ProbeReady}}
+ if w.recoveryAttemptsReady() {
+ t.Fatal("missing revert provider admitted")
+ }
+ if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ oldKey := recoveryAttemptKey(f.start)
+ c := f.read().Claim
+ c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
+ f.putClaim(c)
+ f.start.Admitted = f.read()
+ f.rec.Move.Rollback.ID = c.Op.Intent["rollback"]
+ if recoveryAttemptKey(f.start) == oldKey {
+ t.Fatal("rollback not bound")
+ }
+ if got := f.ops.CurrentAttempt(f.start); got.ID != "" {
+ t.Fatal("reservation escaped its rollback")
+ }
+ if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if j.Steps[oldKey] != "1" || j.Steps[recoveryAttemptKey(f.start)] != "1" {
+ t.Fatal("rollback budgets combined")
+ }
+ for _, mutate := range []func(*OperationStart){
+ func(s *OperationStart) { s.Admitted.Claim.Holder.Incarnation = "other" },
+ func(s *OperationStart) { s.Admitted.Claim.Op.ID = "other" },
+ } {
+ copied := f.start
+ raw, _ := json.Marshal(f.start.Admitted.Claim)
+ var claim hostclaim.Claim
+ _ = json.Unmarshal(raw, &claim)
+ copied.Admitted.Claim = &claim
+ mutate(&copied)
+ if recoveryAttemptKey(copied) == recoveryAttemptKey(f.start) {
+ t.Fatal("budget key omitted identity")
+ }
+ }
+}
+
+func TestRecoveryProbeAttemptsRefuseChangedRollback(t *testing.T) {
+ f := recoveryPAFixture(t, hostclaim.KindRestoring)
+ c := f.read().Claim
+ c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
+ f.putClaim(c)
+ if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("changed rollback admitted: %v", err)
+ }
+}
diff --git a/internal/host/movesrecoveryprobe.go b/internal/host/movesrecoveryprobe.go
index 296c5c84f..13dccacb7 100644
--- a/internal/host/movesrecoveryprobe.go
+++ b/internal/host/movesrecoveryprobe.go
@@ -167,6 +167,12 @@ func (a recoveryProbeAdapter) probe(ctx context.Context, s release.Store, rec pr
// budget and the coverage and generation of its return.
func recoveryProbeMatches(rec provision.Record, req recoveryProbeRequest) error {
m := rec.Move
+ if m != nil && m.ID != "" && m.State == provision.MoveReverting {
+ if m.ReopenedAt != nil || req.RollbackID != "" || req.Coverage != "" || req.Generation != "" || !req.Clean || req.Attempts != recoveryProbeAttempts || req.Record == nil {
+ return errRecoveryProbeRequest
+ }
+ return nil
+ }
if m == nil || m.ID == "" || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID != req.RollbackID || m.Rollback.SourcePhase != provision.RbSourceProbing {
return fmt.Errorf("%w: not rollback %s at probing", errRecoveryProbeRequest, req.RollbackID)
}
@@ -188,9 +194,17 @@ func recoveryProbeMatches(rec provision.Record, req recoveryProbeRequest) error
// incarnation, naming the request's src_gen. Anything else waits.
func (a recoveryProbeAdapter) admits(obs hostclaim.Observed, rec provision.Record, req recoveryProbeRequest) error {
c := obs.Claim
- if obs.Absent || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRestoring || c.Op.ID != rec.Move.ID || c.Op.Intent["rollback"] != req.RollbackID ||
- c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != a.Box || c.Holder.Incarnation != rec.Move.Incarnation {
- return fmt.Errorf("%w: the claim is not rollback %s's probe on %s", hostbox.ErrChanged, req.RollbackID, a.Box)
+ if obs.Absent || c == nil || c.Holder == nil || c.Op.ID != rec.Move.ID || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != a.Box || c.Holder.Incarnation != rec.Move.Incarnation {
+ return hostbox.ErrChanged
+ }
+ if rec.Move.State == provision.MoveReverting {
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Holder.Generation != req.SourceGeneration {
+ return hostbox.ErrChanged
+ }
+ return nil
+ }
+ if c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != req.RollbackID {
+ return hostbox.ErrChanged
}
if req.SourceGeneration != "" && c.Op.Journal[hostclaim.JournalSrcGen] != req.SourceGeneration {
return fmt.Errorf("%w: the claim's src_gen is not the request's", hostbox.ErrChanged)
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go
index a7894f5ab..406b0294e 100644
--- a/internal/host/movesrevert.go
+++ b/internal/host/movesrevert.go
@@ -572,6 +572,10 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool,
return err
}}
out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req)
+ var missing *ClaimDependencyError
+ if errors.As(err, &missing) {
+ return false, nil
+ }
if err != nil {
log.Printf("moves: %s: the revert's probe of %s: %v", host, id, err)
out = recoveryProbeWaiting
@@ -612,6 +616,11 @@ func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool,
if err != nil {
return false, err
}
+ // The probe wrote its attempt count into the same journal.
+ j, err = moves.ReadJournal(cfg.DataDir, host, id)
+ if err != nil {
+ return false, err
+ }
j.Mark(rvStepProbeExhausted, string(body))
if err := j.Write(cfg.DataDir); err != nil {
return false, err
diff --git a/internal/host/plan10_j2_adapter_test.go b/internal/host/plan10_j2_adapter_test.go
index 1350c3864..159aac957 100644
--- a/internal/host/plan10_j2_adapter_test.go
+++ b/internal/host/plan10_j2_adapter_test.go
@@ -13,6 +13,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
@@ -72,8 +73,9 @@ func plan10J2AdapterClaim(t *testing.T, f *recoveryFixture, phase, srcGen string
// probe readiness to the admitted session. The held render (Materialize)
// follows the record.
func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
- for _, dirty := range []bool{false, true} {
- t.Run(fmt.Sprintf("dirty=%v", dirty), func(t *testing.T) {
+ for _, tc := range []struct{ dirty, revert bool }{{}, {dirty: true}, {revert: true}} {
+ dirty := tc.dirty
+ t.Run(fmt.Sprintf("dirty=%v/revert=%v", dirty, tc.revert), func(t *testing.T) {
f := newRecoveryFixture(t)
inc := plan10J2AdapterClaim(t, f, hostclaim.PhaseProbing, "src-gen")
c := f.controller("boot-1", "controller-1", 4101, "start-1")
@@ -82,9 +84,33 @@ func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
rec := plan10J2AdapterRecord(inc, dirty)
var events []string
+ req := plan10J2AdapterRequest(&rec, "src-gen", &events)
+ kind := hostclaim.KindRestoring
+ if tc.revert {
+ kind = hostclaim.KindReverting
+ o, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
+ c.Op = plan10J2Op(hostclaim.KindReverting, hostclaim.PhaseProbing)
+ c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec.Move.State, rec.Move.Rollback = provision.MoveReverting, nil
+ req = recoveryProbeRequest{Clean: true, SourceGeneration: o.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, Record: func(_ context.Context, in provision.MoveInputs) error {
+ events = append(events, "record")
+ return provision.MoveRecordRevertInputs(provision.MoveGuard{ID: rec.Move.ID, Incarnation: inc}, in)(&rec)
+ }}
+ }
+ recorded := func() *provision.MoveInputs {
+ if tc.revert {
+ return rec.Move.RecoveryInputs
+ }
+ return rec.Move.Rollback.RecoveryInputs
+ }
f.materialize = func(context.Context, hostclaim.Observed) error {
events = append(events, "render")
- if rec.Move.Rollback.RecoveryInputs == nil {
+ if recorded() == nil {
t.Error("rendered before the inputs were recorded")
}
return nil
@@ -93,8 +119,8 @@ func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
wake := func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
events = append(events, "wake")
policies = append(policies, p)
- if rec.Move.Rollback.RecoveryInputs == nil || p.Inputs == nil || !reflect.DeepEqual(*p.Inputs, *rec.Move.Rollback.RecoveryInputs) {
- t.Errorf("the wake does not carry the recorded inputs: %+v / %+v", p.Inputs, rec.Move.Rollback.RecoveryInputs)
+ if recorded() == nil || p.Inputs == nil || !reflect.DeepEqual(*p.Inputs, *recorded()) {
+ t.Errorf("the wake does not carry the recorded inputs: %+v / %+v", p.Inputs, recorded())
}
if err := c.wake.admitsOperation(f.read(), p, "box-1"); err != nil {
return err
@@ -130,7 +156,7 @@ func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
return plan10J2AdapterInputs(), nil
}}
- out, err := a.probe(t.Context(), f.claims, rec, plan10J2AdapterRequest(&rec, "src-gen", &events))
+ out, err := a.probe(t.Context(), f.claims, rec, req)
if err != nil || out != recoveryProbeAnswered {
t.Fatalf("probe: %v %v", out, err)
}
@@ -144,7 +170,7 @@ func TestPlan10J2AdapterRecordsInputsBeforeTheGroupDProbe(t *testing.T) {
t.Fatalf("attempt journal calls: %s", got)
}
for _, s := range ops.starts {
- if !s.Policy.Probe || s.Admitted.Claim.Op.Kind != hostclaim.KindRestoring || s.Admitted.Claim.Op.Phase != hostclaim.PhaseProbing || !reflect.DeepEqual(*s.Policy.Inputs, plan10J2AdapterInputs()) {
+ if !s.Policy.Probe || s.Admitted.Claim.Op.Kind != kind || s.Admitted.Claim.Op.Phase != hostclaim.PhaseProbing || !reflect.DeepEqual(*s.Policy.Inputs, plan10J2AdapterInputs()) {
t.Fatalf("attempt bound to %+v", s)
}
}
@@ -389,3 +415,149 @@ func TestPlan10J2ArmPhase1ConnectsRecoveryOnlyWithClaimFlag(t *testing.T) {
})
}
}
+
+func TestPlan10J2RevertRunUsesProductionAdapter(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ wakes := 0
+ a := recoveryProbeAdapter{Box: "ie-box", Claims: true,
+ Route: func(string) (*registry.Route, error) { return &h.route, nil },
+ Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
+ return revertRendered(), nil
+ },
+ Wake: func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
+ wakes++
+ m := h.move(t)
+ if m.RecoveryInputs == nil || !reflect.DeepEqual(*m.RecoveryInputs, *p.Inputs) {
+ t.Fatal("wake before revert inputs")
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
+ t.Fatal("wrong claim")
+ }
+ return nil
+ }}
+ h.cfg.recoveryProbe = a.probe
+ h.pass(t)
+ if wakes != 1 {
+ t.Fatalf("wakes=%d", wakes)
+ }
+ h.reopened(t)
+}
+
+func TestPlan10J2RevertMissingProviderWaitsWithoutInputsOrWake(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ calls := 0
+ a := recoveryProbeAdapter{Box: "ie-box", Claims: true, Ready: func() bool { return false }, Route: func(string) (*registry.Route, error) { return &h.route, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
+ calls++
+ return revertRendered(), nil
+ }, Wake: func(context.Context, registry.Route, activator.StartPolicy) error { calls++; return nil }}
+ h.cfg.recoveryProbe = a.probe
+ h.pass(t)
+ before := h.move(t).UpdatedAt
+ h.clock.Advance(16 * time.Minute)
+ h.pass(t)
+ m := h.move(t)
+ if calls != 0 || m.RecoveryInputs != nil || m.State != provision.MoveReverting || !m.UpdatedAt.Equal(before) {
+ t.Fatalf("unbound provider changed recovery: calls=%d move=%+v", calls, m)
+ }
+}
+
+func TestPlan10J2NinthFailedStartSticksAndKeepsJournal(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ starts, ids := 0, 0
+ var last OperationStart
+ provider := func() *movesProbeAttempts {
+ return &movesProbeAttempts{Claims: h.s, DataDir: h.cfg.DataDir,
+ Records: func(ctx context.Context, account, app, host string) (*provision.Record, error) {
+ return provision.GetRecord(ctx, h.s, account, app, host)
+ },
+ Admitted: func(context.Context, *provision.Record) (int, error) { return 9, nil },
+ NewID: func() string { ids++; return fmt.Sprintf("attempt-%d", ids) }, Now: time.Now}
+ }
+ a := recoveryProbeAdapter{Box: "ie-box", Claims: true, Ready: func() bool { return provider().RecoveryProbeAttemptsReady() }, Route: func(string) (*registry.Route, error) { return &h.route, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
+ return revertRendered(), nil
+ },
+ Wake: func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error {
+ o, err := hostclaim.Read(ctx, h.s, rt.Host)
+ if err != nil {
+ return err
+ }
+ holder := o.Claim.Holder
+ last = OperationStart{Admitted: o, Policy: p, Owner: hostguard.FenceTarget{Host: rt.Host, Incarnation: holder.Incarnation, Stamp: holder.Stamp, Session: holder.Session, Box: "ie-box"}}
+ ops := provider()
+ for i := 0; i < 3; i++ {
+ if err := ops.Reserve(ctx, last, starts); err != nil {
+ return err
+ }
+ starts++
+ if err := ops.Failed(ctx, last, starts-1); err != nil {
+ return err
+ }
+ }
+ return moves.ErrAttemptOver
+ }}
+ h.cfg.recoveryProbe = a.probe
+ for i := 0; i < 3; i++ {
+ h.pass(t)
+ }
+ m := h.move(t)
+ if starts != 9 || m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts {
+ t.Fatalf("starts=%d move=%+v", starts, m)
+ }
+ j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if count, err := recoveryAttemptCount(j, last); err != nil || count != 9 {
+ t.Fatalf("journal count=%d err=%v", count, err)
+ }
+}
+
+func TestPlan10J2RevertAdapterRejectsChangedBinding(t *testing.T) {
+ for _, tc := range []string{"reopened", "generation", "rev_phase", "kind", "executor"} {
+ t.Run(tc, func(t *testing.T) {
+ f := newRecoveryFixture(t)
+ o, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
+ c.Op = plan10J2Op(hostclaim.KindReverting, hostclaim.PhaseProbing)
+ c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
+ switch tc {
+ case "rev_phase":
+ c.Op.Journal[hostclaim.JournalRevPhase] = hostclaim.PhaseProbed
+ case "kind":
+ c.Op.Kind = hostclaim.KindRestoring
+ case "executor":
+ c.Op.Exec = "box-2"
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := provision.Record{Host: recoveryHost, Move: &provision.MoveState{ID: o.Claim.Op.ID, Incarnation: o.Claim.Holder.Incarnation, State: provision.MoveReverting}}
+ req := recoveryProbeRequest{Clean: true, SourceGeneration: o.Claim.Holder.Generation, Attempts: recoveryProbeAttempts, Record: func(context.Context, provision.MoveInputs) error { t.Fatal("recorded invalid revert"); return nil }}
+ if tc == "generation" {
+ req.SourceGeneration = "other"
+ }
+ if tc == "reopened" {
+ now := time.Now()
+ rec.Move.ReopenedAt = &now
+ }
+ a := recoveryProbeAdapter{Claims: true, Box: "box-1", Route: func(string) (*registry.Route, error) { t.Fatal("looked up invalid revert"); return nil, nil }, Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
+ t.Fatal("selected invalid revert")
+ return provision.MoveInputs{}, nil
+ }, Wake: func(context.Context, registry.Route, activator.StartPolicy) error {
+ t.Fatal("woke invalid revert")
+ return nil
+ }}
+ if _, err := a.probe(t.Context(), f.claims, rec, req); err == nil {
+ t.Fatal("invalid revert accepted")
+ }
+ })
+ }
+}
diff --git a/internal/hostclaim/lossrecord.go b/internal/hostclaim/lossrecord.go
index fce0867be..3e496aa6b 100644
--- a/internal/hostclaim/lossrecord.go
+++ b/internal/hostclaim/lossrecord.go
@@ -338,7 +338,7 @@ func CreateLossRecord(ctx context.Context, s release.Store, sg release.Signer, r
// proving that the accepted evidence and consequences prepared earlier are
// byte-for-byte unchanged.
func ReplaceLossRecord(ctx context.Context, s release.Store, sg release.Signer, before ObservedLossRecord, next LossRecord) (ObservedLossRecord, error) {
- if before.ETag == "" || before.Record.Validate() != nil || next.Validate() != nil || !immutableLossFieldsEqual(before.Record, next) || !validLossProgression(before.Record, next) {
+ if err := ValidateLossRecordReplacement(before, next); err != nil {
return ObservedLossRecord{}, ErrLossRecord
}
key, _ := LossRecordKey(next.ID)
@@ -384,3 +384,12 @@ func validLossProgression(before, next LossRecord) bool {
// IsDestroyedLocalOnly reports the complete attested consequence shape, not
// merely its intent marker. Owners and durable records use the same predicate.
func IsDestroyedLocalOnly(c LossConsequences) bool { return destroyedLocalOnly(c) }
+
+// ValidateLossRecordReplacement also guards scoped transports that receive
+// signed bytes directly, without passing through ReplaceLossRecord.
+func ValidateLossRecordReplacement(before ObservedLossRecord, next LossRecord) error {
+ if before.ETag == "" || before.Record.Validate() != nil || next.Validate() != nil || !immutableLossFieldsEqual(before.Record, next) || !validLossProgression(before.Record, next) {
+ return ErrLossRecord
+ }
+ return nil
+}
diff --git a/internal/hostops/move_loss_grant.go b/internal/hostops/move_loss_grant.go
index 03abad6ed..4b7eaee01 100644
--- a/internal/hostops/move_loss_grant.go
+++ b/internal/hostops/move_loss_grant.go
@@ -5,6 +5,7 @@ import (
"context"
"crypto/ed25519"
"encoding/json"
+ "errors"
"reflect"
"slices"
"time"
@@ -36,7 +37,7 @@ import (
// lossBinding is one declaration's scope.
type lossBinding struct {
- lossID, host, opID, kind, box, claimETag, incarnation, account, app string
+ lossID, host, opID, kind, box, claimETag, incarnation, account, app, bucket string
}
// moveLossService is the service a typed method runs, with the moves owner
@@ -133,7 +134,7 @@ func (w *ControlWriter) DeclareMoveLoss(ctx context.Context, signer release.Sign
// The operation has moved on: only an accepted declaration replays.
return hostclaim.Observed{}, hostresolve.ErrResolve
}
- b := lossBinding{lossID: c.ID, host: host, opID: c.Op.ID, kind: c.Op.Kind, box: c.Box, claimETag: c.ClaimETag, incarnation: c.HolderIncarnation, account: account, app: app}
+ b := lossBinding{bucket: bucket, lossID: c.ID, host: host, opID: c.Op.ID, kind: c.Op.Kind, box: c.Box, claimETag: c.ClaimETag, incarnation: c.HolderIncarnation, account: account, app: app}
scoped := &lossScopedStore{raw: w.raw, pub: w.pub, now: w.now, b: b}
if _, err := moveLossService(scoped, w.pub, signer, bucket, w.now).Declare(ctx, actor, host, in); err != nil {
return hostclaim.Observed{}, err
@@ -191,7 +192,7 @@ func (s *lossScopedStore) Put(ctx context.Context, key string, body []byte, etag
case key == "control/host-loss-acceptances/"+s.b.lossID+".json":
err = s.acceptance(ctx, body, etag)
case key == lossKey:
- err = s.lossRecord(body)
+ err = s.lossRecord(ctx, body, etag)
case key == hostactors.Key:
err = s.quarantine(ctx, body, etag)
case key == claimKey:
@@ -247,18 +248,37 @@ func (s *lossScopedStore) acceptance(ctx context.Context, body []byte, etag stri
RosterETag string `json:"roster_etag"`
Loss hostclaim.LossRecord `json:"loss"`
}
- if release.VerifyRootEnvelope(s.pub, body, &a) != nil || a.V != 1 || a.RosterETag == "" || a.Loss.State != "prepared" || !s.bound(a.Loss) {
+ if release.VerifyRootEnvelope(s.pub, body, &a) != nil || a.V != 1 || s.b.bucket == "" || a.Bucket != s.b.bucket || a.RosterETag == "" || a.Loss.State != "prepared" || !s.bound(a.Loss) {
return errMoveLossScope
}
_, err := s.registered(ctx)
return err
}
-func (s *lossScopedStore) lossRecord(body []byte) error {
+func (s *lossScopedStore) lossRecord(ctx context.Context, body []byte, etag string) error {
var r hostclaim.LossRecord
if release.VerifyRootEnvelope(s.pub, body, &r) != nil || !s.bound(r) {
return errMoveLossScope
}
+ if _, err := s.registered(ctx); err != nil {
+ return err
+ }
+ before, err := hostclaim.ReadLossRecord(ctx, s.raw, s.pub, s.b.lossID)
+ if errors.Is(err, hostclaim.ErrLossRecordAbsent) {
+ if etag != "" || r.State != "prepared" {
+ return errMoveLossScope
+ }
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ if before.ETag != etag {
+ return release.ErrCASConflict
+ }
+ if hostclaim.ValidateLossRecordReplacement(before, r) != nil {
+ return errMoveLossScope
+ }
return nil
}
diff --git a/internal/hostops/move_loss_scope_fix_test.go b/internal/hostops/move_loss_scope_fix_test.go
new file mode 100644
index 000000000..66922f47f
--- /dev/null
+++ b/internal/hostops/move_loss_scope_fix_test.go
@@ -0,0 +1,181 @@
+package hostops
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type lossScopeCapture struct {
+ release.Store
+ w *lossWorld
+ state, key, id string
+ body []byte
+ etag string
+ ctx context.Context
+}
+
+func (s *lossScopeCapture) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if key == s.key {
+ loss, err := hostclaim.ReadLossRecord(ctx, s.Store, s.w.pub, s.id)
+ if err == nil && loss.Record.State == s.state {
+ s.ctx = ctx
+ s.body = append([]byte(nil), body...)
+ s.etag = etag
+ return "", errors.New("captured permitted write")
+ }
+ }
+ return s.Store.Put(ctx, key, body, etag)
+}
+
+func scopedLossFixture(t *testing.T, state, kind string) (*lossWorld, *lossScopedStore, *lossScopeCapture) {
+ t.Helper()
+ w := newLossWorld(t, "landing", nil)
+ sub := w.submission(w.inspect(t, "uk-box"))
+ o := w.claim(t)
+ key := w.key
+ if kind == "claim" {
+ key = w.claimK
+ }
+ capture := &lossScopeCapture{Store: w.raw, w: w, state: state, key: key, id: sub.Confirmation}
+ cw, err := NewControlWriter(capture, w.pub, func() time.Time { return w.clock })
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = cw.DeclareMoveLoss(t.Context(), w.signer, lossBucket, "operator", "bab", "app", lossHost, "uk-box", sub)
+ if err == nil || capture.body == nil {
+ t.Fatalf("write not captured: %v", err)
+ }
+ loss, err := hostclaim.ReadLossRecord(t.Context(), w.raw, w.pub, sub.Confirmation)
+ if err != nil || loss.Record.State != state {
+ t.Fatalf("signed declaration: %+v %v", loss, err)
+ }
+ s := &lossScopedStore{raw: w.raw, pub: w.pub, now: func() time.Time { return w.clock }, b: lossBinding{bucket: lossBucket, lossID: sub.Confirmation, host: lossHost, opID: lossOp, kind: hostclaim.KindLanding, box: "uk-box", claimETag: o.ETag, incarnation: w.inc, account: "bab", app: "app"}}
+ return w, s, capture
+}
+
+func TestJ4ScopedWritesComparePreparedAndCommittedTransitions(t *testing.T) {
+ for _, state := range []string{"prepared", "committed"} {
+ for _, kind := range []string{"claim", "record"} {
+ t.Run(state+"/"+kind, func(t *testing.T) {
+ _, s, capture := scopedLossFixture(t, state, kind)
+ if _, err := s.Put(capture.ctx, capture.key, capture.body, capture.etag); err != nil {
+ t.Fatalf("permitted write refused: %v", err)
+ }
+ for _, field := range []string{"identity", "transition", "sequence"} {
+ t.Run(field, func(t *testing.T) {
+ w, s, capture := scopedLossFixture(t, state, kind)
+ var altered map[string]any
+ dec := json.NewDecoder(strings.NewReader(string(capture.body)))
+ dec.UseNumber()
+ if err := dec.Decode(&altered); err != nil {
+ t.Fatal(err)
+ }
+ if kind == "claim" {
+ switch field {
+ case "identity":
+ altered["holder"].(map[string]any)["account"] = "other"
+ case "transition":
+ altered["op"].(map[string]any)["exec"] = "foreign"
+ case "sequence":
+ altered["gen"] = 99999
+ }
+ } else {
+ switch field {
+ case "identity":
+ altered["account"] = "other"
+ case "transition":
+ altered["move"].(map[string]any)["reason"] = "other"
+ case "sequence":
+ altered["record_seq"] = 99999
+ }
+ }
+ body, err := json.Marshal(altered)
+ if err != nil {
+ t.Fatal(err)
+ }
+ before := lossSnapshot(t, w)
+ if _, err = s.Put(capture.ctx, capture.key, body, capture.etag); !errors.Is(err, errMoveLossScope) {
+ t.Fatalf("expected scope comparison refusal, got %v", err)
+ }
+ if lossSnapshot(t, w) != before {
+ t.Fatal("altered write landed")
+ }
+ if _, err = s.Put(capture.ctx, capture.key, capture.body, capture.etag); err != nil {
+ t.Fatalf("equivalent valid write refused after negative: %v", err)
+ }
+ })
+ }
+ })
+ }
+ }
+}
+
+func TestJ4ScopedLossRecordRejectsRegressionAndEvidenceChanges(t *testing.T) {
+ for _, change := range []string{"regress", "evidence", "registration"} {
+ t.Run(change, func(t *testing.T) {
+ w, s, capture := scopedLossFixture(t, "committed", "claim")
+ loss, err := hostclaim.ReadLossRecord(t.Context(), w.raw, w.pub, capture.id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next := loss.Record
+ switch change {
+ case "regress":
+ next.State = "prepared"
+ next.ResultClaimGen = 0
+ case "evidence":
+ next.Actor = "another-operator"
+ case "registration":
+ // Removing the app admission ends the scoped transport's authority.
+ key, _ := AdmissionKey("bab", "app")
+ if err := w.raw.Delete(t.Context(), key); err != nil {
+ t.Fatal(err)
+ }
+ }
+ body, err := release.SignRootEnvelope(w.signer, next)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key, _ := hostclaim.LossRecordKey(capture.id)
+ if _, err = s.Put(t.Context(), key, body, loss.ETag); err == nil {
+ t.Fatal("invalid progression forwarded")
+ }
+ })
+ }
+}
+
+func TestJ4ScopedAcceptanceRejectsOtherBucket(t *testing.T) {
+ w, s, capture := scopedLossFixture(t, "prepared", "record")
+ key := "control/host-loss-acceptances/" + capture.id + ".json"
+ raw, _, err := w.raw.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var a struct {
+ V int `json:"v"`
+ Bucket string `json:"bucket"`
+ RosterETag string `json:"roster_etag"`
+ Loss hostclaim.LossRecord `json:"loss"`
+ }
+ if err := release.VerifyRootEnvelope(w.pub, raw, &a); err != nil {
+ t.Fatal(err)
+ }
+ a.Bucket = "foreign"
+ body, err := release.SignRootEnvelope(w.signer, a)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = s.acceptance(t.Context(), body, ""); !errors.Is(err, errMoveLossScope) {
+ t.Fatalf("other bucket: %v", err)
+ }
+ if err = s.acceptance(t.Context(), raw, ""); err != nil {
+ t.Fatalf("correct bucket refused: %v", err)
+ }
+}
diff --git a/internal/hostops/move_loss_test.go b/internal/hostops/move_loss_test.go
index 7787f45e2..dade6e282 100644
--- a/internal/hostops/move_loss_test.go
+++ b/internal/hostops/move_loss_test.go
@@ -788,17 +788,6 @@ func TestJ4ScopedStoreRefusesEverythingElse(t *testing.T) {
if err := s.Delete(ctx, w.key); !errors.Is(err, ErrContract) {
t.Fatalf("delete = %v", err)
}
- // A claim write that is not the declaration's own transition.
- if _, err := hostclaim.Update(ctx, s, o, func(c *hostclaim.Claim) error { c.Op.Exec = "none"; return nil }); err == nil {
- t.Fatal("arbitrary claim transition accepted")
- }
- // A record write that is not the declaration's own transition.
- if _, err := provision.SetRecordMove(provision.WithRecordSequence(ctx), s, "bab", "app", lossHost, func(r *provision.Record) error {
- r.Move.Reason = provision.RollbackDeclaredLost
- return nil
- }); err == nil {
- t.Fatal("arbitrary record transition accepted")
- }
// A loss record or acceptance for another declaration id.
if _, err := s.Put(ctx, "control/host-loss/loss_2.json", []byte(`{}`), ""); !errors.Is(err, ErrContract) {
t.Fatalf("other loss = %v", err)
diff --git a/internal/hostops/move_recovery_grant.go b/internal/hostops/move_recovery_grant.go
index 6ff97d682..c8d4ca78b 100644
--- a/internal/hostops/move_recovery_grant.go
+++ b/internal/hostops/move_recovery_grant.go
@@ -362,18 +362,15 @@ func (g *MoveRecoveryGrant) Claim(ctx context.Context, action MoveRecoveryClaimA
if m != nil || rec.LastMove == nil || rec.LastMove.ID != g.o.OpID {
return ErrPending
}
- switch rec.LastMove.Outcome {
- case "done", "rolled_back", "aborted":
+ switch {
+ case rec.LastMove.Outcome == "done" && c.Op.Kind == hostclaim.KindLanding && c.Op.Phase == hostclaim.PhaseCommitted:
+ transition = hostclaim.ReleaseMove(g.o.OpID, c.Op.Intent["target"])
+ case rec.LastMove.Outcome == "rolled_back" && (c.Op.Kind == hostclaim.KindRestoring || c.Op.Kind == hostclaim.KindReverting) && c.Op.Phase == hostclaim.PhaseOpened:
+ transition = hostclaim.ReleaseMove(g.o.OpID, c.Op.Intent["source"])
+ case rec.LastMove.Outcome == "aborted" && c.Op.Kind == hostclaim.KindMoving && c.Op.Phase == hostclaim.PhaseAborting:
+ transition = hostclaim.ReleaseAbort(g.o.OpID, c.Op.Intent["source"])
default:
- return ErrIdentity
- }
- if c.Op.Kind == hostclaim.KindMoving {
- if c.Op.Phase != hostclaim.PhaseAborting {
- return ErrPending
- }
- transition = hostclaim.ReleaseAbort(g.o.OpID, c.Op.Exec)
- } else {
- transition = hostclaim.ReleaseMove(g.o.OpID, c.Op.Exec)
+ return ErrPending
}
} else {
if m == nil || m.ID != g.o.OpID || m.Incarnation != g.o.Incarnation {
diff --git a/internal/hostops/move_recovery_grant_test.go b/internal/hostops/move_recovery_grant_test.go
index e9174f9af..470289fc4 100644
--- a/internal/hostops/move_recovery_grant_test.go
+++ b/internal/hostops/move_recovery_grant_test.go
@@ -3,6 +3,7 @@ package hostops
import (
"context"
"errors"
+ "fmt"
"testing"
"time"
@@ -398,7 +399,7 @@ func TestMoveRecoveryPassUsesTypedHookOnlyInsideRunRecovery(t *testing.T) {
}
missing := r
missing.Moves = nil
- if err := f.w.RunRecovery(t.Context(), missing.Pass); !errors.Is(err, ErrContract) {
+ if err := f.w.RunRecovery(t.Context(), missing.Pass); err != nil {
t.Fatalf("missing hook: %v", err)
}
var saved context.Context
@@ -461,3 +462,40 @@ func TestMoveRecoveryAbortCloseFreshGrantThenRelease(t *testing.T) {
t.Fatal(err)
}
}
+
+func TestMoveRecoveryReleaseRequiresMatchingOutcomeAndExecutor(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindLanding, hostclaim.KindRestoring, hostclaim.KindReverting, hostclaim.KindMoving} {
+ for _, outcome := range []string{"done", "rolled_back", "aborted"} {
+ for _, foreign := range []bool{false, true} {
+ t.Run(kind+"/"+outcome+fmt.Sprint(foreign), func(t *testing.T) {
+ valid := !foreign && ((kind == hostclaim.KindLanding && outcome == "done") || ((kind == hostclaim.KindRestoring || kind == hostclaim.KindReverting) && outcome == "rolled_back") || (kind == hostclaim.KindMoving && outcome == "aborted"))
+ w := newC7World(t, func(c *hostclaim.Claim, r *provision.Record, _ *AppAdmission) {
+ c.Op.Kind, c.Op.Phase, c.Op.Exec = kind, hostclaim.PhaseOpened, c.Op.Intent["source"]
+ if kind == hostclaim.KindLanding {
+ c.Op.Phase, c.Op.Exec = hostclaim.PhaseCommitted, c.Op.Intent["target"]
+ }
+ if kind == hostclaim.KindMoving {
+ c.Op.Phase = hostclaim.PhaseAborting
+ }
+ if foreign {
+ c.Op.Exec = "foreign"
+ }
+ c.Obligations = nil
+ r.LastMove = &provision.MoveOutcome{ID: r.Move.ID, Outcome: outcome}
+ r.Move = nil
+ })
+ before := w.claim(t)
+ err := w.control.RunRecovery(t.Context(), func(ctx context.Context) error {
+ return runMoveRecovery(ctx, before, func(ctx context.Context, g *MoveRecoveryGrant) error { return g.Claim(ctx, RecoveryMoveRelease) })
+ })
+ if valid && err != nil {
+ t.Fatal(err)
+ }
+ if !valid && (err == nil || w.claim(t).ETag != before.ETag) {
+ t.Fatalf("invalid release: %v", err)
+ }
+ })
+ }
+ }
+ }
+}
diff --git a/internal/hostops/recovery.go b/internal/hostops/recovery.go
index 43d651505..8b0d0f9d4 100644
--- a/internal/hostops/recovery.go
+++ b/internal/hostops/recovery.go
@@ -76,6 +76,9 @@ func (r Recovery) Pass(ctx context.Context) error {
// Zero time is unknown age, never old enough. Freeing empty parking is safe
// independently of age; this includes Plan 4's holderless sweep retries.
if moveRecoveryKind(op.Kind) {
+ if r.Moves == nil {
+ continue
+ }
err = runMoveRecovery(ctx, obs, r.Moves, r.MoveFenceAuthority)
} else if op.Kind == hostclaim.KindParking && len(obs.Claim.Obligations) == 0 {
err = c.Free(ctx, host, op.ID)
diff --git a/internal/park/move_sweep_test.go b/internal/park/move_sweep_test.go
index 39825933f..c16d68463 100644
--- a/internal/park/move_sweep_test.go
+++ b/internal/park/move_sweep_test.go
@@ -78,7 +78,7 @@ func TestPlan10MovesSweepRequiresPriorRetirement(t *testing.T) {
t.Fatal(readErr)
}
if tc != "valid" && tc != "mutation" {
- if err == nil || fresh.ETag != obs.ETag {
+ if fresh.ETag != obs.ETag {
t.Fatalf("unsafe proof queued: err=%v claim=%+v", err, fresh.Claim)
}
return
@@ -160,3 +160,48 @@ func TestPlan10MovesSweepPreservesInstanceEvidence(t *testing.T) {
t.Fatalf("evidence mixed: %+v", scopes)
}
}
+
+func TestPlan10MovesSweepUnavailableProofKeepsOtherCandidates(t *testing.T) {
+ for _, unavailable := range []string{"nil", "error", "invalid"} {
+ t.Run(unavailable, func(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ o, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, o, func(c *hostclaim.Claim) error { c.Obligations = nil; return hostclaim.FreeAfterPark("op")(c) })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.b.Put("data", "moves/"+f.w.Host+"/retired/staging", []byte("staged"))
+ if unavailable != "nil" {
+ f.e.VerifyMovesSweep = func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (MoveSweepProof, error) {
+ if unavailable == "error" {
+ return MoveSweepProof{}, ErrPending
+ }
+ return MoveSweepProof{}, nil
+ }
+ }
+ d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.e.Now(), f.ob.Park.RequestLimit)
+ if !errors.Is(err, ErrPending) {
+ t.Fatalf("unrelated candidates not queued: %v", err)
+ }
+ if len(d.Candidates) == 0 {
+ t.Fatal("all candidates dropped")
+ }
+ fresh, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(fresh.Claim.Obligations) == 0 {
+ t.Fatal("nothing queued")
+ }
+ for _, ob := range fresh.Claim.Obligations {
+ if ob.Park.Location == "moves/"+f.w.Host {
+ t.Fatal("moves residue queued without proof")
+ }
+ }
+ })
+ }
+}
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index e231dc165..82097b102 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -317,25 +317,27 @@ func (e Engine) sweep(ctx context.Context, expected hostclaim.Observed, expected
return d, err
}
var movesEvidence time.Time
+ candidates := d.Candidates[:0]
for _, candidate := range d.Candidates {
- if candidate.Location != "moves/"+host {
- continue
- }
- if e.VerifyMovesSweep == nil || listedAt.Before(listingStarted) {
- return d, ErrUnsafe
- }
- detached, copyErr := moveSweepObservation(obs)
- if copyErr != nil {
- return d, copyErr
- }
- proof, verifyErr := e.VerifyMovesSweep(ctx, detached, append([]lease.ObjectVersion(nil), vs...))
- if verifyErr != nil {
- return d, verifyErr
- }
- if err := proof.Check(obs, candidate.Incarnation, listingStarted); err != nil {
- return d, err
+ if candidate.Location == "moves/"+host {
+ if e.VerifyMovesSweep == nil || listedAt.Before(listingStarted) {
+ continue
+ }
+ detached, copyErr := moveSweepObservation(obs)
+ if copyErr != nil {
+ return d, copyErr
+ }
+ proof, verifyErr := e.VerifyMovesSweep(ctx, detached, append([]lease.ObjectVersion(nil), vs...))
+ if verifyErr != nil || proof.Check(obs, candidate.Incarnation, listingStarted) != nil {
+ continue
+ }
+ movesEvidence = listedAt
}
- movesEvidence = listedAt
+ candidates = append(candidates, candidate)
+ }
+ d.Candidates = candidates
+ if len(d.Candidates) == 0 {
+ return d, nil
}
var op string
if obs.Claim.Holder != nil {
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index 6a20d41f4..ecd3621dd 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -478,10 +478,10 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostops/console_move_grant.go", "ControlWriter.ConsoleMove", "record-writer", "key:WriteRecord", "21e7a6ddc4dbe5b3bf06e65b39a228e2eb588af1edc43beb09d91dd72570dd44"},
{"internal/hostops/console_move_grant.go", "ControlWriter.consoleMoveClaim", "key-site", "key:RecordKey", "265291bd356de061663d89e5ebe33b74ea32c086f8238736d6f971a5afcab722"},
{"internal/hostops/console_move_grant.go", "ObserveConsoleMove", "key-site", "key:RecordKey", "296077e790e210e9f15eabdfb30f4ba09acc46825aee27824ca333bd470664a4"},
- {"internal/hostops/move_loss_grant.go", "ControlWriter.DeclareMoveLoss", "key-site", "key:LossRecordKey", "4f58b1c06f4a5c425c872131a2808241d779471c86d0aaa6ea91977c5aac7245"},
+ {"internal/hostops/move_loss_grant.go", "ControlWriter.DeclareMoveLoss", "key-site", "key:LossRecordKey", "d70728d98081f746e574939d2e1757934e95b80da237a6e7a102aae409e4abec"},
{"internal/hostops/move_loss_grant.go", "ControlWriter.moveLossClaim", "key-site", "key:RecordKey", "8f7b71bc353330e7b031befa3e04f90bfb712d9cd7192595aa69c1fcd84a45d2"},
{"internal/hostops/move_loss_grant.go", "claimSimulation.Put", "nonrecord", "def:Put", "08f341b2c797aa30cda0e4e56249f2479f1ad032ee086442ba543f84e601c118"},
- {"internal/hostops/move_loss_grant.go", "lossScopedStore.Put", "record-transport", "def:Put,key:IsRecordKey,key:LossRecordKey,key:putSequencedRecord,sel:Put", "88d231d1f91bf0ff0bd4ea5992e5099b0cfcd942fbd1b79cf81e828133fbd64f"},
+ {"internal/hostops/move_loss_grant.go", "lossScopedStore.Put", "record-transport", "def:Put,key:IsRecordKey,key:LossRecordKey,key:putSequencedRecord,sel:Put", "f6f912bbc4c6c67317cfbb012ea4e87ed2fbbdababa91fb640c89c433738d3d5"},
{"internal/hostops/move_loss_grant.go", "readOnlyStore.Put", "nonrecord", "def:Put", "e5f2d05226614d6a23afe06dea4fda7d45958aeac6ee5f42474872b2cbbc8b8e"},
{"internal/hostops/move_loss_owner.go", "claimLossIntent", "key-site", "key:RecordKey", "c3eeb176db52e6642bd026bbf97b2b5f6520d08e7fccbdc8c8a9106851725802"},
{"internal/hostops/move_loss_owner.go", "decodeMoveLossIntent", "key-site", "key:RecordKey", "e868448e696de4614a362f27301e6fa25506394fd7582d3076d126a5f72e93db"},
@@ -536,7 +536,7 @@ var plan9B8CensusInventory = []censusEntry{
// Marks recovery stuck through SetRecordMove; event and serial writes are not record objects.
{"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "4ea620140bbc75b127a893926eefcf58685b40c8362f86ba94f22931a05b449a"},
// Records revert inputs through SetRecordMove; stuck uses the reviewed record helper and journals are local.
- {"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "67f15935f08fe9111b713641724093661e8c9dd2ee421287a1fc6cbe0509faf4"},
+ {"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "55e249b195b2325917763cdc87cfcd44563d95095918d696f14fa5cec1859898"},
// Applies revert record mutations only through SetRecordMove and its sequenced writer.
{"internal/host/movesrevert.go", "revertRun.record", "record-writer", "key:SetRecordMove", "843bea75b1a185cb4294e4e1afc65f44182767b489c6e604b96f829ae59909b0"},
// Starts rollback through SetRecordMove; the companion claim update targets the separate claim namespace.
@@ -691,7 +691,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostclaim/lossrecord.go", "LossRecordKey", "key-site", "key:LossRecordKey", "1b109b49a8fcdb09f01dbf908a173394072f01afb0c9e324023fb9b3f71f98b3"},
{"internal/hostclaim/lossrecord.go", "ReadLossRecord", "key-site", "key:LossRecordKey", "1237dd3fa3e79a81e6c35f371bf545c9a04841a6297c078d96852daf25b7b483"},
{"internal/hostclaim/lossrecord.go", "CreateLossRecord", "foreign-recordkey", "key:LossRecordKey,sel:Put", "e62263731fa572ed6f195a267abb63ff5a1bf44284db3db76a819ccaec238191"},
- {"internal/hostclaim/lossrecord.go", "ReplaceLossRecord", "foreign-recordkey", "key:LossRecordKey,sel:Put", "a863b85023ce2f654fcce618be11f8b21b2bbec4c7f5eaf19348e71f48cbab3e"},
+ {"internal/hostclaim/lossrecord.go", "ReplaceLossRecord", "foreign-recordkey", "key:LossRecordKey,sel:Put", "576589457e4a18b64e6b56892603051982bd01770c355504305e72d53149bc52"},
{"internal/hostclaim/store.go", "Update", "nonrecord", "sel:Put", "68f97a49f9a936228cb26e49d7bed3a4ab17c3a4c4255e32c2d2af1dfccd73a6"},
{"internal/hostevidence/signature.go", "immutable", "nonrecord", "sel:Put", "54fb2bdc230c2cb4193f3b808004b63bcfd092266846fc0c03e61617c3ad0ab5"},
{"internal/hostexec/controller.go", "Controller.run", "nonrecord", "sel:Put", "03df11b375761730700ce30c0db05c57f0828629eda0a5de126ea1a2f2f7501c"},
diff --git a/testdata/claim-moves-copy.md b/testdata/claim-moves-copy.md
index abe62e6b4..c2a550a6d 100644
--- a/testdata/claim-moves-copy.md
+++ b/testdata/claim-moves-copy.md
@@ -1,5 +1,7 @@
-# Claim Plan 5 Moves refusal (approved by Paul, 2026-10-08)
+# Claim moves copy record (HC5-N1 approved by Paul, 2026-10-08; HC5-N2 and HC5-N3 come from claim Plan 10, pending his review)
| id | file | text | status |
|---|---|---|---|
| HC5-N1 | internal/console/claim_retire_helpers.go | instance moves are not available on this console yet: they need the host-claim Moves protocol | approved |
+| HC5-N2 | internal/console/apiinstancemove.go | invalid resolution request | pending |
+| HC5-N3 | internal/console/apiinstancemove.go | could not read the instance | pending |