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..17bed76b 100644 --- a/durable/actor.go +++ b/durable/actor.go @@ -98,11 +98,24 @@ 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), and a persistence -// failure as the error. The kernel transitions the delivery drove do not -// themselves fail. +// 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() + 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 +129,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 +140,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..ea357b31 100644 --- a/durable/errors.go +++ b/durable/errors.go @@ -16,9 +16,10 @@ 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. +// 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. @@ -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/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 e77e5a11..9fe3f0ce 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,34 @@ 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, 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 +// 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 +117,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 +266,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 +383,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 { @@ -356,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 @@ -370,6 +411,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 +441,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 +449,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 +475,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 +506,27 @@ 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. 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 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..5318f305 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,137 @@ 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 := &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")) + 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) + 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 fedfe318..e0927377 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 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() + 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 }