amadan

carlos / platform Public

hostrelocate: a passed probe whose record never reached the claim is recovered from the box's attempt journal, keyed by the admitted session, before preflight or any attempt (final review A2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude ecef64d4d366c3c2468033d8e3e07cae1e1e8a4b parent c53e60b
7 files changed, +269 −7
  • internal/host/relocationassembly.go +1 −0
  • internal/host/relocationattempts.go +45 −0
  • internal/host/relocationattempts_test.go +78 −0
  • internal/host/relocationeffects.go +13 −0
  • internal/hostrelocate/participant.go +38 −7
  • internal/hostrelocate/participant_gate_test.go +72 −0
  • internal/hostrelocate/participant_test.go +22 −0
diff --git a/internal/host/relocationassembly.go b/internal/host/relocationassembly.go
index 861bfa469..700acd375 100644
--- a/internal/host/relocationassembly.go
+++ b/internal/host/relocationassembly.go
@@ -136,6 +136,7 @@ func (a ClaimAssembly) RelocationEffects(cfg RelocationAssemblyConfig, ack func(
VerifyLocal: a.relocationVerifyLocal(cfg),
Receipt: a.relocationReceipt(cfg),
Admit: a.relocationAdmit(cfg),
+ Passed: relocationPassedProbe(cfg.DataDir),
StopProbe: a.relocationStopProbe(cfg),
ReleaseHold: a.relocationReleaseHold(cfg),
}
diff --git a/internal/host/relocationattempts.go b/internal/host/relocationattempts.go
index 563aa2441..09794d666 100644
--- a/internal/host/relocationattempts.go
+++ b/internal/host/relocationattempts.go
@@ -15,6 +15,7 @@ import (
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostrelocate"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/moves"
@@ -371,6 +372,50 @@ func (p *relocationProbeAttempts) Healthy(ctx context.Context, s OperationStart,
return p.settle(ctx, s, n, probeHealthy)
}
+// relocationPassedProbe is hostrelocate.Effects.PassedProbe (final review
+// A2): the attempt journal's healthy start for the claim's holder session,
+// the state Healthy settles once the probe passed, read as the agent
+// reads its own journal. The start must be the executor's record attempt
+// k, bound to the holder's session, generation and stamp, so a start of
+// another attempt or session is never taken for this one. After P's
+// opening the journal exists (OpenProbe wrote it before the app first
+// started), so a missing journal is a refusal like an unreadable one,
+// never "no probe passed": the worker's deadlines decide such an
+// operation, not an absence.
+func relocationPassedProbe(dataDir string) func(context.Context, hostclaim.Observed) (uint64, bool, error) {
+ return func(_ context.Context, o hostclaim.Observed) (uint64, bool, error) {
+ c := o.Claim
+ if dataDir == "" || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || !hostclaim.PreparedOpened(c) {
+ return 0, false, ErrClaimStopUnavailable
+ }
+ h := c.Holder
+ if h.WriterBox != c.Op.Exec || h.Session == 0 {
+ return 0, false, hostbox.ErrChanged
+ }
+ k, err := strconv.Atoi(c.Op.Journal["probe_attempts."+c.Op.Exec])
+ if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
+ return 0, false, hostbox.ErrChanged
+ }
+ path, err := relocationAttemptsPath(dataDir, c.Op.ID)
+ if err != nil {
+ return 0, false, err
+ }
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ return 0, false, fmt.Errorf("%w: the attempt journal of %s after P's opening: %w", hostrelocate.ErrRefused, c.Op.ID, err)
+ }
+ j, err := decodeRelocationAttempts(raw, o.Host, c.Op.ID)
+ if err != nil {
+ return 0, false, err
+ }
+ st, ok := j.Starts[strconv.Itoa(k)]
+ if !ok || j.Sessions[strconv.FormatUint(h.Session, 10)] != k || st.Session != h.Session || st.Generation != h.Generation || st.Stamp != h.Stamp || st.Attempt.Box != c.Op.Exec || st.State != probeHealthy {
+ return 0, false, nil
+ }
+ return h.Session, true, nil
+ }
+}
+
// relocationAttemptReader is root's authoritative attempt reader for kind
// relocating (ruling 48; claim Plan 11 "install and test root-owned
// authoritative attempt readers"): it reads the operation's durable
diff --git a/internal/host/relocationattempts_test.go b/internal/host/relocationattempts_test.go
index 829ea8d8d..2e1623c1b 100644
--- a/internal/host/relocationattempts_test.go
+++ b/internal/host/relocationattempts_test.go
@@ -15,6 +15,7 @@ import (
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostrelocate"
"github.com/carlosframework/platform/internal/provision"
)
@@ -257,3 +258,80 @@ func TestRootReaderNeedsABoundReservedStart(t *testing.T) {
t.Fatalf("an attempt past the bound: %v", err)
}
}
+
+// Final review A2: the passed probe is recovered from the journal only as
+// the healthy start of the executor's record attempt bound to the holder's
+// session, generation and stamp; any other start is no probe; a missing or
+// unreadable journal after P's opening is a refusal, never "no probe".
+func TestRelocationPassedProbeReadsTheHealthyStartForTheAdmittedSession(t *testing.T) {
+ ctx := context.Background()
+ dd := t.TempDir()
+ claim := func() *hostclaim.Claim {
+ return &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Incarnation: "inc", Stamp: 9, Session: 7, Generation: "P1", WriterBox: "dst"},
+ Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "op1", Phase: hostclaim.PhaseRelocatingProbing, Exec: "dst",
+ Journal: map[string]string{"install_stamp": "9", "prepared": "P1", "prepared_opened": "P1", "probe_attempts.dst": "1"}}}
+ }
+ passed := relocationPassedProbe(dd)
+ if _, _, err := passed(ctx, hostclaim.Observed{Host: "shop.example", Claim: claim()}); !errors.Is(err, hostrelocate.ErrRefused) {
+ t.Fatalf("a missing journal after P's opening: %v", err)
+ }
+ write := func(state string, change func(*relocationStart)) {
+ t.Helper()
+ j, _ := readRelocationAttempts(dd, "shop.example", "op1")
+ j.Sessions["7"] = 1
+ st := relocationStart{Session: 7, Generation: "P1", Stamp: 9, State: state, Attempt: hostclaim.WakeRetryAttempt{ID: "a", InvocationID: "b", Session: 7, Box: "dst"}}
+ if change != nil {
+ change(&st)
+ }
+ j.Starts["1"] = st
+ if err := j.write(dd); err != nil {
+ t.Fatal(err)
+ }
+ }
+ write(probeHealthy, nil)
+ if s, ok, err := passed(ctx, hostclaim.Observed{Host: "shop.example", Claim: claim()}); err != nil || !ok || s != 7 {
+ t.Fatalf("the healthy start: %d %v %v", s, ok, err)
+ }
+ for name, tc := range map[string]struct {
+ state string
+ change func(*relocationStart)
+ claim func(*hostclaim.Claim)
+ }{
+ "reserved": {probeReserved, nil, nil},
+ "failed": {probeFailed, nil, nil},
+ "another session": {probeHealthy, func(s *relocationStart) { s.Session = 8; s.Attempt.Session = 8 }, nil},
+ "another generation": {probeHealthy, func(s *relocationStart) { s.Generation = "g2" }, nil},
+ "another stamp": {probeHealthy, func(s *relocationStart) { s.Stamp = 10 }, nil},
+ "another box's attempt": {probeHealthy, func(s *relocationStart) { s.Attempt.Box = "src" }, nil},
+ "a later record attempt": {probeHealthy, nil, func(c *hostclaim.Claim) { c.Op.Journal["probe_attempts.dst"] = "2" }},
+ } {
+ write(tc.state, tc.change)
+ c := claim()
+ if tc.claim != nil {
+ tc.claim(c)
+ }
+ if s, ok, err := passed(ctx, hostclaim.Observed{Host: "shop.example", Claim: c}); err != nil || ok || s != 0 {
+ t.Fatalf("%s: %d %v %v", name, s, ok, err)
+ }
+ }
+ for name, change := range map[string]func(*hostclaim.Claim){
+ "P not opened": func(c *hostclaim.Claim) { delete(c.Op.Journal, "prepared_opened") },
+ "installed": func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseInstalled },
+ "another writer": func(c *hostclaim.Claim) { c.Holder.WriterBox = "src" },
+ "no attempt count": func(c *hostclaim.Claim) { delete(c.Op.Journal, "probe_attempts.dst") },
+ } {
+ write(probeHealthy, nil)
+ c := claim()
+ change(c)
+ if _, ok, err := passed(ctx, hostclaim.Observed{Host: "shop.example", Claim: c}); err == nil || ok {
+ t.Fatalf("%s: %v %v", name, ok, err)
+ }
+ }
+ path, _ := relocationAttemptsPath(dd, "op1")
+ if err := os.WriteFile(path, []byte("{"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok, err := passed(ctx, hostclaim.Observed{Host: "shop.example", Claim: claim()}); err == nil || ok {
+ t.Fatalf("an unreadable journal: %v %v", ok, err)
+ }
+}
diff --git a/internal/host/relocationeffects.go b/internal/host/relocationeffects.go
index 960df4242..b8f32a627 100644
--- a/internal/host/relocationeffects.go
+++ b/internal/host/relocationeffects.go
@@ -90,6 +90,9 @@ type RelocationEffects struct {
VerifyLocal func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error)
Receipt func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, bool, error)
Admit func(context.Context, hostclaim.Observed, bool) (uint64, bool, error)
+ // Passed is the probe this box already passed for the admitted session,
+ // from its attempt journal (relocationPassedProbe; final review A2).
+ Passed func(context.Context, hostclaim.Observed) (uint64, bool, error)
// PreflightKind is root's isolated-launch capability for kind
// relocating (ClaimRoleAssembly.Preflight then PreflightKind; ruling 48).
PreflightKind func(context.Context, hostclaim.Observed) error
@@ -343,6 +346,16 @@ func (e *RelocationEffects) Probe(ctx context.Context, obs hostclaim.Observed, c
return e.Admit(ctx, obs, certified)
}
+// PassedProbe recovers a passed probe whose record never reached the claim
+// (final review A2) from the box's durable attempt journal, keyed by the
+// admitted session.
+func (e *RelocationEffects) PassedProbe(ctx context.Context, obs hostclaim.Observed) (uint64, bool, error) {
+ if e.Passed == nil {
+ return 0, false, ErrClaimStopUnavailable
+ }
+ return e.Passed(ctx, obs)
+}
+
// Preflight refuses before an attempt is counted while root cannot launch
// a relocating probe (ruling 48).
func (e *RelocationEffects) Preflight(ctx context.Context, obs hostclaim.Observed) error {
diff --git a/internal/hostrelocate/participant.go b/internal/hostrelocate/participant.go
index 3901cef7f..6e1504a36 100644
--- a/internal/hostrelocate/participant.go
+++ b/internal/hostrelocate/participant.go
@@ -71,6 +71,17 @@ type Effects interface {
// caller counts an attempt in the claim before each call, so a Probe
// that returned early would spend the attempts without probing.
Probe(ctx context.Context, obs hostclaim.Observed, certified bool) (session uint64, passed bool, err error)
+ // PassedProbe reports the probe this box already made for the claim's
+ // admitted session when its own durable attempt record holds it as
+ // healthy (stage B final review A2): after Probe returned passed, the
+ // claim read or the CAS that records the result may have failed,
+ // leaving the live, passed session with no journaled probe. It is
+ // asked once P is opened and no probe is journaled, before Preflight
+ // or any attempt, so the result is recovered rather than a new attempt
+ // launched (none can be: ruling 58, until claim ask C11). ok is false
+ // when the record holds no healthy start for the holder's session; a
+ // record that cannot be read is the error, never "no probe".
+ PassedProbe(ctx context.Context, obs hostclaim.Observed) (session uint64, ok bool, err error)
// Preflight is called in probing before an attempt is counted (region
// HA ruling 48): root's isolated launch must speak the protocol and
// install an attempt reader for relocating, or nothing is counted.
@@ -203,6 +214,18 @@ func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed,
if !ok {
return p.update(ctx, obs, hostclaim.MarkStuck(id, "the local install receipt is missing after P was admitted (§5.8, B2-5)"))
}
+ // A probe this box already passed whose record never reached
+ // the claim (the read or the CAS after Probe failed) is
+ // recovered from the box's durable attempt record, keyed by
+ // the admitted session, before Preflight can turn the live,
+ // passed session into probe-failed (final review A2).
+ session, passed, err := p.Effects.PassedProbe(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ if passed {
+ return p.recordProbe(ctx, obs, session)
+ }
}
// EnsureLocal may have written the claim: count on a fresh read
// of the same installation.
@@ -245,13 +268,7 @@ func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed,
if err != nil || !passed {
return obs, errors.Join(ErrNotYet, err)
}
- // The probe's admission wrote the claim: record the result on a
- // fresh read, which RecordProbe binds to the admitted session.
- fresh, err := hostclaim.Read(ctx, p.Claims, host)
- if err != nil || fresh.Absent {
- return obs, errors.Join(ErrNotYet, err)
- }
- return p.update(ctx, fresh, hostclaim.RecordProbe(id, hostclaim.Probe{Box: p.Box, Session: session, At: p.Now()}))
+ return p.recordProbe(ctx, obs, session)
case executor && c.Op.Phase == hostclaim.PhaseRelocatingProbing && c.Op.Journal["probe_stop"] == "":
// Ruling 49: the passed probe's session is stopped, with root's
// receipt, before the serving CAS ends it.
@@ -268,6 +285,20 @@ func (p Participant) Step(ctx context.Context, host string) (hostclaim.Observed,
return obs, ErrNotYet
}
+// recordProbe journals the passed probe of session on a fresh read: the
+// probe's admission wrote the claim, and RecordProbe binds the result to
+// the admitted session. At is the time of this record; for a probe
+// recovered from the box's attempt record (PassedProbe) that is later than
+// the pass, never earlier, so the worker's release deadline (ruling 67)
+// is counted from no sooner than the probe passed.
+func (p Participant) recordProbe(ctx context.Context, obs hostclaim.Observed, session uint64) (hostclaim.Observed, error) {
+ fresh, err := hostclaim.Read(ctx, p.Claims, obs.Host)
+ if err != nil || fresh.Absent {
+ return obs, errors.Join(ErrNotYet, err)
+ }
+ return p.update(ctx, fresh, hostclaim.RecordProbe(obs.Claim.Op.ID, hostclaim.Probe{Box: p.Box, Session: session, At: p.Now()}))
+}
+
// ErrLaterAttemptUnlaunchable is Preflight's answer for a cold attempt
// after P's opening while root pins every relocating launch to
// install_stamp (claimLaunchAuthority, claim ask C11; ruling 58).
diff --git a/internal/hostrelocate/participant_gate_test.go b/internal/hostrelocate/participant_gate_test.go
index 0f315899a..314546021 100644
--- a/internal/hostrelocate/participant_gate_test.go
+++ b/internal/hostrelocate/participant_gate_test.go
@@ -8,6 +8,7 @@ import (
"testing"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
)
// driveToProbing runs a drain until dst's participant is in probing with
@@ -83,6 +84,77 @@ func TestParticipantFailsAProbeRootCannotRelaunch(t *testing.T) {
}
}
+// flakyStore is the claim store with one failure armed: the next Get (the
+// read after a passed probe) or Put (the CAS that records it) fails once,
+// as a transient store error would, and the store is itself again.
+type flakyStore struct {
+ release.Store
+ armed bool
+ puts bool // fail a Put rather than a Get
+}
+
+var errFlaky = errors.New("the claim store failed once")
+
+func (s *flakyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.armed && !s.puts {
+ s.armed = false
+ return nil, "", errFlaky
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func (s *flakyStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if s.armed && s.puts {
+ s.armed = false
+ return "", errFlaky
+ }
+ return s.Store.Put(ctx, key, body, ifMatch)
+}
+
+// Final review A2 (Astra 2): after Probe returned passed, the claim read
+// or the CAS that records the result fails once. The next pass recovers
+// the result from the box's durable attempt record, keyed by the admitted
+// session, and journals it: no Preflight, no second attempt, no
+// probe-failed, and the live session is the one journaled. A record that
+// cannot be read is retried, never taken as "no probe".
+func TestParticipantRecoversAPassedProbeTheClaimDidNotRecord(t *testing.T) {
+ for name, puts := range map[string]bool{"the claim read fails once": false, "the CAS fails once": true} {
+ t.Run(name, func(t *testing.T) {
+ f := newFixture(t)
+ ctx := context.Background()
+ flaky := &flakyStore{Store: f.s, puts: puts}
+ src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", flaky: flaky}
+ _, pd := driveToProbing(t, f, src, dst)
+ pd.Claims = flaky
+ if _, err := pd.Step(ctx, host); !errors.Is(err, errFlaky) {
+ t.Fatalf("the pass after the probe passed: %v", err)
+ }
+ c := f.claim().Claim
+ if !hostclaim.PreparedOpened(c) || c.Op.Journal["probe"] != "" || flaky.armed {
+ t.Fatalf("after the failure: opened %v, probe %q, armed %v", hostclaim.PreparedOpened(c), c.Op.Journal["probe"], flaky.armed)
+ }
+ dst.preflightErr = ErrLaterAttemptUnlaunchable // what root would answer a new attempt
+ dst.effects = nil
+ dst.passedErr = errors.New("the attempt journal cannot be read")
+ if _, err := pd.Step(ctx, host); !errors.Is(err, dst.passedErr) {
+ t.Fatalf("an unreadable record: %v", err)
+ }
+ dst.passedErr = nil
+ o, err := pd.Step(ctx, host)
+ if err != nil {
+ t.Fatalf("the recovering pass: %v", err)
+ }
+ var p hostclaim.Probe
+ if o.Claim.Op.Phase != hostclaim.PhaseRelocatingProbing || json.Unmarshal([]byte(o.Claim.Op.Journal["probe"]), &p) != nil || p.Session != o.Claim.Holder.Session || p.Box != "dst" || p.At.IsZero() {
+ t.Fatalf("phase %s probe %q", o.Claim.Op.Phase, o.Claim.Op.Journal["probe"])
+ }
+ if len(dst.certified) != 1 || o.Claim.Op.Journal["probe_attempts.dst"] != "1" || slices.Contains(dst.effects, "preflight") {
+ t.Fatalf("admissions %d attempts %q effects %v", len(dst.certified), o.Claim.Op.Journal["probe_attempts.dst"], dst.effects)
+ }
+ })
+ }
+}
+
// Ruling 62: an install that writes the claim (the typed inventory) has its
// receipt recorded on a fresh read, not a conflict loop on the stale one.
func TestParticipantRecordsItsReceiptAfterAnEffectWroteTheClaim(t *testing.T) {
diff --git a/internal/hostrelocate/participant_test.go b/internal/hostrelocate/participant_test.go
index 7ccba9449..39b6b5389 100644
--- a/internal/hostrelocate/participant_test.go
+++ b/internal/hostrelocate/participant_test.go
@@ -43,6 +43,13 @@ type fakeBox struct {
// writesOnInstall: the install writes the claim (the typed inventory),
// as the production install's RecordInventory does (ruling 62).
writesOnInstall bool
+ // healthy stands for the box's durable attempt journal: the sessions
+ // whose probe passed and settled (final review A2). passedErr is a
+ // journal that cannot be read. flaky, when set, is armed by Probe so
+ // the claim access after it fails once.
+ healthy map[uint64]bool
+ passedErr error
+ flaky *flakyStore
}
func (b *fakeBox) Acknowledge(_ context.Context, o hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
@@ -123,8 +130,23 @@ func (b *fakeBox) Probe(ctx context.Context, o hostclaim.Observed, certified boo
if b.failProbes {
return next.Claim.Holder.Session, false, nil
}
+ if b.healthy == nil {
+ b.healthy = map[uint64]bool{}
+ }
+ b.healthy[next.Claim.Holder.Session] = true
+ if b.flaky != nil {
+ b.flaky.armed = true
+ }
return next.Claim.Holder.Session, true, nil
}
+func (b *fakeBox) PassedProbe(_ context.Context, o hostclaim.Observed) (uint64, bool, error) {
+ b.effects = append(b.effects, "passed-probe")
+ if b.passedErr != nil {
+ return 0, false, b.passedErr
+ }
+ s := o.Claim.Holder.Session
+ return s, b.healthy[s], nil
+}
func (f *fixture) clock() time.Time { return f.now }