diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index cec71a8..afaa543 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -30,10 +30,12 @@ implement the `Store` interface out of tree. - **Deterministic replay.** A recovered instance is byte-identical to one that never crashed, because recovery replays the recorded driving events and nondeterministic results rather than re-executing their sources. -- **Exactly-once effects.** A domain effect is applied exactly once over the +- **Exactly-once effects.** A domain effect from a step (not an initial entry, which is at-least-once, or a parent transition an actor drives, which is not dispatched) is applied exactly once over the instance's lifetime (the live run plus any number of recoveries), even though - the replay loop is at-least-once. Each effect carries a deterministic - `EffectID` and is deduplicated through the `Store`'s dispatch set. + the replay loop is at-least-once. The exception is a crash or failed mark after the + handler returned, or a handoff between owners, where it can apply twice, so keep + handlers idempotent on `EffectInstance` plus the `EffectID`. Each effect carries a + deterministic `EffectID` and is deduplicated through the `Store`'s dispatch set. - **Durability across restart.** Every `Fire` step is write-ahead appended to the `Store` before it is acknowledged, so a crash after a successful `Fire` never loses the step. Periodic checkpoints bound the tail that recovery replays. @@ -74,7 +76,7 @@ the first time, and replayed verbatim on recovery: | **Clock** (timers) | `WithRunnerClock` | `JournalClockRead`, replayed so timers fire at the same recorded instants, wall-clock-independent; armed deadlines survive checkpoint compaction | | **Invoked services** | `WithServiceRegistry` + `Handle.RunService` | `JournalServiceResult`; a recorded result replays through the kernel's settle seam without re-running the service, and an unrecorded one (a context canceled during the service, a failed `Append`, a crash) runs again when the host calls `RunService` | | **Child-machine actors** | `WithActorPalette` + `Handle.DeliverToActor` | `JournalActorMessage`; recovery re-fires a recorded parent transition; a settled actor doesn't re-run, and a running one restarts from its initial state on any `Recover` | -| **Domain effects** | `WithEffectHandler` | dispatch set, applied exactly once via deterministic `EffectID` dedup | +| **Domain effects** | `WithEffectHandler` | dispatch set, a step's effects applied exactly once via deterministic `EffectID` dedup, with the exceptions under Exactly-once effects | Use `WithCheckpointEvery(n)` to tune how often a full snapshot is written: a shorter interval bounds recovery replay, a longer one cuts checkpoint cost. @@ -131,19 +133,21 @@ in by implementing `HistoryStore` (the in-tree `MemStore` does so under 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, or a `Tick`, `RunService` or `DeliverToActor` whose -context ended partway, 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 step that may not have been recorded, a failed effect dispatch, or a `Tick`, `RunService` +or `DeliverToActor` whose context ended partway, 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. `Recover` retries an unapplied effect, and +returns `ErrEffectDispatch` while the handler still fails. ## 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 `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 +observes the JSON-decoded shape of actor done-data on a replayed `onDone`. A +`WithEventData` payload and a service's result are 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`, +which loses precision above 2^53. An error stays an error with the same message, with any +invalid UTF-8 replaced, and `errors.Is` no longer matches it. `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 diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 6d9a73d..9c3ce2b 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -9,6 +9,8 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### Added +- **`EffectInstance`** reads the instance from the ctx an `EffectHandler` receives, because + effect ids repeat across instances. - **Fencing.** `WithFence` on `Store.Append` and `WithCheckpointFence` on `Store.Checkpoint` carry a caller-issued token, and a Store refuses a token lower than the highest it has seen for the instance with `ErrFenced`, writing nothing. @@ -30,6 +32,20 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### Changed - Requires Go 1.27.3. +- **A failed effect dispatch fails the Handle.** The call returns `ErrHandleFailed` + wrapping `ErrEffectDispatch`, before any checkpoint, so the effect can't be compacted + away. `Recover` retries it, and fails with `ErrEffectDispatch` while the handler still + does. A `Fire` after a failed dispatch used to be accepted. A failed `MarkDispatched` + fails the Handle too, and `Recover` applies that effect again. The handler's error now + matches through `errors.Is`. +- **A `KindedEffect` that can't marshal fails `Tick` and `RunService`,** as it already + failed `Fire`, even with no handler wired. +- **A service result reaches `Assign` decoded, live as well as on replay,** so a retried + effect can't differ from the first attempt: a struct arrives as a map, an integer as a + `float64`, which loses precision above 2^53, and an error as an error with the same + message, any invalid UTF-8 replaced, which `errors.Is` no longer matches. An `Error` + method that panics settles with a message saying so. The journal records this decoded + value. - **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 @@ -75,6 +91,14 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. - **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. +- **Effects reach the handler from `Start`, `Tick` and `RunService`.** They were only + dispatched from `Fire`. `Start` applies the initial entry's effects after encoding the + baseline and before writing it, and only if the entry succeeded. Those effects are + at-least-once: a handler failure, a failed baseline write or overlapping `Start` calls + re-apply them, so key any dedupe on `EffectInstance` plus the effect id. Effects from a + parent transition an actor drives, including during `Fire`, are still not dispatched. + The first `Recover` of an older journal applies effects from uncompacted `Tick` and + service steps. - **A step whose context ended partway isn't journaled.** A canceled `RunService`, `Tick` or `DeliverToActor` used to record a result the live instance never applied, which recovery then committed, or for an actor, failed on. Each now returns `ErrHandleFailed` diff --git a/durable/README.md b/durable/README.md index 156b62a..8fc4889 100644 --- a/durable/README.md +++ b/durable/README.md @@ -31,10 +31,12 @@ backends implement the `Store` interface out of tree. - **Deterministic replay**: a recovered instance is byte-identical to one that never crashed, because recovery replays the recorded driving events and nondeterministic results rather than re-executing their sources. -- **Exactly-once effects**: a domain effect is applied exactly once over the +- **Exactly-once effects**: a domain effect from a step (not an initial entry, which is at-least-once, or a parent transition an actor drives, which is not dispatched) is applied exactly once over the instance's lifetime (the live run plus any number of recoveries), even though - the replay loop is at-least-once. Each effect carries a deterministic - `EffectID` and is deduplicated through the `Store`'s dispatch set. + the replay loop is at-least-once. The exception is a crash or failed mark after the + handler returned, or a handoff between owners, where it can apply twice, so keep + handlers idempotent on `EffectInstance` plus the `EffectID`. Each effect carries a + deterministic `EffectID` and is deduplicated through the `Store`'s dispatch set. - **Durability across restart**: every `Fire` step is write-ahead appended to the `Store` before it is acknowledged, so a crash after a successful `Fire` never loses the step. Periodic checkpoints bound the tail recovery replays. @@ -84,7 +86,7 @@ the first time, and replayed verbatim on recovery: | **Clock** (timers) | `WithRunnerClock` | `JournalClockRead`, replayed so timers fire at the same recorded instants, wall-clock-independent; armed deadlines survive checkpoint compaction | | **Invoked services** | `WithServiceRegistry` + `Handle.RunService` | `JournalServiceResult`; a recorded result replays through the kernel's settle seam without re-running the service, and an unrecorded one (a context canceled during the service, a failed `Append`, a crash) runs again when the host calls `RunService` | | **Child-machine actors** | `WithActorPalette` + `Handle.DeliverToActor` | `JournalActorMessage`; recovery re-fires a recorded parent transition; a settled actor doesn't re-run, and a running one restarts from its initial state on any `Recover` | -| **Domain effects** | `WithEffectHandler` | dispatch set, applied exactly once via deterministic `EffectID` dedup | +| **Domain effects** | `WithEffectHandler` | dispatch set, a step's effects applied exactly once via deterministic `EffectID` dedup, with the exceptions under Exactly-once effects | Use `WithCheckpointEvery(n)` to tune how often a full snapshot is written (a shorter interval bounds recovery replay; a longer one cuts checkpoint cost). @@ -142,19 +144,21 @@ otherwise. 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, or a `Tick`, `RunService` or `DeliverToActor` whose -context ended partway, 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 step that may not have been recorded, a failed effect dispatch, or a `Tick`, `RunService` +or `DeliverToActor` whose context ended partway, 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. `Recover` retries an unapplied effect, and +returns `ErrEffectDispatch` while the handler still fails. ## 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 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 +observes the JSON-decoded shape of actor done-data on a replayed `onDone`. A +`WithEventData` payload and a service's result are 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`, +which loses precision above 2^53. An error stays an error with the same message, with any +invalid UTF-8 replaced, and `errors.Is` no longer matches it. `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 diff --git a/durable/actor.go b/durable/actor.go index 716c4dd..ecf89db 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -137,9 +137,9 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef *h.actorBuf = append(*h.actorBuf, entries...) } - step := h.nextStep - rec := Record{Step: step, Entries: h.drainActors()} - if err := h.persistStep(ctx, step, &rec); err != nil { + rec := Record{Step: h.nextStep, Entries: h.drainActors()} + // nil: a parent transition an actor drives isn't dispatched (see doc.go). + if err := h.commitStep(ctx, &rec, nil); err != nil { return true, err } return true, nil diff --git a/durable/doc.go b/durable/doc.go index 93355b2..99ead02 100644 --- a/durable/doc.go +++ b/durable/doc.go @@ -17,9 +17,12 @@ // recovery replays the exact recorded driving events and nondeterministic results // through the same pure transition function. // -// Exactly-once effects: a domain effect emitted by a transition is applied +// Exactly-once effects: a domain effect emitted by a step (Runner.Start's initial entry +// is at-least-once, and a parent transition an actor drives isn't dispatched) is applied // exactly once over the instance's lifetime — the live run plus any number of -// recoveries — even though the runtime's replay loop is at-least-once. Each +// recoveries — even though the runtime's replay loop is at-least-once. The exception is +// a crash or failed mark after the handler returned, or a handoff between owners, where +// an effect can apply twice, so keep handlers idempotent on EffectInstance plus the EffectID. Each // effect is stamped with a deterministic EffectID (step, ordinal, kind) and // deduplicated through the Store's dispatch set. // @@ -118,7 +121,7 @@ // publishing a message — as an Effect value the kernel routes out through // FireResult.Effects without applying. The Runner applies it through the caller- // supplied EffectHandler (WithEffectHandler), exactly once over the instance's -// lifetime. +// lifetime, with the exceptions under Exactly-once effects above. // // Each emitted effect is stamped with a deterministic EffectID derived from its // step, its emission ordinal within that step, and its kind — every component a diff --git a/durable/effect.go b/durable/effect.go index 73a002c..cd82861 100644 --- a/durable/effect.go +++ b/durable/effect.go @@ -34,8 +34,8 @@ import ( // Each dispatchable effect is stamped with an EffectID that is a pure function of // the step it was emitted at, its ordinal among the dispatchable effects of that // step, and its kind: "##". Every component is deterministic -// — the step is the Fire ordinal, emission order is fixed by the machine -// definition, and the kind is the effect's stable discriminant — so the same +// — the step is its journal ordinal (-1 for Start's entry effects), emission order is +// fixed by the machine definition, and the kind is the effect's stable discriminant — so the same // effect is stamped with the identical id on the live run and on every recovery // replay, which re-fires the same events and re-emits the same effects in the // same order. That stable identity is what makes the dedup set meaningful across @@ -49,6 +49,8 @@ import ( // (write-ahead: the intent to dispatch is durable before any dispatch). // 2. Dispatch each effect whose id is not already in the dispatched set, calling // the handler, and mark each id dispatched in the Store as it succeeds. +// 3. Checkpoint, if one is due. This comes last so a checkpoint never compacts an +// effect that hasn't landed. // // This yields exactly-once effect application over the at-least-once delivery the // runtime is built on: @@ -59,20 +61,44 @@ import ( // once) and, because it is then marked, never again. // - Crash AFTER dispatch (id marked): recovery re-emits the effect but finds its // id already marked and skips it — so it is not redispatched (exactly once). -// - Handler error: the effect is left un-marked and the error surfaces wrapped in -// ErrEffectDispatch, so a later recovery retries it (at-least-once until it -// succeeds; exactly-once thereafter). +// - Crash or failed mark AFTER the handler returned: the effect was applied but its +// id isn't marked, so recovery applies it again. The handler should be idempotent +// on EffectInstance plus the id for this window, and for a handoff, when two owners +// can run it at once. +// - Handler error: the effect is left un-marked, the error surfaces wrapped in +// ErrEffectDispatch and the Handle fails with ErrHandleFailed, so a later recovery +// retries it (at-least-once until it succeeds; exactly-once thereafter). // // The dedup set is held by the Store (DispatchStore) so it shares the Record's // durability and survives checkpoint compaction — a delayed redispatch of an // already-applied, since-compacted effect is still recognized and skipped. // EffectHandler applies one emitted domain effect, identified by its stamped -// EffectID. The Runner calls it exactly once per EffectID over the lifetime of an -// instance (see WithEffectHandler). A non-nil error leaves the effect un-marked -// for a later retry and is surfaced to the caller wrapped in ErrEffectDispatch. +// EffectID. The Runner calls it once per EffectID over the lifetime of an instance +// (see WithEffectHandler), with three exceptions: a repeated Start can apply the initial +// entry's effects again (see Runner.Start), an effect applies again if the handler +// returned but the mark didn't land, and two owners can run it at once during a handoff. +// An id is unique only within its instance: EffectInstance reads the instance from ctx. +// A non-nil error leaves the effect un-marked for a later retry and is surfaced to the +// caller wrapped in ErrEffectDispatch. type EffectHandler func(ctx context.Context, effectID string, effect state.Effect) error +// effectInstanceKey is the ctx key for the instance an EffectHandler call belongs to. +type effectInstanceKey struct{} + +// withEffectInstance returns ctx carrying id for the handler. +func withEffectInstance(ctx context.Context, id InstanceID) context.Context { + return context.WithValue(ctx, effectInstanceKey{}, id) +} + +// EffectInstance returns the instance an EffectHandler call belongs to, from the ctx the +// handler received. A handler that keeps its own dedupe record must key on this plus the +// effect id, because effect ids repeat across instances. +func EffectInstance(ctx context.Context) (InstanceID, bool) { + id, ok := ctx.Value(effectInstanceKey{}).(InstanceID) + return id, ok +} + // DispatchStore is the dedup seam for idempotent effect dispatch: the Store // tracks which effect ids an instance has applied, so a (re)dispatch skips the // ones that already landed. It is an additive, optional capability — the core @@ -193,6 +219,24 @@ func (h *Handle[S, E, C]) dispatchEffects(ctx context.Context, des []dispatchabl return h.runner.dispatch(ctx, h.id, des) } +// applyStartEffects invokes the handler for the initial entry's domain effects. Like +// dispatch it is a no-op without a handler or a DispatchStore. It marks nothing, because +// the instance doesn't exist yet and nothing replays these effects. +func (r *Runner[S, E, C]) applyStartEffects(ctx context.Context, id InstanceID, des []dispatchableEffect) error { + if len(des) == 0 || r.cfg.effectHandler == nil { + return nil + } + if _, ok := r.store.(DispatchStore); !ok { + return nil + } + for _, de := range des { + if err := r.cfg.effectHandler(withEffectInstance(ctx, id), de.id, de.effect); err != nil { + return fmt.Errorf("%w: effect %q for %q: %w", ErrEffectDispatch, de.id, id, err) + } + } + return nil +} + // dispatchReplayEffects re-dispatches a replayed step's domain effects during // recovery: it re-stamps the re-emitted effects with the same deterministic ids // (step and emission order reproduce exactly) and applies only those not already @@ -225,8 +269,8 @@ func (r *Runner[S, E, C]) dispatch(ctx context.Context, id InstanceID, des []dis if dispatched[de.id] { continue } - if err := r.cfg.effectHandler(ctx, de.id, de.effect); err != nil { - return fmt.Errorf("%w: effect %q for %q: %v", ErrEffectDispatch, de.id, id, err) + if err := r.cfg.effectHandler(withEffectInstance(ctx, id), de.id, de.effect); err != nil { + return fmt.Errorf("%w: effect %q for %q: %w", ErrEffectDispatch, de.id, id, err) } if err := ds.MarkDispatched(ctx, id, de.id); err != nil { return fmt.Errorf("durable: marking effect %q dispatched for %q: %w", de.id, id, err) diff --git a/durable/effect_test.go b/durable/effect_test.go index bb06556..1be259b 100644 --- a/durable/effect_test.go +++ b/durable/effect_test.go @@ -507,3 +507,182 @@ func TestMemStore_DispatchSet_MarkAndQuery(t *testing.T) { t.Fatalf("unmarked id reported dispatched") } } + +// entryNotifyMachine emits a host effect on entering the state an instance starts +// in, so the effect originates in Start rather than Fire. +func entryNotifyMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("entryNotify"). + Action("notify", func(state.ActionCtx[*runCtx]) (state.Effect, error) { + return notifyEffect{To: "ops@example.com"}, nil + }). + State("ready").OnEntry("notify"). + Initial("ready"). + Quench() +} + +// TestRunner_EffectDispatch_Start_AppliesEntryEffect proves a domain effect emitted +// on entering the initial state reaches the handler. +func TestRunner_EffectDispatch_Start_AppliesEntryEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + r := durable.NewRunner(entryNotifyMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](rec.handler())) + if _, err := r.Start(ctx, "i1", &runCtx{}, state.WithInitialState("ready")); err != nil { + t.Fatalf("Start: %v", err) + } + if n := rec.totalApplies(); n != 1 { + t.Fatalf("handler applied %d effects, want 1 (the initial state's entry notify)", n) + } +} + +// TestRunner_EffectDispatch_TimerTransition_AppliesEffect proves a domain effect +// emitted by a timer-driven transition reaches the handler. +func TestRunner_EffectDispatch_TimerTransition_AppliesEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + clk := state.NewFakeClock(epoch) + r := durable.NewRunner(timerNotifyMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](rec.handler()), + durable.WithRunnerClock[string, string, *runCtx](clk)) + h, err := r.Start(ctx, "i1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err != nil { + t.Fatalf("Fire go: %v", err) + } + clk.Advance(2 * time.Second) + if _, err = h.Tick(ctx); err != nil { + t.Fatalf("Tick: %v", err) + } + if n := rec.totalApplies(); n != 2 { + t.Fatalf("handler applied %d effects, want 2 (the timer transition's notify)", n) + } +} + +// TestRunner_EffectDispatch_ServiceSettle_AppliesEffect proves a domain effect +// emitted by the transition a service result drives reaches the handler. +func TestRunner_EffectDispatch_ServiceSettle_AppliesEffect(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + var calls int64 + fn := counterService(&calls) + r := durable.NewRunner(fetchThenWaitMachine(fn), durable.NewMemStore(), + durable.WithEffectHandler[string, string, svcCtx](rec.handler()), + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))) + h, err := r.Start(ctx, "i1", svcCtx{}, 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) + } + mustRunService(t, h, state.InvokeID("fetchThenWait", "loading", 0)) + if n := rec.totalApplies(); n != 1 { + t.Fatalf("handler applied %d effects, want 1 (the settle transition's notify)", n) + } +} + +// TestRunner_EffectDispatch_FailedThenCheckpointed_AppliedOnce proves an effect +// whose dispatch failed is still applied exactly once, live or on recovery, after +// a later checkpoint compacts its step. +func TestRunner_EffectDispatch_FailedThenCheckpointed_AppliedOnce(t *testing.T) { + ctx := context.Background() + st := durable.NewMemStore() + live := newEffectRecorder() + live.failKinds["demo.notify"] = errors.New("downstream unavailable") + opts := []durable.Option[string, string, *runCtx]{ + durable.WithEffectHandler[string, string, *runCtx](live.handler()), + durable.WithCheckpointEvery[string, string, *runCtx](2), + } + h, err := durable.NewRunner(effectMachine(), st, opts...).Start(ctx, "i1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "go"); err == nil { // step 0: its notify fails + t.Fatalf("Fire go: want the handler error") + } + delete(live.failKinds, "demo.notify") + if _, err = h.Fire(ctx, "bill"); err != nil { // step 1: checkpoint compacts step 0 + // A Handle that refuses Fires after a failed dispatch is resumed by Recover. + if h, err = durable.Recover(ctx, effectMachine(), st, "i1", opts...); err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = h.Fire(ctx, "bill"); err != nil { + t.Fatalf("Fire bill: %v", err) + } + } + + healthy := newEffectRecorder() + if _, err = durable.Recover(ctx, effectMachine(), st, "i1", + durable.WithEffectHandler[string, string, *runCtx](healthy.handler())); err != nil { + t.Fatalf("Recover: %v", err) + } + // go emits notify; bill emits charge and notify. Each applies exactly once. + ids := append(live.ids(), healthy.ids()...) + if len(ids) != 3 { + t.Fatalf("applied %d effect ids across live and recovery, want 3: %v", len(ids), ids) + } + for _, id := range ids { + if n := live.count(id) + healthy.count(id); n != 1 { + t.Fatalf("effect %q applied %d times, want once", id, n) + } + } +} + +// failingEntryMachine emits a host effect on entry, then fails the entry. +func failingEntryMachine() *state.Machine[string, string, *runCtx] { + return state.ForgeFor[*runCtx]("failingEntry"). + Action("notify", func(state.ActionCtx[*runCtx]) (state.Effect, error) { + return notifyEffect{To: "ops@example.com"}, nil + }). + Action("boom", func(state.ActionCtx[*runCtx]) (state.Effect, error) { + return nil, errors.New("boom") + }). + State("ready").OnEntry("notify").OnEntry("boom"). + Initial("ready"). + Quench() +} + +// TestRunner_EffectDispatch_Start_FailedEntryAppliesNothing proves an effect emitted +// before a failed entry action doesn't reach the handler, as with a failed Fire. +func TestRunner_EffectDispatch_Start_FailedEntryAppliesNothing(t *testing.T) { + ctx := context.Background() + rec := newEffectRecorder() + r := durable.NewRunner(failingEntryMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](rec.handler())) + if _, err := r.Start(ctx, "i1", &runCtx{}, state.WithInitialState("ready")); err != nil { + t.Fatalf("Start: %v", err) + } + if n := rec.totalApplies(); n != 0 { + t.Fatalf("handler applied %d effects from a failed entry, want 0", n) + } +} + +// TestEffectInstance_ReachesTheHandler proves a handler can tell instances apart, on +// Start and on a step. +func TestEffectInstance_ReachesTheHandler(t *testing.T) { + ctx := context.Background() + var seen []durable.InstanceID + handler := func(ctx context.Context, _ string, _ state.Effect) error { + id, ok := durable.EffectInstance(ctx) + if !ok { + t.Error("no instance on the handler's ctx") + } + seen = append(seen, id) + return nil + } + r := durable.NewRunner(entryNotifyMachine(), durable.NewMemStore(), + durable.WithEffectHandler[string, string, *runCtx](handler)) + for _, id := range []durable.InstanceID{"a", "b"} { + if _, err := r.Start(ctx, id, &runCtx{}, state.WithInitialState("ready")); err != nil { + t.Fatalf("Start: %v", err) + } + } + if len(seen) != 2 || seen[0] != "a" || seen[1] != "b" { + t.Fatalf("handler saw %v, want [a b]", seen) + } + if _, ok := durable.EffectInstance(ctx); ok { + t.Fatal("instance on a ctx the runner never touched") + } +} diff --git a/durable/errors.go b/durable/errors.go index 928739f..50d04c8 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -55,8 +55,8 @@ var ( // 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, or a Tick, RunService or DeliverToActor's context ended partway, on that - // call and every later one. It wraps the cause. Recover a new + // recorded, a Tick, RunService or DeliverToActor's context ended partway, or an + // effect's dispatch failed, 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(errorPrefix + errorHandleFailed) @@ -70,9 +70,10 @@ var ( ErrInstanceExists = errors.New(errorPrefix + errorInstanceExists) // ErrEffectDispatch wraps a failure returned by the caller-supplied effect - // handler (WithEffectHandler) while applying an emitted effect. The effect was + // handler (WithEffectHandler) while applying an emitted effect. A step's effect was // persisted (write-ahead) but not marked dispatched, so a later recovery - // retries it; callers match this with errors.Is to distinguish a dispatch + // retries it, and the Handle fails with ErrHandleFailed; a failure in Runner.Start + // leaves no instance. Recover fails with this error while the handler still does; callers match this with errors.Is to distinguish a dispatch // failure from a kernel transition or persistence error. ErrEffectDispatch = errors.New(errorPrefix + errorEffectDispatch) ) diff --git a/durable/runner.go b/durable/runner.go index 73ea9af..826a7f5 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -98,7 +98,8 @@ func NewRunner[S comparable, E comparable, C any](m *state.Machine[S, E, C], st // 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. +// newer owner holds the instance: stop, because recovering won't help. A failed effect +// dispatch fails the Handle the same way, and Recover retries the effect. type Handle[S comparable, E comparable, C any] struct { runner *Runner[S, E, C] id InstanceID @@ -150,10 +151,9 @@ func (h *Handle[S, E, C]) drainClock() []state.JournalEntry { // durable seams compose: the scheduler arms or cancels delayed (`after`) timers (and // the Handle's timer table mirrors their deadlines for checkpointing), the service // runner starts or stops invoked services, and the actor system spawns or stops -// child-machine actors. It is the shared absorb the live seam methods (Fire, -// RunService, DeliverToActor, Tick) call for the effects they produced, so a state a -// settle or delivery enters arms whatever it declares — a service whose onDone spawns -// an actor, an actor whose onDone arms a timer, and so on. Absorbing a driver's +// child-machine actors. It is the shared absorb Fire and RunService call for the +// effects they produced, so a state a settle enters arms whatever it declares, such as +// a service whose onDone spawns an actor. Absorbing a driver's // effects is a no-op when that driver is not wired or the effects name none of its // kind. func (h *Handle[S, E, C]) absorbDrivers(ctx context.Context, effects []state.Effect) { @@ -281,6 +281,14 @@ func (h *Handle[S, E, C]) ID() InstanceID { return h.id } // state.WithInitialState) configure the initial configuration. Starting an // InstanceID that already exists in the Store reports ErrInstanceExists rather // than clobbering its recorded baseline. +// +// Start applies the initial entry's domain effects before the baseline write, and only +// if the entry succeeded: a failed entry still returns a Handle and a nil error, and +// Instance().InitialErr reports it. They're at-least-once, unlike a step's: a handler failure or a +// failed baseline write leaves no instance, so calling Start again re-applies every +// effect, as does a crash before the baseline. Overlapping Starts on one id, including a +// stale owner's, each apply them before the Store refuses all but one. To dedupe, key on +// EffectInstance plus the effect id. A handler wired only at Recover never sees them. func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opts ...state.CastOption[S]) (*Handle[S, E, C], error) { if _, _, err := r.store.Load(ctx, id); err == nil { return nil, fmt.Errorf("%w: %q", ErrInstanceExists, id) @@ -301,10 +309,10 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock), state.WithUnboundedHistory[S]()}, opts...) inst := r.machine.Cast(input, castOpts...) - // Persist a baseline checkpoint at baselineStep (below the first fired step) - // so the instance is reconstructable from the Store before any event is fired. - // The baseline is the cast snapshot; the tail then accumulates fired steps on - // top of it. + // Encode a baseline checkpoint at baselineStep (below the first fired step), so the + // instance is reconstructable from the Store before any event is fired. The baseline + // is the cast snapshot; the tail then accumulates fired steps on top of it. Encoding + // comes before the effects, so an instance that can't be encoded applies none. snap, err := state.MarshalSnapshot(inst.Snapshot()) if err != nil { return nil, fmt.Errorf("durable: marshaling start snapshot for %q: %w", id, err) @@ -315,6 +323,16 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt if err != nil { return nil, fmt.Errorf("durable: marshaling start checkpoint envelope for %q: %w", id, err) } + // Apply the initial entry's domain effects before the instance exists: a handler + // failure then leaves nothing behind, so the caller can Start again. Nothing can + // re-derive these effects after the baseline, so they're at-least-once. An entry that + // failed applies none, as a failed Fire dispatches none. + if inst.InitialErr() == nil { + if err := r.applyStartEffects(ctx, id, dispatchableEffects(baselineStep, inst.InitialEffects())); err != nil { + return nil, err + } + } + if err := r.store.Checkpoint(ctx, id, baseEnv, baselineStep, WithCheckpointFence(r.cfg.fence)); err != nil { return nil, fmt.Errorf("durable: checkpointing start baseline for %q: %w", id, err) } @@ -363,7 +381,7 @@ func (r *Runner[S, E, C]) newServiceRunner(inst *state.Instance[S, E, C]) *state if r.cfg.serviceReg == nil { return nil } - return state.NewServiceRunner(inst, r.cfg.serviceReg) + return state.NewServiceRunner(inst, r.cfg.serviceReg).WithResultTransform(replayableOutcome) } // newActorSystem builds the kernel host driver for child-machine actors bound to @@ -382,7 +400,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. Handle.Fire describes -// how a state.WithEventData payload is handled, and when it's refused. It is the +// how a state.WithEventData payload is handled, and when it's refused. Loading can +// return ErrEffectDispatch before the event fires, as Recover can. 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. Calls on one id aren't serialized, and must not overlap @@ -400,7 +419,8 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // 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 journals nothing (no step was produced) and is -// returned to the caller. +// returned to the caller. Between the append and any checkpoint it applies the step's +// domain effects; a failed dispatch fails the Handle with ErrHandleFailed. // // 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 @@ -437,32 +457,8 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO h.absorbDrivers(ctx, res.Effects) step := h.nextStep - // Stamp this step's domain effects with their deterministic ids and carry them - // in the Record so the write-ahead append durably names every effect that must - // be dispatched before any dispatch happens. - des := dispatchableEffects(step, res.Effects) - effEnvs, err := recordEffects(des) - if err != nil { - 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 - // 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, h.fail(err) - } - h.nextStep++ - // Dispatch AFTER the append: a crash in this window leaves the effect recorded - // but un-marked, so recovery redispatches it (at-least-once), deduped by id to - // exactly-once once it lands. - if err := h.dispatchEffects(ctx, des); err != nil { - return res, err - } - // Compact through this step only now that its effects are dispatched, so the - // checkpoint never discards a not-yet-dispatched effect. - if err := h.checkpointStep(ctx, step, &rec); err != nil { + rec := Record{Step: step, Event: []byte(res.Trace.EventPayload), EventData: payload.data, EventDataError: payload.errMsg, Entries: h.drainClock()} + if err := h.commitStep(ctx, &rec, res.Effects); err != nil { return res, err } return res, nil @@ -518,28 +514,40 @@ 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.appendStep(ctx, step, &rec); err != nil { - return results, h.fail(err) + var effects []state.Effect + for i := range results { + effects = append(effects, results[i].Effects...) } - h.nextStep = rec.followingStep() - if err := h.checkpointStep(ctx, step, &rec); err != nil { - return results, err - } - return results, nil + return results, h.commitStep(ctx, &rec, effects) } -// 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. This -// relies on the Append running inline, under the Handle's lock, before the step -// returns: an Append deferred off this goroutine would acknowledge a step whose -// outcome isn't known, and the Handle couldn't fail. -func (h *Handle[S, E, C]) persistStep(ctx context.Context, step int, rec *Record) error { - if err := h.appendStep(ctx, step, rec); err != nil { +// commitStep makes a step durable, in write-ahead order. It stamps the step's domain +// effects and carries them in rec, appends rec, advances the next step, dispatches the +// effects, and only then checkpoints, so a checkpoint never compacts an effect that +// hasn't landed: a crash between append and dispatch leaves it in the tail for recovery +// to apply, at-least-once and deduped by id. +// +// The kernel has already applied the step, so a failed encode or append fails the +// Handle. So does a failed dispatch: the effect stays in the tail, and Recover retries +// it. A failed checkpoint doesn't, because the Record is durable. This relies on the +// Append running inline, under the Handle's lock, before the step returns: an Append +// deferred off this goroutine would acknowledge a step whose outcome isn't known, and +// the Handle couldn't fail. +func (h *Handle[S, E, C]) commitStep(ctx context.Context, rec *Record, effects []state.Effect) error { + des := dispatchableEffects(rec.Step, effects) + envs, err := recordEffects(des) + if err != nil { + return h.fail(err) + } + rec.Effects = envs + if err := h.appendStep(ctx, rec.Step, rec); err != nil { return h.fail(err) } h.nextStep = rec.followingStep() - return h.checkpointStep(ctx, step, rec) + if err := h.dispatchEffects(ctx, des); err != nil { + return h.fail(err) + } + return h.checkpointStep(ctx, rec.Step, rec) } // checkpointDue reports whether the checkpoint policy lands a checkpoint at step. @@ -549,7 +557,7 @@ func (h *Handle[S, E, C]) checkpointDue(step int) bool { // appendStep marshals the due checkpoint snapshot into rec (so the Record carries it) // and write-ahead appends rec, WITHOUT compacting the tail yet. Splitting the append -// from the compaction lets the effect-bearing Fire path dispatch a step's domain +// from the compaction lets commitStep dispatch a step's domain // effects after the durable append but before the checkpoint compacts that step away // — so a crash between append and dispatch leaves the effect recoverable from the // still-present tail rather than orphaned behind the compaction. @@ -593,7 +601,9 @@ func (h *Handle[S, E, C]) checkpointStep(ctx context.Context, step int, rec *Rec // snapshot (firing nothing, no IO), and replays the tail's recorded driving // events through the kernel to reach the instance's live state. The returned // Handle continues recording subsequent Fires. Recover reports ErrInstanceNotFound -// for an instance that was never started, and the context's error if ctx ends. +// for an instance that was never started, and the context's error if ctx ends. It +// applies any recorded effect not yet marked dispatched, so it returns ErrEffectDispatch +// and no Handle while the effect handler keeps failing. func Recover[S comparable, E comparable, C any](ctx context.Context, m *state.Machine[S, E, C], st Store, id InstanceID, opts ...Option[S, E, C]) (*Handle[S, E, C], error) { r := NewRunner(m, st, opts...) return r.recover(ctx, id) @@ -687,10 +697,16 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S // re-fires the identical onDone / onError event with the identical data — // running no service. The settle absorbs its own follow-on StartService // effects, so a chained invoke arms its successor for the next entry. + var settled []state.Effect for _, entry := range serviceEntries(rec) { - if err := replayService(ctx, svc, entry); err != nil { + effects, err := replayService(ctx, svc, entry) + if err != nil { return nil, fmt.Errorf("durable: at step %d for %q: %w", rec.Step, id, err) } + settled = append(settled, effects...) + } + if err := r.dispatchReplayEffects(ctx, id, rec.Step, settled); err != nil { + return nil, err } case rec.Tick: // Re-derive the timers the live tick fired: ticking the replay clock @@ -709,6 +725,13 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S mirrorArmed(armed, tickRes[ri].Effects, now) } } + var ticked []state.Effect + for ri := range tickRes { + ticked = append(ticked, tickRes[ri].Effects...) + } + if err := r.dispatchReplayEffects(ctx, id, rec.Step, ticked); err != nil { + return nil, err + } case len(rec.Event) == 0: continue // a checkpoint-only Record drives no event default: diff --git a/durable/runner_options.go b/durable/runner_options.go index 624051c..f6a6245 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -124,14 +124,18 @@ func WithActorPalette[S comparable, E comparable, C any](palette map[string]stat } // WithEffectHandler binds the seam the Runner calls to apply each emitted domain -// effect — an email, a charge, a published message: any at-most-once side effect -// a transition emits as a domain Effect value. A durable effect is applied +// effect — an email, a charge, a published message: any side effect +// a transition emits as a domain Effect value. Effects from a parent transition an actor +// drives aren't dispatched, and a Store that isn't a DispatchStore applies none. A +// durable effect from a step is applied // exactly once over the whole lifetime of an instance (the live run plus any -// number of recoveries), deduped by a deterministic EffectID: the Runner stamps +// number of recoveries), except that it can apply twice after a crash or failed mark +// once the handler returned, or during a handoff, so keep the handler idempotent on +// EffectInstance plus the id. It is deduped by a deterministic EffectID: the Runner stamps // each emitted effect, write-ahead persists the step Record carrying those ids, // then invokes the handler for every id not already marked dispatched, marking -// each as it succeeds. A handler error is surfaced (wrapped in ErrEffectDispatch) -// and the failing effect is left un-marked, so a later recovery retries it +// each as it succeeds. A handler error fails the Handle with ErrHandleFailed, wrapping +// ErrEffectDispatch, and the failing effect is left un-marked, so a later recovery retries it // (at-least-once until it succeeds; exactly-once once it does). // // The handler receives the stamped EffectID and the live effect value. It is nil @@ -170,7 +174,8 @@ func WithMaxEventDataBytes[S comparable, E comparable, C any](n int) Option[S, E // 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 // owner keeps the same token. A non-positive token is the same as no token, which a -// fenced instance refuses. MarkDispatched is not fenced. +// fenced instance refuses. MarkDispatched is not fenced, so during a handoff an effect +// handler can run concurrently for one effect id. // // The token applies to every instance the Runner drives, so a host with a token per // instance passes it to Recover, or builds a Runner per instance. No fence token is written diff --git a/durable/service.go b/durable/service.go index d61b1f7..6cbf430 100644 --- a/durable/service.go +++ b/durable/service.go @@ -29,8 +29,8 @@ import ( // The Runner wraps the kernel's reusable host driver, state.ServiceRunner, which // turns StartService / StopService effects into real ServiceFn executions and // re-fires each result through the invocation's onDone / onError event. On the live -// path RunService delegates to ServiceRunner.Run (resolve + run + settle), reads -// the raw outcome the service produced, and journals it as a JournalServiceResult +// path RunService delegates to ServiceRunner.Tick (resolve + run + settle), reads +// the outcome in its JSON-decoded form, and journals it as a JournalServiceResult // correlated by the invocation id. On recovery the recorded outcomes are replayed // in order through ServiceRunner.SettleDone / SettleError — the same settle methods // the live run drove — so the kernel re-fires the identical onDone / onError event @@ -55,7 +55,7 @@ type serviceResultPayload struct { // Result is the JSON-encoded successful return of the ServiceFn, present for a // SettleDone outcome. It is decoded to a Go value and re-fired through onDone so // the transition's Assign reads it from AssignCtx.Event exactly as on the live - // run. + // run, which settles with the same decoded form (see replayableOutcome). Result json.RawMessage `json:"result,omitempty"` // Error is the failure message for a SettleError outcome. It is reconstructed // into an error and re-fired through onError so the transition's Assign reads it @@ -80,9 +80,12 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { // records its outcome, and settles it — firing the invocation's onDone (on // success) or onError (on failure) event through the durable instance and // recording the produced step. The service is resolved and executed against the -// registry supplied with WithServiceRegistry; its raw result or error is journaled +// registry supplied with WithServiceRegistry; its result or error is journaled // as a JournalServiceResult correlated by id so recovery replays it without -// re-invoking the service. +// re-invoking the service. The outcome reaches the machine JSON-decoded, live as on +// replay, and the journal holds that form: a struct is a map, an integer a float64 that +// loses precision above 2^53, and an error an error with the same message, which +// errors.Is no longer matches. // // 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). It returns @@ -90,11 +93,13 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { // before the service started. // // Once the service has run, it returns true with any error. A failed Append, a result -// that won't encode, or a ctx that ended during the service returns ErrHandleFailed and -// records nothing: a recovered Handle has the service in flight again, and RunService +// or KindedEffect that won't encode, or a ctx that ended during the service returns +// ErrHandleFailed and records nothing: a recovered Handle has the service in flight again, and RunService // runs it a second time. That error also matches the cause, so test ErrHandleFailed // first: it alone means Recover is needed. A failed Checkpoint returns the plain error, // and the result is recorded. +// A failed effect dispatch also returns ErrHandleFailed, but the result is recorded: +// Recover retries the effect and the service doesn't re-run. func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.FireResult[S], bool, error) { h.mu.Lock() defer h.mu.Unlock() @@ -139,14 +144,45 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire *h.svcBuf = append(*h.svcBuf, entry) } - step := h.nextStep - rec := Record{Step: step, Entries: h.drainServices()} - if err := h.persistStep(ctx, step, &rec); err != nil { + rec := Record{Step: h.nextStep, Entries: h.drainServices()} + if err := h.commitStep(ctx, &rec, res.Effects); err != nil { return res, true, err } return res, true, nil } +// replayableOutcome gives a service's outcome the form a replay supplies, so the live +// settle and a replay can't differ: a result in its JSON-decoded form, an error as an +// error with the same message once any invalid UTF-8 is replaced, which is what the +// round trip through JSON does. An Error method that panics settles with a message +// saying so. A result with no JSON form is left as it is, and fails +// the Handle when recording it fails. +func replayableOutcome(result any, err error) (any, error) { + if err != nil { + msg, merr := errorMessage(err) + if merr != nil { + return nil, merr + } + b, jerr := marshalPayload(msg) + if jerr != nil { + return nil, errors.New(msg) + } + if jerr = json.Unmarshal(b, &msg); jerr != nil { + return nil, errors.New(msg) + } + return nil, errors.New(msg) + } + b, jerr := marshalPayload(result) + if jerr != nil { + return result, nil + } + var decoded any + if jerr = json.Unmarshal(b, &decoded); jerr != nil { + return result, nil + } + return decoded, nil +} + // recordServiceOutcome builds the JournalServiceResult entry for a just-settled // service. It reads the success result through lastResult (valid only during the // settle's synchronous window) and the error through lastErr, encoding whichever @@ -157,7 +193,7 @@ func recordServiceOutcome(id string, lastResult func() (any, bool), lastErr erro msg := lastErr.Error() sp.Error = &msg } else if result, ok := lastResult(); ok { - raw, err := json.Marshal(result) + raw, err := marshalPayload(result) if err != nil { return state.JournalEntry{}, fmt.Errorf("marshaling service result: %w", err) } @@ -179,32 +215,34 @@ func recordServiceOutcome(id string, lastResult func() (any, bool), lastErr erro // the recorded payload and calls SettleDone (with the recorded result) or // SettleError (with the reconstructed error), so the kernel re-fires the same // onDone / onError event with the same data and re-derives identical context. It -// reports an error only when the recorded entry cannot be decoded or names no +// returns the effects the settle emitted, and reports an error only when the recorded entry cannot be decoded or names no // in-flight service (a corrupt or mis-ordered journal). -func replayService[S comparable, E comparable, C any](ctx context.Context, svc *state.ServiceRunner[S, E, C], entry state.JournalEntry) error { +func replayService[S comparable, E comparable, C any](ctx context.Context, svc *state.ServiceRunner[S, E, C], entry state.JournalEntry) ([]state.Effect, error) { if svc == nil { - return errors.New("durable: replaying a service result requires a service registry (supply WithServiceRegistry)") + return nil, errors.New("durable: replaying a service result requires a service registry (supply WithServiceRegistry)") } var sp serviceResultPayload if err := json.Unmarshal(entry.Payload, &sp); err != nil { - return fmt.Errorf("durable: decoding recorded service %q outcome: %w", entry.CorrelationID, err) + return nil, fmt.Errorf("durable: decoding recorded service %q outcome: %w", entry.CorrelationID, err) } if sp.Error != nil { - if _, ok := svc.SettleError(ctx, entry.CorrelationID, errors.New(*sp.Error)); !ok { - return fmt.Errorf("durable: replaying service %q: no in-flight service to settle", entry.CorrelationID) + res, ok := svc.SettleError(ctx, entry.CorrelationID, errors.New(*sp.Error)) + if !ok { + return nil, fmt.Errorf("durable: replaying service %q: no in-flight service to settle", entry.CorrelationID) } - return nil + return res.Effects, nil } var result any if len(sp.Result) > 0 { if err := json.Unmarshal(sp.Result, &result); err != nil { - return fmt.Errorf("durable: decoding recorded service %q result: %w", entry.CorrelationID, err) + return nil, fmt.Errorf("durable: decoding recorded service %q result: %w", entry.CorrelationID, err) } } - if _, ok := svc.SettleDone(ctx, entry.CorrelationID, result); !ok { - return fmt.Errorf("durable: replaying service %q: no in-flight service to settle", entry.CorrelationID) + res, ok := svc.SettleDone(ctx, entry.CorrelationID, result) + if !ok { + return nil, fmt.Errorf("durable: replaying service %q: no in-flight service to settle", entry.CorrelationID) } - return nil + return res.Effects, nil } // serviceEntries returns the recorded service outcomes carried by a single diff --git a/durable/service_test.go b/durable/service_test.go index bf4fb5f..9d9223b 100644 --- a/durable/service_test.go +++ b/durable/service_test.go @@ -472,6 +472,25 @@ func TestService_Determinism(t *testing.T) { } } +// fetchThenWaitMachine routes its service result into a state that arms a timer, +// emitting a host effect on the way, so both originate in RunService. +func fetchThenWaitMachine(fn state.ServiceFn[svcCtx]) *state.Machine[string, string, svcCtx] { + return state.ForgeFor[svcCtx]("fetchThenWait"). + Service("fetch", fn). + Action("notify", func(state.ActionCtx[svcCtx]) (state.Effect, error) { + return notifyEffect{To: "ops@example.com"}, nil + }). + State("idle"). + State("loading").Invoke("fetch", state.WithInvokeOnDone("ok")). + State("waiting"). + State("done").Final(). + Initial("idle"). + Transition("idle").On("start").GoTo("loading"). + Transition("loading").On("ok").GoTo("waiting").Do("notify"). + Transition("waiting").After(5 * time.Second).On("elapsed").GoTo("done"). + Quench() +} + // TestService_ResultFromCanceledTurn_NotCommittedOnRecover proves a service result // whose settle is canceled with the turn's context, and so never reaches the live // instance, is not committed by recovery either. @@ -671,3 +690,39 @@ func (c *cancelingClock) Now() time.Time { } return c.Clock.Now() } + +// TestRunService_AssignSeesTheDecodedResult proves a typed result reaches Assign in the +// form a replay supplies, so the two can't differ. +func TestRunService_AssignSeesTheDecodedResult(t *testing.T) { + type user struct{ Email string } + ctx := context.Background() + var seen []any + fn := func(context.Context, state.ServiceCtx[svcCtx]) (any, error) { return user{Email: "a@b.c"}, nil } + m := state.ForgeFor[svcCtx]("decoded"). + Service("fetch", fn). + Reducer("note", func(in state.AssignCtx[svcCtx]) svcCtx { + seen = append(seen, in.Event) + return in.Entity + }). + State("loading").Invoke("fetch", state.WithInvokeOnDone("done"), state.WithInvokeOnError("fail")). + State("ready").Final(). + State("failed").Final(). + Initial("loading"). + Transition("loading").On("done").GoTo("ready").Assign("note"). + Transition("loading").On("fail").GoTo("failed"). + Quench() + store := durable.NewMemStore() + id := durable.InstanceID("d-1") + opts := []durable.Option[string, string, svcCtx]{durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("loading")) + if err != nil { + t.Fatalf("Start: %v", err) + } + mustRunService(t, h, state.InvokeID("decoded", "loading", 0)) + if _, err = durable.Recover(ctx, m, store, id, opts...); err != nil { + t.Fatalf("Recover: %v", err) + } + if len(seen) != 2 || fmt.Sprintf("%T:%v", seen[0], seen[0]) != fmt.Sprintf("%T:%v", seen[1], seen[1]) { + t.Fatalf("live and replay saw %#v", seen) + } +} diff --git a/durable/store.go b/durable/store.go index 31b12a7..b57254f 100644 --- a/durable/store.go +++ b/durable/store.go @@ -59,7 +59,7 @@ type Record struct { // Effects are the effects the step emitted, each with its deterministically // stamped EffectID, persisted before dispatch so a crash between persist and - // dispatch is recoverable: Resume re-dispatches, deduped by EffectID. + // dispatch is recoverable: Recover re-dispatches, deduped by EffectID. Effects []state.EffectEnvelope // Snapshot, when non-empty, is a full marshaled state.Snapshot captured at diff --git a/durable/timetravel.go b/durable/timetravel.go index 654a28e..24a4eea 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -257,7 +257,7 @@ func replayThrough[S comparable, E comparable, C any]( } case hasServiceEntry(rec): for _, entry := range serviceEntries(rec) { - if err := replayService(ctx, svc, entry); err != nil { + if _, err := replayService(ctx, svc, entry); err != nil { return fmt.Errorf("at step %d: %w", rec.Step, err) } } diff --git a/examples/fooddelivery/fooddelivery.go b/examples/fooddelivery/fooddelivery.go index f7b50b8..9c005ae 100644 --- a/examples/fooddelivery/fooddelivery.go +++ b/examples/fooddelivery/fooddelivery.go @@ -388,10 +388,17 @@ func settleReducer(in state.AssignCtx[Order]) Order { // onDone event's payload — the saga's compensation made observable. func recordRefund(in state.AssignCtx[Order]) Order { c := in.Entity - if amount, ok := in.Event.(int64); ok { - c.Refund = amount - c.Log = append(c.Log, fmt.Sprintf("refunded:%d", amount)) + var amount int64 + switch v := in.Event.(type) { + case int64: + amount = v + case float64: // a durable Runner delivers a service result JSON-decoded + amount = int64(v) + default: + return c } + c.Refund = amount + c.Log = append(c.Log, fmt.Sprintf("refunded:%d", amount)) return c } diff --git a/state/CHANGELOG.md b/state/CHANGELOG.md index e4894a8..562df93 100644 --- a/state/CHANGELOG.md +++ b/state/CHANGELOG.md @@ -13,6 +13,8 @@ counts as an additive (minor) versus breaking (major) change. Use the ### Added +- **`ServiceRunner.WithResultTransform`** rewrites a service's outcome before `Tick` + settles it, so a host that records outcomes can give the live run what a replay will see. - **`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. diff --git a/state/runner.go b/state/runner.go index 699ebeb..1d8d8a2 100644 --- a/state/runner.go +++ b/state/runner.go @@ -97,6 +97,9 @@ type ServiceRunner[S comparable, E comparable, C any] struct { // transition's action. It is atomic so a concurrent reader sees a consistent // snapshot. lastResult atomic.Value // serviceOutcome + + // transform, when set, rewrites a ServiceFn's outcome before Tick settles it. + transform func(any, error) (any, error) } // serviceOutcome is the result/error pair a settled service exposes through @@ -217,6 +220,17 @@ func (r *ServiceRunner[S, E, C]) settle(ctx context.Context, id string, result a return res, true } +// WithResultTransform sets a function that rewrites a ServiceFn's outcome before Tick +// settles it, and returns the runner. A host that records outcomes uses it to give the +// live run the form a replay will see. LastResult and LastError then report the rewritten +// outcome. It also runs on an UnboundRefError, where only the returned error is used. It +// doesn't apply to SettleDone or SettleError, which take the outcome as given. Set it +// before the first Tick: the field isn't guarded. +func (r *ServiceRunner[S, E, C]) WithResultTransform(f func(result any, err error) (any, error)) *ServiceRunner[S, E, C] { + r.transform = f + return r +} + // Tick resolves and runs the in-flight service id against the bound registry, // settling it with the ServiceFn's result or error. It is the ServiceRunner's // advance verb — the host-driver counterpart of Scheduler.Tick and @@ -237,10 +251,17 @@ func (r *ServiceRunner[S, E, C]) Tick(ctx context.Context, id string) []FireResu } fn := r.resolve(rs.src.Name) if fn == nil { - res, _ := r.SettleError(ctx, id, &UnboundRefError{Kind: "service", Name: rs.src.Name}) + var err error = &UnboundRefError{Kind: "service", Name: rs.src.Name} + if r.transform != nil { + _, err = r.transform(nil, err) + } + res, _ := r.SettleError(ctx, id, err) return []FireResult[S]{res} } out, err := fn(ctx, ServiceCtx[C]{Entity: r.inst.entity, Params: rs.src.Params, Input: rs.input}) + if r.transform != nil { + out, err = r.transform(out, err) + } if err != nil { res, _ := r.SettleError(ctx, id, err) return []FireResult[S]{res}