Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions docs/src/content/docs/durable/overview.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand Down
7 changes: 7 additions & 0 deletions durable/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 4 additions & 3 deletions durable/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand Down
40 changes: 21 additions & 19 deletions durable/actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand All @@ -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)
Expand Down
42 changes: 42 additions & 0 deletions durable/actor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package durable_test

import (
"context"
"errors"
"fmt"
"sync/atomic"
"testing"
Expand Down Expand Up @@ -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)
}
}
23 changes: 12 additions & 11 deletions durable/doc.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
//
Expand Down
3 changes: 2 additions & 1 deletion durable/errors.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
44 changes: 32 additions & 12 deletions durable/runner.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading