amadan

carlos / platform Public

host: the moves owners for a landing probe (Plan 3 Task 5c, H2, H12, H13, H14, H16, H18)

Select pins the launch inputs at spawn (release, env report), validates
the prepared inputs (PreparedCheck, the journalled manifest at the seal)
and decides the admission per §7.14: the first probe cold into the move
id from the build's pin, keeping the flip's stamp, sealed; a restarted
attempt into a fresh generation. AdmissionEvidence re-derives it before
the CAS. Materialize renders and materializes the held inputs under
their own selections, checks the render against the pins, then runs the
§5.1 start check. Publish writes the prepared marker with the build's
pin; PublishFenced a restarted attempt's after its root. Readiness is
the claim's probe check. Unregistered in production.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0f47cb508676323b07f64e32957f8b146c42427b parent a2e6a2e
3 files changed, +1526 −0
  • internal/host/movesprobe_ready_linux_test.go +42 −0
  • internal/host/movesprobeowners.go +762 −0
  • internal/host/movesprobeowners_test.go +722 −0
diff --git a/internal/host/movesprobe_ready_linux_test.go b/internal/host/movesprobe_ready_linux_test.go
new file mode 100644
index 000000000..3a4e48325
--- /dev/null
+++ b/internal/host/movesprobe_ready_linux_test.go
@@ -0,0 +1,42 @@
+//go:build linux
+
+package host
+
+import (
+ "net/http"
+ "os"
+ "os/exec"
+ "sync/atomic"
+ "testing"
+)
+
+// TestProbeRefusesForeignPeer (Review Focus 4, ruling H12): the moves owner
+// set binds no readiness of its own, so the probe's readiness is the claim's
+// (claimProbeSocketReady) on the path Select read from the app's manifest. A
+// listener in this test process answers 200 on that path, but its peer is
+// not in the spawned process's tree: the check refuses and the app's
+// listener is never asked.
+func TestProbeRefusesForeignPeer(t *testing.T) {
+ if (&movesProbeOwners{}).MovesOwners(&movesProbeAttempts{}).ProbeReady != nil {
+ t.Fatal("the moves owners replace the claim's probe readiness")
+ }
+ group := plan9D3SelfCgroup(t)
+ status := new(atomic.Int32)
+ status.Store(http.StatusOK)
+ rt, hits := plan9D3Server(t, status)
+ spawned := exec.Command("sleep", "5")
+ if err := spawned.Start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { spawned.Process.Kill(); spawned.Wait() })
+ if err := plan9D3Ready(t, plan9D3Cmd(t, spawned.Process.Pid), rt, "/ready", group); err == nil {
+ t.Fatal("a peer outside the spawned tree was accepted")
+ }
+ if hits.Load() != 0 {
+ t.Fatal("the foreign listener was asked")
+ }
+ // The same listener, answering as the spawned process, passes.
+ if err := plan9D3Ready(t, plan9D3Cmd(t, os.Getpid()), rt, "/ready", group); err != nil {
+ t.Fatalf("the spawned process itself: %v", err)
+ }
+}
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
new file mode 100644
index 000000000..5d15da4b4
--- /dev/null
+++ b/internal/host/movesprobeowners.go
@@ -0,0 +1,762 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The moves owners for a landing probe (moves Phase 1 Plan 3 Task 5c; claim
+// Plan 9 Group D, D5; rulings H2, H4, H13, H14, H16, H18). The claim's
+// dispatcher (ClaimOwnerDispatch) hands this owner set only the move kinds;
+// of those it serves exactly one shape, a landing probe: the claim at
+// landing/probing with this box executing as the move's target, the policy
+// a probe, the registry row held by the move. Everything else is refused
+// with hostbox.ErrUnsupported or hostbox.ErrChanged, before any lease, stop
+// or launch: the committed wake is Task 7's, the recoveries Plan 4's.
+//
+// The claim owns the admission, the restore, the launch, readiness,
+// replication and the publication sequence (claimwakeassembly.go). This
+// owner set supplies what moves knows:
+//
+// - Select (read-only, before admission): the launch-input pin at spawn
+// (§7.7, round-2 finding 14): the live release and the launch
+// environment's report are the pinned ones, else ProbeInputsChanged
+// with the recorded sentence "configuration changed during the move",
+// which the landing turns into its rollback (Task 6); then the
+// admission per §7.14. The first probe is a cold admission into the
+// prepared generation, the move id, after PreparedCheck (.files-gen and
+// <dbPath>.gen name it and the holder's incarnation, and the journalled
+// install manifest is the seal's N); its baseline is the build's pin
+// (genselect.SelectPrepared, never an earlier stay's marker, H14); it
+// keeps the flip's stamp and claims sealed coverage when the seal was
+// built from the stopped database. A restarted attempt, once the move
+// id is admitted, is a cold admission into a fresh generation: no kept
+// stamp, no coverage, its baseline the prepared pin while the move id
+// has published nothing and the certified remote (genselect.Select)
+// once it has (H14). The launch identity is the pinned config and
+// release (H4), CARLOS_MOVE_PROBE=1 is in the child's environment (H18)
+// and the readiness path is the app's moves.ready_path (H12).
+// - AdmissionEvidence: the same decisions, re-derived and compared,
+// against a baseline selected again, immediately before the CAS.
+// - Materialize, before every start, under the admitted spawn selection:
+// the held render and the held release, each under its own claim
+// selection, held-config and held-release, through the existing held
+// paths (appconfig.RenderHeldUnder, release.MaterializeHeldUnder) and
+// MovesClaimPolicy's probing rows (H13); the rendered etags must be the
+// pins (the authoritative check); then the §5.1 start check
+// (moves.StartCheck) against the claim after the admission's CAS.
+// - Publish: the prepared generation's marker, before replication: the
+// move id with the build's pin, superseding the current marker by its
+// exact ETag or created when there is none. PublishFenced: a restarted
+// attempt's fresh generation, after its own first root, as serving
+// publishes it. PublicationEvidence: this process's receipt for the
+// admitted session, read back.
+//
+// Production: nothing registers this owner set yet. The probe is unarmed in
+// production until Plan 5 Task 9 arms the parks; the tests use fakes.
+type movesProbeOwners struct {
+ // Claims is the claim store (the deployment store in production).
+ Claims release.Store
+ Records MovesRecordReader
+ // ReadyPath is the app manifest's moves.ready_path, or "/".
+ ReadyPath func(ctx context.Context, account, app string) (string, error)
+ Data genselect.Reader
+ Bucket, Region, Box string
+ // DataDir is the root of the op journals (moves.JournalDir).
+ DataDir string
+ FilesDir func(host string) (dir, envPath string)
+ Markers ClaimMarkerStore
+ Route func(host string) (*registry.Route, error)
+ // Launch is the held row's launch: the live executable, its digest,
+ // arguments, environment and version (the claim's launch snapshot).
+ Launch func(context.Context, registry.Route) (ClaimServingLaunch, error)
+ // EnvReport is moves.ReportEnv's digest of the held row's launch
+ // environment, the value Inputs.EnvReport pinned at readiness.
+ EnvReport func(context.Context, registry.Route) (string, error)
+ // Works selects a held action's claim work at the admitted capture
+ // (hostbox.WorkSource.SelectAt).
+ Works func(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error)
+ Render func(ctx context.Context, rt registry.Route, op string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error)
+ Release func(ctx context.Context, rt registry.Route, version string, work hostwrite.Work) error
+ NewID func() string
+ Now func() time.Time
+
+ mu sync.Mutex
+ published map[string]ClaimPublicationReceipt
+}
+
+var (
+ _ ClaimWakeInputs = (*movesProbeOwners)(nil)
+ _ ClaimFencedPublication = (*movesProbeOwners)(nil)
+)
+
+// ProbeInputsChanged is a probe refused because a launch input is not the
+// one pinned at readiness. Reason is the landing's recorded rollback
+// sentence (provision.RollbackConfigChanged); the landing records it.
+type ProbeInputsChanged struct{ Reason string }
+
+func (e *ProbeInputsChanged) Error() string {
+ return "host: the probe's launch inputs are not the pinned ones: " + e.Reason
+}
+
+func probeInputsChanged() error {
+ return &ProbeInputsChanged{Reason: provision.RollbackConfigChanged}
+}
+
+// owners is the owner set's ClaimKindOwners.
+func (m *movesProbeOwners) owners() ClaimKindOwners {
+ return ClaimKindOwners{Inputs: m, AdmissionEvidence: m.AdmissionEvidence, PublicationEvidence: m.PublicationEvidence}
+}
+
+// MovesOwners is ClaimMovesOwners for the probe: this owner set and ops.
+// ProbeReady stays nil, so the claim binds its own probe readiness check
+// (claimProbeSocketReady: the declared path, 2xx only, a peer in the
+// started role's cgroup; H12).
+func (m *movesProbeOwners) MovesOwners(ops OperationAttempts) ClaimMovesOwners {
+ return ClaimMovesOwners{ClaimKindOwners: m.owners(), Operations: ops}
+}
+
+func (m *movesProbeOwners) valid() bool {
+ return m != nil && m.Claims != nil && m.Records != nil && m.ReadyPath != nil && m.Data != nil && m.Bucket != "" && m.Region != "" && m.Box != "" &&
+ m.DataDir != "" && m.FilesDir != nil && m.Markers != nil && m.Route != nil && m.Launch != nil && m.EnvReport != nil && m.Works != nil &&
+ m.Render != nil && m.Release != nil && m.NewID != nil && m.Now != nil
+}
+
+// probeLanding is what one landing probe's checks read.
+type probeLanding struct {
+ op string
+ holder *hostclaim.Holder
+ rec *provision.Record
+ seal moves.Seal
+ manifest moves.InstallManifest
+}
+
+// landing checks that o is a landing probe this box may run and reads the
+// record, the seal and the journalled install manifest.
+func (m *movesProbeOwners) landing(ctx context.Context, o hostclaim.Observed) (probeLanding, error) {
+ var v probeLanding
+ if !m.valid() {
+ return v, hostwrite.ErrUnavailable
+ }
+ if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Op.Kind != hostclaim.KindLanding {
+ return v, hostbox.ErrUnsupported
+ }
+ c := o.Claim
+ if c.Host != o.Host || c.Op.ID == "" || c.Op.Phase != hostclaim.PhaseLandingProbing || c.Op.Exec != m.Box || c.Op.Intent["target"] != m.Box ||
+ c.Op.Intent["source"] == m.Box || c.Op.Intent["to"] != m.Region || len(c.Op.Participants) != 2 || c.Op.Participants[1] != m.Box {
+ return v, hostbox.ErrChanged
+ }
+ for _, lost := range c.Lost {
+ if lost == m.Box {
+ return v, hostbox.ErrChanged
+ }
+ }
+ v.op, v.holder = c.Op.ID, c.Holder
+ rec, err := m.Records(ctx, v.holder.Account, v.holder.App, o.Host)
+ if err != nil {
+ return v, err
+ }
+ if rec == nil || rec.Host != o.Host || rec.Move == nil {
+ return v, hostbox.ErrChanged
+ }
+ mv := rec.Move
+ if mv.ID != v.op || mv.Incarnation != v.holder.Incarnation || mv.State != provision.MoveLanding || mv.LandingPhase != provision.LandProbing || mv.Reason != "" || mv.To != m.Region || mv.Inputs == nil || mv.Inputs.Release == "" {
+ return v, hostbox.ErrChanged
+ }
+ v.rec = rec
+ if v.seal, err = journalledSeal(c); err != nil {
+ return v, err
+ }
+ j, err := moves.ReadJournal(m.DataDir, o.Host, v.op)
+ if err != nil {
+ return v, err
+ }
+ raw, ok := j.Steps[landStepInstall]
+ if !ok {
+ return v, fmt.Errorf("%w: the landing journalled no install manifest", hostbox.ErrChanged)
+ }
+ if err := json.Unmarshal([]byte(raw), &v.manifest); err != nil {
+ return v, fmt.Errorf("the journalled install manifest: %w", err)
+ }
+ if v.manifest.Op != v.op || v.manifest.Host != o.Host || v.manifest.N != v.seal.N || v.manifest.N == 0 || v.manifest.SnapshotKey == "" || v.manifest.SnapshotVersion == "" {
+ return v, fmt.Errorf("%w: the install manifest is not this move's build at the seal", hostbox.ErrChanged)
+ }
+ return v, nil
+}
+
+// heldRoute is the registry row the probe runs: this host, held by the move.
+func heldRoute(rt registry.Route, host, op string) error {
+ if rt.Host != host || !registry.IsHold(rt.Status) || rt.HoldOp != op || !filepath.IsAbs(rt.DBPath) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// opAdmitted reports whether the holder already admitted a writer into the
+// move id: a restarted attempt then admits a fresh generation.
+func opAdmitted(h *hostclaim.Holder, op string) bool {
+ if h.Generation == op {
+ return true
+ }
+ for _, a := range h.Admissions {
+ if a.Generation == op {
+ return true
+ }
+ }
+ return false
+}
+
+// predecessorRegion is AdmitWriter's predecessor region: the last
+// admission's, or the move's source region before any.
+func predecessorRegion(h *hostclaim.Holder, from string) string {
+ if n := len(h.Admissions); n > 0 {
+ return h.Admissions[n-1].Region
+ }
+ return from
+}
+
+func (m *movesProbeOwners) request(host string, h *hostclaim.Holder) genselect.Request {
+ return genselect.Request{Bucket: m.Bucket, Region: m.Region, Host: host, Holder: *h}
+}
+
+func preparedPin(v probeLanding) genselect.PreparedPin {
+ return genselect.PreparedPin{Generation: v.op, SnapshotKey: v.manifest.SnapshotKey, SnapshotVersion: v.manifest.SnapshotVersion, SnapshotTXID: v.manifest.N, SnapshotCRC64: v.manifest.SnapshotCRC64}
+}
+
+func markerKey(h *hostclaim.Holder) string {
+ return strings.TrimSuffix(h.Location, "/") + "/generation.json"
+}
+
+// currentMarker is the location's current generation.json version, or nil.
+func (m *movesProbeOwners) currentMarker(ctx context.Context, h *hostclaim.Holder) (*claimMarkerVersion, error) {
+ key := markerKey(h)
+ versions, err := m.Data.ListVersions(ctx, m.Bucket, key)
+ if err != nil {
+ return nil, err
+ }
+ var current *claimMarkerVersion
+ for _, v := range versions {
+ if v.Key == key && v.IsLatest {
+ if current != nil || v.DeleteMarker {
+ return nil, hostbox.ErrChanged
+ }
+ current = &claimMarkerVersion{ID: v.VersionID, ETag: v.ETag}
+ }
+ }
+ return current, nil
+}
+
+// baseline is the admission's baseline (H14): the build's pin for the
+// prepared generation; for a restarted attempt, the prepared pin while the
+// location has no marker (the move id published nothing beyond the build)
+// and the certified remote once it has one.
+func (m *movesProbeOwners) baseline(ctx context.Context, host string, v probeLanding, prepared bool) (*genselect.Plan, error) {
+ if prepared {
+ return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
+ }
+ current, err := m.currentMarker(ctx, v.holder)
+ if err != nil {
+ return nil, err
+ }
+ if current == nil {
+ return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
+ }
+ return genselect.Select(ctx, m.Data, m.request(host, v.holder))
+}
+
+// localMarkers reads .files-gen (nil when absent) and <dbPath>.gen ("" when
+// absent).
+func (m *movesProbeOwners) localMarkers(host, dbPath string) (*moves.FilesGen, string, error) {
+ dir, _ := m.FilesDir(host)
+ if !filepath.IsAbs(dir) {
+ return nil, "", hostwrite.ErrUnavailable
+ }
+ var files *moves.FilesGen
+ raw, err := os.ReadFile(filepath.Join(dir, archive.MarkerName))
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ case err != nil:
+ return nil, "", err
+ default:
+ var fg moves.FilesGen
+ dec := json.NewDecoder(bytes.NewReader(raw))
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&fg); err != nil {
+ return nil, "", fmt.Errorf("%w: %v", moves.ErrStartFiles, err)
+ }
+ files = &fg
+ }
+ db, err := hostbox.ReadGenerationMarker(dbPath + ".gen")
+ if err != nil {
+ return nil, "", err
+ }
+ return files, db, nil
+}
+
+// launchIdentity is the probe's spawn identity (H4): the pinned config and
+// release, bound to the record by MovesClaimPolicy.
+func launchIdentity(in *provision.MoveInputs) hostwrite.Identity {
+ return hostwrite.Identity{Config: HeldConfigIdentity(in.BundleETags, in.OverlayETag).Config, Release: in.Release}
+}
+
+// probeEnv is the launch environment with CARLOS_MOVE_PROBE=1 (H18), the
+// one entry for that name.
+func probeEnv(env []string) []string {
+ out := make([]string, 0, len(env)+1)
+ for _, kv := range env {
+ if !strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
+ out = append(out, kv)
+ }
+ }
+ return append(out, "CARLOS_MOVE_PROBE=1")
+}
+
+// Select chooses the probe's launch inputs and admission. It is read-only.
+func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
+ if !p.Probe {
+ return ClaimLaunchInputs{}, hostbox.ErrUnsupported
+ }
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ pins := v.rec.Move.Inputs
+ if (p.Inputs != nil && !reflect.DeepEqual(*p.Inputs, *pins)) || (p.Prepared != nil && *p.Prepared != v.manifest) {
+ return ClaimLaunchInputs{}, hostbox.ErrChanged
+ }
+ if err := heldRoute(rt, o.Host, v.op); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ h := v.holder
+ prepared := !opAdmitted(h, v.op)
+ admission := hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
+ KeepStamp: h.Stamp > h.Session, Region: m.Region, PredecessorRegion: predecessorRegion(h, v.rec.Move.From)}
+ if prepared {
+ files, db, err := m.localMarkers(o.Host, rt.DBPath)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ if err := moves.PreparedCheck(v.op, *h, files, db); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ admission.NewGeneration = v.op
+ if v.seal.FromDB {
+ admission.Coverage = "sealed"
+ }
+ } else {
+ admission.NewGeneration = m.NewID()
+ if !claimFreshGeneration(h, admission.NewGeneration) {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ }
+ plan, err := m.baseline(ctx, o.Host, v, prepared)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ admission.RemoteMaxTXID = plan.MaxTXID
+ ready, err := m.ReadyPath(ctx, h.Account, h.App)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ // The launch-input pin at spawn: the authoritative check, after
+ // readiness's (Plan 2's configured), since the inputs can change between.
+ l, err := m.Launch(ctx, rt)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ if l.Version != pins.Release {
+ return ClaimLaunchInputs{}, probeInputsChanged()
+ }
+ report, err := m.EnvReport(ctx, rt)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ if report == "" || report != pins.EnvReport {
+ return ClaimLaunchInputs{}, probeInputsChanged()
+ }
+ return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), Baseline: plan, Admission: admission,
+ Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: probeEnv(l.Env),
+ DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit, Prepared: prepared, ReadyPath: ready}, nil
+}
+
+// AdmissionEvidence re-derives Select's admission against the observation
+// and a baseline selected again, immediately before the admission's CAS.
+func (m *movesProbeOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return err
+ }
+ h := v.holder
+ if in.ObservedOpID != v.op || in.ObservedPhase != o.Claim.Op.Phase || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != m.Box || in.Region != m.Region ||
+ in.Mode != "cold" || in.PredecessorCertified || in.LossID != "" || len(in.AdditionalWritesLost) != 0 || in.KeepStamp != (h.Stamp > h.Session) ||
+ in.PredecessorRegion != predecessorRegion(h, v.rec.Move.From) || baseline == nil || in.RemoteMaxTXID != baseline.MaxTXID {
+ return hostbox.ErrChanged
+ }
+ prepared := !opAdmitted(h, v.op)
+ switch {
+ case prepared:
+ coverage := ""
+ if v.seal.FromDB {
+ coverage = "sealed"
+ }
+ if in.NewGeneration != v.op || in.Coverage != coverage {
+ return hostbox.ErrChanged
+ }
+ rt, err := m.route(o.Host, v.op)
+ if err != nil {
+ return err
+ }
+ files, db, err := m.localMarkers(o.Host, rt.DBPath)
+ if err != nil {
+ return err
+ }
+ if err := moves.PreparedCheck(v.op, *h, files, db); err != nil {
+ return err
+ }
+ default:
+ if in.NewGeneration == v.op || !claimFreshGeneration(h, in.NewGeneration) || in.Coverage != "" {
+ return hostbox.ErrChanged
+ }
+ }
+ fresh, err := m.baseline(ctx, o.Host, v, prepared)
+ if err != nil {
+ return err
+ }
+ if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+func (m *movesProbeOwners) route(host, op string) (registry.Route, error) {
+ rt, err := m.Route(host)
+ if err != nil {
+ return registry.Route{}, err
+ }
+ if rt == nil {
+ return registry.Route{}, hostbox.ErrChanged
+ }
+ if err := heldRoute(*rt, host, op); err != nil {
+ return registry.Route{}, err
+ }
+ return *rt, nil
+}
+
+// admittedProbe checks that o is the admitted probe session in's admission
+// made: this box writes the admission's new generation.
+func admittedProbe(o hostclaim.Observed, in ClaimLaunchInputs, box string) error {
+ h := o.Claim.Holder
+ if h.WriterBox != box || h.Generation != in.Admission.NewGeneration || h.Generation == "" {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// heldIdentityWork runs its work under the held write's own identity.
+type heldIdentityWork struct {
+ Work hostwrite.Work
+ ID hostwrite.Identity
+}
+
+func (w heldIdentityWork) With(ctx context.Context, _ hostwrite.Identity, fn func() error) error {
+ if w.Work == nil {
+ return hostwrite.ErrUnavailable
+ }
+ return w.Work.With(ctx, w.ID, fn)
+}
+
+// Materialize runs before every start of the admitted probe: the held
+// render and release under their own selections, then the start check.
+func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return err
+ }
+ if err := admittedProbe(o, in, m.Box); err != nil {
+ return err
+ }
+ if work == nil {
+ return hostwrite.ErrUnavailable
+ }
+ pins := v.rec.Move.Inputs
+ if in.Identity != launchIdentity(pins) {
+ return hostbox.ErrChanged
+ }
+ // The admitted spawn selection still holds under the launch identity.
+ if err := work.With(ctx, in.Identity, func() error { return nil }); err != nil {
+ return err
+ }
+ rt, err := m.route(o.Host, v.op)
+ if err != nil {
+ return err
+ }
+ if rt.DBPath != in.DBPath {
+ return hostbox.ErrChanged
+ }
+ h := v.holder
+ captured := hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}
+ hr, err := m.Render(ctx, rt, v.op, func(pre appconfig.HeldRender) (hostwrite.Work, error) {
+ w, err := m.Works(ctx, o.Host, movesActionHeldConfig, captured)
+ if err != nil {
+ return nil, err
+ }
+ return heldIdentityWork{Work: w, ID: HeldConfigIdentity(pre.BundleETags, pre.OverlayETag)}, nil
+ })
+ if err != nil {
+ return err
+ }
+ if !sameBundle(hr.BundleETags, pins.BundleETags) || hr.OverlayETag != pins.OverlayETag {
+ return probeInputsChanged()
+ }
+ released, err := m.Works(ctx, o.Host, movesActionHeldRelease, captured)
+ if err != nil {
+ return err
+ }
+ if err := m.Release(ctx, rt, pins.Release, heldIdentityWork{Work: released, ID: hostwrite.Identity{Release: pins.Release}}); err != nil {
+ return err
+ }
+ return m.startCheck(ctx, o, in)
+}
+
+// startCheck is §5.1's check against the claim after the admission's CAS,
+// read fresh: generation.json from this region's location, .files-gen and
+// <dbPath>.gen from this box.
+func (m *movesProbeOwners) startCheck(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
+ fresh, err := hostclaim.Read(ctx, m.Claims, o.Host)
+ if err != nil {
+ return err
+ }
+ if !claimSameWriter(o, fresh, hostclaim.SessionContinues) {
+ return hostbox.ErrChanged
+ }
+ marker, err := m.readMarker(ctx, fresh.Claim.Holder)
+ if err != nil {
+ return err
+ }
+ files, db, err := m.localMarkers(o.Host, in.DBPath)
+ if err != nil {
+ return err
+ }
+ return moves.StartCheck(fresh.Claim, marker, files, db)
+}
+
+// readMarker is the location's current generation.json, nil when absent.
+func (m *movesProbeOwners) readMarker(ctx context.Context, h *hostclaim.Holder) (*genselect.Marker, error) {
+ current, err := m.currentMarker(ctx, h)
+ if err != nil || current == nil {
+ return nil, err
+ }
+ raw, err := m.Data.GetVersion(ctx, m.Bucket, markerKey(h), current.ID)
+ if err != nil {
+ return nil, err
+ }
+ var marker genselect.Marker
+ if err := json.Unmarshal(raw, &marker); err != nil {
+ return nil, fmt.Errorf("%w: %v", moves.ErrStartMarker, err)
+ }
+ return &marker, nil
+}
+
+// EmptyAuthority: a move never admits the empty bootstrap.
+func (m *movesProbeOwners) EmptyAuthority(context.Context, hostclaim.Observed, ClaimLaunchInputs) error {
+ return hostbox.ErrUnsupported
+}
+
+// Publish writes the prepared generation's marker before replication: the
+// move id with the build's pin. A marker this box already wrote for the
+// move id is a replay and records the receipt again.
+func (m *movesProbeOwners) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return err
+ }
+ if err := admittedProbe(o, in, m.Box); err != nil {
+ return err
+ }
+ h := v.holder
+ if !in.Prepared || in.Admission.NewGeneration != v.op {
+ return hostbox.ErrChanged
+ }
+ pin := genselect.Pin{SnapshotKey: v.manifest.SnapshotKey, SnapshotVersion: v.manifest.SnapshotVersion, SnapshotTXID: v.manifest.N, SnapshotCRC64: v.manifest.SnapshotCRC64}
+ root := genselect.File{Key: v.manifest.SnapshotKey, VersionID: v.manifest.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: v.manifest.N}
+ bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
+ defer cancel()
+ current, err := m.currentMarker(bounded, h)
+ if err != nil {
+ return err
+ }
+ if current != nil {
+ if marker, err := m.readMarker(bounded, h); err == nil && marker != nil && marker.Op == v.op && marker.Box == m.Box && marker.Incarnation == h.Incarnation && marker.Host == o.Host && marker.Pin == pin {
+ m.record(o.Host, h.Session, ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root})
+ return nil
+ }
+ }
+ body, err := json.Marshal(genselect.Marker{Op: v.op, Incarnation: h.Incarnation, Host: o.Host, Box: m.Box, InstalledAt: m.Now().UTC(), Pin: pin})
+ if err != nil {
+ return err
+ }
+ expected := ""
+ if current != nil {
+ expected = current.ETag
+ }
+ version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, func(ctx context.Context) error { return ctx.Err() })
+ if err != nil {
+ return err
+ }
+ if version == "" {
+ return hostwrite.ErrUnavailable
+ }
+ m.record(o.Host, h.Session, ClaimPublicationReceipt{MarkerVersion: version, Root: root})
+ return nil
+}
+
+// PublishFenced writes a restarted attempt's fresh generation marker after
+// the session's own first root, under the publisher's fence, as the claim's
+// serving publisher does (claimServingOwners.PublishFenced).
+func (m *movesProbeOwners) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if err := admittedProbe(o, in, m.Box); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if pub.Sync == nil || pub.Check == nil {
+ return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
+ }
+ h := v.holder
+ if in.Prepared || h.Generation == v.op {
+ return ClaimPublicationReceipt{}, hostbox.ErrChanged
+ }
+ if err := pub.Sync(ctx); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
+ defer cancel()
+ plan, err := genselect.Select(bounded, m.Data, m.request(o.Host, h))
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ // Only this session's bound replica client writes under the fresh
+ // generation, so its root is the session's own acknowledged upload.
+ if plan.Generation != h.Generation || plan.Root == nil || plan.Root.MinTXID != 1 || plan.Root.VersionID == "" {
+ return ClaimPublicationReceipt{}, hostbox.ErrChanged
+ }
+ root := *plan.Root
+ current, err := m.currentMarker(bounded, h)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if current != nil && plan.Marker != nil && plan.MarkerVersion == current.ID && plan.Marker.Op == h.Generation {
+ if plan.Marker.Box != m.Box || plan.Marker.Incarnation != h.Incarnation {
+ return ClaimPublicationReceipt{}, hostbox.ErrChanged
+ }
+ receipt := ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}
+ m.record(o.Host, h.Session, receipt)
+ return receipt, nil
+ }
+ body, err := json.Marshal(genselect.Marker{Op: h.Generation, Incarnation: h.Incarnation, Host: o.Host, Box: m.Box, InstalledAt: m.Now().UTC()})
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ expected := ""
+ if current != nil {
+ expected = current.ETag
+ }
+ if err := pub.Check(bounded); err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, pub.Check)
+ if err != nil {
+ return ClaimPublicationReceipt{}, err
+ }
+ if version == "" {
+ return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
+ }
+ receipt := ClaimPublicationReceipt{MarkerVersion: version, Root: root}
+ m.record(o.Host, h.Session, receipt)
+ return receipt, nil
+}
+
+func (m *movesProbeOwners) record(host string, session uint64, receipt ClaimPublicationReceipt) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if m.published == nil {
+ m.published = map[string]ClaimPublicationReceipt{}
+ }
+ m.published[claimPublishedKey(host, session)] = receipt
+}
+
+// PublicationEvidence requires this process's receipt for the admitted
+// session: the current marker is exactly the version it wrote, naming the
+// holder's generation, incarnation, host and this box (and, for the move
+// id, the build's pin), and the root it named still exists, current.
+func (m *movesProbeOwners) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
+ v, err := m.landing(ctx, o)
+ if err != nil {
+ return err
+ }
+ h := v.holder
+ if h.WriterBox != m.Box {
+ return hostbox.ErrChanged
+ }
+ m.mu.Lock()
+ receipt, ok := m.published[claimPublishedKey(o.Host, h.Session)]
+ m.mu.Unlock()
+ prefix, err := hostbox.GenerationPrefix(*h)
+ if !ok || err != nil || receipt.MarkerVersion == "" || receipt.Root.VersionID == "" || !strings.HasPrefix(receipt.Root.Key, prefix) {
+ return hostbox.ErrChanged
+ }
+ plan, err := genselect.Select(ctx, m.Data, m.request(o.Host, h))
+ if err != nil {
+ return err
+ }
+ if plan.Generation != h.Generation || plan.Marker == nil || plan.MarkerVersion != receipt.MarkerVersion || plan.Marker.Op != h.Generation || plan.Marker.Box != m.Box || plan.Marker.Incarnation != h.Incarnation || plan.Marker.Host != o.Host {
+ return hostbox.ErrChanged
+ }
+ if h.Generation == v.op && (plan.Marker.SnapshotKey != v.manifest.SnapshotKey || plan.Marker.SnapshotVersion != v.manifest.SnapshotVersion || plan.Marker.SnapshotTXID != v.manifest.N) {
+ return hostbox.ErrChanged
+ }
+ versions, err := m.Data.ListVersions(ctx, m.Bucket, receipt.Root.Key)
+ if err != nil {
+ return err
+ }
+ exact, current := false, false
+ for _, ver := range versions {
+ if ver.Key != receipt.Root.Key || ver.DeleteMarker {
+ continue
+ }
+ exact = exact || ver.VersionID == receipt.Root.VersionID
+ current = current || ver.IsLatest
+ }
+ if !exact || !current {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
diff --git a/internal/host/movesprobeowners_test.go b/internal/host/movesprobeowners_test.go
new file mode 100644
index 000000000..5b07d432d
--- /dev/null
+++ b/internal/host/movesprobeowners_test.go
@@ -0,0 +1,722 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
+)
+
+const (
+ poHost = "shop.bab.oncarlos.com"
+ poOp = "mv_ownersprobe"
+)
+
+// poWork is a selected work that records the identities it runs under.
+type poWork struct {
+ action string
+ f *poFixture
+ shut bool
+}
+
+func (w poWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
+ w.f.order = append(w.f.order, w.action+":"+id.Config+"|"+id.Release)
+ if w.shut {
+ return hostwrite.ErrUnavailable
+ }
+ return fn()
+}
+
+type poFixture struct {
+ t *testing.T
+ claims release.Store
+ bucket *leasetest.VersionedBucket
+ rec *provision.Record
+ dataDir, files string
+ route registry.Route
+ manifest moves.InstallManifest
+ launch ClaimServingLaunch
+ envReport string
+ renderETags map[string]string
+ renderOverlay string
+ order []string
+ owners *movesProbeOwners
+ seq int
+ launches, reads int
+ seal moves.Seal
+ inc string
+ snapshotVersion string
+ sourceGeneration string
+}
+
+// poFlipClaim is the claim as the flip leaves it (moves §7.5): landing at
+// probing with uk-box executing; the holder is still the source's session 15
+// in generation "src" on ie-box, at the flip's stamp 18.
+func (f *poFixture) poFlipClaim() *hostclaim.Claim {
+ f.t.Helper()
+ journal := map[string]string{hostclaim.JournalSeal: f.seal.Encode(), hostclaim.JournalCutSession: "15", hostclaim.JournalEvidenceSource: "moves/bab/jam/" + poHost + "/" + poOp + "/stopped-source.json", hostclaim.JournalSourceStamp: "15", hostclaim.JournalSrcGen: "src"}
+ c := &hostclaim.Claim{V: 2, Host: poHost, Gen: 19,
+ Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + poHost + ".json", Account: "bab", App: "jam", CreatedAt: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC), Incarnation: f.inc, Location: "instances/" + poHost, Generation: "src", Stamp: 18, Session: 15, WriterBox: "ie-box",
+ Admissions: []hostclaim.AdmissionRecord{{Session: 15, Box: "ie-box", Generation: "src", Predecessor: "", Region: "ie", Mode: "cold"}}},
+ Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: poOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}, Journal: journal}}
+ if err := c.Validate(); err != nil {
+ f.t.Fatal(err)
+ }
+ return c
+}
+
+func newPOFixture(t *testing.T) *poFixture {
+ t.Helper()
+ at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+ inc, err := hostclaim.Incarnation("bab", "jam", poHost, at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f := &poFixture{t: t, claims: release.NewDirStore(t.TempDir()), bucket: leasetest.NewVersionedBucket(), dataDir: t.TempDir(), files: t.TempDir(), inc: inc}
+ f.bucket.AddBucket("data")
+ f.seal = moves.Seal{V: moves.SealVersion, N: 42, FromDB: true, SrcCovered: true, SealedAt: at, LeaseUntil: at.Add(time.Minute),
+ Replica: []moves.SealedObject{{Key: "moves/" + poHost + "/" + poOp + "/replica/0009/" + ltx.FormatFilename(1, 42), VersionID: "v1", Size: int64(ltx.HeaderSize)}},
+ Files: []moves.SealedObject{{Key: "moves/" + poHost + "/" + poOp + "/files/archive.sealed", VersionID: "v2", Size: 10}, {Key: "moves/" + poHost + "/" + poOp + "/files/archive.json", VersionID: "v3", Size: 10}}}
+ f.putClaim(f.poFlipClaim())
+ // The prepared generation the build wrote, and its manifest in the op
+ // journal (Plan 2 Task 8).
+ key := "instances/" + poHost + "/g-" + poOp + "/0009/" + ltx.FormatFilename(1, 42)
+ f.snapshotVersion = f.bucket.Put("data", key, f.ltxHeader(1, 42))
+ f.manifest = moves.InstallManifest{V: 1, Op: poOp, Host: poHost, N: 42, SnapshotKey: key, SnapshotVersion: f.snapshotVersion, SnapshotCRC64: 7, DBSHA256: strings.Repeat("d", 64)}
+ j, err := moves.ReadJournal(f.dataDir, poHost, poOp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, _ := json.Marshal(f.manifest)
+ j.Mark(landStepInstall, string(body))
+ if err := j.Write(f.dataDir); err != nil {
+ t.Fatal(err)
+ }
+ f.route = registry.Route{Host: poHost, Kind: "instance", App: "jam", Account: "bab", Status: "landing", HoldOp: poOp, Socket: filepath.Join(f.dataDir, poHost+".sock"), DBPath: filepath.Join(f.dataDir, poHost+".db")}
+ f.writeLocal(poOp, poOp, inc)
+ pins := mpPins()
+ pins.EnvReport = "sha256:env"
+ f.rec = &provision.Record{Host: poHost, Account: "bab", App: "jam", Region: "uk",
+ Move: &provision.MoveState{ID: poOp, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, Landing: &provision.MoveLandingJournal{Attempts: 1}}}
+ f.launch = ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:snapshot", Release: "sha256:bytes"}, Executable: "/opt/carlos/live/" + poHost, ExecutableSHA256: strings.Repeat("b", 64), Args: []string{"--socket", f.route.Socket, "--db", f.route.DBPath}, Env: []string{"A=1", "CARLOS_MOVE_PROBE=0"}, Version: pins.Release}
+ f.envReport = pins.EnvReport
+ f.renderETags, f.renderOverlay = pins.BundleETags, pins.OverlayETag
+ f.owners = &movesProbeOwners{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: "uk", Box: "uk-box", DataDir: f.dataDir,
+ Records: func(context.Context, string, string, string) (*provision.Record, error) { f.reads++; return f.rec, nil },
+ ReadyPath: func(context.Context, string, string) (string, error) { return "/healthz", nil },
+ FilesDir: func(string) (string, string) { return f.files, filepath.Join(f.dataDir, poHost+".env") },
+ Markers: plan9D5Markers{f.bucket},
+ Route: func(string) (*registry.Route, error) { rt := f.route; return &rt, nil },
+ Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
+ f.launches++
+ return f.launch, nil
+ },
+ EnvReport: func(context.Context, registry.Route) (string, error) { return f.envReport, nil },
+ Works: func(_ context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error) {
+ f.order = append(f.order, "select:"+action)
+ return poWork{action: action, f: f}, nil
+ },
+ Render: func(ctx context.Context, rt registry.Route, op string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) {
+ f.order = append(f.order, "render")
+ w, err := work(appconfig.HeldRender{BundleETags: f.renderETags, OverlayETag: f.renderOverlay})
+ if err != nil {
+ return appconfig.HeldRender{}, err
+ }
+ if err := w.With(ctx, hostwrite.Identity{Config: "rendered-content-identity"}, func() error { return nil }); err != nil {
+ return appconfig.HeldRender{}, err
+ }
+ return appconfig.HeldRender{Path: "p", SHA256: "s", BundleETags: f.renderETags, OverlayETag: f.renderOverlay}, nil
+ },
+ Release: func(ctx context.Context, rt registry.Route, version string, work hostwrite.Work) error {
+ f.order = append(f.order, "release:"+version)
+ return work.With(ctx, hostwrite.Identity{Release: "ignored"}, func() error { return nil })
+ },
+ NewID: func() string { f.seq++; return fmt.Sprintf("fresh%d", f.seq) },
+ Now: func() time.Time { return time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC) }}
+ return f
+}
+
+func (f *poFixture) ltxHeader(min, max uint64) []byte {
+ h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}
+ body, err := h.MarshalBinary()
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return body
+}
+
+// writeLocal writes the two local markers the landing left: .files-gen
+// naming files/inc and <dbPath>.gen naming db ("" removes it).
+func (f *poFixture) writeLocal(files, db, inc string) {
+ f.t.Helper()
+ body, _ := json.Marshal(moves.FilesGen{Op: files, Incarnation: inc})
+ if err := os.WriteFile(filepath.Join(f.files, archive.MarkerName), body, 0o600); err != nil {
+ f.t.Fatal(err)
+ }
+ if db == "" {
+ os.Remove(f.route.DBPath + ".gen")
+ return
+ }
+ if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", db); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+func (f *poFixture) putClaim(c *hostclaim.Claim) {
+ f.t.Helper()
+ body, err := json.Marshal(c)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ key, err := hostclaim.Key(poHost)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ etag := ""
+ if _, current, err := f.claims.Get(f.t.Context(), key); err == nil {
+ etag = current
+ }
+ if _, err := f.claims.Put(f.t.Context(), key, body, etag); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+func (f *poFixture) read() hostclaim.Observed {
+ f.t.Helper()
+ o, err := hostclaim.Read(f.t.Context(), f.claims, poHost)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return o
+}
+
+var poProbe = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
+
+// admit applies the selected admission with the claim's own CAS.
+func (f *poFixture) admit(o hostclaim.Observed, in ClaimLaunchInputs) hostclaim.Observed {
+ f.t.Helper()
+ admitted, err := hostclaim.Update(f.t.Context(), f.claims, o, hostclaim.AdmitWriter(in.Admission))
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return admitted
+}
+
+// TestProbeIsAColdAdmissionIntoThePreparedGeneration (owner half; the
+// runtime half is TestMovesProbeOwnersThroughTheClaimRuntime): the first
+// probe's admission is cold, into the move id, keeping the flip's stamp,
+// sealed, from the prepared pin; the restarted attempt's is a fresh
+// generation that keeps no stamp and claims no coverage.
+func TestProbeIsAColdAdmissionIntoThePreparedGeneration(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ o := f.read()
+ in, err := f.owners.Select(ctx, f.route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := hostclaim.Admission{ObservedOpID: poOp, ObservedPhase: hostclaim.PhaseLandingProbing, ObservedSession: 15, ObservedGeneration: "src", Box: "uk-box", Mode: "cold", NewGeneration: poOp, KeepStamp: true, Coverage: "sealed", Region: "uk", PredecessorRegion: "ie", RemoteMaxTXID: 42}
+ if !reflect.DeepEqual(in.Admission, want) {
+ t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
+ }
+ if !in.Prepared || in.Empty || in.Baseline == nil || in.Baseline.Generation != poOp || in.Baseline.MaxTXID != 42 || in.Baseline.Pin == nil || in.Baseline.Pin.SnapshotVersion != f.snapshotVersion {
+ t.Fatalf("baseline: prepared=%v %+v", in.Prepared, in.Baseline)
+ }
+ if in.Identity != mpLaunchIdentity(f.rec.Move.Inputs) || in.ReadyPath != "/healthz" || in.DBPath != f.route.DBPath || in.Version != f.launch.Version || in.Executable != f.launch.Executable || len(in.Roles) != 1 || in.Roles[0] != "main" {
+ t.Fatalf("launch inputs: %+v", in)
+ }
+ probe := 0
+ for _, kv := range in.Env {
+ if strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
+ probe++
+ if kv != "CARLOS_MOVE_PROBE=1" {
+ t.Fatalf("env %q", kv)
+ }
+ }
+ }
+ if probe != 1 || !contains(in.Env, "A=1") {
+ t.Fatalf("env: %v", in.Env)
+ }
+ if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
+ t.Fatal(err)
+ }
+ for name, mutate := range map[string]func(*hostclaim.Admission, *genselect.Plan){
+ "no coverage": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Coverage = "" },
+ "stamp not kept": func(a *hostclaim.Admission, _ *genselect.Plan) { a.KeepStamp = false },
+ "another region": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Region = "ie" },
+ "another box": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Box = "ie-box" },
+ "a stale baseline": func(a *hostclaim.Admission, p *genselect.Plan) { p.MaxTXID, a.RemoteMaxTXID = 41, 41 },
+ "another session": func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedSession = 14 },
+ "warm": func(a *hostclaim.Admission, _ *genselect.Plan) { a.Mode = "warm" },
+ "another new gen": func(a *hostclaim.Admission, _ *genselect.Plan) { a.NewGeneration = "other" },
+ "a loss": func(a *hostclaim.Admission, _ *genselect.Plan) { a.LossID = "loss" },
+ "certified": func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorCertified = true },
+ "predecessor where": func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorRegion = "uk" },
+ } {
+ a, p := in.Admission, *in.Baseline
+ mutate(&a, &p)
+ if err := f.owners.AdmissionEvidence(ctx, o, a, &p); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("%s: %v", name, err)
+ }
+ }
+
+ // The first attempt is admitted; the next attempt over it is a fresh
+ // generation with no kept stamp and no coverage, restored from the
+ // prepared pin while the move id published nothing.
+ admitted := f.admit(o, in)
+ h := admitted.Claim.Holder
+ if h.Generation != poOp || h.Stamp != 18 || h.WriterBox != "uk-box" || h.Admissions[len(h.Admissions)-1].Coverage != "sealed" {
+ t.Fatalf("first admission: %+v", h)
+ }
+ again, err := f.owners.Select(ctx, f.route, admitted, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if a := again.Admission; a.NewGeneration != "fresh1" || a.KeepStamp || a.Coverage != "" || a.ObservedGeneration != poOp || a.ObservedSession != h.Session || a.Mode != "cold" || a.PredecessorRegion != "uk" || again.Prepared {
+ t.Fatalf("restarted admission: %+v prepared=%v", a, again.Prepared)
+ }
+ if again.Baseline == nil || again.Baseline.Generation != poOp || again.Baseline.MaxTXID != 42 {
+ t.Fatalf("restarted baseline: %+v", again.Baseline)
+ }
+ if err := f.owners.AdmissionEvidence(ctx, admitted, again.Admission, again.Baseline); err != nil {
+ t.Fatal(err)
+ }
+ reused := again.Admission
+ reused.NewGeneration = poOp
+ if err := f.owners.AdmissionEvidence(ctx, admitted, reused, again.Baseline); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("the move id twice: %v", err)
+ }
+}
+
+func mpLaunchIdentity(p *provision.MoveInputs) hostwrite.Identity {
+ return hostwrite.Identity{Config: HeldConfigIdentity(p.BundleETags, p.OverlayETag).Config, Release: p.Release}
+}
+
+func contains(xs []string, x string) bool {
+ for _, v := range xs {
+ if v == x {
+ return true
+ }
+ }
+ return false
+}
+
+// TestProbeRefusesUnheldRowOrForeignExec: the row must be held by the
+// move, and the claim must name this box the executor of the move's
+// landing at probing; anything else is refused before any launch is read.
+func TestProbeRefusesUnheldRowOrForeignExec(t *testing.T) {
+ for name, tc := range map[string]struct {
+ route func(*registry.Route)
+ claim func(*hostclaim.Claim)
+ rec func(*provision.Record)
+ policy func(*activator.StartPolicy)
+ }{
+ "a ready row": {route: func(r *registry.Route) { r.Status, r.HoldOp = "ready", "" }},
+ "held by another op": {route: func(r *registry.Route) { r.HoldOp = "mv_other" }},
+ "another host's row": {route: func(r *registry.Route) { r.Host = "other.bab.oncarlos.com" }},
+ "a foreign executor": {claim: func(c *hostclaim.Claim) { c.Op.Exec = "ie-box" }},
+ "not the target": {claim: func(c *hostclaim.Claim) { c.Op.Intent["target"] = "ie-box" }},
+ "another region": {claim: func(c *hostclaim.Claim) { c.Op.Intent["to"] = "ie" }},
+ "not probing": {claim: func(c *hostclaim.Claim) { c.Op.Phase = provision.LandRestored }},
+ "not a landing": {claim: func(c *hostclaim.Claim) { c.Op.Kind = hostclaim.KindReverting }},
+ "declared lost": {claim: func(c *hostclaim.Claim) { c.Lost = []string{"uk-box"} }},
+ "a marked record": {rec: func(r *provision.Record) { r.Move.Reason = provision.RollbackProbeAttempts }},
+ "a record past probing": {rec: func(r *provision.Record) { r.Move.LandingPhase = provision.LandProbed }},
+ "no pinned inputs": {rec: func(r *provision.Record) { r.Move.Inputs = nil }},
+ "not a probe": {policy: func(p *activator.StartPolicy) { p.Probe = false }},
+ "other pins": {policy: func(p *activator.StartPolicy) { p.Inputs = &provision.MoveInputs{Release: "v0"} }},
+ "another manifest": {policy: func(p *activator.StartPolicy) { p.Prepared = &moves.InstallManifest{Op: poOp, N: 41} }},
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newPOFixture(t)
+ if tc.claim != nil {
+ c := f.poFlipClaim()
+ tc.claim(c)
+ f.putClaim(c)
+ }
+ if tc.route != nil {
+ tc.route(&f.route)
+ }
+ if tc.rec != nil {
+ tc.rec(f.rec)
+ }
+ p := poProbe
+ if tc.policy != nil {
+ tc.policy(&p)
+ }
+ if _, err := f.owners.Select(t.Context(), f.route, f.read(), p); err == nil {
+ t.Fatal("selected")
+ }
+ if f.launches != 0 {
+ t.Fatal("the launch was read")
+ }
+ })
+ }
+}
+
+// The prepared inputs are validated before any admission: the local markers
+// name the move id and the holder's incarnation, and the journalled
+// manifest is the seal's.
+func TestMovesProbeOwnersValidateThePreparedInputs(t *testing.T) {
+ for name, tc := range map[string]struct {
+ mutate func(*poFixture)
+ want error
+ }{
+ "files of another op": {func(f *poFixture) { f.writeLocal("mv_other", poOp, f.inc) }, moves.ErrStartFiles},
+ "files of another incarnation": {func(f *poFixture) { f.writeLocal(poOp, poOp, "other") }, moves.ErrStartFiles},
+ "no files marker": {func(f *poFixture) {
+ os.Remove(filepath.Join(f.files, archive.MarkerName))
+ }, moves.ErrStartFiles},
+ "a database of another op": {func(f *poFixture) { f.writeLocal(poOp, "src", f.inc) }, moves.ErrStartDB},
+ "no database marker": {func(f *poFixture) { f.writeLocal(poOp, "", f.inc) }, moves.ErrStartDB},
+ "a manifest at another N": {func(f *poFixture) {
+ f.manifest.N = 41
+ j, _ := moves.ReadJournal(f.dataDir, poHost, poOp)
+ body, _ := json.Marshal(f.manifest)
+ j.Mark(landStepInstall, string(body))
+ j.Write(f.dataDir)
+ }, hostbox.ErrChanged},
+ "no manifest": {func(f *poFixture) {
+ j, _ := moves.ReadJournal(f.dataDir, poHost, poOp)
+ delete(j.Steps, landStepInstall)
+ j.Write(f.dataDir)
+ }, hostbox.ErrChanged},
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newPOFixture(t)
+ tc.mutate(f)
+ if _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe); !errors.Is(err, tc.want) {
+ t.Fatalf("got %v, want %v", err, tc.want)
+ }
+ })
+ }
+}
+
+// The launch-input pin at spawn (§7.7, round-2 finding 14): the live
+// release and the launch environment's report are the ones pinned at
+// readiness, else the probe fails with the recorded rollback sentence
+// before any admission.
+func TestMovesProbeOwnersPinTheLaunchInputs(t *testing.T) {
+ for name, mutate := range map[string]func(*poFixture){
+ "another release": func(f *poFixture) { f.launch.Version = "v9" },
+ "another environment": func(f *poFixture) { f.envReport = "sha256:other" },
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newPOFixture(t)
+ mutate(f)
+ _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe)
+ var changed *ProbeInputsChanged
+ if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged {
+ t.Fatalf("got %v", err)
+ }
+ })
+ }
+}
+
+// TestFirstProbeFromTheSourceGenerationIsAdmitted: from the claim as the
+// flip left it, the prepared inputs validate, the admission names the move
+// id, and Materialize, before every start, renders and materializes the held
+// inputs under their own selections and passes the §5.1 start check after
+// the CAS. A hand-built claim already in the move id whose files name
+// another op is refused there.
+func TestFirstProbeFromTheSourceGenerationIsAdmitted(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ o := f.read()
+ in, err := f.owners.Select(ctx, f.route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(o, in)
+ f.order = nil
+ spawn := poWork{action: "spawn", f: f}
+ if err := f.owners.Materialize(ctx, admitted, in, spawn); err != nil {
+ t.Fatal(err)
+ }
+ pins := f.rec.Move.Inputs
+ launch := mpLaunchIdentity(pins)
+ want := []string{
+ "spawn:" + launch.Config + "|" + launch.Release,
+ "render", "select:held-config", "held-config:" + HeldConfigIdentity(pins.BundleETags, pins.OverlayETag).Config + "|",
+ "select:held-release", "release:" + pins.Release, "held-release:|" + pins.Release,
+ }
+ if !reflect.DeepEqual(f.order, want) {
+ t.Fatalf("order:\n got %q\nwant %q", f.order, want)
+ }
+
+ // A render that is not the pinned configuration fails the probe and
+ // materializes nothing.
+ f.order = nil
+ f.renderOverlay = "e-changed"
+ err = f.owners.Materialize(ctx, admitted, in, spawn)
+ var changed *ProbeInputsChanged
+ if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged || contains(f.order, "select:held-release") {
+ t.Fatalf("a changed render: %v %v", err, f.order)
+ }
+ f.renderOverlay = pins.OverlayETag
+
+ // The start check against the claim after the CAS.
+ f.writeLocal("mv_other", poOp, f.inc)
+ if err := f.owners.Materialize(ctx, admitted, in, spawn); !errors.Is(err, moves.ErrStartFiles) {
+ t.Fatalf("files of another op: %v", err)
+ }
+ f.writeLocal(poOp, "src", f.inc)
+ if err := f.owners.Materialize(ctx, admitted, in, spawn); !errors.Is(err, moves.ErrStartDB) {
+ t.Fatalf("a database the restore did not write: %v", err)
+ }
+ f.writeLocal(poOp, poOp, f.inc)
+ f.bucket.Put("data", "instances/"+poHost+"/generation.json", mustJSON(t, genselect.Marker{Op: "elsewhere", Incarnation: f.inc, Host: poHost, Box: "ie-box", InstalledAt: time.Unix(1, 0).UTC()}))
+ if err := f.owners.Materialize(ctx, admitted, in, spawn); !errors.Is(err, moves.ErrStartMarker) {
+ t.Fatalf("a marker the claim does not expect: %v", err)
+ }
+ // The spawn selection itself refused: nothing is rendered.
+ f.order = nil
+ if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f, shut: true}); !errors.Is(err, hostwrite.ErrUnavailable) || contains(f.order, "render") {
+ t.Fatalf("a refused spawn selection: %v %v", err, f.order)
+ }
+}
+
+func mustJSON(t *testing.T, v any) []byte {
+ t.Helper()
+ b, err := json.Marshal(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+// The prepared marker (moves §7.14 marker row, R10-3): the first probe's
+// admission publishes generation.json naming the move id with the build's
+// pin, superseding an earlier stay's marker by its exact ETag, before
+// replication; the evidence is this process's receipt.
+func TestMovesProbeOwnersPublishThePreparedMarker(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ key := "instances/" + poHost + "/generation.json"
+ f.bucket.Put("data", key, mustJSON(t, genselect.Marker{Op: "earlier", Incarnation: f.inc, Host: poHost, Box: "uk-box", InstalledAt: time.Unix(1, 0).UTC(), Pin: genselect.Pin{SnapshotTXID: 3}}))
+ o := f.read()
+ in, err := f.owners.Select(ctx, f.route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(o, in)
+ if err := f.owners.PublicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("evidence before publication: %v", err)
+ }
+ if err := f.owners.Publish(ctx, admitted, in); err != nil {
+ t.Fatal(err)
+ }
+ body, ok := f.bucket.Current("data", key)
+ var m genselect.Marker
+ if !ok || json.Unmarshal(body, &m) != nil || m.Op != poOp || m.Box != "uk-box" || m.Incarnation != f.inc || m.Host != poHost ||
+ m.Pin != (genselect.Pin{SnapshotKey: f.manifest.SnapshotKey, SnapshotVersion: f.manifest.SnapshotVersion, SnapshotTXID: 42, SnapshotCRC64: 7}) {
+ t.Fatalf("marker: %s", body)
+ }
+ if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
+ t.Fatal(err)
+ }
+ written := f.bucket.LatestID("data", key)
+ if err := f.owners.Publish(ctx, admitted, in); err != nil || f.bucket.LatestID("data", key) != written {
+ t.Fatalf("replay: %v", err)
+ }
+ // A restarted attempt's baseline is now the published move id.
+ again, err := f.owners.Select(ctx, f.route, admitted, poProbe)
+ if err != nil || again.Baseline.Generation != poOp || again.Baseline.Marker == nil || again.Baseline.Marker.Op != poOp {
+ t.Fatalf("restarted baseline after publication: %+v %v", again.Baseline, err)
+ }
+ f.bucket.Put("data", key, mustJSON(t, genselect.Marker{Op: poOp, Incarnation: f.inc, Host: poHost, Box: "ie-box", InstalledAt: time.Unix(2, 0).UTC(), Pin: m.Pin}))
+ if err := f.owners.PublicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("evidence for a marker this session did not write: %v", err)
+ }
+ if err := f.owners.EmptyAuthority(ctx, admitted, in); err == nil {
+ t.Fatal("a move admission claimed the empty bootstrap")
+ }
+}
+
+// A restarted attempt's fresh generation is published fenced, after its own
+// first root, as serving publishes.
+func TestMovesProbeOwnersPublishARestartedAttemptFenced(t *testing.T) {
+ f := newPOFixture(t)
+ ctx := t.Context()
+ o := f.read()
+ in, err := f.owners.Select(ctx, f.route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ first := f.admit(o, in)
+ again, err := f.owners.Select(ctx, f.route, first, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.admit(first, again)
+ synced := 0
+ pub := ClaimPublication{Sync: func(context.Context) error { synced++; return nil }, Check: func(context.Context) error { return nil }}
+ if err := f.owners.Publish(ctx, admitted, again); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("an unfenced marker for a fresh generation: %v", err)
+ }
+ if _, err := f.owners.PublishFenced(ctx, admitted, again, pub); !errors.Is(err, hostbox.ErrChanged) || synced != 1 {
+ t.Fatalf("a marker before the root: %v synced=%d", err, synced)
+ }
+ root := "instances/" + poHost + "/g-" + again.Admission.NewGeneration + "/0009/" + ltx.FormatFilename(1, 44)
+ f.bucket.Put("data", root, f.ltxHeader(1, 44))
+ receipt, err := f.owners.PublishFenced(ctx, admitted, again, pub)
+ if err != nil || receipt.Root.Key != root || receipt.MarkerVersion == "" {
+ t.Fatalf("publish: %+v %v", receipt, err)
+ }
+ if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestMovesProbeOwnersThroughTheClaimRuntime drives the claim's own runtime
+// (claim Plan 9 Group D's ClaimWakeAssembly, the recovery fixture's
+// controller) with this owner set and the moves probe attempts bound, as
+// production binds them: the admission is the prepared cold admission,
+// journalled before any start; the probe start carries the moves attempt
+// journal; and Materialize renders the held inputs before any launch is
+// prepared.
+func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
+ f := newRecoveryFixture(t)
+ ctx := t.Context()
+ journalDir, filesDir := t.TempDir(), t.TempDir()
+ seal := moves.Seal{V: moves.SealVersion, N: 42, FromDB: true, SrcCovered: true, SealedAt: f.now, LeaseUntil: f.now.Add(time.Minute),
+ Replica: []moves.SealedObject{{Key: "moves/" + recoveryHost + "/" + poOp + "/replica/0009/" + ltx.FormatFilename(1, 42), VersionID: "v1", Size: int64(ltx.HeaderSize)}},
+ Files: []moves.SealedObject{{Key: "moves/" + recoveryHost + "/" + poOp + "/files/archive.sealed", VersionID: "v2", Size: 10}}}
+ if _, err := hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: poOp, Phase: hostclaim.PhaseLandingProbing, Exec: "box-1", Participants: []string{"box-0", "box-1"},
+ Intent: map[string]string{"to": "ie", "source": "box-0", "target": "box-1"}, Journal: map[string]string{hostclaim.JournalSeal: seal.Encode()}}
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ o := f.read()
+ h := o.Claim.Holder
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ key := strings.TrimSuffix(h.Location, "/") + "/g-" + poOp + "/0009/" + ltx.FormatFilename(1, 42)
+ hdr, _ := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: 1, MaxTXID: 42}).MarshalBinary()
+ manifest := moves.InstallManifest{V: 1, Op: poOp, Host: recoveryHost, N: 42, SnapshotKey: key, SnapshotVersion: bucket.Put("data", key, hdr)}
+ j, _ := moves.ReadJournal(journalDir, recoveryHost, poOp)
+ body, _ := json.Marshal(manifest)
+ j.Mark(landStepInstall, string(body))
+ if err := j.Write(journalDir); err != nil {
+ t.Fatal(err)
+ }
+ route := registry.Route{Host: recoveryHost, Kind: "instance", App: "jam", Account: "bab", Status: "landing", HoldOp: poOp, Socket: filepath.Join(f.guard.DataDir, "app.sock"), DBPath: filepath.Join(f.guard.DataDir, "state", "data.db")}
+ if err := os.MkdirAll(filepath.Dir(route.DBPath), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := hostbox.WriteGenerationMarker(route.DBPath+".gen", poOp); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(filesDir, archive.MarkerName), mustJSON(t, moves.FilesGen{Op: poOp, Incarnation: h.Incarnation}), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ pins := mpPins()
+ pins.EnvReport = "sha256:env"
+ rec := &provision.Record{Host: recoveryHost, Account: "bab", App: "jam", Region: "ie",
+ Move: &provision.MoveState{ID: poOp, Incarnation: h.Incarnation, From: "uk", To: "ie", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, Landing: &provision.MoveLandingJournal{Attempts: 1}}}
+ records := func(context.Context, string, string, string) (*provision.Record, error) { return rec, nil }
+ var order []string
+ launches := 0
+ owners := &movesProbeOwners{Claims: f.claims, Records: records, Data: bucket, Bucket: "data", Region: "ie", Box: "box-1", DataDir: journalDir, Markers: plan9D5Markers{bucket},
+ ReadyPath: func(context.Context, string, string) (string, error) { return "/ready", nil },
+ FilesDir: func(string) (string, string) { return filesDir, "" },
+ Route: func(string) (*registry.Route, error) { rt := route; return &rt, nil },
+ Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) {
+ return ClaimServingLaunch{Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), Env: []string{"A=1"}, Version: pins.Release}, nil
+ },
+ EnvReport: func(context.Context, registry.Route) (string, error) { return pins.EnvReport, nil },
+ Works: func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
+ order = append(order, "select:"+action)
+ return plan9D5Work{}, nil
+ },
+ Render: func(ctx context.Context, _ registry.Route, _ string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) {
+ order = append(order, "render")
+ if _, err := work(appconfig.HeldRender{BundleETags: pins.BundleETags, OverlayETag: pins.OverlayETag}); err != nil {
+ return appconfig.HeldRender{}, err
+ }
+ return appconfig.HeldRender{BundleETags: pins.BundleETags, OverlayETag: pins.OverlayETag}, nil
+ },
+ Release: func(context.Context, registry.Route, string, hostwrite.Work) error {
+ order = append(order, "release")
+ return nil
+ },
+ NewID: f.newID, Now: func() time.Time { return f.now }}
+ attempts := &movesProbeAttempts{Claims: f.claims, Records: records, DataDir: journalDir, NewID: f.newID, Now: func() time.Time { return f.now },
+ Admitted: func(context.Context, *provision.Record) (int, error) { return 3, nil }}
+ c := f.controller("boot-1", "controller-1", 4101, "start-1")
+ bound := owners.MovesOwners(attempts)
+ c.wake.Inputs = ClaimOwnerDispatch{Moves: &bound.ClaimKindOwners}
+ c.wake.Continue = hostclaim.SessionContinues
+ c.wake.Operations = bound.Operations
+ c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
+ c.wake.Launch = func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error) {
+ launches++
+ return nil, hostwrite.ErrUnavailable
+ }
+ live, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ done := make(chan struct{})
+ go func() { <-live.Done(); close(done) }()
+ c.wake.mu.Lock()
+ c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
+ c.wake.mu.Unlock()
+
+ admitted, err := c.runtime.Admit(ctx, route, o, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ah := admitted.Claim.Holder
+ last := ah.Admissions[len(ah.Admissions)-1]
+ if ah.Generation != poOp || ah.Stamp != h.Stamp || ah.WriterBox != "box-1" || last.Mode != "cold" || last.Coverage != "sealed" || last.Generation != poOp || last.Region != "ie" {
+ t.Fatalf("the admission: %+v %+v", ah, last)
+ }
+ s, err := c.runtime.Prepare(ctx, route, admitted, poProbe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if s.OpenProbe == nil || s.Identity != mpLaunchIdentity(pins) {
+ t.Fatalf("the probe start: %+v", s.Identity)
+ }
+ if opened, err := s.OpenProbe(ctx, f.now); err != nil || !opened.Equal(f.now) {
+ t.Fatalf("open: %v %v", opened, err)
+ }
+ if err := s.Reserve(ctx, 0); err != nil {
+ t.Fatal(err)
+ }
+ if a := s.CurrentAttempt(); a.ID == "" || a.Session != ah.Session || a.ControllerID != "controller-1" {
+ t.Fatalf("the reservation: %+v", a)
+ }
+ if err := s.Materialize(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if want := []string{"render", "select:held-config", "select:held-release", "release"}; !reflect.DeepEqual(order, want) || launches != 0 {
+ t.Fatalf("before the launch: %v launches=%d", order, launches)
+ }
+ if _, err := s.PrepareLaunch(ctx); err == nil || launches != 1 {
+ t.Fatalf("the launch after Materialize: %v %d", err, launches)
+ }
+}