amadan

carlos / platform Public

hostclaim: admission retries re-verify the declaration, recovery intent keeps exact integers, create-only retries compare canonical payloads (Plan 8 review round 1)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 2f390d47639567373a9c558900a1b569b03a6f27 parent 47043c9
4 files changed, +196 −11
  • internal/hostclaim/lossrecord.go +4 −1
  • internal/hostclaim/lossrecord_test.go +131 −0
  • internal/hostclaim/lost.go +27 −10
  • internal/hostclaim/lost_test.go +34 −0

CI passed — run details

diff --git a/internal/hostclaim/lossrecord.go b/internal/hostclaim/lossrecord.go
index 59d2712fc..7891f74ce 100644
--- a/internal/hostclaim/lossrecord.go
+++ b/internal/hostclaim/lossrecord.go
@@ -1,6 +1,7 @@
package hostclaim
import (
+ "bytes"
"context"
"crypto/ed25519"
"encoding/json"
@@ -313,7 +314,9 @@ func CreateLossRecord(ctx context.Context, s release.Store, sg release.Signer, r
etag, err := s.Put(ctx, key, wire, "")
if errors.Is(err, release.ErrCASConflict) {
existing, readErr := ReadLossRecord(ctx, s, sg.RootPublicKey(), record.ID)
- if readErr == nil && reflect.DeepEqual(existing.Record, record) {
+ existingPayload, existingErr := json.Marshal(existing.Record)
+ payload, payloadErr := json.Marshal(record)
+ if readErr == nil && existingErr == nil && payloadErr == nil && bytes.Equal(existingPayload, payload) {
return existing, nil
}
return ObservedLossRecord{}, fmt.Errorf("%w: declaration id already names different evidence", ErrLossRecord)
diff --git a/internal/hostclaim/lossrecord_test.go b/internal/hostclaim/lossrecord_test.go
index 9bb59fa6d..f274a374b 100644
--- a/internal/hostclaim/lossrecord_test.go
+++ b/internal/hostclaim/lossrecord_test.go
@@ -102,6 +102,50 @@ func TestTimerKnowledgeNeverConflatesUnknownWithEmptyComplete(t *testing.T) {
}
}
+func TestLossRecordCreateRetryUsesSerializedEvidence(t *testing.T) {
+ for _, name := range []string{"ordinary timestamps", "spaced recovery intent", "both"} {
+ t.Run(name, func(t *testing.T) {
+ ctx := t.Context()
+ s := release.NewDirStore(t.TempDir())
+ _, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sg := release.RootSigner(priv)
+ record := validLossRecord(t)
+ if name != "spaced recovery intent" {
+ now := time.Now()
+ record.AcceptedAt, record.FencedAt, record.CloudObservation.ObservedAt = now, now, now
+ }
+ if name != "ordinary timestamps" {
+ record.ConfirmedConsequences.RecoveryIntent = json.RawMessage(`{ "kind": "lost-target", "txid": 9007199254740993 }`)
+ }
+ created, err := CreateLossRecord(ctx, s, sg, record)
+ if err != nil {
+ t.Fatal(err)
+ }
+ retry, err := CreateLossRecord(ctx, s, sg, record)
+ if err != nil || retry.ETag != created.ETag {
+ t.Errorf("identical serialized evidence retry = %+v, %v", retry, err)
+ }
+ changed := record
+ changed.FencedAt = changed.FencedAt.Add(time.Nanosecond)
+ if _, err := CreateLossRecord(ctx, s, sg, changed); !errors.Is(err, ErrLossRecord) {
+ t.Fatalf("changed evidence accepted: %v", err)
+ }
+ changed = record
+ changed.ConfirmedConsequences.RecoveryIntent = json.RawMessage(`{"kind":"lost-target","txid":9007199254740992}`)
+ if _, err := CreateLossRecord(ctx, s, sg, changed); !errors.Is(err, ErrLossRecord) {
+ t.Fatalf("changed recovery intent accepted: %v", err)
+ }
+ fresh, err := ReadLossRecord(ctx, s, sg.RootPublicKey(), record.ID)
+ if err != nil || fresh.ETag != created.ETag {
+ t.Fatalf("retry changed stored evidence: %v", err)
+ }
+ })
+ }
+}
+
func TestAdmissionLossRetentionRequiresExactDurableCopy(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
@@ -236,6 +280,93 @@ func TestLossBearingAdmissionRequiresEveryFrozenTargetAndSourceCut(t *testing.T)
if last.LossID != "loss_1" || !reflect.DeepEqual(last.AdditionalWritesLost, []WritesLost{earlier, source}) || last.WritesLost == nil || last.WritesLost.RemoteMaxTXID != 12 || last.WritesLost.Session != 4 {
t.Fatalf("loss admission = %+v", last)
}
+ t.Run("retry validates authorization and evidence", func(t *testing.T) {
+ denied := errors.New("recovery no longer authorized")
+ for _, name := range []string{
+ "identical", "primary txid", "primary txid and baseline", "region", "predecessor region",
+ "observed session", "observed generation", "observed phase", "observed operation",
+ "certified", "coverage", "additional loss", "missing baseline", "changed baseline",
+ "owner denial", "operation", "phase", "executor", "incarnation", "signature", "missing declaration",
+ } {
+ t.Run(name, func(t *testing.T) {
+ candidate := in
+ candidate.AdditionalWritesLost = append([]WritesLost(nil), in.AdditionalWritesLost...)
+ evidence := append([]ValidatedWriteSource(nil), validated...)
+ observation := admitted
+ observation.Claim, err = cloneClaim(admitted.Claim)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := pub
+ var store release.Store = s
+ switch name {
+ case "primary txid":
+ candidate.RemoteMaxTXID++
+ case "primary txid and baseline":
+ candidate.RemoteMaxTXID++
+ evidence[0].RemoteMaxTXID++
+ case "region":
+ candidate.Region = "uk"
+ case "predecessor region":
+ candidate.PredecessorRegion = "uk"
+ case "observed session":
+ candidate.ObservedSession++
+ case "observed generation":
+ candidate.ObservedGeneration = "other"
+ case "observed phase":
+ candidate.ObservedPhase = "other"
+ case "observed operation":
+ candidate.ObservedOpID = "other"
+ case "certified":
+ candidate.PredecessorCertified = true
+ case "coverage":
+ candidate.Coverage = "sealed"
+ case "additional loss":
+ candidate.AdditionalWritesLost[0].RemoteMaxTXID++
+ case "missing baseline":
+ evidence = nil
+ case "changed baseline":
+ evidence[0].Baseline = "other"
+ case "operation":
+ observation.Claim.Op.ID = "later-operation"
+ candidate.ObservedOpID = "later-operation"
+ case "phase":
+ observation.Claim.Op.Phase = "other"
+ candidate.ObservedPhase = "other"
+ case "executor":
+ observation.Claim.Op.Exec = "other"
+ case "incarnation":
+ observation.Claim.Holder.Incarnation = strings.Repeat("b", 64)
+ case "signature":
+ key = make(ed25519.PublicKey, ed25519.PublicKeySize)
+ case "missing declaration":
+ store = release.NewDirStore(t.TempDir())
+ }
+ calls := 0
+ owner := lossRecoveryOwnerFunc(func(c Claim, loss LossRecord, a Admission) error {
+ calls++
+ if name == "owner denial" {
+ return denied
+ }
+ return (testLossRecoveryOwner{}).ValidateLossRecovery(c, loss, a)
+ })
+ retry, err := AdmitWriterWithLoss(ctx, store, key, observation, candidate, record.ID, evidence, owner)
+ if name == "identical" {
+ if err != nil || retry.ETag != admitted.ETag || !reflect.DeepEqual(retry.Claim, admitted.Claim) || calls != 1 {
+ t.Errorf("identical retry: err %v, owner calls %d, observation %+v", err, calls, retry)
+ }
+ } else if err == nil {
+ t.Errorf("invalid retry accepted")
+ } else if name == "owner denial" && !errors.Is(err, denied) {
+ t.Errorf("owner denial = %v", err)
+ }
+ fresh, err := Read(ctx, s, admitted.Host)
+ if err != nil || fresh.ETag != admitted.ETag {
+ t.Fatalf("retry changed claim: %v", err)
+ }
+ })
+ }
+ })
}
func TestTargetOnlyLossAdmissionRequiresColdUncoveredRecovery(t *testing.T) {
diff --git a/internal/hostclaim/lost.go b/internal/hostclaim/lost.go
index 6b361a02a..ab8ec014e 100644
--- a/internal/hostclaim/lost.go
+++ b/internal/hostclaim/lost.go
@@ -1,11 +1,13 @@
package hostclaim
import (
+ "bytes"
"context"
"crypto/ed25519"
"encoding/json"
"errors"
"fmt"
+ "io"
"reflect"
"time"
@@ -110,14 +112,6 @@ func AdmitWriterWithLoss(ctx context.Context, s release.Store, pub ed25519.Publi
if observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || !lossRefPresent(observed.Claim, lossID) {
return Observed{}, ErrLossRecord
}
- for _, prior := range observed.Claim.Holder.Admissions {
- if prior.LossID == lossID {
- if prior.Box == in.Box && prior.Mode == in.Mode && prior.Generation == in.NewGeneration && reflect.DeepEqual(prior.AdditionalWritesLost, in.AdditionalWritesLost) {
- return observed, nil
- }
- return Observed{}, fmt.Errorf("%w: declaration id already used by another admission", ErrWrongState)
- }
- }
loss, err := ReadLossRecord(ctx, s, pub, lossID)
if err != nil {
return Observed{}, err
@@ -134,6 +128,20 @@ func AdmitWriterWithLoss(ctx context.Context, s release.Store, pub ed25519.Publi
if len(h.Admissions) > 0 {
predecessor = h.Admissions[len(h.Admissions)-1]
}
+ for _, prior := range h.Admissions {
+ if prior.LossID != lossID {
+ continue
+ }
+ if prior.Box != in.Box || prior.Mode != in.Mode || prior.Generation != in.NewGeneration ||
+ prior.Region != in.Region || prior.PredecessorRegion != in.PredecessorRegion || prior.Coverage != in.Coverage ||
+ prior.Predecessor != in.ObservedGeneration || prior.WritesLost == nil ||
+ prior.WritesLost.Generation != in.ObservedGeneration || prior.WritesLost.Session != in.ObservedSession ||
+ prior.WritesLost.RemoteMaxTXID != in.RemoteMaxTXID || !reflect.DeepEqual(prior.AdditionalWritesLost, in.AdditionalWritesLost) {
+ return Observed{}, fmt.Errorf("%w: declaration id already used by another admission", ErrWrongState)
+ }
+ predecessor = AdmissionRecord{Box: prior.WritesLost.Box, Generation: prior.WritesLost.Generation, Session: prior.WritesLost.Session, Region: prior.PredecessorRegion}
+ break
+ }
want := map[string]WriteSource{}
for _, source := range loss.Record.WriteSources {
key := fmt.Sprintf("%s\x00%s\x00%d", source.Box, source.Generation, source.Session)
@@ -199,7 +207,11 @@ func AdmitWriterWithLoss(ctx context.Context, s release.Store, pub ed25519.Publi
return Observed{}, err
}
in.LossID, in.lossVerified = lossID, true
- return Update(ctx, s, observed, AdmitWriter(in))
+ next, err := Update(ctx, s, observed, AdmitWriter(in))
+ if errors.Is(err, ErrLossAlreadyApplied) {
+ return observed, nil
+ }
+ return next, err
}
func declareLostTransition(in LostDeclaration, owner LossOperationOwner) func(*Claim) error {
@@ -368,9 +380,14 @@ func cloneClaim(c *Claim) (*Claim, error) {
func canonicalRaw(raw json.RawMessage) (string, error) {
var value any
- if err := json.Unmarshal(raw, &value); err != nil {
+ decoder := json.NewDecoder(bytes.NewReader(raw))
+ decoder.UseNumber()
+ if err := decoder.Decode(&value); err != nil {
return "", err
}
+ if err := decoder.Decode(new(any)); err != io.EOF {
+ return "", ErrLossRecord
+ }
canonical, err := json.Marshal(value)
return string(canonical), err
}
diff --git a/internal/hostclaim/lost_test.go b/internal/hostclaim/lost_test.go
index ad3a148b3..b972411d9 100644
--- a/internal/hostclaim/lost_test.go
+++ b/internal/hostclaim/lost_test.go
@@ -175,6 +175,40 @@ func lossForObservation(t *testing.T, obs Observed, id string, revision uint64,
return r
}
+func TestDeclareLostPreservesLargeRecoveryIntentIntegers(t *testing.T) {
+ ctx := t.Context()
+ s := release.NewDirStore(t.TempDir())
+ pub, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ obs := servingClaim(t, ctx, s)
+ obs, err = Update(ctx, s, obs, func(c *Claim) error {
+ c.Op = Op{Kind: KindDeleting, ID: "delete", Phase: "parking", Exec: "box-a", Participants: []string{"box-a"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ intent := `{"kind":"recover","nested":{"txid":9007199254740993},"positions":[18446744073709551615,-9007199254740993]}`
+ record := lossForObservation(t, obs, "large_integers", 1, "box-a", LossConsequences{
+ RevokeExecution: true, RemoveParticipation: true, RecoveryIntent: json.RawMessage(intent),
+ })
+ if _, err := CreateLossRecord(ctx, s, release.RootSigner(priv), record); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := DeclareLost(ctx, s, pub, obs, record.ID, testLossOwner{}); err != nil {
+ t.Fatal(err)
+ }
+ fresh, err := Read(ctx, s, obs.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := fresh.Claim.Op.Journal["loss."+record.ID]; got != intent {
+ t.Fatalf("persisted recovery intent = %s, want %s", got, intent)
+ }
+}
+
func TestDeclareLostNewerRevisionAppliesSecondConsequences(t *testing.T) {
ctx := t.Context()
s := release.NewDirStore(t.TempDir())