diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 347366e0..4680e2a8 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. @@ -124,10 +125,25 @@ 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 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..5067e0d0 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -21,12 +21,44 @@ 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. + +- **`ErrHandleFailed`** reports a Handle out of step with its journal. + ### 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 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 + 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. @@ -36,11 +68,23 @@ 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 +- **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 + 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 + 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..f720c52f 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. @@ -135,11 +136,26 @@ 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 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/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/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/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..47167b76 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -38,6 +38,25 @@ 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, 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/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..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" @@ -24,9 +25,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 +// 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 // replays only the tail after the latest checkpoint rather than the whole run. @@ -75,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 @@ -92,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 @@ -226,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. @@ -251,8 +291,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...) @@ -268,7 +308,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) } @@ -338,10 +378,12 @@ 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. +// 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 { @@ -352,11 +394,33 @@ 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. +// +// 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) { + 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) + 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 @@ -376,15 +440,15 @@ 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), 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 // 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 @@ -410,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 @@ -431,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) } @@ -468,7 +546,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) } @@ -518,10 +596,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) } @@ -617,7 +695,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) } @@ -693,15 +775,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, @@ -718,6 +801,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..bc5b3f7b 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -4,7 +4,12 @@ import ( "context" "encoding/json" "errors" + "runtime" + "strings" + "sync" + "sync/atomic" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -142,6 +147,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 +481,55 @@ 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] { + 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() +} + +// 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 @@ -505,3 +570,402 @@ func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t * t.Fatalf("stale Handle's Fires were acknowledged at %q, but recovery yields %q", stale.Instance().Current(), got) } } + +// 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) { + 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) + } +} + +// 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) + } +} + +// 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() + _, 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) + } +} + +// 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 } diff --git a/durable/store.go b/durable/store.go index 13a35a03..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 @@ -40,6 +41,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 @@ -74,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 @@ -97,7 +117,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 +150,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 +162,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..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) } @@ -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..8ff02f2a 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) @@ -366,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"`