amadan

carlos / platform Public

hostclaim: spec rev 6.10 admission (refuse creating, KeepStamp, sealed baseline, QueueGenerationRetire, @root)

Co-Authored-By: Codex (gpt-6-astra) <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 090fb7a2d7ac5116f50ca5987666b526f0646ffc parent fdff530
2 files changed, +273 −6
  • internal/hostclaim/generation.go +36 −3
  • internal/hostclaim/generation_test.go +237 −3

CI passed — run details

diff --git a/internal/hostclaim/generation.go b/internal/hostclaim/generation.go
index e61f6ad8e..584843100 100644
--- a/internal/hostclaim/generation.go
+++ b/internal/hostclaim/generation.go
@@ -5,6 +5,8 @@ import (
"fmt"
)
+const RootGeneration = "@root"
+
type Admission struct {
ObservedOpID, ObservedPhase string // Empty values are compared literally.
ObservedSession uint64
@@ -14,6 +16,8 @@ type Admission struct {
NewGeneration string
PredecessorCertified bool
RemoteMaxTXID uint64
+ KeepStamp bool
+ BaselineCoversPredecessor bool
}
type writerPredecessor struct {
@@ -31,6 +35,7 @@ type writerAdmission struct {
Mode string `json:"mode"`
Predecessor writerPredecessor `json:"predecessor"`
WritesLost *writerWritesLost `json:"writes_lost,omitempty"`
+ Baseline string `json:"baseline,omitempty"`
}
type writerWritesLost struct {
@@ -54,13 +59,16 @@ func AdmitWriter(in Admission) func(*Claim) error {
return fmt.Errorf("%w: operation or phase moved since it was observed", ErrWrongState)
}
switch c.Op.Kind {
- case KindDeleting, KindParking, KindPurging, KindPlacing:
+ case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing:
return fmt.Errorf("%w: op kind %q cannot admit a writer", ErrWrongState, c.Op.Kind)
}
h := c.Holder
if h.Session != in.ObservedSession || h.Generation != in.ObservedGeneration {
return fmt.Errorf("%w: writer session or generation moved since it was observed", ErrWrongState)
}
+ if in.KeepStamp && (in.Mode != "cold" || h.Stamp <= h.Session) {
+ return fmt.Errorf("%w: keeping a stamp requires cold admission before its first writer", ErrWrongState)
+ }
switch in.Mode {
case "warm":
if !in.PredecessorCertified || in.NewGeneration != "" {
@@ -90,9 +98,15 @@ func AdmitWriter(in Admission) func(*Claim) error {
entry := writerAdmission{Session: c.Gen + 1, Box: in.Box, Generation: h.Generation, Stamp: h.Stamp, Mode: in.Mode, Predecessor: predecessor}
if in.Mode == "cold" {
entry.Generation = in.NewGeneration
- entry.Stamp = entry.Session
+ if !in.KeepStamp {
+ entry.Stamp = entry.Session
+ }
if !in.PredecessorCertified {
- entry.WritesLost = &writerWritesLost{Box: h.WriterBox, Generation: h.Generation, Session: h.Session, RemoteMaxTXID: in.RemoteMaxTXID}
+ if in.BaselineCoversPredecessor {
+ entry.Baseline = "sealed"
+ } else {
+ entry.WritesLost = &writerWritesLost{Box: h.WriterBox, Generation: h.Generation, Session: h.Session, RemoteMaxTXID: in.RemoteMaxTXID}
+ }
}
}
raw, err := json.Marshal(entry)
@@ -112,3 +126,22 @@ func AdmitWriter(in Admission) func(*Claim) error {
return nil
}
}
+
+func QueueGenerationRetire(observedSession uint64, publishedGeneration, region, retireGeneration string, newID func() string) func(*Claim) error {
+ return func(c *Claim) error {
+ if c.Holder == nil || c.Op.Kind != KindServing || c.Holder.Session != observedSession ||
+ publishedGeneration != c.Holder.Generation || retireGeneration == c.Holder.Generation || region == "" {
+ return fmt.Errorf("%w: generation retirement requires the current published serving session", ErrWrongState)
+ }
+ for _, ob := range c.Obligations {
+ if ob.Kind == "park-generation" && ob.Region == region && ob.Generation == retireGeneration {
+ return fmt.Errorf("%w: generation retirement already queued", ErrWrongState)
+ }
+ }
+ c.Obligations = append(c.Obligations, Obligation{
+ ID: newID(), Op: fmt.Sprintf("session:%d", c.Holder.Session), Kind: "park-generation",
+ Region: region, Generation: retireGeneration, Retirement: newID(), Owner: "region:" + region,
+ })
+ return nil
+ }
+}
diff --git a/internal/hostclaim/generation_test.go b/internal/hostclaim/generation_test.go
index a3d6e1c38..c06250467 100644
--- a/internal/hostclaim/generation_test.go
+++ b/internal/hostclaim/generation_test.go
@@ -93,7 +93,7 @@ func assertWriterJournal(t *testing.T, c *Claim, want string) {
}
func TestAdmitWriterPreservesOperation(t *testing.T) {
- for _, kind := range []string{KindServing, KindCreating, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} {
+ for _, kind := range []string{KindServing, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} {
for _, mode := range []string{"warm", "cold"} {
t.Run(kind+"/"+mode, func(t *testing.T) {
ctx := context.Background()
@@ -137,7 +137,7 @@ func TestAdmitWriterPreservesOperation(t *testing.T) {
}
func TestAdmitWriterRefusals(t *testing.T) {
- for _, name := range []string{"no holder", "no executor", "non-executor", "stale session", "stale generation", "warm with generation", "cold without generation", "cold same generation", "unknown mode", "empty mode", "malformed journal", "warm without certificate", "stale op", "stale phase", "deleting", "parking", "purging", "placing"} {
+ for _, name := range []string{"no holder", "no executor", "non-executor", "stale session", "stale generation", "warm with generation", "warm with current generation", "cold without generation", "cold same generation", "unknown mode", "empty mode", "malformed journal", "warm without certificate", "stale op", "stale phase", "creating", "deleting", "parking", "purging", "placing"} {
t.Run(name, func(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
@@ -151,7 +151,7 @@ func TestAdmitWriterRefusals(t *testing.T) {
c.Op = Op{}
case "no executor":
c.Op.Exec = ""
- case "deleting", "parking", "purging", "placing":
+ case "creating", "deleting", "parking", "purging", "placing":
c.Op.Kind, c.Op.ID = name, "op1"
case "malformed journal":
c.Op.Journal = map[string]string{"writers": "not-json"}
@@ -173,6 +173,8 @@ func TestAdmitWriterRefusals(t *testing.T) {
generation = "old"
case "warm with generation":
mode = "warm"
+ case "warm with current generation":
+ mode, newGeneration = "warm", generation
case "cold without generation":
newGeneration = ""
case "cold same generation":
@@ -346,3 +348,235 @@ func assertAdmissionStored(t *testing.T, ctx context.Context, s release.Store, w
}
return got
}
+
+func TestAdmitWriterKeepStamp(t *testing.T) {
+ for _, name := range []string{"landing", "equal session", "later session", "warm"} {
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o, err := Update(ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a", PredecessorCertified: true}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if name == "landing" || name == "warm" {
+ o, err = Update(ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: KindLanding, ID: "m1", Phase: "probe", Exec: "boxT"}
+ c.Holder.Stamp = c.Gen + 1
+ return nil
+ })
+ } else if name == "later session" {
+ o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: "gen_a", Box: "adi-knitting", Mode: "warm", PredecessorCertified: true}))
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ before := *o.Claim.Holder
+ in := Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: before.Session, ObservedGeneration: before.Generation, Box: o.Claim.Op.Exec, Mode: "cold", NewGeneration: "gen_b", KeepStamp: true, PredecessorCertified: true}
+ if name == "warm" {
+ in.Mode, in.NewGeneration = "warm", ""
+ }
+ next, err := Update(ctx, s, o, AdmitWriter(in))
+ if name != "landing" {
+ if !errors.Is(err, ErrWrongState) {
+ t.Fatalf("KeepStamp admission = %v, want ErrWrongState", err)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if next.Claim.Holder.Stamp != before.Stamp || next.Claim.Holder.Session != next.Claim.Gen || next.Claim.Holder.Generation != "gen_b" {
+ t.Fatalf("landing admission holder = %+v, prior stamp = %d", next.Claim.Holder, before.Stamp)
+ }
+ want := fmt.Sprintf(`[{"session":%d,"box":"boxT","generation":"gen_b","stamp":%d,"mode":"cold","predecessor":{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d}}]`, next.Claim.Gen, before.Stamp, before.Session, before.Stamp)
+ assertWriterJournal(t, next.Claim, want)
+ assertAdmissionStored(t, ctx, s, next)
+ in.ObservedSession, in.ObservedGeneration, in.NewGeneration = next.Claim.Holder.Session, "gen_b", "gen_c"
+ if _, err := Update(ctx, s, next, AdmitWriter(in)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("restarted probe keeping stamp = %v, want ErrWrongState", err)
+ }
+ assertAdmissionStored(t, ctx, s, next)
+ })
+ }
+}
+
+func TestAdmitWriterBaselineCoversPredecessor(t *testing.T) {
+ for _, mode := range []string{"cold", "warm"} {
+ for _, certified := range []bool{false, true} {
+ for _, covered := range []bool{false, true} {
+ if mode == "warm" && !certified {
+ continue
+ }
+ t.Run(fmt.Sprintf("%s/certified=%t/covered=%t", mode, certified, covered), func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o, err := Update(ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a", PredecessorCertified: true}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ before := *o.Claim.Holder
+ var oldEntries []json.RawMessage
+ if err := json.Unmarshal([]byte(o.Claim.Op.Journal["writers"]), &oldEntries); err != nil {
+ t.Fatal(err)
+ }
+ in := Admission{ObservedSession: before.Session, ObservedGeneration: before.Generation, Box: "adi-knitting", Mode: mode, PredecessorCertified: certified, BaselineCoversPredecessor: covered, RemoteMaxTXID: 42}
+ generation, stamp := before.Generation, before.Stamp
+ if mode == "cold" {
+ in.NewGeneration, generation, stamp = "gen_b", "gen_b", o.Claim.Gen+1
+ }
+ o, err = Update(ctx, s, o, AdmitWriter(in))
+ if err != nil {
+ t.Fatal(err)
+ }
+ extra := ""
+ if mode == "cold" && !certified {
+ if covered {
+ extra = `,"baseline":"sealed"`
+ } else {
+ extra = fmt.Sprintf(`,"writes_lost":{"box":"adi-knitting","generation":"gen_a","session":%d,"remote_max_txid":42}`, before.Session)
+ }
+ }
+ last := fmt.Sprintf(`{"session":%d,"box":"adi-knitting","generation":%q,"stamp":%d,"mode":%q,"predecessor":{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d}%s}`, o.Claim.Gen, generation, stamp, mode, before.Session, before.Stamp, extra)
+ assertWriterJournal(t, o.Claim, "["+string(oldEntries[0])+","+last+"]")
+ assertAdmissionStored(t, ctx, s, o)
+ })
+ }
+ }
+ }
+}
+
+func TestQueueGenerationRetire(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o, err := Update(ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a", PredecessorCertified: true}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ session := o.Claim.Holder.Session
+ calls := 0
+ newID := func() string { calls++; return fmt.Sprintf("id%d", calls) }
+ for _, target := range []struct{ region, generation string }{{"ie", "old"}, {"uk", "old"}, {"ie", "older"}} {
+ before := o
+ o, err = Update(ctx, s, o, QueueGenerationRetire(session, "gen_a", target.region, target.generation, newID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := *before.Claim
+ want.Gen++
+ want.Obligations = append(append([]Obligation(nil), want.Obligations...), Obligation{
+ ID: fmt.Sprintf("id%d", calls-1), Op: fmt.Sprintf("session:%d", session), Kind: "park-generation",
+ Region: target.region, Generation: target.generation, Retirement: fmt.Sprintf("id%d", calls), Owner: "region:" + target.region,
+ })
+ if !reflect.DeepEqual(o.Claim, &want) {
+ t.Fatalf("queued claim = %+v, want %+v", o.Claim, &want)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+ }
+ if calls != 6 {
+ t.Fatalf("newID calls = %d, want 6", calls)
+ }
+ if _, err := Update(ctx, s, o, QueueGenerationRetire(session, "gen_a", "ie", "old", newID)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("duplicate retirement = %v, want ErrWrongState", err)
+ }
+ if calls != 6 {
+ t.Fatalf("duplicate consumed IDs: %d", calls)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+}
+
+func TestQueueGenerationRetireRefusals(t *testing.T) {
+ for _, name := range []string{"no holder", "creating", "deleting", "parking", "purging", "placing", "moving", "landing", "reverting", "restoring", "pushing", "stale session", "unpublished", "current generation", "empty region"} {
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o, err := Update(ctx, s, o, func(c *Claim) error {
+ c.Holder.Session, c.Holder.Generation = c.Gen+1, "gen_a"
+ if name == "no holder" {
+ c.Holder, c.Op = nil, Op{}
+ } else if knownKinds[name] {
+ c.Op.Kind, c.Op.ID = name, "op1"
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ session, published, region, retire := o.Claim.Gen, "gen_a", "ie", "old"
+ switch name {
+ case "stale session":
+ session--
+ case "unpublished":
+ published = "old"
+ case "current generation":
+ retire = "gen_a"
+ case "empty region":
+ region = ""
+ }
+ newID := func() string { t.Fatal("refusal allocated an ID"); return "" }
+ if _, err := Update(ctx, s, o, QueueGenerationRetire(session, published, region, retire, newID)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("retirement = %v, want ErrWrongState", err)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+ })
+ }
+}
+
+func TestAdoptAdmitRetireRootGeneration(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o, err := Read(ctx, s, testHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = Update(ctx, s, o, Adopt(newHolder(), "adi-knitting", nil, nil, "adopt1", counter()))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o.Claim.Holder.Generation != "" || RootGeneration != "@root" {
+ t.Fatalf("legacy generation = %q, root sentinel = %q", o.Claim.Holder.Generation, RootGeneration)
+ }
+ before := *o.Claim.Holder
+ o = assertAdmissionStored(t, ctx, s, o)
+ o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedSession: before.Session, ObservedGeneration: "", Box: "adi-knitting", Mode: "cold", NewGeneration: "G1", PredecessorCertified: true}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := fmt.Sprintf(`[{"session":%d,"box":"adi-knitting","generation":"G1","stamp":%d,"mode":"cold","predecessor":{"session":%d,"box":"","generation":"","stamp":%d}}]`, o.Claim.Gen, o.Claim.Gen, before.Session, before.Stamp)
+ assertWriterJournal(t, o.Claim, want)
+ if _, err := Update(ctx, s, o, QueueGenerationRetire(o.Claim.Holder.Session, "", "ie", RootGeneration, counter())); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("retiring unfinished admission baseline = %v, want ErrWrongState", err)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+ o, err = Update(ctx, s, o, QueueGenerationRetire(o.Claim.Holder.Session, "G1", "ie", RootGeneration, counter()))
+ if err != nil {
+ t.Fatal(err)
+ }
+ o = assertAdmissionStored(t, ctx, s, o)
+ if len(o.Claim.Obligations) != 1 {
+ t.Fatalf("obligations = %+v", o.Claim.Obligations)
+ }
+ ob := o.Claim.Obligations[0]
+ if ob.Generation != "@root" || ob.Kind != "park-generation" || ob.Region != "ie" || ob.Owner != "region:ie" || ob.Op != fmt.Sprintf("session:%d", o.Claim.Holder.Session) {
+ t.Fatalf("root obligation = %+v", ob)
+ }
+ holder := *o.Claim.Holder
+ o, err = Update(ctx, s, o, CompleteObligation(ob.Op, ob.ID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(o.Claim.Obligations) != 0 || !reflect.DeepEqual(*o.Claim.Holder, holder) || o.Claim.Op.Kind != KindServing {
+ t.Fatalf("completed root retirement = %+v", o.Claim)
+ }
+ stored, err := Read(ctx, s, testHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if stored.ETag != o.ETag || len(stored.Claim.Obligations) != 0 || !reflect.DeepEqual(*stored.Claim.Holder, holder) || stored.Claim.Op.Kind != KindServing {
+ t.Fatalf("stored root retirement = %+v", stored)
+ }
+}