fix(durable): lock the Handle and refuse steps after a failed Append

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

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

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

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
 }
 

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)

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)
+	}
+}