amadan

carlos / platform Public

release: control/serial — the deploy signal object (spec §3)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell f93666a33e8f35aac3f532f4d01038a36c73a049 parent bb62815
2 files changed, +130 −0
  • internal/release/serial.go +74 −0
  • internal/release/serial_test.go +56 −0
diff --git a/internal/release/serial.go b/internal/release/serial.go
new file mode 100644
index 000000000..7c315b788
--- /dev/null
+++ b/internal/release/serial.go
@@ -0,0 +1,74 @@
+// Package-file doc: control/serial is the deploy signal (carlos-deploy spec
+// §3): an UNSIGNED counter the console bumps after any bucket write a box
+// loop consumes. It carries no authority — it can only trigger verified work
+// sooner — which is why it is the one object in the bucket outside the
+// Ed25519 envelope. Boxes react to any CHANGE, not only increments, so a
+// reset or a lost-increment race can never wedge a box into ignoring bumps.
+package release
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "time"
+)
+
+// SerialKey deliberately sits at the bucket top level: nothing that lists or
+// parses app shapes (ChannelsPrefix, appPrefix) ever sees it.
+const SerialKey = "control/serial"
+
+type SerialDoc struct {
+ Schema int `json:"schema"`
+ Serial int64 `json:"serial"`
+ BumpedAt string `json:"bumped_at"`
+}
+
+// ReadSerial returns the current serial, its etag for CAS, and nil for a
+// missing object (zero doc, "" etag) — absence is the ordinary pre-first-bump
+// state, not an error. A present-but-unparseable object also reads as the
+// zero doc WITH its real etag, so the next bump overwrites the garbage.
+func ReadSerial(ctx context.Context, s Store) (SerialDoc, string, error) {
+ body, etag, err := s.Get(ctx, SerialKey)
+ if errors.Is(err, os.ErrNotExist) {
+ return SerialDoc{}, "", nil
+ }
+ if err != nil {
+ return SerialDoc{}, "", err
+ }
+ var doc SerialDoc
+ if json.Unmarshal(body, &doc) != nil {
+ return SerialDoc{}, etag, nil // garbage: recoverable, next bump replaces it
+ }
+ return doc, etag, nil
+}
+
+// BumpSerial increments the serial with a small CAS retry loop. Callers must
+// treat failure as log-only: a lost bump costs latency (the hourly floor
+// still converges everything), never correctness, and must never fail the
+// user-facing write that triggered it.
+func BumpSerial(ctx context.Context, s Store, now time.Time) error {
+ var lastErr error
+ for attempt := 0; attempt < 4; attempt++ {
+ doc, etag, err := ReadSerial(ctx, s)
+ if err != nil {
+ return err
+ }
+ next := SerialDoc{Schema: 1, Serial: doc.Serial + 1, BumpedAt: now.UTC().Format(time.RFC3339)}
+ body, err := json.Marshal(next)
+ if err != nil {
+ return err
+ }
+ if _, err = s.Put(ctx, SerialKey, body, etag); err == nil {
+ return nil
+ } else if errors.Is(err, ErrCASConflict) {
+ lastErr = err
+ continue // someone else bumped — their change signal serves ours too,
+ // but retry so OUR write's announcement isn't silently coalesced away
+ } else {
+ return err
+ }
+ }
+ return fmt.Errorf("bump %s: lost CAS race %d times: %w", SerialKey, 4, lastErr)
+}
diff --git a/internal/release/serial_test.go b/internal/release/serial_test.go
new file mode 100644
index 000000000..8499a07cd
--- /dev/null
+++ b/internal/release/serial_test.go
@@ -0,0 +1,56 @@
+package release
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSerialBumpFromMissing(t *testing.T) {
+ s := NewDirStore(t.TempDir())
+ ctx := context.Background()
+ doc, etag, err := ReadSerial(ctx, s)
+ if err != nil || doc.Serial != 0 || etag != "" {
+ t.Fatalf("missing serial should read as zero: %+v %q %v", doc, etag, err)
+ }
+ if err := BumpSerial(ctx, s, time.Unix(1754640000, 0).UTC()); err != nil {
+ t.Fatal(err)
+ }
+ doc, etag, err = ReadSerial(ctx, s)
+ if err != nil || doc.Serial != 1 || doc.Schema != 1 || etag == "" {
+ t.Fatalf("first bump: %+v %q %v", doc, etag, err)
+ }
+ if doc.BumpedAt == "" {
+ t.Fatal("bumped_at not stamped")
+ }
+}
+
+func TestSerialBumpIncrements(t *testing.T) {
+ s := NewDirStore(t.TempDir())
+ ctx := context.Background()
+ for i := 0; i < 3; i++ {
+ if err := BumpSerial(ctx, s, time.Unix(1754640000, 0).UTC()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ doc, _, _ := ReadSerial(ctx, s)
+ if doc.Serial != 3 {
+ t.Fatalf("serial = %d, want 3", doc.Serial)
+ }
+}
+
+func TestSerialGarbageObjectStillBumps(t *testing.T) {
+ // A bucket-writer can corrupt the object; a bump must recover, not wedge.
+ s := NewDirStore(t.TempDir())
+ ctx := context.Background()
+ if _, err := s.Put(ctx, SerialKey, []byte("not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := BumpSerial(ctx, s, time.Unix(1754640000, 0).UTC()); err != nil {
+ t.Fatal(err)
+ }
+ doc, _, err := ReadSerial(ctx, s)
+ if err != nil || doc.Serial != 1 {
+ t.Fatalf("recovery bump: %+v %v", doc, err)
+ }
+}