amadan

carlos / platform Public

host: the landing wake after the commit (Plan 3 Task 7, Task 5's committed wake)

At landing/committed, when no process serves the host (the probe's own
entry is never restarted), the moves owners admit a cold continuation
into a fresh generation from the certified remote: no kept stamp, no
coverage, the row's ordinary launch without the probe variable, the
start check before every start, fenced publication as serving's. The
attempts journal its starts per admitted session with no record attempt
or settle. MovesClaimPolicyWithInputs grants the committed spawn only
through an injected input check over a committed record;
MovesClaimPolicy (production) refuses it until the claim binds one.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4e1f93d5bb6e731793bb89a72218efc3b302d963 parent 580a677
6 files changed, +426 −23
  • internal/host/movespolicy.go +41 −0
  • internal/host/movesprobe_policy_test.go +43 −0
  • internal/host/movesprobeattempts.go +65 −13
  • internal/host/movesprobeattempts_test.go +35 −0
  • internal/host/movesprobeowners.go +145 −10
  • internal/host/movesprobeowners_committed_test.go +97 −0
diff --git a/internal/host/movespolicy.go b/internal/host/movespolicy.go
index 00060ebad..b5eaa465b 100644
--- a/internal/host/movespolicy.go
+++ b/internal/host/movespolicy.go
@@ -233,7 +233,19 @@ var movesPolicyActions = map[string]bool{
// MovesClaimPolicy builds hostbox.Policy.Moves for the moves plans over a
// record reader: see the file comment for what it grants.
+// MovesClaimPolicy is MovesClaimPolicyWithInputs with no committed input
+// check: the committed wake's spawn is refused (production, until the claim
+// binds one; H4).
func MovesClaimPolicy(read MovesRecordReader) func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error {
+ return MovesClaimPolicyWithInputs(read, nil)
+}
+
+// MovesClaimPolicyWithInputs builds hostbox.Policy.Moves. committed is the
+// committed wake's input check (Plan 3 Task 7, H4): a fresh spawn selection
+// at landing/committed carries the ordinary launch identity, and committed
+// says whether it is the launch this host's row would run, as the serving
+// InputPolicy does for serving. Nil refuses every committed spawn.
+func MovesClaimPolicyWithInputs(read MovesRecordReader, committed func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error) func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error {
return func(ctx context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
refuse := func(reason string, cause error) error {
r := &MovesPolicyRefusal{Host: s.Host, Op: s.Op, Action: s.Action, Box: s.Box, Reason: reason, Cause: cause}
@@ -292,6 +304,9 @@ func MovesClaimPolicy(read MovesRecordReader) func(context.Context, *hostclaim.C
return refuse("this box is declared lost", nil)
}
}
+ if s.Phase == hostclaim.PhaseLandingCommitted && s.Action == movesActionSpawn {
+ return movesCommittedSpawn(ctx, read, committed, c, s, id, refuse)
+ }
if !movesPhaseRights[s.Phase][s.Action] {
return refuse("the phase does not permit the action", nil)
}
@@ -548,3 +563,29 @@ func heldTimersReplacer(writes hostwrite.Config, reg *registry.Registry) func(ct
return schedule.SelectedTimers{Store: reg, Work: work, Context: ctx, Host: host}.ReplaceTimers(host, rows)
}
}
+
+// movesCommittedSpawn is the committed wake's spawn (Plan 3 Task 7): the
+// ordinary launch identity, over a committed record of this move, accepted by
+// the committed input check.
+func movesCommittedSpawn(ctx context.Context, read MovesRecordReader, committed func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity, refuse func(string, error) error) error {
+ if committed == nil {
+ return refuse("the committed launch has no input check", nil)
+ }
+ if !recoveryIdentityFits(s.Action, id) {
+ return refuse("the identity does not fit the action", nil)
+ }
+ if read == nil {
+ return refuse("no record reader", nil)
+ }
+ rec, err := read(ctx, c.Holder.Account, c.Holder.App, c.Host)
+ if err != nil {
+ return refuse("the record cannot be read", err)
+ }
+ if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != s.Op || rec.Move.Incarnation != c.Holder.Incarnation || rec.Move.State != provision.MoveCommitted {
+ return refuse("the record is not this committed move", nil)
+ }
+ if err := committed(ctx, c, s, id); err != nil {
+ return refuse("the launch is not the committed one", err)
+ }
+ return nil
+}
diff --git a/internal/host/movesprobe_policy_test.go b/internal/host/movesprobe_policy_test.go
index c77a5e23d..173bfbb7a 100644
--- a/internal/host/movesprobe_policy_test.go
+++ b/internal/host/movesprobe_policy_test.go
@@ -109,3 +109,46 @@ func TestMovesClaimPolicyProbeSessionContinues(t *testing.T) {
t.Fatal("a backward step was granted")
}
}
+
+// Task 7, the committed wake: at landing/committed a fresh selection's spawn
+// is granted only with the ordinary launch identity the committed input
+// check accepts, against a committed record of this move. Without the check
+// (production until the claim binds one) it is refused.
+func TestMovesClaimPolicyCommittedSpawn(t *testing.T) {
+ ctx := t.Context()
+ committed := mpRecord()
+ committed.Move.State, committed.Move.LandingPhase = provision.MoveCommitted, ""
+ reader := &mpReader{rec: committed}
+ checked := 0
+ ordinary := hostwrite.Identity{Config: "sha256:env", Release: "sha256:bytes"}
+ inputs := func(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
+ checked++
+ if id != ordinary {
+ return errors.New("not the launch")
+ }
+ return nil
+ }
+ policy := hostbox.Policy{Moves: MovesClaimPolicyWithInputs(reader.read, inputs), Held: func(context.Context, string) (string, error) { return "", nil }}
+ c := mpClaim(hostclaim.KindLanding, hostclaim.PhaseLandingCommitted)
+ s := mpSelection(c, "spawn")
+ if err := policy.Check(ctx, c, s, ordinary); err != nil || checked != 1 {
+ t.Fatalf("the committed wake's spawn: %v checked=%d", err, checked)
+ }
+ if err := policy.Check(ctx, c, s, hostwrite.Identity{Config: "x", Release: "y"}); err == nil {
+ t.Fatal("another launch was granted")
+ }
+ for _, action := range []string{"held-config", "held-release", "restore", "install"} {
+ if err := policy.Check(ctx, c, mpSelection(c, action), ordinary); err == nil {
+ t.Fatalf("committed/%s was granted", action)
+ }
+ }
+ reader.rec = mpProbingRecord()
+ if err := policy.Check(ctx, c, s, ordinary); err == nil {
+ t.Fatal("a committed spawn over a record that is not committed was granted")
+ }
+ reader.rec = committed
+ unchecked := hostbox.Policy{Moves: MovesClaimPolicy(reader.read), Held: func(context.Context, string) (string, error) { return "", nil }}
+ if err := unchecked.Check(ctx, c, s, ordinary); err == nil {
+ t.Fatal("a committed spawn without an input check was granted")
+ }
+}
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go
index b00fbb8fe..29239567b 100644
--- a/internal/host/movesprobeattempts.go
+++ b/internal/host/movesprobeattempts.go
@@ -50,8 +50,10 @@ import (
// Every method refuses (hostbox.ErrChanged) once the claim no longer names
// the admitted writer (op id, kind, session, stamp, generation; phase
// probing) or the record no longer names this landing at probing, unmarked,
-// at attempt k. Only probes are served: a landing wake after the commit is
-// Task 7's and is refused (hostbox.ErrUnsupported).
+// at attempt k. A landing wake after the commit (Plan 3 Task 7) is served
+// too: its starts are journalled per admitted session (probe_start_committed-
+// <session>), bound to the claim at landing/committed and a committed record,
+// with no record attempt, no window and no settle.
type movesProbeAttempts struct {
Claims release.Store
Records MovesRecordReader
@@ -103,15 +105,19 @@ func (p *movesProbeAttempts) valid() bool {
}
// bound is every method's check: it returns the record's attempt index k for
-// the admitted probe s, or a refusal.
+// the admitted probe s, or 0 for a committed wake (Plan 3 Task 7), or a
+// refusal.
func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, error) {
if !p.valid() {
return 0, hostwrite.ErrUnavailable
}
a := s.Admitted
- if !s.Policy.Probe || a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding {
+ if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding {
return 0, hostbox.ErrUnsupported
}
+ if !s.Policy.Probe {
+ return 0, p.boundCommitted(ctx, s)
+ }
h := a.Claim.Holder
if s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation {
return 0, hostbox.ErrChanged
@@ -145,6 +151,41 @@ func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int,
return k, nil
}
+// boundCommitted is bound for a committed wake: the claim still names the
+// admitted writer at landing/committed, and the record is this move,
+// committed. Its starts are journalled per session (startKey).
+func (p *movesProbeAttempts) boundCommitted(ctx context.Context, s OperationStart) error {
+ a := s.Admitted
+ h := a.Claim.Holder
+ if a.Claim.Op.Phase != hostclaim.PhaseLandingCommitted || s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation {
+ return hostbox.ErrChanged
+ }
+ fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
+ if err != nil {
+ return err
+ }
+ if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseLandingCommitted {
+ return hostbox.ErrChanged
+ }
+ rec, err := p.Records(ctx, h.Account, h.App, a.Host)
+ if err != nil {
+ return err
+ }
+ if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation || rec.Move.State != provision.MoveCommitted {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+// startKey names a start's journal step: the record attempt k for a probe,
+// the admitted session for a committed wake (k 0).
+func startKey(s OperationStart, k int) string {
+ if k == 0 {
+ return "committed-" + strconv.FormatUint(s.Owner.Session, 10)
+ }
+ return strconv.Itoa(k)
+}
+
func (p *movesProbeAttempts) journal(s OperationStart) (*moves.Journal, error) {
return moves.ReadJournal(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
}
@@ -157,6 +198,9 @@ func probeKey(s OperationStart) string {
// Absent, it is written into j when write is set (the caller writes j) and
// refused otherwise.
func bindSession(j *moves.Journal, s OperationStart, k int, write bool) error {
+ if k == 0 {
+ return nil // a committed wake has no record attempt
+ }
step := probeStepSession + strconv.FormatUint(s.Owner.Session, 10)
raw, ok := j.Steps[step]
switch {
@@ -177,6 +221,9 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr
if err != nil {
return time.Time{}, err
}
+ if k == 0 {
+ return time.Time{}, hostbox.ErrUnsupported
+ }
if proposed.IsZero() {
return time.Time{}, hostwrite.ErrUnavailable
}
@@ -209,24 +256,24 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr
}
// readStart returns attempt k's journalled start, or false.
-func readStart(j *moves.Journal, k int) (probeStart, bool, error) {
- raw, ok := j.Steps[probeStepStart+strconv.Itoa(k)]
+func readStart(j *moves.Journal, key string) (probeStart, bool, error) {
+ raw, ok := j.Steps[probeStepStart+key]
if !ok {
return probeStart{}, false, nil
}
var st probeStart
if err := json.Unmarshal([]byte(raw), &st); err != nil {
- return probeStart{}, false, fmt.Errorf("moves journal %s/%s: %s%d: %w", j.Host, j.Op, probeStepStart, k, err)
+ return probeStart{}, false, fmt.Errorf("moves journal %s/%s: %s%s: %w", j.Host, j.Op, probeStepStart, key, err)
}
return st, true, nil
}
-func writeStart(dataDir string, j *moves.Journal, k int, st probeStart) error {
+func writeStart(dataDir string, j *moves.Journal, key string, st probeStart) error {
raw, err := json.Marshal(st)
if err != nil {
return err
}
- j.Mark(probeStepStart+strconv.Itoa(k), string(raw))
+ j.Mark(probeStepStart+key, string(raw))
return j.Write(dataDir)
}
@@ -258,7 +305,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
if err := bindSession(j, s, k, n == 0); err != nil {
return err
}
- st, ok, err := readStart(j, k)
+ st, ok, err := readStart(j, startKey(s, k))
if err != nil {
return err
}
@@ -293,7 +340,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
return hostwrite.ErrUnavailable
}
- if err := writeStart(p.DataDir, j, k, st); err != nil {
+ if err := writeStart(p.DataDir, j, startKey(s, k), st); err != nil {
return err
}
}
@@ -348,7 +395,7 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
if err := bindSession(j, s, k, false); err != nil {
return err
}
- st, ok, err := readStart(j, k)
+ st, ok, err := readStart(j, startKey(s, k))
if err != nil {
return err
}
@@ -357,7 +404,7 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
return hostbox.ErrChanged
}
st.State = state
- return writeStart(p.DataDir, j, k, st)
+ return writeStart(p.DataDir, j, startKey(s, k), st)
}
// Healthy accepts start n after readiness, replication and publication:
@@ -378,6 +425,11 @@ func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n in
if err != nil {
return err
}
+ if k == 0 {
+ // A committed wake: serving's readiness has passed, with
+ // replication and publication; there is no probe settle.
+ return p.settle(ctx, s, n, probeHealthy)
+ }
if again == nil || !probeAlive(cmd) {
return hostbox.ErrChanged
}
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go
index 2ccffa49e..885421b8f 100644
--- a/internal/host/movesprobeattempts_test.go
+++ b/internal/host/movesprobeattempts_test.go
@@ -337,3 +337,38 @@ func TestMovesProbeAttemptsHealthySettlesAndAsksAgain(t *testing.T) {
t.Fatalf("journalled healthy: %v", j.Steps)
}
}
+
+// Task 7: the committed wake's starts are journalled per admitted session,
+// with no record attempt (the probe's are over); Healthy accepts at once.
+func TestMovesProbeAttemptsCommittedWake(t *testing.T) {
+ f := newPAFixture(t)
+ ctx := t.Context()
+ c := paClaim(t)
+ c.Op.Phase = hostclaim.PhaseLandingCommitted
+ f.putClaim(c)
+ f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
+ s := f.start
+ s.Admitted, s.Policy.Probe = f.read(), false
+ s.Owner.ClaimETag = s.Admitted.ETag
+ if err := f.ops.Reserve(ctx, s, 0); err != nil {
+ t.Fatal(err)
+ }
+ first := f.ops.CurrentAttempt(s)
+ if first.ID == "" || first.Session != s.Owner.Session {
+ t.Fatalf("reservation: %+v", first)
+ }
+ if err := f.ops.Failed(ctx, s, 0); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.ops.Reserve(ctx, s, 1); err != nil || f.ops.CurrentAttempt(s).ID == first.ID {
+ t.Fatalf("start 1: %v", err)
+ }
+ asked := 0
+ if err := f.ops.Healthy(ctx, s, 1, nil, func(context.Context) error { asked++; return nil }); err != nil || asked != 0 {
+ t.Fatalf("healthy: %v asked=%d", err, asked)
+ }
+ f.rec.Move.State = provision.MoveLanding
+ if err := f.ops.Reserve(ctx, s, 0); err == nil {
+ t.Fatal("a committed wake's start over a landing record")
+ }
+}
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index 3e03f9b76..a51391589 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -173,11 +173,14 @@ func (m *movesProbeOwners) valid() bool {
// 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
+ // committed is a landing wake after the commit (Plan 3 Task 7): the
+ // claim at landing/committed, the record committed.
+ committed bool
+ 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
@@ -191,7 +194,14 @@ func (m *movesProbeOwners) landing(ctx context.Context, o hostclaim.Observed) (p
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 ||
+ switch c.Op.Phase {
+ case hostclaim.PhaseLandingProbing:
+ case hostclaim.PhaseLandingCommitted:
+ v.committed = true
+ default:
+ return v, hostbox.ErrChanged
+ }
+ if c.Host != o.Host || c.Op.ID == "" || 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
}
@@ -209,7 +219,14 @@ func (m *movesProbeOwners) landing(ctx context.Context, o hostclaim.Observed) (p
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 == "" {
+ if mv.ID != v.op || mv.Incarnation != v.holder.Incarnation || mv.To != m.Region || mv.Inputs == nil || mv.Inputs.Release == "" {
+ return v, hostbox.ErrChanged
+ }
+ if v.committed {
+ if mv.State != provision.MoveCommitted {
+ return v, hostbox.ErrChanged
+ }
+ } else if mv.State != provision.MoveLanding || mv.LandingPhase != provision.LandProbing || mv.Reason != "" {
return v, hostbox.ErrChanged
}
v.rec = rec
@@ -473,13 +490,19 @@ func probeEnv(env []string) []string {
// 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
}
+ if v.committed {
+ if p.Probe || p.Prepared != nil || p.Inputs != nil {
+ return ClaimLaunchInputs{}, hostbox.ErrUnsupported
+ }
+ return m.selectCommitted(ctx, rt, o, v)
+ }
+ if !p.Probe {
+ return ClaimLaunchInputs{}, hostbox.ErrUnsupported
+ }
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
@@ -548,6 +571,9 @@ func (m *movesProbeOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Ob
if err != nil {
return err
}
+ if v.committed {
+ return m.committedEvidence(ctx, o, v, in, baseline)
+ }
if err := m.admissible(ctx, v); err != nil {
return err
}
@@ -649,6 +675,9 @@ func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed
if work == nil {
return hostwrite.ErrUnavailable
}
+ if v.committed {
+ return m.materializeCommitted(ctx, o, v, in, work)
+ }
pins := v.rec.Move.Inputs
if in.Identity != launchIdentity(pins) {
return hostbox.ErrChanged
@@ -1017,3 +1046,109 @@ func (m *movesProbeOwners) PublicationEvidence(ctx context.Context, o hostclaim.
}
return nil
}
+
+// The committed wake (Plan 3 Task 7; Task 5's concern). After the commit the
+// claim stays landing/committed until the source's finalization closes the
+// move, and Group D admits a landing wake there (K6's committed row). A wake
+// happens only when no process serves the host: the probe's own entry, while
+// it lives, is the serving process and is never restarted. The wake is a
+// cold admission into a fresh generation from the certified remote (the
+// landing's published marker, genselect.Select), continuing the admitted
+// session's generation: no kept stamp, no coverage. It launches the row's
+// ordinary launch (the claim's launch snapshot, its own identity, which
+// MovesClaimPolicyWithInputs's committed check accepts), never the probe
+// variable, and the held inputs are not rendered again: after the open the
+// row is config sync's.
+
+// committedRoute is the row a committed wake runs: this host's, held by the
+// move until the open, then ready.
+func committedRoute(rt registry.Route, host, op string) error {
+ if rt.Host != host || !filepath.IsAbs(rt.DBPath) {
+ return hostbox.ErrChanged
+ }
+ if (registry.IsHold(rt.Status) && rt.HoldOp == op) || (rt.Status == "ready" && rt.HoldOp == "") {
+ return nil
+ }
+ return hostbox.ErrChanged
+}
+
+// withoutProbeVars is a launch environment with the probe-only variables
+// removed.
+func withoutProbeVars(env []string) []string {
+ out := make([]string, 0, len(env))
+ for _, kv := range env {
+ if k, _, _ := strings.Cut(kv, "="); !probeOnlyVars[k] {
+ out = append(out, kv)
+ }
+ }
+ return out
+}
+
+func (m *movesProbeOwners) committedAdmission(o hostclaim.Observed, v probeLanding, gen string, plan *genselect.Plan) hostclaim.Admission {
+ h := v.holder
+ return hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
+ NewGeneration: gen, KeepStamp: h.Stamp > h.Session, Region: m.Region, PredecessorRegion: predecessorRegion(h, v.rec.Move.From), RemoteMaxTXID: plan.MaxTXID}
+}
+
+func (m *movesProbeOwners) selectCommitted(ctx context.Context, rt registry.Route, o hostclaim.Observed, v probeLanding) (ClaimLaunchInputs, error) {
+ if err := committedRoute(rt, o.Host, v.op); err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ h := v.holder
+ if h.WriterBox != m.Box {
+ return ClaimLaunchInputs{}, hostbox.ErrChanged
+ }
+ gen := m.NewID()
+ if gen == v.op || !claimFreshGeneration(h, gen) {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ plan, err := genselect.Select(ctx, m.Data, m.request(o.Host, h))
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ l, err := m.Launch(ctx, rt)
+ if err != nil {
+ return ClaimLaunchInputs{}, err
+ }
+ if l.Identity.Config == "" || l.Identity.Release == "" || !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
+ return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
+ }
+ return ClaimLaunchInputs{Roles: []string{"main"}, Identity: l.Identity, Baseline: plan, Admission: m.committedAdmission(o, v, gen, plan),
+ Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: withoutProbeVars(l.Env),
+ DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit}, nil
+}
+
+func (m *movesProbeOwners) committedEvidence(ctx context.Context, o hostclaim.Observed, v probeLanding, in hostclaim.Admission, baseline *genselect.Plan) error {
+ h := v.holder
+ if baseline == nil || h.WriterBox != m.Box || in.NewGeneration == v.op || !claimFreshGeneration(h, in.NewGeneration) || in.PredecessorCertified || in.LossID != "" || len(in.AdditionalWritesLost) != 0 {
+ return hostbox.ErrChanged
+ }
+ if !reflect.DeepEqual(in, m.committedAdmission(o, v, in.NewGeneration, baseline)) {
+ return hostbox.ErrChanged
+ }
+ fresh, err := genselect.Select(ctx, m.Data, m.request(o.Host, h))
+ if err != nil {
+ return err
+ }
+ if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) || !reflect.DeepEqual(fresh.Pin, baseline.Pin) {
+ return hostbox.ErrChanged
+ }
+ return nil
+}
+
+func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim.Observed, v probeLanding, in ClaimLaunchInputs, work hostwrite.Work) error {
+ if err := writerFence(work, in.Identity)(ctx); err != nil {
+ return err
+ }
+ rt, err := m.Route(o.Host)
+ if err != nil {
+ return err
+ }
+ if rt == nil || rt.DBPath != in.DBPath {
+ return hostbox.ErrChanged
+ }
+ if err := committedRoute(*rt, o.Host, v.op); err != nil {
+ return err
+ }
+ return m.startCheck(ctx, o, in)
+}
diff --git a/internal/host/movesprobeowners_committed_test.go b/internal/host/movesprobeowners_committed_test.go
new file mode 100644
index 000000000..d9d1fbb2d
--- /dev/null
+++ b/internal/host/movesprobeowners_committed_test.go
@@ -0,0 +1,97 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/superfly/ltx"
+)
+
+var poWake = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second}
+
+// committedAfterProbe is the owners' fixture after a passing probe and the
+// commit: the move id admitted and published, the claim at
+// landing/committed, the record committed, the row open.
+func committedAfterProbe(t *testing.T) (*poFixture, hostclaim.Observed) {
+ t.Helper()
+ f := newPOFixture(t)
+ admitted, in := f.selectAndAdmit()
+ if err := f.owners.Materialize(t.Context(), admitted, in, newFenceWork()); err != nil {
+ t.Fatal(err)
+ }
+ c := *f.read().Claim
+ c.Op.Phase = hostclaim.PhaseLandingCommitted
+ f.putClaim(&c)
+ f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
+ f.route.Status, f.route.HoldOp = "ready", ""
+ return f, f.read()
+}
+
+// Task 7 (Task 5's concern): the landing wake after the commit is a cold
+// admission into a fresh generation from the certified remote, continuing
+// the admitted session's generation; it carries the ordinary launch, no
+// probe variable, and is evidenced, materialized, start-checked and
+// published as serving's are.
+func TestMovesProbeOwnersCommittedWake(t *testing.T) {
+ f, o := committedAfterProbe(t)
+ ctx := t.Context()
+ h := o.Claim.Holder
+ in, err := f.owners.Select(ctx, f.route, o, poWake)
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := in.Admission
+ if a.ObservedOpID != poOp || a.ObservedPhase != hostclaim.PhaseLandingCommitted || a.ObservedSession != h.Session || a.ObservedGeneration != poOp ||
+ a.Mode != "cold" || a.NewGeneration == "" || a.NewGeneration == poOp || a.KeepStamp || a.Coverage != "" || a.Region != "uk" || a.PredecessorRegion != "uk" {
+ t.Fatalf("the committed admission: %+v", a)
+ }
+ if in.Prepared || in.Baseline == nil || in.Baseline.Generation != poOp || in.Baseline.MaxTXID != 42 || a.RemoteMaxTXID != 42 {
+ t.Fatalf("the committed baseline: %+v", in.Baseline)
+ }
+ if in.Identity != f.launch.Identity {
+ t.Fatalf("the committed wake's identity %+v, want the launch's %+v", in.Identity, f.launch.Identity)
+ }
+ for _, kv := range in.Env {
+ if strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
+ t.Fatalf("the committed wake carries %s", kv)
+ }
+ }
+ if err := f.owners.AdmissionEvidence(ctx, o, a, in.Baseline); err != nil {
+ t.Fatal(err)
+ }
+ stale := *in.Baseline
+ stale.MaxTXID = 41
+ if err := f.owners.AdmissionEvidence(ctx, o, a, &stale); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("a stale baseline: %v", err)
+ }
+ admitted := f.admit(o, in)
+ f.writeLocal(poOp, admitted.Claim.Holder.Generation, f.inc)
+ if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
+ t.Fatalf("materialize: %v", err)
+ }
+ root := "instances/" + poHost + "/g-" + a.NewGeneration + "/0009/" + ltx.FormatFilename(1, 43)
+ f.bucket.Put("data", root, f.ltxHeader(1, 43))
+ pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}
+ if _, err := f.owners.PublishFenced(ctx, admitted, in, pub); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
+ t.Fatal(err)
+ }
+ // A probe is never admitted at committed, and a committed wake never at
+ // probing; a record that is not committed refuses.
+ if _, err := f.owners.Select(ctx, f.route, o, poProbe); err == nil {
+ t.Fatal("a probe at committed")
+ }
+ f.rec.Move.State = provision.MoveLanding
+ if _, err := f.owners.Select(ctx, f.route, o, poWake); err == nil {
+ t.Fatal("a committed wake over a landing record")
+ }
+}