diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 347366e0..8dfd056b 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -91,6 +91,7 @@ Two stdlib-only reference implementations ship in-tree: - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance subdirectories, each an append-only journal, an atomic checkpoint, an idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file. + `Append` refuses a Record over its 16 MiB line limit with `ErrRecordTooLarge`. Each append flushes to stable storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity. Use it for durability across restarts without a database. @@ -126,8 +127,13 @@ in by implementing `HistoryStore` (the in-tree `MemStore` does so under ## A note on serialized payloads -Events, service done-data, actor done-data, and actor messages are recorded as -their JSON form. A parent reducer that type-asserts a non-JSON Go type from -`AssignCtx.Event` observes the JSON-decoded shape on a replayed `onDone`. A +Events, service done-data, actor done-data, and actor messages are recorded as their +JSON form. A parent reducer that type-asserts a non-JSON Go type from `AssignCtx.Event` +observes the JSON-decoded shape on a replayed `onDone`. A `WithEventData` payload is +JSON-decoded live as well as on replay, so a reducer gets the same value both ways: a +struct is a map and an integer a `float64`. An error stays an error with the same +message, with any invalid UTF-8 replaced. `Handle.Fire` refuses a payload it can't +record, or one over `WithMaxEventDataBytes` (8 MiB by default), with +`ErrEventDataEncoding`, and nothing applies. A typed-codec option to carry the concrete Go value across the journal boundary is reserved for a later, additive change. diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 74a844a0..a4e26561 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -21,12 +21,30 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. fence the old owner: a stale write before then is accepted, and the new owner's write at that step gets `ErrStepConflict`. +- **`WithMaxEventDataBytes`** caps a `WithEventData` payload. `ErrEventDataEncoding` reports + a refused payload, and `ErrRecordTooLarge` a Record too large for a `FileStore` line. + `Record.EventData` and `Record.EventDataError` hold the payload. + ### Changed - Requires Go 1.27.3. -- **A `Store` implementation has new work to do.** It must enforce fence tokens and - return `ErrStepConflict`, or `WithOwnerFence` protects nothing. Existing - implementations still compile, so nothing flags this. +- **A `WithEventData` payload reaches `Assign` decoded, live as well as on replay.** A + struct arrives as a map and an integer as a `float64`, which loses precision above + 2^53, so an assertion to the original + type now fails at `Fire`, not at `Recover`. An error arrives as an error with the same + message, any invalid UTF-8 replaced, and `errors.Is` no longer matches it. +- **`Handle.Fire` can refuse a payload.** One with no JSON form that decodes again, or + over `WithMaxEventDataBytes` (8 MiB by default), returns the new `ErrEventDataEncoding` + before the kernel runs. Fires of that size used to pass, because the payload wasn't + journaled. +- **`FileStore.Append` refuses a Record over its 16 MiB line limit** with the new + `ErrRecordTooLarge`. It used to write a line `Load` couldn't read back unless a + checkpoint compacted it away. An oversized line now drops its `Snapshot` first, which + recovery never reads, so large checkpoints keep working. +- **A `Store` implementation has new work to do.** It must enforce fence tokens, return + `ErrStepConflict` and keep `Record.EventData` and `Record.EventDataError`, or + `WithOwnerFence` protects nothing and payloads are lost. Existing implementations + still compile, so nothing flags this. - **`MemStore.Append` without `WithIdempotencyKey` refuses a Record JSON can't encode,** as `FileStore` already did. @@ -41,6 +59,10 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### Fixed +- **`WithEventData` survives recovery.** `Handle.Fire` records the payload in the new + `Record.EventData` and `Record.EventDataError`, and `Recover` and `StateAt` supply it + on replay. This needs `state.FireEventData`, so `durable` requires `state` v1.1.0 or + later, and its `require` must move there before a `durable` tag. - **Recording-clock buffer race.** Closing a step's `Record` (draining the recording clock) and the clock-read index reads used to mirror timer deadlines now take the recording clock's mutex, so they are safe against a `Now()` the diff --git a/durable/README.md b/durable/README.md index faad6cd2..9b9858c6 100644 --- a/durable/README.md +++ b/durable/README.md @@ -101,6 +101,7 @@ Two stdlib-only reference implementations ship in-tree: - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance subdirectories, each an append-only journal, an atomic checkpoint, an idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file. + `Append` refuses a Record over its 16 MiB line limit with `ErrRecordTooLarge`. Each append flushes to stable storage; each checkpoint uses write-temp-plus-rename for crash-safe atomicity. Use it for durability across restarts without a database. @@ -137,9 +138,14 @@ otherwise. ## Caveat: serialized payloads -Events, service done-data, actor done-data, and actor messages are recorded as -their JSON form. A parent reducer that type-asserts a non-JSON Go type from -`AssignCtx.Event` observes the JSON-decoded shape on the replayed `onDone`. A +Events, service done-data, actor done-data, and actor messages are recorded as their +JSON form. A parent reducer that type-asserts a non-JSON Go type from `AssignCtx.Event` +observes the JSON-decoded shape on a replayed `onDone`. A `WithEventData` payload is +JSON-decoded live as well as on replay, so a reducer gets the same value both ways: a +struct is a map and an integer a `float64`. An error stays an error with the same +message, with any invalid UTF-8 replaced. `Handle.Fire` refuses a payload it can't +record, or one over `WithMaxEventDataBytes` (8 MiB by default), with +`ErrEventDataEncoding`, and nothing applies. A typed-codec option to carry the concrete Go value across the journal boundary is reserved for a later, additive change. diff --git a/durable/doc.go b/durable/doc.go index b4d5b6c9..00e5a8fa 100644 --- a/durable/doc.go +++ b/durable/doc.go @@ -31,10 +31,11 @@ // // # Caveats // -// Payloads are serialized at the journal boundary. Event, service done-data, -// actor done-data, and actor messages are recorded as their JSON form; a parent -// reducer that type-asserts a non-JSON Go type from AssignCtx.Event observes the -// JSON-decoded shape on the replayed onDone. A typed-codec option to carry the +// Payloads are serialized at the journal boundary. Event, service done-data, actor +// done-data, and actor messages are recorded as their JSON form; a parent reducer that +// type-asserts a non-JSON Go type from AssignCtx.Event observes the JSON-decoded shape +// on a replayed onDone. A WithEventData payload is JSON-decoded live as well as on +// replay, so a reducer gets the same value both ways. A typed-codec option to carry the // concrete Go value across the boundary is reserved for a later, additive change. // // The kernel stays pure and dependency-free. The durable module is host-side: diff --git a/durable/errors.go b/durable/errors.go index 7b9d0752..d08e5f47 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -14,6 +14,8 @@ const ( errorInstanceExists = "instance already exists" errorEffectDispatch = "effect dispatch failed" errorStepOutOfRange = "time-travel step out of range" + errorEventDataEncoding = "event data can't be recorded" + errorRecordTooLarge = "record too large for the journal" ) // Sentinel errors a Store reports. Callers match them with errors.Is. @@ -38,6 +40,18 @@ var ( // a newer owner has taken the instance over. Nothing is written. ErrFenced = errors.New(errorPrefix + errorFenced) + // ErrEventDataEncoding wraps a Handle.Fire refused before the kernel ran, because + // its state.WithEventData payload has no JSON form that decodes again, panics while + // encoding (such as an error that's a typed nil), or exceeds WithMaxEventDataBytes. + // Nothing applied, so the Handle stays usable. + ErrEventDataEncoding = errors.New(errorPrefix + errorEventDataEncoding) + + // ErrRecordTooLarge is reported by FileStore.Append for a Record whose journal line + // exceeds the store's line limit, which a later Load could not read back, even + // without its Snapshot. No journal line is written, but a Handle's kernel has already + // applied the step. + ErrRecordTooLarge = errors.New(errorPrefix + errorRecordTooLarge) + // ErrCheckpointNotAdvancing is reported by Store.Checkpoint when throughStep // does not advance beyond the instance's current checkpoint. ErrCheckpointNotAdvancing = errors.New(errorPrefix + errorCheckpointNotAdvancing) diff --git a/durable/event_data_test.go b/durable/event_data_test.go new file mode 100644 index 00000000..058bafd1 --- /dev/null +++ b/durable/event_data_test.go @@ -0,0 +1,402 @@ +package durable_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "strings" + "testing" + "time" + + "github.com/stablekernel/crucible/durable" + "github.com/stablekernel/crucible/state" +) + +func TestRecover_FireWithEventData_FileStoreAndCheckpointTail(t *testing.T) { + ctx := context.Background() + m := quantityMachine() + dir := t.TempDir() + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("qty-1") + runner := durable.NewRunner(m, st, durable.WithCheckpointEvery[string, string, *runCtx](2)) + h, err := runner.Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + for _, q := range []float64{3, 4, 5} { // the checkpoint lands on the second + if _, err = h.Fire(ctx, "add", state.WithEventData(q)); err != nil { + t.Fatalf("Fire: %v", err) + } + } + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + recovered, err := durable.Recover(ctx, m, reopened, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if live, got := handleSnapshot(t, h), handleSnapshot(t, recovered); got != live { + t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) + } +} + +func TestStateAt_ReplaysEventData(t *testing.T) { + ctx := context.Background() + m := quantityMachine() + store := durable.NewMemStore(durable.WithHistory()) + const id = durable.InstanceID("qty-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(3.0)); err != nil { + t.Fatalf("Fire: %v", err) + } + view, err := durable.StateAt(ctx, m, store, id, 0) + if err != nil { + t.Fatalf("StateAt: %v", err) + } + got, err := state.MarshalSnapshot(view.Snapshot()) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if live := handleSnapshot(t, h); string(got) != live { + t.Fatalf("view differs from live:\n live: %s\n got: %s", live, got) + } +} + +func TestFire_EventDataWithNoJSONForm_RefusedBeforeTheKernelRuns(t *testing.T) { + ctx := context.Background() + m := quantityMachine() + store := durable.NewMemStore() + const id = durable.InstanceID("qty-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + before := handleSnapshot(t, h) + if _, err = h.Fire(ctx, "add", state.WithEventData(make(chan int))); err == nil { + t.Fatal("Fire with an unencodable payload succeeded") + } + if got := handleSnapshot(t, h); got != before { + t.Fatalf("refused Fire moved the instance:\n before: %s\n after: %s", before, got) + } + if n := tailLen(t, store, id); n != 0 { + t.Fatalf("records = %d, want 0", n) + } +} + +func TestFire_NilEventData_ReachesReplay(t *testing.T) { + ctx := context.Background() + var seen []any + m := state.ForgeFor[*runCtx]("nil-data"). + Reducer("note", func(in state.AssignCtx[*runCtx]) *runCtx { + seen = append(seen, in.Event) + return in.Entity + }). + State("open"). + Initial("open"). + Transition("open").On("add").GoTo("open").Assign("note"). + Quench() + store := durable.NewMemStore() + const id = durable.InstanceID("nil-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(nil)); err != nil { + t.Fatalf("Fire: %v", err) + } + if _, err = durable.Recover(ctx, m, store, id); err != nil { + t.Fatalf("Recover: %v", err) + } + if len(seen) != 2 || seen[0] != nil || seen[1] != nil { + t.Fatalf("assign saw %v, want nil live and nil on replay", seen) + } +} + +// A Fire with no payload keeps today's on-disk Record, so existing journals and +// their digests don't change shape. +func TestRecord_NoEventData_OmittedFromJSON(t *testing.T) { + b, err := json.Marshal(durable.Record{Step: 1}) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + if strings.Contains(string(b), "EventData") { + t.Fatalf("record JSON carries EventData: %s", b) + } +} + +// payloadMachine records what each Fire's Assign saw, so a test can compare the live +// run with a replay. +func payloadMachine(seen *[]any) *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("payload"). + Reducer("note", func(in state.AssignCtx[*runCtx]) *runCtx { + *seen = append(*seen, in.Event) + return in.Entity + }). + State("open"). + Initial("open"). + Transition("open").On("add").GoTo("open").Assign("note"). + Quench() +} + +// TestFire_EventData_LiveEqualsReplay proves Assign sees one value for each kind of +// payload, whether the step ran live or was replayed. +func TestFire_EventData_LiveEqualsReplay(t *testing.T) { + type order struct{ Qty int } + tests := []struct { + name string + payload any + want string + }{ + {"an int", 3, "float64:3"}, + {"a large int", int64(1<<53 + 1), "float64:9.007199254740992e+15"}, + {"a struct", order{Qty: 2}, "map[string]interface {}:map[Qty:2]"}, + {"an error", errors.New("card declined"), "*errors.errorString:card declined"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + var seen []any + m := payloadMachine(&seen) + store := durable.NewMemStore() + const id = durable.InstanceID("p-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(tt.payload)); err != nil { + t.Fatalf("Fire: %v", err) + } + if _, err = durable.Recover(ctx, m, store, id); err != nil { + t.Fatalf("Recover: %v", err) + } + if len(seen) != 2 { + t.Fatalf("assign ran %d times, want 2", len(seen)) + } + for i, v := range seen { + if got := fmt.Sprintf("%T:%v", v, v); got != tt.want { + t.Fatalf("assign %d saw %s, want %s", i, got, tt.want) + } + } + }) + } +} + +// TestFire_NoEventData_AssignSeesTheEvent proves a Fire with no payload keeps the +// event in AssignCtx.Event. +func TestFire_NoEventData_AssignSeesTheEvent(t *testing.T) { + ctx := context.Background() + var seen []any + h, err := durable.NewRunner(payloadMachine(&seen), durable.NewMemStore()). + Start(ctx, "p-1", &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add"); err != nil { + t.Fatalf("Fire: %v", err) + } + if len(seen) != 1 || seen[0] != "add" { + t.Fatalf("assign saw %v, want the event", seen) + } +} + +// panicMarshaler panics while encoding. +type panicMarshaler struct{} + +func (panicMarshaler) MarshalJSON() ([]byte, error) { panic("boom") } + +// nilError panics on a nil receiver, as an error with a pointer receiver can. +type nilError struct{ msg string } + +func (e *nilError) Error() string { return e.msg } + +// TestFire_ErrorEventData_InvalidUTF8_LiveEqualsReplay proves an error message reads +// the same live and from a reopened FileStore, whose JSON replaces invalid UTF-8. +func TestFire_ErrorEventData_InvalidUTF8_LiveEqualsReplay(t *testing.T) { + ctx := context.Background() + var seen []any + m := payloadMachine(&seen) + dir := t.TempDir() + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("p-1") + h, err := durable.NewRunner(m, st).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(errors.New("bad\xffbyte"))); err != nil { + t.Fatalf("Fire: %v", err) + } + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, err = durable.Recover(ctx, m, reopened, id); err != nil { + t.Fatalf("Recover: %v", err) + } + if len(seen) != 2 || seen[0].(error).Error() != seen[1].(error).Error() { + t.Fatalf("live and replay saw %q", seen) + } +} + +// TestFire_EventData_Refused proves a payload that can't be recorded refuses the Fire +// before the kernel runs, and the Handle stays usable. +func TestFire_EventData_Refused(t *testing.T) { + tooBig, ok := new(big.Int).SetString("1"+strings.Repeat("0", 400), 10) + if !ok { + t.Fatal("SetString") + } + tests := []struct { + name string + payload any + }{ + {"no JSON form", make(chan int)}, + {"a number no float64 holds", tooBig}, + {"over the limit", strings.Repeat("x", 100)}, + {"an error over the limit", errors.New(strings.Repeat("x", 100))}, + {"an error over the limit once encoded", errors.New(strings.Repeat("<", 20))}, + {"a typed-nil error", (*nilError)(nil)}, + {"a panicking marshaler", panicMarshaler{}}, + {"a nil embedded marshaler", struct{ *time.Time }{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + var seen []any + store := durable.NewMemStore() + runner := durable.NewRunner(payloadMachine(&seen), store, + durable.WithMaxEventDataBytes[string, string, *runCtx](64)) + const id = durable.InstanceID("p-1") + h, err := runner.Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(tt.payload)); !errors.Is(err, durable.ErrEventDataEncoding) { + t.Fatalf("Fire = %v, want ErrEventDataEncoding", err) + } + if len(seen) != 0 { + t.Fatalf("assign ran on a refused Fire: %v", seen) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(1.0)); err != nil { + t.Fatalf("Fire after the refusal: %v", err) + } + }) + } +} + +// TestFire_EventData_LeavesTheCallersOptionsAlone proves the decoded option doesn't +// land in spare capacity of the caller's slice. +func TestFire_EventData_LeavesTheCallersOptionsAlone(t *testing.T) { + ctx := context.Background() + var seen []any + h, err := durable.NewRunner(payloadMachine(&seen), durable.NewMemStore()). + Start(ctx, "p-1", &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + opts := make([]state.FireOption, 1, 2) + opts[0] = state.WithEventData(1) + if _, err = h.Fire(ctx, "add", opts...); err != nil { + t.Fatalf("Fire: %v", err) + } + if spare := opts[:2][1]; spare != nil { + t.Fatal("Fire wrote into the caller's slice") + } +} + +// TestFileStore_RecordOverTheLineLimit_IsRefused proves a Record Load couldn't read back +// is refused, leaving the instance readable. +func TestFileStore_RecordOverTheLineLimit_IsRefused(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("p-1") + if _, err = st.Append(ctx, id, durable.Record{Step: 0}); err != nil { + t.Fatalf("Append: %v", err) + } + big := durable.Record{Step: 1, EventData: make([]byte, 13<<20)} + if _, err = st.Append(ctx, id, big); !errors.Is(err, durable.ErrRecordTooLarge) { + t.Fatalf("Append = %v, want ErrRecordTooLarge", err) + } + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, tail, err := reopened.Load(ctx, id); err != nil || len(tail) != 1 { + t.Fatalf("Load = (%d records, %v), want (1, nil)", len(tail), err) + } +} + +// TestFileStore_LargeRecordWithSnapshot_StaysReadable proves a Record over the line +// limit only because of its Snapshot is kept without it, so Load still works if the +// checkpoint never lands. +func TestFileStore_LargeRecordWithSnapshot_StaysReadable(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("p-1") + rec := durable.Record{Step: 0, Event: []byte(`"go"`), Snapshot: make([]byte, 13<<20)} + if _, err = st.Append(ctx, id, rec); err != nil { + t.Fatalf("Append: %v", err) + } + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + _, tail, err := reopened.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(tail) != 1 || string(tail[0].Event) != `"go"` || len(tail[0].Snapshot) != 0 { + t.Fatalf("tail = %d records, want the step without its Snapshot", len(tail)) + } + if _, err = reopened.Append(ctx, id, rec); err != nil { + t.Fatalf("retry of the same Record: %v", err) + } +} + +// TestFileStore_LargeSnapshotRecord_SurvivesCompaction proves compaction doesn't rewrite +// the Snapshot an oversized Record dropped, which would leave a line Load can't read. +func TestFileStore_LargeSnapshotRecord_SurvivesCompaction(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + st, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("NewFileStore: %v", err) + } + const id = durable.InstanceID("p-1") + for step := range 4 { + if _, err = st.Append(ctx, id, durable.Record{Step: step}); err != nil { + t.Fatalf("Append: %v", err) + } + } + if _, err = st.Append(ctx, id, durable.Record{Step: 4, Snapshot: make([]byte, 13<<20)}); err != nil { + t.Fatalf("Append: %v", err) + } + if err = st.Checkpoint(ctx, id, []byte("{}"), 2); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + reopened, err := durable.NewFileStore(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + if _, tail, err := reopened.Load(ctx, id); err != nil || len(tail) != 2 { + t.Fatalf("Load = (%d records, %v), want (2, nil)", len(tail), err) + } +} diff --git a/durable/fence_test.go b/durable/fence_test.go index d921bb8c..6302dbb4 100644 --- a/durable/fence_test.go +++ b/durable/fence_test.go @@ -154,7 +154,7 @@ func TestStore_Append_EmptyAndNilFieldsAreTheSameRecord(t *testing.T) { if _, err := st.Append(ctx, id, durable.Record{Step: 0}); err != nil { t.Fatalf("Append: %v", err) } - same := durable.Record{Step: 0, Event: []byte{}, Snapshot: []byte{}, Entries: []state.JournalEntry{}} + same := durable.Record{Step: 0, Event: []byte{}, EventData: []byte{}, Snapshot: []byte{}, Entries: []state.JournalEntry{}} if _, err := st.Append(ctx, id, same); err != nil { t.Fatalf("retry: %v", err) } diff --git a/durable/filestore.go b/durable/filestore.go index b7a9462d..7f46d6f0 100644 --- a/durable/filestore.go +++ b/durable/filestore.go @@ -361,9 +361,8 @@ func scanLines(path, label string, fn func(line []byte)) (err error) { return nil } -// maxLineBytes bounds a single journal/ledger line so a corrupt length can't -// exhaust memory; a marshaled Record (including a checkpoint envelope) stays well -// under it in practice. +// maxLineBytes bounds a single journal/ledger line so a corrupt length can't exhaust +// memory. Append keeps every line under it, dropping a Record's Snapshot or refusing it. const maxLineBytes = 16 * 1024 * 1024 // appliedEntry is one line of the idempotency ledger. @@ -400,14 +399,27 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. ErrStepOutOfOrder, rec.Step, inst.maxStep, id) } - if err = inst.ensureDir(); err != nil { - return 0, err - } - inst.existed = true line, err := json.Marshal(cloneRecord(rec)) if err != nil { return 0, fmt.Errorf("durable: marshaling record step %d for %q: %w", rec.Step, id, err) } + kept := cloneRecord(rec) + if len(line) >= maxLineBytes && len(rec.Snapshot) > 0 { + // Recovery reads the checkpoint, never a tail Record's Snapshot, so an oversized + // Record drops it rather than leaving a line Load can't read. The in-memory tail + // drops it too, because compaction rewrites the journal from there. + kept.Snapshot = nil + if line, err = json.Marshal(kept); err != nil { + return 0, fmt.Errorf("durable: marshaling record step %d for %q: %w", rec.Step, id, err) + } + } + if len(line) >= maxLineBytes { + return 0, fmt.Errorf("%w: step %d for %q is %d bytes", ErrRecordTooLarge, rec.Step, id, len(line)) + } + if err = inst.ensureDir(); err != nil { + return 0, err + } + inst.existed = true if err = appendLine(filepath.Join(inst.dir, journalFile), line); err != nil { return 0, fmt.Errorf("durable: appending journal for %q: %w", id, err) } @@ -421,7 +433,7 @@ func (s *FileStore) Append(_ context.Context, id InstanceID, rec Record, opts .. return 0, fmt.Errorf("durable: appending applied ledger for %q: %w", id, err) } - inst.tail = append(inst.tail, cloneRecord(rec)) + inst.tail = append(inst.tail, kept) inst.maxStep = rec.Step inst.seq = seq inst.applied[key] = appliedRecord{seq: seq, digest: digest} diff --git a/durable/memstore.go b/durable/memstore.go index 289245fa..af746094 100644 --- a/durable/memstore.go +++ b/durable/memstore.go @@ -363,6 +363,13 @@ func cloneRecord(rec Record) Record { if rec.Event != nil { out.Event = append([]byte(nil), rec.Event...) } + if rec.EventData != nil { + out.EventData = append([]byte(nil), rec.EventData...) + } + if rec.EventDataError != nil { + msg := *rec.EventDataError + out.EventDataError = &msg + } if len(rec.Entries) > 0 { out.Entries = append(out.Entries, rec.Entries...) } diff --git a/durable/runner.go b/durable/runner.go index 67e7ab02..9e1aa048 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -24,9 +24,10 @@ import ( // // For an event-driven machine every transition is a pure function of // (configuration, context, event payload, machine definition). The only input a -// Runner must record to reproduce a run is therefore the driving event: each -// Fire appends a Record{Step, Event} where Step is the produced Trace ordinal and -// Event is the kernel's structured Trace.EventPayload. Periodically — governed by +// Runner must record to reproduce a run is therefore the driving event and any +// payload it carried: each Fire appends a Record{Step, Event} where Step is the +// produced Trace ordinal and Event is the kernel's structured Trace.EventPayload, +// plus the payload's JSON form for a state.WithEventData Fire. Periodically — governed by // the checkpoint policy (WithCheckpointEvery) — the Runner also persists a full // marshaled Snapshot and compacts the journal tail through that step, so recovery // replays only the tail after the latest checkpoint rather than the whole run. @@ -338,7 +339,8 @@ func (r *Runner[S, E, C]) newActorSystem(inst *state.Instance[S, E, C]) *state.A } // Fire drives one event through a durable instance identified by id, loading and -// replaying it from the Store first, then recording the step. It is the +// replaying it from the Store first, then recording the step. Handle.Fire describes +// how a state.WithEventData payload is handled, and when it's refused. It is the // stateless entry point (no Handle required); for a hot path that fires many // events in sequence, hold a Handle from Start or Recover and use Handle.Fire to // avoid reloading between steps. @@ -356,7 +358,24 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // checkpoint policy is due — persists a full Snapshot and compacts the tail. A // kernel transition error is recorded as a no-op (no step was produced) and // returned to the caller. +// +// A state.WithEventData payload reaches the transition's Assign in its JSON-decoded +// form, live and on replay alike, so both see the same value: a struct arrives as a +// map and an integer as a float64, which loses precision above 2^53. An error arrives +// as an error with the same message, with any invalid UTF-8 replaced, though errors.Is +// no longer matches it. Keep its message, not the error, in context: a checkpoint turns +// an error into a map, and Recover fails on an error-typed field. A payload that can't take this form, an error whose Error panics (such as +// a typed nil), or +// exceeds WithMaxEventDataBytes, refuses the Fire with ErrEventDataEncoding before the +// kernel runs. The payload is read on the calling goroutine with no lock of its own, so +// don't mutate it concurrently with Fire. func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireOption) (state.FireResult[S], error) { + // Encode the WithEventData payload before the kernel runs, so a payload that + // can't be recorded refuses the Fire rather than applying a step it can't record. + payload, opts, err := encodeEventData(opts, h.runner.cfg.maxEventData) + if err != nil { + return state.FireResult[S]{}, fmt.Errorf("durable: recording event data for %q: %w", h.id, err) + } res := h.inst.Fire(ctx, event, opts...) if res.Err != nil { return res, res.Err @@ -378,7 +397,7 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO if err != nil { return res, err } - rec := Record{Step: step, Event: []byte(res.Trace.EventPayload), Entries: h.drainClock(), Effects: effEnvs} + rec := Record{Step: step, Event: []byte(res.Trace.EventPayload), EventData: payload.data, EventDataError: payload.errMsg, Entries: h.drainClock(), Effects: effEnvs} // Write-ahead append the step Record (with its effect ids) BEFORE dispatching, but // hold off compacting it: dispatch happens between the append and the checkpoint so // a crash mid-dispatch leaves the effect recorded in the still-present tail for @@ -617,7 +636,11 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S if err != nil { return nil, fmt.Errorf("durable: decoding recorded event at step %d for %q: %w", rec.Step, id, err) } - res := inst.Fire(ctx, event) + fireOpts, err := replayFireOptions(rec) + if err != nil { + return nil, fmt.Errorf("durable: decoding recorded event data at step %d for %q: %w", rec.Step, id, err) + } + res := inst.Fire(ctx, event, fireOpts...) if res.Err != nil { return nil, fmt.Errorf("durable: replaying step %d for %q: %w", rec.Step, id, res.Err) } @@ -718,6 +741,110 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S }, nil } +// eventPayload is the recorded form of a WithEventData payload. +type eventPayload struct { + data []byte // the JSON form, nil for none or for an error payload + errMsg *string // an error payload's message +} + +// encodeEventData returns the recorded form of the WithEventData payload in opts and +// the options to Fire with, which carry the payload as replay will see it, so the live +// run and a replay can't differ. With no payload it returns opts unchanged, because +// adding one would replace the event in AssignCtx.Event. It copies opts rather than +// appending, which could write into the caller's slice. +func encodeEventData(opts []state.FireOption, maxBytes int) (eventPayload, []state.FireOption, error) { + data, ok := state.FireEventData(opts...) + if !ok { + return eventPayload{}, opts, nil + } + var payload eventPayload + encoded := data + e, isErr := data.(error) + if isErr { + msg, err := errorMessage(e) + if err != nil { + return payload, nil, fmt.Errorf("%w: %w", ErrEventDataEncoding, err) + } + encoded = msg + } + b, err := marshalPayload(encoded) + if err != nil { + return payload, nil, fmt.Errorf("%w: %w", ErrEventDataEncoding, err) + } + if len(b) > maxBytes { + return payload, nil, eventDataTooLarge(len(b), maxBytes) + } + if isErr { + // Take the message back from its JSON form, which is what a Store keeps: JSON + // replaces invalid UTF-8, so the raw message could differ from a replay's. + var msg string + if err = json.Unmarshal(b, &msg); err != nil { + return payload, nil, fmt.Errorf("%w: %w", ErrEventDataEncoding, err) + } + payload.errMsg = &msg + } else { + payload.data = b + } + decoded, err := payload.fireOption() + if err != nil { + return payload, nil, fmt.Errorf("%w: %w", ErrEventDataEncoding, err) + } + live := make([]state.FireOption, 0, len(opts)+1) + return payload, append(append(live, opts...), decoded), nil +} + +// errorMessage returns e.Error(), or an error if it panics, as a typed nil can. +func errorMessage(e error) (msg string, err error) { + defer func() { + if p := recover(); p != nil { + err = fmt.Errorf("error payload panicked: %v", p) + } + }() + return e.Error(), nil +} + +// marshalPayload returns v's JSON form, or an error if a custom marshaler panics, so +// a caller's payload can't panic out of Fire. +func marshalPayload(v any) (b []byte, err error) { + defer func() { + if p := recover(); p != nil { + err = fmt.Errorf("payload panicked while encoding: %v", p) + } + }() + return json.Marshal(v) +} + +// eventDataTooLarge reports a payload over the Runner's cap. +func eventDataTooLarge(size, maxBytes int) error { + return fmt.Errorf("%w: payload is %d bytes, over the %d byte limit", ErrEventDataEncoding, size, maxBytes) +} + +// fireOption returns the option carrying the payload as a Fire sees it: live, and when +// a replay re-Fires the step. +func (p eventPayload) fireOption() (state.FireOption, error) { + if p.errMsg != nil { + return state.WithEventData(errors.New(*p.errMsg)), nil + } + var data any + if err := json.Unmarshal(p.data, &data); err != nil { + return nil, err + } + return state.WithEventData(data), nil +} + +// replayFireOptions returns the options that re-Fire rec's event with the payload +// the live Fire carried, or none when it carried none. +func replayFireOptions(rec *Record) ([]state.FireOption, error) { + if len(rec.EventData) == 0 && rec.EventDataError == nil { + return nil, nil + } + opt, err := eventPayload{data: rec.EventData, errMsg: rec.EventDataError}.fireOption() + if err != nil { + return nil, err + } + return []state.FireOption{opt}, nil +} + // baselineStep is the Step of the start baseline checkpoint, recorded before any // event so a freshly started instance is loadable. It sits below the first fired // step (0). diff --git a/durable/runner_options.go b/durable/runner_options.go index 7d5e4d60..a84886e4 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -22,6 +22,8 @@ type runnerConfig[S comparable, E comparable, C any] struct { checkpointEvery int // eventCodec reconstructs a recorded event for replay; it defaults to JSON. eventCodec EventCodec[E] + // maxEventData caps a WithEventData payload's JSON form, in bytes. + maxEventData int // clock is the real time seam wrapped by the recording clock on the live path // and replaced by the replay clock on recovery. It defaults to // state.SystemClock(); a test supplies a deterministic fake. @@ -145,6 +147,25 @@ func WithEffectHandler[S comparable, E comparable, C any](h EffectHandler) Optio } } +// defaultMaxEventDataBytes keeps a payload alone inside FileStore's 16 MiB journal line, +// after a Record encodes its JSON form in base64. +const defaultMaxEventDataBytes = 8 << 20 + +// WithMaxEventDataBytes sets the largest state.WithEventData payload Handle.Fire +// accepts, measured as its JSON form, or for an error, its message's. A larger one +// refuses the Fire with ErrEventDataEncoding before the kernel runs. It defaults to +// 8 MiB, which leaves a payload room in FileStore's journal line; raise it for a Store +// with no such limit. The cap covers the payload only, so a Record can still exceed a +// line with its Event, Entries or Effects, and FileStore then reports ErrRecordTooLarge +// after the kernel ran. A non-positive n is ignored. +func WithMaxEventDataBytes[S comparable, E comparable, C any](n int) Option[S, E, C] { + return func(c *runnerConfig[S, E, C]) { + if n > 0 { + c.maxEventData = n + } + } +} + // WithOwnerFence sets the fence token a Runner carries on every Append and // Checkpoint, so a Handle that has lost the instance to a newer owner gets ErrFenced // instead of committing. Pass the token the host's lease issued; a retry by the same @@ -168,6 +189,8 @@ func resolveRunner[S comparable, E comparable, C any](opts ...Option[S, E, C]) r c := runnerConfig[S, E, C]{ eventCodec: jsonEventCodec[E]{}, clock: state.SystemClock(), + + maxEventData: defaultMaxEventDataBytes, } for _, opt := range opts { opt(&c) diff --git a/durable/runner_test.go b/durable/runner_test.go index c5ea4492..a139451d 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -142,6 +142,17 @@ func recoveredSnapshotBytes[C any](t *testing.T, m *state.Machine[string, string return b } +// handleSnapshot returns the marshaled snapshot of a durable Handle's instance, so +// a live Handle and a recovered one can be compared byte-for-byte. +func handleSnapshot[C any](t *testing.T, h *durable.Handle[string, string, C]) string { + t.Helper() + b, err := state.MarshalSnapshot(h.Instance().Snapshot()) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return string(b) +} + // TestRecover_ByteIdenticalAcrossFixtures is the acceptance gate: a recovered // instance's marshaled snapshot is byte-identical to a never-durable live run // across linear, branching, context-mutating, and parallel machines. @@ -465,6 +476,23 @@ func TestRunner_Start_RejectsDuplicate(t *testing.T) { } } +// quantityMachine folds a quantity carried as event data into Count. The quantity +// is a float64 because event data crosses the journal as JSON (doc.go, Caveats). +func quantityMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("quantity"). + Reducer("addQuantity", func(in state.AssignCtx[*runCtx]) *runCtx { + c := *in.Entity + if q, ok := in.Event.(float64); ok { + c.Count += int(q) + } + return &c + }). + State("open"). + Initial("open"). + Transition("open").On("add").GoTo("open").Assign("addQuantity"). + Quench() +} + // TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory proves // two Handles on one instance (a stalled worker, and the worker that recovered the // instance after it) cannot both have their Fires acknowledged while the journal @@ -505,3 +533,27 @@ func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t * t.Fatalf("stale Handle's Fires were acknowledged at %q, but recovery yields %q", stale.Instance().Current(), got) } } + +// TestRecover_FireWithEventData_MatchesLive proves an instance driven with +// state.WithEventData recovers to the live state, because the Handle journals the data. +func TestRecover_FireWithEventData_MatchesLive(t *testing.T) { + ctx := context.Background() + m := quantityMachine() + store := durable.NewMemStore() + id := durable.InstanceID("qty-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("open")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "add", state.WithEventData(3.0)); err != nil { + t.Fatalf("Fire: %v", err) + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if live, got := handleSnapshot(t, h), handleSnapshot(t, recovered); got != live { + t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) + } +} diff --git a/durable/store.go b/durable/store.go index 13a35a03..1ae61605 100644 --- a/durable/store.go +++ b/durable/store.go @@ -40,6 +40,16 @@ type Record struct { // additive EventPayload contract). Event []byte + // EventData is the JSON form of the payload the Fire at this Step carried + // through state.WithEventData, or empty when it carried none. The live Fire and + // a replay both supply the decoded form, so the transition's Assign sees the same + // value: a struct as a map and an integer as a float64. + EventData []byte `json:",omitempty"` + + // EventDataError is the message of an error payload, which has no useful JSON + // form. The live Fire and a replay both supply an error with this message. + EventDataError *string `json:",omitempty"` + // Entries are the nondeterministic results recorded during this step — the // invoked-service done/error payloads, actor messages, clock reads, and // randomness draws the kernel consumed — in the order they resolved. Replay @@ -97,7 +107,8 @@ type Record struct { // Step survives a Checkpoint that compacts that Step away, so a delayed retry of an // already-checkpointed Step is still recognized rather than rejected as out of // order, and a different Record there is still a conflict. Two Records are the -// same when every field matches, with a nil and an empty field counting as equal. +// same when every field matches, with a nil and an empty field counting as equal, +// except EventDataError, where nil means no error and an empty message is an error. // An Append carrying WithIdempotencyKey (a backend reads it with // AppendIdempotencyKey) dedups on that key alone, so the caller vouches for the // Record: no comparison is made. Keys starting "step:" are reserved for the Store's @@ -129,6 +140,10 @@ type Record struct { // is the full Record history. For an unknown instance, Load reports // ErrInstanceNotFound. // +// Fields: a Store keeps every Record field, including EventData and EventDataError. A +// Store that drops one of those loses that payload on recovery. Snapshot is the +// exception: recovery reads the checkpoint, so a Store may drop a tail Record's copy. +// // All methods are safe for concurrent use by multiple goroutines. type Store interface { // Append persists rec for the instance and returns its monotonic per-instance @@ -137,7 +152,8 @@ type Store interface { // different Record there is rejected with ErrStepConflict. Records must be // appended in increasing Step order; an out-of-order Step is rejected with // ErrStepOutOfOrder. A fence token below the highest seen is rejected with - // ErrFenced. + // ErrFenced. A Store with a size limit reports a Record over it, as FileStore does + // with ErrRecordTooLarge. Append(ctx context.Context, id InstanceID, rec Record, opts ...AppendOption) (seq int64, err error) // Load returns the instance's latest checkpoint Snapshot bytes (nil if the diff --git a/durable/timetravel.go b/durable/timetravel.go index 9bec1218..00cf8932 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -264,7 +264,11 @@ func replayThrough[S comparable, E comparable, C any]( if err != nil { return fmt.Errorf("decoding recorded event at step %d: %w", rec.Step, err) } - res := inst.Fire(ctx, event) + fireOpts, err := replayFireOptions(rec) + if err != nil { + return fmt.Errorf("decoding recorded event data at step %d: %w", rec.Step, err) + } + res := inst.Fire(ctx, event, fireOpts...) if res.Err != nil { return fmt.Errorf("replaying step %d: %w", rec.Step, res.Err) } diff --git a/state/CHANGELOG.md b/state/CHANGELOG.md index aca9a773..e4894a80 100644 --- a/state/CHANGELOG.md +++ b/state/CHANGELOG.md @@ -11,6 +11,11 @@ counts as an additive (minor) versus breaking (major) change. Use the ## [Unreleased] +### Added + +- **`FireEventData`** reports the payload a `WithEventData` option carries, so a host + that records a Fire can persist it. It's additive, so the next release is v1.1.0. + ### Changed - Requires Go 1.27.3. diff --git a/state/fire_event_data_test.go b/state/fire_event_data_test.go new file mode 100644 index 00000000..165015e6 --- /dev/null +++ b/state/fire_event_data_test.go @@ -0,0 +1,30 @@ +package state_test + +import ( + "testing" + + "github.com/stablekernel/crucible/state" +) + +func TestFireEventData(t *testing.T) { + tests := []struct { + name string + opts []state.FireOption + wantData any + wantOK bool + }{ + {"no options", nil, nil, false}, + {"no payload option", []state.FireOption{state.CollectAll()}, nil, false}, + {"a payload", []state.FireOption{state.WithEventData(3)}, 3, true}, + {"a nil payload is still set", []state.FireOption{state.WithEventData(nil)}, nil, true}, + {"the last payload wins", []state.FireOption{state.WithEventData(1), state.WithEventData(2)}, 2, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + data, ok := state.FireEventData(tt.opts...) + if data != tt.wantData || ok != tt.wantOK { + t.Fatalf("FireEventData = (%v, %v), want (%v, %v)", data, ok, tt.wantData, tt.wantOK) + } + }) + } +} diff --git a/state/options.go b/state/options.go index 3a73846d..c6a5a074 100644 --- a/state/options.go +++ b/state/options.go @@ -282,6 +282,18 @@ func WithEventData(data any) FireOption { } } +// FireEventData reports the payload a WithEventData option in opts carries, and +// whether one is set, without firing anything. A host that records a Fire, such as +// the durable Runner, uses it to persist the payload so a replay can supply it again. +// When more than one option sets a payload the last wins, matching Fire. +func FireEventData(opts ...FireOption) (any, bool) { + cfg := fireConfig{} + for _, o := range opts { + o(&cfg) + } + return cfg.eventData, cfg.hasData +} + // VerifyOption configures Verify. type VerifyOption func(*verifyConfig)