carlos / platform Public

Merge remote-tracking branch 'origin/main' into feat/hostclaim-region-ha

claude agent, operated by paul@keymail.dev 1cbcdab00eb9c399c21d3d1fa62c074a48b60958 parent 2ad9775parent ca591df
10 files changed, +756 −34
  • approved_moves_copy_test.go +1 −0
  • internal/host/movesfinal.go +97 −6
  • internal/host/movesland_test.go +7 −0
  • internal/host/moveslandcommit.go +35 −21
  • internal/host/movesparks.go +123 −0
  • internal/host/movesparks_test.go +399 −0
  • internal/host/movesrevert.go +17 −7
  • internal/moves/parkscope.go +29 −0
  • internal/moves/parkscope_test.go +46 −0
  • internal/provision/plan9_b8_census_test.go +2 −0

CI passed — run details

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 1cb1a1556..14d7bac4b 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -589,6 +589,7 @@ var movesScanFiles = []string{
"internal/host/movesmeasure.go",
"internal/host/movesmeasure_linux.go",
"internal/host/movesmeasure_other.go",
+ "internal/host/movesparks.go",
"internal/host/movespass.go",
"internal/host/movesphase1.go",
"internal/host/movespolicy.go",
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 2f7a05e3d..3c3ed9d32 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -42,7 +42,10 @@ import (
// 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;
+// did not record is ever moved or removed. The plan and the renames run
+// inside the host guard with the claim and the record read fresh there
+// (parkLocal; Plan 5 Task 3, carried follow-up (d)), so a call that
+// resumes after authority moved on renames nothing;
// 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
@@ -52,7 +55,11 @@ import (
// 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/.
+// retirement, so a finished source keeps nothing under .moves/. Then the
+// source tries the close (closeMove, movesparks.go): it closes only when
+// the target has opened and both parks are acknowledged already, and
+// this is the last pass that routes the record here, so otherwise the
+// close is the target's.
//
// 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
@@ -270,7 +277,7 @@ func (f *sourceFinal) run(ctx context.Context) error {
return err
}
}
- return f.parkJournal()
+ return f.parkJournalAndClose(ctx)
}
obs, err := hostclaim.Read(ctx, f.s, host)
if err != nil {
@@ -299,8 +306,7 @@ func (f *sourceFinal) run(ctx context.Context) error {
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 {
+ if err := f.parkLocal(ctx, holder, obs.Claim); err != nil {
return fmt.Errorf("park the source's local artifacts: %w", err)
}
if err := f.deleteRow(ctx); err != nil {
@@ -334,7 +340,16 @@ func (f *sourceFinal) run(ctx context.Context) error {
return err
}
}
- return f.parkJournal()
+ return f.parkJournalAndClose(ctx)
+}
+
+// parkJournalAndClose is step 4's end: the journal's park, then the close
+// when the target has opened and nothing is owed (closeMove).
+func (f *sourceFinal) parkJournalAndClose(ctx context.Context) error {
+ if err := f.parkJournal(); err != nil {
+ return err
+ }
+ return f.w.closeMove(ctx, f.s, f.rec, f.fault)
}
// sourceOf is the claim this finalization runs under: this move's landing,
@@ -487,6 +502,82 @@ func (f *sourceFinal) installation(ctx context.Context) (string, error) {
return "unregistered", nil
}
+// parkLocal is step 2's moves-owned rename (parkLocalArtifacts) of what the
+// source kept at the flat paths, its plan and its renames inside the host
+// guard (Plan 5 Task 3, carried follow-up (d); the pattern of Plan 4's
+// installer, installRun.guardedWith, and its rollback's parkLeftovers). In
+// the guard, with the claim and the record read fresh:
+// - the claim is still this move's commit with this box its source
+// (sourceOf), and this box does not execute the host;
+// - no inventory entry of this box claims a path the rename would move;
+// - the record is still this move's commit.
+//
+// The guard is the installation record's state: none (an unregistered
+// source with no record, WithNoInstallation), or the source's stamp, active
+// or retired by step 2's registered half (RetireInstallationWith, which
+// retires an active one first, so the guard refuses this box any later
+// write at it). A retirement already journalled is done and takes no guard.
+func (f *sourceFinal) parkLocal(ctx context.Context, holder hostclaim.Holder, c *hostclaim.Claim) error {
+ j := f.j
+ if _, done := j.Steps[finalStepLocal]; done {
+ return nil
+ }
+ cfg := f.cfg()
+ host, id := f.rec.Host, f.g.ID
+ paths := sourceArtifactPaths(host)
+ check := func(fresh *hostclaim.Claim) error {
+ if err := f.sourceOf(hostclaim.Observed{Host: host, Claim: fresh}); err != nil {
+ return fmt.Errorf("%w: %v", errFinalChanged, err)
+ }
+ if fresh.Op.Exec == cfg.Box {
+ return fmt.Errorf("%w: this box executes the host again", errFinalChanged)
+ }
+ for _, p := range paths {
+ if _, ok := fresh.Inventory[cfg.Box].Paths[p]; ok {
+ return fmt.Errorf("%w: the inventory claims %s", errFinalChanged, p)
+ }
+ }
+ rec, err := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
+ if err != nil {
+ return fmt.Errorf("re-read the record: %w", err)
+ }
+ if rec == nil || rec.Move == nil || rec.Move.ID != id || rec.Move.Incarnation != f.g.Incarnation || rec.Move.State != provision.MoveCommitted {
+ return errFinalChanged
+ }
+ return nil
+ }
+ work := func() error {
+ _, err := parkLocalArtifacts(ctx, j, f.mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: holder.Incarnation,
+ Host: host, Op: id, Step: finalStepLocal, Paths: paths, NewID: obligationID, Fault: f.fault})
+ return err
+ }
+ guard := cfg.Guard
+ if guard.Claims == nil {
+ guard.Claims = hostguard.StoreReader{Store: f.s}
+ }
+ // The guard's record directory: a Phase 1 source without host claims
+ // may not have one yet (the landing makes the same two, ensureLocalDirs).
+ for _, d := range []string{"hostguard", ".parked"} {
+ if err := os.MkdirAll(filepath.Join(guard.DataDir, d), 0o770); err != nil {
+ return err
+ }
+ }
+ captured, err := hostguard.Capture(guard.DataDir, host)
+ switch {
+ case errors.Is(err, hostguard.ErrNoRecord):
+ return guard.WithNoInstallation(ctx, host, f.g.Incarnation, check, work)
+ case err != nil:
+ return fmt.Errorf("the source's installation record: %w", err)
+ }
+ stamp := captured.Stamp
+ if stamp == 0 {
+ if stamp, err = strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64); err != nil || stamp == 0 {
+ return fmt.Errorf("move %s: the claim journals no source stamp", id)
+ }
+ }
+ return guard.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: f.g.Incarnation, Stamp: stamp}, check, work)
+}
+
// sourceArtifactPaths are the source's artifacts at the flat paths (G11),
// relative to the data directory: the installation's (installArtifactPaths:
// the database with its marker, journal, quarantine and litestream shadow,
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
index 7efa1d5fa..e4c39e4ab 100644
--- a/internal/host/movesland_test.go
+++ b/internal/host/movesland_test.go
@@ -90,6 +90,10 @@ type landingHarness struct {
releaseNoop bool
// releaseErr makes ReleaseHeld fail.
releaseErr error
+ // ownAuthorize drops the park authorizer the target would inherit from
+ // the source's configuration (the finalization harness sets one), so
+ // the target's parks use their own stand-ins (Plan 5 Task 3).
+ ownAuthorize bool
}
// fakeParkLeases refuses a second acquisition of a held location.
@@ -346,6 +350,9 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
if h.parkView == nil {
c.ParkView = nil
}
+ if h.ownAuthorize {
+ c.ParkAuthorize = nil
+ }
c.OwnLease = h.leases.Acquire
if h.store != nil {
c.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return h.store, nil }
diff --git a/internal/host/moveslandcommit.go b/internal/host/moveslandcommit.go
index 43a124116..2a564e71b 100644
--- a/internal/host/moveslandcommit.go
+++ b/internal/host/moveslandcommit.go
@@ -34,7 +34,12 @@ import (
// released (ReleaseHeld), so an edge reload in between keeps the entry
// (Task 5's open race). The entry from the probe is the serving process:
// it is never restarted;
-// - MoveOpened, the first time only.
+// - MoveOpened, the first time only;
+// - the staging park and the close (§7.9 steps 4 and 5, Plan 5 Task 3,
+// movesparks.go): the lane's park through laneParkTarget at
+// landing/committed, then closeMove. A claim that already records this
+// move's release (a crash between the close's two halves) skips the
+// rest and only clears the record.
//
// Unarmed in production with the rest of the landing: landingArmed fails
// there (ruling G17), and the open needs ReleaseHeld.
@@ -87,6 +92,15 @@ func (w *movesWorker) committedTarget(ctx context.Context, s release.Store, rec
return nil
}
l := &landingRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, guard: cfg.Guard}
+ obs, err := hostclaim.Read(ctx, s, rec.Host)
+ if err != nil {
+ return fmt.Errorf("read the host claim: %w", err)
+ }
+ if !obs.Absent && hostclaim.RequireReleasedMove(m.ID)(obs.Claim) == nil {
+ // The close released the claim and stopped before the record's
+ // half (§7.5): only that is left.
+ return w.closeMove(ctx, s, rec, l.fault)
+ }
if err := l.commitClaim(ctx); err != nil {
return err
}
@@ -95,17 +109,22 @@ func (w *movesWorker) committedTarget(ctx context.Context, s release.Store, rec
}
if cfg.ReleaseHeld == nil {
w.notice(rec.Host+"#open", "moves: %s: the open of %s waits: this box cannot release the hold", rec.Host, m.ID)
- return nil
+ } else {
+ stamped, err := l.open(ctx)
+ if err != nil {
+ return err
+ }
+ if stamped {
+ log.Printf("moves: %s: landing %s is open", rec.Host, m.ID)
+ nudgeSerial(ctx, s)
+ }
}
- stamped, err := l.open(ctx)
- if err != nil {
+ // §7.9 step 4: the lane's park the commit queued (movesparks.go).
+ if err := w.laneParkTarget(ctx, s, rec, hostclaim.KindLanding, hostclaim.PhaseCommitted); err != nil {
return err
}
- if stamped {
- log.Printf("moves: %s: landing %s is open", rec.Host, m.ID)
- nudgeSerial(ctx, s)
- }
- return nil
+ // §7.9 step 5: the close, once both stamps are in and nothing is owed.
+ return w.closeMove(ctx, s, rec, l.fault)
}
// commitClaim is the claim half of the commit (§7.5): written by whoever
@@ -145,20 +164,15 @@ func (l *landingRun) commitClaim(ctx context.Context) error {
return nil
}
-// commitScopes are the parks the commit queues, through Plan 5 Task 2's
-// scope helpers: the source region's location scoped from the flip (H7,
-// moves.SourceLocationScope), and the staging lane scoped to the seal's
-// bound (moves.StagingScope).
+// commitScopes are the parks the commit queues: moves.CommitScopes, the one
+// builder the console's pass uses too (Plan 5 Task 3, K9), so the box and
+// the console write the same obligations. The source region's location is
+// scoped from the flip (H7), the staging lane to the seal's bound.
func (l *landingRun) commitScopes(c *hostclaim.Claim) (source, staging hostclaim.ParkScope, err error) {
- m := l.rec.Move
- if m == nil || m.FlippedAt == nil {
- return source, staging, errors.New("the source's location park needs the flip's time")
- }
- seal, err := journalledSeal(c)
- if err != nil {
- return source, staging, err
+ if l.rec.Move == nil {
+ return source, staging, errors.New("the commit's parks need the record's move")
}
- return moves.SourceLocationScope(*c.Holder, *m.FlippedAt), moves.StagingScope(l.rec.Host, l.g.ID, c.Holder.Incarnation, seal.Bound()), nil
+ return moves.CommitScopes(c, l.rec.Move.FlippedAt)
}
// open is the conditional open, proved (fix round 1, finding 3): under the
diff --git a/internal/host/movesparks.go b/internal/host/movesparks.go
new file mode 100644
index 000000000..0e515bf47
--- /dev/null
+++ b/internal/host/movesparks.go
@@ -0,0 +1,123 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The target's staging park after the commit and the close (moves spec §7.9
+// steps 4 and 5; Plan 5 Task 3 with rulings K4, K6, K9 and K15), behind
+// CARLOS_MOVES_PHASE1. Both run in Plan 3's committed arm on the target
+// (committedTarget, moveslandcommit.go), routed as it is; there is no second
+// committed dispatcher.
+//
+// Step 4, the staging park. The commit queued the lane's park-replica for
+// the target region in its own write, scoped to moves/<host>/<op> with the
+// seal's bound as its evidence (moves.CommitScopes, the one builder the
+// console's pass uses too, K9). After the claim half and the open, every
+// pass while the record is committed runs it through laneParkTarget with the
+// claim at landing/committed (K6, the abort's and the revert's dispatcher):
+// the park engine over this box's own bucket archives the lane, answers
+// ErrPending until seal bound + SignatureAge + RequestLimit, and then
+// acknowledges on a clean listing. A late request found by the verification
+// listing is a further retirement event under the same scope.
+//
+// Step 5, the close (closeMove). Any participant that sees opened_at and
+// source_parked_at on the record and no obligation of the op in the claim
+// closes: the terminal event `done` (moves.LogEvent when the hook is wired,
+// Plan 6; a log line until then), then hostclaim.ReleaseMove to serving under
+// the target (§19 K8: the claim first), then provision.MoveClose (the record
+// last: LastMove{Outcome: "done"}, index D11). A crash between the two
+// leaves a finished move on the record that admits nothing (MoveRequest
+// refuses while Move is set), and the next pass of the target finds the
+// claim recording this move's release and writes only the record's close.
+// The target calls it every committed pass; the source calls it in the pass
+// that stamps source_parked_at, which is the last pass that routes the
+// record to it. The console's recovery pass also tries (Task 7).
+//
+// Production (K4). The staging park and the close are production-live only
+// after Plan 5 Task 9 (park arming) and claim Plans 10 and 11. The committed
+// arm runs only where the landing is armed (landingArmed), and production
+// wires neither the park engine's leases nor Plan 5's view and authorizer
+// until Task 9; the commit and the open themselves need the probe wake of
+// claim Plan 11's launcher; and the source's location park is acknowledged
+// by C4's regional runner, whose moves evidence is claim Plan 10's. Until
+// then the tests arm everything with fakes.
+
+// closeMove is the close (§7.9 step 5; see the file comment). fault is the
+// caller's test seam, passed "after-close-release" between the claim's
+// release and the record's close. It does nothing until the record carries
+// both stamps and the claim owes nothing under the op.
+func (w *movesWorker) closeMove(ctx context.Context, s release.Store, rec provision.Record, fault func(string) error) error {
+ host, id, inc := rec.Host, rec.Move.ID, rec.Move.Incarnation
+ fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, host)
+ if err != nil {
+ return fmt.Errorf("re-read the record: %w", err)
+ }
+ if fresh == nil {
+ return nil
+ }
+ m := fresh.Move
+ if m == nil || m.Legacy() || m.ID != id || m.Incarnation != inc || m.State != provision.MoveCommitted {
+ return nil
+ }
+ if m.OpenedAt == nil || m.SourceParkedAt == nil {
+ w.notice(host+"#close", "moves: %s: move %s closes once the target has opened and the source has parked (opened %t, parked %t)", host, id, m.OpenedAt != nil, m.SourceParkedAt != nil)
+ return nil
+ }
+ obs, err := hostclaim.Read(ctx, s, host)
+ if err != nil {
+ return fmt.Errorf("read the host claim: %w", err)
+ }
+ if obs.Absent {
+ return fmt.Errorf("move %s has no host claim", id)
+ }
+ c := obs.Claim
+ if hostclaim.RequireReleasedMove(id)(c) != nil {
+ if c.Op.ID != id || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || len(c.Op.Participants) != 2 {
+ w.notice(host+"#close", "moves: %s: move %s waits to close: the claim is %s/%q (%s), not this move's commit", host, id, c.Op.Kind, c.Op.Phase, c.Op.ID)
+ return nil
+ }
+ if c.Holder == nil || c.Holder.Incarnation != inc {
+ return fmt.Errorf("move %s: the host claim names another incarnation", id)
+ }
+ owed := 0
+ for _, o := range c.Obligations {
+ if o.Op == id {
+ owed++
+ }
+ }
+ if owed > 0 {
+ w.notice(host+"#close", "moves: %s: move %s waits for %d park(s) of the move to be acknowledged before it closes", host, id, owed)
+ return nil
+ }
+ w.logEvent(ctx, *fresh, "done")
+ target := c.Op.Participants[1]
+ if _, err := hostclaim.Update(ctx, s, obs, hostclaim.ReleaseMove(id, target)); err != nil {
+ return fmt.Errorf("release the claim: %w", err)
+ }
+ log.Printf("moves: %s: the claim is serving under %s after move %s", host, target, id)
+ if fault != nil {
+ if err := fault("after-close-release"); err != nil {
+ return err
+ }
+ }
+ }
+ g := provision.MoveGuard{ID: id, Incarnation: inc}
+ if _, err := provision.SetRecordMove(ctx, s, fresh.Account, fresh.App, host, provision.MoveClose(g, w.now())); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: the close of move %s refused (%v); the record moved on", host, id, err)
+ return nil
+ }
+ return fmt.Errorf("close the move: %w", err)
+ }
+ log.Printf("moves: %s: move %s is done; the instance is in %s", host, id, m.To)
+ nudgeSerial(ctx, s)
+ return nil
+}
diff --git a/internal/host/movesparks_test.go b/internal/host/movesparks_test.go
new file mode 100644
index 000000000..53f8e434a
--- /dev/null
+++ b/internal/host/movesparks_test.go
@@ -0,0 +1,399 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The target's staging park after the commit and the close (moves spec §7.9
+// steps 4 and 5; Plan 5 Task 3). The fixture is the source's finalization
+// harness (newFinalHarness: a landing driven to restored, the source box ie
+// with its real registry row and artifacts), given a probe so that the
+// TARGET box (uk) probes, commits and opens through Plan 3's committed arm,
+// not by hand. Both boxes run the real park engine over the fake versioned
+// buckets with one fake lease table. Nothing in production runs the
+// source region's acknowledgement yet (C4's regional runner), so
+// regionalAck stands in for it.
+
+func newCloseHarness(t *testing.T) *finalHarness {
+ t.Helper()
+ h := newFinalHarness(t)
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
+ rec.Move.CutAttempts = 2
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ p := &probeLog{t: t, h: h.landingHarness}
+ h.probe, h.probeLive = p.wake, p.liveProbe
+ h.ownAuthorize = true
+ return h
+}
+
+// chore finds the op's park-replica in region: the staging lane when lane
+// is true, the location otherwise.
+func (h *finalHarness) chore(t *testing.T, region string, lane bool) *hostclaim.Obligation {
+ t.Helper()
+ for _, o := range h.claim(t).Obligations {
+ if o.Op == p1ID && o.Kind == "park-replica" && o.Region == region && o.Park != nil && isStagingLocation(o.Park.Location) == lane {
+ return &o
+ }
+ }
+ return nil
+}
+
+// regionalAck is C4's regional runner for the source region: the park
+// engine over data-ie, run once at the harness's clock.
+func (h *finalHarness) regionalAck(t *testing.T, o *hostclaim.Obligation) error {
+ t.Helper()
+ e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-ie", Region: "ie", Leases: h.leases, Now: h.clock.Now,
+ NewID: obligationID, View: sourceTestView, Authorize: sourceTestAuthorize}
+ return e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID})
+}
+
+// clockTo moves the harness's clock to at, never back.
+func (h *finalHarness) clockTo(t *testing.T, at time.Time) {
+ t.Helper()
+ if d := at.Sub(h.clock.now); d > 0 {
+ h.clock.Advance(d)
+ }
+ if h.clock.now.After(at.Add(time.Second)) {
+ t.Fatalf("fixture: the clock is already at %v, past %v", h.clock.now, at)
+ }
+}
+
+// clockPast moves the harness's clock past at, if it is not there already.
+func (h *finalHarness) clockPast(at time.Time) {
+ if d := at.Sub(h.clock.now); d >= 0 {
+ h.clock.Advance(d + time.Second)
+ }
+}
+
+// committedAndOpen asserts the target committed and opened the landing.
+func (h *finalHarness) committedAndOpen(t *testing.T) {
+ t.Helper()
+ m := h.read(t, p1Host).Move
+ if m == nil || m.State != provision.MoveCommitted || m.OpenedAt == nil {
+ t.Fatalf("the target did not commit and open: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
+ t.Fatalf("the claim: %+v", c.Op)
+ }
+}
+
+// notClosed asserts the move is still open on both halves.
+func (h *finalHarness) notClosed(t *testing.T, why string) {
+ t.Helper()
+ if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveCommitted {
+ t.Fatalf("%s: the record closed: %+v", why, m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
+ t.Fatalf("%s: the claim was released: %+v", why, c.Op)
+ }
+}
+
+// done asserts the close: the claim serving under the target with this
+// move recorded as released and nothing owed, the record at rest in uk with
+// LastMove done.
+func (h *finalHarness) done(t *testing.T) {
+ t.Helper()
+ rec := h.read(t, p1Host)
+ if rec.Move != nil || rec.Region != "uk" || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "done" {
+ t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "uk-box" || hostclaim.RequireReleasedMove(p1ID)(c) != nil || len(c.Obligations) != 0 {
+ t.Fatalf("the claim after the close: %+v obligations %+v", c.Op, c.Obligations)
+ }
+}
+
+// TestStagingParkAcknowledgesAtSealBoundPlusH (Review Focus 1, witness
+// wStagingScopeDeadline; §7.9 step 4, K6): the commit queues the lane's
+// park scoped to the seal's bound T, and the target runs it through
+// laneParkTarget. At T + 16 min the run is ErrPending and nothing is
+// acknowledged; a restarted worker at T + 17 min + 1 s acknowledges, and
+// the scope it acknowledged is still T's (quiet_after = T + H).
+func TestStagingParkAcknowledgesAtSealBoundPlusH(t *testing.T) {
+ h := newCloseHarness(t)
+ h.land(t)
+ h.committedAndOpen(t)
+ seal, err := journalledSeal(h.claim(t))
+ if err != nil {
+ t.Fatal(err)
+ }
+ bound := seal.Bound()
+ o := h.chore(t, "uk", true)
+ if o == nil {
+ t.Fatal("the commit queued no staging park")
+ }
+ if want := moves.StagingScope(p1Host, p1ID, h.inc, bound); *o.Park != want || o.Owner != "region:uk" {
+ t.Fatalf("the staging park %+v %+v, want %+v", o, o.Park, want)
+ }
+ // The first pass archived the lane; nothing is acknowledged yet.
+ lane := moves.ScopeLocation(p1Host, p1ID) + "/"
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, lane) {
+ t.Fatalf("the lane still holds %s after the target's pass", k)
+ }
+ }
+ // T + 16 min: the engine answers ErrPending, and the pass acknowledges
+ // nothing.
+ h.clockTo(t, bound.Add(16*time.Minute))
+ e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now,
+ NewID: obligationID, View: standInView("uk", roleTarget, h.read(t, p1Host).Move), Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
+ if err := e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); !errors.Is(err, park.ErrPending) {
+ t.Fatalf("at T + 16 min the run answered %v, want ErrPending", err)
+ }
+ h.land(t)
+ if again := h.chore(t, "uk", true); again == nil || !again.Park.EvidenceAt.Equal(bound) {
+ t.Fatalf("at T + 16 min: %+v", again)
+ }
+ // A restarted target at T + 17 min + 1 s acknowledges.
+ h.freshTarget()
+ h.clockTo(t, bound.Add(park.SignatureAge+moves.ParkRequestLimit+time.Second))
+ h.land(t)
+ if again := h.chore(t, "uk", true); again != nil {
+ t.Fatalf("the staging park is still owed past T + H: %+v", again)
+ }
+ if q := h.claim(t).QuietAfter["uk"]; !q.Equal(bound.Add(park.SignatureAge + moves.ParkRequestLimit)) {
+ t.Fatalf("quiet_after %v, want T + H = %v", q, bound.Add(park.SignatureAge+moves.ParkRequestLimit))
+ }
+ if n := len(h.vs.callsOn("copy-version data-uk/parked/" + h.inc + "/" + p1Host + "/" + o.Retirement + "/moves/")); n == 0 {
+ t.Fatal("nothing was archived under the lane's retirement")
+ }
+}
+
+// TestCloseWaitsForBothObligations (§7.9 step 5): with the target open and
+// the source parked, the close still waits while the source region's
+// location park is owed; once both parks are acknowledged the claim is
+// serving under uk-box and the record at rest in uk.
+func TestCloseWaitsForBothObligations(t *testing.T) {
+ h := newCloseHarness(t)
+ h.land(t)
+ h.committedAndOpen(t)
+ h.sourcePass(t)
+ if m := h.read(t, p1Host).Move; m.SourceParkedAt == nil {
+ t.Fatalf("the source did not finalize: %+v", m)
+ }
+ h.land(t)
+ h.notClosed(t, "both parks owed")
+ // Past the staging lane's horizon the target acknowledges it; the
+ // source's location park is still owed (no regional runner ran it).
+ lane := h.chore(t, "uk", true)
+ h.clockPast(lane.Park.EvidenceAt.Add(park.SignatureAge + lane.Park.RequestLimit))
+ h.land(t)
+ if h.chore(t, "uk", true) != nil {
+ t.Fatal("the staging park was not acknowledged")
+ }
+ src := h.chore(t, "ie", false)
+ if src == nil {
+ t.Fatal("fixture: the source's location park was acknowledged by the finalization")
+ }
+ h.land(t)
+ h.notClosed(t, "the source's location park owed")
+ // The source region's runner acknowledges after its horizon; then the
+ // target's next pass closes.
+ h.clockPast(src.Park.EvidenceAt.Add(park.SignatureAge + src.Park.RequestLimit))
+ if err := h.regionalAck(t, src); err != nil {
+ t.Fatalf("the source region's acknowledgement: %v", err)
+ }
+ if h.chore(t, "ie", false) != nil {
+ t.Fatal("the source's location park was not acknowledged")
+ }
+ h.land(t)
+ h.done(t)
+ // A further pass of either box changes nothing.
+ before := h.claim(t).Gen
+ h.land(t)
+ h.sourcePass(t)
+ if after := h.claim(t).Gen; after != before {
+ t.Fatalf("a pass after the close wrote the claim (%d to %d)", before, after)
+ }
+}
+
+// TestCloseIsRecordLast (§7.5, §19 K8): the close releases the claim before
+// it clears the record. A crash between leaves a finished move on the
+// record, which admits nothing (no new move starts over it), and the next
+// pass clears it without writing the claim again.
+func TestCloseIsRecordLast(t *testing.T) {
+ h := newCloseHarness(t)
+ h.land(t)
+ h.sourcePass(t)
+ h.clock.Advance(24 * time.Hour)
+ if err := h.regionalAck(t, h.chore(t, "ie", false)); err != nil {
+ t.Fatal(err)
+ }
+ h.landAt = "after-close-release"
+ h.crashLand(t)
+ rec := h.read(t, p1Host)
+ if rec.Move == nil || rec.Move.State != provision.MoveCommitted || rec.LastMove != nil {
+ t.Fatalf("the record cleared before the claim released: %+v %+v", rec.Move, rec.LastMove)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "uk-box" || hostclaim.RequireReleasedMove(p1ID)(c) != nil {
+ t.Fatalf("the claim at the crash point: %+v", c.Op)
+ }
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRequest("ie", h.clock.Now())); !errors.Is(err, provision.ErrMoveConflict) {
+ t.Fatalf("a new move started over the finished one: %v", err)
+ }
+ gen := c.Gen
+ h.freshTarget()
+ h.land(t)
+ h.done(t)
+ if after := h.claim(t).Gen; after != gen {
+ t.Fatalf("the record's close wrote the claim again (%d to %d)", gen, after)
+ }
+}
+
+// Any participant that sees both stamps and nothing owed closes (§7.9 step
+// 5): the source's pass that stamps source_parked_at closes too when the
+// target has opened and both parks are already acknowledged.
+func TestTheSourceClosesWhenItFinishesLast(t *testing.T) {
+ h := newCloseHarness(t)
+ h.landFault = func(point string) error {
+ if point == "after-close-release" {
+ t.Fatal("the target released before the source parked")
+ }
+ return nil
+ }
+ h.land(t)
+ h.committedAndOpen(t)
+ h.clock.Advance(24 * time.Hour)
+ h.land(t)
+ if h.chore(t, "uk", true) != nil {
+ t.Fatal("the staging park was not acknowledged")
+ }
+ h.notClosed(t, "the source has not parked")
+ // The source's location park is acknowledged in the finalization's own
+ // run, past its horizon; the same pass stamps and closes.
+ h.sourcePass(t)
+ h.done(t)
+}
+
+// K9: the console's commit (Task 7a) builds its scopes from the record and
+// the claim alone, through moves.CommitScopes with FlippedAt, so whichever
+// of the box and the console commits first writes the same obligations,
+// byte for byte, and the other's replay is no change.
+func TestConsoleCommitMatchesTheBoxs(t *testing.T) {
+ h, _ := probedLanding(t)
+ var pre *hostclaim.Claim
+ h.landFault = func(point string) error {
+ switch point {
+ case "before-commit":
+ pre = h.claim(t)
+ case "after-claim-committed":
+ return errCrash
+ }
+ return nil
+ }
+ h.land(t)
+ if pre == nil || pre.Op.Phase != hostclaim.PhaseLandingProbed {
+ t.Fatalf("fixture: the claim before the commit: %+v", pre)
+ }
+ box := h.claim(t)
+ if box.Op.Phase != hostclaim.PhaseCommitted {
+ t.Fatalf("the box did not commit the claim: %+v", box.Op)
+ }
+ rec := h.read(t, p1Host)
+ source, staging, err := moves.CommitScopes(pre, rec.Move.FlippedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var minted []string
+ for _, o := range box.Obligations[len(pre.Obligations):] {
+ minted = append(minted, o.ID, o.Retirement)
+ }
+ newID := func() string {
+ if len(minted) == 0 {
+ t.Fatal("the console's commit minted more ids than the box's")
+ }
+ id := minted[0]
+ minted = minted[1:]
+ return id
+ }
+ console := pre
+ if err := hostclaim.Commit(p1ID, rec.Move.From, pre.Op.Intent["source"], rec.Move.To, source, staging, newID)(console); err != nil {
+ t.Fatal(err)
+ }
+ got, _ := json.Marshal(console.Obligations)
+ want, _ := json.Marshal(box.Obligations)
+ if !bytes.Equal(got, want) || len(minted) != 0 {
+ t.Fatalf("the console's commit queues\n%s\nthe box's\n%s", got, want)
+ }
+ // Without the flip's time there is no source scope (H7).
+ if _, _, err := moves.CommitScopes(pre, nil); err == nil {
+ t.Fatal("scopes without the flip")
+ }
+}
+
+// Carried follow-up (d) (Plan 3 Task 8; Task 3 rulings): the finalization's
+// moves-owned rename runs inside the host guard with the claim and the
+// record read fresh there, so authority that changes before it renames
+// nothing at the live paths and journals no plan.
+func TestSourceFinalizationRenameIsGuarded(t *testing.T) {
+ for name, change := range map[string]func(t *testing.T, h *finalHarness){
+ "the claim's executor is this box again": func(t *testing.T, h *finalHarness) {
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "ie-box" })
+ },
+ "the claim has released the move": func(t *testing.T, h *finalHarness) {
+ rewriteClaimOp(t, h.s, func(op *hostclaim.Op) {
+ *op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box", Journal: map[string]string{}}
+ })
+ },
+ "the inventory claims a path": func(t *testing.T, h *finalHarness) {
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
+ if c.Inventory == nil {
+ c.Inventory = map[string]hostclaim.InventoryEntry{}
+ }
+ c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 9}}
+ })
+ },
+ "the record is no longer this commit": func(t *testing.T, h *finalHarness) {
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
+ rec.Move.ID = p1ID2
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ h := newFinalHarness(t)
+ h.commit(t)
+ changed := false
+ inner := h.cfg.finalFault
+ h.cfg.finalFault = func(point string) error {
+ if point == "after-"+finalStepInstallation && !changed {
+ changed = true
+ change(t, h)
+ }
+ return inner(point)
+ }
+ h.sourcePass(t)
+ if !changed {
+ t.Fatalf("the rename's guard point was never reached: %v", h.finalHits)
+ }
+ want := []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}
+ if got := h.live(t); strings.Join(got, " ") != strings.Join(want, " ") {
+ t.Fatalf("live paths moved: %v, want %v", got, want)
+ }
+ if got := h.parked(t); len(got) != 0 {
+ t.Fatalf("parked: %v", got)
+ }
+ if j := h.sourceJournal(t); j.Done(finalStepLocal+"-plan") || j.Done(finalStepLocal) {
+ t.Fatalf("a plan was journalled: %v", j.Steps)
+ }
+ })
+ }
+}
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go
index 57537e4dd..a7894f5ab 100644
--- a/internal/host/movesrevert.go
+++ b/internal/host/movesrevert.go
@@ -840,12 +840,15 @@ func (w *movesWorker) sweepRevertJournal(dataDir, host, op string, j *moves.Jour
// participantRound): the lane's park-replica the claim owes this region, run
// by the park engine over this box's own bucket until it is acknowledged
// after its horizon. It serves a revert (the claim reverting, any phase; the
-// park hostclaim.Revert queued) and an abort (moving/aborting; the park
-// hostclaim.Abort queued, Plan 5 Task 1); Task 3 adds the committed landing.
-// No phases means any phase of kind. The view and the authorizer are Plan
-// 5's when wired, else the stand-ins: the source's region authoritative
-// (neither outcome flipped), and only this move's lane park under kind and
-// its phases.
+// park hostclaim.Revert queued), an abort (moving/aborting; the park
+// hostclaim.Abort queued, Plan 5 Task 1) and a committed landing
+// (landing/committed; the park hostclaim.Commit queued, Plan 5 Task 3, §7.9
+// step 4, called from the committed arm). No phases means any phase of
+// kind. The view and the authorizer are Plan 5's when wired, else the
+// stand-ins: the source's region authoritative (see the view below), and
+// only this move's lane park under kind and its phases, with this box not
+// executing the host unless the landing is committed (the target serves it
+// then).
func (w *movesWorker) laneParkTarget(ctx context.Context, s release.Store, rec provision.Record, kind string, phases ...string) error {
cfg := w.cfg
m := rec.Move
@@ -888,14 +891,21 @@ func (w *movesWorker) laneParkTarget(ctx context.Context, s release.Store, rec p
if err != nil {
return err
}
+ committed := kind == hostclaim.KindLanding
view, auth := cfg.ParkView, cfg.ParkAuthorize
if view == nil {
+ // The stand-in names From authoritative, after a commit too. For a
+ // lane, which is never the holder's location, the engine reads the
+ // authoritative region only to decide whether it must protect the
+ // location's live generations (park.protected), and a stand-in has
+ // no marker to protect them with. Plan 5's view (Task 9) answers
+ // the target's region after the commit, with its marker.
view = standInView(cfg.Region, roleTarget, m)
}
if auth == nil {
auth = func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
c := obs.Claim
- if obs.Absent || c == nil || !inOp(c.Op) || c.Op.Exec == cfg.Box ||
+ if obs.Absent || c == nil || !inOp(c.Op) || (!committed && c.Op.Exec == cfg.Box) ||
o.Op != id || o.Kind != "park-replica" || o.Region != cfg.Region || o.Park == nil || o.Park.Location != lane {
return fmt.Errorf("%w: not this move's lane park", park.ErrUnsafe)
}
diff --git a/internal/moves/parkscope.go b/internal/moves/parkscope.go
index f1611d6f0..bd2acdcbc 100644
--- a/internal/moves/parkscope.go
+++ b/internal/moves/parkscope.go
@@ -1,6 +1,8 @@
package moves
import (
+ "errors"
+ "fmt"
"strings"
"time"
@@ -72,3 +74,30 @@ func ReturnLaneScope(host, rbID, inc string, returnSealBound time.Time) hostclai
func laneScope(host, lane, inc string, bound time.Time) hostclaim.ParkScope {
return hostclaim.ParkScope{Location: ScopeLocation(host, lane), Incarnation: inc, EvidenceAt: bound.UTC(), RequestLimit: ParkRequestLimit}
}
+
+// CommitScopes are the two parks a commit queues (hostclaim.Commit), built
+// from the claim and the record's flip time alone (Plan 5 Task 3, ruling
+// K9): the source region's location from the flip (SourceLocationScope,
+// ruling H7) and the move's own lane to the seal the claim journals
+// (StagingScope with the seal's bound). The target box and the console's
+// pass (Task 7a) both call it, so whichever commits first writes the
+// obligations the other's replay finds; a scope built any other way would
+// be refused as "another scope" by the queue helpers. It refuses without
+// the flip's time, the holder or the journalled seal.
+func CommitScopes(c *hostclaim.Claim, flippedAt *time.Time) (source, staging hostclaim.ParkScope, err error) {
+ if c == nil || c.Holder == nil {
+ return source, staging, errors.New("a commit's parks need the claim's holder")
+ }
+ if flippedAt == nil || flippedAt.IsZero() {
+ return source, staging, errors.New("the source's location park needs the flip's time")
+ }
+ raw, ok := c.Op.Journal[hostclaim.JournalSeal]
+ if !ok {
+ return source, staging, fmt.Errorf("%w: the claim carries no seal", ErrBadSeal)
+ }
+ seal, err := DecodeSeal(raw)
+ if err != nil {
+ return source, staging, err
+ }
+ return SourceLocationScope(*c.Holder, *flippedAt), StagingScope(c.Host, c.Op.ID, c.Holder.Incarnation, seal.Bound()), nil
+}
diff --git a/internal/moves/parkscope_test.go b/internal/moves/parkscope_test.go
index 5c3ee56e1..64d524028 100644
--- a/internal/moves/parkscope_test.go
+++ b/internal/moves/parkscope_test.go
@@ -56,3 +56,49 @@ func TestScopeHelpersUseTheGenuineBound(t *testing.T) {
t.Fatal("a zero bound made a valid scope; the helper must not fill one in")
}
}
+
+// CommitScopes (Plan 5 Task 3, K9) is the one builder of a commit's two
+// parks, for the box and the console alike: the source's location from the
+// flip (SourceLocationScope) and the move's lane to the seal's bound
+// (StagingScope), read from the claim and the record's FlippedAt only. It
+// refuses without the flip's time, without a holder, and without a seal.
+func TestCommitScopes(t *testing.T) {
+ const host = "shop.bab.example.com"
+ inc := strings.Repeat("ab", 32)
+ seal := sealForTest()
+ flipped := seal.SealedAt.Add(40 * time.Second)
+ c := &hostclaim.Claim{Host: host,
+ Holder: &hostclaim.Holder{Location: "instances/" + host + "/" + inc + "/", Incarnation: inc},
+ Op: hostclaim.Op{ID: "move-1", Kind: hostclaim.KindLanding, Journal: map[string]string{hostclaim.JournalSeal: seal.Encode()}}}
+ source, staging, err := CommitScopes(c, &flipped)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if want := SourceLocationScope(*c.Holder, flipped); source != want {
+ t.Fatalf("source %+v, want %+v", source, want)
+ }
+ if want := StagingScope(host, "move-1", inc, seal.Bound()); staging != want {
+ t.Fatalf("staging %+v, want %+v", staging, want)
+ }
+ for name, edit := range map[string]func(c *hostclaim.Claim) *time.Time{
+ "no flip": func(*hostclaim.Claim) *time.Time { return nil },
+ "no holder": func(c *hostclaim.Claim) *time.Time { c.Holder = nil; return &flipped },
+ "no seal": func(c *hostclaim.Claim) *time.Time { delete(c.Op.Journal, hostclaim.JournalSeal); return &flipped },
+ "no claim": nil,
+ } {
+ t.Run(name, func(t *testing.T) {
+ cc := *c
+ cc.Op.Journal = map[string]string{hostclaim.JournalSeal: seal.Encode()}
+ if edit == nil {
+ if _, _, err := CommitScopes(nil, &flipped); err == nil {
+ t.Fatal("scopes without a claim")
+ }
+ return
+ }
+ at := edit(&cc)
+ if _, _, err := CommitScopes(&cc, at); err == nil {
+ t.Fatal("scopes built")
+ }
+ })
+ }
+}
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index 529492fd8..3b57c68a9 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -505,6 +505,8 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/host/moveslandcommit.go", "landingRun.probed", "record-writer", "key:SetRecordMove", "6f961790d8fe169af68e2f70790c400d37c2388a4f62c26921f22b8b47c859de"},
// Stamps opened_at through SetRecordMove (MoveOpened) inside the hold lock, after the proved release.
{"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "65e6908056c3055d5d0df2c27bb88943e31c08691a2173a090a9edb94c8c0b49"},
+ // Closes a committed move through SetRecordMove (MoveClose, LastMove done) after the claim's ReleaseMove; never writes the claim after the record.
+ {"internal/host/movesparks.go", "movesWorker.closeMove", "record-writer", "key:SetRecordMove", "849d114cf80a45004e9d01c66fcf11516381b046fe0f7c40294a628a3b922b90"},
// Closes the rollback record through SetRecordMove; claim release and local journal parking use separate storage.
{"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "fff2e28c78db68bccb16da6ab70ca877b51243b0aa56b9cb2c513c6f5660f7f7"},
// Advances the source recovery phase through SetRecordMove and its sequenced writer.