amadan

carlos / platform Public

hostclaim: a malformed journal counter or stamp, replacements included, is an error and never 0, and the relocation worker refuses on it before it looks for a spare

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude cb7059c32f87dd2e6b46be41b1ea0f0d58ff4ec2 parent b3381d2
4 files changed, +141 −23
  • internal/hostclaim/relocation.go +56 −10
  • internal/hostclaim/relocation_test.go +39 −6
  • internal/hostrelocate/worker.go +16 −7
  • internal/hostrelocate/worker_test.go +30 −0
diff --git a/internal/hostclaim/relocation.go b/internal/hostclaim/relocation.go
index 5238422c9..0c27547b1 100644
--- a/internal/hostclaim/relocation.go
+++ b/internal/hostclaim/relocation.go
@@ -238,9 +238,29 @@ func readJournal(op Op, key string, v any) bool {
return ok && json.Unmarshal([]byte(raw), v) == nil
}
-func journalUint(op Op, key string) uint64 {
- n, _ := strconv.ParseUint(op.Journal[key], 10, 64)
- return n
+// journalUint is a journaled counter or stamp: absent is 0, its starting
+// value; a value that does not parse is an error, never 0 (absence or
+// malformed data is never evidence).
+func journalUint(op Op, key string) (uint64, error) {
+ raw, ok := op.Journal[key]
+ if !ok || raw == "" {
+ return 0, nil
+ }
+ n, err := strconv.ParseUint(raw, 10, 64)
+ if err != nil {
+ return 0, fmt.Errorf("%w: journal %s is malformed: %q", ErrRelocation, key, raw)
+ }
+ return n, nil
+}
+
+// Replacements is the number of target replacements the operation has
+// used (§5.8's bound, MaxTargetReplacements). A malformed count is an
+// error, never 0.
+func Replacements(c *Claim) (uint64, error) {
+ if c == nil {
+ return 0, ErrMalformed
+ }
+ return journalUint(c.Op, "replacements")
}
func requireRelocation(c *Claim, opID string, phases ...string) error {
@@ -844,8 +864,12 @@ func ToInstalled(opID string, a PreparedAuthority) func(*Claim) error {
if err := requireAcks(c); err != nil {
return err
}
+ stamp, err := journalUint(c.Op, "install_stamp")
+ if err != nil {
+ return err
+ }
c.Op.Phase = PhaseInstalled
- a.InstallStamp = journalUint(c.Op, "install_stamp")
+ a.InstallStamp = stamp
return journalJSON(c, "prepared_authority", a)
}
}
@@ -881,8 +905,12 @@ func RecordLocalInstall(opID string, l LocalInstall) func(*Claim) error {
if err := requireRelocation(c, opID, PhaseInstalled); err != nil {
return err
}
+ stamp, err := journalUint(c.Op, "install_stamp")
+ if err != nil {
+ return err
+ }
if l.Box != c.Op.Exec || l.Enrollment != c.Op.Intent["target_enrollment"] ||
- l.InstallStamp == 0 || l.InstallStamp != journalUint(c.Op, "install_stamp") || !isHex64(l.JournalSHA256) {
+ l.InstallStamp == 0 || l.InstallStamp != stamp || !isHex64(l.JournalSHA256) {
return fmt.Errorf("%w: receipt is not this installation's", ErrRelocation)
}
if preparedOpened(c) {
@@ -907,9 +935,13 @@ func ToProbing(opID string, now time.Time) func(*Claim) error {
if err := requireRelocation(c, opID, PhaseInstalled); err != nil {
return err
}
+ stamp, err := journalUint(c.Op, "install_stamp")
+ if err != nil {
+ return err
+ }
var l LocalInstall
if !readJournal(c.Op, "local_install", &l) || l.Box != c.Op.Exec || l.Enrollment != c.Op.Intent["target_enrollment"] ||
- l.InstallStamp != journalUint(c.Op, "install_stamp") {
+ l.InstallStamp != stamp {
return fmt.Errorf("%w: no local install receipt for the executor", ErrRelocation)
}
if err := requireAcks(c); err != nil {
@@ -938,7 +970,10 @@ func BeginProbeAttempt(opID, box string) func(*Claim) error {
return fmt.Errorf("%w: only the executor probes", ErrRelocation)
}
key := "probe_attempts." + box
- n := journalUint(c.Op, key)
+ n, err := journalUint(c.Op, key)
+ if err != nil {
+ return err
+ }
if n >= MaxProbeAttempts {
return ErrProbeAttemptsExhausted
}
@@ -953,8 +988,12 @@ func RecordProbe(opID string, p Probe) func(*Claim) error {
if err := requireRelocation(c, opID, PhaseProbing); err != nil {
return err
}
+ stamp, err := journalUint(c.Op, "install_stamp")
+ if err != nil {
+ return err
+ }
if p.Box != c.Op.Exec || p.Box != c.Holder.WriterBox || p.Session == 0 || p.Session != c.Holder.Session ||
- p.Session < journalUint(c.Op, "install_stamp") {
+ p.Session < stamp {
return fmt.Errorf("%w: probe is not of the executor's admitted session", ErrRelocation)
}
return journalJSON(c, "probe", p)
@@ -1060,7 +1099,10 @@ func replaceTargetSlot(c *Claim, t Target) error {
}
old := c.Op.Intent["target_box"]
if old != "" && old != t.Box {
- n := journalUint(c.Op, "replacements")
+ n, err := journalUint(c.Op, "replacements")
+ if err != nil {
+ return err
+ }
if n >= MaxTargetReplacements {
return ErrReplacementsExceeded
}
@@ -1197,7 +1239,11 @@ func fenceTargetStuck(ctx context.Context, s release.Store, observed Observed, o
}
reason := noSpare
if reason == "" {
- if journalUint(c.Op, "replacements") < MaxTargetReplacements {
+ n, err := journalUint(c.Op, "replacements")
+ if err != nil {
+ return err
+ }
+ if n < MaxTargetReplacements {
return fmt.Errorf("%w: replacements remain; replace the target instead", ErrRelocation)
}
reason = fmt.Sprintf("%s terminated after %d target replacements", old, MaxTargetReplacements)
diff --git a/internal/hostclaim/relocation_test.go b/internal/hostclaim/relocation_test.go
index 98bf47e02..c7579dcb0 100644
--- a/internal/hostclaim/relocation_test.go
+++ b/internal/hostclaim/relocation_test.go
@@ -107,8 +107,17 @@ func relPrepared(gen string, maxTXID uint64, filesSeq int64) PreparedAuthority {
return a
}
+// journalN is journalUint for a journal the test's own transitions wrote.
+func journalN(op Op, key string) uint64 {
+ n, err := journalUint(op, key)
+ if err != nil {
+ panic(err)
+ }
+ return n
+}
+
func localInstall(c *Claim) LocalInstall {
- return LocalInstall{Box: c.Op.Exec, Enrollment: relEnrollment(c.Op.Exec), InstallStamp: journalUint(c.Op, "install_stamp"), FilesManifestVersion: 3, JournalSHA256: strings.Repeat("b", 64)}
+ return LocalInstall{Box: c.Op.Exec, Enrollment: relEnrollment(c.Op.Exec), InstallStamp: journalN(c.Op, "install_stamp"), FilesManifestVersion: 3, JournalSHA256: strings.Repeat("b", 64)}
}
// toProbing drives an operation at pinning (with its pin) through
@@ -182,7 +191,7 @@ func TestRelocationDrainEndToEnd(t *testing.T) {
o = step(t, s, o, ToInstalled("rel-1", relPrepared("P1", 40, 7)))
o = step(t, s, o, RecordLocalInstall("rel-1", localInstall(o.Claim)))
o = step(t, s, o, ToProbing("rel-1", relObserved))
- install := journalUint(o.Claim.Op, "install_stamp")
+ install := journalN(o.Claim.Op, "install_stamp")
if o.Claim.Op.Exec != relTarget || o.Claim.Holder.Stamp != install || !slices.Contains(o.Claim.Touched.Boxes, relTarget) {
t.Fatalf("installation: exec %s stamp %d install %d", o.Claim.Op.Exec, o.Claim.Holder.Stamp, install)
}
@@ -417,7 +426,7 @@ func TestRelocationTwoReplacementsThenStuck(t *testing.T) {
t.Fatalf("third replacement: %v", err)
}
o = step(t, s, o, MarkStuck("rel-1", "two target replacements used"))
- if journalUint(o.Claim.Op, "replacements") != 2 {
+ if journalN(o.Claim.Op, "replacements") != 2 {
t.Fatal("stuck reset the count")
}
if _, err := Update(ctx, s, o, ToInstalling("rel-1", Target{relSpare2, relEnrollment(relSpare2)}, "P9")); !errors.Is(err, ErrRelocation) {
@@ -425,6 +434,30 @@ func TestRelocationTwoReplacementsThenStuck(t *testing.T) {
}
}
+// Final review: a malformed replacements count is an error, never 0
+// (absence or malformed data is never evidence). Absent is 0.
+func TestRelocationMalformedReplacementsRefuses(t *testing.T) {
+ ctx := context.Background()
+ s, o := relAtPhase(t, PhasePinning)
+ if n, err := Replacements(o.Claim); err != nil || n != 0 {
+ t.Fatalf("absent: %d %v", n, err)
+ }
+ o = step(t, s, o, func(c *Claim) error { c.Op.Journal["replacements"] = "two"; return nil })
+ if _, err := Replacements(o.Claim); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("malformed count: %v", err)
+ }
+ if _, err := ReplaceTarget(ctx, s, o, "rel-1", putFence(t, s, relTarget), nil, Target{relSpare, relEnrollment(relSpare)}, "", relAuthority{}); !errors.Is(err, ErrRelocation) || errors.Is(err, ErrReplacementsExceeded) {
+ t.Fatalf("a replacement over a malformed count: %v", err)
+ }
+ after, err := Read(ctx, s, o.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after.ETag != o.ETag || after.Claim.Op.Intent["target_box"] != relTarget {
+ t.Fatalf("the claim moved: %+v", after.Claim.Op)
+ }
+}
+
func TestRelocationReplacementInReleasingAcknowledgesBeforePinning(t *testing.T) {
ctx := context.Background()
s, o := relAtPhase(t, PhaseReleasing)
@@ -727,7 +760,7 @@ func TestRelocationProbeAttemptsAreBounded(t *testing.T) {
for i := 0; i < MaxProbeAttempts; i++ {
o = step(t, s, o, BeginProbeAttempt("rel-1", relTarget))
}
- if journalUint(o.Claim.Op, "probe_attempts."+relTarget) != MaxProbeAttempts {
+ if journalN(o.Claim.Op, "probe_attempts."+relTarget) != MaxProbeAttempts {
t.Fatalf("count: %v", o.Claim.Op.Journal)
}
if _, err := Update(ctx, s, o, BeginProbeAttempt("rel-1", relTarget)); !errors.Is(err, ErrProbeAttemptsExhausted) {
@@ -741,7 +774,7 @@ func TestRelocationProbeAttemptsAreBounded(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if journalUint(o.Claim.Op, "probe_attempts."+relSpare) != 0 {
+ if journalN(o.Claim.Op, "probe_attempts."+relSpare) != 0 {
t.Fatal("the replacement inherited attempts")
}
}
@@ -1016,7 +1049,7 @@ func TestRelocationPreparedManifestIsSequenceZero(t *testing.T) {
o = step(t, s, o, ToInstalled("rel-1", relPrepared("P1", 40, 7)))
var a PreparedAuthority
readJournal(o.Claim.Op, "prepared_authority", &a)
- if a.FilesSeq != 0 || a.SourceFilesSeq != 7 || a.InstallStamp == 0 || a.InstallStamp != journalUint(o.Claim.Op, "install_stamp") {
+ if a.FilesSeq != 0 || a.SourceFilesSeq != 7 || a.InstallStamp == 0 || a.InstallStamp != journalN(o.Claim.Op, "install_stamp") {
t.Fatalf("journaled authority: %+v", a)
}
}
diff --git a/internal/hostrelocate/worker.go b/internal/hostrelocate/worker.go
index 0768f578b..1614e27c8 100644
--- a/internal/hostrelocate/worker.go
+++ b/internal/hostrelocate/worker.go
@@ -20,7 +20,6 @@ import (
"fmt"
"os"
"slices"
- "strconv"
"time"
"github.com/carlosframework/platform/internal/hostactors"
@@ -353,7 +352,9 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro
if c.Op.Journal["release"] == "" {
return w.releaseOverdue(ctx, obs, "probe_failed_since")
}
- if exhausted(c) {
+ if done, err := exhausted(c); err != nil {
+ return obs, err
+ } else if done {
return w.stuckOn(ctx, obs, hostclaim.ErrReplacementsExceeded)
}
next, err := w.next(ctx, obs)
@@ -503,7 +504,11 @@ func (w Worker) replace(ctx context.Context, obs hostclaim.Observed, ref hostcla
// with the replacements used, no spare is looked for; the dead
// target's fence is consumed into a stuck pinning (the loss retained,
// its installation's authority revoked, no probe to wait for).
- if exhausted(obs.Claim) {
+ done, err := exhausted(obs.Claim)
+ if err != nil {
+ return obs, err
+ }
+ if done {
out, err := hostclaim.FenceExhaustedTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, w.Fences)
if err == nil {
w.Notify(ctx)
@@ -550,10 +555,14 @@ func (w Worker) update(ctx context.Context, obs hostclaim.Observed, tr func(*hos
return hostclaim.Update(ctx, w.Claims, obs, tr)
}
-// exhausted reports whether the operation has used its target replacements.
-func exhausted(c *hostclaim.Claim) bool {
- n, _ := strconv.ParseUint(c.Op.Journal["replacements"], 10, 64)
- return n >= hostclaim.MaxTargetReplacements
+// exhausted reports whether the operation has used its target
+// replacements. A malformed count is an error, never 0.
+func exhausted(c *hostclaim.Claim) (bool, error) {
+ n, err := hostclaim.Replacements(c)
+ if err != nil {
+ return false, err
+ }
+ return n >= hostclaim.MaxTargetReplacements, nil
}
// acked is the D6 barrier as the worker reads it before a CAS that needs it.
diff --git a/internal/hostrelocate/worker_test.go b/internal/hostrelocate/worker_test.go
index 514dab20d..408aba755 100644
--- a/internal/hostrelocate/worker_test.go
+++ b/internal/hostrelocate/worker_test.go
@@ -549,6 +549,36 @@ func TestWorkerMarksExhaustionStuckBeforeLookingForASpare(t *testing.T) {
}
}
+// Final review: a malformed replacements count is an error, never 0. The
+// worker refuses before it looks for a spare, and the claim is unchanged.
+func TestWorkerRefusesAMalformedReplacementsCount(t *testing.T) {
+ f := newFixture(t)
+ if _, err := f.w.Start(context.Background(), host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
+ t.Fatal(err)
+ }
+ f.participant(func(c *hostclaim.Claim) error {
+ if c.Op.Journal == nil {
+ c.Op.Journal = map[string]string{}
+ }
+ c.Op.Journal["replacements"] = "two"
+ return nil
+ })
+ chose := 0
+ f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) {
+ chose++
+ return hostclaim.Target{}, errors.New("no eligible box")
+ }
+ f.now = f.now.Add(ReleaseDeadline + time.Minute)
+ f.terminate("dst")
+ before := f.claim()
+ if _, err := f.w.Step(context.Background(), host); !errors.Is(err, hostclaim.ErrRelocation) {
+ t.Fatalf("step over a malformed count: %v", err)
+ }
+ if after := f.claim(); after.ETag != before.ETag || chose != 0 {
+ t.Fatalf("the claim moved (%v) or a spare was looked for (%d)", after.ETag != before.ETag, chose)
+ }
+}
+
func TestWorkerFencesAnExhaustedTargetThatDiedAfterItsLocalInstall(t *testing.T) {
f := newFixture(t)
ctx := context.Background()