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

docs/src/content/docs/durable/overview.md

@@ -91,6 +91,7 @@ Two stdlib-only reference implementations ship in-tree:
 - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance
   subdirectories, each an append-only journal, an atomic checkpoint, an
   idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file.
+  `Append` refuses a Record over its 16 MiB line limit with `ErrRecordTooLarge`.
   Each append flushes to stable storage; each checkpoint uses
   write-temp-plus-rename for crash-safe atomicity.
   Use it for durability across restarts without a database.
@@ -124,10 +125,25 @@ 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 JSON form. A parent reducer that type-asserts a non-JSON Go type from
-`AssignCtx.Event` observes the JSON-decoded shape on a replayed `onDone`. A
+Events, service done-data, actor done-data, and actor messages are recorded as their
+JSON form. A parent reducer that type-asserts a non-JSON Go type from `AssignCtx.Event`
+observes the JSON-decoded shape on a replayed `onDone`. A `WithEventData` payload is
+JSON-decoded live as well as on replay, so a reducer gets the same value both ways: a
+struct is a map and an integer a `float64`. An error stays an error with the same
+message, with any invalid UTF-8 replaced. `Handle.Fire` refuses a payload it can't
+record, or one over `WithMaxEventDataBytes` (8 MiB by default), with
+`ErrEventDataEncoding`, and nothing applies. A
 typed-codec option to carry the concrete Go value across the journal boundary is
 reserved for a later, additive change.

durable/CHANGELOG.md

@@ -21,12 +21,44 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.
   fence the old owner: a stale write before then is accepted, and the new owner's
   write at that step gets `ErrStepConflict`.
 
+- **`WithMaxEventDataBytes`** caps a `WithEventData` payload. `ErrEventDataEncoding` reports
+  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 `Store` implementation has new work to do.** It must enforce fence tokens and
-  return `ErrStepConflict`, or `WithOwnerFence` protects nothing. Existing
-  implementations still compile, so nothing flags this.
+- **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
+  type now fails at `Fire`, not at `Recover`. An error arrives as an error with the same
+  message, any invalid UTF-8 replaced, and `errors.Is` no longer matches it.
+- **`Handle.Fire` can refuse a payload.** One with no JSON form that decodes again, or
+  over `WithMaxEventDataBytes` (8 MiB by default), returns the new `ErrEventDataEncoding`
+  before the kernel runs. Fires of that size used to pass, because the payload wasn't
+  journaled.
+- **`FileStore.Append` refuses a Record over its 16 MiB line limit** with the new
+  `ErrRecordTooLarge`. It used to write a line `Load` couldn't read back unless a
+  checkpoint compacted it away. An oversized line now drops its `Snapshot` first, which
+  recovery never reads, so large checkpoints keep working.
+- **A `Store` implementation has new work to do.** It must enforce fence tokens, return
+  `ErrStepConflict` and keep `Record.EventData` and `Record.EventDataError`, or
+  `WithOwnerFence` protects nothing and payloads are lost. Existing implementations
+  still compile, so nothing flags this.
 - **`MemStore.Append` without `WithIdempotencyKey` refuses a Record JSON can't encode,**
   as `FileStore` already did.
 
@@ -36,11 +68,23 @@ and this module adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.
   The same record is still an idempotent no-op, and an Append with
   `WithIdempotencyKey` still dedups on the key alone. A `FileStore` ledger line from
   before this change has no digest, so its step keeps the old no-op behavior. A
-  recovered instance that reuses a step number now fails with `ErrStepConflict`
-  instead of silently dropping the step.
+  recovered instance that reuses a step number with a different record now fails with
+  `ErrStepConflict` instead of silently dropping the step.
 
 ### 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
+  recorded step, and high past a multi-Trace actor delivery, which skipped ordinals. A
+  checkpoint written before this change can still reuse a step: a different record fails
+  with `ErrStepConflict`, and an identical one is acknowledged without a new Record.
+- **`WithEventData` survives recovery.** `Handle.Fire` records the payload in the new
+  `Record.EventData` and `Record.EventDataError`, and `Recover` and `StateAt` supply it
+  on replay. This needs `state.FireEventData`, so `durable` requires `state` v1.1.0 or
+  later, and its `require` must move there before a `durable` tag.
 - **Recording-clock buffer race.** Closing a step's `Record` (draining the
   recording clock) and the clock-read index reads used to mirror timer deadlines
   now take the recording clock's mutex, so they are safe against a `Now()` the

durable/README.md

@@ -101,6 +101,7 @@ Two stdlib-only reference implementations ship in-tree:
 - **`FileStore`** (`NewFileStore`): on-disk, a directory of per-instance
   subdirectories, each an append-only journal, an atomic checkpoint, an
   idempotency ledger, a dispatched-effect log and, once fenced, a `fence` file.
+  `Append` refuses a Record over its 16 MiB line limit with `ErrRecordTooLarge`.
   Each append flushes to stable storage; each checkpoint uses
   write-temp-plus-rename for crash-safe atomicity.
   Use it for durability across restarts without a database.
@@ -135,11 +136,26 @@ 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 JSON form. A parent reducer that type-asserts a non-JSON Go type from
-`AssignCtx.Event` observes the JSON-decoded shape on the replayed `onDone`. A
+Events, service done-data, actor done-data, and actor messages are recorded as their
+JSON form. A parent reducer that type-asserts a non-JSON Go type from `AssignCtx.Event`
+observes the JSON-decoded shape on a replayed `onDone`. A `WithEventData` payload is
+JSON-decoded live as well as on replay, so a reducer gets the same value both ways: a
+struct is a map and an integer a `float64`. An error stays an error with the same
+message, with any invalid UTF-8 replaced. `Handle.Fire` refuses a payload it can't
+record, or one over `WithMaxEventDataBytes` (8 MiB by default), with
+`ErrEventDataEncoding`, and nothing applies. A
 typed-codec option to carry the concrete Go value across the journal boundary is
 reserved for a later, additive change.
 

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
 }
 

durable/checkpoint.go

@@ -61,8 +61,8 @@ type pendingTimerWire struct {
 }
 
 // checkpointEnvelope is the durable layer's wrapper around the bytes a Store keeps
-// for a checkpoint: the opaque state.Snapshot bytes plus the pending-timer table
-// armed at the checkpoint instant. The durable layer fully owns the checkpoint
+// for a checkpoint: the opaque state.Snapshot bytes, the pending-timer table armed at
+// the checkpoint instant, and the next step ordinal. The durable layer fully owns the checkpoint
 // bytes (it both marshals them at Checkpoint and unmarshals them at recover), so
 // wrapping the kernel snapshot in this envelope is additive and invisible to the
 // Store, which treats checkpoint bytes as opaque. A checkpoint with no armed timer
@@ -77,14 +77,21 @@ type checkpointEnvelope struct {
 	// recovery whose compacted tail no longer carries the arming ScheduleAfter can
 	// still re-arm them at their recorded instants.
 	Timers []pendingTimerWire `json:"timers,omitempty"`
+	// NextStep is the step ordinal the first Record after this checkpoint takes. A
+	// recovery whose compacted tail is empty has no Record to derive it from, and the
+	// restored Trace count doesn't match the ordinals: a tick barrier consumes one and
+	// leaves no Trace, and an actor delivery can leave several Traces on one ordinal.
+	// Zero means none was recorded: a Start baseline, or a checkpoint from before this
+	// field.
+	NextStep int `json:"nextStep,omitempty"`
 }
 
 // marshalCheckpoint wraps the marshaled kernel snapshot and the armed pending
 // timers into a checkpoint envelope's bytes. Timers are emitted in scheduleID
 // order so the persisted form is deterministic regardless of the map iteration
-// order they were collected in.
-func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, error) {
-	env := checkpointEnvelope{Snapshot: snap}
+// order they were collected in. nextStep is the ordinal that follows the checkpoint.
+func marshalCheckpoint(snap []byte, timers map[string]pendingTimer, nextStep int) ([]byte, error) {
+	env := checkpointEnvelope{Snapshot: snap, NextStep: nextStep}
 	if len(timers) > 0 {
 		ids := make([]string, 0, len(timers))
 		for id := range timers {
@@ -113,18 +120,19 @@ func marshalCheckpoint(snap []byte, timers map[string]pendingTimer) ([]byte, err
 }
 
 // unmarshalCheckpoint splits a checkpoint envelope's bytes back into the kernel
-// snapshot bytes and the persisted pending-timer wires. It tolerates a bare
-// (pre-envelope) snapshot: bytes that are not a checkpoint envelope are returned
-// as the snapshot with no timers, so an older checkpoint still restores. A
-// checkpoint envelope is distinguished by carrying a non-empty "snapshot" member.
-func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, error) {
+// snapshot bytes, the persisted pending-timer wires and the next step ordinal. It
+// tolerates a bare (pre-envelope) snapshot: bytes that are not a checkpoint envelope
+// are returned as the snapshot with no timers and a zero ordinal, so an older
+// checkpoint still restores. A checkpoint envelope is distinguished by carrying a
+// non-empty "snapshot" member.
+func unmarshalCheckpoint(b []byte) ([]byte, []pendingTimerWire, int, error) {
 	var env checkpointEnvelope
 	if err := json.Unmarshal(b, &env); err != nil || len(env.Snapshot) == 0 {
 		// Not an envelope (or an envelope without the wrapped snapshot member): treat
 		// the bytes as a bare kernel snapshot, the pre-timer-survival checkpoint form.
-		return b, nil, nil
+		return b, nil, 0, nil
 	}
-	return env.Snapshot, env.Timers, nil
+	return env.Snapshot, env.Timers, env.NextStep, nil
 }
 
 // decodeTimerEvent decodes a persisted timer event wire back into the machine's

durable/checkpoint_internal_test.go

@@ -17,11 +17,11 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) {
 	timers := map[string]pendingTimer{
 		"m:a:after:0": {id: "m:a:after:0", due: due, event: "elapsed"},
 	}
-	b, err := marshalCheckpoint(snap, timers)
+	b, err := marshalCheckpoint(snap, timers, 0)
 	if err != nil {
 		t.Fatalf("marshalCheckpoint: %v", err)
 	}
-	gotSnap, gotWires, err := unmarshalCheckpoint(b)
+	gotSnap, gotWires, _, err := unmarshalCheckpoint(b)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint: %v", err)
 	}
@@ -45,7 +45,7 @@ func TestMarshalCheckpoint_RoundTrip(t *testing.T) {
 // armed, so an event-driven checkpoint's envelope stays minimal.
 func TestMarshalCheckpoint_NoTimers(t *testing.T) {
 	snap := []byte(`{"machine":"m"}`)
-	b, err := marshalCheckpoint(snap, nil)
+	b, err := marshalCheckpoint(snap, nil, 0)
 	if err != nil {
 		t.Fatalf("marshalCheckpoint: %v", err)
 	}
@@ -64,7 +64,7 @@ func TestMarshalCheckpoint_NoTimers(t *testing.T) {
 // checkpoint still restores.
 func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) {
 	bare := []byte(`{"machine":"m","current":"a","configuration":["a"]}`)
-	snap, wires, err := unmarshalCheckpoint(bare)
+	snap, wires, _, err := unmarshalCheckpoint(bare)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint: %v", err)
 	}
@@ -81,7 +81,7 @@ func TestUnmarshalCheckpoint_BareSnapshotFallback(t *testing.T) {
 // any real decode failure when it unmarshals the kernel snapshot).
 func TestUnmarshalCheckpoint_Garbage(t *testing.T) {
 	garbage := []byte(`not json`)
-	snap, wires, err := unmarshalCheckpoint(garbage)
+	snap, wires, _, err := unmarshalCheckpoint(garbage)
 	if err != nil {
 		t.Fatalf("unmarshalCheckpoint should tolerate garbage: %v", err)
 	}
@@ -155,7 +155,7 @@ func TestMarshalCheckpoint_BadEventErrors(t *testing.T) {
 	timers := map[string]pendingTimer{
 		"bad": {id: "bad", due: time.Unix(0, 1), event: make(chan int)},
 	}
-	if _, err := marshalCheckpoint([]byte(`{}`), timers); err == nil {
+	if _, err := marshalCheckpoint([]byte(`{}`), timers, 0); err == nil {
 		t.Fatal("want encode error for unmarshalable timer event, got nil")
 	}
 }
@@ -174,3 +174,18 @@ func TestTickInstant(t *testing.T) {
 		t.Fatalf("tickInstant: ok=%v got=%d", ok, got.UnixNano())
 	}
 }
+
+// TestCheckpoint_NextStepRoundTrips proves the envelope carries the ordinal after
+// the checkpoint, and that a checkpoint without one reports zero, the old form.
+func TestCheckpoint_NextStepRoundTrips(t *testing.T) {
+	b, err := marshalCheckpoint([]byte(`{"v":1}`), nil, 9)
+	if err != nil {
+		t.Fatalf("marshalCheckpoint: %v", err)
+	}
+	if _, _, next, err := unmarshalCheckpoint(b); err != nil || next != 9 {
+		t.Fatalf("unmarshalCheckpoint = (%d, %v), want (9, nil)", next, err)
+	}
+	if _, _, next, err := unmarshalCheckpoint([]byte(`{"snapshot":{"v":1}}`)); err != nil || next != 0 {
+		t.Fatalf("old envelope = (%d, %v), want (0, nil)", next, err)
+	}
+}

durable/doc.go

@@ -31,10 +31,11 @@
 //
 // # Caveats
 //
-// Payloads are serialized at the journal boundary. Event, service done-data,
-// actor done-data, and actor messages are recorded as their JSON form; a parent
-// reducer that type-asserts a non-JSON Go type from AssignCtx.Event observes the
-// JSON-decoded shape on the replayed onDone. A typed-codec option to carry the
+// Payloads are serialized at the journal boundary. Event, service done-data, actor
+// done-data, and actor messages are recorded as their JSON form; a parent reducer that
+// type-asserts a non-JSON Go type from AssignCtx.Event observes the JSON-decoded shape
+// on a replayed onDone. A WithEventData payload is JSON-decoded live as well as on
+// replay, so a reducer gets the same value both ways. A typed-codec option to carry the
 // concrete Go value across the boundary is reserved for a later, additive change.
 //
 // The kernel stays pure and dependency-free. The durable module is host-side: