amadan
host: abort parks staging from the target region and waits for the obligation (§7.11, joint N13)
The source's aborting arm takes the staging lease, journals abort_bound once (now + SignatureLife + DefaultSkew, or a later seal bound), CASes hostclaim.Abort with moves.AbortScope (moving/aborting, the lane's park-replica owned by the target region), then retires its own state: snapshots by version through DeleteVersion, receipts, ledger, journal last. It releases with the new hostclaim.ReleaseAbort only once no obligation remains, and clears the record with MoveGuardedAbortComplete. The source never deletes from the target bucket. The target region's box reaches an aborting record through the participant rule (aborting joins movesParticipantStates and the target-without-row exception) and parks the lane with laneParkTarget, Plan 4's revertLanePark generalised by claim kind and phases. Unarmed in production until Task 9, as the revert's. ReleaseMove no longer releases moving/aborting (ruling K7); the legacy MoveAbortComplete refuses a per-op move. moves.AbortAfter (6h) and AbortDue measure from max(RequestedAt, ReadyAt) for a preparing move (K18). An aborting move is no longer live for the target's report caches. Census: interimAbort's row replaced by abortSource's. Spec note for K19. Rulings K5, K6, K7, K15, K17, K18, K19, K21.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
18 files changed,
+1243
−196
approved_moves_copy_test.go+2 −0docs/superpowers/specs/2026-10-02-move-any-app-design.md+5 −1internal/host/movesabort.go+283 −0internal/host/movesabort_test.go+487 −0internal/host/movesfinal.go+13 −9internal/host/movespass.go+6 −0internal/host/movesphase1.go+3 −84internal/host/movesphase1_test.go+0 −61internal/host/movesreport.go+7 −1internal/host/movesrevert.go+28 −16internal/hostclaim/move.go+93 −14internal/hostclaim/move_abort_test.go+188 −0internal/hostclaim/move_test.go+15 −7internal/moves/abort.go+39 −0internal/moves/abort_test.go+46 −0internal/provision/move_test.go+19 −0internal/provision/plan9_b8_census_test.go+1 −1internal/provision/provision.go+8 −2
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index 321ed1234..11de97d30 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -571,6 +571,7 @@ var movesScanFiles = []string{ |
| "internal/host/movesfloor.go", |
| "internal/host/movesfloor_linux.go", |
| "internal/host/movesfloor_other.go", |
| + "internal/host/movesabort.go", |
| "internal/host/movesarchive.go", |
| "internal/host/movescut.go", |
| "internal/host/movesfinal.go", |
| @@ -611,6 +612,7 @@ var movesScanFiles = []string{ |
| "internal/lease/mirror.go", |
| "internal/lease/staging.go", |
| "internal/lease/versioned.go", |
| + "internal/moves/abort.go", |
| "internal/moves/archive/archive.go", |
| "internal/moves/archive/extract.go", |
| "internal/moves/archive/limits.go", |
diff --git a/docs/superpowers/specs/2026-10-02-move-any-app-design.md b/docs/superpowers/specs/2026-10-02-move-any-app-design.md| index 26471345f..a767b1266 100644 |
| --- a/docs/superpowers/specs/2026-10-02-move-any-app-design.md |
| +++ b/docs/superpowers/specs/2026-10-02-move-any-app-design.md |
| @@ -3063,7 +3063,11 @@ an unreachable target never needs a human. |
| Delete and prefix-scoped List on `moves/*` only), so it cannot park target |
| staging. Abort therefore adds a `park-staging` obligation **owned by the |
| target region's box**, not before the staging horizon of §7.3. The source |
| -parks its own snapshots. The claim stays `moving` until every obligation |
| +parks its own snapshots. (Plan 5 note, ruling K19: "parks" here means the |
| +source retires them from its own bucket, every version of |
| +`move-snapshots/<host>/<op>/` through `DeleteVersion`; that prefix is outside |
| +the park scope grammar, and F3's retention of snapshots covers completed |
| +moves only.) The claim stays `moving` until every obligation |
| has acknowledged, then releases to `serving`. The model gives the source |
| the target's park and records an IAM violation on every abort. |
diff --git a/internal/host/movesabort.go b/internal/host/movesabort.go| new file mode 100644 |
| index 000000000..e8d86cb79 |
| --- /dev/null |
| +++ b/internal/host/movesabort.go |
| @@ -0,0 +1,283 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "os" |
| + "sort" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The abort (moves spec §7.11, joint N13; Plan 5 Task 1 with rulings K5, K6, |
| +// K7, K19 and K21), behind CARLOS_MOVES_PHASE1. It replaces Plan 1's interim |
| +// abort, which released the claim at once and left the staging lane in the |
| +// target bucket with no owner. |
| +// |
| +// Who retires what. The source cannot read the target's moves/* nor write |
| +// its parked/ (M17), so it never touches the lane: the lane is parked by the |
| +// target region's box. The source retires only its own state. |
| +// |
| +// The source, for a record aborting in its region (phase1MoveRound): |
| +// |
| +// 1. with the claim at this move's moving/"" and this box executing, it |
| +// takes the staging lease, so no staging writer is mid-round and none |
| +// can sign another request (K5); |
| +// 2. it journals abort_bound once: the later of now + SignatureLife + |
| +// DefaultSkew and any seal's bound. Every staging request is signed at a |
| +// check that leaves its whole signature life before the lease's |
| +// SafeUntil, and none is signed after this moment while the lease is |
| +// held, so no request can land after the bound. A replay reads the |
| +// journalled bound and never computes a later one (revertClaim's |
| +// pattern); |
| +// 3. it CASes hostclaim.Abort with moves.AbortScope(host, op, inc, bound): |
| +// moving/aborting, the lane's park-replica queued for the target region, |
| +// owned by it. Execution stays with the source, which serves throughout; |
| +// 4. only then it retires its own state: the move's snapshots in its own |
| +// bucket, every version through DeleteVersion (K19: they are outside the |
| +// park grammar, so "the source parks its own snapshots" means it |
| +// retires them from its own bucket), then the receipts and the ledger, |
| +// and last the op's journal directory (the bound is in the claim by |
| +// then); |
| +// 5. it waits until no obligation under the op remains, then releases with |
| +// hostclaim.ReleaseAbort (K7: from moving/aborting only, so never before |
| +// the lane park was queued), and last clears the record with |
| +// provision.MoveGuardedAbortComplete (§7.5: the claim releases first, |
| +// the record clears second). A round that finds the claim already |
| +// serving and recording this move's release completes the record; any |
| +// other serving claim is left alone. |
| +// |
| +// The target region's box, for a record aborting whose To is its region: |
| +// movesParticipantRecord answers it as the target on the state alone (the |
| +// record never flipped, so the target has neither the row nor a journal) |
| +// and laneParkTarget runs the lane's park-replica through the park engine |
| +// over its own bucket until the engine acknowledges it after the horizon, |
| +// abort_bound + SignatureAge + RequestLimit. A late request found by the |
| +// engine's verification listing is parked as a further retirement event. |
| +// |
| +// A dead target region (K21). If no box of the target region runs the lane |
| +// park, the obligation stays owed, the claim stays moving/aborting and |
| +// nothing releases. That blocks a delete (and any other operation) of the |
| +// host until the region returns. Nothing needs to be done by hand; the |
| +// overdue page is Task 8's. |
| +// |
| +// Arming. The source's half needs only what a preparing round already has. |
| +// The target's lane park is unarmed in production, as the revert's is, |
| +// until Task 9 wires the park engine's leases (revertLaneArmed); the tests |
| +// inject it. Plan 1's DeletePrefix path stays for legacy records only |
| +// (legacyMoveRound); this path never deletes from the target bucket. |
| + |
| +// abortStepBound is the abort's journalled writers' horizon (K5). |
| +const abortStepBound = "abort_bound" |
| + |
| +// abFault is the abort's test seam after a side effect. |
| +func (w *movesWorker) abFault(point string) error { |
| + if f := w.cfg.abFault; f != nil { |
| + return f(point) |
| + } |
| + return nil |
| +} |
| + |
| +// abortSource is phase1MoveRound's MoveAborting arm. |
| +func (w *movesWorker) abortSource(ctx context.Context, s release.Store, rec provision.Record) error { |
| + cfg := w.cfg |
| + id := rec.Move.ID |
| + if cfg.Box == "" { |
| + return errors.New("the abort needs this box's label") |
| + } |
| + obs, err := hostclaim.Read(ctx, s, rec.Host) |
| + if err != nil { |
| + return fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent { |
| + w.notice(rec.Host+"#abort", "moves: %s: aborting move %s has no host claim; left for the console pass", rec.Host, id) |
| + return nil |
| + } |
| + op := obs.Claim.Op |
| + mine := op.Kind == hostclaim.KindMoving && op.ID == id |
| + switch { |
| + case mine && op.Exec != cfg.Box: |
| + w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim's executor is %q, not this box; left for the console pass", rec.Host, id, op.Exec) |
| + return nil |
| + case mine && op.Phase == hostclaim.PhaseMoving: |
| + ok, err := w.abortClaim(ctx, s, rec) |
| + if err != nil || !ok { |
| + return err |
| + } |
| + if obs, err = hostclaim.Read(ctx, s, rec.Host); err != nil { |
| + return fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseAborting { |
| + return nil // another writer moved it; the next pass reads it again |
| + } |
| + fallthrough |
| + case mine && op.Phase == hostclaim.PhaseAborting: |
| + if moveJournalPresent(cfg.DataDir, rec.Host, id) { |
| + if err := w.retireOwnMoveState(ctx, rec); err != nil { |
| + return err |
| + } |
| + } |
| + if released, err := w.releaseAbort(ctx, s, rec, obs); err != nil || !released { |
| + return err |
| + } |
| + case hostclaim.RequireReleasedMove(id)(obs.Claim) == nil: |
| + // This abort released the claim before a crash: finish. The own |
| + // state went before the release; retiring it again is a no-op. |
| + if err := w.retireOwnMoveState(ctx, rec); err != nil { |
| + return err |
| + } |
| + default: |
| + w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim is %s/%q (%s) and records no release of this move; left for the console pass", rec.Host, id, op.Kind, op.Phase, op.ID) |
| + return nil |
| + } |
| + g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation} |
| + if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedAbortComplete(g, cfg.Now())); err != nil { |
| + if errors.Is(err, provision.ErrMoveConflict) { |
| + log.Printf("moves: %s: abort of %s: the record moved on (%v)", rec.Host, id, err) |
| + return nil |
| + } |
| + return err |
| + } |
| + nudgeSerial(ctx, s) |
| + log.Printf("moves: %s: move %s aborted; the instance stays in %q", rec.Host, id, rec.Region) |
| + return nil |
| +} |
| + |
| +// abortClaim is steps 1 to 3: the staging lease, the journalled bound and |
| +// the Abort CAS. ok is false when the pass should wait (another worker holds |
| +// the lease). Without a deployment bucket no round ever took the lease and |
| +// none staged, so there is no lease to take and the bound alone holds. |
| +func (w *movesWorker) abortClaim(ctx context.Context, s release.Store, rec provision.Record) (bool, error) { |
| + cfg := w.cfg |
| + host, id := rec.Host, rec.Move.ID |
| + if cfg.StagingLease != nil { |
| + _, rel, err := cfg.StagingLease(ctx, rec.Move.To, host) |
| + if err != nil { |
| + var he *lease.HeldError |
| + if errors.As(err, &he) { |
| + w.notice(host+"#abort", "moves: %s: aborting move %s waits: another worker holds the staging lease", host, id) |
| + return false, nil |
| + } |
| + return false, fmt.Errorf("staging lease: %w", err) |
| + } |
| + defer func() { |
| + rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + defer cancel() |
| + if err := rel(rctx); err != nil { |
| + log.Printf("moves: %s: release the staging lease: %v", host, err) |
| + } |
| + }() |
| + } |
| + // Under the lease, the claim again: still this move's, before the seal. |
| + obs, err := hostclaim.Read(ctx, s, host) |
| + if err != nil { |
| + return false, fmt.Errorf("read the host claim: %w", err) |
| + } |
| + if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.ID != id || obs.Claim.Op.Phase != hostclaim.PhaseMoving || obs.Claim.Op.Exec != cfg.Box { |
| + return true, nil |
| + } |
| + if obs.Claim.Holder == nil { |
| + return false, errors.New("the abort needs the claim's holder") |
| + } |
| + j, err := moves.ReadJournal(cfg.DataDir, host, id) |
| + if err != nil { |
| + return false, err |
| + } |
| + var bound time.Time |
| + if raw, ok := j.Steps[abortStepBound]; ok { |
| + if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil { |
| + bound = time.Time{} |
| + } |
| + } |
| + if bound.IsZero() { |
| + bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC() |
| + if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) { |
| + bound = seal.Bound().UTC() |
| + } |
| + j.Mark(abortStepBound, bound.Format(time.RFC3339Nano)) |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return false, err |
| + } |
| + if err := w.abFault("after-abort-bound"); err != nil { |
| + return false, err |
| + } |
| + } |
| + scope := moves.AbortScope(host, id, obs.Claim.Holder.Incarnation, bound) |
| + if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Abort(id, rec.Move.To, scope, obligationID)); err != nil { |
| + return false, fmt.Errorf("abort the claim: %w", err) |
| + } |
| + log.Printf("moves: %s: the claim aborts move %s; the staging lane in %s is queued for its park there", host, id, rec.Move.To) |
| + if err := w.abFault("after-abort-claim"); err != nil { |
| + return false, err |
| + } |
| + return true, nil |
| +} |
| + |
| +// releaseAbort is step 5: the release once no obligation of the op remains. |
| +// released is false while the lane park is still owed. |
| +func (w *movesWorker) releaseAbort(ctx context.Context, s release.Store, rec provision.Record, obs hostclaim.Observed) (bool, error) { |
| + id := rec.Move.ID |
| + for _, o := range obs.Claim.Obligations { |
| + if o.Op == id { |
| + w.notice(rec.Host+"#abort-wait", "moves: %s: aborting move %s waits for %s to park the staging lane (%s %s)", rec.Host, id, rec.Move.To, o.Kind, o.Region+o.Box) |
| + return false, nil |
| + } |
| + } |
| + if _, err := hostclaim.Update(ctx, s, obs, hostclaim.ReleaseAbort(id, w.cfg.Box)); err != nil { |
| + return false, fmt.Errorf("release the host claim: %w", err) |
| + } |
| + if err := w.abFault("after-release-abort"); err != nil { |
| + return false, err |
| + } |
| + return true, nil |
| +} |
| + |
| +// retireOwnMoveState is step 4: the move's snapshots in this box's own |
| +// bucket, every version through DeleteVersion (K19), then its receipts and |
| +// ledger, and last its journal directory, which holds the bound. Idempotent. |
| +func (w *movesWorker) retireOwnMoveState(ctx context.Context, rec provision.Record) error { |
| + cfg := w.cfg |
| + if cfg.Versioned == nil { |
| + return errors.New("the abort needs this box's versioned store to retire its snapshots") |
| + } |
| + vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return err |
| + } |
| + prefix := moves.SnapshotsPrefix(rec.Host, rec.Move.ID) |
| + vers, err := vs.ListVersions(ctx, cfg.OwnBucket, prefix) |
| + if err != nil { |
| + return fmt.Errorf("list the move's snapshots: %w", err) |
| + } |
| + for _, v := range vers { |
| + if err := vs.DeleteVersion(ctx, cfg.OwnBucket, v.Key, v.VersionID); err != nil { |
| + return fmt.Errorf("delete %s version %s: %w", v.Key, v.VersionID, err) |
| + } |
| + } |
| + own, err := w.mirror(ctx, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return err |
| + } |
| + keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)} |
| + sort.Strings(keys) |
| + for _, k := range keys { |
| + if err := own.Delete(ctx, cfg.OwnBucket, k); err != nil && !errors.Is(err, lease.ErrMirrorNotFound) { |
| + return fmt.Errorf("delete %s: %w", k, err) |
| + } |
| + } |
| + if err := w.abFault("after-abort-own-state"); err != nil { |
| + return err |
| + } |
| + // The move's local directory: its journal and the archive kept for it. |
| + if err := os.RemoveAll(moves.JournalDir(cfg.DataDir, rec.Host, rec.Move.ID)); err != nil { |
| + return fmt.Errorf("remove the move's local directory: %w", err) |
| + } |
| + return w.abFault("after-abort-journal") |
| +} |
diff --git a/internal/host/movesabort_test.go b/internal/host/movesabort_test.go| new file mode 100644 |
| index 000000000..d0d65877f |
| --- /dev/null |
| +++ b/internal/host/movesabort_test.go |
| @@ -0,0 +1,487 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "os" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/park" |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The abort (moves spec §7.11; Plan 5 Task 1). The fixture is the Phase 1 |
| +// harness after one preparing round, so the lane holds staged objects in the |
| +// target bucket (data-uk) and the source's own bucket (data-ie) holds its |
| +// receipts, its ledger and a snapshot. The target region's box (uk-box) runs |
| +// the lane park through the real park engine over the same buckets with a |
| +// fake lease table. Everything else is real: the source's abort, hostclaim's |
| +// Abort and ReleaseAbort, and provision's guarded transitions. |
| + |
| +type abortHarness struct { |
| + *phase1Harness |
| + tdir string |
| + tleases *fakeParkLeases |
| + tw *movesWorker |
| + abAt string |
| + abDone bool |
| + abHits []string |
| +} |
| + |
| +func newAbortHarness(t *testing.T) *abortHarness { |
| + t.Helper() |
| + h := &abortHarness{phase1Harness: newPhase1Harness(t), tdir: t.TempDir(), tleases: &fakeParkLeases{}} |
| + h.start(t, "manual") |
| + h.pass(t) |
| + if len(h.laneKeys()) == 0 { |
| + t.Fatal("the fixture staged nothing") |
| + } |
| + h.f.put("data-ie", moves.SnapshotsPrefix(p1Host, p1ID)+"files-x.sealed", "s") |
| + h.cfg.abFault = func(point string) error { |
| + h.abHits = append(h.abHits, point) |
| + if point == h.abAt && !h.abDone { |
| + h.abDone = true |
| + return errCrash |
| + } |
| + return nil |
| + } |
| + return h |
| +} |
| + |
| +// abort is the console's abort door: the guarded record CAS. |
| +func (h *abortHarness) abort(t *testing.T) { |
| + t.Helper() |
| + g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| + if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// laneCfg is the target region's box with its lane park armed. |
| +func (h *abortHarness) laneCfg() moveCfg { |
| + c := h.targetCfg() |
| + c.DataDir = h.tdir |
| + c.ParkLeases = h.tleases |
| + return c |
| +} |
| + |
| +// tPass runs one pass as the target region's box. |
| +func (h *abortHarness) tPass(t *testing.T) { |
| + t.Helper() |
| + if h.tw == nil { |
| + h.tw = &movesWorker{mirrors: map[string]lease.MirrorStore{}} |
| + } |
| + h.tw.cfg = h.laneCfg() |
| + if err := movesPass(context.Background(), h.s, h.tw); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// laneChore is the lane's park the abort queued in the target region. |
| +func (h *abortHarness) laneChore(t *testing.T) *hostclaim.Obligation { |
| + t.Helper() |
| + for _, o := range h.claim(t).Obligations { |
| + if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) { |
| + return &o |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// laneKeys are the objects still in the move's lane in the target bucket. |
| +func (h *abortHarness) laneKeys() []string { |
| + var out []string |
| + for k := range h.f.buckets["data-uk"] { |
| + if strings.HasPrefix(k, moves.ScopeLocation(p1Host, p1ID)+"/") { |
| + out = append(out, k) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// archived counts the objects under one retirement's archive of the lane. |
| +func (h *abortHarness) archived(inc, ret string) int { |
| + prefix := "parked/" + inc + "/" + p1Host + "/" + ret + "/moves/" |
| + n := 0 |
| + for k := range h.f.buckets["data-uk"] { |
| + if strings.HasPrefix(k, prefix) { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// targetDeletes are every delete request anyone sent to the target bucket: |
| +// a plain delete, a guarded staging delete, or a delete by version. |
| +func (h *abortHarness) targetDeletes() []string { |
| + var out []string |
| + for _, op := range h.f.ops { |
| + if strings.HasPrefix(op, "delete data-uk/") { |
| + out = append(out, op) |
| + } |
| + } |
| + for _, c := range h.f.stagingCalls() { |
| + if c.Op == "delete" && c.Bucket == "data-uk" { |
| + out = append(out, "staging delete "+c.Key) |
| + } |
| + } |
| + out = append(out, h.vs.callsOn("delete-version data-uk/")...) |
| + return out |
| +} |
| + |
| +func (h *abortHarness) pastHorizon(t *testing.T, o *hostclaim.Obligation) { |
| + t.Helper() |
| + until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit) |
| + if d := until.Sub(h.clock.now); d > 0 { |
| + h.clock.Advance(d + time.Second) |
| + } |
| +} |
| + |
| +// waiting asserts the abort is still waiting for the lane park: the record |
| +// aborting, the claim moving/aborting under the source, the chore owed. |
| +func (h *abortHarness) waiting(t *testing.T) { |
| + t.Helper() |
| + if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting { |
| + t.Fatalf("the record is not aborting: %+v", m) |
| + } |
| + c := h.claim(t) |
| + if c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseAborting || c.Op.ID != p1ID || c.Op.Exec != "ie-box" { |
| + t.Fatalf("the claim while the abort waits: %+v", c.Op) |
| + } |
| +} |
| + |
| +// closed asserts the end: the record at rest in ie with LastMove aborted, |
| +// the claim serving under the source with this move recorded as released. |
| +func (h *abortHarness) closed(t *testing.T) { |
| + t.Helper() |
| + rec := h.read(t, p1Host) |
| + if rec.Move != nil || rec.Region != "ie" || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "aborted" { |
| + t.Fatalf("the record after the abort: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove) |
| + } |
| + c := h.claim(t) |
| + if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" || hostclaim.RequireReleasedMove(p1ID)(c) != nil || len(c.Obligations) != 0 { |
| + t.Fatalf("the claim after the abort: %+v obligations %+v", c.Op, c.Obligations) |
| + } |
| +} |
| + |
| +// ownStateGone asserts the source retired its own state for the move: the |
| +// snapshots, receipts and ledger in its bucket, and its journal directory. |
| +func (h *abortHarness) ownStateGone(t *testing.T) { |
| + t.Helper() |
| + for k := range h.f.buckets["data-ie"] { |
| + if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) || k == moves.ReceiptsKey(p1Host, p1ID) || k == moves.LedgerKey(p1Host, p1ID) { |
| + t.Fatalf("%s survived the abort", k) |
| + } |
| + } |
| + if _, err := os.Stat(moves.JournalDir(h.cfg.DataDir, p1Host, p1ID)); !os.IsNotExist(err) { |
| + t.Fatalf("the source's journal directory is still there: %v", err) |
| + } |
| +} |
| + |
| +// TestAbortParksFromTheTargetAndWaits (Review Focus 2, §7.11, joint N13): the |
| +// source's passes never send a delete to the target bucket; the source takes |
| +// the claim to moving/aborting with the lane's park queued for the target |
| +// region, and retires its own snapshots (by version), receipts, ledger and |
| +// journal. The target region's box archives moves/<host>/<id>/... under |
| +// parked/<inc>/<host>/<ret>/moves/... and acknowledges only after the |
| +// horizon. Only then does the source release the claim to serving under |
| +// itself, and the record clear. |
| +func TestAbortParksFromTheTargetAndWaits(t *testing.T) { |
| + h := newAbortHarness(t) |
| + lane := h.laneKeys() |
| + h.abort(t) |
| + deletes := len(h.targetDeletes()) |
| + acquired := h.lease.acquired |
| + decided := h.clock.now |
| + h.pass(t) |
| + h.waiting(t) |
| + o := h.laneChore(t) |
| + if o == nil { |
| + t.Fatal("no lane park queued") |
| + } |
| + if o.Owner != "region:uk" || o.Park.Incarnation != h.inc || o.Park.RequestLimit != moves.ParkRequestLimit { |
| + t.Fatalf("the lane park: %+v %+v", o, o.Park) |
| + } |
| + if min := decided.Add(moves.SignatureLife + moves.DefaultSkew); o.Park.EvidenceAt.Before(min) || o.Park.EvidenceAt.After(min.Add(time.Second)) { |
| + t.Fatalf("the park's horizon starts at %v, want the decision's time plus a signature's life and the skew, %v", o.Park.EvidenceAt, min) |
| + } |
| + if h.lease.acquired != acquired+1 || h.lease.released != h.lease.acquired { |
| + t.Fatalf("the staging lease: acquired %d (was %d), released %d", h.lease.acquired, acquired, h.lease.released) |
| + } |
| + h.ownStateGone(t) |
| + if got := h.vs.callsOn("delete-version data-ie/" + moves.SnapshotsPrefix(p1Host, p1ID)); len(got) == 0 { |
| + t.Fatal("the snapshots were not retired by version") |
| + } |
| + if got := h.laneKeys(); len(got) != len(lane) { |
| + t.Fatalf("the source touched the lane: %d keys, had %d", len(got), len(lane)) |
| + } |
| + // The source waits: another pass changes nothing and deletes nothing there. |
| + gen := h.claim(t).Gen |
| + h.pass(t) |
| + h.waiting(t) |
| + if h.claim(t).Gen != gen { |
| + t.Fatal("a waiting source wrote the claim") |
| + } |
| + if got := h.targetDeletes(); len(got) != deletes { |
| + t.Fatalf("the source sent deletes to the target bucket: %v", got[deletes:]) |
| + } |
| + // The target region's box archives the lane; the horizon has not passed. |
| + h.tPass(t) |
| + if left := h.laneKeys(); len(left) != 0 { |
| + t.Fatalf("the lane still holds %v", left) |
| + } |
| + if n := h.archived(o.Park.Incarnation, o.Retirement); n < len(lane) { |
| + t.Fatalf("%d archived, the lane had %d", n, len(lane)) |
| + } |
| + if h.laneChore(t) == nil { |
| + t.Fatal("the lane park was acknowledged before its horizon") |
| + } |
| + h.pass(t) |
| + h.waiting(t) |
| + // Past the horizon the target acknowledges; the source then releases. |
| + h.pastHorizon(t, o) |
| + h.tPass(t) |
| + if h.laneChore(t) != nil { |
| + t.Fatal("the lane park was not acknowledged after its horizon") |
| + } |
| + h.waiting(t) |
| + h.pass(t) |
| + h.closed(t) |
| + before := h.claim(t) |
| + h.pass(t) |
| + h.tPass(t) |
| + if after := h.claim(t); after.Gen != before.Gen { |
| + t.Fatalf("a pass after the abort wrote the claim (%d to %d)", before.Gen, after.Gen) |
| + } |
| +} |
| + |
| +// TestAbortScopeIsTheWritersHorizon (witness wStagingQuiet; ruling K5): the |
| +// abort's horizon covers every staging request the source signed before it, |
| +// and a staging PUT signed before the abort that lands after the park's |
| +// first pass is found by the verification listing and parked as a further |
| +// retirement event. The park acknowledges only after abort_bound + H. |
| +func TestAbortScopeIsTheWritersHorizon(t *testing.T) { |
| + h := newAbortHarness(t) |
| + signed := h.f.stagingCalls() |
| + if len(signed) == 0 { |
| + t.Fatal("the fixture signed no staging request") |
| + } |
| + h.abort(t) |
| + h.pass(t) |
| + o := h.laneChore(t) |
| + if o == nil { |
| + t.Fatal("no lane park queued") |
| + } |
| + bound := o.Park.EvidenceAt |
| + for _, c := range signed { |
| + if c.SignAt.Add(moves.SignatureLife + moves.DefaultSkew).After(bound) { |
| + t.Fatalf("a staging request signed at %v can land after the abort's horizon %v", c.SignAt, bound) |
| + } |
| + } |
| + // The park's first pass archives the lane. |
| + h.tPass(t) |
| + first := o.Retirement |
| + if len(h.laneKeys()) != 0 || h.laneChore(t) == nil { |
| + t.Fatalf("after the first pass: lane %v", h.laneKeys()) |
| + } |
| + // A request signed before the abort lands now, in the lane. |
| + late := moves.ScopeLocation(p1Host, p1ID) + "/replica/0000/0000000000000009-0000000000000009.ltx" |
| + h.f.store("data-uk", late, fakeMoveObject{lease.MirrorObject{ETag: "late", Size: 4}, "late"}) |
| + h.f.setVersion("data-uk", late, h.f.nextVersion()) |
| + // One second before abort_bound + H nothing is acknowledged, whatever |
| + // the listing says. |
| + deadline := bound.Add(park.SignatureAge + o.Park.RequestLimit) |
| + if d := deadline.Add(-time.Second).Sub(h.clock.now); d > 0 { |
| + h.clock.Advance(d) |
| + } |
| + h.tPass(t) |
| + if h.laneChore(t) == nil { |
| + t.Fatal("the lane park was acknowledged before abort_bound + H") |
| + } |
| + h.pass(t) |
| + h.waiting(t) |
| + // Past the horizon the listing finds the late write: a further retirement. |
| + h.pastHorizon(t, o) |
| + h.tPass(t) |
| + o2 := h.laneChore(t) |
| + if o2 == nil { |
| + t.Fatal("the park acknowledged over the late write") |
| + } |
| + if o2.Retirement == first || !o2.Park.EvidenceAt.Equal(bound) { |
| + t.Fatalf("the further event: retirement %q (first %q), horizon %v", o2.Retirement, first, o2.Park.EvidenceAt) |
| + } |
| + h.tPass(t) |
| + if h.laneChore(t) != nil { |
| + t.Fatal("the further retirement was not acknowledged") |
| + } |
| + if _, ok := h.f.buckets["data-uk"][late]; ok { |
| + t.Fatal("the late write is still in the lane") |
| + } |
| + if n := h.archived(o.Park.Incarnation, o2.Retirement); n != 1 { |
| + t.Fatalf("the further retirement archived %d objects, want the late write", n) |
| + } |
| + if q := h.claim(t).QuietAfter["uk"]; !q.Equal(deadline) { |
| + t.Fatalf("quiet_after %v, want %v", q, deadline) |
| + } |
| + h.pass(t) |
| + h.closed(t) |
| +} |
| + |
| +// TestAbortWaitsForAnUnreachableTargetRegion (ruling K21): with no box of |
| +// the target region running the lane park, the claim stays moving/aborting |
| +// and nothing releases, however long the source waits. A delete cannot |
| +// begin over it, because the claim is not serving. |
| +func TestAbortWaitsForAnUnreachableTargetRegion(t *testing.T) { |
| + h := newAbortHarness(t) |
| + h.abort(t) |
| + h.pass(t) |
| + o := h.laneChore(t) |
| + if o == nil { |
| + t.Fatal("no lane park queued") |
| + } |
| + for i := 0; i < 4; i++ { |
| + h.clock.Advance(12 * time.Hour) |
| + h.pass(t) |
| + h.waiting(t) |
| + if h.laneChore(t) == nil { |
| + t.Fatal("the lane park vanished with no target pass") |
| + } |
| + } |
| + c := h.claim(t) |
| + if err := hostclaim.StartGuard(c, []string{"ie"}, []string{"ie-box"}); err == nil { |
| + t.Fatal("a new operation could start over the waiting abort") |
| + } |
| + // An unarmed target box (production until Task 9) parks nothing either. |
| + unarmed := h.laneCfg() |
| + unarmed.ParkLeases = nil |
| + w := &movesWorker{cfg: unarmed, mirrors: map[string]lease.MirrorStore{}} |
| + if err := movesPass(context.Background(), h.s, w); err != nil { |
| + t.Fatal(err) |
| + } |
| + if h.laneChore(t) == nil || len(h.laneKeys()) == 0 { |
| + t.Fatal("an unarmed target box parked the lane") |
| + } |
| + // Once the region returns, the abort completes. |
| + h.tPass(t) |
| + h.pastHorizon(t, o) |
| + h.tPass(t) |
| + h.pass(t) |
| + h.closed(t) |
| +} |
| + |
| +// TestAbortReplays: a crash after any of the abort's side effects, then a |
| +// restarted agent, still reaches the close, with the journalled bound named |
| +// once (the lane park's scope never changes), one lane park queued, and the |
| +// source's own state retired. |
| +func TestAbortReplays(t *testing.T) { |
| + for _, point := range []string{"after-abort-bound", "after-abort-claim", "after-abort-own-state", "after-abort-journal", "after-release-abort"} { |
| + t.Run(point, func(t *testing.T) { |
| + h := newAbortHarness(t) |
| + h.abort(t) |
| + h.abAt = point |
| + if err := movesPass(context.Background(), h.s, h.w()); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i := 0; i < 6 && !h.abDone; i++ { |
| + h.tPass(t) |
| + if o := h.laneChore(t); o != nil { |
| + h.pastHorizon(t, o) |
| + } |
| + h.pass(t) |
| + } |
| + if !h.abDone { |
| + t.Fatalf("never reached %s (%v)", point, h.abHits) |
| + } |
| + var bound time.Time |
| + if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID); err == nil { |
| + if raw := j.Steps[abortStepBound]; raw != "" { |
| + bound, _ = time.Parse(time.RFC3339Nano, raw) |
| + } |
| + } |
| + if o := h.laneChore(t); o != nil { |
| + if !bound.IsZero() && !o.Park.EvidenceAt.Equal(bound) { |
| + t.Fatalf("the park's horizon %v is not the journalled bound %v", o.Park.EvidenceAt, bound) |
| + } |
| + bound = o.Park.EvidenceAt |
| + } |
| + h.clock.Advance(time.Minute) |
| + h.worker1 = nil |
| + for i := 0; i < 8 && h.read(t, p1Host).Move != nil; i++ { |
| + h.pass(t) |
| + if o := h.laneChore(t); o != nil { |
| + if !bound.IsZero() && !o.Park.EvidenceAt.Equal(bound) { |
| + t.Fatalf("the replay named another horizon: %v, journalled %v", o.Park.EvidenceAt, bound) |
| + } |
| + h.tPass(t) |
| + h.pastHorizon(t, o) |
| + h.tPass(t) |
| + } |
| + } |
| + h.closed(t) |
| + h.ownStateGone(t) |
| + }) |
| + } |
| +} |
| + |
| +// An aborting record never flipped: its region is the source's, so the |
| +// target region's box has neither the row nor a journal of the move. The |
| +// participant rule answers it as the target on the state alone, as for a |
| +// revert: its one piece of work is the lane park (ruling K6). The source's |
| +// region answers nothing without the row (its abort is phase1MoveRound's), |
| +// and a third region nothing. |
| +func TestAbortingTargetIsAParticipant(t *testing.T) { |
| + cfg := moveCfg{Region: "uk", DataDir: t.TempDir(), OwnsRoute: func(string) bool { return false }} |
| + rec := provision.Record{Host: p1Host, Region: "ie", Move: &provision.MoveState{ID: p1ID, State: provision.MoveAborting, From: "ie", To: "uk"}} |
| + if got := movesParticipantRecord(cfg, rec); got != roleTarget { |
| + t.Fatalf("an aborting record's target: %q", got) |
| + } |
| + cfg.Region = "ie" |
| + if got := movesParticipantRecord(cfg, rec); got != roleNone { |
| + t.Fatalf("the source's region without the row: %q", got) |
| + } |
| + cfg.Region = "fr" |
| + if got := movesParticipantRecord(cfg, rec); got != roleNone { |
| + t.Fatalf("a third region: %q", got) |
| + } |
| +} |
| + |
| +// The abort's crash boundary between the claim's release and the record's |
| +// clear: a round that finds the record aborting and the claim serving |
| +// completes the record only when the serving op records this move as the |
| +// one it released (RequireReleasedMove). A serving claim without that |
| +// evidence was released by someone else, and the record is left alone. |
| +func TestAbortCompletesOnlyWithItsReleaseEvidence(t *testing.T) { |
| + ctx := context.Background() |
| + h := newAbortHarness(t) |
| + h.abort(t) |
| + obs, err := hostclaim.Read(ctx, h.s, p1Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + scope := moves.AbortScope(p1Host, p1ID, h.inc, h.clock.now.Add(time.Hour)) |
| + if obs, err = hostclaim.Update(ctx, h.s, obs, hostclaim.Abort(p1ID, "uk", scope, obligationID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if obs, err = hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, obs.Claim.Obligations[0].ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err = hostclaim.Update(ctx, h.s, obs, hostclaim.ReleaseAbort(p1ID, "ie-box")); err != nil { |
| + t.Fatal(err) |
| + } |
| + h.pass(t) |
| + h.closed(t) |
| + h.ownStateGone(t) |
| + |
| + // A serving claim with no evidence of this release: left alone. |
| + h2 := newAbortHarness(t) |
| + h2.abort(t) |
| + rewriteClaimOp(t, h2.s, func(op *hostclaim.Op) { *op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "ie-box"} }) |
| + h2.pass(t) |
| + if got := h2.read(t, p1Host); got.Move == nil || got.Move.State != provision.MoveAborting { |
| + t.Fatalf("a serving claim without evidence completed the record: %+v", got.Move) |
| + } |
| +} |
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go| index b65ae5a78..b1cd2a78d 100644 |
| --- a/internal/host/movesfinal.go |
| +++ b/internal/host/movesfinal.go |
| @@ -84,12 +84,13 @@ const ( |
| const finalConfigAckName = "config-ack-source.json" |
| // movesParticipantStates are the record states after the flip (and a |
| -// revert) in which a participant may have work for a record whose region |
| -// no longer names it, or whose row it has already deleted. |
| +// revert or an abort) in which a participant may have work for a record |
| +// whose region no longer names it, or whose row it has already deleted. |
| var movesParticipantStates = map[string]bool{ |
| provision.MoveCommitted: true, |
| provision.MoveRollingBack: true, |
| provision.MoveReverting: true, |
| + provision.MoveAborting: true, |
| } |
| // movesParticipantRecord is the routing rule for the states after the flip |
| @@ -117,10 +118,10 @@ func movesParticipantRecord(cfg moveCfg, rec provision.Record) string { |
| if cfg.OwnsRoute == nil || cfg.OwnsRoute(rec.Host) || moveJournalPresent(cfg.DataDir, rec.Host, m.ID) { |
| return role |
| } |
| - // A revert never flipped: its target has neither the row nor a journal, |
| - // and its one piece of work is the lane park the claim owes its region |
| - // (movesrevert.go, Plan 4 Task 6). |
| - if role == roleTarget && m.State == provision.MoveReverting { |
| + // A revert or an abort never flipped: its target has neither the row |
| + // nor a journal, and its one piece of work is the lane park the claim |
| + // owes its region (laneParkTarget; Plan 4 Task 6, Plan 5 Task 1). |
| + if role == roleTarget && (m.State == provision.MoveReverting || m.State == provision.MoveAborting) { |
| return role |
| } |
| return roleNone |
| @@ -139,8 +140,9 @@ func moveJournalPresent(dataDir, host, op string) bool { |
| // participantRound is a post-flip participant's step for a record the |
| // region rule did not admit (movesPass): the source's finalization of a |
| -// committed move, and the target's half of a rollback (movesrollback.go, |
| -// J10). The rest is Plan 4's later tasks'. |
| +// committed move, the target's half of a rollback (movesrollback.go, J10), |
| +// and the target region's lane park of a revert or an abort |
| +// (laneParkTarget). The rest is Plan 4's later tasks'. |
| func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec provision.Record, role string) error { |
| switch { |
| case role == roleSource && rec.Move.State == provision.MoveCommitted: |
| @@ -148,7 +150,9 @@ func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec |
| case role == roleTarget && rec.Move.State == provision.MoveRollingBack: |
| return w.rollbackTarget(ctx, s, rec) |
| case role == roleTarget && rec.Move.State == provision.MoveReverting: |
| - return w.revertLanePark(ctx, s, rec) |
| + return w.laneParkTarget(ctx, s, rec, hostclaim.KindReverting) |
| + case role == roleTarget && rec.Move.State == provision.MoveAborting: |
| + return w.laneParkTarget(ctx, s, rec, hostclaim.KindMoving, hostclaim.PhaseAborting) |
| } |
| return nil |
| } |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index f1ebed17b..2bf3d96d5 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -618,6 +618,12 @@ type moveCfg struct { |
| // production. |
| RestatusHold func(host, status, op string, validate func() error) (bool, error) |
| rvFault func(point string) error |
| + |
| + // The abort (§7.11, movesabort.go): the source uses the staging lease, |
| + // Versioned and its own bucket above; the target region's lane park is |
| + // laneParkTarget's, armed as the revert's (revertLaneArmed). abFault is |
| + // its test seam, nil in production. |
| + abFault func(point string) error |
| } |
| // defaultMoveReadyDelta is the readiness bar when |
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go| index d997b5f3f..485677065 100644 |
| --- a/internal/host/movesphase1.go |
| +++ b/internal/host/movesphase1.go |
| @@ -10,7 +10,6 @@ import ( |
| "log" |
| "os" |
| "path/filepath" |
| - "sort" |
| "strings" |
| "time" |
| @@ -185,7 +184,9 @@ func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec p |
| case provision.MovePreparing, provision.MoveReady: |
| return w.preparingRound(ctx, s, rec, fleet) |
| case provision.MoveAborting: |
| - return w.interimAbort(ctx, s, rec) |
| + // The abort's source half (movesabort.go, Plan 5 Task 1): the lane |
| + // is the target region's to park. |
| + return w.abortSource(ctx, s, rec) |
| case provision.MoveCutting: |
| return w.runCut(ctx, s, rec, fleet) |
| case provision.MoveReverting: |
| @@ -955,85 +956,3 @@ func readReport(ctx context.Context, s release.Store, key string, into any) (boo |
| } |
| return true, nil |
| } |
| - |
| -// interimAbort is the abort's box half until Plan 5 replaces it (§7.11): |
| -// the source deletes its own snapshots, receipts and ledger for the move, |
| -// releases the claim back to serving (K8), and then completes the record. |
| -// The staging lane in the target bucket is left for the target's park. |
| -// |
| -// The crash boundary between the claim release and the record write: a |
| -// round that finds the record aborting and the claim already serving |
| -// completes the record only when the serving op records this move as the |
| -// one it released (RequireReleasedMove). A serving claim without that |
| -// evidence was released by someone else (the console pass, Plan 5), and |
| -// the record is left to it. |
| -func (w *movesWorker) interimAbort(ctx context.Context, s release.Store, rec provision.Record) error { |
| - cfg := w.cfg |
| - id := rec.Move.ID |
| - g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation} |
| - obs, err := hostclaim.Read(ctx, s, rec.Host) |
| - if err != nil { |
| - return fmt.Errorf("read the host claim: %w", err) |
| - } |
| - if obs.Absent { |
| - w.notice(rec.Host+"#abort", "moves: %s: aborting move %s has no host claim; left for the console pass", rec.Host, id) |
| - return nil |
| - } |
| - c := obs.Claim |
| - switch { |
| - case c.Op.Kind == hostclaim.KindMoving && c.Op.ID == id: |
| - if err := w.deleteOwnMoveState(ctx, rec); err != nil { |
| - return err |
| - } |
| - box := cfg.Box |
| - if box == "" { |
| - box = c.Op.Exec |
| - } |
| - if _, err := hostclaim.Update(ctx, s, obs, hostclaim.ReleaseMove(id, box)); err != nil { |
| - return fmt.Errorf("release the host claim: %w", err) |
| - } |
| - case hostclaim.RequireReleasedMove(id)(c) == nil: |
| - // The claim was released by this abort before a crash: finish. |
| - if err := w.deleteOwnMoveState(ctx, rec); err != nil { |
| - return err |
| - } |
| - default: |
| - w.notice(rec.Host+"#abort", "moves: %s: aborting move %s: the claim is %s/%s and records no release of this move; left for the console pass", rec.Host, id, c.Op.Kind, c.Op.ID) |
| - return nil |
| - } |
| - if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedAbortComplete(g, cfg.Now())); err != nil { |
| - if errors.Is(err, provision.ErrMoveConflict) { |
| - log.Printf("moves: %s: abort of %s: the record moved on (%v)", rec.Host, id, err) |
| - return nil |
| - } |
| - return err |
| - } |
| - nudgeSerial(ctx, s) |
| - log.Printf("moves: %s: move %s aborted; the instance stays in %q", rec.Host, id, rec.Region) |
| - return nil |
| -} |
| - |
| -// deleteOwnMoveState removes the move's snapshots, receipts and ledger from |
| -// this box's own bucket. Idempotent. |
| -func (w *movesWorker) deleteOwnMoveState(ctx context.Context, rec provision.Record) error { |
| - cfg := w.cfg |
| - own, err := w.mirror(ctx, cfg.OwnBucketRegion) |
| - if err != nil { |
| - return err |
| - } |
| - if err := lease.DeletePrefix(ctx, own, cfg.OwnBucket, moves.SnapshotsPrefix(rec.Host, rec.Move.ID)); err != nil { |
| - return fmt.Errorf("delete the move's snapshots: %w", err) |
| - } |
| - keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)} |
| - sort.Strings(keys) |
| - for _, k := range keys { |
| - if err := own.Delete(ctx, cfg.OwnBucket, k); err != nil && !errors.Is(err, lease.ErrMirrorNotFound) { |
| - return fmt.Errorf("delete %s: %w", k, err) |
| - } |
| - } |
| - // The move's local directory: its journal and the archive kept for it. |
| - if err := os.RemoveAll(moves.JournalDir(cfg.DataDir, rec.Host, rec.Move.ID)); err != nil { |
| - return fmt.Errorf("remove the move's local directory: %w", err) |
| - } |
| - return nil |
| -} |
diff --git a/internal/host/movesphase1_test.go b/internal/host/movesphase1_test.go| index 4c1fb6e5a..a6efc9a4c 100644 |
| --- a/internal/host/movesphase1_test.go |
| +++ b/internal/host/movesphase1_test.go |
| @@ -199,67 +199,6 @@ func TestStagingMoveJSONReplayAfterLostLedger(t *testing.T) { |
| } |
| } |
| -// The interim abort (§7.11): the source deletes its own snapshots, receipts |
| -// and ledger, releases the claim to serving, then completes the record; |
| -// the staging lane is left for the target's park. A crash between the two |
| -// writes completes on the next round only with the release's evidence, and |
| -// a serving claim without it is left to the console pass. |
| -func TestInterimAbortReleasesTheClaim(t *testing.T) { |
| - ctx := context.Background() |
| - h := newPhase1Harness(t) |
| - h.start(t, "manual") |
| - h.pass(t) |
| - h.f.put("data-ie", moves.SnapshotsPrefix(p1Host, p1ID)+"files-x.sealed", "s") |
| - g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| - if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil { |
| - t.Fatal(err) |
| - } |
| - lane := len(stagedKeys(h, moves.StagingPrefix(p1Host, p1ID))) |
| - h.pass(t) |
| - got := h.read(t, p1Host) |
| - if got.Move != nil || got.LastMove == nil || got.LastMove.Outcome != "aborted" { |
| - t.Fatalf("abort did not complete: %+v %+v", got.Move, got.LastMove) |
| - } |
| - if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" { |
| - t.Fatalf("claim after the abort: %+v", c.Op) |
| - } |
| - for _, k := range []string{moves.ReceiptsKey(p1Host, p1ID), moves.LedgerKey(p1Host, p1ID), moves.SnapshotsPrefix(p1Host, p1ID) + "files-x.sealed"} { |
| - if _, ok := h.f.buckets["data-ie"][k]; ok { |
| - t.Fatalf("%s survived the abort", k) |
| - } |
| - } |
| - if n := len(stagedKeys(h, moves.StagingPrefix(p1Host, p1ID))); n != lane || n == 0 { |
| - t.Fatalf("the source touched the lane: %d keys, had %d", n, lane) |
| - } |
| - |
| - // The crash case: the claim released, the record still aborting. |
| - h2 := newPhase1Harness(t) |
| - h2.start(t, "manual") |
| - if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil { |
| - t.Fatal(err) |
| - } |
| - obs, _ := hostclaim.Read(ctx, h2.s, p1Host) |
| - if _, err := hostclaim.Update(ctx, h2.s, obs, hostclaim.ReleaseMove(p1ID, "ie-box")); err != nil { |
| - t.Fatal(err) |
| - } |
| - h2.pass(t) |
| - if got := h2.read(t, p1Host); got.Move != nil { |
| - t.Fatalf("the replay did not complete the record: %+v", got.Move) |
| - } |
| - |
| - // A serving claim with no evidence of this release: left alone. |
| - h3 := newPhase1Harness(t) |
| - h3.start(t, "manual") |
| - if _, err := provision.SetRecordMove(ctx, h3.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil { |
| - t.Fatal(err) |
| - } |
| - rewriteClaimOp(t, h3.s, func(op *hostclaim.Op) { *op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "ie-box"} }) |
| - h3.pass(t) |
| - if got := h3.read(t, p1Host); got.Move == nil || got.Move.State != provision.MoveAborting { |
| - t.Fatalf("a serving claim without evidence completed the record: %+v", got.Move) |
| - } |
| -} |
| - |
| // rewriteClaimOp edits p1Host's claim op in place, as another writer would. |
| func rewriteClaimOp(t *testing.T, s release.Store, edit func(*hostclaim.Op)) { |
| t.Helper() |
diff --git a/internal/host/movesreport.go b/internal/host/movesreport.go| index 147dbe336..b1ffb51a3 100644 |
| --- a/internal/host/movesreport.go |
| +++ b/internal/host/movesreport.go |
| @@ -133,7 +133,13 @@ func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, rec |
| // A move into this box is live from preparing through the cut and |
| // the landing (ruling G26): the held refusal a landing's configured |
| // phase writes, and T5's verdict, stay until the move is over. The |
| - // report itself is written only while preparing or ready. |
| + // report itself is written only while preparing or ready. An |
| + // aborting move is over for this box: it never flipped, so no |
| + // landing wrote a refusal, and its record stays aborting only until |
| + // the lane park is acknowledged (Plan 5 Task 1). |
| + if m.State == provision.MoveAborting { |
| + continue |
| + } |
| live[m.ID] = true |
| if m.State != provision.MovePreparing && m.State != provision.MoveReady { |
| continue |
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go| index 3a2f6e0c4..57537e4dd 100644 |
| --- a/internal/host/movesrevert.go |
| +++ b/internal/host/movesrevert.go |
| @@ -68,7 +68,7 @@ import ( |
| // the claim steps to opened. |
| // - opened: the claim waits for the lane's park the revert queued in the |
| // target region (§7.11), which the target region's box runs |
| -// (revertLanePark) and the park engine acknowledges after its horizon. |
| +// (laneParkTarget) and the park engine acknowledges after its horizon. |
| // Then ReleaseMove to serving under this box, and MoveRevertClose: |
| // LastMove rolled_back, offline from cut_at to the reopen. The source's |
| // journal of the op is parked in the same pass; a crash before that |
| @@ -77,13 +77,13 @@ import ( |
| // A claim this move already released (RequireReleasedMove: a crash before the |
| // record's close) only closes. |
| // |
| -// The target region's half (revertLanePark). A revert never flipped, so its |
| +// The target region's half (laneParkTarget). A revert never flipped, so its |
| // target box has neither the row nor a journal of the move; the participant |
| // rule answers it as the target on the state alone, and its one piece of work |
| // is the lane's park-replica the claim owes its region: the park engine over |
| // its own bucket archives moves/<host>/<op>/ under |
| // parked/<inc>/<host>/<ret>/moves/ and acknowledges once the horizon has |
| -// passed (exactly as Plan 5 Task 1 dispatches an abort's). |
| +// passed (laneParkTarget dispatches an abort's the same way, Plan 5 Task 1). |
| // |
| // Every record write goes through a provision transition (the record_seq |
| // rule); every claim write through a hostclaim transition. |
| @@ -835,28 +835,40 @@ func (w *movesWorker) sweepRevertJournal(dataDir, host, op string, j *moves.Jour |
| return parkRevertJournal(dataDir, host, op, true) |
| } |
| -// revertLanePark is the target region's half of a revert (movesPass through |
| -// participantRound): the lane's park-replica the revert queued in this |
| -// region, run by the park engine over this box's own bucket until it is |
| -// acknowledged after its horizon. The view and the authorizer are Plan 5's |
| -// when wired, else the stand-ins: the source's region authoritative (a revert |
| -// never flipped), and only this move's reverting lane park. |
| -func (w *movesWorker) revertLanePark(ctx context.Context, s release.Store, rec provision.Record) error { |
| +// laneParkTarget is the target region's half of a move outcome that leaves |
| +// the move's own staging lane behind in its bucket (movesPass through |
| +// participantRound): the lane's park-replica the claim owes this region, run |
| +// by the park engine over this box's own bucket until it is acknowledged |
| +// after its horizon. It serves a revert (the claim reverting, any phase; the |
| +// park hostclaim.Revert queued) and an abort (moving/aborting; the park |
| +// hostclaim.Abort queued, Plan 5 Task 1); Task 3 adds the committed landing. |
| +// No phases means any phase of kind. The view and the authorizer are Plan |
| +// 5's when wired, else the stand-ins: the source's region authoritative |
| +// (neither outcome flipped), and only this move's lane park under kind and |
| +// its phases. |
| +func (w *movesWorker) laneParkTarget(ctx context.Context, s release.Store, rec provision.Record, kind string, phases ...string) error { |
| cfg := w.cfg |
| m := rec.Move |
| - if m == nil || m.Legacy() || m.State != provision.MoveReverting || m.To != cfg.Region || m.From == cfg.Region { |
| + if m == nil || m.Legacy() || m.To != cfg.Region || m.From == cfg.Region { |
| return nil |
| } |
| + notice := rec.Host + "#lane-" + kind |
| + if kind == hostclaim.KindReverting { |
| + notice = rec.Host + "#rv-lane" |
| + } |
| if !revertLaneArmed(cfg) { |
| - w.notice(rec.Host+"#rv-lane", "moves: %s: the park of move %s's lane waits: this box's park engine is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.ID) |
| + w.notice(notice, "moves: %s: the park of move %s's lane waits: this box's park engine is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.ID) |
| return nil |
| } |
| host, id := rec.Host, m.ID |
| + inOp := func(op hostclaim.Op) bool { |
| + return op.ID == id && op.Kind == kind && (len(phases) == 0 || containsString(phases, op.Phase)) |
| + } |
| obs, err := hostclaim.Read(ctx, s, host) |
| if err != nil { |
| return fmt.Errorf("read the host claim: %w", err) |
| } |
| - if obs.Absent || obs.Claim.Op.ID != id || obs.Claim.Op.Kind != hostclaim.KindReverting { |
| + if obs.Absent || !inOp(obs.Claim.Op) { |
| return nil // the claim half has not followed yet, or it is over |
| } |
| if h := obs.Claim.Holder; h == nil || h.Incarnation != m.Incarnation { |
| @@ -883,9 +895,9 @@ func (w *movesWorker) revertLanePark(ctx context.Context, s release.Store, rec p |
| if auth == nil { |
| auth = func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| c := obs.Claim |
| - if obs.Absent || c == nil || c.Op.ID != id || c.Op.Kind != hostclaim.KindReverting || c.Op.Exec == cfg.Box || |
| + if obs.Absent || c == nil || !inOp(c.Op) || c.Op.Exec == cfg.Box || |
| o.Op != id || o.Kind != "park-replica" || o.Region != cfg.Region || o.Park == nil || o.Park.Location != lane { |
| - return fmt.Errorf("%w: not this revert's lane park", park.ErrUnsafe) |
| + return fmt.Errorf("%w: not this move's lane park", park.ErrUnsafe) |
| } |
| return nil |
| } |
| @@ -896,7 +908,7 @@ func (w *movesWorker) revertLanePark(ctx context.Context, s release.Store, rec p |
| err := e.Run(ctx, park.Work{Host: host, ID: o.ID, Op: id}) |
| switch { |
| case errors.Is(err, park.ErrPending): |
| - w.notice(host+"#rv-lane", "moves: %s: move %s's lane is archived; its park is acknowledged after its horizon", host, id) |
| + w.notice(notice, "moves: %s: move %s's lane is archived; its park is acknowledged after its horizon", host, id) |
| case errors.Is(err, park.ErrChanged): |
| case err != nil: |
| return fmt.Errorf("park move %s's lane: %w", id, err) |
diff --git a/internal/hostclaim/move.go b/internal/hostclaim/move.go| index ad0bfbf61..4640e7e71 100644 |
| --- a/internal/hostclaim/move.go |
| +++ b/internal/hostclaim/move.go |
| @@ -17,7 +17,7 @@ import ( |
| const ( |
| PhaseMoving = "" // moving, before the seal |
| PhaseSealed = "sealed" // moving, after the cut's staged phase |
| - PhaseAborting = "aborting" // moving, the interim abort has begun |
| + PhaseAborting = "aborting" // moving, the abort has begun (Abort): the lane park is queued |
| ) |
| // Journal keys the cut and the flip write on the claim's op (moves spec |
| @@ -133,22 +133,24 @@ func RequireMoveOp(opID, kind, phase string) func(*Claim) error { |
| } |
| // ReleaseMove is K8: a move hands the host back to serving. It is legal |
| -// from the interim abort (moving before the seal, phase "" or "aborting"), |
| -// and from the end of each move outcome, read from op.phase (J1, J6): a |
| -// committed landing, an opened rollback (restoring) and an opened revert |
| -// (reverting). In those three the executor named must be the one executing. |
| -// It is legal only once no obligation under this op remains. The serving op |
| -// keeps exec and records which move released it: operation-bound completion |
| -// evidence for the abort's crash boundary (RequireReleasedMove). |
| +// from moving before the seal (phase "": the request door's compensation for |
| +// a move whose record start failed), and from the end of each move outcome, |
| +// read from op.phase (J1, J6): a committed landing, an opened rollback |
| +// (restoring) and an opened revert (reverting). In those three the executor |
| +// named must be the one executing. An aborting op is ReleaseAbort's (Plan 5 |
| +// ruling K7), never this. It is legal only once no obligation under this op |
| +// remains. The serving op keeps exec and records which move released it: |
| +// operation-bound completion evidence for the abort's crash boundary |
| +// (RequireReleasedMove). |
| func ReleaseMove(opID, execBox string) func(*Claim) error { |
| return func(c *Claim) error { |
| if err := RequireOp(opID)(c); err != nil { |
| return err |
| } |
| - abort := c.Op.Kind == KindMoving && (c.Op.Phase == PhaseMoving || c.Op.Phase == PhaseAborting) |
| + door := c.Op.Kind == KindMoving && c.Op.Phase == PhaseMoving |
| ended := (c.Op.Kind == KindLanding && c.Op.Phase == PhaseCommitted) || |
| ((c.Op.Kind == KindRestoring || c.Op.Kind == KindReverting) && c.Op.Phase == PhaseOpened) |
| - if !abort && !ended { |
| + if !door && !ended { |
| return fmt.Errorf("%w: op is %s/%q, a release needs moving before the seal, a committed landing or an opened recovery", ErrWrongState, c.Op.Kind, c.Op.Phase) |
| } |
| if execBox == "" { |
| @@ -157,16 +159,93 @@ func ReleaseMove(opID, execBox string) func(*Claim) error { |
| if ended && c.Op.Exec != execBox { |
| return fmt.Errorf("%w: exec is %q, the release names %q", ErrWrongState, c.Op.Exec, execBox) |
| } |
| - for _, o := range c.Obligations { |
| - if o.Op == opID { |
| - return fmt.Errorf("%w: %s %s%s is still owed under this move", ErrPendingChore, o.Kind, o.Region, o.Box) |
| + return releaseToServing(c, opID, execBox) |
| + } |
| +} |
| + |
| +// releaseToServing is every move release's last step: legal only once no |
| +// obligation under the op remains; the serving op keeps the executor and |
| +// records the move it released. |
| +func releaseToServing(c *Claim, opID, execBox string) error { |
| + for _, o := range c.Obligations { |
| + if o.Op == opID { |
| + return fmt.Errorf("%w: %s %s%s is still owed under this move", ErrPendingChore, o.Kind, o.Region, o.Box) |
| + } |
| + } |
| + c.Op = Op{Kind: KindServing, Exec: execBox, Journal: map[string]string{journalReleasedMove: opID}} |
| + return nil |
| +} |
| + |
| +// Abort is the claim half of a move's abort (moves spec §7.11, joint N13; |
| +// Plan 5 Task 1): this move's moving op before the seal, with the source |
| +// still executing, goes to moving/aborting, and the park of its own staging |
| +// lane is queued in the same write for the target region, owned by that |
| +// region (QueueStagingPark's chore, index D6). The source cannot read or |
| +// park the target's moves/*, so the lane is the target region's to retire; |
| +// execution stays with the source, which serves throughout. |
| +// |
| +// The scope is moves/<host>/<op> with the holder's incarnation and the |
| +// journalled abort bound the caller computed (moves.AbortScope, ruling K5). |
| +// A replay finds the op already aborting: it is a no-op when the same lane |
| +// park is owed, or when none is (the target already acknowledged it), and |
| +// is refused when a lane park with another scope is owed. It allocates no |
| +// ids. A cut past requested cannot abort, and neither can an op whose |
| +// execution was given up (that is a revert). |
| +func Abort(opID, targetRegion string, scope ParkScope, newID func() string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := RequireOp(opID)(c); err != nil { |
| + return err |
| + } |
| + if c.Op.Kind != KindMoving || (c.Op.Phase != PhaseMoving && c.Op.Phase != PhaseAborting) || c.Op.Exec == "" { |
| + return fmt.Errorf("%w: op is %s/%q exec %q, an abort needs moving before the seal with the source executing", ErrWrongState, c.Op.Kind, c.Op.Phase, c.Op.Exec) |
| + } |
| + if targetRegion == "" || c.Op.Intent["to"] != targetRegion { |
| + return fmt.Errorf("%w: %q is not the move's target region", ErrWrongState, targetRegion) |
| + } |
| + var err error |
| + if scope, err = parkQueueGuard(c, opID, targetRegion, scope, newID); err != nil { |
| + return err |
| + } |
| + // Restates moves.ScopeLocation(host, op), as QueueStagingPark does. |
| + if scope.Location != "moves/"+c.Host+"/"+opID { |
| + return fmt.Errorf("%w: an abort parks this move's own lane, moves/%s/%s", ErrMalformed, c.Host, opID) |
| + } |
| + if c.Op.Phase == PhaseAborting { |
| + for _, o := range c.Obligations { |
| + if o.Op == opID && o.Kind == "park-replica" && o.Region == targetRegion && o.Park != nil && isStagingLocation(o.Park.Location) && !sameParkScope(*o.Park, scope) { |
| + return fmt.Errorf("%w: %s already owes its lane park in %s with another scope", ErrWrongState, opID, targetRegion) |
| + } |
| } |
| + return nil |
| + } |
| + if _, err := queueReplicaPark(c, opID, targetRegion, scope, newID); err != nil { |
| + return err |
| } |
| - c.Op = Op{Kind: KindServing, Exec: execBox, Journal: map[string]string{journalReleasedMove: opID}} |
| + c.Op.Phase = PhaseAborting |
| return nil |
| } |
| } |
| +// ReleaseAbort is the abort's release (ruling K7): from moving/aborting |
| +// only, so always after Abort queued the lane park, with the named source |
| +// still executing, and only once no obligation under the op remains (the |
| +// target region acknowledged the lane park). The serving op keeps the |
| +// source as its executor and records the move it released. |
| +func ReleaseAbort(opID, sourceBox string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := RequireMoveOp(opID, KindMoving, PhaseAborting)(c); err != nil { |
| + return err |
| + } |
| + if sourceBox == "" { |
| + return fmt.Errorf("%w: a release names the executor", ErrMalformed) |
| + } |
| + if c.Op.Exec != sourceBox { |
| + return fmt.Errorf("%w: exec is %q, the abort's release names %q", ErrWrongState, c.Op.Exec, sourceBox) |
| + } |
| + return releaseToServing(c, opID, sourceBox) |
| + } |
| +} |
| + |
| // RequireReleasedMove is the replay rule for a crashed abort: nil when |
| // the claim is serving and its op records that move opID released it, |
| // else ErrStaleOp. A plain serving claim, or one released by another |
diff --git a/internal/hostclaim/move_abort_test.go b/internal/hostclaim/move_abort_test.go| new file mode 100644 |
| index 000000000..69eb51cc6 |
| --- /dev/null |
| +++ b/internal/hostclaim/move_abort_test.go |
| @@ -0,0 +1,188 @@ |
| +package hostclaim |
| + |
| +import ( |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// Abort (moves spec §7.11, Plan 5 Task 1): moving before the seal goes to |
| +// moving/aborting, and the lane's park-replica is queued in the same write, |
| +// owned by the target region and scoped to moves/<host>/<op>. Execution |
| +// stays with the source. A replay allocates no ids; a replay with another |
| +// scope is refused, not overwritten. |
| +func TestAbortQueuesTheTargetsLanePark(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + ids := 0 |
| + newID := func() string { ids++; return "id" + strings.Repeat("x", ids) } |
| + if err := Abort(moveA, "uk", lane, newID)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.Op.Kind != KindMoving || c.Op.Phase != PhaseAborting || c.Op.Exec != "ie-box" || c.Op.ID != moveA { |
| + t.Fatalf("op after the abort: %+v", c.Op) |
| + } |
| + if len(c.Obligations) != 1 { |
| + t.Fatalf("obligations: %+v", c.Obligations) |
| + } |
| + o := c.Obligations[0] |
| + if o.Op != moveA || o.Kind != "park-replica" || o.Region != "uk" || o.Owner != "region:uk" || o.Box != "" || o.Park == nil || !sameParkScope(*o.Park, lane) || o.Retirement == "" { |
| + t.Fatalf("the lane park: %+v", o) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatalf("the claim after the abort must validate: %v", err) |
| + } |
| + // A replay with the same scope is a no-op and allocates nothing. |
| + before, n := string(mustJSON(t, c)), ids |
| + if err := Abort(moveA, "uk", lane, newID)(c); err != nil || ids != n || string(mustJSON(t, c)) != before { |
| + t.Fatalf("replay: %v, ids %d to %d", err, n, ids) |
| + } |
| + // Another scope for the same lane is refused, not overwritten. |
| + later := lane |
| + later.EvidenceAt = lane.EvidenceAt.Add(time.Second) |
| + if err := Abort(moveA, "uk", later, newID)(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != before { |
| + t.Fatalf("replay with another scope: %v", err) |
| + } |
| + // Once the target acknowledged the park, a late replay queues nothing. |
| + if err := CompleteObligation(moveA, o.ID)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + n = ids |
| + if err := Abort(moveA, "uk", lane, newID)(c); err != nil || len(c.Obligations) != 0 || ids != n { |
| + t.Fatalf("replay after the acknowledgement: %v, obligations %+v", err, c.Obligations) |
| + } |
| +} |
| + |
| +// Abort is legal only on this move's moving op before the seal, with the |
| +// source still executing, for the move's own target region and its own |
| +// lane; every refusal leaves the claim as it was and allocates nothing. |
| +func TestAbortRefusals(t *testing.T) { |
| + for name, tc := range map[string]struct { |
| + edit func(c *Claim) |
| + op string |
| + region string |
| + loc func(c *Claim) string |
| + inc string |
| + zero bool |
| + nilID bool |
| + want error |
| + }{ |
| + "another op": {op: moveB, want: ErrStaleOp}, |
| + "sealed": {edit: func(c *Claim) { c.Op.Phase = PhaseSealed }, want: ErrWrongState}, |
| + "execution given up": {edit: func(c *Claim) { c.Op.Exec = "" }, want: ErrWrongState}, |
| + "a landing": {edit: func(c *Claim) { c.Op.Kind = KindLanding }, want: ErrWrongState}, |
| + "a revert": {edit: func(c *Claim) { c.Op.Kind = KindReverting }, want: ErrWrongState}, |
| + "another region": {region: "se", want: ErrWrongState}, |
| + "no region": {region: "-", want: ErrWrongState}, |
| + "the live location": {loc: func(c *Claim) string { return "instances/" + c.Host + "/" + c.Holder.Incarnation }, want: ErrMalformed}, |
| + "the staging prefix": {loc: func(c *Claim) string { return "moves/" + c.Host }, want: ErrMalformed}, |
| + "another move's lane": {loc: func(c *Claim) string { return "moves/" + c.Host + "/" + moveB }, want: ErrMalformed}, |
| + "another incarnation": {inc: strings.Repeat("a", 64), want: ErrMalformed}, |
| + "no bound": {zero: true, want: ErrMalformed}, |
| + "no way to mint ids": {nilID: true, want: ErrMalformed}, |
| + "a dirty claim (no move)": {edit: func(c *Claim) { c.Op = Op{Kind: KindServing, Exec: "ie-box"} }, want: ErrStaleOp}, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + if tc.edit != nil { |
| + tc.edit(c) |
| + } |
| + location := "moves/" + c.Host + "/" + moveA |
| + if tc.loc != nil { |
| + location = tc.loc(c) |
| + } |
| + s := scope(location) |
| + if tc.inc != "" { |
| + s.Incarnation = tc.inc |
| + } |
| + if tc.zero { |
| + s.EvidenceAt = time.Time{} |
| + } |
| + op, region := moveA, "uk" |
| + if tc.op != "" { |
| + op = tc.op |
| + } |
| + if tc.region == "-" { |
| + region = "" |
| + } else if tc.region != "" { |
| + region = tc.region |
| + } |
| + ids := 0 |
| + newID := func() string { ids++; return "x" } |
| + if tc.nilID { |
| + newID = nil |
| + } |
| + before := string(mustJSON(t, c)) |
| + err := Abort(op, region, s, newID)(c) |
| + if !errors.Is(err, tc.want) { |
| + t.Fatalf("err %v, want %v", err, tc.want) |
| + } |
| + if string(mustJSON(t, c)) != before || ids != 0 { |
| + t.Fatalf("a refused abort changed the claim or minted %d ids", ids) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// ReleaseAbort (ruling K7) is the abort's only release: from moving/aborting, |
| +// with the source still executing, once no chore of the op remains. The |
| +// serving op keeps the source as its executor and records the move it |
| +// released. ReleaseMove's moving/"" stays the door's, and it no longer |
| +// releases an aborting op. |
| +func TestReleaseAbortFollowsTheLanePark(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + // Before the abort's CAS there is nothing to release from. |
| + if err := ReleaseAbort(moveA, "ie-box")(c); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("release before the abort: %v", err) |
| + } |
| + if err := Abort(moveA, "uk", lane, counter())(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := string(mustJSON(t, c)) |
| + if err := ReleaseAbort(moveA, "ie-box")(c); !errors.Is(err, ErrPendingChore) || string(mustJSON(t, c)) != before { |
| + t.Fatalf("release with the lane park owed: %v", err) |
| + } |
| + if err := ReleaseMove(moveA, "ie-box")(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != before { |
| + t.Fatalf("ReleaseMove released an aborting op: %v", err) |
| + } |
| + if err := CompleteObligation(moveA, c.Obligations[0].ID)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, rel := range map[string]func(*Claim) error{ |
| + "another box": ReleaseAbort(moveA, "uk-box"), |
| + "no box": ReleaseAbort(moveA, ""), |
| + "another op": ReleaseAbort(moveB, "ie-box"), |
| + "ReleaseMove's": ReleaseMove(moveA, "ie-box"), |
| + } { |
| + mid := string(mustJSON(t, c)) |
| + if err := rel(c); err == nil || string(mustJSON(t, c)) != mid { |
| + t.Fatalf("%s: released (%v)", name, err) |
| + } |
| + } |
| + if err := ReleaseAbort(moveA, "ie-box")(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.Op.Kind != KindServing || c.Op.Exec != "ie-box" || c.Op.ID != "" || c.Op.Phase != "" || len(c.Op.Participants) != 0 || len(c.Op.Journal) != 1 { |
| + t.Fatalf("op after the release: %+v", c.Op) |
| + } |
| + if err := RequireReleasedMove(moveA)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatalf("the claim after the release must validate: %v", err) |
| + } |
| + // Execution moved away (it never does in an abort) refuses. |
| + d, scope := parkFixture(t) |
| + if err := Abort(moveA, "uk", scope("moves/"+d.Host+"/"+moveA), counter())(d); err != nil { |
| + t.Fatal(err) |
| + } |
| + d.Obligations = nil |
| + d.Op.Exec = "uk-box" |
| + if err := ReleaseAbort(moveA, "ie-box")(d); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("exec is not the source: %v", err) |
| + } |
| +} |
diff --git a/internal/hostclaim/move_test.go b/internal/hostclaim/move_test.go| index 465dff847..078a828e0 100644 |
| --- a/internal/hostclaim/move_test.go |
| +++ b/internal/hostclaim/move_test.go |
| @@ -147,7 +147,7 @@ func TestTouchAppendsOnceUnderTheOp(t *testing.T) { |
| func TestReleaseMoveFromMovingNeedsNoChores(t *testing.T) { |
| now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| - for _, phase := range []string{PhaseMoving, PhaseAborting} { |
| + for _, phase := range []string{PhaseMoving} { |
| c := movingTestClaim(t) |
| if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil { |
| t.Fatal(err) |
| @@ -176,6 +176,14 @@ func TestReleaseMoveFromMovingNeedsNoChores(t *testing.T) { |
| t.Fatalf("claim after release must validate: %v", err) |
| } |
| } |
| + // Not from an aborting move (ruling K7: that is ReleaseAbort's, after |
| + // the lane park the abort queued). |
| + a := movingTestClaim(t) |
| + _ = BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(a) |
| + a.Op.Phase = PhaseAborting |
| + if err := ReleaseMove(moveA, "ie-box")(a); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("release from aborting: %v", err) |
| + } |
| // Not from serving, and not from a sealed move. |
| s := movingTestClaim(t) |
| if err := ReleaseMove(moveA, "ie-box")(s); !errors.Is(err, ErrStaleOp) { |
| @@ -874,7 +882,7 @@ func TestLocationParkOnlyWhereTheRegionIsRetired(t *testing.T) { |
| "rolled-back target after the source": {KindRestoring, "claimed", "uk", "uk-box", true}, |
| "source before the cut": {KindMoving, PhaseMoving, "ie", "ie-box", false}, |
| "source sealed": {KindMoving, PhaseSealed, "ie", "ie-box", false}, |
| - "source in the interim abort": {KindMoving, PhaseAborting, "ie", "ie-box", false}, |
| + "source in the abort": {KindMoving, PhaseAborting, "ie", "ie-box", false}, |
| "source while the landing checks": {KindLanding, "checking", "ie", "ie-box", false}, |
| "source while the landing probes": {KindLanding, "probing", "ie", "ie-box", false}, |
| "source while reverting": {KindReverting, "evidence", "ie", "ie-box", false}, |
| @@ -1782,11 +1790,10 @@ func TestReleaseMoveWaitsForObligations(t *testing.T) { |
| states := map[string]struct { |
| kind, phase, exec string |
| }{ |
| - "committed landing": {KindLanding, PhaseCommitted, "uk-box"}, |
| - "opened rollback": {KindRestoring, PhaseOpened, "ie-box"}, |
| - "opened revert": {KindReverting, PhaseOpened, "ie-box"}, |
| - "interim abort": {KindMoving, PhaseAborting, "ie-box"}, |
| - "interim abort before cut": {KindMoving, PhaseMoving, "ie-box"}, |
| + "committed landing": {KindLanding, PhaseCommitted, "uk-box"}, |
| + "opened rollback": {KindRestoring, PhaseOpened, "ie-box"}, |
| + "opened revert": {KindReverting, PhaseOpened, "ie-box"}, |
| + "the door, before the seal": {KindMoving, PhaseMoving, "ie-box"}, |
| } |
| for name, st := range states { |
| c := targetWroteLanding(t) |
| @@ -1832,6 +1839,7 @@ func TestReleaseMoveWaitsForObligations(t *testing.T) { |
| "reverting probed": {KindReverting, PhaseProbed}, |
| "reverting, rev only": {KindReverting, PhaseSealed}, |
| "moving sealed": {KindMoving, PhaseSealed}, |
| + "moving aborting": {KindMoving, PhaseAborting}, |
| } { |
| c := targetWroteLanding(t) |
| c.Op.Kind, c.Op.Phase, c.Op.Exec = st.kind, st.phase, "ie-box" |
diff --git a/internal/moves/abort.go b/internal/moves/abort.go| new file mode 100644 |
| index 000000000..60c136e20 |
| --- /dev/null |
| +++ b/internal/moves/abort.go |
| @@ -0,0 +1,39 @@ |
| +package moves |
| + |
| +import ( |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// AbortAfter is how long a Phase 1 move may stay preparing before it aborts |
| +// itself (spec §7.11: "a move whose readiness never comes aborts itself |
| +// after a bound, so an unreachable target never needs a human"; Plan 5 |
| +// ruling K18). §13 pages an operator at 60 minutes with no promotable round; |
| +// the abort comes at six hours. The console's sweep (Plan 5 Task 7) CASes |
| +// provision.MoveGuardedAbort when AbortDue says so; nothing here writes. |
| +const AbortAfter = 6 * time.Hour |
| + |
| +// AbortDue reports whether rec's move is due its automatic abort at now: a |
| +// Phase 1 move (not legacy) still preparing, AbortAfter or more after the |
| +// later of its RequestedAt and its ReadyAt. |
| +// |
| +// The bound is measured from max(RequestedAt, ReadyAt) (ruling K18), not |
| +// from Move.UpdatedAt, which every record write stamps and so says nothing |
| +// about readiness. Today a preparing move has no ReadyAt (nothing demotes a |
| +// ready move to preparing), so the bound runs from the request; taking the |
| +// later of the two keeps it right should a demotion ever stamp one. A ready |
| +// move is not due: readiness came, and the cut is the operator's (or the |
| +// automatic cut's). A move with no request time is never due: there is no |
| +// evidence to measure from. |
| +func AbortDue(rec provision.Record, now time.Time) bool { |
| + m := rec.Move |
| + if m == nil || m.Legacy() || m.State != provision.MovePreparing || m.RequestedAt.IsZero() { |
| + return false |
| + } |
| + from := m.RequestedAt |
| + if m.ReadyAt != nil && m.ReadyAt.After(from) { |
| + from = *m.ReadyAt |
| + } |
| + return !now.Before(from.Add(AbortAfter)) |
| +} |
diff --git a/internal/moves/abort_test.go b/internal/moves/abort_test.go| new file mode 100644 |
| index 000000000..af8d45594 |
| --- /dev/null |
| +++ b/internal/moves/abort_test.go |
| @@ -0,0 +1,46 @@ |
| +package moves |
| + |
| +import ( |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The auto-abort bound (spec §7.11, ruling K18): a Phase 1 move still |
| +// preparing AbortAfter after max(RequestedAt, ReadyAt) is due; one that is |
| +// not yet there, or is in any other state, or is a legacy move, is not. |
| +func TestAbortDue(t *testing.T) { |
| + if AbortAfter != 6*time.Hour { |
| + t.Fatalf("AbortAfter = %v, want 6h", AbortAfter) |
| + } |
| + req := time.Date(2026, 10, 9, 8, 0, 0, 0, time.UTC) |
| + rec := func(state string, ready *time.Time) provision.Record { |
| + return provision.Record{Host: "a.go.example.com", Move: &provision.MoveState{ID: "mv_abcdefghijklmnopqrstuvwxyz", State: state, RequestedAt: req, ReadyAt: ready}} |
| + } |
| + at := func(d time.Duration) time.Time { return req.Add(d) } |
| + later := req.Add(2 * time.Hour) |
| + for name, c := range map[string]struct { |
| + rec provision.Record |
| + now time.Time |
| + want bool |
| + }{ |
| + "preparing, just short": {rec(provision.MovePreparing, nil), at(AbortAfter - time.Second), false}, |
| + "preparing, at the bound": {rec(provision.MovePreparing, nil), at(AbortAfter), true}, |
| + "preparing, well past": {rec(provision.MovePreparing, nil), at(3 * AbortAfter), true}, |
| + "the bound is from the later time": {rec(provision.MovePreparing, &later), at(AbortAfter + time.Hour), false}, |
| + "and reached from it": {rec(provision.MovePreparing, &later), later.Add(AbortAfter), true}, |
| + "a non-UTC clock": {rec(provision.MovePreparing, nil), at(AbortAfter).In(time.FixedZone("x", -7*3600)), true}, |
| + "ready": {rec(provision.MoveReady, &later), at(10 * AbortAfter), false}, |
| + "cutting": {rec(provision.MoveCutting, &later), at(10 * AbortAfter), false}, |
| + "aborting": {rec(provision.MoveAborting, nil), at(10 * AbortAfter), false}, |
| + "landing": {rec(provision.MoveLanding, nil), at(10 * AbortAfter), false}, |
| + "legacy": {provision.Record{Move: &provision.MoveState{State: provision.MovePreparing, RequestedAt: req}}, at(10 * AbortAfter), false}, |
| + "no move": {provision.Record{}, at(10 * AbortAfter), false}, |
| + "no request time": {provision.Record{Move: &provision.MoveState{ID: "mv_abcdefghijklmnopqrstuvwxyz", State: provision.MovePreparing}}, at(10 * AbortAfter), false}, |
| + } { |
| + if got := AbortDue(c.rec, c.now); got != c.want { |
| + t.Errorf("%s: AbortDue = %v, want %v", name, got, c.want) |
| + } |
| + } |
| +} |
diff --git a/internal/provision/move_test.go b/internal/provision/move_test.go| index 4a692bd50..b28f67c22 100644 |
| --- a/internal/provision/move_test.go |
| +++ b/internal/provision/move_test.go |
| @@ -1621,3 +1621,22 @@ func TestRevertInputsRecordedBeforeTheRender(t *testing.T) { |
| t.Fatalf("recovery_inputs missing: %s", raw) |
| } |
| } |
| + |
| +// The legacy abort completion is the legacy record's only (Plan 5 Task 1): |
| +// it refuses a per-op move, which completes through MoveGuardedAbortComplete |
| +// once the target region has parked its lane. A legacy aborting record still |
| +// completes, unchanged. |
| +func TestLegacyAbortCompleteRefusesAPerOpMove(t *testing.T) { |
| + g := MoveGuard{ID: "mv_abcdefghijklmnopqrstuvwxyz", Incarnation: strings.Repeat("ab", 32)} |
| + rec := &Record{Move: &MoveState{ID: g.ID, Incarnation: g.Incarnation, State: MoveAborting}} |
| + if err := MoveAbortComplete()(rec); !errors.Is(err, ErrMoveConflict) || rec.Move == nil { |
| + t.Fatalf("legacy completion of a per-op abort: %v %+v", err, rec.Move) |
| + } |
| + if err := MoveGuardedAbortComplete(g, time.Now())(rec); err != nil || rec.Move != nil { |
| + t.Fatalf("guarded completion: %v %+v", err, rec.Move) |
| + } |
| + legacy := &Record{Move: &MoveState{To: "uk", State: MoveAborting}} |
| + if err := MoveAbortComplete()(legacy); err != nil || legacy.Move != nil { |
| + t.Fatalf("legacy completion of a legacy abort: %v %+v", err, legacy.Move) |
| + } |
| +} |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index 50ceb077a..fa42f8606 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -627,6 +627,7 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"internal/host/fleetbind.go", "WriteFleetBinding", "nonrecord", "sel:Put", "44ab984c2c9b30abcc938132118f61a6922ccc0cd40ed2b6cb61c7654aa0a1c3"}, |
| {"internal/host/logship.go", "logshipFlush", "nonrecord", "sel:Put", "a2fa594c1206028af170a2d0cbedef55ba00a2f1c698a6853e1214cc3b172dd9"}, |
| {"internal/host/metrics_loop.go", "putAlertState", "nonrecord", "sel:Put", "d5b6e8b34dc5a731d6bbafeac0b7d0c5c20706d6f413dff38088b3321b35a26c"}, |
| + {"internal/host/movesabort.go", "movesWorker.abortSource", "record-writer", "key:SetRecordMove", "5ba846213f439e5e55acac31ce7b7fe4f74810ed46834cea158d9d32d8ad7dbe"}, |
| {"internal/host/movesarchive.go", "cutRun.publishCutArchive", "nonrecord", "sel:CopyVersioned,sel:PutIfAbsent", "6c7fcfd2d4b2a0319ac15ecbfe003ed0292fe160457fd5704be508c2ab302a1b"}, |
| {"internal/host/movescut.go", "cutRun.cutTransition", "record-writer", "key:SetRecordMove", "d0d67399aa784036de43b3c143e88c3d85f8b3d9e6345f84908d3b58ff56191e"}, |
| {"internal/host/movescut.go", "publishStoppedReport", "nonrecord", "sel:Put", "c710682f31d99127ad76524add246fda9f640c63934940e45c79ea25ba36eb3c"}, |
| @@ -645,7 +646,6 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"internal/host/movesphase1.go", "movesWorker.refreshCrossRate", "nonrecord", "sel:CopyVersioned", "7e03e680bd61be26dddcf168cacd2427bb535571dfd326c4a3130ebe65ea1968"}, |
| {"internal/host/movesphase1.go", "movesWorker.writeSourceReport", "nonrecord", "sel:PutReplace", "370407cca2c1b91f6e5e7f135238f04c0dce9d7986e6e1ee9bebbf88cad0b9e0"}, |
| {"internal/host/movesphase1.go", "movesWorker.decideReady", "record-writer", "key:SetRecordMove", "c228e6f38eca6ed10501538031d3260f016a3f26ce501d8544f1581ae7668d8f"}, |
| - {"internal/host/movesphase1.go", "movesWorker.interimAbort", "record-writer", "key:SetRecordMove", "9e875647dfdd2531c6857a4730245526c90c000986b2c9ea655bfcc4df43c02e"}, |
| {"internal/host/movesreport.go", "movesWorker.targetReport", "nonrecord", "sel:PutReplace", "3f6aec125f1bd40271adf78833ac3268e536cbc5e8af3731b16f382669cd955b"}, |
| {"internal/host/movesstore.go", "laneProbe", "nonrecord", "sel:PutIfAbsent", "f4a17a1143e7370b9039b2e66e736606015c46619bab92c553ce7b1ad6b7918b"}, |
| {"internal/host/snapshot_loop.go", "registrySnapshotPass", "nonrecord", "sel:Put", "6b01b46b361c3a2d1d432659a0d234c01975849b9a8ac5571858fdd78d376849"}, |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index cda7ff114..0154b3918 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -2416,13 +2416,19 @@ func MoveAbort() func(*Record) error { |
| } |
| } |
| -// MoveAbortComplete is the source box's half of an abort: staging is gone, |
| -// the move field clears. |
| +// MoveAbortComplete is the source box's half of a legacy abort: staging is |
| +// gone, the move field clears. It refuses a per-op move (Plan 5 Task 1): that |
| +// completes through MoveGuardedAbortComplete once the target region has |
| +// parked its lane and the claim is released, and its source never deletes |
| +// the target's staging. |
| func MoveAbortComplete() func(*Record) error { |
| return func(rec *Record) error { |
| if rec.Move == nil || rec.Move.State != MoveAborting { |
| return fmt.Errorf("%w: abort-complete is only legal from aborting", ErrMoveConflict) |
| } |
| + if !rec.Move.Legacy() { |
| + return fmt.Errorf("%w: a move with an id completes its abort through the guarded transition", ErrMoveConflict) |
| + } |
| rec.Move = nil |
| return nil |
| } |