From 27c5701591eef1a1f826f428fd65d060054819a8 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 07:56:50 -0700 Subject: [PATCH 1/3] fix(durable): journal WithEventData payloads for recovery WithEventData payloads are recorded and supplied again by Recover and StateAt, through the additive state.FireEventData. The live Fire now uses the decoded payload too, so the live run and a replay can't differ. An error payload is rebuilt from its message. A payload that can't be recorded, or is over WithMaxEventDataBytes, refuses the Fire before the kernel runs. FileStore refuses a Record too large for it to read back. Relates to #214. Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 12 +- durable/CHANGELOG.md | 28 +- durable/README.md | 12 +- durable/doc.go | 9 +- durable/errors.go | 12 + durable/event_data_test.go | 402 ++++++++++++++++++++++ durable/fence_test.go | 2 +- durable/filestore.go | 28 +- durable/memstore.go | 7 + durable/runner.go | 139 +++++++- durable/runner_options.go | 23 ++ durable/runner_test.go | 52 +++ durable/store.go | 20 +- durable/timetravel.go | 6 +- state/CHANGELOG.md | 5 + state/fire_event_data_test.go | 30 ++ state/options.go | 12 + 17 files changed, 768 insertions(+), 31 deletions(-) create mode 100644 durable/event_data_test.go create mode 100644 state/fire_event_data_test.go 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..283eb638 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -38,6 +38,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("crucible/durable: event data can't be recorded") + + // 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("crucible/durable: record too large for the journal") + // 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..413e3d59 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("crucible/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..764735b7 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) (data any, ok bool) { + cfg := fireConfig{} + for _, o := range opts { + o(&cfg) + } + return cfg.eventData, cfg.hasData +} + // VerifyOption configures Verify. type VerifyOption func(*verifyConfig) From 4ac4984a6ca6dd1b410a0a199f471d2c74c23187 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 08:56:14 -0700 Subject: [PATCH 2/3] fix(durable): carry the next step through a checkpoint Recovery derived the next step from the restored Trace count when a checkpoint had compacted the tail. A tick barrier consumes an ordinal but leaves no Trace, so the count fell short and the next Fire reused a recorded step. The checkpoint envelope now records the step that follows it. A checkpoint written before this change keeps the old fallback. Relates to #214. Signed-off-by: Ray Karnes --- durable/CHANGELOG.md | 10 +- durable/checkpoint.go | 32 ++++-- durable/checkpoint_internal_test.go | 27 ++++- durable/runner.go | 35 +++--- durable/runner_test.go | 168 ++++++++++++++++++++++++++++ durable/store.go | 14 ++- durable/timetravel.go | 2 +- state/options.go | 4 +- state/snapshot.go | 4 +- 9 files changed, 252 insertions(+), 44 deletions(-) diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index a4e26561..543dc021 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -54,11 +54,17 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. The same record is still an idempotent no-op, and an Append with `WithIdempotencyKey` still dedups on the key alone. A `FileStore` ledger line from before this change has no digest, so its step keeps the old no-op behavior. A - recovered instance that reuses a step number now fails with `ErrStepConflict` - instead of silently dropping the step. + recovered instance that reuses a step number with a different record now fails with + `ErrStepConflict` instead of silently dropping the step. ### Fixed +- **The next step survives a checkpoint that compacts tick barriers.** A checkpoint + now records the step that follows it, in a `nextStep` field of its envelope. Recovery + used the Trace count, which runs low past a tick barrier, so the next Fire could reuse a + recorded step, and high past a multi-Trace actor delivery, which skipped ordinals. A + checkpoint written before this change can still reuse a step: a different record fails + with `ErrStepConflict`, and an identical one is acknowledged without a new Record. - **`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 diff --git a/durable/checkpoint.go b/durable/checkpoint.go index a3a12b6a..c65eb830 100644 --- a/durable/checkpoint.go +++ b/durable/checkpoint.go @@ -61,8 +61,8 @@ type pendingTimerWire struct { } // checkpointEnvelope is the durable layer's wrapper around the bytes a Store keeps -// for a checkpoint: the opaque state.Snapshot bytes plus the pending-timer table -// armed at the checkpoint instant. The durable layer fully owns the checkpoint +// for a checkpoint: the opaque state.Snapshot bytes, the pending-timer table armed at +// the checkpoint instant, and the next step ordinal. The durable layer fully owns the checkpoint // bytes (it both marshals them at Checkpoint and unmarshals them at recover), so // wrapping the kernel snapshot in this envelope is additive and invisible to the // Store, which treats checkpoint bytes as opaque. A checkpoint with no armed timer @@ -77,14 +77,21 @@ type checkpointEnvelope struct { // recovery whose compacted tail no longer carries the arming ScheduleAfter can // still re-arm them at their recorded instants. Timers []pendingTimerWire `json:"timers,omitempty"` + // NextStep is the step ordinal the first Record after this checkpoint takes. A + // recovery whose compacted tail is empty has no Record to derive it from, and the + // restored Trace count doesn't match the ordinals: a tick barrier consumes one and + // leaves no Trace, and an actor delivery can leave several Traces on one ordinal. + // Zero means none was recorded: a Start baseline, or a checkpoint from before this + // field. + NextStep int `json:"nextStep,omitempty"` } // marshalCheckpoint wraps the marshaled kernel snapshot and the armed pending // timers into a checkpoint envelope's bytes. Timers are emitted in scheduleID // order so the persisted form is deterministic regardless of the map iteration -// order they were collected in. -func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, error) { - env := checkpointEnvelope{Snapshot: snap} +// order they were collected in. nextStep is the ordinal that follows the checkpoint. +func marshalCheckpoint(snap []byte, timers map[string]pendingTimer, nextStep int) ([]byte, error) { + env := checkpointEnvelope{Snapshot: snap, NextStep: nextStep} if len(timers) > 0 { ids := make([]string, 0, len(timers)) for id := range timers { @@ -113,18 +120,19 @@ func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, err } // unmarshalCheckpoint splits a checkpoint envelope's bytes back into the kernel -// snapshot bytes and the persisted pending-timer wires. It tolerates a bare -// (pre-envelope) snapshot: bytes that are not a checkpoint envelope are returned -// as the snapshot with no timers, so an older checkpoint still restores. A -// checkpoint envelope is distinguished by carrying a non-empty "snapshot" member. -func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, error) { +// snapshot bytes, the persisted pending-timer wires and the next step ordinal. It +// tolerates a bare (pre-envelope) snapshot: bytes that are not a checkpoint envelope +// are returned as the snapshot with no timers and a zero ordinal, so an older +// checkpoint still restores. A checkpoint envelope is distinguished by carrying a +// non-empty "snapshot" member. +func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, int, error) { var env checkpointEnvelope if err := json.Unmarshal(b, &env); err != nil || len(env.Snapshot) == 0 { // Not an envelope (or an envelope without the wrapped snapshot member): treat // the bytes as a bare kernel snapshot, the pre-timer-survival checkpoint form. - return b, nil, nil + return b, nil, 0, nil } - return env.Snapshot, env.Timers, nil + return env.Snapshot, env.Timers, env.NextStep, nil } // decodeTimerEvent decodes a persisted timer event wire back into the machine's diff --git a/durable/checkpoint_internal_test.go b/durable/checkpoint_internal_test.go index 45f61931..5ae66b4f 100644 --- a/durable/checkpoint_internal_test.go +++ b/durable/checkpoint_internal_test.go @@ -17,11 +17,11 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) { timers := map[string]pendingTimer{ "m:a:after:0": {id: "m:a:after:0", due: due, event: "elapsed"}, } - b, err := marshalCheckpoint(snap, timers) + b, err := marshalCheckpoint(snap, timers, 0) if err != nil { t.Fatalf("marshalCheckpoint: %v", err) } - gotSnap, gotWires, err := unmarshalCheckpoint(b) + gotSnap, gotWires, _, err := unmarshalCheckpoint(b) if err != nil { t.Fatalf("unmarshalCheckpoint: %v", err) } @@ -45,7 +45,7 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) { // armed, so an event-driven checkpoint's envelope stays minimal. func TestMarshalCheckpoint_NoTimers(t *testing.T) { snap := []byte(`{"machine":"m"}`) - b, err := marshalCheckpoint(snap, nil) + b, err := marshalCheckpoint(snap, nil, 0) if err != nil { t.Fatalf("marshalCheckpoint: %v", err) } @@ -64,7 +64,7 @@ func TestMarshalCheckpoint_NoTimers(t *testing.T) { // checkpoint still restores. func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) { bare := []byte(`{"machine":"m","current":"a","configuration":["a"]}`) - snap, wires, err := unmarshalCheckpoint(bare) + snap, wires, _, err := unmarshalCheckpoint(bare) if err != nil { t.Fatalf("unmarshalCheckpoint: %v", err) } @@ -81,7 +81,7 @@ func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) { // any real decode failure when it unmarshals the kernel snapshot). func TestUnmarshalCheckpoint_Garbage(t *testing.T) { garbage := []byte(`not json`) - snap, wires, err := unmarshalCheckpoint(garbage) + snap, wires, _, err := unmarshalCheckpoint(garbage) if err != nil { t.Fatalf("unmarshalCheckpoint should tolerate garbage: %v", err) } @@ -155,7 +155,7 @@ func TestMarshalCheckpoint_BadEventErrors(t *testing.T) { timers := map[string]pendingTimer{ "bad": {id: "bad", due: time.Unix(0, 1), event: make(chan int)}, } - if _, err := marshalCheckpoint([]byte(`{}`), timers); err == nil { + if _, err := marshalCheckpoint([]byte(`{}`), timers, 0); err == nil { t.Fatal("want encode error for unmarshalable timer event, got nil") } } @@ -174,3 +174,18 @@ func TestTickInstant(t *testing.T) { t.Fatalf("tickInstant: ok=%v got=%d", ok, got.UnixNano()) } } + +// TestCheckpoint_NextStepRoundTrips proves the envelope carries the ordinal after +// the checkpoint, and that a checkpoint without one reports zero, the old form. +func TestCheckpoint_NextStepRoundTrips(t *testing.T) { + b, err := marshalCheckpoint([]byte(`{"v":1}`), nil, 9) + if err != nil { + t.Fatalf("marshalCheckpoint: %v", err) + } + if _, _, next, err := unmarshalCheckpoint(b); err != nil || next != 9 { + t.Fatalf("unmarshalCheckpoint = (%d, %v), want (9, nil)", next, err) + } + if _, _, next, err := unmarshalCheckpoint([]byte(`{"snapshot":{"v":1}}`)); err != nil || next != 0 { + t.Fatalf("old envelope = (%d, %v), want (0, nil)", next, err) + } +} diff --git a/durable/runner.go b/durable/runner.go index 9e1aa048..e77e5a11 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -26,7 +26,7 @@ import ( // (configuration, context, event payload, machine definition). The only input a // 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, +// step's 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 @@ -252,8 +252,8 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt buf := make([]state.JournalEntry, 0) recClock := newRecordingClock(r.cfg.clock, &buf) // Force unbounded full-trace history: the runner journals each step's Trace - // (notably the EventPayload) and derives both the step ordinal and the checkpoint - // snapshots from the instance's retained Trace history, so a recover can replay it + // (notably the EventPayload) and derives the checkpoint snapshots from the + // instance's retained Trace history, so a recover can replay it // byte-identically. WithUnboundedHistory elevates to full trace and retains every // step; it is prepended so a caller's opts cannot downgrade it. castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock), state.WithUnboundedHistory[S]()}, opts...) @@ -269,7 +269,7 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt } // Wrap the baseline snapshot in a checkpoint envelope (no timers armed yet) so // every checkpoint a recover loads carries the same durable envelope shape. - baseEnv, err := marshalCheckpoint(snap, nil) + baseEnv, err := marshalCheckpoint(snap, nil, 0) if err != nil { return nil, fmt.Errorf("durable: marshaling start checkpoint envelope for %q: %w", id, err) } @@ -354,7 +354,7 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // Fire drives one event through the Handle's live instance and records the step: // it Fires the kernel, appends a Record carrying the driving event at the -// produced Trace ordinal (write-ahead, before returning), and — when the +// step's ordinal (write-ahead, before returning), and — when the // 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. @@ -487,7 +487,7 @@ func (h *Handle[S, E, C]) appendStep(ctx context.Context, step int, rec *Record) // Wrap the kernel snapshot with the absolute deadlines of the timers armed at // this checkpoint, so a recovery whose compacted tail no longer carries their // arming ScheduleAfter can re-arm them at their recorded instants. - env, err := marshalCheckpoint(snap, h.timers) + env, err := marshalCheckpoint(snap, h.timers, rec.followingStep()) if err != nil { return fmt.Errorf("durable: marshaling checkpoint envelope at step %d for %q: %w", step, h.id, err) } @@ -537,10 +537,10 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S return nil, fmt.Errorf("durable: instance %q has no checkpoint to restore", id) } - // Unwrap the durable checkpoint envelope: the kernel snapshot bytes plus the - // absolute deadlines of the timers that were armed at the checkpoint. A bare + // Unwrap the durable checkpoint envelope: the kernel snapshot bytes, the absolute + // deadlines of the timers armed at the checkpoint, and the next step ordinal. A bare // (pre-envelope) checkpoint yields the snapshot with no persisted timers. - kernelSnap, persistedTimers, err := unmarshalCheckpoint(snapBytes) + kernelSnap, persistedTimers, checkpointNext, err := unmarshalCheckpoint(snapBytes) if err != nil { return nil, fmt.Errorf("durable: unmarshaling checkpoint envelope for %q: %w", id, err) } @@ -716,15 +716,16 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S // scheduler-tick barrier, which consumed one ordinal per timer it fired plus the // barrier's own. The next ordinal is therefore the last tail Record's Step plus // its own span. When the tail is empty (a fresh checkpoint compacted every - // Record), no actor or tick step survives in it, so the restored Trace count — - // one per pre-checkpoint step — gives the next ordinal directly. - next := len(inst.History()) + // Record), the checkpoint carries the next ordinal. A Start baseline carries none and + // has no Traces, so its count of zero is right. A checkpoint from before this field + // also falls back to the Trace count, which can be off (see + // checkpointEnvelope.NextStep). + next := checkpointNext + if next == 0 { + next = len(inst.History()) + } if n := len(tail); n > 0 { - last := &tail[n-1] - next = last.Step + 1 - if last.Tick { - next = last.Step + last.TickSteps + 1 - } + next = tail[n-1].followingStep() } return &Handle[S, E, C]{ runner: r, diff --git a/durable/runner_test.go b/durable/runner_test.go index a139451d..2d33819a 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -476,6 +477,23 @@ func TestRunner_Start_RejectsDuplicate(t *testing.T) { } } +// pollingMachine re-arms a one-second timer every time it fires, so a host timer +// loop keeps producing steps, while `poke` bumps Count on the same state. +func pollingMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("polling"). + Action("bump", func(c state.ActionCtx[*runCtx]) (state.Effect, error) { + c.Entity.Count++ + return nil, nil + }). + State("idle"). + State("polling"). + Initial("idle"). + Transition("idle").On("start").GoTo("polling"). + Transition("polling").On("poke").GoTo("polling").Do("bump"). + Transition("polling").After(time.Second).On("tick").GoTo("polling").Reenter(). + Quench() +} + // 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] { @@ -557,3 +575,153 @@ func TestRecover_FireWithEventData_MatchesLive(t *testing.T) { t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) } } + +// TestRecover_StepAfterTickCheckpoint_IsPersisted proves the next step ordinal +// survives a checkpoint that compacts tick barriers: a Fire after recovery must +// not reuse a recorded step, which the Store would drop or refuse. +func TestRecover_StepAfterTickCheckpoint_IsPersisted(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("poll-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{ + durable.WithRunnerClock[string, string, *runCtx](clk), + durable.WithCheckpointEvery[string, string, *runCtx](8), + } + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + // Four timer-firing ticks put barriers on steps 1, 3, 5 and 7; the checkpoint + // at step 7 compacts the whole tail. + for range 4 { + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + + _, tail, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(tail) != 0 { + t.Fatalf("tail holds %d records, want the checkpoint to have compacted them", len(tail)) + } + + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover(again): %v", err) + } + if live, got := handleSnapshot(t, resumed), handleSnapshot(t, recovered); got != live { + t.Fatalf("the acknowledged poke was not persisted:\n live: %s\n got: %s", live, got) + } +} + +// lastStep returns the Step of the newest Record in a history-keeping store. +func lastStep(t *testing.T, store *durable.MemStore, id durable.InstanceID) int { + t.Helper() + _, records, err := store.History(context.Background(), id) + if err != nil { + t.Fatalf("History: %v", err) + } + if len(records) == 0 { + t.Fatal("no records") + } + return records[len(records)-1].Step +} + +// TestRecover_NextStep_IsTheLiveOrdinal proves the ordinal after a compacted tail is +// exactly the one a live run would take, not merely one past the journal. +func TestRecover_NextStep_IsTheLiveOrdinal(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore(durable.WithHistory()) + id := durable.InstanceID("poll-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{ + durable.WithRunnerClock[string, string, *runCtx](clk), + durable.WithCheckpointEvery[string, string, *runCtx](8), + } + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + for range 4 { // barriers on steps 1, 3, 5 and 7, each spanning two ordinals + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + } + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + if got := lastStep(t, store, id); got != 9 { + t.Fatalf("poke recorded at step %d, want 9", got) + } +} + +// oldEnvelopeStore serves checkpoints as a version before nextStep wrote them. +type oldEnvelopeStore struct{ *durable.MemStore } + +func (s oldEnvelopeStore) Load(ctx context.Context, id durable.InstanceID) ([]byte, []durable.Record, error) { + snap, tail, err := s.MemStore.Load(ctx, id) + if err != nil { + return nil, nil, err + } + var env map[string]json.RawMessage + if err = json.Unmarshal(snap, &env); err != nil { + return nil, nil, err + } + delete(env, "nextStep") + snap, err = json.Marshal(env) + return snap, tail, err +} + +// TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount proves a checkpoint from before +// the field still recovers, from the restored Trace count. +func TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + mem := durable.NewMemStore(durable.WithHistory()) + store := oldEnvelopeStore{mem} + id := durable.InstanceID("poll-1") + opts := []durable.Option[string, string, *runCtx]{durable.WithCheckpointEvery[string, string, *runCtx](3)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + for _, ev := range []string{"start", "poke", "poke"} { // the checkpoint compacts all three + if _, err = h.Fire(ctx, ev); err != nil { + t.Fatalf("Fire(%s): %v", ev, err) + } + } + resumed, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = resumed.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire(poke): %v", err) + } + if got := lastStep(t, mem, id); got != 3 { + t.Fatalf("poke recorded at step %d, want 3", got) + } +} diff --git a/durable/store.go b/durable/store.go index 1ae61605..31b12a79 100644 --- a/durable/store.go +++ b/durable/store.go @@ -24,8 +24,9 @@ type InstanceID string // after an upgrade reads as a conflict. It must also be copied in cloneRecord, or the // digest ignores it. The same applies to state.JournalEntry and state.EffectEnvelope. type Record struct { - // Step is the Fire ordinal this Record was produced at, indexing the - // instance's recorded Traces. It is the per-instance idempotency key: a + // Step is the ordinal this Record was produced at. Ordinals rise with each Record, + // and a tick barrier consumes one per timer it fired plus its own, so they don't + // index the instance's Traces. It is the per-instance idempotency key: a // Store treats a second Append of the same Record at the same Step as // already-applied. Step int @@ -84,6 +85,15 @@ type Record struct { TickSteps int } +// followingStep returns the step ordinal after r. A tick barrier consumes one +// ordinal per timer it fired plus its own; every other Record consumes one. +func (r *Record) followingStep() int { + if r.Tick { + return r.Step + r.TickSteps + 1 + } + return r.Step + 1 +} + // Store is the durable-execution persistence seam. It records an instance's // per-step Records and periodic Snapshot checkpoints, and reconstructs the // instance from the latest checkpoint plus the tail of Records appended after diff --git a/durable/timetravel.go b/durable/timetravel.go index 00cf8932..4d5be559 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -202,7 +202,7 @@ func restoreBaseline[S comparable, E comparable, C any]( baseline []byte, repClock state.Clock, ) (*state.Instance[S, E, C], error) { - kernelSnap, _, err := unmarshalCheckpoint(baseline) + kernelSnap, _, _, err := unmarshalCheckpoint(baseline) if err != nil { return nil, fmt.Errorf("unmarshaling baseline envelope: %w", err) } diff --git a/state/options.go b/state/options.go index 764735b7..8ff02f2a 100644 --- a/state/options.go +++ b/state/options.go @@ -378,8 +378,8 @@ func WithRestoreFullTrace[S comparable]() RestoreOption[S] { // WithRestoreUnboundedHistory restores an instance in full trace mode with // unbounded history retention, mirroring WithUnboundedHistory at Cast. The durable -// runner uses it on recover: it reconstructs the step ordinal and checkpoint -// snapshots from the retained Trace history, so post-recovery Fires must keep +// runner uses it on recover: it reconstructs checkpoint snapshots from the retained +// Trace history, so post-recovery Fires must keep // retaining every trace exactly as a freshly started durable instance does. func WithRestoreUnboundedHistory[S comparable]() RestoreOption[S] { return func(cfg *restoreConfig[S]) { diff --git a/state/snapshot.go b/state/snapshot.go index 4bbf7119..4bcbf732 100644 --- a/state/snapshot.go +++ b/state/snapshot.go @@ -227,8 +227,8 @@ const ( // Clock.Now() reads, and host randomness — each correlated by a stable id reused // from the effect that armed it (invocationID / actorInvocationID / scheduleID). type JournalEntry struct { - // Step is the Fire ordinal the result resolved at, indexing the instance's - // recorded Traces, so replay applies the recorded value at the right step. + // Step is the ordinal the result resolved at, so replay applies the recorded value + // at the right step. Step int `json:"step"` // Kind classifies which nondeterministic source produced the result. Kind JournalKind `json:"kind"` From 67d9182fcb7663a9939b1d73c2888697b84d3622 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 09:00:40 -0700 Subject: [PATCH 3/3] fix(durable): lock the Handle and refuse steps after a failed Append Fire, Tick, RunService and DeliverToActor now hold a lock for the whole step, so a host's timer loop no longer loses acknowledged steps. A step that applied to the kernel but wasn't appended leaves the instance ahead of the journal, so the Handle refuses further steps with ErrHandleFailed until it's recovered. Relates to #214. Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 10 + durable/CHANGELOG.md | 16 ++ durable/README.md | 10 + durable/actor.go | 20 +- durable/errors.go | 9 +- durable/runner.go | 85 ++++++-- durable/runner_test.go | 244 ++++++++++++++++++++++ durable/service.go | 20 +- 8 files changed, 390 insertions(+), 24 deletions(-) diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 8dfd056b..4680e2a8 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -125,6 +125,16 @@ Time-travel needs the full record history through the target step. A `Store` opt in by implementing `HistoryStore` (the in-tree `MemStore` does so under `WithHistory`); `StateAt` otherwise falls back to the latest checkpoint plus tail. +## Handles across goroutines + +`Fire`, `Tick`, `RunService` and `DeliverToActor` are safe to call from different +goroutines: each holds the Handle's lock for the whole step. A host callback must not +call back into its Handle. `Instance()` is outside the lock. + +A step that may not have been recorded fails the Handle with `ErrHandleFailed`. `Recover` +a new Handle with the same options to continue, unless the cause is `ErrFenced`: then a +newer owner holds the instance, so stop. + ## A note on serialized payloads Events, service done-data, actor done-data, and actor messages are recorded as their diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 543dc021..5067e0d0 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -25,9 +25,23 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. a refused payload, and `ErrRecordTooLarge` a Record too large for a `FileStore` line. `Record.EventData` and `Record.EventDataError` hold the payload. +- **`ErrHandleFailed`** reports a Handle out of step with its journal. + ### Changed - Requires Go 1.27.3. +- **A Handle's steps are serialized across goroutines.** `Fire`, `Tick`, `RunService` and + `DeliverToActor` hold a lock for the whole step, so a host's timer loop no longer loses + acknowledged steps. A host callback must not call back into its Handle, or it + deadlocks. `Instance()` isn't covered: read it only while no step is running. A `Tick`, + `RunService` or `DeliverToActor` whose context ended while it waited returns that error + and changes nothing. +- **A step that may not have been recorded fails the Handle.** That call and every later + one return `ErrHandleFailed`, wrapping the cause. `Recover` a new Handle with the same + options to continue. A service whose result wasn't recorded is in flight again, so + `RunService` runs it a second time. An `InvokeActor` actor restarts from its initial + state, and a `Spawn` actor doesn't return. If the cause is `ErrFenced`, a newer owner + holds the instance: stop. - **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 @@ -59,6 +73,8 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### Fixed +- **A failed checkpoint no longer makes the next step conflict.** `Tick`, `RunService` + and `DeliverToActor` now advance the next step as soon as the append lands. - **The next step survives a checkpoint that compacts tick barriers.** A checkpoint now records the step that follows it, in a `nextStep` field of its envelope. Recovery used the Trace count, which runs low past a tick barrier, so the next Fire could reuse a diff --git a/durable/README.md b/durable/README.md index 9b9858c6..f720c52f 100644 --- a/durable/README.md +++ b/durable/README.md @@ -136,6 +136,16 @@ opts in by implementing `HistoryStore` (the in-tree `MemStore` does so under `WithHistory`); `StateAt` falls back to `Load` (latest checkpoint plus tail) otherwise. +## Handles across goroutines + +`Fire`, `Tick`, `RunService` and `DeliverToActor` are safe to call from different +goroutines: each holds the Handle's lock for the whole step. A host callback must not +call back into its Handle. `Instance()` is outside the lock. + +A step that may not have been recorded fails the Handle with `ErrHandleFailed`. `Recover` +a new Handle with the same options to continue, unless the cause is `ErrFenced`: then a +newer owner holds the instance, so stop. + ## Caveat: serialized payloads Events, service done-data, actor done-data, and actor messages are recorded as their diff --git a/durable/actor.go b/durable/actor.go index 792ba00d..41abf380 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -99,10 +99,21 @@ type actorMessagePayload struct { // re-instantiating the actor. // // It returns whether the actor was found running (false for a ref that names no -// live actor — already settled, stopped, or never spawned), and a persistence -// failure as the error. The kernel transitions the delivery drove do not -// themselves fail. +// live actor — already settled, stopped, or never spawned — or with an error when the +// Handle has failed or ctx has ended), and a persistence failure as the error. The +// kernel transitions the delivery drove do not themselves fail. After a failed Append +// the result wasn't recorded: on a recovered Handle an InvokeActor actor restarts from its +// initial state, and a Spawn actor doesn't return. Resending a recorded delivery repeats +// any message it sends the parent. func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef, event any) (bool, error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return false, err + } + if err := ctx.Err(); err != nil { + return false, err + } if h.actors == nil { return false, fmt.Errorf("durable: no actor palette wired for %q (supply WithActorPalette)", h.id) } @@ -116,7 +127,7 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef output, settled, settleErr := h.actorOutcome() entries, err := recordActorTransitions(ref.ID, h.inst.History(), before, output, settled, settleErr) if err != nil { - return true, fmt.Errorf("durable: recording actor %q result for %q: %w", ref.ID, h.id, err) + return true, h.fail(fmt.Errorf("durable: recording actor %q result for %q: %w", ref.ID, h.id, err)) } if h.actorBuf != nil { *h.actorBuf = append(*h.actorBuf, entries...) @@ -127,7 +138,6 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef if err := h.persistStep(ctx, step, &rec); err != nil { return true, err } - h.nextStep++ return true, nil } diff --git a/durable/errors.go b/durable/errors.go index 283eb638..47167b76 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -47,9 +47,16 @@ var ( // 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. + // applied the step, so the Handle fails with ErrHandleFailed. ErrRecordTooLarge = errors.New("crucible/durable: record too large for the journal") + // ErrHandleFailed is reported by a Handle's Fire, Tick, RunService and + // DeliverToActor once a step applied to the live instance but may not have been + // recorded, on that call and every later one. It wraps the cause. Recover a new + // Handle with the same options to continue, unless the cause is ErrFenced: then a + // newer owner holds the instance, so stop. + ErrHandleFailed = errors.New("crucible/durable: handle is out of step with the journal") + // 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/runner.go b/durable/runner.go index e77e5a11..442fec6f 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sync" "time" "github.com/stablekernel/crucible/state" @@ -76,11 +77,33 @@ func NewRunner[S comparable, E comparable, C any](m *state.Machine[S, E, C], st // clock that journals every reading the scheduler consumes, so timer-driven // transitions are durable and replay wall-clock-independent. Obtain a Handle from // Runner.Start or Recover. +// +// Fire, Tick, RunService and DeliverToActor are safe to call from different +// goroutines: each holds the Handle's lock for the whole step, including the Store +// write, effect dispatch and any service it runs. Any host callback run during a step +// (an effect handler, service, guard, action, reducer or actor behavior) must not call +// back into the same Handle, or it deadlocks. A step waits for the lock without watching +// its context, so a slow service delays other steps past their deadlines; a Tick, +// RunService or DeliverToActor whose context has ended by then returns that error and +// changes nothing, and the kernel refuses a Fire. Instance returns the +// live instance outside that lock, so read it only while no step is running. A +// stateless Runner.Fire builds its own Handle, so it isn't serialized with this one. +// +// A step that applied to the kernel but may not have been recorded leaves the Handle out +// of step with the journal. That call and every later one return ErrHandleFailed, +// wrapping the cause. Recover a new Handle with the same options to continue. A service +// whose result wasn't recorded is in flight again, so RunService runs it a second time. +// A running InvokeActor actor restarts from its initial state on any Recover, losing every +// delivery so far, and a Spawn actor doesn't return. If the cause is ErrFenced, a +// newer owner holds the instance: stop, because recovering won't help. type Handle[S comparable, E comparable, C any] struct { - runner *Runner[S, E, C] - id InstanceID - inst *state.Instance[S, E, C] + runner *Runner[S, E, C] + id InstanceID + inst *state.Instance[S, E, C] + + mu sync.Mutex // serializes the steps, and guards the instance and every field below nextStep int + failed error // the error that took the Handle out of step with the journal, nil until then sched *state.Scheduler[S, E, C] recClock *recordingClock // owns the clock-read accumulator and its mutex @@ -93,6 +116,21 @@ type Handle[S comparable, E comparable, C any] struct { actorBuf *[]state.JournalEntry // recording actor-transition accumulator, flushed per step } +// refuse reports ErrHandleFailed once the Handle is out of step with the journal. +func (h *Handle[S, E, C]) refuse() error { + if h.failed == nil { + return nil + } + return fmt.Errorf("%w: %q: %w", ErrHandleFailed, h.id, h.failed) +} + +// fail records err as the reason the Handle is out of step with the journal, and +// returns the same refusal every later step gets. +func (h *Handle[S, E, C]) fail(err error) error { + h.failed = err + return h.refuse() +} + // drainClock returns and clears the clock readings accumulated since the last // drain, so each recorded step carries exactly the readings consumed during it. // The drain is taken under the recording clock's mutex, so closing a step's @@ -227,7 +265,8 @@ func (h *Handle[S, E, C]) dropFiredTimers(now time.Time) { // Instance returns the underlying kernel Instance the Handle wraps, for reads // such as Configuration, Snapshot, or Current. Drive it through the Handle's Fire // (or the Runner's) so steps continue to be recorded; firing the bare Instance -// bypasses durability. +// bypasses durability. Read it only while no step is running: it's outside the Handle's +// lock. After ErrHandleFailed it shows state the journal may not hold. func (h *Handle[S, E, C]) Instance() *state.Instance[S, E, C] { return h.inst } // ID returns the InstanceID the Handle records under. @@ -343,7 +382,8 @@ func (r *Runner[S, E, C]) newActorSystem(inst *state.Instance[S, E, C]) *state.A // 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. +// avoid reloading between steps. Calls on one id aren't serialized, and must not overlap +// each other or a live Handle. func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts ...state.FireOption) (state.FireResult[S], error) { h, err := r.recover(ctx, id) if err != nil { @@ -370,6 +410,11 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // 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) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return state.FireResult[S]{}, err + } // 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) @@ -395,7 +440,7 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO des := dispatchableEffects(step, res.Effects) effEnvs, err := recordEffects(des) if err != nil { - return res, err + return res, h.fail(err) } 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 @@ -403,7 +448,7 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO // a crash mid-dispatch leaves the effect recorded in the still-present tail for // recovery to redispatch — never orphaned behind the compaction. if err := h.appendStep(ctx, step, &rec); err != nil { - return res, err + return res, h.fail(err) } h.nextStep++ // Dispatch AFTER the append: a crash in this window leaves the effect recorded @@ -429,6 +474,16 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO // count of timers fired — so recovery re-derives the same timers at their // recorded instants. It returns the FireResults of the timers it fired, in order. func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return nil, err + } + // A context that ended in the lock queue must stop here: the scheduler would consume + // a due timer whose Fire the kernel then refuses. + if err := ctx.Err(); err != nil { + return nil, err + } results := h.sched.Tick(ctx) step := h.nextStep @@ -450,20 +505,24 @@ func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], erro if len(rec.Entries) == 0 && rec.TickSteps == 0 { return results, nil } - if err := h.persistStep(ctx, step, &rec); err != nil { + if err := h.appendStep(ctx, step, &rec); err != nil { + return results, h.fail(err) + } + h.nextStep = rec.followingStep() + if err := h.checkpointStep(ctx, step, &rec); err != nil { return results, err } - h.nextStep += len(results) + 1 return results, nil } -// persistStep checkpoints if the policy is due, then write-ahead appends rec -// before the step is acknowledged to the caller. The checkpoint is taken at the -// barrier's own step so a Load after it returns this Snapshot plus the later tail. +// persistStep write-ahead appends rec, advances the next step, then checkpoints if +// the policy is due. A failed append fails the Handle, because the kernel already +// applied the step; a failed checkpoint doesn't, because the Record is durable. func (h *Handle[S, E, C]) persistStep(ctx context.Context, step int, rec *Record) error { if err := h.appendStep(ctx, step, rec); err != nil { - return err + return h.fail(err) } + h.nextStep = rec.followingStep() return h.checkpointStep(ctx, step, rec) } diff --git a/durable/runner_test.go b/durable/runner_test.go index 2d33819a..bc5b3f7b 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -4,6 +4,10 @@ import ( "context" "encoding/json" "errors" + "runtime" + "strings" + "sync" + "sync/atomic" "testing" "time" @@ -511,6 +515,21 @@ func quantityMachine() *state.Machine[string, string, *runCtx] { Quench() } +// flakyAppendStore fails the first Append of one step, then delegates. +type flakyAppendStore struct { + durable.Store + failStep int + failed bool +} + +func (s *flakyAppendStore) Append(ctx context.Context, id durable.InstanceID, rec durable.Record, opts ...durable.AppendOption) (int64, error) { + if rec.Step == s.failStep && !s.failed { + s.failed = true + return 0, errors.New("store unavailable") + } + return s.Store.Append(ctx, id, rec, opts...) +} + // 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 @@ -552,6 +571,71 @@ func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t * } } +// TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps proves the "Tick from its +// own timer loop" pattern that drainClock documents holds while requests Fire on +// the same Handle: the recovered instance matches the live one, and every +// acknowledged Fire survives. +func TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps(t *testing.T) { + if runtime.GOMAXPROCS(0) < 2 { + t.Skip("needs two CPUs to interleave Fire and Tick") + } + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("poll-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{durable.WithRunnerClock[string, string, *runCtx](clk)} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + + const n = 200 + var acked, fired, failed atomic.Int64 + var wg sync.WaitGroup + wg.Add(2) + go func() { // the host's timer loop + defer wg.Done() + for range n { + clk.Advance(time.Second) + results, terr := h.Tick(ctx) + if terr != nil { + failed.Add(1) + } + fired.Add(int64(len(results))) + } + }() + go func() { // request traffic + defer wg.Done() + for range n { + if _, ferr := h.Fire(ctx, "poke"); ferr != nil { + failed.Add(1) + continue + } + acked.Add(1) + } + }() + wg.Wait() + if failed.Load() != 0 || acked.Load() != n || fired.Load() == 0 { + t.Fatalf("failed %d calls, acknowledged %d of %d Fires, fired %d timers", failed.Load(), acked.Load(), n, fired.Load()) + } + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got, want := recovered.Instance().Entity().Count, int(acked.Load()); got != want { + t.Fatalf("recovered %d pokes, want %d acknowledged", got, want) + } + 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) + } +} + // 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) { @@ -629,6 +713,36 @@ func TestRecover_StepAfterTickCheckpoint_IsPersisted(t *testing.T) { } } +// TestRunner_FireAfterFailedAppend_MatchesJournal proves a Fire whose Append failed +// acknowledges nothing further: the Handle refuses, and recovery matches the journal. +func TestRunner_FireAfterFailedAppend_MatchesJournal(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := &flakyAppendStore{Store: durable.NewMemStore(), failStep: 1} + id := durable.InstanceID("poll-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { // step 1: the Append fails + t.Fatalf("Fire(poke) = %v, want ErrHandleFailed", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("second Fire(poke) = %v, want ErrHandleFailed", err) + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Entity().Count; got != 0 { + t.Fatalf("recovered %d pokes, want 0: none was recorded", got) + } +} + // lastStep returns the Step of the newest Record in a history-keeping store. func lastStep(t *testing.T, store *durable.MemStore, id durable.InstanceID) int { t.Helper() @@ -725,3 +839,133 @@ func TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount(t *testing.T) { t.Fatalf("poke recorded at step %d, want 3", got) } } + +// TestHandle_FailedAppend_RefusesEveryStep proves the failed Handle stays refused, +// names the cause, and that a recovered Handle continues from the journal. +func TestHandle_FailedAppend_RefusesEveryStep(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := &flakyAppendStore{Store: durable.NewMemStore(), failStep: 1} + id := durable.InstanceID("poll-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Fire(poke) = %v, want ErrHandleFailed", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) || !strings.Contains(err.Error(), "store unavailable") { + t.Fatalf("Fire after the failure = %v, want ErrHandleFailed naming the cause", err) + } + if _, err = h.Tick(ctx); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Tick after the failure = %v, want ErrHandleFailed", err) + } + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = recovered.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire on the recovered Handle: %v", err) + } +} + +// failingCheckpointStore fails the next Checkpoint after armed is set. +type failingCheckpointStore struct { + durable.Store + armed bool +} + +func (s *failingCheckpointStore) Checkpoint(ctx context.Context, id durable.InstanceID, snapshot []byte, throughStep int, opts ...durable.CheckpointOption) error { + if s.armed { + s.armed = false + return errors.New("checkpoint unavailable") + } + return s.Store.Checkpoint(ctx, id, snapshot, throughStep, opts...) +} + +// TestHandle_FailedCheckpointOnTick_KeepsGoing proves a failed checkpoint doesn't fail +// the Handle or make the next step conflict: the Record is durable. +func TestHandle_FailedCheckpointOnTick_KeepsGoing(t *testing.T) { + ctx := context.Background() + store := &failingCheckpointStore{Store: durable.NewMemStore()} + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{ + durable.WithRunnerClock[string, string, *runCtx](clk), + durable.WithCheckpointEvery[string, string, *runCtx](2), + } + h, err := durable.NewRunner(pollingMachine(), store, opts...).Start(ctx, "poll-1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + store.armed = true + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err == nil || errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Tick = %v, want the plain checkpoint error", err) + } + if _, err = h.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire after the failed checkpoint: %v", err) + } +} + +// TestHandle_FencedOwner_Stops proves a Handle that lost its instance reports the fence +// through ErrHandleFailed, so a host can tell it must stop rather than recover. +func TestHandle_FencedOwner_Stops(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + const id = durable.InstanceID("poll-1") + owner := func(token int64) durable.Option[string, string, *runCtx] { + return durable.WithOwnerFence[string, string, *runCtx](token) + } + old, err := durable.NewRunner(m, store, owner(1)).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = old.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + newer, err := durable.Recover(ctx, m, store, id, owner(2)) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = newer.Fire(ctx, "poke"); err != nil { + t.Fatalf("newer Fire: %v", err) + } + _, err = old.Fire(ctx, "poke") + if !errors.Is(err, durable.ErrHandleFailed) || !errors.Is(err, durable.ErrFenced) { + t.Fatalf("old Fire = %v, want ErrHandleFailed wrapping ErrFenced", err) + } +} + +// TestHandle_ContextEndedBeforeTheStep_ChangesNothing proves an ended context stops a +// Tick before the scheduler consumes a due timer. +func TestHandle_ContextEndedBeforeTheStep_ChangesNothing(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{durable.WithRunnerClock[string, string, *runCtx](clk)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, "poll-1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + clk.Advance(time.Second) + ended, cancel := context.WithCancel(ctx) + cancel() + if _, err = h.Tick(ended); !errors.Is(err, context.Canceled) { + t.Fatalf("Tick = %v, want context.Canceled", err) + } + results, err := h.Tick(ctx) + if err != nil || len(results) != 1 { + t.Fatalf("Tick after = (%d results, %v), want the timer still due", len(results), err) + } +} diff --git a/durable/service.go b/durable/service.go index fedfe318..aacb91bc 100644 --- a/durable/service.go +++ b/durable/service.go @@ -84,10 +84,21 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { // re-invoking the service. // // It returns the routed FireResult and true, or the zero result and false when id -// names no in-flight service (already settled, stopped, or never started). A -// persistence failure is returned as the error; the kernel transition itself does -// not fail (a settled service always routes onDone or onError). +// names no in-flight service (already settled, stopped, or never started), or false +// and the error when the Handle has failed or ctx has ended. A persistence failure is +// returned as the error; the kernel transition itself does not fail (a settled service +// always routes onDone or onError). After a failed Append the result wasn't recorded, so +// a recovered Handle has the service in flight again, and RunService runs it a second +// time. func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.FireResult[S], bool, error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return state.FireResult[S]{}, false, err + } + if err := ctx.Err(); err != nil { + return state.FireResult[S]{}, false, err + } if h.svc == nil { return state.FireResult[S]{}, false, fmt.Errorf("durable: no service registry wired for %q (supply WithServiceRegistry)", h.id) } @@ -109,7 +120,7 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire entry, err := recordServiceOutcome(id, h.svc.LastResult, h.svc.LastError()) if err != nil { - return res, true, fmt.Errorf("durable: recording service %q result for %q: %w", id, h.id, err) + return res, true, h.fail(fmt.Errorf("durable: recording service %q result for %q: %w", id, h.id, err)) } if h.svcBuf != nil { *h.svcBuf = append(*h.svcBuf, entry) @@ -120,7 +131,6 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire if err := h.persistStep(ctx, step, &rec); err != nil { return res, true, err } - h.nextStep++ return res, true, nil }