amadan

carlos / platform Public

host: the source's finalization park after the commit (§7.9 step 3)

The source of a committed move parks its location through the park engine
(horizon from the flip, H7), retires a registered installation through
park.Local under the guard or parks unregistered artifacts by a moves-owned
journalled rename into the engine's layout (Paul's H5(b)), deletes its row
and timers rows in one registry transaction under WithHostHoldLock (H5(d)),
moves its config ack into the report prefix, stamps source_parked_at and
parks its op journal last. Post-flip records reach it through
movesParticipantRecord (H5(a)), which Plan 4 reuses with
parkLocalArtifacts. Unarmed in production until the park engine's leases
and Plan 5's view exist (G17).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1a9442f793e40f62337ad3680a5a623a0b868fb7 parent 302665b
5 files changed, +1793 −1
  • approved_moves_copy_test.go +1 −0
  • internal/host/movesfinal.go +935 −0
  • internal/host/movesfinal_test.go +828 −0
  • internal/host/movespass.go +26 −0
  • internal/host/movesphase1_harness_test.go +3 −1
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 3ff23c842..97371d3f9 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -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",
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
new file mode 100644
index 000000000..562786496
--- /dev/null
+++ b/internal/host/movesfinal.go
@@ -0,0 +1,935 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "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" // source_parked_at is stamped
+ 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()
+}
+
+func journalComponent(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+
+// 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. In production ParkLeases is not (ruling G17).
+func sourceFinalArmed(cfg moveCfg) bool {
+ return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" &&
+ cfg.ParkLeases != 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.Done(finalStepParked) {
+ if err := f.mark(finalStepParked, ""); 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.Done(finalStepParked) {
+ if err := f.stampParked(ctx); err != nil {
+ return err
+ }
+ if err := f.mark(finalStepParked, ""); 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. The stand-in authorizers use the same rule.
+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
+}
+
+// authorizeChore is park.Local.Authorize's stand-in for the source until
+// Plan 5: this move's own chore, for this box or this region, under sourceOf.
+func (f *sourceFinal) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
+ if err := f.sourceOf(hostclaim.Observed{Host: f.rec.Host, Claim: c}); err != nil {
+ return err
+ }
+ cfg := f.cfg()
+ switch {
+ case o.Op != f.g.ID:
+ case o.Kind == "park-local" && o.Box == cfg.Box:
+ return nil
+ case o.Kind == "park-replica" && o.Region == cfg.Region:
+ return nil
+ }
+ return fmt.Errorf("%w: not this source's chore", park.ErrUnsafe)
+}
+
+// authorizeObligation is park.Engine.Authorize's stand-in: the same rule
+// over an observation.
+func (f *sourceFinal) authorizeObligation(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ if obs.Absent {
+ return park.ErrUnsafe
+ }
+ return f.authorizeChore(ctx, obs.Claim, o)
+}
+
+// view is park.Engine.View's stand-in for the source's location park: the
+// target's region is authoritative after the commit, so this region's whole
+// location is retired and every version under it is selected (D6, H24: in
+// Phase 1 the scope is the root, instances/<host>, with every incarnation's
+// residue under it).
+func (f *sourceFinal) view(_ 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: f.cfg().Region, AuthoritativeRegion: f.rec.Move.To,
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
+}
+
+// engine is the park engine over this box's own bucket: the wired leases,
+// and the view and authorizer Plan 5 supplies, or the stand-ins.
+func (f *sourceFinal) engine(vs lease.VersionedStore) park.Engine {
+ cfg := f.cfg()
+ view, auth := cfg.ParkView, cfg.ParkAuthorize
+ if view == nil {
+ view = f.view
+ }
+ if auth == nil {
+ auth = f.authorizeObligation
+ }
+ return park.Engine{Store: vs, Claims: f.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
+ Now: f.now, NewID: obligationID, View: view, Authorize: auth, 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 {
+ local.Authorize = f.authorizeChore
+ }
+ 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"}
+}
+
+// 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
+// renames it; after it nothing of the move is left under .moves/.
+func (f *sourceFinal) parkJournal() error {
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ retirement := f.j.Steps[finalStepLocal]
+ if retirement == "" {
+ return fmt.Errorf("move %s: the journal records no local retirement for its own park", id)
+ }
+ plan, err := journalledLocalPlan(f.j, finalStepLocal)
+ if err != nil {
+ return err
+ }
+ dst := filepath.Join(parkedArtifactsDir(cfg.DataDir, plan.Incarnation, host, retirement), ".moves", host, id)
+ if f.j.Steps[finalStepJournal] != dst {
+ if err := f.mark(finalStepJournal, dst); err != nil {
+ return err
+ }
+ }
+ if err := parkOpJournal(cfg.DataDir, plan.Incarnation, host, id, 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, id)
+ return nil
+}
+
+// ---- the moves-owned local rename (ruling H5(b)) ----
+
+// 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"
+
+// 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 both parents synced; then Step records
+// the retirement id. A replay resumes from the journal:
+// - an artifact already at its destination with its identity is 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 {
+ 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 := syncPath(filepath.Dir(src)); err != nil {
+ return "", err
+ }
+ if err := syncPath(filepath.Dir(dst)); err != nil {
+ return "", err
+ }
+ if err := fault("after-rename:" + it.Path); err != nil {
+ return "", err
+ }
+ }
+ if err := syncPath(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 the parent of every directory
+// it made, so a rename into it survives a crash.
+func mkdirSynced(root, dir string) error {
+ var made []string
+ for d := dir; d != root && d != filepath.Dir(d); d = filepath.Dir(d) {
+ if _, err := os.Lstat(d); err == nil {
+ break
+ } else if !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ made = append(made, d)
+ }
+ if err := os.MkdirAll(dir, 0o770); err != nil {
+ return err
+ }
+ for _, d := range made {
+ if err := syncPath(filepath.Dir(d)); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// 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 is done; a destination already there 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) {
+ 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
+ }
+ if err := syncPath(filepath.Dir(src)); err != nil {
+ return err
+ }
+ return syncPath(filepath.Dir(dst))
+}
diff --git a/internal/host/movesfinal_test.go b/internal/host/movesfinal_test.go
new file mode 100644
index 000000000..168ffc9a2
--- /dev/null
+++ b/internal/host/movesfinal_test.go
@@ -0,0 +1,828 @@
+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"} {
+ 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.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(sourceArtifactPaths(p1Host), 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", 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", 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) != 6 || 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) != 6 {
+ 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-rename:" + p1Host + ".db", "after-rename:" + p1Host + ".db.gen", "after-rename:" + p1Host + ".db.corrupt",
+ "after-rename:" + p1Host, "after-rename:" + p1Host + ".env", "after-" + finalStepLocal,
+ "after-row-pending", "after-row-delete", "after-" + finalStepRowDeleted, "after-config-ack-copy", "after-" + finalStepConfigAck,
+ "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
+ 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)
+ h.cfg.ParkLeases = nil
+ if sourceFinalArmed(h.cfg) {
+ t.Fatal("armed without park leases")
+ }
+ h.sourcePass(t)
+ h.untouched(t, "unarmed")
+ 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 || 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")
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index b570dd677..dc823a37f 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() {
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