amadan

carlos / platform Public

host: the claim says whether a probe attempt admitted (Plan 3 Task 6 fix round 2)

1. probe_admitting_<k> journals the claim's writer session and admission
   count before the call. On restart attempt k is admitted, so consumed,
   when Task 5's binding exists or the claim's writer moved on since; only
   an unchanged claim reuses k. Closes the crash between the admission
   CAS and Reserve's binding.
2. Before an admitted attempt without probe_passed_<k> is consumed, the
   activator's live entry is asked: this move's, awake, with the claim's
   writer, the attempt passed; its pass is journalled and the phase writes
   follow. Closes a failed write of probe_passed_<k>.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude fa2f5ab892520857e824c9076c52b5524640678c parent 1214e14
2 files changed, +168 −19
  • internal/host/movesland_probe_test.go +64 −3
  • internal/host/moveslandprobe.go +104 −16
diff --git a/internal/host/movesland_probe_test.go b/internal/host/movesland_probe_test.go
index 45b44d76e..78f53dd0d 100644
--- a/internal/host/movesland_probe_test.go
+++ b/internal/host/movesland_probe_test.go
@@ -33,8 +33,11 @@ type probeLog struct {
before func(n int)
admissions []probeAdmission
openWindow bool
- last activator.ProbeResult
- live bool
+ // noBind leaves the admission without Task 5's binding: a crash
+ // between the admission CAS and Reserve(0).
+ noBind bool
+ last activator.ProbeResult
+ live bool
}
// probeAdmission is one admission the fake made: the record attempt it was
@@ -75,7 +78,9 @@ func (p *probeLog) wake(ctx context.Context, cfg activator.ProbeConfig) (activat
if err != nil {
return activator.ProbeResult{}, err
}
- j.Mark(probeStepSession+strconv.FormatUint(ah.Session, 10), strconv.Itoa(k))
+ if !p.noBind {
+ j.Mark(probeStepSession+strconv.FormatUint(ah.Session, 10), strconv.Itoa(k))
+ }
if p.openWindow {
j.Mark(probeStepOpened+strconv.Itoa(k), p.h.clock.Now().UTC().Format(time.RFC3339Nano))
}
@@ -412,3 +417,59 @@ func TestLandingProbeAdmittedAttemptIsConsumedOnRestart(t *testing.T) {
t.Fatalf("the only attempt was admitted again: %+v admissions %+v", m, p1.admissions)
}
}
+
+// Fix round 2, ruling 1: the claim, not the binding, says whether attempt k
+// was admitted. A crash after the admission CAS and before Reserve's
+// binding leaves probe_admitting_<k> and a claim whose writer session moved
+// on: the restart consumes k and never admits under it again.
+func TestLandingProbeAdmissionWithoutBindingIsConsumed(t *testing.T) {
+ h, p := installedLanding(t, 2)
+ p.noBind = true
+ p.answers = []error{errors.New("the controller died after the admission")}
+ h.land(t)
+ if len(p.admissions) != 1 || !h.landJournal(t).Done("probe_admitting_1") || attemptAdmitted(h.landJournal(t), 1) {
+ t.Fatalf("the crash: admissions %+v journal %v", p.admissions, h.landJournal(t).Steps)
+ }
+ h.freshTarget()
+ p.noBind = false
+ h.land(t)
+ h.atPhase(t, provision.LandProbed)
+ if len(p.admissions) != 2 || p.admissions[1].attempt != 2 {
+ t.Fatalf("index 1 was admitted twice: %+v", p.admissions)
+ }
+
+ // probe_admitting_<k> with the claim unchanged (no admission happened):
+ // the same index is used.
+ h2, p2 := installedLanding(t, 2)
+ h2.landAt = "before-probe-1"
+ h2.crashLand(t)
+ if !h2.landJournal(t).Done("probe_admitting_1") {
+ t.Fatalf("no admitting step before the call: %v", h2.landJournal(t).Steps)
+ }
+ h2.freshTarget()
+ h2.land(t)
+ if m := h2.atPhase(t, provision.LandProbed); m.Landing.Attempts != 1 || len(p2.admissions) != 1 || p2.admissions[0].attempt != 1 {
+ t.Fatalf("an unadmitted attempt was not reused: %+v %+v", m.Landing, p2.admissions)
+ }
+}
+
+// Fix round 2, ruling 2: a probe that passed but whose probe_passed_<k>
+// write failed is found through the activator's live entry: its writer is
+// the claim's, so the attempt passed, and the next pass reaches probed
+// without a second ProbeWake.
+func TestLandingProbePassFoundThroughTheLiveEntry(t *testing.T) {
+ h, p := installedLanding(t, 2)
+ h.landAt = "before-journal-passed"
+ h.crashLand(t)
+ if j := h.landJournal(t); j.Done("probe_passed_1") || len(p.configs) != 1 || !p.live {
+ t.Fatalf("the crash point: %v calls %d live %v", j.Steps, len(p.configs), p.live)
+ }
+ h.land(t)
+ h.atPhase(t, provision.LandProbed)
+ if len(p.configs) != 1 || len(p.admissions) != 1 || !h.landJournal(t).Done("probe_passed_1") {
+ t.Fatalf("probed again: %d calls %d admissions", len(p.configs), len(p.admissions))
+ }
+ if c := h.claim(t); c.Holder.Session != p.last.Session {
+ t.Fatalf("the probed writer: %+v", c.Holder)
+ }
+}
diff --git a/internal/host/moveslandprobe.go b/internal/host/moveslandprobe.go
index be7c1ae91..60acc0ce6 100644
--- a/internal/host/moveslandprobe.go
+++ b/internal/host/moveslandprobe.go
@@ -46,6 +46,12 @@ import (
// is admitted once Task 5's journalled session binding (probe_session_<s> =
// k) exists for it, and an admitted attempt is consumed on restart, failed
// unless it passed. Only a counted attempt with no binding is made again.
+// Fix round 2: the claim is the source of truth for the admission itself.
+// probe_admitting_<k> records the claim's writer before the call; a claim
+// whose writer moved on since admitted under k though Reserve's binding
+// never became durable. Before an admitted attempt with no probe_passed_<k>
+// is consumed, the activator's live entry is asked: this move's, awake,
+// with the claim's writer, it passed, and its pass is journalled then.
//
// A pass is journalled (probe_passed_<k>: the admitted session, stamp and
// generation) before the probing to probed record write (ruling 1). A pass
@@ -60,10 +66,20 @@ import (
// probeStepFailed marks attempt k over in the op journal; probeStepPassed
// names attempt k's passing writer (probePassed).
const (
- probeStepFailed = "probe_failed_"
- probeStepPassed = "probe_passed_"
+ probeStepFailed = "probe_failed_"
+ probeStepPassed = "probe_passed_"
+ probeStepAdmitting = "probe_admitting_"
)
+// probeAdmitting is a probe_admitting_<k> step's value, written before the
+// call: the claim's writer session and admission count the attempt started
+// from. The claim is the source of truth for whether the attempt admitted
+// (fix round 2, ruling 1).
+type probeAdmitting struct {
+ Session uint64 `json:"session"`
+ Admissions int `json:"admissions"`
+}
+
// probePassed is a probe_passed_<k> step's value: the admitted writer the
// passing attempt left serving.
type probePassed struct {
@@ -85,10 +101,10 @@ func MovesRecordAdmittedAttempts(_ context.Context, rec *provision.Record) (int,
}
// attemptSpent reports whether counted attempt k is over: its failure is
-// journalled, a session was admitted under it (Task 5's binding), or its
-// window opened and has passed.
-func (l *landingRun) attemptSpent(j *moves.Journal, k int) bool {
- if j.Done(probeStepFailed+strconv.Itoa(k)) || attemptAdmitted(j, k) {
+// journalled, it was admitted (attemptAdmittedBy), or its window opened and
+// has passed.
+func (l *landingRun) attemptSpent(j *moves.Journal, k int, h *hostclaim.Holder) bool {
+ if j.Done(probeStepFailed+strconv.Itoa(k)) || l.attemptAdmittedBy(j, k, h) {
return true
}
raw, ok := j.Steps[probeStepOpened+strconv.Itoa(k)]
@@ -111,6 +127,66 @@ func attemptAdmitted(j *moves.Journal, k int) bool {
return false
}
+// attemptAdmittedBy reports whether attempt k admitted a writer: Task 5's
+// binding exists for it, or probe_admitting_<k> was written and the claim
+// has since moved on (another writer session, or a newer cold admission on
+// this box). probe_admitting_<k> with the claim unchanged is an attempt that
+// never admitted, and only that may reuse its index.
+func (l *landingRun) attemptAdmittedBy(j *moves.Journal, k int, h *hostclaim.Holder) bool {
+ if attemptAdmitted(j, k) {
+ return true
+ }
+ raw, ok := j.Steps[probeStepAdmitting+strconv.Itoa(k)]
+ if !ok || h == nil {
+ return false
+ }
+ var a probeAdmitting
+ if err := json.Unmarshal([]byte(raw), &a); err != nil {
+ // Unreadable: never risk a second admission under k.
+ return true
+ }
+ if h.Session != a.Session {
+ return true
+ }
+ if len(h.Admissions) > a.Admissions {
+ last := h.Admissions[len(h.Admissions)-1]
+ return last.Box == l.cfg().Box && last.Mode == "cold"
+ }
+ return false
+}
+
+// livePass reports the activator's live probe entry when it is this move's
+// and its writer is the claim's current one: an attempt that passed though
+// its probe_passed_<k> never became durable (fix round 2, ruling 2).
+func (l *landingRun) livePass(h *hostclaim.Holder) (activator.ProbeResult, bool) {
+ live := l.cfg().ProbeLive
+ if live == nil || h == nil || h.WriterBox != l.cfg().Box {
+ return activator.ProbeResult{}, false
+ }
+ op, res, ok := live(l.rec.Host)
+ if !ok || op != l.g.ID || res.Session != h.Session || res.Stamp != h.Stamp || res.Generation != h.Generation {
+ return activator.ProbeResult{}, false
+ }
+ return res, true
+}
+
+// journalPassed writes probe_passed_<k> for the passing writer res.
+func (l *landingRun) journalPassed(k int, res activator.ProbeResult) error {
+ if err := l.fault("before-journal-passed"); err != nil {
+ return err
+ }
+ raw, err := json.Marshal(probePassed{Session: res.Session, Stamp: res.Stamp, Generation: res.Generation})
+ if err != nil {
+ return err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return err
+ }
+ j.Mark(probeStepPassed+strconv.Itoa(k), string(raw))
+ return j.Write(l.cfg().DataDir)
+}
+
// failAttempt journals attempt k over.
func (l *landingRun) failAttempt(k int) error {
j, err := l.journal()
@@ -198,7 +274,16 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) {
return false, err
}
k++
- } else if l.attemptSpent(j, k) {
+ } else if l.attemptSpent(j, k, obs.Claim.Holder) {
+ if !j.Done(probeStepFailed + strconv.Itoa(k)) {
+ if res, ok := l.livePass(obs.Claim.Holder); ok {
+ log.Printf("moves: %s: landing %s attempt %d passed (its live writer is the claim's); its phase writes follow", l.rec.Host, l.g.ID, k)
+ if err := l.journalPassed(k, res); err != nil {
+ return false, err
+ }
+ return l.passed(ctx)
+ }
+ }
if err := l.failAttempt(k); err != nil {
return false, err
}
@@ -218,10 +303,21 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) {
}
counted = k
}
+ // The claim's writer before the call: whether the attempt admitted is
+ // the claim's to say after a crash (ruling 1).
+ before, err := l.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ admitting, err := json.Marshal(probeAdmitting{Session: before.Claim.Holder.Session, Admissions: len(before.Claim.Holder.Admissions)})
+ if err != nil {
+ return false, err
+ }
if j, err = l.journal(); err != nil {
return false, err
}
j.Phase, j.Attempt = provision.LandProbing, k
+ j.Mark(probeStepAdmitting+strconv.Itoa(k), string(admitting))
if err := j.Write(cfg.DataDir); err != nil {
return false, err
}
@@ -237,15 +333,7 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) {
switch {
case err == nil:
log.Printf("moves: %s: landing %s passed its probe on attempt %d of %d", l.rec.Host, l.g.ID, k, n)
- raw, err := json.Marshal(probePassed{Session: res.Session, Stamp: res.Stamp, Generation: res.Generation})
- if err != nil {
- return false, err
- }
- if j, err = l.journal(); err != nil {
- return false, err
- }
- j.Mark(probeStepPassed+strconv.Itoa(k), string(raw))
- if err := j.Write(cfg.DataDir); err != nil {
+ if err := l.journalPassed(k, res); err != nil {
return false, err
}
return l.passed(ctx)