amadan

carlos / platform Public

hostclaim: holder.admissions, Coverage, PublishGeneration, Adopt writer_box (spec rev 6.12)

Co-Authored-By: Codex (gpt-6-astra) <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 34224a29a2665dab32b07d206210d9e8647b35ea parent 090fb7a
4 files changed, +413 −225
  • internal/hostclaim/adopt.go +1 −1
  • internal/hostclaim/claim.go +31 −12
  • internal/hostclaim/generation.go +72 −75
  • internal/hostclaim/generation_test.go +309 −137

CI passed — run details

diff --git a/internal/hostclaim/adopt.go b/internal/hostclaim/adopt.go
index 29c5cb8ad..d4492f7a6 100644
--- a/internal/hostclaim/adopt.go
+++ b/internal/hostclaim/adopt.go
@@ -17,7 +17,7 @@ func Adopt(nh NewHolder, servingBox string, otherRegions, otherBoxes []string, o
return err
}
c.Holder = &Holder{Family: nh.Family, Key: nh.Key, Account: nh.Account, App: nh.App, Name: nh.Name,
- Incarnation: inc, CreatedAt: nh.CreatedAt, Location: "instances/" + nh.Host, Stamp: c.Gen + 1}
+ Incarnation: inc, CreatedAt: nh.CreatedAt, Location: "instances/" + nh.Host, Stamp: c.Gen + 1, WriterBox: servingBox}
c.Op = Op{Kind: KindServing, Exec: servingBox, Participants: []string{servingBox}}
c.Touched = Touched{Complete: true, Regions: []string{nh.Region}, Boxes: []string{servingBox}}
appendParkObligations(c, opID, otherRegions, otherBoxes, newID)
diff --git a/internal/hostclaim/claim.go b/internal/hostclaim/claim.go
index 2df8c344f..b20bd783f 100644
--- a/internal/hostclaim/claim.go
+++ b/internal/hostclaim/claim.go
@@ -44,18 +44,37 @@ var (
)
type Holder struct {
- Family string `json:"family"` // "instance" | "sidecar"
- Key string `json:"key"` // exact record key
- Account string `json:"account"`
- App string `json:"app"`
- Name string `json:"name,omitempty"` // sidecars only
- Incarnation string `json:"incarnation"`
- CreatedAt time.Time `json:"created_at"`
- Location string `json:"location"`
- Generation string `json:"generation,omitempty"`
- Stamp uint64 `json:"stamp,omitempty"`
- Session uint64 `json:"session,omitempty"`
- WriterBox string `json:"writer_box,omitempty"`
+ Family string `json:"family"` // "instance" | "sidecar"
+ Key string `json:"key"` // exact record key
+ Account string `json:"account"`
+ App string `json:"app"`
+ Name string `json:"name,omitempty"` // sidecars only
+ Incarnation string `json:"incarnation"`
+ CreatedAt time.Time `json:"created_at"`
+ Location string `json:"location"`
+ Generation string `json:"generation,omitempty"`
+ Stamp uint64 `json:"stamp,omitempty"`
+ Session uint64 `json:"session,omitempty"`
+ WriterBox string `json:"writer_box,omitempty"`
+ Admissions []AdmissionRecord `json:"admissions,omitempty"`
+}
+
+type AdmissionRecord struct {
+ Session uint64 `json:"session"`
+ Box string `json:"box"`
+ Generation string `json:"generation"`
+ Predecessor string `json:"predecessor,omitempty"`
+ Mode string `json:"mode"`
+ Coverage string `json:"coverage,omitempty"`
+ Covered bool `json:"covered"`
+ WritesLost *WritesLost `json:"writes_lost,omitempty"`
+}
+
+type WritesLost struct {
+ Box string `json:"box"`
+ Generation string `json:"generation"`
+ Session uint64 `json:"session"`
+ RemoteMaxTXID uint64 `json:"remote_max_txid"`
}
type Op struct {
diff --git a/internal/hostclaim/generation.go b/internal/hostclaim/generation.go
index 584843100..ac646e139 100644
--- a/internal/hostclaim/generation.go
+++ b/internal/hostclaim/generation.go
@@ -1,9 +1,6 @@
package hostclaim
-import (
- "encoding/json"
- "fmt"
-)
+import "fmt"
const RootGeneration = "@root"
@@ -17,39 +14,13 @@ type Admission struct {
PredecessorCertified bool
RemoteMaxTXID uint64
KeepStamp bool
- BaselineCoversPredecessor bool
-}
-
-type writerPredecessor struct {
- Session uint64 `json:"session"`
- Box string `json:"box"`
- Generation string `json:"generation"`
- Stamp uint64 `json:"stamp"`
-}
-
-type writerAdmission struct {
- Session uint64 `json:"session"`
- Box string `json:"box"`
- Generation string `json:"generation"`
- Stamp uint64 `json:"stamp"`
- Mode string `json:"mode"`
- Predecessor writerPredecessor `json:"predecessor"`
- WritesLost *writerWritesLost `json:"writes_lost,omitempty"`
- Baseline string `json:"baseline,omitempty"`
-}
-
-type writerWritesLost struct {
- Box string `json:"box"`
- Generation string `json:"generation"`
- Session uint64 `json:"session"`
- RemoteMaxTXID uint64 `json:"remote_max_txid"`
+ Coverage string
}
// AdmitWriter is §6.4a's writer admission CAS within the current operation.
// Warm admission keeps the installation; cold admission installs a fresh
// generation and stamp. The caller checks the evidence required for warm mode
-// and supplies PredecessorCertified. Op.Journal is map[string]string; its
-// "writers" value holds a JSON array string.
+// and supplies PredecessorCertified.
func AdmitWriter(in Admission) func(*Claim) error {
return func(c *Claim) error {
if c.Holder == nil || c.Op.Exec == "" || c.Op.Exec != in.Box {
@@ -82,66 +53,92 @@ func AdmitWriter(in Admission) func(*Claim) error {
return fmt.Errorf("%w: unknown writer admission mode %q", ErrWrongState, in.Mode)
}
- var writers []json.RawMessage
- if raw, ok := c.Op.Journal["writers"]; ok {
- if err := json.Unmarshal([]byte(raw), &writers); err != nil {
- return fmt.Errorf("%w: writer journal: %v", ErrWrongState, err)
- }
+ switch in.Coverage {
+ case "", "sealed", "clean-return":
+ default:
+ return ErrWrongState
}
- predecessor := writerPredecessor{Session: h.Session, Box: h.WriterBox, Generation: h.Generation, Stamp: h.Stamp}
- if len(writers) > 0 {
- var last writerPredecessor
- if err := json.Unmarshal(writers[len(writers)-1], &last); err != nil {
- return fmt.Errorf("%w: previous writer: %v", ErrWrongState, err)
- }
+ predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation}
+ if len(h.Admissions) > 0 {
+ predecessor = h.Admissions[len(h.Admissions)-1]
+ }
+ entry := AdmissionRecord{
+ Session: c.Gen + 1, Box: in.Box, Generation: h.Generation,
+ Predecessor: predecessor.Generation, Mode: in.Mode, Coverage: in.Coverage,
}
- 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
if !in.KeepStamp {
- entry.Stamp = entry.Session
+ h.Stamp = entry.Session
}
- if !in.PredecessorCertified {
- if in.BaselineCoversPredecessor {
- entry.Baseline = "sealed"
- } else {
- entry.WritesLost = &writerWritesLost{Box: h.WriterBox, Generation: h.Generation, Session: h.Session, RemoteMaxTXID: in.RemoteMaxTXID}
- }
+ if !in.PredecessorCertified && in.Coverage == "" {
+ entry.WritesLost = &WritesLost{Box: predecessor.Box, Generation: predecessor.Generation, Session: predecessor.Session, RemoteMaxTXID: in.RemoteMaxTXID}
}
}
- raw, err := json.Marshal(entry)
- if err != nil {
- return err
+ h.Admissions = append(h.Admissions, entry)
+ for len(h.Admissions) > 8 && h.Admissions[0].Covered {
+ h.Admissions = h.Admissions[1:]
}
- journal, err := json.Marshal(append(writers, raw))
- if err != nil {
- return err
- }
- if c.Op.Journal == nil {
- c.Op.Journal = make(map[string]string)
- }
- c.Op.Journal["writers"] = string(journal)
- h.Session, h.Generation, h.Stamp = entry.Session, entry.Generation, entry.Stamp
- h.WriterBox = in.Box
+ h.Session, h.Generation, h.WriterBox = entry.Session, entry.Generation, in.Box
return nil
}
}
-func QueueGenerationRetire(observedSession uint64, publishedGeneration, region, retireGeneration string, newID func() string) func(*Claim) error {
+func PublishGeneration(observedOpID, observedPhase string, observedSession uint64, publishedGeneration, region 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)
+ if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase ||
+ c.Holder.Session != observedSession || c.Holder.Generation != publishedGeneration || region == "" {
+ return 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)
+ switch c.Op.Kind {
+ case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing:
+ return ErrWrongState
+ }
+ h := c.Holder
+ last := -1
+ for i, record := range h.Admissions {
+ if record.Generation == publishedGeneration {
+ last = i
+ }
+ }
+ if last == -1 {
+ return ErrWrongState
+ }
+ activeGeneration := publishedGeneration
+ if activeGeneration == "" {
+ activeGeneration = RootGeneration
+ }
+ op := c.Op.ID
+ if op == "" {
+ op = fmt.Sprintf("session:%d", h.Session)
+ }
+ for i := 0; i <= last; i++ {
+ record := &h.Admissions[i]
+ if record.Covered {
+ continue
+ }
+ record.Covered = true
+ predecessor := record.Predecessor
+ if predecessor == "" {
+ predecessor = RootGeneration
+ }
+ if predecessor == activeGeneration {
+ continue
+ }
+ queued := false
+ for _, ob := range c.Obligations {
+ if ob.Kind == "park-generation" && ob.Region == region && ob.Generation == predecessor {
+ queued = true
+ break
+ }
+ }
+ if !queued {
+ c.Obligations = append(c.Obligations, Obligation{
+ ID: newID(), Op: op, Kind: "park-generation", Region: region,
+ Generation: predecessor, Retirement: newID(), Owner: "region:" + region,
+ })
}
}
- 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 c06250467..923e302bc 100644
--- a/internal/hostclaim/generation_test.go
+++ b/internal/hostclaim/generation_test.go
@@ -12,7 +12,7 @@ import (
"github.com/carlosframework/platform/internal/release"
)
-func TestAdmitWriterWarmColdJournal(t *testing.T) {
+func TestAdmitWriterWarmColdAdmissions(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
o := servingClaim(t, ctx, s)
@@ -28,7 +28,7 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) {
if err := json.Unmarshal(raw, &fields); err != nil {
t.Fatal(err)
}
- for _, name := range []string{"session", "writer_box"} {
+ for _, name := range []string{"session", "writer_box", "admissions"} {
if _, ok := fields[name]; ok {
t.Fatalf("zero %s must be omitted", name)
}
@@ -41,8 +41,8 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) {
if o.Claim.Holder.WriterBox != "adi-knitting" || o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold || o.Claim.Holder.Session != cold || cold <= stamp {
t.Fatalf("cold holder = %+v, gen = %d", o.Claim.Holder, cold)
}
- first := fmt.Sprintf(`{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d,"mode":"cold","predecessor":{"session":0,"box":"","generation":"","stamp":%d}}`, cold, cold, stamp)
- assertWriterJournal(t, o.Claim, "["+first+"]")
+ first := AdmissionRecord{Session: cold, Box: "adi-knitting", Generation: "gen_a", Mode: "cold"}
+ assertAdmissions(t, o.Claim, []AdmissionRecord{first})
o, err = Update(ctx, s, o, func(c *Claim) error {
c.Op.Exec = "boxT"
return nil
@@ -58,8 +58,8 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) {
if o.Claim.Holder.WriterBox != "boxT" || o.Claim.Holder.Session != warm || warm <= cold || o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold {
t.Fatalf("warm holder = %+v, gen = %d", o.Claim.Holder, warm)
}
- second := fmt.Sprintf(`{"session":%d,"box":"boxT","generation":"gen_a","stamp":%d,"mode":"warm","predecessor":{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d}}`, warm, cold, cold, cold)
- assertWriterJournal(t, o.Claim, "["+first+","+second+"]")
+ second := AdmissionRecord{Session: warm, Box: "boxT", Generation: "gen_a", Mode: "warm", Predecessor: "gen_a"}
+ assertAdmissions(t, o.Claim, []AdmissionRecord{first, second})
o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedSession: warm, ObservedGeneration: "gen_a", Box: "boxT", Mode: "cold", NewGeneration: "gen_b", PredecessorCertified: true}))
if err != nil {
t.Fatal(err)
@@ -67,8 +67,8 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) {
if o.Claim.Holder.Generation != "gen_b" || o.Claim.Holder.Stamp != o.Claim.Gen || o.Claim.Holder.Session != o.Claim.Gen {
t.Fatalf("second cold holder = %+v, gen = %d", o.Claim.Holder, o.Claim.Gen)
}
- third := fmt.Sprintf(`{"session":%d,"box":"boxT","generation":"gen_b","stamp":%d,"mode":"cold","predecessor":{"session":%d,"box":"boxT","generation":"gen_a","stamp":%d}}`, o.Claim.Gen, o.Claim.Gen, warm, cold)
- assertWriterJournal(t, o.Claim, "["+first+","+second+","+third+"]")
+ third := AdmissionRecord{Session: o.Claim.Gen, Box: "boxT", Generation: "gen_b", Mode: "cold", Predecessor: "gen_a"}
+ assertAdmissions(t, o.Claim, []AdmissionRecord{first, second, third})
stored, err := Read(ctx, s, testHost)
if err != nil {
t.Fatal(err)
@@ -78,17 +78,13 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) {
}
}
-func assertWriterJournal(t *testing.T, c *Claim, want string) {
+func assertAdmissions(t *testing.T, c *Claim, want []AdmissionRecord) {
t.Helper()
- var gotJSON, wantJSON any
- if err := json.Unmarshal([]byte(c.Op.Journal["writers"]), &gotJSON); err != nil {
- t.Fatal(err)
- }
- if err := json.Unmarshal([]byte(want), &wantJSON); err != nil {
- t.Fatal(err)
+ if !reflect.DeepEqual(c.Holder.Admissions, want) {
+ t.Fatalf("admissions = %+v, want %+v", c.Holder.Admissions, want)
}
- if !reflect.DeepEqual(gotJSON, wantJSON) {
- t.Fatalf("writers = %s, want %s", c.Op.Journal["writers"], want)
+ if _, ok := c.Op.Journal["writers"]; ok {
+ t.Fatal("admission wrote an operation writers journal")
}
}
@@ -127,9 +123,9 @@ func TestAdmitWriterPreservesOperation(t *testing.T) {
want.Holder.Generation = generation
want.Holder.Stamp = want.Gen
}
- want.Op.Journal = map[string]string{"source": "unchanged", "writers": next.Claim.Op.Journal["writers"]}
- if want.Op.Journal["writers"] == "" || !reflect.DeepEqual(next.Claim, &want) {
- t.Fatalf("admission changed fields outside holder session/generation/stamp and writers journal: got %+v, want %+v", next.Claim, &want)
+ want.Holder.Admissions = []AdmissionRecord{{Session: want.Gen, Box: "boxT", Generation: generation, Mode: mode}}
+ if !reflect.DeepEqual(next.Claim, &want) {
+ t.Fatalf("admission changed fields outside holder writer fields and admissions: got %+v, want %+v", next.Claim, &want)
}
})
}
@@ -137,7 +133,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", "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"} {
+ 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", "unknown coverage", "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())
@@ -153,8 +149,6 @@ func TestAdmitWriterRefusals(t *testing.T) {
c.Op.Exec = ""
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"}
}
return nil
})
@@ -188,6 +182,8 @@ func TestAdmitWriterRefusals(t *testing.T) {
switch name {
case "warm without certificate":
in.Mode, in.NewGeneration, in.PredecessorCertified = "warm", "", false
+ case "unknown coverage":
+ in.Coverage = "remote"
case "stale op":
in.ObservedOpID = "old"
case "stale phase":
@@ -327,8 +323,11 @@ func TestAdmitWriterSurvivesOperationReplacement(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- 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},"writes_lost":{"box":"adi-knitting","generation":"gen_a","session":%d,"remote_max_txid":42}}]`, o.Claim.Gen, o.Claim.Gen, writer.Session, writer.Stamp, writer.Session)
- assertWriterJournal(t, o.Claim, want)
+ want := append(append([]AdmissionRecord(nil), writer.Admissions...), AdmissionRecord{
+ Session: o.Claim.Gen, Box: "boxT", Generation: "gen_b", Mode: "cold", Predecessor: "gen_a",
+ WritesLost: &WritesLost{Box: "adi-knitting", Generation: "gen_a", Session: writer.Session, RemoteMaxTXID: 42},
+ })
+ assertAdmissions(t, o.Claim, want)
if o.Claim.Holder.WriterBox != "boxT" {
t.Fatalf("new writer = %q", o.Claim.Holder.WriterBox)
}
@@ -390,8 +389,10 @@ func TestAdmitWriterKeepStamp(t *testing.T) {
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)
+ want := append(append([]AdmissionRecord(nil), before.Admissions...), AdmissionRecord{
+ Session: next.Claim.Gen, Box: "boxT", Generation: "gen_b", Mode: "cold", Predecessor: "gen_a",
+ })
+ assertAdmissions(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) {
@@ -402,45 +403,30 @@ func TestAdmitWriterKeepStamp(t *testing.T) {
}
}
-func TestAdmitWriterBaselineCoversPredecessor(t *testing.T) {
+func TestAdmitWriterCoverage(t *testing.T) {
for _, mode := range []string{"cold", "warm"} {
for _, certified := range []bool{false, true} {
- for _, covered := range []bool{false, true} {
+ for _, coverage := range []string{"", "sealed", "clean-return"} {
if mode == "warm" && !certified {
continue
}
- t.Run(fmt.Sprintf("%s/certified=%t/covered=%t", mode, certified, covered), func(t *testing.T) {
+ t.Run(fmt.Sprintf("%s/certified=%t/coverage=%s", mode, certified, coverage), 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)
- }
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a", PredecessorCertified: true}))
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
+ in := Admission{ObservedSession: before.Session, ObservedGeneration: before.Generation, Box: "adi-knitting", Mode: mode, PredecessorCertified: certified, Coverage: coverage, RemoteMaxTXID: 42}
+ generation := before.Generation
if mode == "cold" {
- in.NewGeneration, generation, stamp = "gen_b", "gen_b", o.Claim.Gen+1
+ in.NewGeneration, generation = "gen_b", "gen_b"
}
- o, err = Update(ctx, s, o, AdmitWriter(in))
- if err != nil {
- t.Fatal(err)
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(in))
+ last := AdmissionRecord{Session: o.Claim.Gen, Box: "adi-knitting", Generation: generation, Predecessor: "gen_a", Mode: mode, Coverage: coverage}
+ if mode == "cold" && !certified && coverage == "" {
+ last.WritesLost = &WritesLost{Box: "adi-knitting", Generation: "gen_a", Session: before.Session, RemoteMaxTXID: 42}
}
- 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+"]")
+ assertAdmissions(t, o.Claim, append(append([]AdmissionRecord(nil), before.Admissions...), last))
assertAdmissionStored(t, ctx, s, o)
})
}
@@ -448,135 +434,321 @@ func TestAdmitWriterBaselineCoversPredecessor(t *testing.T) {
}
}
-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}))
+func updateGenerationTest(t *testing.T, ctx context.Context, s release.Store, o Observed, tr func(*Claim) error) Observed {
+ t.Helper()
+ next, err := Update(ctx, s, o, tr)
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,
+ return assertAdmissionStored(t, ctx, s, next)
+}
+
+func TestAdmitWriterCleanReturnPredecessor(t *testing.T) {
+ for _, coverage := range []string{"", "clean-return"} {
+ t.Run(coverage, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: KindLanding, ID: "move", Phase: "probe", Exec: "boxT"}
+ return nil
+ })
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedOpID: "move", ObservedPhase: "probe", Box: "boxT", Mode: "cold", NewGeneration: "GT"}))
+ target := o.Claim.Holder.Admissions[0]
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Holder.Generation = "GS"
+ c.Op = Op{Kind: KindServing, Exec: "boxS"}
+ return nil
+ })
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedSession: target.Session, ObservedGeneration: "GS", Box: "boxS", Mode: "cold", NewGeneration: "GS2", Coverage: coverage, RemoteMaxTXID: 42}))
+ last := AdmissionRecord{Session: o.Claim.Gen, Box: "boxS", Generation: "GS2", Predecessor: "GT", Mode: "cold", Coverage: coverage}
+ if coverage == "" {
+ last.WritesLost = &WritesLost{Box: "boxT", Generation: "GT", Session: target.Session, RemoteMaxTXID: 42}
+ }
+ assertAdmissions(t, o.Claim, []AdmissionRecord{target, last})
})
- 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)
+}
+
+func TestAdoptAdmitPublishRootGeneration(t *testing.T) {
+ for _, replaceOp := range []bool{false, true} {
+ t.Run(fmt.Sprintf("replaceOp=%t", replaceOp), func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o, err := Read(ctx, s, testHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o = updateGenerationTest(t, ctx, s, o, Adopt(newHolder(), "boxS", nil, nil, "adopt", counter()))
+ if o.Claim.Holder.WriterBox != "boxS" || o.Claim.Holder.Generation != "" || RootGeneration != "@root" {
+ t.Fatalf("adopted holder = %+v", o.Claim.Holder)
+ }
+ for _, generation := range []string{"", RootGeneration} {
+ newID := func() string { t.Fatal("unadmitted publication allocated an ID"); return "" }
+ if _, err := Update(ctx, s, o, PublishGeneration("", "", 0, generation, "ie", newID)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("publication before admission = %v", err)
+ }
+ assertAdmissionStored(t, ctx, s, o)
+ }
+ if replaceOp {
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: KindLanding, ID: "move", Phase: "probe", Exec: "boxT"}
+ return nil
+ })
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: KindServing, Exec: "boxT"}
+ return nil
+ })
+ }
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: o.Claim.Op.Exec, Mode: "cold", NewGeneration: "G1", RemoteMaxTXID: 42}))
+ assertAdmissions(t, o.Claim, []AdmissionRecord{{
+ Session: o.Claim.Gen, Box: o.Claim.Op.Exec, Generation: "G1", Mode: "cold",
+ WritesLost: &WritesLost{Box: "boxS", Session: 0, Generation: "", RemoteMaxTXID: 42},
+ }})
+ o = updateGenerationTest(t, ctx, s, o, PublishGeneration("", "", o.Claim.Holder.Session, "G1", "ie", counter()))
+ if !o.Claim.Holder.Admissions[0].Covered || len(o.Claim.Obligations) != 1 {
+ t.Fatalf("published claim = %+v", o.Claim)
+ }
+ ob := o.Claim.Obligations[0]
+ if ob.Generation != RootGeneration || 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)
+ }
+ stored, err := Read(ctx, s, testHost)
+ if err != nil || stored.ETag != o.ETag || len(stored.Claim.Obligations) != 0 || !reflect.DeepEqual(*stored.Claim.Holder, holder) || stored.Claim.Op.Kind != KindServing {
+ t.Fatalf("completed root retirement = %+v, %v", stored, err)
+ }
+ })
}
- if calls != 6 {
- t.Fatalf("duplicate consumed IDs: %d", calls)
+}
+
+func TestPublishGenerationChain(t *testing.T) {
+ for _, kind := range []string{KindServing, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} {
+ t.Run(kind, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ if kind != KindServing {
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: kind, ID: "move", Phase: "probe", Exec: "boxT", Journal: map[string]string{"source": "unchanged"}}
+ return nil
+ })
+ }
+ for _, generation := range []string{"G1", "G2"} {
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase,
+ ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
+ Box: o.Claim.Op.Exec, Mode: "cold", NewGeneration: generation}))
+ }
+ before := o
+ calls := 0
+ newID := func() string { calls++; return fmt.Sprintf("id%d", calls) }
+ publish := PublishGeneration(o.Claim.Op.ID, o.Claim.Op.Phase, o.Claim.Holder.Session, "G2", "ie", newID)
+ o = updateGenerationTest(t, ctx, s, o, publish)
+ op := before.Claim.Op.ID
+ if op == "" {
+ op = fmt.Sprintf("session:%d", before.Claim.Holder.Session)
+ }
+ want := *before.Claim
+ holder := *want.Holder
+ want.Holder = &holder
+ want.Gen++
+ holder.Admissions = append([]AdmissionRecord(nil), holder.Admissions...)
+ for i := range holder.Admissions {
+ holder.Admissions[i].Covered = true
+ }
+ want.Obligations = []Obligation{
+ {ID: "id1", Op: op, Kind: "park-generation", Region: "ie", Generation: RootGeneration, Retirement: "id2", Owner: "region:ie"},
+ {ID: "id3", Op: op, Kind: "park-generation", Region: "ie", Generation: "G1", Retirement: "id4", Owner: "region:ie"},
+ }
+ if calls != 4 || !reflect.DeepEqual(o.Claim, &want) {
+ t.Fatalf("publication = %+v, want %+v; ID calls %d", o.Claim, &want, calls)
+ }
+ o = updateGenerationTest(t, ctx, s, o, publish)
+ want.Gen++
+ if calls != 4 || !reflect.DeepEqual(o.Claim, &want) {
+ t.Fatalf("republication changed coverage or chores: %+v; calls %d", o.Claim, 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"} {
+func TestPublishGenerationRefusals(t *testing.T) {
+ for _, name := range []string{"no holder", "creating", "deleting", "parking", "purging", "placing", "stale op", "stale phase", "stale session", "wrong generation", "empty region", "no matching admission"} {
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"
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "G1"}))
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op = Op{Kind: KindLanding, ID: "move", Phase: "probe", Exec: "boxT"}
if name == "no holder" {
c.Holder, c.Op = nil, Op{}
} else if knownKinds[name] {
- c.Op.Kind, c.Op.ID = name, "op1"
+ c.Op.Kind = name
+ } else if name == "no matching admission" {
+ c.Holder.Generation = "unrecorded"
}
return nil
})
- if err != nil {
- t.Fatal(err)
+ op, phase, session, generation, region := o.Claim.Op.ID, o.Claim.Op.Phase, uint64(0), "G1", "ie"
+ if o.Claim.Holder != nil {
+ session, generation = o.Claim.Holder.Session, o.Claim.Holder.Generation
}
- session, published, region, retire := o.Claim.Gen, "gen_a", "ie", "old"
switch name {
+ case "stale op":
+ op = ""
+ case "stale phase":
+ phase = ""
case "stale session":
session--
- case "unpublished":
- published = "old"
- case "current generation":
- retire = "gen_a"
+ case "wrong generation":
+ generation = "old"
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)
+ if _, err := Update(ctx, s, o, PublishGeneration(op, phase, session, generation, region, newID)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("publication = %v, want ErrWrongState", err)
}
assertAdmissionStored(t, ctx, s, o)
})
}
}
-func TestAdoptAdmitRetireRootGeneration(t *testing.T) {
+func TestPublishGenerationCoverageBoundary(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)
+ o := servingClaim(t, ctx, s)
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Holder.Generation = "G2"
+ c.Holder.Admissions = []AdmissionRecord{
+ {Generation: "old", Predecessor: "already-covered", Covered: true},
+ {Generation: "G1", Predecessor: ""},
+ {Generation: "G1", Predecessor: RootGeneration},
+ {Generation: "G2", Predecessor: "G1"},
+ {Generation: "G2", Predecessor: "G2", Mode: "warm"},
+ {Generation: "later", Predecessor: "G2"},
+ }
+ c.Obligations = []Obligation{
+ {ID: "existing", Op: "old-op", Kind: "park-generation", Region: "ie", Generation: "G1", Retirement: "ret1", Owner: "region:ie"},
+ {ID: "other-region", Op: "old-op", Kind: "park-generation", Region: "uk", Generation: RootGeneration, Retirement: "ret2", Owner: "region:uk"},
+ }
+ return nil
+ })
+ before := o
+ calls := 0
+ newID := func() string { calls++; return fmt.Sprintf("id%d", calls) }
+ o = updateGenerationTest(t, ctx, s, o, PublishGeneration("", "", 0, "G2", "ie", newID))
+ want := append(append([]Obligation(nil), before.Claim.Obligations...), Obligation{
+ ID: "id1", Op: "session:0", Kind: "park-generation", Region: "ie", Generation: RootGeneration, Retirement: "id2", Owner: "region:ie",
+ })
+ if calls != 2 || !reflect.DeepEqual(o.Claim.Obligations, want) {
+ t.Fatalf("deduplicated chores = %+v, want %+v; calls %d", o.Claim.Obligations, want, calls)
}
- if o.Claim.Holder.Generation != "" || RootGeneration != "@root" {
- t.Fatalf("legacy generation = %q, root sentinel = %q", o.Claim.Holder.Generation, RootGeneration)
+ for i, record := range o.Claim.Holder.Admissions {
+ if record.Covered != (i < 5) {
+ t.Fatalf("record %d covered = %t", i, record.Covered)
+ }
}
- 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)
+}
+
+func TestPublishGenerationActiveRoot(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "warm", PredecessorCertified: true}))
+ o = updateGenerationTest(t, ctx, s, o, PublishGeneration("", "", o.Claim.Holder.Session, "", "ie", func() string {
+ t.Fatal("active root must not be retired")
+ return ""
+ }))
+ if !o.Claim.Holder.Admissions[0].Covered || len(o.Claim.Obligations) != 0 {
+ t.Fatalf("active root publication = %+v", o.Claim)
}
- 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)
+}
+
+func TestAdmitWriterTrimsCoveredPrefix(t *testing.T) {
+ for _, name := range []string{"all covered", "uncovered first", "uncovered middle"} {
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ for i := 1; i <= 10; i++ {
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation,
+ Box: o.Claim.Op.Exec, Mode: "cold", NewGeneration: fmt.Sprintf("G%d", i)}))
+ }
+ if len(o.Claim.Holder.Admissions) != 10 {
+ t.Fatalf("uncovered records were trimmed: %+v", o.Claim.Holder.Admissions)
+ }
+ o = updateGenerationTest(t, ctx, s, o, PublishGeneration("", "", o.Claim.Holder.Session, "G10", "ie", counter()))
+ if name != "all covered" {
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ i := 0
+ if name == "uncovered middle" {
+ i = 1
+ }
+ c.Holder.Admissions[i].Covered = false
+ return nil
+ })
+ }
+ before := o.Claim.Holder.Admissions
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: "G10",
+ Box: o.Claim.Op.Exec, Mode: "cold", NewGeneration: "G11", PredecessorCertified: true}))
+ drop := 3
+ if name == "uncovered first" {
+ drop = 0
+ } else if name == "uncovered middle" {
+ drop = 1
+ }
+ want := append(append([]AdmissionRecord(nil), before[drop:]...), AdmissionRecord{Session: o.Claim.Gen, Box: o.Claim.Op.Exec, Generation: "G11", Predecessor: "G10", Mode: "cold"})
+ assertAdmissions(t, o.Claim, want)
+ })
}
- assertAdmissionStored(t, ctx, s, o)
- o, err = Update(ctx, s, o, QueueGenerationRetire(o.Claim.Holder.Session, "G1", "ie", RootGeneration, counter()))
+}
+
+func TestAdmissionRecordJSON(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "G1", RemoteMaxTXID: 42}))
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: "G1", Box: "adi-knitting", Mode: "cold", NewGeneration: "G2", Coverage: "sealed", PredecessorCertified: true}))
+ o = updateGenerationTest(t, ctx, s, o, PublishGeneration("", "", o.Claim.Holder.Session, "G2", "ie", counter()))
+ raw, err := json.Marshal(o.Claim.Holder)
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)
+ var holder map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &holder); err != nil {
+ t.Fatal(err)
}
- holder := *o.Claim.Holder
- o, err = Update(ctx, s, o, CompleteObligation(ob.Op, ob.ID))
- if err != nil {
+ var got []map[string]any
+ if err := json.Unmarshal(holder["admissions"], &got); 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)
+ want := []map[string]any{
+ {"session": float64(o.Claim.Holder.Admissions[0].Session), "box": "adi-knitting", "generation": "G1", "mode": "cold", "covered": true,
+ "writes_lost": map[string]any{"box": "", "generation": "", "session": float64(0), "remote_max_txid": float64(42)}},
+ {"session": float64(o.Claim.Holder.Session), "box": "adi-knitting", "generation": "G2", "predecessor": "G1", "mode": "cold", "coverage": "sealed", "covered": true},
}
- stored, err := Read(ctx, s, testHost)
- if err != nil {
- t.Fatal(err)
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("admissions JSON = %s, want %+v", holder["admissions"], want)
}
- 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)
+}
+
+func TestAdmitWriterIgnoresLegacyWriterJournal(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ o = updateGenerationTest(t, ctx, s, o, func(c *Claim) error {
+ c.Op.Journal = map[string]string{"writers": "not-json", "source": "unchanged"}
+ return nil
+ })
+ before := o.Claim.Op
+ o = updateGenerationTest(t, ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "G1"}))
+ if !reflect.DeepEqual(o.Claim.Op, before) || len(o.Claim.Holder.Admissions) != 1 {
+ t.Fatalf("legacy journal affected admission: %+v", o.Claim)
}
}