From 0dee0b571e38bda4f07ba929a0f098c0b3f010be Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 09:18:22 -0700 Subject: [PATCH] fix(durable): don't journal a step whose context ended partway A context ended during RunService, Tick or DeliverToActor stops the kernel's Fire, but the result was journaled anyway. Recovery then committed something the live instance never applied, or for an actor, failed on every attempt. Each now returns ErrHandleFailed and records nothing. Recover and StateAt fail on a canceled context instead of returning state short of the journal. Relates to #214. Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 7 +- durable/CHANGELOG.md | 7 + durable/README.md | 7 +- durable/actor.go | 40 +++-- durable/actor_test.go | 42 +++++ durable/doc.go | 23 +-- durable/errors.go | 3 +- durable/runner.go | 44 +++-- durable/runner_options.go | 18 +- durable/service.go | 41 +++-- durable/service_test.go | 201 ++++++++++++++++++++++ durable/timetravel.go | 12 +- 12 files changed, 371 insertions(+), 74 deletions(-) diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 4680e2a8..cec71a8f 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -72,8 +72,8 @@ the first time, and replayed verbatim on recovery: | Seam | Wire with | Recorded as | | --- | --- | --- | | **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`; the service runs once, then recovery replays its result through the kernel's settle seam | -| **Child-machine actors** | `WithActorPalette` + `Handle.DeliverToActor` | `JournalActorMessage`; the behavior runs once, then recovery re-fires the recorded parent transition | +| **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 | Use `WithCheckpointEvery(n)` to tune how often a full snapshot is written: a @@ -131,7 +131,8 @@ 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 fails the Handle with `ErrHandleFailed`. `Recover` +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. diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 5067e0d0..6d9a73df 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -75,6 +75,13 @@ 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. +- **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` + wrapping the error and records nothing. After `Recover` the service stays in flight and + must tolerate running again, the timer is re-armed, and the actor restarts. `Recover` + and `StateAt` now fail on a canceled context, rather than returning state short of the + journal. - **The next step survives a checkpoint that compacts tick barriers.** A checkpoint now records the step that follows it, in a `nextStep` field of its envelope. Recovery used the Trace count, which runs low past a tick barrier, so the next Fire could reuse a diff --git a/durable/README.md b/durable/README.md index f720c52f..156b62a1 100644 --- a/durable/README.md +++ b/durable/README.md @@ -82,8 +82,8 @@ the first time, and replayed verbatim on recovery: | Seam | Wire with | Recorded as | | --- | --- | --- | | **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`; the service runs once, then recovery replays its result through the kernel's settle seam | -| **Child-machine actors** | `WithActorPalette` + `Handle.DeliverToActor` | `JournalActorMessage`; the behavior runs once, then recovery re-fires the recorded parent transition | +| **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 | Use `WithCheckpointEvery(n)` to tune how often a full snapshot is written (a @@ -142,7 +142,8 @@ 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 fails the Handle with `ErrHandleFailed`. `Recover` +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. diff --git a/durable/actor.go b/durable/actor.go index 17bed76b..716c4dd5 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -17,15 +17,14 @@ import ( // NOT a pure function of (configuration, context, event payload, machine // definition) and MUST be recorded to reproduce a run. // -// # Run exactly once, replay the result +// # Run once per recorded result, replay the result // -// A durable actor's behavior runs exactly once, on the live path, and the result -// it routes into the parent is recorded; on recovery the recorded result is -// replayed back through the SAME parent transition the live run drove, and the -// actor behavior is NEVER re-instantiated. This is the standard durable-execution -// guarantee: the child machine (with its side effects and its nondeterministic -// output) runs once, and recovery is a pure replay against the parent kernel rather -// than a re-execution of the actor tree. +// A durable actor's behavior runs on the live path and the result it routes into the +// parent is recorded; on recovery the recorded result is replayed back through the SAME +// parent transition the live run drove, so a settled actor never re-runs. This is the +// standard durable-execution guarantee: recovery is a pure replay against the parent +// kernel rather than a re-execution of the actor tree. An actor still running at recovery +// restarts from its initial state, because its own state isn't journaled. // // The Runner wraps the kernel's reusable host driver, state.ActorSystem, which // turns SpawnActor / StopActor effects into running child actors, owns each actor's @@ -90,23 +89,21 @@ type actorMessagePayload struct { HasData bool `json:"hasData,omitempty"` } -// DeliverToActor routes event into the running actor identified by ref exactly -// once, runs the actor (settling it when it reaches its final state), records each +// DeliverToActor routes event into the running actor identified by ref, runs the actor (settling it when it reaches its final state), records each // parent transition the delivery drove, and persists the produced step. The actor // behavior is resolved and run against the palette supplied with WithActorPalette; // the done-data, error, or parent-driving message it routes is journaled as a // JournalActorMessage correlated by the actor id so recovery replays it without // re-instantiating the actor. // -// It returns false for a ref that names no live actor (already settled, stopped, or -// never spawned), and false with the error when the Handle had already failed or ctx -// has ended. If the actor received the event but the delivery couldn't be recorded or -// appended, it returns true and ErrHandleFailed: the Handle has failed. A failed -// checkpoint returns true and that error, and the Handle carries on. The kernel -// transitions the delivery drove do not themselves fail. When a delivery 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. +// It returns whether the actor was found running: false for a ref that names no live +// actor (already settled, stopped, or never spawned), or false and an error when no +// palette is wired, the Handle has failed, or ctx ended before the delivery. Once the +// actor ran it returns true with any error. A failed Append, a result that won't encode, +// or a ctx that ended during the delivery returns ErrHandleFailed and records nothing: on +// a recovered Handle an InvokeActor actor restarts from its initial state, and a Spawn +// actor doesn't return. A failed Checkpoint returns the plain error, and the delivery is +// recorded. 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() @@ -125,6 +122,11 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef if !delivered { return false, nil } + // A context that ended during the delivery can stop a parent Fire partway, and a + // refused Fire leaves a Trace replay can't re-fire. Record nothing. + if err := ctx.Err(); err != nil { + return true, h.fail(fmt.Errorf("durable: delivering to actor %q for %q: %w", ref.ID, h.id, err)) + } output, settled, settleErr := h.actorOutcome() entries, err := recordActorTransitions(ref.ID, h.inst.History(), before, output, settled, settleErr) diff --git a/durable/actor_test.go b/durable/actor_test.go index f8af0cd5..ef3cd6f5 100644 --- a/durable/actor_test.go +++ b/durable/actor_test.go @@ -2,6 +2,7 @@ package durable_test import ( "context" + "errors" "fmt" "sync/atomic" "testing" @@ -704,3 +705,44 @@ func TestActor_Determinism(t *testing.T) { t.Fatalf("two live runs diverged:\n run1=%s\n run2=%s", s1, s2) } } + +// TestDeliverToActor_ContextEndedPartway_RecordsNothing proves a delivery whose context +// ended records nothing, so every later Recover still works. +func TestDeliverToActor_ContextEndedPartway_RecordsNothing(t *testing.T) { + ctx := context.Background() + m := supervisorMachine() + store := durable.NewMemStore() + id := durable.InstanceID("sup-cancel") + turn, cancel := context.WithCancel(ctx) + defer cancel() + behavior := func(map[string]any) (state.ActorInstance, error) { + child := state.ForgeFor[actorCtx]("child"). + State("working"). + State("done").Final(). + Initial("working"). + Transition("working").On("finish").GoTo("done"). + Quench() + inst := child.Cast(actorCtx{}, state.WithInitialState[string]("working")) + return state.NewActor(inst, func(*state.Instance[string, string, actorCtx]) any { + cancel() // the turn ends after the child finishes, before the parent settles + return "late" + }), nil + } + opts := []durable.Option[string, string, actorCtx]{ + durable.WithActorPalette[string, string, actorCtx](map[string]state.ActorBehavior{"spawnChild": behavior}), + } + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, actorCtx{}, state.WithInitialState("supervising")) + if err != nil { + t.Fatalf("Start: %v", err) + } + ref, ok := h.ActorRef(state.ActorID("supervisor", "supervising", 0)) + if !ok { + t.Fatal("no actor ref for spawned child") + } + if _, err = h.DeliverToActor(turn, ref, "finish"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("DeliverToActor = %v, want ErrHandleFailed", err) + } + if _, err = durable.Recover(ctx, m, store, id, opts...); err != nil { + t.Fatalf("Recover: %v", err) + } +} diff --git a/durable/doc.go b/durable/doc.go index 00e5a8fa..93355b2f 100644 --- a/durable/doc.go +++ b/durable/doc.go @@ -99,17 +99,18 @@ // was compacted out of the tail is re-armed from its persisted deadline rather // than lost. // -// Invoked services (WithServiceRegistry): a service (`invoke`) runs exactly once -// on the live path; its result is journaled as a JournalServiceResult. On -// recovery the recorded result is replayed back through the same kernel settle -// seam — the service is never re-invoked and the same onDone or onError event -// re-fires with the same data. -// -// Child-machine actors (WithActorPalette): an actor's behavior runs exactly once -// on the live path; each parent transition the delivery drives is journaled as a -// JournalActorMessage. On recovery the recorded transition is re-fired directly -// through the parent with the recorded done-data — the actor behavior is never -// re-instantiated. +// Invoked services (WithServiceRegistry): a service (`invoke`) runs on the live path; +// its result is journaled as a JournalServiceResult. On recovery the recorded result is +// replayed back through the same kernel settle seam — that service is never re-invoked +// and the same onDone or onError event re-fires with the same data. A result that +// wasn't recorded has nothing to replay, so the service runs again when the host calls +// RunService. +// +// Child-machine actors (WithActorPalette): an actor's behavior runs on the live path; +// each parent transition the delivery drives is journaled as a JournalActorMessage. On +// recovery the recorded transition is re-fired directly through the parent with the +// recorded done-data, so a settled actor never re-runs. An actor still running at +// recovery restarts from its initial state, because its own state isn't journaled. // // # Idempotent effect dispatch // diff --git a/durable/errors.go b/durable/errors.go index ea357b31..928739f7 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -55,7 +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, on that call and every later one. It wraps the cause. Recover a new + // 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 // 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) diff --git a/durable/runner.go b/durable/runner.go index 9fe3f0ce..73ea9af3 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -39,15 +39,16 @@ import ( // deadline or testing dueness — is journaled and returned verbatim on recovery, // making timer-driven transitions durable and replay wall-clock-independent. // -// Invoked services are the second recorded seam: a service runs exactly once on -// the live path (Handle.RunService, against the registry supplied with -// WithServiceRegistry) and its result is journaled as a JournalServiceResult; on -// recovery the recorded result is replayed back through the kernel's settle seam, -// so the service is never re-invoked and the same onDone / onError event re-fires -// with the same data. +// Invoked services are the second recorded seam: a service runs on the live path +// (Handle.RunService, against the registry supplied with WithServiceRegistry) and its +// result is journaled as a JournalServiceResult; on recovery the recorded result is +// replayed back through the kernel's settle seam, so that service is never re-invoked +// and the same onDone / onError event re-fires with the same data. A result that wasn't +// recorded has nothing to replay, so the service runs again when the host calls +// RunService. // // Child-machine actors are the third recorded seam: an actor's behavior runs -// exactly once on the live path (Handle.DeliverToActor, against the palette +// on the live path (Handle.DeliverToActor, against the palette // supplied with WithActorPalette) and each parent transition the delivery drives — // the actor's onDone / onError, or a message it sends — is journaled as a // JournalActorMessage; on recovery the recorded transition is replayed by re-firing @@ -89,10 +90,11 @@ func NewRunner[S comparable, E comparable, C any](m *state.Machine[S, E, C], st // 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, which can be a context that ended during the Store write. Recover -// a new Handle with the same options to continue. A service +// A step that applied to the kernel but may not have been recorded, or a Tick, RunService +// or DeliverToActor whose context ended partway, leaves the Handle out of step with the +// journal. That call and every later one return ErrHandleFailed, wrapping the cause, +// which can be a context that ended during the Store write. 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 @@ -486,6 +488,16 @@ func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], erro return nil, err } results := h.sched.Tick(ctx) + // A context that ended partway makes the kernel refuse a due timer's Fire after the + // scheduler consumed it. Journaling that would commit on recovery what live never + // applied, so record nothing: Recover re-arms the timer. + if err := ctx.Err(); err != nil { + for i := range results { + if results[i].Err != nil { + return results, h.fail(fmt.Errorf("durable: ticking %q: %w", h.id, err)) + } + } + } step := h.nextStep rec := Record{Step: step, Tick: true, TickSteps: len(results), Entries: h.drainClock()} @@ -581,7 +593,7 @@ 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. +// for an instance that was never started, and the context's error if ctx ends. 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) @@ -652,6 +664,11 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S armed := map[string]pendingTimer{} for i := range tail { + // A canceled context can stop a replayed settle or tick with no error, leaving + // state short of the journal, so check it before each Record. + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("durable: recovering %q: %w", id, err) + } rec := &tail[i] switch { case hasActorEntry(rec): @@ -790,6 +807,9 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S if n := len(tail); n > 0 { next = tail[n-1].followingStep() } + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("durable: recovering %q: %w", id, err) + } return &Handle[S, E, C]{ runner: r, id: id, diff --git a/durable/runner_options.go b/durable/runner_options.go index a84886e4..624051c5 100644 --- a/durable/runner_options.go +++ b/durable/runner_options.go @@ -91,10 +91,10 @@ func WithRunnerClock[S comparable, E comparable, C any](c state.Clock) Option[S, } // WithServiceRegistry binds the invoked-service implementations the Runner runs on -// the live path and resolves on recovery. A durable service runs exactly once — -// live — and its result is recorded; on recovery the recorded result is replayed -// through the same settle seam and the service is never re-invoked, so the registry -// is consulted only to execute a service the first time. Supply it for any machine +// the live path and resolves on recovery. A durable service runs live and its result is +// recorded; on recovery the recorded result is replayed through the same settle seam and +// that service is never re-invoked, so the registry is consulted only to execute a +// service whose result isn't recorded. Supply it for any machine // that declares `invoke` states; a purely event-driven or timer-driven machine // needs none. The registry binds the same ServiceFns the machine declares, by name // (state.NewRegistry().Service(name, fn)). @@ -108,11 +108,11 @@ func WithServiceRegistry[S comparable, E comparable, C any](reg *state.Registry[ // WithActorPalette binds the child-machine actor behaviors the Runner spawns and // runs on the live path and resolves on recovery, keyed by the actor src name (the -// name passed to InvokeActor or the Spawn built-in). A durable actor's behavior -// runs exactly once — live — and its done-data, error, or parent-driving message is -// recorded; on recovery the recorded result is replayed back through the same parent -// transition and the behavior is never re-instantiated, so the palette is consulted -// only to run an actor the first time. Supply it for any machine that spawns child +// name passed to InvokeActor or the Spawn built-in). A durable actor's behavior runs +// live and its done-data, error, or parent-driving message is recorded; on recovery the +// recorded result is replayed back through the same parent transition and that behavior +// is never re-instantiated, so the palette is consulted only to run an actor whose +// result isn't recorded. Supply it for any machine that spawns child // actors; a machine that spawns none needs none. Each behavior is the actor-model // analog of a ServiceFn, registered by the same src name the machine declares. func WithActorPalette[S comparable, E comparable, C any](palette map[string]state.ActorBehavior) Option[S, E, C] { diff --git a/durable/service.go b/durable/service.go index e0927377..d61b1f73 100644 --- a/durable/service.go +++ b/durable/service.go @@ -16,14 +16,15 @@ import ( // (configuration, context, event payload, machine definition) and MUST be recorded // to reproduce a run. // -// # Run exactly once, replay the result +// # Run once per recorded result, replay the result // -// A durable service runs exactly once, on the live path, and its result is -// recorded; on recovery the recorded result is replayed back through the same -// settle seam, and the service is NEVER re-invoked. This is the standard -// durable-execution guarantee: the external call (with its side effects and its -// nondeterministic output) happens once, and recovery is a pure replay against the -// kernel rather than a re-execution of the host's services. +// A durable service runs on the live path and its result is recorded; on recovery the +// recorded result is replayed back through the same settle seam, and that service is +// NEVER re-invoked. This is the standard durable-execution guarantee: recovery is a pure +// replay against the kernel rather than a re-execution of the host's services. A result +// that wasn't recorded, after a context canceled during the service, a failed Append or a +// crash, has nothing to replay, so it stays in flight, runs again when the host calls +// RunService, and must tolerate a second run. // // The Runner wraps the kernel's reusable host driver, state.ServiceRunner, which // turns StartService / StopService effects into real ServiceFn executions and @@ -75,7 +76,7 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { return out } -// RunService runs the in-flight invoked service identified by id exactly once, +// RunService runs the in-flight invoked service identified by id, // 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 @@ -84,12 +85,16 @@ 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), or false -// and the error when the Handle had already failed or ctx has ended. If the service -// settled but its outcome couldn't be recorded or appended, it returns the result, -// true and ErrHandleFailed: the Handle has failed. A failed checkpoint returns the -// result, true and that error, and the Handle carries on. A recovered Handle has an -// unrecorded service in flight again, and RunService runs it a second time. +// names no in-flight service (already settled, stopped, or never started). It returns +// false and an error when no registry is wired, the Handle has failed, or ctx ended +// 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 +// 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. func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.FireResult[S], bool, error) { h.mu.Lock() defer h.mu.Unlock() @@ -112,6 +117,14 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire } res := results[0] + // A context canceled during the service stops the settle's Fire before it applies, + // so the instance never saw this result: journaling it would commit it on recovery. + // The runner has already dropped the service, so this Handle can't run it again; + // Recover re-arms it. + if res.Err != nil && ctx.Err() != nil { + return res, true, h.fail(fmt.Errorf("durable: settling service %q for %q: %w", id, h.id, res.Err)) + } + // The settle re-fired the invocation's onDone / onError, which may enter a state // that arms a timer, starts another service, or spawns an actor. Absorb those // effects into every driver so the next seam method finds them armed — the service diff --git a/durable/service_test.go b/durable/service_test.go index df1d4db8..bf4fb5f0 100644 --- a/durable/service_test.go +++ b/durable/service_test.go @@ -6,6 +6,7 @@ import ( "fmt" "sync/atomic" "testing" + "time" "github.com/stablekernel/crucible/durable" "github.com/stablekernel/crucible/state" @@ -470,3 +471,203 @@ func TestService_Determinism(t *testing.T) { t.Fatalf("two live runs diverged:\n run1=%s\n run2=%s", s1, s2) } } + +// 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. +func TestService_ResultFromCanceledTurn_NotCommittedOnRecover(t *testing.T) { + ctx := context.Background() + turnCtx, cancel := context.WithCancel(ctx) + defer cancel() + fn := func(context.Context, state.ServiceCtx[svcCtx]) (any, error) { + cancel() // the turn's deadline passes while the call is in flight + return "late", nil + } + m := fetchMachine(fn) + store := durable.NewMemStore() + id := durable.InstanceID("fetch-1") + opts := []durable.Option[string, string, svcCtx]{durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, 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) + } + _, _, _ = h.RunService(turnCtx, state.InvokeID("fetch", "loading", 0)) + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + // Compare state, not bytes: live keeps the Trace of the refused settle. + if live, got := h.Instance().Current(), recovered.Instance().Current(); got != live || live != "loading" { + t.Fatalf("recovered %q, live %q, want both loading: the canceled turn's result was committed", got, live) + } + if notes := recovered.Instance().Entity().Notes; len(notes) != 0 { + t.Fatalf("recovered notes %v, want none", notes) + } +} + +// TestService_CanceledTurn_RecoverRerunsTheService proves the canceled result isn't +// journaled, the Handle refuses further steps, and a recovered Handle can run the +// service to completion. +func TestService_CanceledTurn_RecoverRerunsTheService(t *testing.T) { + ctx := context.Background() + turnCtx, cancel := context.WithCancel(ctx) + defer cancel() + calls := 0 + fn := func(context.Context, state.ServiceCtx[svcCtx]) (any, error) { + calls++ + cancel() + return "late", nil + } + m := fetchMachine(fn) + store := durable.NewMemStore() + id := durable.InstanceID("fetch-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("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + invoke := state.InvokeID("fetch", "loading", 0) + if _, _, err = h.RunService(turnCtx, invoke); !errors.Is(err, context.Canceled) { + t.Fatalf("RunService = %v, want context.Canceled", err) + } + if _, err = h.Fire(ctx, "start"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Fire after the cancel = %v, want ErrHandleFailed", err) + } + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, ok, err := recovered.RunService(ctx, invoke); err != nil || !ok { + t.Fatalf("RunService after Recover = (%v, %v)", ok, err) + } + if calls != 2 { + t.Fatalf("service ran %d times, want 2", calls) + } +} + +// TestRecover_CanceledContext_DoesNotLoseARecordedResult proves a canceled Recover +// reports the cancellation rather than a Handle short of the journal. +func TestRecover_CanceledContext_DoesNotLoseARecordedResult(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchMachine(fn) + store := durable.NewMemStore() + id := durable.InstanceID("fetch-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("fetch", "loading", 0)) + canceled, cancel := context.WithCancel(ctx) + cancel() + if _, err = durable.Recover(canceled, m, store, id, opts...); !errors.Is(err, context.Canceled) { + t.Fatalf("Recover = %v, want context.Canceled", err) + } +} + +// TestStateAt_CanceledContext_ReportsIt proves a canceled historical read reports the +// cancellation, not a view short of the journal. +func TestStateAt_CanceledContext_ReportsIt(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := fetchMachine(fn) + store := durable.NewMemStore(durable.WithHistory()) + id := durable.InstanceID("fetch-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("fetch", "loading", 0)) + canceled, cancel := context.WithCancel(ctx) + cancel() + if _, err = durable.StateAt(canceled, m, store, id, 0, opts...); !errors.Is(err, context.Canceled) { + t.Fatalf("StateAt = %v, want context.Canceled", err) + } +} + +// TestRecover_CanceledContext_ChainedServices_ReportsIt proves a canceled replay of two +// chained services reports the cancellation, not a journal that looks corrupt. +func TestRecover_CanceledContext_ChainedServices_ReportsIt(t *testing.T) { + ctx := context.Background() + var calls int64 + fn := counterService(&calls) + m := chainMachine(fn, fn) + store := durable.NewMemStore() + id := durable.InstanceID("chain-1") + opts := []durable.Option[string, string, svcCtx]{durable.WithServiceRegistry[string, string, svcCtx](chainRegistry(fn, fn))} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, svcCtx{}, state.WithInitialState("first")) + if err != nil { + t.Fatalf("Start: %v", err) + } + mustRunService(t, h, state.InvokeID("chain", "first", 0)) + mustRunService(t, h, state.InvokeID("chain", "second", 0)) + canceled, cancel := context.WithCancel(ctx) + cancel() + if _, err = durable.Recover(canceled, m, store, id, opts...); !errors.Is(err, context.Canceled) { + t.Fatalf("Recover = %v, want context.Canceled", err) + } +} + +// TestTick_ContextEndedPartway_RecordsNothing proves a due timer the kernel refused isn't +// journaled: the Handle fails and the journal doesn't grow. +func TestTick_ContextEndedPartway_RecordsNothing(t *testing.T) { + ctx := context.Background() + m := timerMachine() + store := durable.NewMemStore() + id := durable.InstanceID("timer-1") + turn, cancel := context.WithCancel(ctx) + defer cancel() + clk := &cancelingClock{Clock: state.NewFakeClock(epoch)} + opts := []durable.Option[string, string, *timerCtx]{durable.WithRunnerClock[string, string, *timerCtx](clk)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, id, &timerCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "arm"); err != nil { + t.Fatalf("Fire(arm): %v", err) + } + _, before, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + clk.Clock.(*state.FakeClock).Advance(6 * time.Second) + clk.onNow = cancel // the turn ends as the Tick reads the clock + if _, err = h.Tick(turn); !errors.Is(err, durable.ErrHandleFailed) || !errors.Is(err, context.Canceled) { + t.Fatalf("Tick = %v, want ErrHandleFailed wrapping context.Canceled", err) + } + clk.onNow = nil + _, after, err := store.Load(ctx, id) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(after) != len(before) { + t.Fatalf("journal grew from %d to %d records", len(before), len(after)) + } +} + +// cancelingClock runs onNow once on the next clock read. +type cancelingClock struct { + state.Clock + onNow func() +} + +func (c *cancelingClock) Now() time.Time { + if c.onNow != nil { + f := c.onNow + c.onNow = nil + f() + } + return c.Clock.Now() +} diff --git a/durable/timetravel.go b/durable/timetravel.go index 4d5be559..654a28e7 100644 --- a/durable/timetravel.go +++ b/durable/timetravel.go @@ -101,7 +101,8 @@ func (v *TimeTravelView[S, E, C]) Step() int { return v.step } // historical read applies no side effect. // // step must be in [BaselineStep, lastRecordedStep]; a step outside that range -// reports ErrStepOutOfRange, and an unknown instance reports ErrInstanceNotFound. +// reports ErrStepOutOfRange, an unknown instance reports ErrInstanceNotFound, and an +// ended ctx reports its error. func StateAt[S comparable, E comparable, C any]( ctx context.Context, m *state.Machine[S, E, C], @@ -241,6 +242,11 @@ func replayThrough[S comparable, E comparable, C any]( } for i := range bounded { + // A canceled context can stop a replayed settle or tick with no error, leaving + // state short of the journal, so check it before each Record. + if err := ctx.Err(); err != nil { + return err + } rec := &bounded[i] switch { case hasActorEntry(rec): @@ -282,7 +288,9 @@ func replayThrough[S comparable, E comparable, C any]( // applies no side effect. } } - return nil + // A canceled context can stop a replayed settle or tick without an error reaching + // here, which would hand back a view short of the journal. + return ctx.Err() } // boundRecords returns the prefix of records whose first ordinal falls at or below