amadan

carlos / platform Public

host: bind a probe session to one record attempt (Plan 3 Task 5 fix round 1, finding 5)

The first Reserve(0) or OpenProbe journals probe_session_<session> = k.
Once the record's index moves past k, OpenProbe, Reserve, Failed and
Healthy refuse and CurrentAttempt answers the zero attempt: a new record
attempt needs a new cold admission. The test that let one session open a
second window now requires the refusal.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b02fd3b400452c10a5e7b3b91d0b06a4669951ac parent bd99ffd
2 files changed, +142 −13
  • internal/host/movesprobeattempts.go +83 −10
  • internal/host/movesprobeattempts_test.go +59 −3
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go
index e9dae6fed..b00fbb8fe 100644
--- a/internal/host/movesprobeattempts.go
+++ b/internal/host/movesprobeattempts.go
@@ -41,6 +41,11 @@ import (
// - 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.
+// - probe_session_<session>: the record attempt k the admitted session is
+// bound to, written by its first Reserve(0) or OpenProbe (fix round 1,
+// finding 5). A session serves one record attempt: once the record's
+// index moves past k every method refuses and CurrentAttempt answers
+// the zero attempt, so a new record attempt needs a new cold admission.
//
// Every method refuses (hostbox.ErrChanged) once the claim no longer names
// the admitted writer (op id, kind, session, stamp, generation; phase
@@ -62,15 +67,23 @@ type movesProbeAttempts struct {
Settle, Ready time.Duration
mu sync.Mutex
- current map[string]hostclaim.WakeRetryAttempt
+ current map[string]boundAttempt
+}
+
+// boundAttempt is a session's last reservation and the record attempt the
+// session is bound to.
+type boundAttempt struct {
+ k int
+ attempt hostclaim.WakeRetryAttempt
}
var _ OperationAttempts = (*movesProbeAttempts)(nil)
// Journal step names and start states.
const (
- probeStepOpened = "probe_opened_"
- probeStepStart = "probe_start_"
+ probeStepOpened = "probe_opened_"
+ probeStepStart = "probe_start_"
+ probeStepSession = "probe_session_"
probeReserved = "reserved"
probeFailed = "failed"
@@ -140,6 +153,24 @@ func probeKey(s OperationStart) string {
return s.Admitted.Host + "\x00" + strconv.FormatUint(s.Owner.Session, 10)
}
+// bindSession checks the session's journalled record attempt against k.
+// 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 {
+ step := probeStepSession + strconv.FormatUint(s.Owner.Session, 10)
+ raw, ok := j.Steps[step]
+ switch {
+ case ok && raw == strconv.Itoa(k):
+ return nil
+ case ok:
+ return fmt.Errorf("%w: session %d is bound to record attempt %s, not %d", hostbox.ErrChanged, s.Owner.Session, raw, k)
+ case !write:
+ return fmt.Errorf("%w: session %d is bound to no record attempt", hostbox.ErrChanged, s.Owner.Session)
+ }
+ j.Mark(step, strconv.Itoa(k))
+ return nil
+}
+
// 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)
@@ -155,8 +186,14 @@ func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, pr
if err != nil {
return time.Time{}, err
}
+ if err := bindSession(j, s, k, true); err != nil {
+ return time.Time{}, err
+ }
step := probeStepOpened + strconv.Itoa(k)
if raw, ok := j.Steps[step]; ok {
+ if err := j.Write(p.DataDir); err != nil {
+ return time.Time{}, err
+ }
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)
@@ -218,6 +255,9 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
if err != nil {
return err
}
+ if err := bindSession(j, s, k, n == 0); err != nil {
+ return err
+ }
st, ok, err := readStart(j, k)
if err != nil {
return err
@@ -240,6 +280,13 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
}
st = probeStart{Session: s.Owner.Session, N: n, State: probeReserved}
}
+ if _, bound := j.Steps[probeStepSession+strconv.FormatUint(s.Owner.Session, 10)]; bound && st.Attempt.ID != "" {
+ // Reserve(0) validated an existing reservation; the binding it may
+ // have just written is made durable.
+ if err := j.Write(p.DataDir); err != nil {
+ return err
+ }
+ }
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}
@@ -251,21 +298,33 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in
}
}
if p.current == nil {
- p.current = map[string]hostclaim.WakeRetryAttempt{}
+ p.current = map[string]boundAttempt{}
}
- p.current[probeKey(s)] = st.Attempt
+ p.current[probeKey(s)] = boundAttempt{k: k, attempt: 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.
+// It reads the record again (the interface gives it no context, so its own
+// bounded one): once the record's index is past the bound attempt it answers
+// the zero attempt.
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)]
+ cur, ok := p.current[probeKey(s)]
+ p.mu.Unlock()
+ if !ok {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if k, err := p.bound(ctx, s); err != nil || k != cur.k {
+ return hostclaim.WakeRetryAttempt{}
+ }
+ return cur.attempt
}
// Failed settles start n, already stopped and drained, as failed.
@@ -286,11 +345,15 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
if err != nil {
return err
}
+ if err := bindSession(j, s, k, false); 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)] {
+ cur := p.current[probeKey(s)]
+ if !ok || st.Session != s.Owner.Session || st.N != n || st.State != probeReserved || cur.k != k || st.Attempt != cur.attempt {
return hostbox.ErrChanged
}
st.State = state
@@ -302,7 +365,17 @@ func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int
// 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 {
+ k, err := p.bound(ctx, s)
+ if err != nil {
+ return err
+ }
+ p.mu.Lock()
+ j, err := p.journal(s)
+ if err == nil {
+ err = bindSession(j, s, k, false)
+ }
+ p.mu.Unlock()
+ if err != nil {
return err
}
if again == nil || !probeAlive(cmd) {
@@ -326,7 +399,7 @@ func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n in
return fmt.Errorf("%w: the app exited during the settle", hostbox.ErrChanged)
}
asked, cancel := context.WithTimeout(ctx, ready)
- err := again(asked)
+ err = again(asked)
cancel()
if err != nil {
return err
diff --git a/internal/host/movesprobeattempts_test.go b/internal/host/movesprobeattempts_test.go
index b1b3def6b..2ccffa49e 100644
--- a/internal/host/movesprobeattempts_test.go
+++ b/internal/host/movesprobeattempts_test.go
@@ -127,17 +127,73 @@ func TestMovesProbeAttemptsOpenProbeIsJournalledOnce(t *testing.T) {
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)
}
+ // The next record attempt is a new cold admission: the same session
+ // never opens its window.
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)
+ if _, err := f.ops.OpenProbe(ctx, f.start, t2); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("the next record attempt on the same session: %v", err)
+ }
+ next := f.readmit(21)
+ if got, err := f.ops.OpenProbe(ctx, next, t2); err != nil || !got.Equal(t2) {
+ t.Fatalf("the next record attempt's admission: %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") {
+ if err != nil || !j.Done("probe_opened_1") || !j.Done("probe_opened_2") || j.Steps["probe_session_20"] != "1" || j.Steps["probe_session_21"] != "2" {
t.Fatalf("journal: %+v %v", j, err)
}
}
+// readmit stores the claim after a new cold admission at session and
+// returns that admission's start.
+func (f *paFixture) readmit(session uint64) OperationStart {
+ f.t.Helper()
+ c := paClaim(f.t)
+ c.Holder.Session, c.Gen = session, session
+ c.Holder.Generation = "fresh"
+ c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: session, Box: "uk-box", Generation: "fresh", Predecessor: paOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
+ f.putClaim(c)
+ s := f.start
+ s.Admitted = f.read()
+ s.Owner.Session, s.Owner.ClaimETag = session, s.Admitted.ETag
+ return s
+}
+
+// A session is bound to the record attempt it first journalled (at
+// Reserve(0) or OpenProbe): once the record's index moves past it, every
+// method refuses and CurrentAttempt answers nothing, even across a
+// controller restart.
+func TestMovesProbeAttemptsBindTheSessionToOneRecordAttempt(t *testing.T) {
+ f := newPAFixture(t)
+ ctx := t.Context()
+ if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
+ t.Fatal(err)
+ }
+ if f.ops.CurrentAttempt(f.start).ID == "" {
+ t.Fatal("no reservation")
+ }
+ f.rec.Move.Landing.Attempts = 2
+ if a := f.ops.CurrentAttempt(f.start); a.ID != "" {
+ t.Fatalf("a reservation of record attempt 1 answered for attempt 2: %+v", a)
+ }
+ if err := f.ops.Failed(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("Failed: %v", err)
+ }
+ if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("Reserve(0): %v", err)
+ }
+ if _, err := f.attempts().OpenProbe(ctx, f.start, time.Now()); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("OpenProbe after a restart: %v", err)
+ }
+ if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("Healthy: %v", err)
+ }
+ next := f.readmit(21)
+ if err := f.ops.Reserve(ctx, next, 0); err != nil {
+ t.Fatalf("the new admission: %v", 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