amadan

carlos / platform Public

ledger: Append — chained entries advancing HEAD by CAS

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 8b8734ade0c48e6d260245f00dbe719cfdfd0419 parent 96be71b
2 files changed, +232 −0
  • internal/ledger/append.go +102 −0
  • internal/ledger/append_test.go +130 −0
diff --git a/internal/ledger/append.go b/internal/ledger/append.go
new file mode 100644
index 000000000..43dd13b8d
--- /dev/null
+++ b/internal/ledger/append.go
@@ -0,0 +1,102 @@
+package ledger
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type Appended struct {
+ Key string `json:"key"`
+ SHA256 string `json:"sha256"`
+ Count int `json:"count"`
+}
+
+// appendAttempts bounds the CAS retry loop. Conflicts are rare (two
+// writers accruing the same app's ledger in the same moment); four
+// attempts outlasts any plausible burst without ever spinning.
+const appendAttempts = 4
+
+// Append validates, chains, and writes one entry, advancing HEAD with CAS.
+// A lost HEAD race re-reads and re-chains (up to 4 attempts). An entry
+// object orphaned by a lost race is harmless: nothing references it, and
+// Verify walks only the chain.
+func Append(ctx context.Context, st release.Store, sqid, app, name, writer string,
+ payload json.RawMessage, now func() time.Time) (Appended, error) {
+ if !ValidName(name) {
+ return Appended{}, fmt.Errorf("ledger: invalid ledger name %q", name)
+ }
+ if len(payload) > MaxPayload {
+ return Appended{}, fmt.Errorf("ledger: payload %d bytes exceeds cap %d", len(payload), MaxPayload)
+ }
+ if !json.Valid(payload) {
+ return Appended{}, fmt.Errorf("ledger: payload is not valid JSON")
+ }
+
+ var lastErr error
+ for range appendAttempts {
+ headKey := HeadKey(sqid, app, name)
+ hb, etag, err := st.Get(ctx, headKey)
+ var head Head
+ exists := err == nil
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ // First entry: prev fields stay empty, HEAD is create-only.
+ case err != nil:
+ return Appended{}, err
+ default:
+ if err := json.Unmarshal(hb, &head); err != nil {
+ return Appended{}, fmt.Errorf("ledger: %s corrupt: %w", headKey, err)
+ }
+ }
+
+ ts := now().UTC().Format(time.RFC3339)
+ e := Entry{
+ Schema: Schema, Ledger: name, Writer: writer, TS: ts,
+ Payload: payload, PrevKey: head.HeadKey, PrevSHA256: head.HeadSHA256,
+ }
+ eb, err := json.Marshal(e)
+ if err != nil {
+ return Appended{}, err
+ }
+ sha := SHA256Hex(eb)
+ key := EntryKey(sqid, app, name, ts, sha)
+ if _, err := st.Put(ctx, key, eb, ""); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ // Same second, same bytes-shape collision: retry re-reads
+ // HEAD and produces a differently-chained entry.
+ lastErr = err
+ continue
+ }
+ return Appended{}, err
+ }
+
+ nh := Head{Schema: Schema, UpdatedAt: ts, HeadKey: key, HeadSHA256: sha, Count: head.Count + 1}
+ nhb, err := json.Marshal(nh)
+ if err != nil {
+ return Appended{}, err
+ }
+ ifMatch := ""
+ if exists {
+ ifMatch = etag
+ }
+ if _, err := st.Put(ctx, headKey, nhb, ifMatch); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ // Lost the race: another writer advanced HEAD. Our entry
+ // object is now an unreferenced orphan — harmless in a
+ // versioned bucket; Verify walks only the chain. Re-read
+ // and re-chain.
+ lastErr = err
+ continue
+ }
+ return Appended{}, err
+ }
+ return Appended{Key: key, SHA256: sha, Count: nh.Count}, nil
+ }
+ return Appended{}, fmt.Errorf("ledger: append lost %d CAS races: %w", appendAttempts, lastErr)
+}
diff --git a/internal/ledger/append_test.go b/internal/ledger/append_test.go
new file mode 100644
index 000000000..53b3aaedd
--- /dev/null
+++ b/internal/ledger/append_test.go
@@ -0,0 +1,130 @@
+package ledger
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func fixedNow() time.Time { return time.Date(2026, 8, 3, 21, 4, 0, 0, time.UTC) }
+
+func TestAppendFirstAndSecond(t *testing.T) {
+ st := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+
+ a1, err := Append(ctx, st, "bab", "platform", "carbon", "paul@keymail.dev",
+ json.RawMessage(`{"n":1}`), fixedNow)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // HEAD points at the entry, count 1, and the stored bytes hash to
+ // the recorded sha.
+ hb, _, err := st.Get(ctx, HeadKey("bab", "platform", "carbon"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var h Head
+ if err := json.Unmarshal(hb, &h); err != nil {
+ t.Fatal(err)
+ }
+ if h.HeadKey != a1.Key || h.Count != 1 || h.Schema != Schema {
+ t.Fatalf("head = %+v, appended = %+v", h, a1)
+ }
+ eb, _, err := st.Get(ctx, a1.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if SHA256Hex(eb) != a1.SHA256 || h.HeadSHA256 != a1.SHA256 {
+ t.Fatal("stored bytes do not hash to the recorded sha")
+ }
+ var e1 Entry
+ if err := json.Unmarshal(eb, &e1); err != nil {
+ t.Fatal(err)
+ }
+ if e1.PrevKey != "" || e1.PrevSHA256 != "" || e1.Writer != "paul@keymail.dev" {
+ t.Fatalf("first entry = %+v", e1)
+ }
+
+ // Second append chains to the first.
+ later := func() time.Time { return fixedNow().Add(time.Minute) }
+ a2, err := Append(ctx, st, "bab", "platform", "carbon", "paul@keymail.dev",
+ json.RawMessage(`{"n":2}`), later)
+ if err != nil {
+ t.Fatal(err)
+ }
+ eb2, _, _ := st.Get(ctx, a2.Key)
+ var e2 Entry
+ if err := json.Unmarshal(eb2, &e2); err != nil {
+ t.Fatal(err)
+ }
+ if e2.PrevKey != a1.Key || e2.PrevSHA256 != a1.SHA256 || a2.Count != 2 {
+ t.Fatalf("second entry does not chain: %+v (count %d)", e2, a2.Count)
+ }
+}
+
+func TestAppendRejects(t *testing.T) {
+ st := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ if _, err := Append(ctx, st, "bab", "platform", "Not A Name", "w",
+ json.RawMessage(`{}`), fixedNow); err == nil {
+ t.Fatal("bad ledger name accepted")
+ }
+ big := json.RawMessage(`"` + string(make([]byte, MaxPayload)) + `"`)
+ if _, err := Append(ctx, st, "bab", "platform", "carbon", "w", big, fixedNow); err == nil {
+ t.Fatal("oversized payload accepted")
+ }
+ if _, err := Append(ctx, st, "bab", "platform", "carbon", "w",
+ json.RawMessage(`{not json`), fixedNow); err == nil {
+ t.Fatal("invalid JSON payload accepted")
+ }
+}
+
+// casOnceStore makes the FIRST head Put lose its race, as if another
+// writer advanced HEAD between our read and our write.
+type casOnceStore struct {
+ release.Store
+ tripped bool
+}
+
+func (c *casOnceStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if !c.tripped && key == HeadKey("bab", "platform", "carbon") {
+ c.tripped = true
+ // Simulate the race: a competing append lands first.
+ if _, err := Append(ctx, c.Store, "bab", "platform", "carbon", "rival",
+ json.RawMessage(`{"rival":true}`), fixedNow); err != nil {
+ return "", err
+ }
+ return "", release.ErrCASConflict
+ }
+ return c.Store.Put(ctx, key, body, ifMatch)
+}
+
+func TestAppendRetriesLostRace(t *testing.T) {
+ st := &casOnceStore{Store: release.NewDirStore(t.TempDir())}
+ ctx := context.Background()
+ later := func() time.Time { return fixedNow().Add(time.Minute) }
+ a, err := Append(ctx, st, "bab", "platform", "carbon", "me",
+ json.RawMessage(`{"mine":true}`), later)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if a.Count != 2 {
+ t.Fatalf("count = %d, want 2 (rival's entry then ours)", a.Count)
+ }
+ eb, _, err := st.Get(ctx, a.Key)
+ if errors.Is(err, os.ErrNotExist) || err != nil {
+ t.Fatal(err)
+ }
+ var e Entry
+ if err := json.Unmarshal(eb, &e); err != nil {
+ t.Fatal(err)
+ }
+ if e.PrevKey == "" {
+ t.Fatal("retried entry did not re-chain onto the rival's head")
+ }
+}