diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 4680e2a..cec71a8 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 5067e0d..6d9a73d 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 f720c52..156b62a 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 17bed76..716c4dd 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 f8af0cd..ef3cd6f 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 00e5a8f..93355b2 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 ea357b3..928739f 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 9fe3f0c..73ea9af 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 a84886e..624051c 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 e092737..d61b1f7 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 df1d4db..bf4fb5f 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 4d5be55..654a28e 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