amadan

carlos / platform Public

host: the moves probe attempts, bound to the record's attempt (Plan 3 Task 5b, H2, H10)

OperationAttempts for a landing probe: OpenProbe journals the first start
once per record attempt and survives a controller restart; Reserve,
CurrentAttempt and Failed journal the starts of one admission, bound to
Landing.Attempts within moves.AttemptsFor; Healthy is the 20 s settle
with the process alive and a second readiness request. Box-local journal
only: no record or claim write.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a2e6a2e1db9d9ceb7a382e0f4f1bda1167f83fc4 parent 210b7d9
2 files changed, +628 −0
  • internal/host/movesprobeattempts.go +345 −0
  • internal/host/movesprobeattempts_test.go +283 −0
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go
new file mode 100644
index 000000000..e9dae6fed
--- /dev/null
+++ b/internal/host/movesprobeattempts.go
@@ -0,0 +1,345 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "os/exec"
+ "strconv"
+ "sync"
+ "syscall"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// movesProbeAttempts is the moves OperationAttempts implementation for a
+// landing probe (moves Phase 1 Plan 3 Task 5b; claim Plan 9 Group D, D2;
+// rulings H2, H10). The claim binds its methods into the admitted probe's
+// activator.ClaimStart; this type journals, counts and settles the probe's
+// starts in the box-local op journal (moves.Journal, <DataDir>/.moves/<host>/
+// <op>/), and writes no record and no claim.
+//
+// One record attempt is one admission. The landing (Task 6) counts it on the
+// record (provision.MoveProbeAttempt, Landing.Attempts = k) before it asks the
+// activator for the probe; every method here is bound to that k, refused
+// unless 1 <= k <= moves.AttemptsFor("landing", admitted), where admitted is
+// what T11 admitted for this move (Admitted, H10). Within an attempt the
+// activator makes up to five starts (startAdmitted).
+//
+// Steps it journals, per record attempt k:
+//
+// - probe_opened_<k>: the first app start's time, RFC 3339. OpenProbe
+// writes it once and returns it to every later start, a restarted
+// controller included, so the 30 s window is the attempt's and never
+// restarts.
+// - probe_start_<k>: the current start's reservation (probeStart): the
+// admitted session, the start's number n and the role launcher's
+// WakeRetryAttempt, and its state, reserved, failed or healthy.
+//
+// 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).
+type movesProbeAttempts struct {
+ Claims release.Store
+ Records MovesRecordReader
+ DataDir string
+ // Admitted is the number of probe attempts T11 admitted for the move
+ // (H10: the target's report, or the record once MoveAdmit carries it).
+ Admitted func(context.Context, *provision.Record) (int, error)
+ NewID func() string
+ Now func() time.Time
+ // Settle is the wait before the second readiness request; zero is
+ // moves.SettleSeconds. Ready bounds that request; zero is
+ // moves.ProbeSeconds.
+ Settle, Ready time.Duration
+
+ mu sync.Mutex
+ current map[string]hostclaim.WakeRetryAttempt
+}
+
+var _ OperationAttempts = (*movesProbeAttempts)(nil)
+
+// Journal step names and start states.
+const (
+ probeStepOpened = "probe_opened_"
+ probeStepStart = "probe_start_"
+
+ probeReserved = "reserved"
+ probeFailed = "failed"
+ probeHealthy = "healthy"
+)
+
+// probeStart is a probe_start_<k> step's value.
+type probeStart struct {
+ Session uint64 `json:"session"`
+ N int `json:"n"`
+ Attempt hostclaim.WakeRetryAttempt `json:"attempt"`
+ State string `json:"state"`
+}
+
+func (p *movesProbeAttempts) valid() bool {
+ return p != nil && p.Claims != nil && p.Records != nil && p.DataDir != "" && p.Admitted != nil && p.NewID != nil && p.Now != nil
+}
+
+// bound is every method's check: it returns the record's attempt index k for
+// the admitted probe s, 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 {
+ return 0, hostbox.ErrUnsupported
+ }
+ 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
+ }
+ fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
+ if err != nil {
+ return 0, err
+ }
+ if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseLandingProbing {
+ return 0, hostbox.ErrChanged
+ }
+ rec, err := p.Records(ctx, h.Account, h.App, a.Host)
+ if err != nil {
+ return 0, err
+ }
+ if rec == nil || rec.Host != a.Host || rec.Move == nil {
+ return 0, hostbox.ErrChanged
+ }
+ m := rec.Move
+ if m.ID != a.Claim.Op.ID || m.Incarnation != h.Incarnation || m.State != provision.MoveLanding || m.LandingPhase != provision.LandProbing || m.Reason != "" || m.Landing == nil {
+ return 0, hostbox.ErrChanged
+ }
+ admitted, err := p.Admitted(ctx, rec)
+ if err != nil {
+ return 0, err
+ }
+ k := m.Landing.Attempts
+ if k < 1 || k > moves.AttemptsFor(moves.ProbeLanding, admitted) {
+ return 0, hostbox.ErrChanged
+ }
+ return k, nil
+}
+
+func (p *movesProbeAttempts) journal(s OperationStart) (*moves.Journal, error) {
+ return moves.ReadJournal(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
+}
+
+func probeKey(s OperationStart) string {
+ return s.Admitted.Host + "\x00" + strconv.FormatUint(s.Owner.Session, 10)
+}
+
+// OpenProbe journals the attempt's first start time, once.
+func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return time.Time{}, err
+ }
+ if proposed.IsZero() {
+ return time.Time{}, hostwrite.ErrUnavailable
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := p.journal(s)
+ if err != nil {
+ return time.Time{}, err
+ }
+ step := probeStepOpened + strconv.Itoa(k)
+ if raw, ok := j.Steps[step]; ok {
+ opened, err := time.Parse(time.RFC3339Nano, raw)
+ if err != nil {
+ return time.Time{}, fmt.Errorf("moves journal %s/%s: %s: %w", s.Admitted.Host, s.Admitted.Claim.Op.ID, step, err)
+ }
+ return opened, nil
+ }
+ opened := proposed.UTC()
+ j.Mark(step, opened.Format(time.RFC3339Nano))
+ if err := j.Write(p.DataDir); err != nil {
+ return time.Time{}, err
+ }
+ return opened, nil
+}
+
+// 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)]
+ 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 st, true, nil
+}
+
+func writeStart(dataDir string, j *moves.Journal, k int, st probeStart) error {
+ raw, err := json.Marshal(st)
+ if err != nil {
+ return err
+ }
+ j.Mark(probeStepStart+strconv.Itoa(k), string(raw))
+ return j.Write(dataDir)
+}
+
+// sameStartOwner reports whether a journalled reservation is the one this
+// admission's owner made for these launch inputs.
+func sameStartOwner(a hostclaim.WakeRetryAttempt, s OperationStart, digest string) bool {
+ o := s.Owner
+ return a.ID != "" && a.InvocationID != "" && a.Session == o.Session && a.Box == o.Box && a.Enrollment == o.Enrollment && a.BootID == o.BootID && a.ControllerID == o.ControllerID && a.LaunchInputs == digest
+}
+
+// Reserve journals start n of this admission before it runs. Start 0 is
+// reserved by the first Reserve(0) of the admission and validated by every
+// later one; start n > 0 only after start n-1 was settled failed.
+func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) error {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return err
+ }
+ digest, err := ClaimLaunchDigest(s.Inputs)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := p.journal(s)
+ if err != nil {
+ return err
+ }
+ st, ok, err := readStart(j, k)
+ if err != nil {
+ return err
+ }
+ same := ok && st.Session == s.Owner.Session
+ switch {
+ case n < 0:
+ return hostbox.ErrChanged
+ case n == 0 && same:
+ if st.N != 0 || st.State != probeReserved || !sameStartOwner(st.Attempt, s, digest) {
+ return hostbox.ErrChanged
+ }
+ case n == 0:
+ // The admission's first start; an earlier admission's journalled
+ // start, under another session, is over.
+ st = probeStart{Session: s.Owner.Session, N: 0, State: probeReserved}
+ default:
+ if !same || st.N != n-1 || st.State != probeFailed || st.Attempt.LaunchInputs != digest {
+ return hostbox.ErrChanged
+ }
+ st = probeStart{Session: s.Owner.Session, N: n, State: probeReserved}
+ }
+ if st.Attempt.ID == "" {
+ o := s.Owner
+ st.Attempt = hostclaim.WakeRetryAttempt{ID: p.NewID(), Box: o.Box, Enrollment: o.Enrollment, BootID: o.BootID, ControllerID: o.ControllerID, Session: o.Session, InvocationID: p.NewID(), LaunchInputs: digest}
+ 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 {
+ return err
+ }
+ }
+ if p.current == nil {
+ p.current = map[string]hostclaim.WakeRetryAttempt{}
+ }
+ p.current[probeKey(s)] = st.Attempt
+ return nil
+}
+
+// CurrentAttempt is the reservation the last Reserve journalled for this
+// admission; the zero attempt, which the role launcher refuses, before any.
+func (p *movesProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetryAttempt {
+ if p == nil {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.current[probeKey(s)]
+}
+
+// Failed settles start n, already stopped and drained, as failed.
+func (p *movesProbeAttempts) Failed(ctx context.Context, s OperationStart, n int) error {
+ recovery, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+ defer cancel()
+ return p.settle(recovery, s, n, probeFailed)
+}
+
+func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int, state string) error {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ j, err := p.journal(s)
+ if err != nil {
+ return err
+ }
+ st, ok, err := readStart(j, k)
+ if err != nil {
+ return err
+ }
+ if !ok || st.Session != s.Owner.Session || st.N != n || st.State != probeReserved || st.Attempt != p.current[probeKey(s)] {
+ return hostbox.ErrChanged
+ }
+ st.State = state
+ return writeStart(p.DataDir, j, k, st)
+}
+
+// Healthy accepts start n after readiness, replication and publication:
+// the 20 s settle with the process alive, then the probe's own readiness
+// check again (again, the claim's ClaimProbeReadiness against the started
+// role), with the process still alive after it (moves §7.7).
+func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n int, cmd *exec.Cmd, again func(context.Context) error) error {
+ if _, err := p.bound(ctx, s); err != nil {
+ return err
+ }
+ if again == nil || !probeAlive(cmd) {
+ return hostbox.ErrChanged
+ }
+ settle, ready := p.Settle, p.Ready
+ if settle <= 0 {
+ settle = time.Duration(moves.SettleSeconds * float64(time.Second))
+ }
+ if ready <= 0 {
+ ready = time.Duration(moves.ProbeSeconds * float64(time.Second))
+ }
+ timer := time.NewTimer(settle)
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return ctx.Err()
+ case <-timer.C:
+ }
+ if !probeAlive(cmd) {
+ return fmt.Errorf("%w: the app exited during the settle", hostbox.ErrChanged)
+ }
+ asked, cancel := context.WithTimeout(ctx, ready)
+ err := again(asked)
+ cancel()
+ if err != nil {
+ return err
+ }
+ if !probeAlive(cmd) {
+ return fmt.Errorf("%w: the app exited after the second request", hostbox.ErrChanged)
+ }
+ return p.settle(ctx, s, n, probeHealthy)
+}
+
+// probeAlive reports whether cmd's process has not exited. Signal 0 checks
+// existence without delivering anything; once the activator's Wait has
+// reaped the process it reports os.ErrProcessDone.
+func probeAlive(cmd *exec.Cmd) bool {
+ return cmd != nil && cmd.Process != nil && cmd.Process.Signal(syscall.Signal(0)) == nil
+}
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go
new file mode 100644
index 000000000..b1b3def6b
--- /dev/null
+++ b/internal/host/movesprobeattempts_test.go
@@ -0,0 +1,283 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os/exec"
+ "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/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ paHost = "shop.bab.oncarlos.com"
+ paOp = "mv_probeattempts"
+)
+
+// paClaim is the landing claim after the first probe's admission: uk-box
+// executes at probing and is the writer of session 20 in generation paOp,
+// admitted cold from the source's session 15.
+func paClaim(t *testing.T) *hostclaim.Claim {
+ t.Helper()
+ at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+ inc, err := hostclaim.Incarnation("bab", "jam", paHost, at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := &hostclaim.Claim{V: 2, Host: paHost, Gen: 20,
+ Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + paHost + ".json", Account: "bab", App: "jam", CreatedAt: at, Incarnation: inc, Location: "instances/" + paHost, Generation: paOp, Stamp: 18, Session: 20, WriterBox: "uk-box",
+ Admissions: []hostclaim.AdmissionRecord{
+ {Session: 15, Box: "ie-box", Generation: "src", Predecessor: "", Region: "ie", Mode: "cold"},
+ {Session: 20, Box: "uk-box", Generation: paOp, Predecessor: "src", Region: "uk", PredecessorRegion: "ie", Mode: "cold", Coverage: "sealed"}}},
+ Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: paOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}}
+ if err := c.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ return c
+}
+
+type paFixture struct {
+ t *testing.T
+ claims release.Store
+ rec *provision.Record
+ ops *movesProbeAttempts
+ start OperationStart
+ ids int
+ admit int
+ dataDir string
+}
+
+func newPAFixture(t *testing.T) *paFixture {
+ t.Helper()
+ f := &paFixture{t: t, claims: release.NewDirStore(t.TempDir()), dataDir: t.TempDir(), admit: 2}
+ c := paClaim(t)
+ f.putClaim(c)
+ o := f.read()
+ f.rec = &provision.Record{Host: paHost, Account: "bab", App: "jam", Region: "uk",
+ Move: &provision.MoveState{ID: paOp, Incarnation: c.Holder.Incarnation, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: mpPins(), Landing: &provision.MoveLandingJournal{Attempts: 1}}}
+ f.ops = f.attempts()
+ in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Executable: "/opt/carlos/live/" + paHost, ExecutableSHA256: strings.Repeat("a", 64), Args: []string{"--socket", "/s"}, Env: []string{"CARLOS_MOVE_PROBE=1"}, ReadyPath: "/ready"}
+ h := o.Claim.Holder
+ f.start = OperationStart{Admitted: o, Policy: activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}, Inputs: in,
+ Owner: hostguard.FenceTarget{Host: paHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Op: paOp, Phase: hostclaim.PhaseLandingProbing, Box: "uk-box", Region: "uk", Enrollment: "enr-uk", BootID: "boot-1", ControllerID: "ctl-1", ClaimETag: o.ETag}}
+ return f
+}
+
+// attempts is a fresh controller's provider over the same durable state.
+func (f *paFixture) attempts() *movesProbeAttempts {
+ return &movesProbeAttempts{Claims: f.claims, DataDir: f.dataDir, Settle: 20 * time.Millisecond, Ready: time.Second,
+ Records: func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil },
+ Admitted: func(context.Context, *provision.Record) (int, error) { return f.admit, nil },
+ NewID: func() string { f.ids++; return "id-" + strings.Repeat("x", f.ids) },
+ Now: func() time.Time { return time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC) }}
+}
+
+func (f *paFixture) putClaim(c *hostclaim.Claim) {
+ f.t.Helper()
+ body, err := json.Marshal(c)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ key, err := hostclaim.Key(paHost)
+ 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 *paFixture) read() hostclaim.Observed {
+ f.t.Helper()
+ o, err := hostclaim.Read(f.t.Context(), f.claims, paHost)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ return o
+}
+
+// OpenProbe journals the first start's time once per record attempt and
+// returns it to every later start, a restarted controller included; the
+// next record attempt opens its own window.
+func TestMovesProbeAttemptsOpenProbeIsJournalledOnce(t *testing.T) {
+ f := newPAFixture(t)
+ ctx := t.Context()
+ t1 := time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC)
+ got, err := f.ops.OpenProbe(ctx, f.start, t1)
+ if err != nil || !got.Equal(t1) {
+ t.Fatalf("open: %v %v", got, err)
+ }
+ if got, err := f.ops.OpenProbe(ctx, f.start, t1.Add(5*time.Second)); err != nil || !got.Equal(t1) {
+ t.Fatalf("a second start: %v %v", got, err)
+ }
+ if got, err := f.attempts().OpenProbe(ctx, f.start, t1.Add(9*time.Second)); err != nil || !got.Equal(t1) {
+ t.Fatalf("after a controller restart: %v %v", got, err)
+ }
+ f.rec.Move.Landing.Attempts = 2
+ t2 := t1.Add(time.Minute)
+ if got, err := f.ops.OpenProbe(ctx, f.start, t2); err != nil || !got.Equal(t2) {
+ t.Fatalf("the next record attempt: %v %v", got, err)
+ }
+ j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
+ if err != nil || !j.Done("probe_opened_1") || !j.Done("probe_opened_2") {
+ t.Fatalf("journal: %+v %v", j, err)
+ }
+}
+
+// Reserve, CurrentAttempt and Failed are bound to the record's attempt
+// index and the admitted session: the reservation carries the owner's
+// tuple and the launch digest, is stable between reservations, and a next
+// start is reserved only after the last one was settled.
+func TestMovesProbeAttemptsReserveBindsTheRecordsAttempt(t *testing.T) {
+ f := newPAFixture(t)
+ ctx := t.Context()
+ if a := f.ops.CurrentAttempt(f.start); a.ID != "" {
+ t.Fatalf("an attempt before any reservation: %+v", a)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ first := f.ops.CurrentAttempt(f.start)
+ digest, _ := ClaimLaunchDigest(f.start.Inputs)
+ o := f.start.Owner
+ if first.ID == "" || first.InvocationID == "" || first.ID == first.InvocationID || first.Session != o.Session || first.Box != o.Box || first.Enrollment != o.Enrollment || first.BootID != o.BootID || first.ControllerID != o.ControllerID || first.LaunchInputs != digest {
+ t.Fatalf("reservation: %+v", first)
+ }
+ if again := f.ops.CurrentAttempt(f.start); again != first {
+ t.Fatalf("CurrentAttempt moved: %+v", again)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != first {
+ t.Fatalf("Reserve(0) again is a validation: %v", err)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("start 1 before start 0 settled: %v", err)
+ }
+ if err := f.ops.Failed(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("settling a start that was not reserved: %v", err)
+ }
+ if err := f.ops.Failed(ctx, f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 1); err != nil {
+ t.Fatal(err)
+ }
+ second := f.ops.CurrentAttempt(f.start)
+ if second.ID == first.ID || second.InvocationID == first.InvocationID || second.Session != o.Session || second.LaunchInputs != digest {
+ t.Fatalf("start 1: %+v", second)
+ }
+ changed := f.start
+ changed.Inputs.Env = []string{"CARLOS_MOVE_PROBE=1", "OTHER=1"}
+ if err := f.ops.Failed(ctx, f.start, 1); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.ops.Reserve(ctx, changed, 2); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("other launch inputs: %v", err)
+ }
+}
+
+// Every method refuses once the record or the claim no longer names the
+// admitted probe: the record's attempt index outside what T11 admitted, a
+// rollback reason, the record past probing, another session.
+func TestMovesProbeAttemptsRefuseWhenTheProbeMovedOn(t *testing.T) {
+ for name, mutate := range map[string]func(*paFixture){
+ "no attempt counted": func(f *paFixture) { f.rec.Move.Landing = nil },
+ "beyond the admitted": func(f *paFixture) { f.rec.Move.Landing.Attempts = 3 },
+ "none admitted": func(f *paFixture) { f.admit = 0 },
+ "a rollback is marked": func(f *paFixture) { f.rec.Move.Reason = provision.RollbackProbeAttempts },
+ "the record is probed": func(f *paFixture) { f.rec.Move.LandingPhase = provision.LandProbed },
+ "another move": func(f *paFixture) { f.rec.Move.ID = "mv_other" },
+ "another session": func(f *paFixture) {
+ c := paClaim(f.t)
+ c.Holder.Session, c.Gen = 21, 21
+ c.Holder.Admissions[1].Session = 21
+ f.putClaim(c)
+ },
+ "the claim is probed": func(f *paFixture) {
+ c := paClaim(f.t)
+ c.Op.Phase = hostclaim.PhaseLandingProbed
+ f.putClaim(c)
+ },
+ "a committed wake": func(f *paFixture) { f.start.Policy.Probe = false },
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newPAFixture(t)
+ mutate(f)
+ ctx := t.Context()
+ if _, err := f.ops.OpenProbe(ctx, f.start, time.Now()); err == nil {
+ t.Fatal("OpenProbe")
+ }
+ if err := f.ops.Reserve(ctx, f.start, 0); err == nil {
+ t.Fatal("Reserve")
+ }
+ if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); err == nil {
+ t.Fatal("Healthy")
+ }
+ })
+ }
+}
+
+// Healthy is the 20 s settle with the process alive, then the second
+// readiness request (moves §7.7); the start is journalled healthy only
+// after both.
+func TestMovesProbeAttemptsHealthySettlesAndAsksAgain(t *testing.T) {
+ f := newPAFixture(t)
+ ctx := t.Context()
+ if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ cmd := exec.Command("sleep", "30")
+ if err := cmd.Start(); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { cmd.Process.Kill(); cmd.Wait() })
+ began := time.Now()
+ asked := 0
+ if err := f.ops.Healthy(ctx, f.start, 0, cmd, func(context.Context) error {
+ asked++
+ if time.Since(began) < f.ops.Settle {
+ t.Error("asked before the settle")
+ }
+ return nil
+ }); err != nil || asked != 1 {
+ t.Fatalf("healthy: %v asked=%d", err, asked)
+ }
+ j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
+ if err != nil || !strings.Contains(j.Steps["probe_start_1"], `"state":"healthy"`) {
+ t.Fatalf("journal: %+v %v", j, err)
+ }
+
+ // A second request that fails, or a process that died in the settle,
+ // is not healthy.
+ g := newPAFixture(t)
+ if err := g.ops.Reserve(ctx, g.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ refused := errors.New("503")
+ if err := g.ops.Healthy(ctx, g.start, 0, cmd, func(context.Context) error { return refused }); !errors.Is(err, refused) {
+ t.Fatalf("a failing second request: %v", err)
+ }
+ dead := exec.Command("/bin/true")
+ if err := dead.Run(); err != nil {
+ t.Fatal(err)
+ }
+ if err := g.ops.Healthy(ctx, g.start, 0, dead, func(context.Context) error { return nil }); err == nil {
+ t.Fatal("a dead process was healthy")
+ }
+ if j, _ := moves.ReadJournal(g.dataDir, paHost, paOp); strings.Contains(j.Steps["probe_start_1"], "healthy") {
+ t.Fatalf("journalled healthy: %v", j.Steps)
+ }
+}