amadan

carlos / platform Public

hostclaim: AdmitWriter replaces NewGeneration (writer sessions, spec rev 6.8 §6.4a)

holder.session moves forward on every admission; warm keeps the generation,
cold installs a new generation and stamp; works inside any op and journals
the predecessor.
Co-Authored-By: Codex (gpt-6-astra) <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude a67d9f4744d7ec86c3468cf1f6064a902f5cb068 parent cabbd41
3 files changed, +247 −26
  • internal/hostclaim/claim.go +1 −0
  • internal/hostclaim/generation.go +73 −14
  • internal/hostclaim/generation_test.go +173 −12

CI passed — run details

diff --git a/internal/hostclaim/claim.go b/internal/hostclaim/claim.go
index d7429affe..9d867bca1 100644
--- a/internal/hostclaim/claim.go
+++ b/internal/hostclaim/claim.go
@@ -54,6 +54,7 @@ type Holder struct {
Location string `json:"location"`
Generation string `json:"generation,omitempty"`
Stamp uint64 `json:"stamp,omitempty"`
+ Session uint64 `json:"session,omitempty"`
}
type Op struct {
diff --git a/internal/hostclaim/generation.go b/internal/hostclaim/generation.go
index 0a82cd8c1..28c5fe4a3 100644
--- a/internal/hostclaim/generation.go
+++ b/internal/hostclaim/generation.go
@@ -1,24 +1,83 @@
package hostclaim
-import "fmt"
+import (
+ "encoding/json"
+ "fmt"
+)
-// NewGeneration is §6.4a's cold recovery: an installing CAS that gives a
-// writer whose predecessor may still have uploads in flight a fresh
-// generation, and with it a fresh stamp (§2, R6-4). It is never a
-// same-generation restart.
-func NewGeneration(observedGeneration, newGeneration, execBox string) func(*Claim) error {
+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"`
+}
+
+// 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.
+func AdmitWriter(observedSession uint64, observedGeneration, box, mode, newGeneration string) func(*Claim) error {
return func(c *Claim) error {
- if c.Holder == nil || c.Op.Kind != KindServing || c.Op.Exec == "" || c.Op.Exec != execBox {
- return fmt.Errorf("%w: only the serving executor may start a generation", ErrWrongState)
+ if c.Holder == nil || c.Op.Exec == "" || c.Op.Exec != box {
+ return fmt.Errorf("%w: only the executor may admit a writer", ErrWrongState)
+ }
+ h := c.Holder
+ if h.Session != observedSession || h.Generation != observedGeneration {
+ return fmt.Errorf("%w: writer session or generation moved since it was observed", ErrWrongState)
+ }
+ switch mode {
+ case "warm":
+ if newGeneration != "" {
+ return fmt.Errorf("%w: a warm admission keeps the generation", ErrWrongState)
+ }
+ case "cold":
+ if newGeneration == "" || newGeneration == h.Generation {
+ return fmt.Errorf("%w: a cold recovery needs a new generation", ErrWrongState)
+ }
+ default:
+ return fmt.Errorf("%w: unknown writer admission mode %q", ErrWrongState, 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)
+ }
+ }
+ predecessor := writerPredecessor{Session: h.Session, 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.Box = last.Box
+ }
+ entry := writerAdmission{Session: c.Gen + 1, Box: box, Generation: h.Generation, Stamp: h.Stamp, Mode: mode, Predecessor: predecessor}
+ if mode == "cold" {
+ entry.Generation = newGeneration
+ entry.Stamp = entry.Session
+ }
+ raw, err := json.Marshal(entry)
+ if err != nil {
+ return err
}
- if c.Holder.Generation != observedGeneration {
- return fmt.Errorf("%w: generation moved since it was observed", ErrWrongState)
+ journal, err := json.Marshal(append(writers, raw))
+ if err != nil {
+ return err
}
- if newGeneration == "" || newGeneration == c.Holder.Generation {
- return fmt.Errorf("%w: a cold recovery needs a new generation", ErrWrongState)
+ if c.Op.Journal == nil {
+ c.Op.Journal = make(map[string]string)
}
- c.Holder.Generation = newGeneration
- c.Holder.Stamp = c.Gen + 1 // the resulting gen
+ c.Op.Journal["writers"] = string(journal)
+ h.Session, h.Generation, h.Stamp = entry.Session, entry.Generation, entry.Stamp
return nil
}
}
diff --git a/internal/hostclaim/generation_test.go b/internal/hostclaim/generation_test.go
index 9b3a320b3..8c46b0fd6 100644
--- a/internal/hostclaim/generation_test.go
+++ b/internal/hostclaim/generation_test.go
@@ -2,29 +2,190 @@ package hostclaim
import (
"context"
+ "encoding/json"
"errors"
+ "fmt"
+ "reflect"
"testing"
"github.com/carlosframework/platform/internal/release"
)
-func TestNewGenerationInstallsOnce(t *testing.T) {
+func TestAdmitWriterWarmColdJournal(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
o := servingClaim(t, ctx, s)
- before := o.Claim.Holder.Stamp
- o2, err := Update(ctx, s, o, NewGeneration("", "gen_a", "adi-knitting"))
- if err != nil || o2.Claim.Holder.Generation != "gen_a" || o2.Claim.Holder.Stamp != o2.Claim.Gen || o2.Claim.Holder.Stamp <= before {
- t.Fatalf("new generation = %+v, %v", o2.Claim.Holder, err)
+ stamp := o.Claim.Holder.Stamp
+ if o.Claim.Holder.Session != 0 {
+ t.Fatalf("legacy session = %d, want zero", o.Claim.Holder.Session)
}
- // A second recovery that observed the old generation must not install.
- if _, err := Update(ctx, s, o2, NewGeneration("", "gen_b", "adi-knitting")); !errors.Is(err, ErrWrongState) {
- t.Fatalf("stale recovery = %v, want ErrWrongState", err)
+ raw, err := json.Marshal(o.Claim.Holder)
+ if err != nil {
+ t.Fatal(err)
}
- if _, err := Update(ctx, s, o2, NewGeneration("gen_a", "gen_c", "some-other-box")); !errors.Is(err, ErrWrongState) {
- t.Fatalf("recovery by a non-executor = %v, want ErrWrongState", err)
+ var fields map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &fields); err != nil {
+ t.Fatal(err)
}
- if _, err := Update(ctx, s, o2, NewGeneration("gen_a", "gen_a", "adi-knitting")); !errors.Is(err, ErrWrongState) {
- t.Fatalf("same-generation restart = %v, want ErrWrongState", err)
+ if _, ok := fields["session"]; ok {
+ t.Fatal("zero session must be omitted")
+ }
+ o, err = Update(ctx, s, o, AdmitWriter(0, "", "adi-knitting", "cold", "gen_a"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ cold := o.Claim.Gen
+ if 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+"]")
+ o, err = Update(ctx, s, o, func(c *Claim) error {
+ c.Op.Exec = "boxT"
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = Update(ctx, s, o, AdmitWriter(cold, "gen_a", "boxT", "warm", ""))
+ if err != nil {
+ t.Fatal(err)
+ }
+ warm := o.Claim.Gen
+ if 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+"]")
+ o, err = Update(ctx, s, o, AdmitWriter(warm, "gen_a", "boxT", "cold", "gen_b"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ 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+"]")
+ stored, err := Read(ctx, s, testHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(stored, o) {
+ t.Fatalf("stored admission = %+v, want %+v", stored, o)
+ }
+}
+
+func assertWriterJournal(t *testing.T, c *Claim, want string) {
+ 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(gotJSON, wantJSON) {
+ t.Fatalf("writers = %s, want %s", c.Op.Journal["writers"], want)
+ }
+}
+
+func TestAdmitWriterPreservesOperation(t *testing.T) {
+ for _, kind := range []string{KindServing, KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} {
+ for _, mode := range []string{"warm", "cold"} {
+ t.Run(kind+"/"+mode, 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.Op = Op{Kind: kind, ID: "m1", Phase: "probe", Exec: "boxT", Participants: []string{"boxS", "boxT"},
+ Intent: map[string]string{"target": "boxT"}, Journal: map[string]string{"source": "unchanged"},
+ By: "operator", StartedAt: c.Holder.CreatedAt}
+ c.Obligations = []Obligation{{ID: "ob1", Op: "m1", Kind: "park-local", Box: "boxS", Owner: "box:boxS", Retirement: "rt1"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ generation := ""
+ if mode == "cold" {
+ generation = "gen_a"
+ }
+ next, err := Update(ctx, s, o, AdmitWriter(0, "", "boxT", mode, generation))
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := *o.Claim
+ holder := *want.Holder
+ want.Holder = &holder
+ want.Gen++
+ want.Holder.Session = want.Gen
+ if mode == "cold" {
+ 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)
+ }
+ })
+ }
+ }
+}
+
+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"} {
+ 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.Gen + 1
+ c.Holder.Generation = "gen_a"
+ switch name {
+ case "no holder":
+ c.Holder = nil
+ c.Op = Op{}
+ case "no executor":
+ c.Op.Exec = ""
+ case "malformed journal":
+ c.Op.Journal = map[string]string{"writers": "not-json"}
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ session, generation, box, mode, newGeneration := o.Claim.Gen, "gen_a", "adi-knitting", "cold", "gen_b"
+ switch name {
+ case "no executor":
+ box = ""
+ case "non-executor":
+ box = "boxT"
+ case "stale session":
+ session--
+ case "stale generation":
+ generation = "old"
+ case "warm with generation":
+ mode = "warm"
+ case "cold without generation":
+ newGeneration = ""
+ case "cold same generation":
+ newGeneration = generation
+ case "unknown mode":
+ mode = "hot"
+ case "empty mode":
+ mode = ""
+ }
+ if _, err := Update(ctx, s, o, AdmitWriter(session, generation, box, mode, newGeneration)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("admit writer = %v, want ErrWrongState", err)
+ }
+ stored, err := Read(ctx, s, testHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(stored, o) {
+ t.Fatalf("refused admission changed stored claim or ETag: got %+v, want %+v", stored, o)
+ }
+ })
}
}