From fea7db33071fa968fc597e63507099a66d86a5e9 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Thu, 5 Nov 2026 09:00:40 -0700 Subject: [PATCH 1/2] fix(durable): lock the Handle and refuse steps after a failed Append Fire, Tick, RunService and DeliverToActor now hold a lock for the whole step, so a host's timer loop no longer loses acknowledged steps. A step that applied to the kernel but wasn't appended leaves the instance ahead of the journal, so the Handle refuses further steps with ErrHandleFailed until it's recovered. Relates to #214. Signed-off-by: Ray Karnes --- docs/src/content/docs/durable/overview.md | 10 + durable/CHANGELOG.md | 16 ++ durable/README.md | 10 + durable/actor.go | 20 +- durable/errors.go | 10 +- durable/runner.go | 85 ++++++-- durable/runner_test.go | 244 ++++++++++++++++++++++ durable/service.go | 20 +- 8 files changed, 391 insertions(+), 24 deletions(-) diff --git a/docs/src/content/docs/durable/overview.md b/docs/src/content/docs/durable/overview.md index 8dfd056b..4680e2a8 100644 --- a/docs/src/content/docs/durable/overview.md +++ b/docs/src/content/docs/durable/overview.md @@ -125,6 +125,16 @@ Time-travel needs the full record history through the target step. A `Store` opt in by implementing `HistoryStore` (the in-tree `MemStore` does so under `WithHistory`); `StateAt` otherwise falls back to the latest checkpoint plus tail. +## Handles across goroutines + +`Fire`, `Tick`, `RunService` and `DeliverToActor` are safe to call from different +goroutines: each holds the Handle's lock for the whole step. A host callback must not +call back into its Handle. `Instance()` is outside the lock. + +A step that may not have been recorded fails the Handle with `ErrHandleFailed`. `Recover` +a new Handle with the same options to continue, unless the cause is `ErrFenced`: then a +newer owner holds the instance, so stop. + ## A note on serialized payloads Events, service done-data, actor done-data, and actor messages are recorded as their diff --git a/durable/CHANGELOG.md b/durable/CHANGELOG.md index 543dc021..5067e0d0 100644 --- a/durable/CHANGELOG.md +++ b/durable/CHANGELOG.md @@ -25,9 +25,23 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. a refused payload, and `ErrRecordTooLarge` a Record too large for a `FileStore` line. `Record.EventData` and `Record.EventDataError` hold the payload. +- **`ErrHandleFailed`** reports a Handle out of step with its journal. + ### Changed - Requires Go 1.27.3. +- **A Handle's steps are serialized across goroutines.** `Fire`, `Tick`, `RunService` and + `DeliverToActor` hold a lock for the whole step, so a host's timer loop no longer loses + acknowledged steps. A host callback must not call back into its Handle, or it + deadlocks. `Instance()` isn't covered: read it only while no step is running. A `Tick`, + `RunService` or `DeliverToActor` whose context ended while it waited returns that error + and changes nothing. +- **A step that may not have been recorded fails the Handle.** That call and every later + one return `ErrHandleFailed`, wrapping the cause. `Recover` a new Handle with the same + options to continue. A service whose result wasn't recorded is in flight again, so + `RunService` runs it a second time. An `InvokeActor` actor restarts from its initial + state, and a `Spawn` actor doesn't return. If the cause is `ErrFenced`, a newer owner + holds the instance: stop. - **A `WithEventData` payload reaches `Assign` decoded, live as well as on replay.** A struct arrives as a map and an integer as a `float64`, which loses precision above 2^53, so an assertion to the original @@ -59,6 +73,8 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0. ### Fixed +- **A failed checkpoint no longer makes the next step conflict.** `Tick`, `RunService` + and `DeliverToActor` now advance the next step as soon as the append lands. - **The next step survives a checkpoint that compacts tick barriers.** A checkpoint now records the step that follows it, in a `nextStep` field of its envelope. Recovery used the Trace count, which runs low past a tick barrier, so the next Fire could reuse a diff --git a/durable/README.md b/durable/README.md index 9b9858c6..f720c52f 100644 --- a/durable/README.md +++ b/durable/README.md @@ -136,6 +136,16 @@ opts in by implementing `HistoryStore` (the in-tree `MemStore` does so under `WithHistory`); `StateAt` falls back to `Load` (latest checkpoint plus tail) otherwise. +## Handles across goroutines + +`Fire`, `Tick`, `RunService` and `DeliverToActor` are safe to call from different +goroutines: each holds the Handle's lock for the whole step. A host callback must not +call back into its Handle. `Instance()` is outside the lock. + +A step that may not have been recorded fails the Handle with `ErrHandleFailed`. `Recover` +a new Handle with the same options to continue, unless the cause is `ErrFenced`: then a +newer owner holds the instance, so stop. + ## Caveat: serialized payloads Events, service done-data, actor done-data, and actor messages are recorded as their diff --git a/durable/actor.go b/durable/actor.go index 792ba00d..41abf380 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -99,10 +99,21 @@ type actorMessagePayload struct { // re-instantiating the actor. // // It returns whether the actor was found running (false for a ref that names no -// live actor — already settled, stopped, or never spawned), and a persistence -// failure as the error. The kernel transitions the delivery drove do not -// themselves fail. +// live actor — already settled, stopped, or never spawned — or with an error when the +// Handle has failed or ctx has ended), and a persistence failure as the error. The +// kernel transitions the delivery drove do not themselves fail. After a failed Append +// the result wasn't recorded: on a recovered Handle an InvokeActor actor restarts from its +// initial state, and a Spawn actor doesn't return. Resending a recorded delivery repeats +// any message it sends the parent. func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef, event any) (bool, error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return false, err + } + if err := ctx.Err(); err != nil { + return false, err + } if h.actors == nil { return false, fmt.Errorf("durable: no actor palette wired for %q (supply WithActorPalette)", h.id) } @@ -116,7 +127,7 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef output, settled, settleErr := h.actorOutcome() entries, err := recordActorTransitions(ref.ID, h.inst.History(), before, output, settled, settleErr) if err != nil { - return true, fmt.Errorf("durable: recording actor %q result for %q: %w", ref.ID, h.id, err) + return true, h.fail(fmt.Errorf("durable: recording actor %q result for %q: %w", ref.ID, h.id, err)) } if h.actorBuf != nil { *h.actorBuf = append(*h.actorBuf, entries...) @@ -127,7 +138,6 @@ func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef if err := h.persistStep(ctx, step, &rec); err != nil { return true, err } - h.nextStep++ return true, nil } diff --git a/durable/errors.go b/durable/errors.go index d08e5f47..5ac19d7e 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -16,6 +16,7 @@ const ( errorStepOutOfRange = "time-travel step out of range" errorEventDataEncoding = "event data can't be recorded" errorRecordTooLarge = "record too large for the journal" + errorHandleFailed = "handle is out of step with the journal" ) // Sentinel errors a Store reports. Callers match them with errors.Is. @@ -49,9 +50,16 @@ var ( // ErrRecordTooLarge is reported by FileStore.Append for a Record whose journal line // exceeds the store's line limit, which a later Load could not read back, even // without its Snapshot. No journal line is written, but a Handle's kernel has already - // applied the step. + // applied the step, so the Handle fails with ErrHandleFailed. ErrRecordTooLarge = errors.New(errorPrefix + errorRecordTooLarge) + // ErrHandleFailed is reported by a Handle's Fire, Tick, RunService and + // DeliverToActor once a step applied to the live instance but may not have been + // recorded, on that call and every later one. It wraps the cause. Recover a new + // Handle with the same options to continue, unless the cause is ErrFenced: then a + // newer owner holds the instance, so stop. + ErrHandleFailed = errors.New(errorPrefix + errorHandleFailed) + // ErrCheckpointNotAdvancing is reported by Store.Checkpoint when throughStep // does not advance beyond the instance's current checkpoint. ErrCheckpointNotAdvancing = errors.New(errorPrefix + errorCheckpointNotAdvancing) diff --git a/durable/runner.go b/durable/runner.go index e77e5a11..442fec6f 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sync" "time" "github.com/stablekernel/crucible/state" @@ -76,11 +77,33 @@ func NewRunner[S comparable, E comparable, C any](m *state.Machine[S, E, C], st // clock that journals every reading the scheduler consumes, so timer-driven // transitions are durable and replay wall-clock-independent. Obtain a Handle from // Runner.Start or Recover. +// +// Fire, Tick, RunService and DeliverToActor are safe to call from different +// goroutines: each holds the Handle's lock for the whole step, including the Store +// write, effect dispatch and any service it runs. Any host callback run during a step +// (an effect handler, service, guard, action, reducer or actor behavior) must not call +// back into the same Handle, or it deadlocks. A step waits for the lock without watching +// its context, so a slow service delays other steps past their deadlines; a Tick, +// RunService or DeliverToActor whose context has ended by then returns that error and +// changes nothing, and the kernel refuses a Fire. Instance returns the +// live instance outside that lock, so read it only while no step is running. A +// stateless Runner.Fire builds its own Handle, so it isn't serialized with this one. +// +// A step that applied to the kernel but may not have been recorded leaves the Handle out +// of step with the journal. That call and every later one return ErrHandleFailed, +// wrapping the cause. Recover a new Handle with the same options to continue. A service +// whose result wasn't recorded is in flight again, so RunService runs it a second time. +// A running InvokeActor actor restarts from its initial state on any Recover, losing every +// delivery so far, and a Spawn actor doesn't return. If the cause is ErrFenced, a +// newer owner holds the instance: stop, because recovering won't help. type Handle[S comparable, E comparable, C any] struct { - runner *Runner[S, E, C] - id InstanceID - inst *state.Instance[S, E, C] + runner *Runner[S, E, C] + id InstanceID + inst *state.Instance[S, E, C] + + mu sync.Mutex // serializes the steps, and guards the instance and every field below nextStep int + failed error // the error that took the Handle out of step with the journal, nil until then sched *state.Scheduler[S, E, C] recClock *recordingClock // owns the clock-read accumulator and its mutex @@ -93,6 +116,21 @@ type Handle[S comparable, E comparable, C any] struct { actorBuf *[]state.JournalEntry // recording actor-transition accumulator, flushed per step } +// refuse reports ErrHandleFailed once the Handle is out of step with the journal. +func (h *Handle[S, E, C]) refuse() error { + if h.failed == nil { + return nil + } + return fmt.Errorf("%w: %q: %w", ErrHandleFailed, h.id, h.failed) +} + +// fail records err as the reason the Handle is out of step with the journal, and +// returns the same refusal every later step gets. +func (h *Handle[S, E, C]) fail(err error) error { + h.failed = err + return h.refuse() +} + // drainClock returns and clears the clock readings accumulated since the last // drain, so each recorded step carries exactly the readings consumed during it. // The drain is taken under the recording clock's mutex, so closing a step's @@ -227,7 +265,8 @@ func (h *Handle[S, E, C]) dropFiredTimers(now time.Time) { // Instance returns the underlying kernel Instance the Handle wraps, for reads // such as Configuration, Snapshot, or Current. Drive it through the Handle's Fire // (or the Runner's) so steps continue to be recorded; firing the bare Instance -// bypasses durability. +// bypasses durability. Read it only while no step is running: it's outside the Handle's +// lock. After ErrHandleFailed it shows state the journal may not hold. func (h *Handle[S, E, C]) Instance() *state.Instance[S, E, C] { return h.inst } // ID returns the InstanceID the Handle records under. @@ -343,7 +382,8 @@ func (r *Runner[S, E, C]) newActorSystem(inst *state.Instance[S, E, C]) *state.A // how a state.WithEventData payload is handled, and when it's refused. It is the // stateless entry point (no Handle required); for a hot path that fires many // events in sequence, hold a Handle from Start or Recover and use Handle.Fire to -// avoid reloading between steps. +// avoid reloading between steps. Calls on one id aren't serialized, and must not overlap +// each other or a live Handle. func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts ...state.FireOption) (state.FireResult[S], error) { h, err := r.recover(ctx, id) if err != nil { @@ -370,6 +410,11 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // kernel runs. The payload is read on the calling goroutine with no lock of its own, so // don't mutate it concurrently with Fire. func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireOption) (state.FireResult[S], error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return state.FireResult[S]{}, err + } // Encode the WithEventData payload before the kernel runs, so a payload that // can't be recorded refuses the Fire rather than applying a step it can't record. payload, opts, err := encodeEventData(opts, h.runner.cfg.maxEventData) @@ -395,7 +440,7 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO des := dispatchableEffects(step, res.Effects) effEnvs, err := recordEffects(des) if err != nil { - return res, err + return res, h.fail(err) } rec := Record{Step: step, Event: []byte(res.Trace.EventPayload), EventData: payload.data, EventDataError: payload.errMsg, Entries: h.drainClock(), Effects: effEnvs} // Write-ahead append the step Record (with its effect ids) BEFORE dispatching, but @@ -403,7 +448,7 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO // a crash mid-dispatch leaves the effect recorded in the still-present tail for // recovery to redispatch — never orphaned behind the compaction. if err := h.appendStep(ctx, step, &rec); err != nil { - return res, err + return res, h.fail(err) } h.nextStep++ // Dispatch AFTER the append: a crash in this window leaves the effect recorded @@ -429,6 +474,16 @@ func (h *Handle[S, E, C]) Fire(ctx context.Context, event E, opts ...state.FireO // count of timers fired — so recovery re-derives the same timers at their // recorded instants. It returns the FireResults of the timers it fired, in order. func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return nil, err + } + // A context that ended in the lock queue must stop here: the scheduler would consume + // a due timer whose Fire the kernel then refuses. + if err := ctx.Err(); err != nil { + return nil, err + } results := h.sched.Tick(ctx) step := h.nextStep @@ -450,20 +505,24 @@ func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], erro if len(rec.Entries) == 0 && rec.TickSteps == 0 { return results, nil } - if err := h.persistStep(ctx, step, &rec); err != nil { + if err := h.appendStep(ctx, step, &rec); err != nil { + return results, h.fail(err) + } + h.nextStep = rec.followingStep() + if err := h.checkpointStep(ctx, step, &rec); err != nil { return results, err } - h.nextStep += len(results) + 1 return results, nil } -// persistStep checkpoints if the policy is due, then write-ahead appends rec -// before the step is acknowledged to the caller. The checkpoint is taken at the -// barrier's own step so a Load after it returns this Snapshot plus the later tail. +// persistStep write-ahead appends rec, advances the next step, then checkpoints if +// the policy is due. A failed append fails the Handle, because the kernel already +// applied the step; a failed checkpoint doesn't, because the Record is durable. func (h *Handle[S, E, C]) persistStep(ctx context.Context, step int, rec *Record) error { if err := h.appendStep(ctx, step, rec); err != nil { - return err + return h.fail(err) } + h.nextStep = rec.followingStep() return h.checkpointStep(ctx, step, rec) } diff --git a/durable/runner_test.go b/durable/runner_test.go index 2d33819a..bc5b3f7b 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -4,6 +4,10 @@ import ( "context" "encoding/json" "errors" + "runtime" + "strings" + "sync" + "sync/atomic" "testing" "time" @@ -511,6 +515,21 @@ func quantityMachine() *state.Machine[string, string, *runCtx] { Quench() } +// flakyAppendStore fails the first Append of one step, then delegates. +type flakyAppendStore struct { + durable.Store + failStep int + failed bool +} + +func (s *flakyAppendStore) Append(ctx context.Context, id durable.InstanceID, rec durable.Record, opts ...durable.AppendOption) (int64, error) { + if rec.Step == s.failStep && !s.failed { + s.failed = true + return 0, errors.New("store unavailable") + } + return s.Store.Append(ctx, id, rec, opts...) +} + // TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory proves // two Handles on one instance (a stalled worker, and the worker that recovered the // instance after it) cannot both have their Fires acknowledged while the journal @@ -552,6 +571,71 @@ func TestRecover_TwoHandlesOnOneInstance_JournalMatchesAnAcknowledgedHistory(t * } } +// TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps proves the "Tick from its +// own timer loop" pattern that drainClock documents holds while requests Fire on +// the same Handle: the recovered instance matches the live one, and every +// acknowledged Fire survives. +func TestRunner_FireDuringTimerLoop_LosesNoAcknowledgedSteps(t *testing.T) { + if runtime.GOMAXPROCS(0) < 2 { + t.Skip("needs two CPUs to interleave Fire and Tick") + } + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + id := durable.InstanceID("poll-1") + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{durable.WithRunnerClock[string, string, *runCtx](clk)} + runner := durable.NewRunner(m, store, opts...) + h, err := runner.Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + + const n = 200 + var acked, fired, failed atomic.Int64 + var wg sync.WaitGroup + wg.Add(2) + go func() { // the host's timer loop + defer wg.Done() + for range n { + clk.Advance(time.Second) + results, terr := h.Tick(ctx) + if terr != nil { + failed.Add(1) + } + fired.Add(int64(len(results))) + } + }() + go func() { // request traffic + defer wg.Done() + for range n { + if _, ferr := h.Fire(ctx, "poke"); ferr != nil { + failed.Add(1) + continue + } + acked.Add(1) + } + }() + wg.Wait() + if failed.Load() != 0 || acked.Load() != n || fired.Load() == 0 { + t.Fatalf("failed %d calls, acknowledged %d of %d Fires, fired %d timers", failed.Load(), acked.Load(), n, fired.Load()) + } + + recovered, err := durable.Recover(ctx, m, store, id, opts...) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got, want := recovered.Instance().Entity().Count, int(acked.Load()); got != want { + t.Fatalf("recovered %d pokes, want %d acknowledged", got, want) + } + if live, got := handleSnapshot(t, h), handleSnapshot(t, recovered); got != live { + t.Fatalf("recovered instance differs from live:\n live: %s\n got: %s", live, got) + } +} + // TestRecover_FireWithEventData_MatchesLive proves an instance driven with // state.WithEventData recovers to the live state, because the Handle journals the data. func TestRecover_FireWithEventData_MatchesLive(t *testing.T) { @@ -629,6 +713,36 @@ func TestRecover_StepAfterTickCheckpoint_IsPersisted(t *testing.T) { } } +// TestRunner_FireAfterFailedAppend_MatchesJournal proves a Fire whose Append failed +// acknowledges nothing further: the Handle refuses, and recovery matches the journal. +func TestRunner_FireAfterFailedAppend_MatchesJournal(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := &flakyAppendStore{Store: durable.NewMemStore(), failStep: 1} + id := durable.InstanceID("poll-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { // step 1: the Append fails + t.Fatalf("Fire(poke) = %v, want ErrHandleFailed", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("second Fire(poke) = %v, want ErrHandleFailed", err) + } + + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := recovered.Instance().Entity().Count; got != 0 { + t.Fatalf("recovered %d pokes, want 0: none was recorded", got) + } +} + // lastStep returns the Step of the newest Record in a history-keeping store. func lastStep(t *testing.T, store *durable.MemStore, id durable.InstanceID) int { t.Helper() @@ -725,3 +839,133 @@ func TestRecover_CheckpointWithoutNextStep_UsesTheTraceCount(t *testing.T) { t.Fatalf("poke recorded at step %d, want 3", got) } } + +// TestHandle_FailedAppend_RefusesEveryStep proves the failed Handle stays refused, +// names the cause, and that a recovered Handle continues from the journal. +func TestHandle_FailedAppend_RefusesEveryStep(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := &flakyAppendStore{Store: durable.NewMemStore(), failStep: 1} + id := durable.InstanceID("poll-1") + h, err := durable.NewRunner(m, store).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Fire(poke) = %v, want ErrHandleFailed", err) + } + if _, err = h.Fire(ctx, "poke"); !errors.Is(err, durable.ErrHandleFailed) || !strings.Contains(err.Error(), "store unavailable") { + t.Fatalf("Fire after the failure = %v, want ErrHandleFailed naming the cause", err) + } + if _, err = h.Tick(ctx); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Tick after the failure = %v, want ErrHandleFailed", err) + } + recovered, err := durable.Recover(ctx, m, store, id) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = recovered.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire on the recovered Handle: %v", err) + } +} + +// failingCheckpointStore fails the next Checkpoint after armed is set. +type failingCheckpointStore struct { + durable.Store + armed bool +} + +func (s *failingCheckpointStore) Checkpoint(ctx context.Context, id durable.InstanceID, snapshot []byte, throughStep int, opts ...durable.CheckpointOption) error { + if s.armed { + s.armed = false + return errors.New("checkpoint unavailable") + } + return s.Store.Checkpoint(ctx, id, snapshot, throughStep, opts...) +} + +// TestHandle_FailedCheckpointOnTick_KeepsGoing proves a failed checkpoint doesn't fail +// the Handle or make the next step conflict: the Record is durable. +func TestHandle_FailedCheckpointOnTick_KeepsGoing(t *testing.T) { + ctx := context.Background() + store := &failingCheckpointStore{Store: durable.NewMemStore()} + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{ + durable.WithRunnerClock[string, string, *runCtx](clk), + durable.WithCheckpointEvery[string, string, *runCtx](2), + } + h, err := durable.NewRunner(pollingMachine(), store, opts...).Start(ctx, "poll-1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + store.armed = true + clk.Advance(time.Second) + if _, err = h.Tick(ctx); err == nil || errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Tick = %v, want the plain checkpoint error", err) + } + if _, err = h.Fire(ctx, "poke"); err != nil { + t.Fatalf("Fire after the failed checkpoint: %v", err) + } +} + +// TestHandle_FencedOwner_Stops proves a Handle that lost its instance reports the fence +// through ErrHandleFailed, so a host can tell it must stop rather than recover. +func TestHandle_FencedOwner_Stops(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + const id = durable.InstanceID("poll-1") + owner := func(token int64) durable.Option[string, string, *runCtx] { + return durable.WithOwnerFence[string, string, *runCtx](token) + } + old, err := durable.NewRunner(m, store, owner(1)).Start(ctx, id, &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = old.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + newer, err := durable.Recover(ctx, m, store, id, owner(2)) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if _, err = newer.Fire(ctx, "poke"); err != nil { + t.Fatalf("newer Fire: %v", err) + } + _, err = old.Fire(ctx, "poke") + if !errors.Is(err, durable.ErrHandleFailed) || !errors.Is(err, durable.ErrFenced) { + t.Fatalf("old Fire = %v, want ErrHandleFailed wrapping ErrFenced", err) + } +} + +// TestHandle_ContextEndedBeforeTheStep_ChangesNothing proves an ended context stops a +// Tick before the scheduler consumes a due timer. +func TestHandle_ContextEndedBeforeTheStep_ChangesNothing(t *testing.T) { + ctx := context.Background() + m := pollingMachine() + store := durable.NewMemStore() + clk := state.NewFakeClock(epoch) + opts := []durable.Option[string, string, *runCtx]{durable.WithRunnerClock[string, string, *runCtx](clk)} + h, err := durable.NewRunner(m, store, opts...).Start(ctx, "poll-1", &runCtx{}, state.WithInitialState("idle")) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err = h.Fire(ctx, "start"); err != nil { + t.Fatalf("Fire(start): %v", err) + } + clk.Advance(time.Second) + ended, cancel := context.WithCancel(ctx) + cancel() + if _, err = h.Tick(ended); !errors.Is(err, context.Canceled) { + t.Fatalf("Tick = %v, want context.Canceled", err) + } + results, err := h.Tick(ctx) + if err != nil || len(results) != 1 { + t.Fatalf("Tick after = (%d results, %v), want the timer still due", len(results), err) + } +} diff --git a/durable/service.go b/durable/service.go index fedfe318..aacb91bc 100644 --- a/durable/service.go +++ b/durable/service.go @@ -84,10 +84,21 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { // re-invoking the service. // // It returns the routed FireResult and true, or the zero result and false when id -// names no in-flight service (already settled, stopped, or never started). A -// persistence failure is returned as the error; the kernel transition itself does -// not fail (a settled service always routes onDone or onError). +// names no in-flight service (already settled, stopped, or never started), or false +// and the error when the Handle has failed or ctx has ended. A persistence failure is +// returned as the error; the kernel transition itself does not fail (a settled service +// always routes onDone or onError). After a failed Append the result wasn't recorded, so +// a recovered Handle has the service in flight again, and RunService runs it a second +// time. func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.FireResult[S], bool, error) { + h.mu.Lock() + defer h.mu.Unlock() + if err := h.refuse(); err != nil { + return state.FireResult[S]{}, false, err + } + if err := ctx.Err(); err != nil { + return state.FireResult[S]{}, false, err + } if h.svc == nil { return state.FireResult[S]{}, false, fmt.Errorf("durable: no service registry wired for %q (supply WithServiceRegistry)", h.id) } @@ -109,7 +120,7 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire entry, err := recordServiceOutcome(id, h.svc.LastResult, h.svc.LastError()) if err != nil { - return res, true, fmt.Errorf("durable: recording service %q result for %q: %w", id, h.id, err) + return res, true, h.fail(fmt.Errorf("durable: recording service %q result for %q: %w", id, h.id, err)) } if h.svcBuf != nil { *h.svcBuf = append(*h.svcBuf, entry) @@ -120,7 +131,6 @@ func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.Fire if err := h.persistStep(ctx, step, &rec); err != nil { return res, true, err } - h.nextStep++ return res, true, nil } From 2f8b57c886d7f85c4937611d3c48deefe600e005 Mon Sep 17 00:00:00 2001 From: Ray Karnes Date: Mon, 9 Nov 2026 11:04:01 -0700 Subject: [PATCH 2/2] test(durable): cover the failed-Handle latch on RunService and DeliverToActor RunService and DeliverToActor refuse steps after a failed Append and return an ended context's error, but only Fire and Tick were tested. New tests fail an Append under each and end the context before each, and check nothing ran or was journaled. The Tick context test now also checks that nothing was appended. The RunService and DeliverToActor docs said a failed step returns false; they return true with ErrHandleFailed once the service settled or the actor received the event. The docs now say so, name a context that ends during the Store write as a possible cause, and note that the latch relies on the Append running inline. Signed-off-by: Ray Karnes --- durable/actor.go | 14 +-- durable/errors.go | 2 +- durable/handle_failure_test.go | 164 +++++++++++++++++++++++++++++++++ durable/runner.go | 10 +- durable/runner_test.go | 6 +- durable/service.go | 10 +- 6 files changed, 190 insertions(+), 16 deletions(-) create mode 100644 durable/handle_failure_test.go diff --git a/durable/actor.go b/durable/actor.go index 41abf380..17bed76b 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -98,12 +98,14 @@ type actorMessagePayload struct { // JournalActorMessage correlated by the actor id so recovery replays it without // re-instantiating the actor. // -// It returns whether the actor was found running (false for a ref that names no -// live actor — already settled, stopped, or never spawned — or with an error when the -// Handle has failed or ctx has ended), and a persistence failure as the error. The -// kernel transitions the delivery drove do not themselves fail. After a failed Append -// the result wasn't recorded: on a recovered Handle an InvokeActor actor restarts from its -// initial state, and a Spawn actor doesn't return. Resending a recorded delivery repeats +// 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. func (h *Handle[S, E, C]) DeliverToActor(ctx context.Context, ref state.ActorRef, event any) (bool, error) { h.mu.Lock() diff --git a/durable/errors.go b/durable/errors.go index 5ac19d7e..ea357b31 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -19,7 +19,7 @@ const ( errorHandleFailed = "handle is out of step with the journal" ) -// Sentinel errors a Store reports. Callers match them with errors.Is. +// Sentinel errors a Store or a Handle reports. Callers match them with errors.Is. var ( // ErrInstanceNotFound is reported by Store.Load for an instance that has // never been written. diff --git a/durable/handle_failure_test.go b/durable/handle_failure_test.go new file mode 100644 index 00000000..ed57686c --- /dev/null +++ b/durable/handle_failure_test.go @@ -0,0 +1,164 @@ +package durable_test + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + + "github.com/stablekernel/crucible/durable" + "github.com/stablekernel/crucible/state" +) + +// fetchID is the invocation id of the fetch service the single-service machine runs. +var fetchID = state.InvokeID("fetch", "loading", 0) + +// armedAppendStore counts Appends and, once armed, fails the next one. +type armedAppendStore struct { + durable.Store + armed bool + appends int +} + +func (s *armedAppendStore) Append(ctx context.Context, id durable.InstanceID, rec durable.Record, opts ...durable.AppendOption) (int64, error) { + if s.armed { + s.armed = false + return 0, errors.New("store unavailable") + } + s.appends++ + return s.Store.Append(ctx, id, rec, opts...) +} + +// startFetch starts the single-service machine over store and moves it into the +// state that invokes the service. +func startFetch(t *testing.T, store durable.Store, calls *int64) *durable.Handle[string, string, svcCtx] { + t.Helper() + ctx := context.Background() + fn := counterService(calls) + h, err := durable.NewRunner(fetchMachine(fn), store, + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))). + Start(ctx, "fetch-1", 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) + } + return h +} + +// startTwin starts the two-actor machine over store, with childA running. +func startTwin(t *testing.T, store durable.Store, runs *int64) (*durable.Handle[string, string, actorCtx], state.ActorRef) { + t.Helper() + palette := map[string]state.ActorBehavior{"childA": counterChild(runs), "childB": counterChild(runs)} + h, err := durable.NewRunner(twoActorMachine(), store, + durable.WithActorPalette[string, string, actorCtx](palette)). + Start(context.Background(), "twin-1", actorCtx{}, state.WithInitialState("first")) + if err != nil { + t.Fatalf("Start: %v", err) + } + ref, ok := h.ActorRef(state.ActorID("twin", "first", 0)) + if !ok { + t.Fatal("no ref for childA") + } + return h, ref +} + +// TestHandle_RunService_FailedAppend_FailsHandle proves a failed Append of a service +// outcome fails the Handle: RunService reports the settle with ErrHandleFailed, every +// later step is refused, and a recovered Handle runs the service again. +func TestHandle_RunService_FailedAppend_FailsHandle(t *testing.T) { + ctx := context.Background() + var calls int64 + store := &armedAppendStore{Store: durable.NewMemStore()} + h := startFetch(t, store, &calls) + + store.armed = true + _, ok, err := h.RunService(ctx, fetchID) + if !ok || !errors.Is(err, durable.ErrHandleFailed) || !strings.Contains(err.Error(), "store unavailable") { + t.Fatalf("RunService = (%v, %v), want true and ErrHandleFailed naming the cause", ok, err) + } + if _, ok, err = h.RunService(ctx, fetchID); ok || !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("RunService after the failure = (%v, %v), want false and ErrHandleFailed", ok, err) + } + if _, err = h.Fire(ctx, "start"); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Fire after the failure = %v, want ErrHandleFailed", err) + } + + fn := counterService(&calls) + rh, err := durable.Recover(ctx, fetchMachine(fn), store, "fetch-1", + durable.WithServiceRegistry[string, string, svcCtx](fetchRegistry(fn))) + if err != nil { + t.Fatalf("Recover: %v", err) + } + if got := rh.Instance().Current(); got != "loading" { + t.Fatalf("recovered state = %q, want loading (the outcome wasn't recorded)", got) + } + mustRunService(t, rh, fetchID) + if got := atomic.LoadInt64(&calls); got != 2 { + t.Fatalf("service ran %d times, want 2 (once live, once after recovery)", got) + } +} + +// TestHandle_DeliverToActor_FailedAppend_FailsHandle proves a failed Append of an +// actor delivery fails the Handle: DeliverToActor reports the delivery with +// ErrHandleFailed, and every later step is refused. +func TestHandle_DeliverToActor_FailedAppend_FailsHandle(t *testing.T) { + ctx := context.Background() + var runs int64 + store := &armedAppendStore{Store: durable.NewMemStore()} + h, ref := startTwin(t, store, &runs) + + store.armed = true + delivered, err := h.DeliverToActor(ctx, ref, "finish") + if !delivered || !errors.Is(err, durable.ErrHandleFailed) || !strings.Contains(err.Error(), "store unavailable") { + t.Fatalf("DeliverToActor = (%v, %v), want true and ErrHandleFailed naming the cause", delivered, err) + } + if delivered, err = h.DeliverToActor(ctx, ref, "finish"); delivered || !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("DeliverToActor after the failure = (%v, %v), want false and ErrHandleFailed", delivered, err) + } + if _, err = h.Tick(ctx); !errors.Is(err, durable.ErrHandleFailed) { + t.Fatalf("Tick after the failure = %v, want ErrHandleFailed", err) + } +} + +// TestHandle_RunService_ContextEnded_ChangesNothing proves an ended context stops +// RunService before the service runs or anything is journaled. +func TestHandle_RunService_ContextEnded_ChangesNothing(t *testing.T) { + var calls int64 + store := &armedAppendStore{Store: durable.NewMemStore()} + h := startFetch(t, store, &calls) + before := store.appends + + ended, cancel := context.WithCancel(context.Background()) + cancel() + if _, ok, err := h.RunService(ended, fetchID); ok || !errors.Is(err, context.Canceled) { + t.Fatalf("RunService = (%v, %v), want false and context.Canceled", ok, err) + } + if got := atomic.LoadInt64(&calls); got != 0 || store.appends != before { + t.Fatalf("after an ended context: %d service runs and %d new appends, want 0 and 0", got, store.appends-before) + } + mustRunService(t, h, fetchID) +} + +// TestHandle_DeliverToActor_ContextEnded_ChangesNothing proves an ended context stops +// DeliverToActor before the actor sees the event or anything is journaled. +func TestHandle_DeliverToActor_ContextEnded_ChangesNothing(t *testing.T) { + var runs int64 + store := &armedAppendStore{Store: durable.NewMemStore()} + h, ref := startTwin(t, store, &runs) + before, state0 := store.appends, h.Instance().Current() + + ended, cancel := context.WithCancel(context.Background()) + cancel() + if delivered, err := h.DeliverToActor(ended, ref, "finish"); delivered || !errors.Is(err, context.Canceled) { + t.Fatalf("DeliverToActor = (%v, %v), want false and context.Canceled", delivered, err) + } + if store.appends != before || h.Instance().Current() != state0 { + t.Fatalf("after an ended context: %d new appends, state %q, want 0 and %q", store.appends-before, h.Instance().Current(), state0) + } + if delivered, err := h.DeliverToActor(context.Background(), ref, "finish"); !delivered || err != nil { + t.Fatalf("DeliverToActor after = (%v, %v), want the actor still running", delivered, err) + } +} diff --git a/durable/runner.go b/durable/runner.go index 442fec6f..9fe3f0ce 100644 --- a/durable/runner.go +++ b/durable/runner.go @@ -91,7 +91,8 @@ func NewRunner[S comparable, E comparable, C any](m *state.Machine[S, E, C], st // // A step that applied to the kernel but may not have been recorded leaves the Handle out // of step with the journal. That call and every later one return ErrHandleFailed, -// wrapping the cause. Recover a new Handle with the same options to continue. A service +// 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 @@ -396,7 +397,7 @@ func (r *Runner[S, E, C]) Fire(ctx context.Context, id InstanceID, event E, opts // it Fires the kernel, appends a Record carrying the driving event at the // step's ordinal (write-ahead, before returning), and — when the // checkpoint policy is due — persists a full Snapshot and compacts the tail. A -// kernel transition error is recorded as a no-op (no step was produced) and +// kernel transition error journals nothing (no step was produced) and is // returned to the caller. // // A state.WithEventData payload reaches the transition's Assign in its JSON-decoded @@ -517,7 +518,10 @@ func (h *Handle[S, E, C]) Tick(ctx context.Context) ([]state.FireResult[S], erro // persistStep write-ahead appends rec, advances the next step, then checkpoints if // the policy is due. A failed append fails the Handle, because the kernel already -// applied the step; a failed checkpoint doesn't, because the Record is durable. +// applied the step; a failed checkpoint doesn't, because the Record is durable. This +// relies on the Append running inline, under the Handle's lock, before the step +// returns: an Append deferred off this goroutine would acknowledge a step whose +// outcome isn't known, and the Handle couldn't fail. func (h *Handle[S, E, C]) persistStep(ctx context.Context, step int, rec *Record) error { if err := h.appendStep(ctx, step, rec); err != nil { return h.fail(err) diff --git a/durable/runner_test.go b/durable/runner_test.go index bc5b3f7b..5318f305 100644 --- a/durable/runner_test.go +++ b/durable/runner_test.go @@ -948,7 +948,7 @@ func TestHandle_FencedOwner_Stops(t *testing.T) { func TestHandle_ContextEndedBeforeTheStep_ChangesNothing(t *testing.T) { ctx := context.Background() m := pollingMachine() - store := durable.NewMemStore() + store := &armedAppendStore{Store: durable.NewMemStore()} clk := state.NewFakeClock(epoch) opts := []durable.Option[string, string, *runCtx]{durable.WithRunnerClock[string, string, *runCtx](clk)} h, err := durable.NewRunner(m, store, opts...).Start(ctx, "poll-1", &runCtx{}, state.WithInitialState("idle")) @@ -959,11 +959,15 @@ func TestHandle_ContextEndedBeforeTheStep_ChangesNothing(t *testing.T) { t.Fatalf("Fire(start): %v", err) } clk.Advance(time.Second) + before := store.appends ended, cancel := context.WithCancel(ctx) cancel() if _, err = h.Tick(ended); !errors.Is(err, context.Canceled) { t.Fatalf("Tick = %v, want context.Canceled", err) } + if store.appends != before { + t.Fatalf("Tick on an ended context appended %d Records, want 0", store.appends-before) + } results, err := h.Tick(ctx) if err != nil || len(results) != 1 { t.Fatalf("Tick after = (%d results, %v), want the timer still due", len(results), err) diff --git a/durable/service.go b/durable/service.go index aacb91bc..e0927377 100644 --- a/durable/service.go +++ b/durable/service.go @@ -85,11 +85,11 @@ func (h *Handle[S, E, C]) drainServices() []state.JournalEntry { // // 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 has failed or ctx has ended. A persistence failure is -// returned as the error; the kernel transition itself does not fail (a settled service -// always routes onDone or onError). After a failed Append the result wasn't recorded, so -// a recovered Handle has the service in flight again, and RunService runs it a second -// time. +// 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. func (h *Handle[S, E, C]) RunService(ctx context.Context, id string) (state.FireResult[S], bool, error) { h.mu.Lock() defer h.mu.Unlock()