amadan

carlos / platform Public

moves-phase1-3b

Squashed from moves-phase1-3b into main.
claude 4a51a6daac3582709ac2af89bd9071eb0e7375c3 parent 81d0cc6
12 files changed, +2597 −7
  • approved_moves_copy_test.go +4 −1
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +3 −0
  • docs/superpowers/specs/2026-10-02-move-any-app-design.md +9 −5
  • internal/host/movesfinal.go +724 −0
  • internal/host/movesfinal_test.go +998 −0
  • internal/host/movespass.go +31 −0
  • internal/host/movesphase1_harness_test.go +3 −1
  • internal/host/movesretire.go +337 −0
  • internal/moves/startcheck.go +162 −0
  • internal/moves/startcheck_test.go +232 −0
  • internal/registry/hold_test.go +64 −0
  • internal/registry/registry.go +30 −0

CI passed — run details

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index ef0477bd0..bb7ec9b31 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -70,7 +70,7 @@ var movesCopyIDs = []string{
"MV-N163", "MV-N164", "MV-N165", "MV-N166", "MV-N167", "MV-N168",
"MV-N169", "MV-N170", "MV-N171", "MV-N172", "MV-N173", "MV-N174",
"MV-N175", "MV-N176", "MV-N177", "MV-N178", "MV-N179", "MV-N180", "MV-N181",
- "MV-N182", "MV-N183",
+ "MV-N182", "MV-N183", "MV-N184", "MV-N185", "MV-N186",
}
// copyRow is one row of the record. The File cell is the file the sentence is
@@ -570,6 +570,7 @@ var movesScanFiles = []string{
"internal/host/movesfloor_other.go",
"internal/host/movesarchive.go",
"internal/host/movescut.go",
+ "internal/host/movesfinal.go",
"internal/host/movesguard.go",
"internal/host/movesheldenv.go",
"internal/host/movesheldlock.go",
@@ -585,6 +586,7 @@ var movesScanFiles = []string{
"internal/host/movespolicy.go",
"internal/host/movesprobe.go",
"internal/host/movesreport.go",
+ "internal/host/movesretire.go",
"internal/host/movesstage.go",
"internal/host/movesstore.go",
"internal/host/refusalwatch.go",
@@ -627,6 +629,7 @@ var movesScanFiles = []string{
"internal/moves/sealedclient.go",
"internal/moves/source.go",
"internal/moves/staging.go",
+ "internal/moves/startcheck.go",
"internal/moves/switch.go",
"internal/moves/timers.go",
"internal/provision/provision.go",
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 40607c9e1..0baaab1e0 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -237,6 +237,9 @@ Columns: id, file, text, status.
|---|---|---|---|
| MV-N182 | internal/provision/provision.go | the probe ran out of attempts | pending |
| MV-N183 | internal/provision/provision.go | the commit came after the deadline | pending |
+| MV-N184 | internal/moves/startcheck.go | this box holds another generation of the instance | pending |
+| MV-N185 | internal/moves/startcheck.go | the instance's files belong to another generation | pending |
+| MV-N186 | internal/moves/startcheck.go | the instance's database belongs to another generation | pending |
## Plan 4
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 36c92b479..26471345f 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
@@ -1838,7 +1838,7 @@ field.
| marker | where | written | content |
|---|---|---|---|
| `generation.json` | `<location>generation.json`, beside `lock.json` | **last**, after the install manifest is complete and verified (§7.6); **conditionally**: If-Match the marker it supersedes, or If-None-Match when there is none (model 29) | `{"op","incarnation","host","box","installed_at","manifest_sha256","snapshot_key","snapshot_version","snapshot_txid","snapshot_crc64"}`: the pinned root, the install's snapshot identity and the floor a successor snapshot must clear (R3-1, R4-1, R5-3) |
-| `<dbPath>.gen` | beside the local database | after a restore completes | `{"op","incarnation","location"}` |
+| `<dbPath>.gen` | beside the local database | after a restore completes | the generation's name as plain text, one line: the format main already writes (`hostbox.WriteGenerationMarker`). It carries no incarnation and no location; the claim's holder vouches for both (Paul's ruling, moves Phase 1 Plan 3) |
| `.files-gen` | inside the files directory | **last**, after files, `<host>.env`, timers and the carried registry columns are all durable (§7.6) | `{"op","incarnation"}` |
`generation.json` is the one reused key a generation writes: everything
@@ -1888,10 +1888,14 @@ is the region of the holder's current installation:
generation for every probe, R10-3); a lagging marker is read through
`holder.admissions` (§5.1 step 9); and its incarnation to be the
holder's;
-2. `.files-gen`, when the files directory exists, to name the same
- generation and incarnation;
-3. `<dbPath>.gen` to name the same generation and location. A warm
- database that does not is from an earlier stay on this box: the
+2. `.files-gen`, when the files directory exists, to name the holder's
+ incarnation and either `holder.generation` or its cold predecessor
+ (the newest cold admission into `holder.generation`, warm self-links
+ skipped), which is the prepared generation a restarted probe was restored
+ from; no older generation counts;
+3. `<dbPath>.gen` to name the same generation, exactly: the plain name
+ carries no location, which the claim's `holder.location` vouches for.
+ A warm database that does not is from an earlier stay on this box: the
activator parks it (§6.3) and restores cold.
Rev 5's check compared the three markers with each other only. Astra
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
new file mode 100644
index 000000000..e02647890
--- /dev/null
+++ b/internal/host/movesfinal.go
@@ -0,0 +1,724 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The source's finalization after the commit (moves spec §7.9 step 3; Plan 3
+// Task 8 with rulings H5, H7, H19 and H24), behind CARLOS_MOVES_PHASE1. Once
+// the record is committed the target's region is authoritative, and the
+// source box has four things left to do for the move, in this order:
+//
+// 1. the remote half: the source region's location park (the park-replica
+// the commit queued for instances/<host>, D6, H24) is run through the
+// park engine over this box's own bucket until its archive is complete
+// (park.Engine.Parked). Its acknowledgement after the horizon is the
+// engine's or the regional runner's (C4), never a reason to wait here.
+// The horizon starts at the flip (H7), which is after the source's last
+// write to its location: a park whose EvidenceAt is earlier is refused;
+// 2. the local half (ruling H5(b), Paul): a source whose installation is
+// registered (the commit queued its park-local, because the inventory
+// lists paths of the journalled source stamp S) retires it through
+// park.Local.RetireInstallation under the host guard, which completes the
+// chore itself (H19). Everything else the source kept at the live flat
+// paths, registered or not, is parked by a moves-owned journalled rename
+// into the engine's layout, .parked/<incarnation>/<host>/<retirement>/
+// artifacts/ (parkLocalArtifacts): the retirement id and every artifact's
+// identity are journalled before the first rename, and nothing the plan
+// did not record is ever moved or removed;
+// 3. the row and its timers rows go in one registry transaction, only while
+// the move holds the row (registry.DeleteIfHoldWithTimers, H5(d)), under
+// provision.WithHostHoldLock with the record re-read there; the journal
+// says row-deleted "pending" before it and "done" after it, and a replay
+// that finds "pending" and no row of this op's treats the delete as done
+// (round-2 finding 12). The box's config ack is copied into the move's
+// report prefix as config-ack-source.json and deleted (§6.3);
+// 4. MoveSourceParked stamps source_parked_at, and the serial is nudged.
+// Last of all the op's own journal directory is renamed into the same
+// retirement, so a finished source keeps nothing under .moves/.
+//
+// Routing (ruling H5(a)): after the flip the record's region is the target's
+// and after step 3 this box has no row, so the region-and-row rule that
+// admits every other Phase 1 round would never bring the replay of a crash
+// between steps 3 and 4 back here. movesParticipantRecord is the rule for the
+// states after the flip: this box's region is the move's From (or To), and it
+// still has the row or the op's journal. Plan 4's rollback visits reuse it.
+//
+// Arming. Every hook that exists on main is wired (the guard, park.Local, the
+// registry's delete, the versioned store), but the park engine's leases and
+// Plan 5's view and authorizer are not (ruling G17, as for the landing), so
+// sourceFinalArmed answers false in production and the pass only logs. The
+// tests arm it with fakes and the stand-ins below. No production record
+// reaches committed before Task 7 and Group D anyway.
+
+// The finalization's op journal steps (moves.Journal, the source's own
+// journal of this op, which the cut began).
+const (
+ finalStepRemote = "source-park-remote" // the location park's archive is complete, or acknowledged
+ finalStepInstallation = "source-park-installation" // retired, completed or unregistered
+ finalStepLocal = "source-park-local" // parkLocalArtifacts' step: the retirement id
+ finalStepRowDeleted = "row-deleted" // "pending" before the delete, "done" after it
+ finalStepConfigAck = "config-ack-source" // the config ack is copied and deleted
+ finalStepParked = "source-parked" // "pending" before the stamp's CAS, "done" after it
+ finalStepJournal = "source-park-journal" // where the journal is going, written just before it goes
+)
+
+// finalConfigAckName is the config ack's copy under the move's report prefix.
+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.
+var movesParticipantStates = map[string]bool{
+ provision.MoveCommitted: true,
+ provision.MoveRollingBack: true,
+ provision.MoveReverting: true,
+}
+
+// movesParticipantRecord is the routing rule for the states after the flip
+// (ruling H5(a), shared with Plan 4's J10 and J13): it answers this box's role
+// in rec's move, roleSource when this box's region is the move's From,
+// roleTarget when it is the move's To, and roleNone otherwise. A role is
+// answered only while the box still has something of the move's: its row
+// (OwnsRoute) or the op's journal directory (moveJournalPresent). The caller
+// decides what a role does in a state; this only says the box is a
+// participant with work it can find.
+func movesParticipantRecord(cfg moveCfg, rec provision.Record) string {
+ m := rec.Move
+ if m == nil || m.Legacy() || !movesParticipantStates[m.State] || m.From == m.To {
+ return roleNone
+ }
+ role := roleNone
+ switch cfg.Region {
+ case m.From:
+ role = roleSource
+ case m.To:
+ role = roleTarget
+ default:
+ return roleNone
+ }
+ if cfg.OwnsRoute == nil || cfg.OwnsRoute(rec.Host) || moveJournalPresent(cfg.DataDir, rec.Host, m.ID) {
+ return role
+ }
+ return roleNone
+}
+
+// moveJournalPresent reports whether this box holds the op's journal
+// directory, <dataDir>/.moves/<host>/<op>/. A name that could leave its
+// directory is never present.
+func moveJournalPresent(dataDir, host, op string) bool {
+ if dataDir == "" || !journalComponent(host) || !journalComponent(op) {
+ return false
+ }
+ fi, err := os.Lstat(moves.JournalDir(dataDir, host, op))
+ return err == nil && fi.IsDir()
+}
+
+// participantRound is a post-flip participant's step for a record the
+// region rule did not admit (movesPass). Only the source's finalization of a
+// committed move is built; the rest is Plan 4's.
+func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec provision.Record, role string) error {
+ if role == roleSource && rec.Move.State == provision.MoveCommitted {
+ return w.finalizeSource(ctx, s, rec)
+ }
+ return nil
+}
+
+// sourceFinalArmed reports whether this box can run the finalization: every
+// hook it calls is wired, the park engine's leases and Plan 5's view and
+// authorizer included. Nothing stands in for a missing owner in production
+// (fix round 1): the tests inject their own. In production none of the
+// three is wired yet (ruling G17).
+func sourceFinalArmed(cfg moveCfg) bool {
+ return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" &&
+ cfg.ParkLeases != nil && cfg.ParkView != nil && cfg.ParkAuthorize != nil &&
+ cfg.Versioned != nil && cfg.Route != nil && cfg.DeleteHeld != nil
+}
+
+// sourceLocationParkScope is the source region's location park the commit
+// queues (Task 7 builds the commit; Task 8 reads it): holder.location, the
+// holder's incarnation, the cut's request limit, and a horizon that starts at
+// the flip (ruling H7). The flip comes after ExecToNone, after which the
+// source never writes its location again; the stop evidence's CheckedAt does
+// not, because the final checkpoint writes after it.
+func sourceLocationParkScope(rec provision.Record, holder hostclaim.Holder) (hostclaim.ParkScope, error) {
+ if rec.Move == nil || rec.Move.FlippedAt == nil {
+ return hostclaim.ParkScope{}, errors.New("the source's location park needs the flip's time")
+ }
+ return hostclaim.ParkScope{Location: strings.TrimSuffix(holder.Location, "/"), Incarnation: holder.Incarnation,
+ EvidenceAt: rec.Move.FlippedAt.UTC(), RequestLimit: cutRequestLimit}, nil
+}
+
+// errFinalChanged is the record or claim no longer being this move's commit
+// when a step re-reads it.
+var errFinalChanged = errors.New("the record is no longer this move's commit")
+
+// sourceFinal is one pass's state for one committed record on its source.
+type sourceFinal struct {
+ w *movesWorker
+ s release.Store
+ rec provision.Record
+ g provision.MoveGuard
+ j *moves.Journal
+}
+
+// finalizeSource is the source's §7.9 step 3 for a committed record. It runs
+// only when the record is committed and either this box's row is held by the
+// move or the journal has row-deleted (Review Focus 5, round-2 finding 12):
+// a landing record, a row held by another op, or no row and no journalled
+// delete leaves everything as it is.
+func (w *movesWorker) finalizeSource(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m := rec.Move
+ if m == nil || m.Legacy() || m.State != provision.MoveCommitted || m.From != cfg.Region || m.To == cfg.Region {
+ return nil
+ }
+ if !sourceFinalArmed(cfg) {
+ w.notice(rec.Host+"#final", "moves: %s: the source's finalization of %s waits: this box's finalization is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.ID)
+ return nil
+ }
+ f := &sourceFinal{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}}
+ return f.run(ctx)
+}
+
+func (f *sourceFinal) cfg() moveCfg { return f.w.cfg }
+
+func (f *sourceFinal) now() time.Time {
+ if n := f.cfg().Now; n != nil {
+ return n()
+ }
+ return time.Now()
+}
+
+// fault is the test seam after a side effect.
+func (f *sourceFinal) fault(point string) error {
+ if fn := f.cfg().finalFault; fn != nil {
+ return fn(point)
+ }
+ return nil
+}
+
+// mark journals a step durably, then passes its fault point.
+func (f *sourceFinal) mark(step, value string) error {
+ f.j.Mark(step, value)
+ if err := f.j.Write(f.cfg().DataDir); err != nil {
+ return err
+ }
+ return f.fault("after-" + step)
+}
+
+func (f *sourceFinal) run(ctx context.Context) error {
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ j, err := moves.ReadJournal(cfg.DataDir, host, id)
+ if err != nil {
+ return err
+ }
+ f.j = j
+ rt, err := cfg.Route(host)
+ if err != nil {
+ return fmt.Errorf("read this box's row: %w", err)
+ }
+ held := rt != nil && rt.HoldOp == id
+ if !held && !j.Done(finalStepRowDeleted) {
+ f.w.notice(host+"#final-row", "moves: %s: the source's finalization of %s waits: this box's row is not held by the move", host, id)
+ return nil
+ }
+ if held && rt.DBPath != "" && rt.DBPath != provision.DefaultDBPath(cfg.DataDir, host) {
+ return fmt.Errorf("move %s: the source's row is not at the flat path (%s)", id, rt.DBPath)
+ }
+ // source_parked_at is stamped only after every step above it, so once it
+ // is only the journal's own park is left, and the claim may already
+ // have moved on to the close.
+ if f.rec.Move.SourceParkedAt != nil {
+ if j.Steps[finalStepParked] != "done" {
+ if err := f.mark(finalStepParked, "done"); err != nil {
+ return err
+ }
+ }
+ return f.parkJournal()
+ }
+ obs, err := hostclaim.Read(ctx, f.s, host)
+ if err != nil {
+ return fmt.Errorf("read the host claim: %w", err)
+ }
+ if err := f.sourceOf(obs); err != nil {
+ f.w.notice(host+"#final-claim", "moves: %s: the source's finalization of %s waits: %v", host, id, err)
+ return nil
+ }
+ holder := *obs.Claim.Holder
+ if !j.Done(finalStepRemote) {
+ done, err := f.remote(ctx, obs)
+ if err != nil || !done {
+ return err
+ }
+ if err := f.mark(finalStepRemote, ""); err != nil {
+ return err
+ }
+ }
+ if !j.Done(finalStepInstallation) {
+ how, err := f.installation(ctx)
+ if err != nil {
+ return err
+ }
+ if err := f.mark(finalStepInstallation, how); err != nil {
+ return err
+ }
+ }
+ if _, err := parkLocalArtifacts(ctx, j, f.mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: holder.Incarnation,
+ Host: host, Op: id, Step: finalStepLocal, Paths: sourceArtifactPaths(host), NewID: obligationID, Fault: f.fault}); err != nil {
+ return fmt.Errorf("park the source's local artifacts: %w", err)
+ }
+ if err := f.deleteRow(ctx); err != nil {
+ return err
+ }
+ if !j.Done(finalStepConfigAck) {
+ if err := f.configAck(ctx); err != nil {
+ return err
+ }
+ if err := f.mark(finalStepConfigAck, ""); err != nil {
+ return err
+ }
+ }
+ if j.Steps[finalStepParked] != "done" {
+ // "pending" is the stamp step reached: from here the journal is
+ // the local sweep's to finish if the close overtakes this box
+ // (sweepFinalJournals).
+ if j.Steps[finalStepParked] != "pending" {
+ j.Mark(finalStepParked, "pending")
+ if err := j.Write(cfg.DataDir); err != nil {
+ return err
+ }
+ if err := f.fault("after-source-parked-pending"); err != nil {
+ return err
+ }
+ }
+ if err := f.stampParked(ctx); err != nil {
+ return err
+ }
+ if err := f.mark(finalStepParked, "done"); err != nil {
+ return err
+ }
+ }
+ return f.parkJournal()
+}
+
+// sourceOf is the claim this finalization runs under: this move's landing,
+// committed, this box its source with its evidence journalled, and the
+// record's incarnation.
+func (f *sourceFinal) sourceOf(obs hostclaim.Observed) error {
+ cfg := f.cfg()
+ if obs.Absent || obs.Claim == nil {
+ return fmt.Errorf("%w: no host claim", park.ErrUnsafe)
+ }
+ c := obs.Claim
+ if c.Op.ID != f.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted {
+ return fmt.Errorf("%w: the claim is %s/%q for %q, not this move's commit", park.ErrUnsafe, c.Op.Kind, c.Op.Phase, c.Op.ID)
+ }
+ if len(c.Op.Participants) != 2 || c.Op.Participants[0] != cfg.Box || c.Op.Intent["source"] != cfg.Box {
+ return fmt.Errorf("%w: this box is not the move's source in the claim", park.ErrUnsafe)
+ }
+ if c.Op.Journal[hostclaim.JournalEvidenceSource] == "" {
+ return fmt.Errorf("%w: the claim does not journal the source's evidence", park.ErrUnsafe)
+ }
+ if c.Holder == nil || c.Holder.Incarnation != f.g.Incarnation {
+ return fmt.Errorf("%w: the claim's holder is not this move's incarnation", park.ErrUnsafe)
+ }
+ return nil
+}
+
+// engine is the park engine over this box's own bucket, with the wired
+// leases, view and authorizer (sourceFinalArmed requires all three).
+func (f *sourceFinal) engine(vs lease.VersionedStore) park.Engine {
+ cfg := f.cfg()
+ return park.Engine{Store: vs, Claims: f.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
+ Now: f.now, NewID: obligationID, View: cfg.ParkView, Authorize: cfg.ParkAuthorize, After: cfg.parkFault}
+}
+
+// remote is step 1. done is true once the source region's location park has
+// its archive complete, or is no longer in the claim (acknowledged: the
+// commit queues it in its own write, so a committed claim without it has
+// had it acknowledged). The engine takes the location's lease itself.
+func (f *sourceFinal) remote(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ var chore *hostclaim.Obligation
+ for i := range obs.Claim.Obligations {
+ o := &obs.Claim.Obligations[i]
+ if o.Op == id && o.Kind == "park-replica" && o.Region == cfg.Region && o.Park != nil && !isStagingLocation(o.Park.Location) {
+ chore = o
+ }
+ }
+ if chore == nil {
+ return true, nil
+ }
+ if flipped := f.rec.Move.FlippedAt; flipped == nil || chore.Park.EvidenceAt.Before(*flipped) {
+ return false, fmt.Errorf("move %s: the source's location park starts its horizon at %s, before the flip (ruling H7)", id, chore.Park.EvidenceAt.Format(time.RFC3339))
+ }
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return false, err
+ }
+ e := f.engine(vs)
+ work := park.Work{Host: host, ID: chore.ID, Op: id}
+ if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) {
+ return false, fmt.Errorf("park the source's location: %w", err)
+ }
+ parked, err := e.Parked(ctx, work)
+ if errors.Is(err, park.ErrChanged) {
+ return true, nil
+ }
+ if err != nil {
+ return false, fmt.Errorf("the source's location park: %w", err)
+ }
+ if !parked {
+ f.w.notice(host+"#final-remote", "moves: %s: the source's finalization of %s waits for its location park's archive", host, id)
+ return false, nil
+ }
+ log.Printf("moves: %s: the source's location %s is parked", host, chore.Park.Location)
+ return true, nil
+}
+
+func isStagingLocation(location string) bool { return strings.HasPrefix(location, "moves/") }
+
+// installation is step 2's registered half (ruling H5(b)). With the box's
+// park-local and inventory paths of the journalled source stamp S, the
+// installation is retired under the host guard, which completes the chore; a
+// park-local with no such paths left is completed; no park-local is an
+// unregistered source, whose artifacts are all parkLocalArtifacts'. It
+// answers which, for the journal.
+func (f *sourceFinal) installation(ctx context.Context) (string, error) {
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ obs, err := hostclaim.Read(ctx, f.s, host)
+ if err != nil {
+ return "", fmt.Errorf("read the host claim: %w", err)
+ }
+ if err := f.sourceOf(obs); err != nil {
+ return "", fmt.Errorf("%w: %v", errFinalChanged, err)
+ }
+ c := obs.Claim
+ stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
+ if err != nil || stamp == 0 {
+ return "", fmt.Errorf("move %s: the claim journals no source stamp", id)
+ }
+ var chore *hostclaim.Obligation
+ for i := range c.Obligations {
+ if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
+ chore = o
+ }
+ }
+ registered := hostclaim.InventoryHasStamp(c, cfg.Box, stamp)
+ switch {
+ case chore != nil && registered:
+ captured, err := hostguard.Capture(cfg.DataDir, host)
+ if err != nil {
+ return "", fmt.Errorf("the source's installation record: %w", err)
+ }
+ // A retry after the guard retired the record and before the claim
+ // completed the chore captures stamp 0; the stamp is the claim's S.
+ if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != stamp) {
+ return "", fmt.Errorf("move %s: the source's installation record is at %s/%d, the claim's source is %s/%d", id, captured.Incarnation, captured.Stamp, c.Holder.Incarnation, stamp)
+ }
+ guard := cfg.Guard
+ if guard.Claims == nil {
+ guard.Claims = hostguard.StoreReader{Store: f.s}
+ }
+ local := cfg.Local
+ local.Guard, local.Claims = guard, f.s
+ if local.Box == "" {
+ local.Box = cfg.Box
+ }
+ if local.Authorize == nil {
+ // park.Local asks the same owner the engine does, over the claim
+ // it read under the guard.
+ authorize := cfg.ParkAuthorize
+ local.Authorize = func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
+ return authorize(ctx, hostclaim.Observed{Host: host, Claim: c}, o)
+ }
+ }
+ if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: stamp}, chore.ID, id); err != nil {
+ return "", fmt.Errorf("retire the source's installation: %w", err)
+ }
+ log.Printf("moves: %s: the source's registered installation (stamp %d) is retired", host, stamp)
+ return "retired", nil
+ case chore != nil:
+ if _, err := hostclaim.Update(ctx, f.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, stamp)); err != nil {
+ return "", fmt.Errorf("complete the source's local park: %w", err)
+ }
+ return "completed", nil
+ case registered:
+ return "", fmt.Errorf("move %s: this box's inventory lists the source's paths, and the commit queued no park-local for them", id)
+ }
+ return "unregistered", nil
+}
+
+// sourceArtifactPaths are the source's artifacts at the flat paths (G11),
+// relative to the data directory: the database with its marker, journal,
+// quarantine and litestream shadow, the files directory and <host>.env.
+func sourceArtifactPaths(host string) []string {
+ db := host + ".db"
+ return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", host, host + ".env",
+ host + ".config.env", host + "-sidecar.env"}
+}
+
+// deleteRow is step 3's row and timers delete, under the host's hold lock
+// with the record re-read there (the lock reconcile's hold decision holds).
+func (f *sourceFinal) deleteRow(ctx context.Context) error {
+ if f.j.Steps[finalStepRowDeleted] == "done" {
+ return nil
+ }
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ var err error
+ provision.WithHostHoldLock(host, func() {
+ fresh, e := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
+ if e != nil {
+ err = fmt.Errorf("re-read the record: %w", e)
+ return
+ }
+ if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != f.g.Incarnation || m.State != provision.MoveCommitted {
+ err = errFinalChanged
+ return
+ }
+ if f.j.Steps[finalStepRowDeleted] != "pending" {
+ f.j.Mark(finalStepRowDeleted, "pending")
+ if err = f.j.Write(cfg.DataDir); err != nil {
+ return
+ }
+ if err = f.fault("after-row-pending"); err != nil {
+ return
+ }
+ }
+ // false: no row is held by this op. With "pending" journalled that is
+ // this delete, already done; a row another op holds was never this
+ // move's, and is left as it is.
+ deleted, e := cfg.DeleteHeld(host, id)
+ if e != nil {
+ err = fmt.Errorf("delete the source's row: %w", e)
+ return
+ }
+ if deleted {
+ log.Printf("moves: %s: the source's row and its timers are deleted", host)
+ }
+ if err = f.fault("after-row-delete"); err != nil {
+ return
+ }
+ err = f.mark(finalStepRowDeleted, "done")
+ })
+ return err
+}
+
+// configAck copies this box's config ack for the host into the move's report
+// prefix and deletes it (§6.3). Absent is nothing to copy, or a replay after
+// the delete.
+func (f *sourceFinal) configAck(ctx context.Context) error {
+ cfg := f.cfg()
+ key := appconfig.ConfigAckKey(cfg.Box, f.rec.Host)
+ body, _, err := f.s.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("read the config ack: %w", err)
+ }
+ // The copy is create-only. A replay after the copy finds it: the same
+ // body is this copy; another (the ack was rewritten in between) is
+ // replaced under its etag, so the copy is the last ack the box wrote.
+ dst := moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.g.ID) + finalConfigAckName
+ if _, err := f.s.Put(ctx, dst, body, ""); errors.Is(err, release.ErrCASConflict) {
+ had, etag, gerr := f.s.Get(ctx, dst)
+ if gerr != nil {
+ return fmt.Errorf("read the config ack's copy: %w", gerr)
+ }
+ if string(had) != string(body) {
+ if _, err := f.s.Put(ctx, dst, body, etag); err != nil {
+ return fmt.Errorf("copy the config ack: %w", err)
+ }
+ }
+ } else if err != nil {
+ return fmt.Errorf("copy the config ack: %w", err)
+ }
+ if err := f.fault("after-config-ack-copy"); err != nil {
+ return err
+ }
+ if err := f.s.Delete(ctx, key); err != nil {
+ return fmt.Errorf("delete the config ack: %w", err)
+ }
+ return nil
+}
+
+// stampParked is step 4: MoveSourceParked, then the nudge. A record that has
+// moved past this commit (closed, or another move) had it stamped, since the
+// close needs it; this commit without the stamp is an error.
+func (f *sourceFinal) stampParked(ctx context.Context) error {
+ host := f.rec.Host
+ _, err := provision.SetRecordMove(ctx, f.s, f.rec.Account, f.rec.App, host, provision.MoveSourceParked(f.g, f.now()))
+ if errors.Is(err, provision.ErrMoveConflict) {
+ fresh, gerr := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
+ if gerr != nil {
+ return fmt.Errorf("re-read the record: %w", gerr)
+ }
+ if m := fresh.Move; m != nil && m.ID == f.g.ID && m.SourceParkedAt == nil {
+ return fmt.Errorf("stamp source_parked_at: %w", err)
+ }
+ err = nil
+ }
+ if err != nil {
+ return fmt.Errorf("stamp source_parked_at: %w", err)
+ }
+ if err := f.fault("after-source-parked-record"); err != nil {
+ return err
+ }
+ nudgeSerial(ctx, f.s)
+ log.Printf("moves: %s: the source has parked what it kept for %s", host, f.g.ID)
+ return nil
+}
+
+// parkJournal renames the op's journal directory into the local retirement,
+// after source_parked_at is stamped: until then it is what brings a replay
+// back here. A crash before the rename leaves the journal, and the next pass
+// (or the local sweep) renames it; after it nothing of the move is left
+// under .moves/.
+func (f *sourceFinal) parkJournal() error {
+ return parkFinalJournal(f.cfg().DataDir, f.rec.Host, f.g.ID, f.j, f.mark)
+}
+
+// parkFinalJournal parks a finalized op's journal into the retirement its
+// own local park recorded, journalling the destination just before it goes.
+func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step, value string) error) error {
+ retirement := j.Steps[finalStepLocal]
+ if retirement == "" {
+ return fmt.Errorf("move %s: the journal records no local retirement for its own park", op)
+ }
+ plan, err := journalledLocalPlan(j, finalStepLocal)
+ if err != nil {
+ return err
+ }
+ dst := filepath.Join(parkedArtifactsDir(dataDir, plan.Incarnation, host, retirement), ".moves", host, op)
+ if j.Steps[finalStepJournal] != dst {
+ if err := mark(finalStepJournal, dst); err != nil {
+ return err
+ }
+ }
+ if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil {
+ return fmt.Errorf("park the op journal: %w", err)
+ }
+ log.Printf("moves: %s: the source's finalization of %s is complete", host, op)
+ return nil
+}
+
+// sweepFinalJournals is the local sweep of this box's op journals (fix round
+// 1, ruling 1), independent of any active move: a crash after the stamp and
+// before the journal's park, then the close, leaves a journal no record
+// routes back here (movesParticipantRecord answers nothing for a record with
+// no move). Each journal under <DataDir>/.moves/<host>/<op>/ that has reached
+// the stamp step is finished here:
+// - the stamp journalled "done": the journal is parked;
+// - "pending", and the listed record still carries this op: with
+// source_parked_at set it is journalled done and parked; without, it is
+// left to the finalization's own pass;
+// - "pending", and the listed record no longer carries this op (closed, or
+// another move since): the stamp landed, because the only transition
+// that clears a committed move is MoveClose, which refuses unless
+// source_parked_at is set (the aborts end at cutting/requested); so the
+// stamp is journalled done and the journal parked;
+// - a host the listing does not name (its record deleted, or unreadable)
+// is left as it is.
+//
+// A journal that has not reached the stamp step (no source-parked step: a
+// landing's, a cut's, a finalization still under way) is never touched, and
+// nothing is ever deleted: a journal is only renamed into its retirement.
+func (w *movesWorker) sweepFinalJournals(recs []provision.Record) {
+ dd := w.cfg.DataDir
+ if dd == "" {
+ return
+ }
+ hosts, err := os.ReadDir(filepath.Join(dd, ".moves"))
+ if err != nil {
+ if !errors.Is(err, os.ErrNotExist) {
+ w.notice("#final-sweep", "moves: the op journal sweep cannot list %s: %v", filepath.Join(dd, ".moves"), err)
+ }
+ return
+ }
+ byHost := map[string][]provision.Record{}
+ for _, rec := range recs {
+ byHost[rec.Host] = append(byHost[rec.Host], rec)
+ }
+ for _, hd := range hosts {
+ host := hd.Name()
+ if !hd.IsDir() || !journalComponent(host) {
+ continue
+ }
+ ops, err := os.ReadDir(filepath.Join(dd, ".moves", host))
+ if err != nil {
+ w.notice(host+"#final-sweep", "moves: %s: the op journal sweep cannot list its journals: %v", host, err)
+ continue
+ }
+ for _, od := range ops {
+ op := od.Name()
+ if !od.IsDir() || !journalComponent(op) {
+ continue
+ }
+ if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil {
+ w.logTrouble(host, fmt.Errorf("the op journal sweep, %s: %w", op, err))
+ }
+ }
+ }
+}
+
+// sweepFinalJournal is one journal of sweepFinalJournals.
+func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error {
+ j, err := moves.ReadJournal(dataDir, host, op)
+ if err != nil {
+ return err
+ }
+ stamp, reached := j.Steps[finalStepParked]
+ if !reached {
+ return nil
+ }
+ if stamp != "done" {
+ if len(recs) == 0 {
+ w.notice(host+"#final-sweep-"+op, "moves: %s: the journal of %s reached its stamp, and no listed record names the host; it is left", host, op)
+ return nil
+ }
+ for _, rec := range recs {
+ if m := rec.Move; m != nil && m.ID == op && m.SourceParkedAt == nil {
+ return nil // the finalization's own pass stamps it
+ }
+ }
+ }
+ mark := func(step, value string) error {
+ j.Mark(step, value)
+ if err := j.Write(dataDir); err != nil {
+ return err
+ }
+ if fn := w.cfg.finalFault; fn != nil {
+ return fn("after-" + step)
+ }
+ return nil
+ }
+ if stamp != "done" {
+ if err := mark(finalStepParked, "done"); err != nil {
+ return err
+ }
+ }
+ return parkFinalJournal(dataDir, host, op, j, mark)
+}
diff --git a/internal/host/movesfinal_test.go b/internal/host/movesfinal_test.go
new file mode 100644
index 000000000..236dc2510
--- /dev/null
+++ b/internal/host/movesfinal_test.go
@@ -0,0 +1,998 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "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"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// The source's finalization harness: the landing harness driven through
+// restored, then committed the way Task 7 will (the record's probing and
+// probed steps, MoveCommit, and hostclaim.Commit with the source's location
+// park scoped by sourceLocationParkScope), and the SOURCE box (ie-box) given
+// a real registry holding its row "moving" under the move with two timers,
+// the config ack the box last wrote, and the database's marker and
+// quarantine beside the database the cut harness left at the flat path.
+// The source's park engine runs over the fake versioned store and the same
+// fake lease table as the landing's (fakeParkLeases).
+
+const finalAckBody = `{"v":1,"host":"a.go.example.com"}`
+
+type finalHarness struct {
+ *landingHarness
+ reg *registry.Registry
+ finalAt string // the source's crash point; "" crashes nowhere
+ finalDone bool
+ finalHits []string
+ crashAll map[string]bool // crash once at every point not in it (the replay test)
+ deletes int // DeleteHeld calls that deleted
+}
+
+func newFinalHarness(t *testing.T) *finalHarness {
+ t.Helper()
+ h := &finalHarness{landingHarness: newLandingHarness(t)}
+ h.flip(t)
+ h.land(t)
+ h.installed(t)
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ h.reg = reg
+ src := h.phase1Harness.route
+ src.Status, src.HoldOp = "moving", p1ID
+ if err := reg.Upsert(src); err != nil {
+ t.Fatal(err)
+ }
+ for _, tm := range []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000},
+ {Host: p1Host, Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000}} {
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ dd := h.cfg.DataDir
+ for name, body := range map[string]string{p1Host + ".db.gen": "cur", p1Host + ".db.corrupt": "quarantined",
+ p1Host + ".config.env": "SECRET=1\n", p1Host + "-sidecar.env": "SIDE=1\n"} {
+ if err := os.WriteFile(filepath.Join(dd, name), []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if _, err := h.s.Put(context.Background(), appconfig.ConfigAckKey("ie-box", p1Host), []byte(finalAckBody), ""); err != nil {
+ t.Fatal(err)
+ }
+ h.cfg.Route = func(host string) (*registry.Route, error) { return reg.Get(host) }
+ h.cfg.OwnsRoute = func(host string) bool {
+ rt, err := reg.Get(host)
+ return err == nil && rt != nil && rt.Provisioned
+ }
+ h.cfg.DeleteHeld = func(host, op string) (bool, error) {
+ ok, err := reg.DeleteIfHoldWithTimers(host, op)
+ if ok {
+ h.deletes++
+ }
+ return ok, err
+ }
+ h.cfg.Guard = hostguard.Guard{DataDir: dd, LockDir: t.TempDir()}
+ h.cfg.Local = park.Local{DataDir: dd, Box: "ie-box"}
+ h.cfg.ParkLeases = h.leases
+ h.cfg.ParkView, h.cfg.ParkAuthorize = sourceTestView, sourceTestAuthorize
+ h.cfg.parkFault = h.lcfg.parkFault
+ h.cfg.finalFault = func(point string) error {
+ h.finalHits = append(h.finalHits, point)
+ if h.crashAll != nil && !h.crashAll[point] {
+ h.crashAll[point] = true
+ return errCrash
+ }
+ if point == h.finalAt && !h.finalDone {
+ h.finalDone = true
+ return errCrash
+ }
+ return nil
+ }
+ return h
+}
+
+// advanceToProbed walks the record and the claim from restored through
+// probing to probed, as Task 6 will.
+func (h *finalHarness) advanceToProbed(t *testing.T) {
+ t.Helper()
+ ctx := context.Background()
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ for _, step := range [][2]string{{provision.LandRestored, provision.LandProbing}, {provision.LandProbing, provision.LandProbed}} {
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveLandingPhase(g, step[0], step[1], nil)); err != nil {
+ t.Fatal(err)
+ }
+ obs, err := hostclaim.Read(ctx, h.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.SetMovePhase(p1ID, hostclaim.KindLanding, step[0], step[1])); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// commitRecord is the record's commit alone (MoveCommit).
+func (h *finalHarness) commitRecord(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.MoveCommit(g, h.clock.Now())); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// commitClaim is the claim's commit (hostclaim.Commit), with the source's
+// location park scoped from the flip (H7) and the target's staging lane.
+func (h *finalHarness) commitClaim(t *testing.T) {
+ t.Helper()
+ ctx := context.Background()
+ rec := h.read(t, p1Host)
+ obs, err := hostclaim.Read(ctx, h.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ scope, err := sourceLocationParkScope(rec, *obs.Claim.Holder)
+ if err != nil {
+ t.Fatal(err)
+ }
+ staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
+ if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.Commit(p1ID, "ie", "ie-box", "uk", scope, staging, obligationID)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (h *finalHarness) commit(t *testing.T) {
+ t.Helper()
+ h.advanceToProbed(t)
+ h.commitRecord(t)
+ h.commitClaim(t)
+}
+
+// sourcePass runs one pass as the source box.
+func (h *finalHarness) sourcePass(t *testing.T) { t.Helper(); h.pass(t) }
+
+// crashSource runs one source pass that must reach the crash point.
+func (h *finalHarness) crashSource(t *testing.T) {
+ t.Helper()
+ h.sourcePass(t)
+ if !h.finalDone {
+ t.Fatalf("the pass never reached %s (points reached: %v)", h.finalAt, h.finalHits)
+ }
+}
+
+func (h *finalHarness) sourceJournal(t *testing.T) *moves.Journal {
+ t.Helper()
+ j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return j
+}
+
+// live lists which of the source's artifacts are at their live paths.
+func (h *finalHarness) live(t *testing.T) []string {
+ t.Helper()
+ var out []string
+ for _, p := range append(testSourcePaths(), filepath.Join(".moves", p1Host, p1ID)) {
+ if _, err := os.Lstat(filepath.Join(h.cfg.DataDir, p)); err == nil {
+ out = append(out, p)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ }
+ return out
+}
+
+// parked lists every regular file under .parked/<inc>/<host>/, as
+// "<retirement>/<rest>", sorted.
+func (h *finalHarness) parked(t *testing.T) []string {
+ t.Helper()
+ root := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host)
+ var out []string
+ filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
+ if err == nil && !d.IsDir() {
+ rel, _ := filepath.Rel(root, p)
+ out = append(out, filepath.ToSlash(rel))
+ }
+ return nil
+ })
+ sort.Strings(out)
+ return out
+}
+
+// untouched asserts that nothing of the source's was finalized.
+func (h *finalHarness) untouched(t *testing.T, why string) {
+ t.Helper()
+ if got, want := h.live(t), []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}; strings.Join(got, " ") != strings.Join(want, " ") {
+ t.Fatalf("%s: live artifacts %v, want %v", why, got, want)
+ }
+ if got := h.parked(t); len(got) != 0 {
+ t.Fatalf("%s: parked %v", why, got)
+ }
+ if rt, err := h.reg.Get(p1Host); err != nil || rt == nil {
+ t.Fatalf("%s: the row: %+v %v", why, rt, err)
+ }
+ if tm, _ := h.reg.TimersFor(p1Host); len(tm) != 2 {
+ t.Fatalf("%s: timers %+v", why, tm)
+ }
+ if m := h.read(t, p1Host).Move; m == nil || m.SourceParkedAt != nil {
+ t.Fatalf("%s: record %+v", why, m)
+ }
+ if _, _, err := h.s.Get(context.Background(), appconfig.ConfigAckKey("ie-box", p1Host)); err != nil {
+ t.Fatalf("%s: the config ack: %v", why, err)
+ }
+ for _, step := range []string{finalStepRemote, finalStepInstallation, finalStepLocal + "-plan", finalStepRowDeleted} {
+ if h.sourceJournal(t).Done(step) {
+ t.Fatalf("%s: the journal has %s", why, step)
+ }
+ }
+ if calls := h.vs.callsOn("parked/"); len(calls) != 0 {
+ t.Fatalf("%s: the source's location was parked: %v", why, calls)
+ }
+}
+
+// finalized asserts the end state of a source finalization of an
+// unregistered source: every artifact in one retirement under the engine's
+// layout with the op's journal and a receipt naming the op, nothing live,
+// the row and its timers gone, the config ack moved into the report
+// prefix, source_parked_at stamped, and the source's location parked.
+func (h *finalHarness) finalized(t *testing.T) string {
+ t.Helper()
+ ctx := context.Background()
+ if got := h.live(t); len(got) != 0 {
+ t.Fatalf("still live: %v", got)
+ }
+ parked := h.parked(t)
+ if len(parked) == 0 {
+ t.Fatal("nothing parked")
+ }
+ ret, _, _ := strings.Cut(parked[0], "/")
+ var rest []string
+ for _, p := range parked {
+ r, tail, _ := strings.Cut(p, "/")
+ if r != ret {
+ t.Fatalf("more than one retirement: %v", parked)
+ }
+ rest = append(rest, tail)
+ }
+ jpath := "artifacts/.moves/" + p1Host + "/" + p1ID + "/journal.json"
+ want := []string{"artifacts/" + p1Host + "/upload.txt", "artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.corrupt", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env",
+ "artifacts/" + p1Host + ".config.env", "artifacts/" + p1Host + "-sidecar.env", localParkReceiptName}
+ var got []string
+ sawJournal := false
+ for _, r := range rest {
+ switch {
+ case r == jpath:
+ sawJournal = true
+ case strings.HasPrefix(r, "artifacts/.moves/"):
+ // the op journal directory's other contents (the archive scratch)
+ default:
+ got = append(got, r)
+ }
+ }
+ sort.Strings(want)
+ if strings.Join(got, " ") != strings.Join(want, " ") || !sawJournal {
+ t.Fatalf("parked %v (journal %v), want %v and the journal", got, sawJournal, want)
+ }
+ art := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, "artifacts")
+ for name, body := range map[string]string{p1Host + ".db": "the stopped database", p1Host + "/upload.txt": "hello", p1Host + ".env": "APP_MODE=prod\n", p1Host + ".db.gen": "cur"} {
+ if b, err := os.ReadFile(filepath.Join(art, name)); err != nil || string(b) != body {
+ t.Fatalf("parked %s: %q %v", name, b, err)
+ }
+ }
+ var receipt localParkReceipt
+ if b, err := os.ReadFile(filepath.Join(filepath.Dir(art), localParkReceiptName)); err != nil || json.Unmarshal(b, &receipt) != nil || receipt.Op != p1ID || receipt.Retirement != ret || receipt.Incarnation != h.inc {
+ t.Fatalf("receipt %+v %v", receipt, err)
+ }
+ var pj moves.Journal
+ if b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, jpath)); err != nil || json.Unmarshal(b, &pj) != nil {
+ t.Fatalf("the parked journal: %v", err)
+ }
+ if pj.Steps[finalStepRowDeleted] != "done" || !pj.Done(finalStepParked) || pj.Steps[finalStepLocal] != ret || pj.Steps[finalStepInstallation] != "unregistered" {
+ t.Fatalf("the parked journal's steps: %v", pj.Steps)
+ }
+ if rt, err := h.reg.Get(p1Host); err != nil || rt != nil {
+ t.Fatalf("the row: %+v %v", rt, err)
+ }
+ if tm, err := h.reg.TimersFor(p1Host); err != nil || len(tm) != 0 {
+ t.Fatalf("timers: %+v %v", tm, err)
+ }
+ if _, _, err := h.s.Get(ctx, appconfig.ConfigAckKey("ie-box", p1Host)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the config ack is still there: %v", err)
+ }
+ if b, _, err := h.s.Get(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID)+finalConfigAckName); err != nil || string(b) != finalAckBody {
+ t.Fatalf("the config ack's copy: %q %v", b, err)
+ }
+ m := h.read(t, p1Host).Move
+ if m == nil || m.State != provision.MoveCommitted || m.SourceParkedAt == nil {
+ t.Fatalf("record: %+v", m)
+ }
+ // The source's location is archived and gone from its live keys; its
+ // park-replica is still owed until the horizon (the regional runner's).
+ for k := range h.f.buckets["data-ie"] {
+ if strings.HasPrefix(k, cutRoot+"/") {
+ t.Fatalf("still at the source's location: %s", k)
+ }
+ }
+ archived := 0
+ for k := range h.f.buckets["data-ie"] {
+ if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") && strings.Contains(k, "/instances/") {
+ archived++
+ }
+ }
+ if archived == 0 {
+ t.Fatal("nothing archived from the source's location")
+ }
+ if held := h.leases.holding(); len(held) != 0 {
+ t.Fatalf("leases still held: %v", held)
+ }
+ return ret
+}
+
+// Review Focus 5: the source finalizes only a committed record whose row
+// this move still holds (or whose journal says the row is deleted).
+func TestSourceParkWaitsForCommit(t *testing.T) {
+ t.Run("a landing record", func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.advanceToProbed(t)
+ h.sourcePass(t)
+ h.untouched(t, "landing/probed")
+ })
+ t.Run("a committed record whose claim is not committed", func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.advanceToProbed(t)
+ h.commitRecord(t)
+ h.sourcePass(t)
+ h.untouched(t, "claim at probed")
+ })
+ t.Run("a committed record whose row another op holds", func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ if ok, err := h.reg.ReleaseHold(p1Host, p1ID); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ if err := h.reg.SetHold(p1Host, "restoring", p1ID2); err != nil {
+ t.Fatal(err)
+ }
+ h.sourcePass(t)
+ h.untouched(t, "row held by another op")
+ if rt, _ := h.reg.Get(p1Host); rt.HoldOp != p1ID2 {
+ t.Fatalf("the other op's row: %+v", rt)
+ }
+ })
+ t.Run("a committed record with no row and no journalled delete", func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ if err := h.reg.Delete(p1Host); err != nil {
+ t.Fatal(err)
+ }
+ h.sourcePass(t)
+ if got := h.live(t); len(got) != 8 || len(h.parked(t)) != 0 || h.read(t, p1Host).Move.SourceParkedAt != nil {
+ t.Fatalf("finalized without the row: live %v parked %v", got, h.parked(t))
+ }
+ })
+ t.Run("then the commit", func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.advanceToProbed(t)
+ h.sourcePass(t)
+ h.commitRecord(t)
+ h.commitClaim(t)
+ h.sourcePass(t)
+ h.finalized(t)
+ })
+}
+
+// The end to end finalization of an unregistered source (ruling H5(b)): the
+// database, its marker and quarantine, the files directory and the env are
+// gone from the live paths and in one retirement under the engine's layout;
+// the row and its timers rows are gone; source_parked_at is set; the op
+// journal is parked last; a further pass changes nothing.
+func TestSourceParkRetiresInstallationAndRow(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ if c := h.claim(t); len(c.Obligations) != 2 {
+ t.Fatalf("the commit queued %+v (want the source's location and the target's lane, no park-local)", c.Obligations)
+ }
+ h.sourcePass(t)
+ h.finalized(t)
+ if h.deletes != 1 {
+ t.Fatalf("the row was deleted %d times", h.deletes)
+ }
+ before := h.parked(t)
+ h.freshWorker()
+ h.sourcePass(t)
+ if after := h.parked(t); strings.Join(after, " ") != strings.Join(before, " ") || h.deletes != 1 {
+ t.Fatalf("a further pass changed the park: %v", after)
+ }
+ if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
+ t.Fatal("a finalized source still routes the record here")
+ }
+}
+
+// Past the horizon the engine acknowledges the source's location park in
+// the same run, and the finalization goes on.
+func TestSourceParkPastTheHorizonAcknowledges(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ h.clock.Advance(park.SignatureAge + cutRequestLimit + time.Minute)
+ h.sourcePass(t)
+ h.finalized(t)
+ for _, o := range h.claim(t).Obligations {
+ if o.Kind == "park-replica" && o.Region == "ie" {
+ t.Fatalf("the source's location park is still owed: %+v", o)
+ }
+ }
+}
+
+// The remote half comes first: while the archive is incomplete (a delete
+// the store accepts but does not land) nothing local moves.
+func TestSourceParkWaitsForTheRemoteHalf(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ stuck := true
+ h.vs.skipDelete = func(key string) bool { return stuck && strings.HasPrefix(key, cutRoot+"/") }
+ h.sourcePass(t)
+ if got := h.live(t); len(got) != 8 {
+ t.Fatalf("the local half ran before the remote: live %v", got)
+ }
+ if j := h.sourceJournal(t); j.Done(finalStepRemote) || j.Done(finalStepLocal+"-plan") {
+ t.Fatalf("journal: %v", j.Steps)
+ }
+ if rt, _ := h.reg.Get(p1Host); rt == nil {
+ t.Fatal("the row went before the remote half")
+ }
+ stuck = false
+ h.sourcePass(t)
+ h.finalized(t)
+}
+
+// Round-2 finding 12: a crash between the row's delete and the journal's
+// "done" (and before MoveSourceParked) is replayed: the record's region is
+// the target's and the row is gone, so only the op journal routes the
+// record back here, and the replay treats the pending delete as done and
+// stamps source_parked_at.
+func TestSourceParkSurvivesACrashAfterTheRowDelete(t *testing.T) {
+ for _, point := range []string{"after-row-pending", "after-row-delete"} {
+ t.Run(point, func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ h.finalAt = point
+ h.crashSource(t)
+ if j := h.sourceJournal(t); j.Steps[finalStepRowDeleted] != "pending" {
+ t.Fatalf("journal: %v", j.Steps)
+ }
+ rt, _ := h.reg.Get(p1Host)
+ if (point == "after-row-delete") != (rt == nil) {
+ t.Fatalf("the row after %s: %+v", point, rt)
+ }
+ if h.read(t, p1Host).Move.SourceParkedAt != nil {
+ t.Fatal("stamped before the crash point")
+ }
+ if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleSource {
+ t.Fatal("the journal does not route the record back to the source")
+ }
+ h.freshWorker()
+ h.sourcePass(t)
+ h.finalized(t)
+ if h.deletes != 1 {
+ t.Fatalf("deleted %d times", h.deletes)
+ }
+ })
+ }
+}
+
+// Every crash point, each once, with a fresh worker after each: the replay
+// converges on the same end state, with each artifact renamed once, the row
+// deleted once and one retirement.
+func TestSourceParkReplayIsIdempotent(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ h.crashAll = map[string]bool{}
+ h.parkAt = "delete"
+ for i := 0; i < 40 && len(h.live(t)) > 0; i++ {
+ h.freshWorker()
+ h.sourcePass(t)
+ }
+ h.finalized(t)
+ if h.deletes != 1 {
+ t.Fatalf("deleted %d times", h.deletes)
+ }
+ want := []string{"after-" + finalStepRemote, "after-" + finalStepInstallation, "after-" + finalStepLocal + "-plan",
+ "after-" + finalStepLocal,
+ "after-row-pending", "after-row-delete", "after-" + finalStepRowDeleted, "after-config-ack-copy", "after-" + finalStepConfigAck,
+ "after-source-parked-pending", "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
+ for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env"} {
+ want = append(want, "after-rename-unsynced:"+p, "after-rename:"+p)
+ }
+ for _, p := range want {
+ if !h.crashAll[p] {
+ t.Errorf("never crashed at %s (crashed at %v)", p, h.crashAll)
+ }
+ }
+ if !h.parkDone {
+ t.Error("the park engine never crashed at its delete")
+ }
+}
+
+// A registered source (the commit queued its park-local because the
+// inventory lists paths of the journalled stamp S) retires its installation
+// through park.Local under the guard, which completes the park-local; what
+// the inventory does not list is parked by the moves-owned rename.
+func TestSourceParkRetiresARegisteredInstallation(t *testing.T) {
+ h := newFinalHarness(t)
+ dd := h.cfg.DataDir
+ for _, d := range []string{"hostguard", ".parked"} {
+ if err := os.MkdirAll(filepath.Join(dd, d), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rec := `{"v":1,"host":"` + p1Host + `","incarnation":"` + h.inc + `","active_stamp":1,"retired_stamps":[]}`
+ if err := os.WriteFile(filepath.Join(dd, "hostguard", p1Host+".json"), []byte(rec), 0o660); err != nil {
+ t.Fatal(err)
+ }
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
+ if c.Op.Journal[hostclaim.JournalSourceStamp] != "1" {
+ t.Fatalf("fixture: source stamp %q", c.Op.Journal[hostclaim.JournalSourceStamp])
+ }
+ if c.Inventory == nil {
+ c.Inventory = map[string]hostclaim.InventoryEntry{}
+ }
+ c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}
+ })
+ h.commit(t)
+ var chore hostclaim.Obligation
+ for _, o := range h.claim(t).Obligations {
+ if o.Kind == "park-local" {
+ chore = o
+ }
+ }
+ if chore.Box != "ie-box" {
+ t.Fatalf("the commit queued no park-local for the source: %+v", h.claim(t).Obligations)
+ }
+ h.sourcePass(t)
+ c := h.claim(t)
+ for _, o := range c.Obligations {
+ if o.Kind == "park-local" {
+ t.Fatalf("the park-local is still owed: %+v", o)
+ }
+ }
+ if len(c.Inventory["ie-box"].Paths) != 0 {
+ t.Fatalf("inventory: %+v", c.Inventory["ie-box"])
+ }
+ if got := h.live(t); len(got) != 0 {
+ t.Fatalf("still live: %v", got)
+ }
+ parked := strings.Join(h.parked(t), " ")
+ for _, p := range []string{chore.Retirement + "/artifacts/" + p1Host + ".db", chore.Retirement + "/artifacts/" + p1Host + "/upload.txt", chore.Retirement + "/.journal.json"} {
+ if !strings.Contains(parked, p) {
+ t.Fatalf("park.Local's retirement lacks %s: %s", p, parked)
+ }
+ }
+ j := h.parkedJournal(t)
+ if j.Steps[finalStepInstallation] != "retired" {
+ t.Fatalf("installation step %q", j.Steps[finalStepInstallation])
+ }
+ ret := j.Steps[finalStepLocal]
+ if ret == chore.Retirement || !strings.Contains(parked, ret+"/artifacts/"+p1Host+".db.corrupt") {
+ t.Fatalf("the unregistered quarantine is not in the moves-owned retirement %s: %s", ret, parked)
+ }
+ if captured, err := hostguard.Capture(dd, p1Host); err != nil || captured.Stamp != 0 {
+ t.Fatalf("the installation record is still active: %+v %v", captured, err)
+ }
+ if rt, _ := h.reg.Get(p1Host); rt != nil || h.read(t, p1Host).Move.SourceParkedAt == nil {
+ t.Fatal("the row or the stamp")
+ }
+}
+
+// parkedJournal reads the op journal from wherever the finalization parked it.
+func (h *finalHarness) parkedJournal(t *testing.T) moves.Journal {
+ t.Helper()
+ var j moves.Journal
+ for _, p := range h.parked(t) {
+ if strings.HasSuffix(p, "/.moves/"+p1Host+"/"+p1ID+"/journal.json") {
+ b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, p))
+ if err != nil || json.Unmarshal(b, &j) != nil {
+ t.Fatalf("parked journal: %v", err)
+ }
+ return j
+ }
+ }
+ t.Fatalf("no parked journal in %v", h.parked(t))
+ return j
+}
+
+// The rename moves only what its plan recorded: after the plan, an
+// artifact replaced at its live path (another identity) and one that
+// appears where the plan found nothing are both left where they are,
+// untouched; everything recorded is parked.
+func TestSourceParkLeavesWhatItDidNotRecord(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ dd := h.cfg.DataDir
+ if err := os.Remove(filepath.Join(dd, p1Host+".db.corrupt")); err != nil {
+ t.Fatal(err)
+ }
+ h.finalAt = "after-" + finalStepLocal + "-plan"
+ h.crashSource(t)
+ envPath := filepath.Join(dd, p1Host+".env")
+ if err := os.Rename(envPath, envPath+".old"); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(envPath, []byte("SOMEONE=else\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dd, p1Host+".db.corrupt"), []byte("new"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ h.freshWorker()
+ h.sourcePass(t)
+ for name, body := range map[string]string{p1Host + ".env": "SOMEONE=else\n", p1Host + ".db.corrupt": "new"} {
+ if b, err := os.ReadFile(filepath.Join(dd, name)); err != nil || string(b) != body {
+ t.Fatalf("%s was touched: %q %v", name, b, err)
+ }
+ }
+ parked := strings.Join(h.parked(t), " ")
+ if strings.Contains(parked, p1Host+".env") || strings.Contains(parked, p1Host+".db.corrupt") || !strings.Contains(parked, "artifacts/"+p1Host+".db ") {
+ t.Fatalf("parked %s", parked)
+ }
+ if h.read(t, p1Host).Move.SourceParkedAt == nil {
+ t.Fatal("the finalization did not complete")
+ }
+}
+
+// H7: the source's location park's horizon starts at the flip, so a final
+// checkpoint request signed after the stop evidence was collected, and
+// before the flip, is inside it; a park scoped from the stop evidence is
+// refused before the engine runs.
+func TestSourceLocationParkScopeCoversTheFinalCheckpoint(t *testing.T) {
+ checked := time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)
+ flipped := checked.Add(40 * time.Second)
+ signed := checked.Add(30 * time.Second) // the final checkpoint, after CheckedAt
+ rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, FlippedAt: &flipped}}
+ scope, err := sourceLocationParkScope(rec, hostclaim.Holder{Location: cutRoot + "/", Incarnation: "inc"})
+ if err != nil || scope.Location != cutRoot || !scope.EvidenceAt.Equal(flipped) || scope.RequestLimit != cutRequestLimit {
+ t.Fatalf("scope %+v %v", scope, err)
+ }
+ horizon := scope.EvidenceAt.Add(park.SignatureAge + scope.RequestLimit)
+ if latest := signed.Add(park.SignatureAge + cutRequestLimit); latest.After(horizon) {
+ t.Fatalf("a request signed at %s can land until %s, after the horizon %s", signed, latest, horizon)
+ }
+ if early := checked.Add(park.SignatureAge + cutRequestLimit); !signed.Add(park.SignatureAge + cutRequestLimit).After(early) {
+ t.Fatal("fixture: the stop evidence's horizon would have covered it too")
+ }
+ if _, err := sourceLocationParkScope(provision.Record{Move: &provision.MoveState{}}, hostclaim.Holder{}); err == nil {
+ t.Fatal("a scope without the flip")
+ }
+
+ h := newFinalHarness(t)
+ h.advanceToProbed(t)
+ h.commitRecord(t)
+ rec = h.read(t, p1Host)
+ obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ early := hostclaim.ParkScope{Location: cutRoot, Incarnation: h.inc, EvidenceAt: rec.Move.FlippedAt.Add(-time.Minute), RequestLimit: cutRequestLimit}
+ staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
+ if _, err := hostclaim.Update(context.Background(), h.s, obs, hostclaim.Commit(p1ID, "ie", "ie-box", "uk", early, staging, obligationID)); err != nil {
+ t.Fatal(err)
+ }
+ h.sourcePass(t)
+ h.untouched(t, "a horizon before the flip")
+}
+
+// The routing rule for the states after the flip (H5(a)): this box's
+// region names the role, and the row or the op's journal says it has work.
+func TestMovesParticipantRecord(t *testing.T) {
+ dd := t.TempDir()
+ owns := false
+ cfg := moveCfg{Region: "ie", DataDir: dd, OwnsRoute: func(string) bool { return owns }}
+ rec := func(state, from, to string) provision.Record {
+ return provision.Record{Host: p1Host, Region: to, Move: &provision.MoveState{ID: p1ID, State: state, From: from, To: to}}
+ }
+ if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleNone {
+ t.Fatalf("no row and no journal: %q", got)
+ }
+ owns = true
+ for _, c := range []struct {
+ rec provision.Record
+ want string
+ }{
+ {rec(provision.MoveCommitted, "ie", "uk"), roleSource},
+ {rec(provision.MoveRollingBack, "ie", "uk"), roleSource},
+ {rec(provision.MoveReverting, "ie", "uk"), roleSource},
+ {rec(provision.MoveCommitted, "uk", "ie"), roleTarget},
+ {rec(provision.MoveCommitted, "fr", "uk"), roleNone},
+ {rec(provision.MoveLanding, "ie", "uk"), roleNone},
+ {rec(provision.MovePreparing, "ie", "uk"), roleNone},
+ {rec(provision.MoveCommitted, "ie", "ie"), roleNone},
+ {provision.Record{Host: p1Host, Move: &provision.MoveState{State: provision.MoveCommitted, From: "ie", To: "uk"}}, roleNone},
+ {provision.Record{Host: p1Host}, roleNone},
+ } {
+ if got := movesParticipantRecord(cfg, c.rec); got != c.want {
+ t.Errorf("%+v: %q, want %q", c.rec.Move, got, c.want)
+ }
+ }
+ owns = false
+ j := &moves.Journal{Host: p1Host, Op: p1ID}
+ if err := j.Write(dd); err != nil {
+ t.Fatal(err)
+ }
+ if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleSource {
+ t.Fatalf("the op's journal: %q", got)
+ }
+ if got := movesParticipantRecord(cfg, provision.Record{Host: p1Host, Move: &provision.MoveState{ID: "../x", State: provision.MoveCommitted, From: "ie", To: "uk"}}); got != roleNone {
+ t.Fatalf("an unusable op id: %q", got)
+ }
+}
+
+// Production leaves the finalization unarmed (no park engine leases until
+// Plan 5, ruling G17) but wires the registry's delete; an unarmed pass
+// changes nothing.
+func TestSourceParkWaitsUnarmedInProduction(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ for name, unarm := range map[string]func(*moveCfg){
+ "no park leases": func(c *moveCfg) { c.ParkLeases = nil },
+ "no park view": func(c *moveCfg) { c.ParkView = nil },
+ "no park authorizer": func(c *moveCfg) { c.ParkAuthorize = nil },
+ } {
+ saved := h.cfg
+ unarm(&h.cfg)
+ if sourceFinalArmed(h.cfg) {
+ t.Fatalf("%s: armed", name)
+ }
+ h.sourcePass(t)
+ h.untouched(t, name)
+ h.cfg = saved
+ }
+ var prod moveCfg
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
+ armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
+ if prod.DeleteHeld == nil || prod.Guard.DataDir == "" || prod.Local.DataDir == "" || prod.Versioned == nil {
+ t.Fatalf("production wiring: %+v", prod)
+ }
+ prod.Route = func(string) (*registry.Route, error) { return nil, nil }
+ if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || sourceFinalArmed(prod) {
+ t.Fatal("production armed the finalization")
+ }
+}
+
+// The rename helper on its own: absent candidates, a refused path, and a
+// destination that holds something else.
+func TestParkLocalArtifactsRefusals(t *testing.T) {
+ dd := t.TempDir()
+ j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
+ mark := func(step, value string) error { j.Mark(step, value); return nil }
+ p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", NewID: func() string { return "r1" }}
+ for _, bad := range []string{"../x", "/abs", ".parked/x", "hostguard", "a/../b", ""} {
+ p.Paths = []string{bad}
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
+ t.Errorf("%q was accepted", bad)
+ }
+ }
+ if len(j.Steps) != 0 {
+ t.Fatalf("a refused park journalled %v", j.Steps)
+ }
+ if err := os.WriteFile(filepath.Join(dd, "a"), []byte("a"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ p.Paths = []string{"a", "absent"}
+ p.Fault = func(point string) error {
+ if point == "after-rename:a" {
+ return errCrash
+ }
+ return nil
+ }
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
+ t.Fatalf("crash: %v", err)
+ }
+ // Something else at the destination stops the park; nothing more moves.
+ dst := filepath.Join(parkedArtifactsDir(dd, "inc", p1Host, "r1"), "a")
+ if err := os.Rename(dst, filepath.Join(dd, "a")); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(dst, []byte("other"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ p.Fault = nil
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
+ t.Fatal("a foreign destination was accepted")
+ }
+ if b, _ := os.ReadFile(filepath.Join(dd, "a")); string(b) != "a" {
+ t.Fatalf("the live artifact moved: %q", b)
+ }
+ os.Remove(dst)
+ if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
+ t.Fatalf("resume: %q %v", r, err)
+ }
+ if _, err := os.Lstat(filepath.Join(dd, "a")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a is still live: %v", err)
+ }
+ // Another plan under the same step is refused.
+ p.Paths = []string{"b"}
+ delete(j.Steps, "s")
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
+ t.Fatal("a plan for other paths was accepted")
+ }
+}
+
+// testSourcePaths is what the tests expect the source to keep at its flat
+// live paths and park: the database with its marker, journal, quarantine
+// and litestream shadow, the files directory, <host>.env, and the two
+// rendered env files (fix round 1, ruling 3).
+func testSourcePaths() []string {
+ db := p1Host + ".db"
+ return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", p1Host, p1Host + ".env",
+ p1Host + ".config.env", p1Host + "-sidecar.env"}
+}
+
+// sourceTestView and sourceTestAuthorize are the park view and authorizer
+// the tests inject for the source's location park, in place of Plan 5's
+// owners (fix round 1, ruling 4: production never substitutes stand-ins).
+// The target's region is authoritative after the commit, so every version
+// under the location is selected; a chore is authorized only when it is
+// this move's, for ie-box or ie, under the move's committed landing with
+// ie-box its source.
+func sourceTestView(_ context.Context, obs hostclaim.Observed) (park.View, error) {
+ if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
+ return park.View{}, park.ErrUnsafe
+ }
+ location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
+ return park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "uk",
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
+}
+
+func sourceTestAuthorize(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ c := obs.Claim
+ if obs.Absent || c == nil || c.Op.ID != p1ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted ||
+ len(c.Op.Participants) != 2 || c.Op.Participants[0] != "ie-box" || o.Op != p1ID {
+ return park.ErrUnsafe
+ }
+ if (o.Kind == "park-local" && o.Box == "ie-box") || (o.Kind == "park-replica" && o.Region == "ie") {
+ return nil
+ }
+ return park.ErrUnsafe
+}
+
+// Fix round 1, ruling 1: a crash after source_parked_at is stamped and
+// before the journal is parked, then the close (which clears the move, so
+// no record routes here any more), then a fresh source pass: the local
+// sweep parks the journal, because the close proves the stamp landed.
+func TestSourceParkJournalSurvivesTheClose(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ h.finalAt = "after-source-parked-record"
+ h.crashSource(t)
+ if j := h.sourceJournal(t); j.Done(finalStepParked) && j.Steps[finalStepParked] == "done" {
+ t.Fatalf("fixture: the stamp is journalled done: %v", j.Steps)
+ }
+ ctx := context.Background()
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ for _, mut := range []func(*provision.Record) error{provision.MoveOpened(g, h.clock.Now()), provision.MoveClose(g, h.clock.Now())} {
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, mut); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID {
+ t.Fatalf("fixture: not closed: %+v", rec)
+ }
+ if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
+ t.Fatal("fixture: a closed record still routes here")
+ }
+ h.freshWorker()
+ h.sourcePass(t)
+ if got := h.live(t); len(got) != 0 {
+ t.Fatalf("still live after the close: %v", got)
+ }
+ j := h.parkedJournal(t)
+ if j.Steps[finalStepParked] != "done" {
+ t.Fatalf("the parked journal's stamp: %v", j.Steps)
+ }
+}
+
+// The sweep never touches a journal that has not reached the stamp, nor one
+// whose stamp is pending while the record still carries the op unstamped: a
+// journal of another op (a landing's, say) under a record with no move, and
+// this op's journal stopped right after it journalled the stamp pending,
+// before the stamp's CAS.
+func TestSourceParkSweepLeavesUnstampedJournals(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ // Another op's journal that would park cleanly if the sweep took it: a
+ // complete local plan and retirement, everything but the stamp step.
+ plan, err := json.Marshal(localParkPlan{Retirement: "r0", Incarnation: h.inc})
+ if err != nil {
+ t.Fatal(err)
+ }
+ otherSteps := map[string]string{finalStepLocal + "-plan": string(plan), finalStepLocal: "r0", finalStepRowDeleted: "done"}
+ other := &moves.Journal{Host: p1Host, Op: p1ID2, Steps: otherSteps}
+ if err := other.Write(h.cfg.DataDir); err != nil {
+ t.Fatal(err)
+ }
+ h.finalAt = "after-source-parked-pending"
+ h.crashSource(t)
+ if j := h.sourceJournal(t); j.Steps[finalStepParked] != "pending" || h.read(t, p1Host).Move.SourceParkedAt != nil {
+ t.Fatalf("fixture: %v", j.Steps)
+ }
+ h.cfg.ParkLeases = nil // unarmed: only the sweep runs
+ h.freshWorker()
+ h.sourcePass(t)
+ for _, op := range []string{p1ID, p1ID2} {
+ if !moveJournalPresent(h.cfg.DataDir, p1Host, op) {
+ t.Fatalf("the sweep took %s's journal before its stamp", op)
+ }
+ }
+ if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2); err != nil || len(j.Steps) != len(otherSteps) {
+ t.Fatalf("the sweep wrote into another op's journal: %v %v", j.Steps, err)
+ }
+}
+
+// Fix round 1, ruling 2: a replay that finds an artifact already at its
+// destination (a stop after the rename, before either sync) syncs the
+// source's parent and every destination parent before it goes on; and the
+// rename offers an interruption point between the rename and the syncs.
+func TestParkLocalArtifactsSyncsOnReplay(t *testing.T) {
+ dd := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(dd, "d", "e"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dd, "d", "e", "a"), []byte("a"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
+ mark := func(step, value string) error { j.Mark(step, value); return nil }
+ var points []string
+ p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", Paths: []string{"d/e/a"}, NewID: func() string { return "r1" },
+ Fault: func(point string) error {
+ points = append(points, point)
+ if point == "after-rename-unsynced:d/e/a" {
+ return errCrash
+ }
+ return nil
+ }}
+ if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
+ t.Fatalf("no interruption between the rename and its syncs: %v (points %v)", err, points)
+ }
+ var synced []string
+ saved := retireSync
+ retireSync = func(path string) error { synced = append(synced, path); return saved(path) }
+ t.Cleanup(func() { retireSync = saved })
+ p.Fault = nil
+ if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
+ t.Fatalf("replay: %q %v", r, err)
+ }
+ art := parkedArtifactsDir(dd, "inc", p1Host, "r1")
+ for _, want := range []string{filepath.Join(dd, "d", "e"), filepath.Join(art, "d", "e"), filepath.Join(art, "d"), art, filepath.Dir(art), filepath.Join(dd, ".parked")} {
+ found := false
+ for _, s := range synced {
+ found = found || s == want
+ }
+ if !found {
+ t.Errorf("the replay did not sync %s (synced %v)", want, synced)
+ }
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index b570dd677..12f4ba48b 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -315,6 +315,10 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
cfg.ReplaceTimers = heldTimersReplacer(writes, reg)
cfg.SetReplication = reg.SetReplication
cfg.SetNextDue = reg.SetNextDue
+ // The source's finalization (movesfinal.go): a plain registry call
+ // under the host's hold lock, not a claim-selected write (ruling
+ // H5(d)). Unarmed with ParkLeases, as the landing is.
+ cfg.DeleteHeld = reg.DeleteIfHoldWithTimers
}
if eb != nil {
// A box with no delegated cgroup has no manager, and Reconstruct
@@ -546,6 +550,16 @@ type moveCfg struct {
// and the park engine's (park.Engine.After). Nil in production.
landFault func(point string) error
parkFault func(step string) error
+
+ // The source's finalization (§7.9 step 3, movesfinal.go). DeleteHeld
+ // deletes this box's row and its timers rows in one transaction, only
+ // while op holds the row (registry.DeleteIfHoldWithTimers, ruling
+ // H5(d)). The finalization also uses Guard, Local, ParkLeases,
+ // ParkView, ParkAuthorize, Versioned and Route above; ParkLeases nil
+ // leaves it unarmed in production (sourceFinalArmed). finalFault is
+ // its test seam, nil in production.
+ DeleteHeld func(host, op string) (bool, error)
+ finalFault func(point string) error
}
// defaultMoveReadyDelta is the readiness bar when
@@ -617,6 +631,18 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
continue
}
if rec.Region != cfg.Region || (cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host)) {
+ // After the flip the record's region is the target's, and
+ // after the source's finalization deletes its row OwnsRoute
+ // no longer holds: a participant that still has the row or
+ // the op's journal is visited by the post-flip rule (ruling
+ // H5(a), movesfinal.go), never by phase1MoveRound.
+ if role := movesParticipantRecord(cfg, rec); role != roleNone {
+ if err := w.participantRound(ctx, s, rec, role); err != nil {
+ w.logTrouble(rec.Host, err)
+ continue
+ }
+ w.trouble.Delete(rec.Host)
+ }
continue
}
if !loadFleet() {
@@ -678,6 +704,11 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
w.evictArchives(nil) // no move into this box: no verdict to keep
w.retireHeldRefusals(nil)
}
+ if cfg.Enabled {
+ // The source's op journals that reached the stamp, whoever's record
+ // no longer routes here (movesfinal.go, fix round 1).
+ w.sweepFinalJournals(recs)
+ }
w.liftFences(recs)
return nil
}
diff --git a/internal/host/movesphase1_harness_test.go b/internal/host/movesphase1_harness_test.go
index 812f8f9c2..0aea92faf 100644
--- a/internal/host/movesphase1_harness_test.go
+++ b/internal/host/movesphase1_harness_test.go
@@ -231,7 +231,9 @@ func (v *fakeVersioned) DeleteVersion(_ context.Context, bucket, key, versionID
if v.skipDelete != nil && v.skipDelete(key) {
return nil
}
- if _, ok := v.f.buckets[bucket][key]; ok && v.f.versionOf(bucket, key) == versionID {
+ // "null" is how S3 names an object written while the bucket was not
+ // versioning (versionID), and deleting that version removes it.
+ if _, ok := v.f.buckets[bucket][key]; ok && v.versionID(bucket, key) == versionID {
delete(v.f.buckets[bucket], key)
}
return nil
diff --git a/internal/host/movesretire.go b/internal/host/movesretire.go
new file mode 100644
index 000000000..9dd91bf26
--- /dev/null
+++ b/internal/host/movesretire.go
@@ -0,0 +1,337 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/moves"
+)
+
+// The moves-owned local retirement (ruling H5(b), the pattern of Plan 2's
+// extraction staging): artifacts the claim's inventory does not list, so
+// park.Local cannot retire them, renamed into the park engine's local
+// layout under a journalled plan, and an op's own journal directory parked
+// after them. The source's finalization (movesfinal.go) uses it; Plan 4's
+// rollback reuses it.
+
+func journalComponent(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+
+// localArtifactPark is one moves-owned local retirement (ruling H5(b), the
+// pattern of Plan 2's extraction staging): artifacts the claim's inventory
+// does not list, so park.Local cannot retire them, renamed into the park
+// engine's local layout, .parked/<incarnation>/<host>/<retirement>/artifacts/,
+// keeping their paths. Plan 4's rollback reuses it.
+type localArtifactPark struct {
+ DataDir, Incarnation, Host, Op string
+ // Paths are the candidates, relative to DataDir. Absent ones are recorded
+ // as absent and never touched.
+ Paths []string
+ // Step is the journal step that records the retirement id once every
+ // rename is durable; Step+"-plan" records the plan before the first one.
+ Step string
+ NewID func() string
+ // Fault is the test seam, after the plan and after each rename.
+ Fault func(point string) error
+}
+
+// localParkPlan is what is journalled before the first rename: the
+// retirement id and every candidate's identity ("dev:ino"; "" when absent).
+type localParkPlan struct {
+ Retirement string `json:"retirement"`
+ Incarnation string `json:"incarnation"`
+ Items []localParkItem `json:"items"`
+}
+
+type localParkItem struct {
+ Path string `json:"path"`
+ ID string `json:"id,omitempty"`
+}
+
+// localParkReceipt sits beside artifacts/ in the retirement, so a later
+// reader can find a moves-owned retirement by its op (D6, I57).
+type localParkReceipt struct {
+ V int `json:"v"`
+ Op string `json:"op"`
+ Host string `json:"host"`
+ Incarnation string `json:"incarnation"`
+ Retirement string `json:"retirement"`
+ Paths []string `json:"paths"`
+}
+
+const localParkReceiptName = "moves-retirement.json"
+
+// retireSync is the fsync every moves-owned retirement makes durable with
+// (syncPath); a variable so a test can see which directories were synced.
+var retireSync = syncPath
+
+// parkedEventDir and parkedArtifactsDir are the engine's local layout
+// (park.Local's parkLocal): one directory per retirement event.
+func parkedEventDir(dataDir, incarnation, host, retirement string) string {
+ return filepath.Join(dataDir, ".parked", incarnation, host, retirement)
+}
+
+func parkedArtifactsDir(dataDir, incarnation, host, retirement string) string {
+ return filepath.Join(parkedEventDir(dataDir, incarnation, host, retirement), "artifacts")
+}
+
+func journalledLocalPlan(j *moves.Journal, step string) (localParkPlan, error) {
+ var plan localParkPlan
+ raw, ok := j.Steps[step+"-plan"]
+ if !ok {
+ return plan, fmt.Errorf("the journal has no %s plan", step)
+ }
+ if err := json.Unmarshal([]byte(raw), &plan); err != nil {
+ return plan, fmt.Errorf("the journal's %s plan: %w", step, err)
+ }
+ if !journalComponent(plan.Retirement) || !journalComponent(plan.Incarnation) {
+ return plan, fmt.Errorf("the journal's %s plan names an unusable retirement", step)
+ }
+ return plan, nil
+}
+
+// artifactPathOK is a relative path under the data directory that is not
+// the guard's or the park's own.
+func artifactPathOK(p string) bool {
+ if p == "" || filepath.IsAbs(p) || filepath.Clean(p) != p || strings.ContainsAny(p, "\\\x00") {
+ return false
+ }
+ for _, c := range strings.Split(p, "/") {
+ if !journalComponent(c) {
+ return false
+ }
+ }
+ first, _, _ := strings.Cut(p, "/")
+ return first != ".parked" && first != "hostguard"
+}
+
+// parkLocalArtifacts runs one moves-owned local retirement and returns its
+// retirement id. In order: the plan (a fresh retirement id and each
+// candidate's identity) is journalled before anything moves; the retirement's
+// directories and its receipt are made; each recorded artifact is renamed to
+// the same path under artifacts/ and its source's parent and every
+// destination directory up to the data directory synced; then Step records
+// the retirement id. A replay resumes from the journal:
+// - an artifact already at its destination with its identity is synced as
+// a fresh rename is (a stop between the rename and its syncs leaves the
+// rename undurable, fix round 1), and is then done;
+// - one still at its live path with its identity is renamed;
+// - one recorded absent, or whose live path now holds something with
+// another identity, is not this park's: it is left where it is, never
+// moved or removed;
+// - a destination that holds anything else is an error, and nothing more
+// is moved.
+func parkLocalArtifacts(ctx context.Context, j *moves.Journal, mark func(step, value string) error, p localArtifactPark) (string, error) {
+ if r, ok := j.Steps[p.Step]; ok {
+ return r, nil
+ }
+ if p.DataDir == "" || !journalComponent(p.Incarnation) || !journalComponent(p.Host) || p.Op == "" || p.Step == "" || p.NewID == nil {
+ return "", errors.New("a local park names its data directory, incarnation, host, op and step")
+ }
+ for _, path := range p.Paths {
+ if !artifactPathOK(path) {
+ return "", fmt.Errorf("a local park cannot retire %q", path)
+ }
+ }
+ fault := func(point string) error {
+ if p.Fault != nil {
+ return p.Fault(point)
+ }
+ return nil
+ }
+ plan, err := journalledLocalPlan(j, p.Step)
+ if err != nil {
+ if _, planned := j.Steps[p.Step+"-plan"]; planned {
+ return "", err
+ }
+ if plan, err = p.plan(); err != nil {
+ return "", err
+ }
+ body, err := json.Marshal(plan)
+ if err != nil {
+ return "", err
+ }
+ if err := mark(p.Step+"-plan", string(body)); err != nil {
+ return "", err
+ }
+ }
+ if plan.Incarnation != p.Incarnation || len(plan.Items) != len(p.Paths) {
+ return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
+ }
+ for i, it := range plan.Items {
+ if it.Path != p.Paths[i] {
+ return "", fmt.Errorf("the journal's %s plan is not this park's", p.Step)
+ }
+ }
+ dest := parkedArtifactsDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
+ if err := mkdirSynced(p.DataDir, dest); err != nil {
+ return "", err
+ }
+ receipt, err := json.Marshal(localParkReceipt{V: 1, Op: p.Op, Host: p.Host, Incarnation: p.Incarnation, Retirement: plan.Retirement, Paths: p.Paths})
+ if err != nil {
+ return "", err
+ }
+ event := parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)
+ if err := writeFileAtomic(filepath.Join(event, localParkReceiptName), receipt, 0o640); err != nil {
+ return "", err
+ }
+ for _, it := range plan.Items {
+ if err := ctx.Err(); err != nil {
+ return "", err
+ }
+ src, dst := filepath.Join(p.DataDir, it.Path), filepath.Join(dest, it.Path)
+ if it.ID == "" {
+ if _, err := os.Lstat(src); err == nil {
+ log.Printf("moves: %s: %s appeared after the park recorded it absent; it is not this park's and is left where it is", p.Host, it.Path)
+ }
+ continue
+ }
+ if done, err := sameIdentity(dst, it.ID); err != nil {
+ return "", err
+ } else if done {
+ if err := syncRename(p.DataDir, src, dst); err != nil {
+ return "", err
+ }
+ if err := fault("after-rename:" + it.Path); err != nil {
+ return "", err
+ }
+ continue
+ }
+ if _, err := os.Lstat(dst); err == nil {
+ return "", fmt.Errorf("%s holds what this park did not put there", dst)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return "", err
+ }
+ ours, err := sameIdentity(src, it.ID)
+ if err != nil {
+ return "", err
+ }
+ if !ours {
+ log.Printf("moves: %s: %s is no longer the artifact the park recorded; what is there is left where it is", p.Host, it.Path)
+ continue
+ }
+ if err := mkdirSynced(p.DataDir, filepath.Dir(dst)); err != nil {
+ return "", err
+ }
+ if err := os.Rename(src, dst); err != nil {
+ return "", fmt.Errorf("park %s: %w", it.Path, err)
+ }
+ if err := fault("after-rename-unsynced:" + it.Path); err != nil {
+ return "", err
+ }
+ if err := syncRename(p.DataDir, src, dst); err != nil {
+ return "", err
+ }
+ if err := fault("after-rename:" + it.Path); err != nil {
+ return "", err
+ }
+ }
+ if err := retireSync(dest); err != nil {
+ return "", err
+ }
+ if err := mark(p.Step, plan.Retirement); err != nil {
+ return "", err
+ }
+ log.Printf("moves: %s: the local artifacts of %s are parked in %s", p.Host, p.Op, event)
+ return plan.Retirement, nil
+}
+
+// plan records a fresh retirement id and each candidate's identity. The
+// retirement's directory must not exist yet: the id is this park's alone.
+func (p localArtifactPark) plan() (localParkPlan, error) {
+ plan := localParkPlan{Retirement: p.NewID(), Incarnation: p.Incarnation}
+ if !journalComponent(plan.Retirement) {
+ return plan, errors.New("no retirement id")
+ }
+ if _, err := os.Lstat(parkedEventDir(p.DataDir, p.Incarnation, p.Host, plan.Retirement)); err == nil {
+ return plan, fmt.Errorf("retirement %s already exists", plan.Retirement)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return plan, err
+ }
+ for _, path := range p.Paths {
+ id, err := pathIdentity(filepath.Join(p.DataDir, path))
+ if errors.Is(err, os.ErrNotExist) {
+ id, err = "", nil
+ }
+ if err != nil {
+ return plan, err
+ }
+ plan.Items = append(plan.Items, localParkItem{Path: path, ID: id})
+ }
+ return plan, nil
+}
+
+// mkdirSynced makes dir (under root) and syncs it and every directory above
+// it up to root. Every call syncs the whole chain, not only the directories
+// it made: a replay after a stop between MkdirAll and the syncs finds the
+// directories there and must still make their entries durable.
+func mkdirSynced(root, dir string) error {
+ if err := os.MkdirAll(dir, 0o770); err != nil {
+ return err
+ }
+ return syncChain(root, dir)
+}
+
+// syncChain syncs dir and every directory above it up to and including root.
+// dir must be root or under it.
+func syncChain(root, dir string) error {
+ root, dir = filepath.Clean(root), filepath.Clean(dir)
+ if rel, err := filepath.Rel(root, dir); err != nil || rel == ".." || strings.HasPrefix(rel, "../") {
+ return fmt.Errorf("%s is not under %s", dir, root)
+ }
+ for d := dir; ; d = filepath.Dir(d) {
+ if err := retireSync(d); err != nil {
+ return err
+ }
+ if d == root || d == filepath.Dir(d) {
+ return nil
+ }
+ }
+}
+
+// syncRename makes a rename from src to dst durable: the source's parent,
+// and every destination directory up to root.
+func syncRename(root, src, dst string) error {
+ if err := retireSync(filepath.Dir(src)); err != nil {
+ return err
+ }
+ return syncChain(root, filepath.Dir(dst))
+}
+
+// parkOpJournal renames the op's journal directory into the retirement's
+// artifacts, at .moves/<host>/<op>. The directory is the op's own (its name
+// is the op id). Absent with the destination there is a replay after the
+// rename, which is synced again and done; absent with no destination is
+// nothing to park; a destination already there beside the journal is
+// refused.
+func parkOpJournal(dataDir, incarnation, host, op, retirement string) error {
+ src := moves.JournalDir(dataDir, host, op)
+ dst := filepath.Join(parkedArtifactsDir(dataDir, incarnation, host, retirement), ".moves", host, op)
+ if _, err := os.Lstat(src); errors.Is(err, os.ErrNotExist) {
+ if _, err := os.Lstat(dst); err == nil {
+ return syncRename(dataDir, src, dst)
+ }
+ return nil
+ } else if err != nil {
+ return err
+ }
+ if _, err := os.Lstat(dst); err == nil {
+ return fmt.Errorf("%s already holds a journal", dst)
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ if err := mkdirSynced(dataDir, filepath.Dir(dst)); err != nil {
+ return err
+ }
+ if err := os.Rename(src, dst); err != nil {
+ return err
+ }
+ return syncRename(dataDir, src, dst)
+}
diff --git a/internal/moves/startcheck.go b/internal/moves/startcheck.go
new file mode 100644
index 000000000..c7586cdf8
--- /dev/null
+++ b/internal/moves/startcheck.go
@@ -0,0 +1,162 @@
+package moves
+
+import (
+ "errors"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+// The start check (spec §5.1, INV-4, J2-R2-8): before every spawn the three
+// markers a generation leaves on a box are compared with the host claim. A
+// marker that merely exists proves nothing; the claim says which generation
+// this box may run.
+//
+// The three markers have three formats (Paul's ruling on H11):
+//
+// - generation.json is genselect.Marker, read from the box's own location;
+// absent is the origin.
+// - .files-gen is the JSON {op, incarnation} (FilesGen); absent is "no
+// files directory yet".
+// - <dbPath>.gen is the plain generation name hostbox.WriteGenerationMarker
+// writes and ColdRestore rewrites; "" is absent. It carries no incarnation
+// and no location: the claim's holder and the restore vouch for those.
+//
+// Refusals are typed and carry member-safe reasons, never marker contents.
+var (
+ // ErrStartMarker: generation.json does not name a generation the claim
+ // expects, or names another incarnation.
+ ErrStartMarker = errors.New("this box holds another generation of the instance")
+ // ErrStartFiles: .files-gen does not name a generation and incarnation
+ // the claim expects.
+ ErrStartFiles = errors.New("the instance's files belong to another generation")
+ // ErrStartDB: the database's marker does not name the claim's generation.
+ ErrStartDB = errors.New("the instance's database belongs to another generation")
+)
+
+// FilesGen is the .files-gen body (§5.1): the op and the incarnation.
+type FilesGen struct {
+ Op string `json:"op"`
+ Incarnation string `json:"incarnation"`
+}
+
+// originOr names the origin generation the way the claim's admissions do.
+func originOr(g string) string {
+ if g == "" {
+ return hostclaim.RootGeneration
+ }
+ return g
+}
+
+// coldAdmission is the newest admission into generation g that is not a warm
+// self-link (F from F, the warm keep genselect's unfinished skips). Its
+// predecessor is where g came from.
+func coldAdmission(h *hostclaim.Holder, g string) (hostclaim.AdmissionRecord, bool) {
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ a := h.Admissions[i]
+ if originOr(a.Generation) != g || a.Mode == "warm" {
+ continue
+ }
+ return a, true
+ }
+ return hostclaim.AdmissionRecord{}, false
+}
+
+// markerGenerations is the set of generations generation.json may name for
+// this holder: holder.generation, as the admission's CAS left it, and the
+// lagging ones. A marker is published once per admission, at Replicate, so
+// after a restarted attempt (a fresh generation restored from the prepared
+// one) or a run of admissions that published nothing, the marker still names
+// an earlier generation. Walking holder.admissions from the newest, skipping
+// warm self-links, each cold admission into an accepted generation adds its
+// predecessor. The origin is hostclaim.RootGeneration here.
+func markerGenerations(h *hostclaim.Holder) map[string]bool {
+ cur := originOr(h.Generation)
+ ok := map[string]bool{cur: true}
+ for {
+ a, found := coldAdmission(h, cur)
+ if !found {
+ return ok
+ }
+ next := originOr(a.Predecessor)
+ if ok[next] { // damaged record: predecessors that loop
+ return ok
+ }
+ ok[next] = true
+ cur = next
+ }
+}
+
+// filesGenerationOK reports whether .files-gen may name g: exactly
+// holder.generation, or the cold predecessor of holder.generation (the
+// prepared generation a restarted attempt was restored from, whose files
+// are still the ones installed). Nothing further back counts: files from
+// before the preparation are stale.
+func filesGenerationOK(h *hostclaim.Holder, g string) bool {
+ if g == "" || g == hostclaim.RootGeneration {
+ return false
+ }
+ if g == h.Generation {
+ return true
+ }
+ a, found := coldAdmission(h, originOr(h.Generation))
+ return found && a.Predecessor == g
+}
+
+// StartCheck is the §5.1 start check against the claim, run after the
+// admission CAS and before every spawn.
+//
+// 1. marker (generation.json; nil is the origin) names a generation the
+// claim expects (holder.generation or, lagging, an earlier link of the
+// admission chain) and the holder's incarnation.
+// 2. filesGen, when present, names holder.generation or its cold
+// predecessor (a restarted attempt's files still name the prepared
+// generation; nothing older) and the holder's incarnation.
+// 3. dbGen, when not "", is exactly holder.generation: the restore rewrites
+// it, so it is current after every restore. A warm database that fails
+// here is from an earlier stay; the caller parks it and restores cold.
+//
+// A claim with no holder expects no generation, so every start is refused
+// with ErrStartMarker.
+func StartCheck(c *hostclaim.Claim, marker *genselect.Marker, filesGen *FilesGen, dbGen string) error {
+ if c == nil || c.Holder == nil {
+ return ErrStartMarker
+ }
+ h := c.Holder
+ want := markerGenerations(h)
+ if marker == nil {
+ if !want[hostclaim.RootGeneration] {
+ return ErrStartMarker
+ }
+ } else if marker.Op == "" || marker.Op == hostclaim.RootGeneration ||
+ !want[marker.Op] || marker.Incarnation != h.Incarnation {
+ return ErrStartMarker
+ }
+ if filesGen != nil {
+ if !filesGenerationOK(h, filesGen.Op) || filesGen.Incarnation != h.Incarnation {
+ return ErrStartFiles
+ }
+ }
+ if dbGen != "" && dbGen != h.Generation {
+ return ErrStartDB
+ }
+ return nil
+}
+
+// PreparedCheck validates a landing probe's inputs before admission (Plan 3
+// Task 5 step 3): the files marker and the database marker both exist and
+// name the prepared generation (the move id), and the files marker names the
+// holder's incarnation. The database marker is the plain name (H11); the
+// location is vouched for by the holder.
+func PreparedCheck(prepared string, holder hostclaim.Holder, filesGen *FilesGen, dbGen string) error {
+ if prepared == "" || prepared == hostclaim.RootGeneration {
+ return ErrStartFiles
+ }
+ if filesGen == nil || filesGen.Op != prepared || filesGen.Incarnation != holder.Incarnation {
+ return ErrStartFiles
+ }
+ if dbGen != prepared {
+ return ErrStartDB
+ }
+ return nil
+}
diff --git a/internal/moves/startcheck_test.go b/internal/moves/startcheck_test.go
new file mode 100644
index 000000000..c99281ce0
--- /dev/null
+++ b/internal/moves/startcheck_test.go
@@ -0,0 +1,232 @@
+package moves
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func startClaim(gen string, adm ...hostclaim.AdmissionRecord) *hostclaim.Claim {
+ return &hostclaim.Claim{Holder: &hostclaim.Holder{
+ Incarnation: "inc-1", Location: "loc/", Generation: gen, Admissions: adm,
+ }}
+}
+
+func startMarker(op string) *genselect.Marker {
+ return &genselect.Marker{Op: op, Incarnation: "inc-1"}
+}
+
+func TestStartCheckOriginEverywhere(t *testing.T) {
+ if err := StartCheck(startClaim(""), nil, nil, ""); err != nil {
+ t.Fatalf("origin everywhere: %v", err)
+ }
+}
+
+func TestStartCheckMarkerNamesTheHoldersGeneration(t *testing.T) {
+ c := startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ fg := &FilesGen{Op: "g2", Incarnation: "inc-1"}
+ if err := StartCheck(c, startMarker("g2"), fg, "g2"); err != nil {
+ t.Fatalf("all three name the holder's generation: %v", err)
+ }
+}
+
+func TestStartCheckLaggingMarker(t *testing.T) {
+ // The latest admission replaced g1 with g2 and has published nothing.
+ c := startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ if err := StartCheck(c, startMarker("g1"), nil, ""); err != nil {
+ t.Fatalf("marker naming the latest admission's predecessor: %v", err)
+ }
+ // A restarted probe: the fresh generation F restored from the prepared M;
+ // the files still name M.
+ c = startClaim("F", hostclaim.AdmissionRecord{Generation: "M", Predecessor: "g0"},
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "M"})
+ fg := &FilesGen{Op: "M", Incarnation: "inc-1"}
+ if err := StartCheck(c, startMarker("M"), fg, "F"); err != nil {
+ t.Fatalf("restarted attempt: %v", err)
+ }
+}
+
+func TestStartCheckChainOfAdmissionsWithoutMarkers(t *testing.T) {
+ c := startClaim("g3",
+ hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"},
+ hostclaim.AdmissionRecord{Generation: "g3", Predecessor: "g2"})
+ for _, op := range []string{"g3", "g2", "g1"} {
+ if err := StartCheck(c, startMarker(op), nil, ""); err != nil {
+ t.Errorf("marker %s through the chain: %v", op, err)
+ }
+ }
+ if err := StartCheck(c, startMarker("g0"), nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("marker beyond the chain = %v, want ErrStartMarker", err)
+ }
+ // The chain is by predecessor link, not by position: an admission whose
+ // generation the chain never reaches adds nothing.
+ c = startClaim("g3",
+ hostclaim.AdmissionRecord{Generation: "x2", Predecessor: "x1"},
+ hostclaim.AdmissionRecord{Generation: "g3", Predecessor: "g2"})
+ if err := StartCheck(c, startMarker("x1"), nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("marker off the chain = %v, want ErrStartMarker", err)
+ }
+}
+
+func TestStartCheckChainTerminatesOnACycle(t *testing.T) {
+ // A damaged record whose predecessors loop must still end.
+ c := startClaim("g2",
+ hostclaim.AdmissionRecord{Generation: "g1", Predecessor: "g2"},
+ hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ if err := StartCheck(c, startMarker("g1"), nil, ""); err != nil {
+ t.Errorf("marker in the cycle: %v", err)
+ }
+ if err := StartCheck(c, startMarker("g9"), nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("marker outside the cycle = %v, want ErrStartMarker", err)
+ }
+}
+
+func TestStartCheckRejectsFilesBeforePreparedGeneration(t *testing.T) {
+ // M was prepared over S; F is the restarted attempt restored from M.
+ // Files from before the preparation are stale, even though S is an
+ // ancestor of the holder's generation.
+ c := startClaim("F",
+ hostclaim.AdmissionRecord{Generation: "M", Predecessor: "S", Mode: "cold"},
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "M", Mode: "cold"})
+ if err := StartCheck(c, startMarker("M"), &FilesGen{Op: "S", Incarnation: "inc-1"}, "F"); !errors.Is(err, ErrStartFiles) {
+ t.Errorf("files naming S = %v, want ErrStartFiles", err)
+ }
+ if err := StartCheck(c, startMarker("M"), &FilesGen{Op: "M", Incarnation: "inc-1"}, "F"); err != nil {
+ t.Errorf("files naming M: %v", err)
+ }
+ if err := StartCheck(c, startMarker("M"), &FilesGen{Op: "F", Incarnation: "inc-1"}, "F"); err != nil {
+ t.Errorf("files naming F: %v", err)
+ }
+}
+
+func TestStartCheckLaggingFilesAfterWarmAdmission(t *testing.T) {
+ // Cold F from M, then a warm re-admission of F (F from F): the files
+ // still name M.
+ c := startClaim("F",
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "M", Mode: "cold"},
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "F", Mode: "warm"})
+ if err := StartCheck(c, startMarker("F"), &FilesGen{Op: "M", Incarnation: "inc-1"}, "F"); err != nil {
+ t.Errorf("files naming M after a warm admission: %v", err)
+ }
+ if err := StartCheck(c, startMarker("F"), &FilesGen{Op: "other", Incarnation: "inc-1"}, "F"); !errors.Is(err, ErrStartFiles) {
+ t.Errorf("files naming another generation = %v, want ErrStartFiles", err)
+ }
+}
+
+func TestStartCheckLaggingMarkerAfterWarmAdmission(t *testing.T) {
+ c := startClaim("F",
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "M", Mode: "cold"},
+ hostclaim.AdmissionRecord{Generation: "F", Predecessor: "F", Mode: "warm"})
+ if err := StartCheck(c, startMarker("M"), nil, ""); err != nil {
+ t.Errorf("marker naming M after a warm admission: %v", err)
+ }
+}
+
+func TestStartCheckOriginAfterAFirstAdmission(t *testing.T) {
+ // The first admission left the origin: its predecessor is the origin, so
+ // an absent marker is lagging, not wrong.
+ c := startClaim("g1", hostclaim.AdmissionRecord{Generation: "g1"})
+ if err := StartCheck(c, nil, nil, ""); err != nil {
+ t.Fatalf("absent marker, predecessor the origin: %v", err)
+ }
+ // With no admission reaching the origin, absent is refused.
+ c = startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ if err := StartCheck(c, nil, nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("absent marker with a published generation = %v, want ErrStartMarker", err)
+ }
+}
+
+func TestStartCheckMarkerRefusals(t *testing.T) {
+ c := startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ for name, m := range map[string]*genselect.Marker{
+ "unrelated generation": startMarker("other"),
+ "another incarnation": {Op: "g2", Incarnation: "inc-2"},
+ "empty op": {Incarnation: "inc-1"},
+ "root as an op": startMarker(hostclaim.RootGeneration),
+ } {
+ if err := StartCheck(c, m, nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("%s = %v, want ErrStartMarker", name, err)
+ }
+ }
+ if err := StartCheck(&hostclaim.Claim{}, nil, nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("no holder = %v, want ErrStartMarker", err)
+ }
+ if err := StartCheck(nil, nil, nil, ""); !errors.Is(err, ErrStartMarker) {
+ t.Errorf("nil claim = %v, want ErrStartMarker", err)
+ }
+}
+
+func TestStartCheckFilesGen(t *testing.T) {
+ c := startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ m := startMarker("g2")
+ for name, fg := range map[string]*FilesGen{
+ "another op": {Op: "other", Incarnation: "inc-1"},
+ "another incarnation": {Op: "g2", Incarnation: "inc-2"},
+ "empty": {},
+ } {
+ if err := StartCheck(c, m, fg, ""); !errors.Is(err, ErrStartFiles) {
+ t.Errorf("%s = %v, want ErrStartFiles", name, err)
+ }
+ }
+ // The origin's own files carry no .files-gen, and a stray one is refused.
+ if err := StartCheck(startClaim(""), nil, &FilesGen{Op: "g1", Incarnation: "inc-1"}, ""); !errors.Is(err, ErrStartFiles) {
+ t.Errorf("files-gen on an origin claim = %v, want ErrStartFiles", err)
+ }
+}
+
+func TestStartCheckDBGen(t *testing.T) {
+ c := startClaim("g2", hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g1"})
+ m := startMarker("g2")
+ // The .gen is plain and exact: a lagging name is a warm database from an
+ // earlier stay, even though the chain would admit it elsewhere.
+ for _, g := range []string{"other", "g1", hostclaim.RootGeneration} {
+ if err := StartCheck(c, m, nil, g); !errors.Is(err, ErrStartDB) {
+ t.Errorf(".gen %q = %v, want ErrStartDB", g, err)
+ }
+ }
+ if err := StartCheck(c, m, nil, "g2"); err != nil {
+ t.Errorf(".gen g2: %v", err)
+ }
+ if err := StartCheck(startClaim(""), nil, nil, "g1"); !errors.Is(err, ErrStartDB) {
+ t.Errorf(".gen on an origin claim = %v, want ErrStartDB", err)
+ }
+}
+
+func TestStartCheckReportsTheFirstFailingMarker(t *testing.T) {
+ c := startClaim("g2")
+ err := StartCheck(c, startMarker("other"), &FilesGen{Op: "other", Incarnation: "inc-1"}, "other")
+ if !errors.Is(err, ErrStartMarker) || errors.Is(err, ErrStartFiles) || errors.Is(err, ErrStartDB) {
+ t.Errorf("err = %v, want only ErrStartMarker", err)
+ }
+}
+
+func TestPreparedCheck(t *testing.T) {
+ h := hostclaim.Holder{Incarnation: "inc-1", Location: "loc/", Generation: "g1"}
+ ok := &FilesGen{Op: "mv-1", Incarnation: "inc-1"}
+ if err := PreparedCheck("mv-1", h, ok, "mv-1"); err != nil {
+ t.Fatalf("both name the prepared generation: %v", err)
+ }
+ cases := []struct {
+ name string
+ prepared string
+ fg *FilesGen
+ db string
+ want error
+ }{
+ {"files-gen absent", "mv-1", nil, "mv-1", ErrStartFiles},
+ {"files-gen another op", "mv-1", &FilesGen{Op: "mv-2", Incarnation: "inc-1"}, "mv-1", ErrStartFiles},
+ {"files-gen another incarnation", "mv-1", &FilesGen{Op: "mv-1", Incarnation: "inc-2"}, "mv-1", ErrStartFiles},
+ {"gen absent", "mv-1", ok, "", ErrStartDB},
+ {"gen another op", "mv-1", ok, "mv-2", ErrStartDB},
+ {"gen the holder's generation", "mv-1", ok, "g1", ErrStartDB},
+ {"no prepared generation", "", &FilesGen{}, "", ErrStartFiles},
+ {"root as prepared", hostclaim.RootGeneration, &FilesGen{Op: hostclaim.RootGeneration, Incarnation: "inc-1"}, hostclaim.RootGeneration, ErrStartFiles},
+ }
+ for _, tc := range cases {
+ if err := PreparedCheck(tc.prepared, h, tc.fg, tc.db); !errors.Is(err, tc.want) {
+ t.Errorf("%s = %v, want %v", tc.name, err, tc.want)
+ }
+ }
+}
diff --git a/internal/registry/hold_test.go b/internal/registry/hold_test.go
index 34d016f56..688f87076 100644
--- a/internal/registry/hold_test.go
+++ b/internal/registry/hold_test.go
@@ -185,6 +185,70 @@ func TestDeleteIfHold(t *testing.T) {
}
}
+// The source's finalization (moves spec §7.9 step 3, ruling H5(d)): the row
+// and the host's timer rows go in one transaction, and only while the op
+// holds the row. Any refusal leaves both, and the neighbour's timers stay.
+func TestDeleteIfHoldWithTimers(t *testing.T) {
+ r := holdFixture(t, "h")
+ if err := r.Upsert(Route{Host: "n", Kind: "instance", Socket: "/n", App: "a", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ for _, tm := range []Timer{{Host: "h", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000},
+ {Host: "h", Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000},
+ {Host: "n", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}} {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ timers := func(host string) int {
+ t.Helper()
+ got, err := r.TimersFor(host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(got)
+ }
+ // Not held, held by another op, an empty op: nothing goes.
+ for _, c := range []struct{ name, hold, op string }{{"unheld", "", holdOpA}, {"another op", holdOpB, holdOpA}, {"empty op", holdOpA, ""}} {
+ if c.hold != "" {
+ if err := r.SetHold("h", "moving", c.hold); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", c.op); err != nil || ok {
+ t.Fatalf("%s: %v %v", c.name, ok, err)
+ }
+ if rt, _ := r.Get("h"); rt == nil || timers("h") != 2 {
+ t.Fatalf("%s: the row or its timers changed: %+v, %d timers", c.name, rt, timers("h"))
+ }
+ if c.hold != "" {
+ if ok, err := r.ReleaseHold("h", c.hold); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ }
+ }
+ if err := r.SetHold("h", "moving", holdOpA); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || !ok {
+ t.Fatalf("delete under the holding op: %v %v", ok, err)
+ }
+ if rt, _ := r.Get("h"); rt != nil || timers("h") != 0 {
+ t.Fatalf("after the delete: %+v, %d timers", rt, timers("h"))
+ }
+ if rt, _ := r.Get("n"); rt == nil || timers("n") != 1 {
+ t.Fatalf("the neighbour changed: %+v, %d timers", rt, timers("n"))
+ }
+ // A replay finds no row held by the op and deletes nothing, timers
+ // written since included.
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "late", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || ok || timers("h") != 1 {
+ t.Fatalf("replay: %v %v, %d timers", ok, err, timers("h"))
+ }
+}
+
func TestReleaseHoldEmptyOpMatchesNothing(t *testing.T) {
r := holdFixture(t, "h")
if ok, err := r.ReleaseHold("h", ""); err != nil || ok {
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index f8642d645..db75d890e 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -1148,6 +1148,36 @@ func (r *Registry) DeleteIfHold(host, op string) (bool, error) {
return n > 0, err
}
+// DeleteIfHoldWithTimers is the source's finalization (moves spec §7.9 step
+// 3, ruling H5(d)): host's row and every timer row of host are deleted in one
+// transaction, and only while op holds the row. false means nothing matched
+// and nothing was deleted, timers included, so a timers clear never runs
+// ahead of the row it belongs to or lands on a successor's rows.
+func (r *Registry) DeleteIfHoldWithTimers(host, op string) (bool, error) {
+ if op == "" {
+ return false, nil
+ }
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ res, err := tx.Exec(`DELETE FROM routes WHERE host = ? AND hold_op = ?`, host, op)
+ if err != nil {
+ return false, err
+ }
+ if n, err := res.RowsAffected(); err != nil || n == 0 {
+ return false, err
+ }
+ if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
+ return false, err
+ }
+ if err := tx.Commit(); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
// HoldRows returns every row held by an operation (a non-empty hold_op), ordered by
// host. Over a fresh connection, like List, so a reader in another process
// sees the latest commit.