diff --git a/baselib/actor/AGENTS.md b/baselib/actor/AGENTS.md index cb4a2129c..ab33b18ce 100644 --- a/baselib/actor/AGENTS.md +++ b/baselib/actor/AGENTS.md @@ -30,6 +30,11 @@ crash-safe at-least-once delivery with exactly-once deduplication. - `DurableActor` — Actor variant with crash-safe mailbox backed by SQL persistence. Provides `Wait(ctx)` to block until the actor stops and `StopAndWait(ctx)` to request a graceful shutdown and then wait. - `DurableActorConfig[M, R]` — Configuration struct for `DurableActor`: behavior, store, codec, clock, DLO, WaitGroup, `TellRetryPolicy`, lease/heartbeat/poll durations, max attempts, cleanup timeout, deduplication TTL, and `NumWorkers`. - `DurableActorConfig.NumWorkers` — How many concurrent worker loops drain the actor's single mailbox. Default and any value `<= 1` is one worker (strictly-sequential processing). A value `> 1` turns the actor into a competing-consumer pool: that many goroutines each lease distinct messages via `LeaseNextMailboxMessage`, so independent messages run in parallel while per-correlation-key FIFO still keeps same-key messages ordered. Only for behaviors whose handlers are concurrency-safe and hold no writer across their side effects (e.g. the serverconn egress sender on the Read/Commit path). `NewDurableActor` **fails closed** with `ErrConcurrentClassicBehavior` when `NumWorkers > 1` is paired with a classic (`Left`) `ActorBehavior`, since the classic path wraps the whole `Receive` in one write transaction and assumes sequential delivery; pools are only valid on the Read/Commit (`TxBehavior`) path. The test-only `DurableActorConfig.AllowConcurrentClassicBehavior()` escape hatch bypasses the guard for the egress benchmark that measures the forbidden config; production code must never call it. +- `DurableActorConfig.MaxRestarts` / `RestartWindow` — BEAM-style restart intensity budget for the supervision kernel. It defaults OFF: `DefaultMaxRestarts` is `UnlimitedRestarts` (-1) and a zero `MaxRestarts` normalizes to it, so a panicking actor restarts for as long as it keeps panicking. That is deliberate. Restarting forever is no worse than the nack-and-continue loop supervision replaces (both are rate-limited by the nack backoff), whereas a finite budget adds a failure mode the runtime did not have: the actor dies permanently and keeps looking alive to anyone who is not watching. Set a finite budget ONLY where the owner wires `Watch` and reacts to `TerminationRestartIntensityExceeded`; `RecommendedMaxRestarts` (5) over `DefaultRestartWindow` (60s) is the value to reach for when you do. Restart timestamps are tracked in a sliding window off the config's injected clock. +- `(*DurableActor).Watch(ctx) <-chan TerminationInfo` — Registers a terminal lifecycle watcher. The returned channel receives exactly one `TerminationInfo` and is then closed. Delivery is non-blocking by construction (a single-use buffer-of-one channel written once), so a slow or absent watcher can never park the actor's shutdown path (#1093 invariant). Registering after termination returns a channel already loaded with the notification, so there is no race with a stopping actor; cancelling `ctx` deregisters the watcher and closes its channel with no notification. The notification is published when the supervision loop exits, or by `Stop` for an actor that was never started; an actor that is neither started nor stopped never publishes one. +- `TerminationInfo` / `TerminationReason` — What a watcher observes: `TerminationStopped` (Stop/StopAndWait), `TerminationContextCancelled` (lifetime context died without a Stop; reserved for a future externally-owned-context constructor), `TerminationRestartIntensityExceeded` (restart budget spent, `Err` carries the panic, `RestartsExhausted` is true), `TerminationRestartFailed` (checkpoint reload or RestartMessage enqueue failed). `Restarts` counts restarts over the actor's whole lifetime. +- `MessageCodec.Supports(typeID)` — Reports whether the codec has a constructor registered for a TLV type. The supervision path uses it to decide, BEFORE tearing a generation down, whether the checkpoint hand-off is even possible; an actor whose codec never registered a `RestartMessage` degrades to cycling its worker generation with no `OnStop` and no restore, since a teardown it cannot be rebuilt from is strictly worse than leaving the behavior running. +- `PrependRestartMessageWithID` — `PrependRestartMessage` with the enqueued row ID returned. Supervision deletes the row it enqueued last time before writing the next, so a run of restarts leaves at most one restart row in the mailbox. A restart row carries `MaxAttempts` 1 and the runtime never nacks one (a nacked row at `attempts == max_attempts` is neither leasable nor reapable, so it would strand): a failed restart turn dead-letters instead, which makes restore handlers responsible for their own idempotency. - `DefaultDurableActorConfig[M, R]()` — Constructor returning a `DurableActorConfig` with safe defaults (30s lease, 10 max attempts, 1s poll floor / 30s poll ceiling, DefaultTellRetryPolicy). - `DurableActorConfig.PollInterval` / `MaxPollInterval` — Floor and ceiling of the idle mailbox poll backoff (defaults 1s / 30s). The fallback poll is NOT the delivery path: a same-process enqueue signals the mailbox's wake channel, and the store's post-commit `RegisterMailboxWake` callback rouses the exact mailbox a committed transaction enqueued into, so delivery latency is unaffected by how far the backoff has decayed. Each consecutive empty poll roughly doubles the wait from `PollInterval` up to `MaxPollInterval`; any wake or successfully claimed message snaps it back to the floor. This matters at scale because on a Postgres-backed store every empty poll is a full SERIALIZABLE write transaction that updates no rows, so thousands of resident-but-idle actors polling at a fixed 1Hz become a pure transaction tax. The timer is never stopped: since `RegisterMailboxWake` is same-process only, the poll remains the sole discovery mechanism for a row enqueued by another process or replica, which makes `MaxPollInterval` the worst-case cross-process/cross-replica delivery latency. A zero value normalizes to the default and a ceiling below the floor is raised to the floor (constant cadence, never a shrinking wait). - `TellRetryPolicy` — Function type `func(attempts int, lastErr error) (bool, time.Duration)` determining retry behavior for failed Tell messages. Return `(false, _)` to dead-letter immediately. @@ -93,6 +98,14 @@ crash-safe at-least-once delivery with exactly-once deduplication. decisions must use `Delivery.EffectiveAttempts()` so the in-flight peeked attempt is counted before a nack can raise the row to `max_attempts`. - `Tell` with a `DurableActor` persists the message before returning (crash-safe enqueue). +- **Panic means restart, not redeliver.** A behavior that *returns* an error is an ordinary message failure: it nacks and retries per `TellRetryPolicy`. A behavior that *panics* is treated as corrupted in-memory state: the recovered value becomes a `behaviorPanic`, the delivery's normal ack/nack/dead-letter bookkeeping runs first (so the poison message burns its attempt), and only then does the worker hand the panic to supervision. Supervision cancels the current worker generation (draining ALL workers, not just the panicking one), runs the behavior's `OnStop` bounded by `CleanupTimeout`, reloads the persisted FSM checkpoint, prepends a `RestartMessage` at `RestartPriority`, and starts a fresh generation. The nack-before-restart ordering is load-bearing: it is what makes a deterministic poison message climb to `max_attempts` and dead-letter instead of crash-looping the actor forever. +- **A restart reuses the behavior INSTANCE; the clean slate is the handler's job.** The framework does not rebuild the behavior. It stops the workers, optionally calls `OnStop`, and redelivers a `RestartMessage`; the same Go value keeps serving afterwards with whatever fields the panic left behind. The actor is therefore clean exactly when its `RestartMessage` handler rebuilds every piece of in-memory state from the durable row, and stale otherwise. This is not hypothetical: the behaviors that adopt supervision (`credit.opBehavior`, `oor.sessionBehavior`, `oor.oorRegistryBehavior`, `unroll.behavior`) all carry a reload seam, and a handler that returns Ok without using it leaves the actor exactly as far ahead of durable truth as the panic left it. A behavior with no in-memory turn state (the `serverconn` egress sender) may consume the message as a no-op, but it should say so and say why. +- **A restart message is not retried.** It is enqueued with `MaxAttempts` 1, and the runtime dead-letters (rather than nacks) a restart turn that fails, because a nacked row at `attempts == max_attempts` strands forever. Restore handlers get exactly one shot per restart and must be idempotent. +- **`OnStop` may run mid-life and more than once.** A supervised restart calls it before the rebuild, so implementations must be idempotent and must leave the behavior able to serve a new generation rather than assuming it is being discarded. A panic escaping `OnStop` is recovered (it is invoked precisely when the behavior's invariants are broken) and terminates the actor with `TerminationRestartFailed` rather than taking the process down. +- **A panicking turn's own writes are rolled back.** On the classic path the whole `Receive` runs inside one framework transaction, so supervision returns the panic from that transaction to force a rollback and redoes the message's ack/nack bookkeeping outside it. Committing the partial writes alongside the nack would persist exactly the torn state the restart exists to escape, and the checkpoint reload would then hand it straight back. +- **Restart preserves public identity.** The restart runs on an internal generation context derived from the actor's lifetime context, deliberately bypassing the `Once`-guarded `Start`/`Stop`. The actor keeps its ID, its `DurableMailbox` (so senders keep enqueueing across the restart gap, and the mailbox's promise registry survives), and its cached `Ref`. Callers holding an `ActorRef` observe nothing beyond a pause in processing. +- **In-flight Ask promises across a restart.** The panicking turn's promise is completed with the panic error by the normal result handling. A sibling worker's turn sees its generation context cancelled, returns a context error, and has its promise completed with that error; its durable bookkeeping still runs on a detached context. A message not yet handed to the behavior is simply redelivered afterwards and its caller still gets the eventual result. `DurableAsk` responses travel through the outbox, so a restart only delays them. +- **Exceeding the restart budget is terminal, which is why it is off by default.** Once a finite `MaxRestarts` is exhausted inside `RestartWindow`, supervision logs at error level (an internal-bug class, so the level rule allows it), cancels the actor's lifetime context so further sends fail fast rather than piling into a mailbox nothing will drain, tears the actor down, and publishes `TerminationRestartIntensityExceeded` to watchers. Nothing else notices: an actor that dies this way still holds its ID and its mailbox rows, so a finite budget without a `Watch` observer converts a visible crash loop into invisible permanent death. - Outbox messages are dispatched only after state is persisted (outbox pattern). - **Outbox fold p-model.** For tx-aware stores, outbox delivery is `claim -> (target mailbox enqueue + CompleteOutbox) in one write tx`. If the @@ -101,7 +114,7 @@ crash-safe at-least-once delivery with exactly-once deduplication. transaction failure even when the inner Tell/Complete operations returned nil, because begin/commit failures happen outside those operation-level logs. - `ServiceKey` lookup via `Receptionist` is type-safe: mismatched types return `ErrServiceKeyTypeMismatch`. -- `RestartMessage` has `RestartPriority` (MaxInt32) ensuring it is processed before all other messages on recovery. +- **`RestartMessage` ordering, and how it holds under a pool.** `RestartMessage` carries `RestartPriority` (MaxInt32), which makes the claim query hand it out before every other row. Under `NumWorkers > 1` that orders the CLAIMS but not the TURNS: launching the whole pool at once lets one worker take the restart while a sibling takes the row behind it, so a normal turn runs against a behavior instance that is still rebuilding itself from the checkpoint. The guarantee is therefore enforced by a **single-worker warm-up barrier**: a generation launches one worker first, and the rest of the pool waits until that worker has resolved the restart hand-off. The barrier holds unconditionally for a row supervision enqueued itself (a supervised restart), and for the boot hand-off, which an owner prepends before `Start` and which the actor cannot see, it orders the first claim and releases on an idle tick. It cannot wedge a pool: a first claim of anything other than a restart releases it before that message is processed, a restore that fails or panics releases it, whatever ends the warm-up worker releases it, and a `Stop` mid-barrier releases it through the generation context. A single-worker actor is already strictly sequential and gets no barrier at all. - Transaction context (`WithTx`/`RequireTx`) enables same-DB-transaction joining between actors and their callers. - `Mailbox.Send` returns the exact failure error (`ErrMailboxClosed`, `ErrActorTerminated`, `context.Canceled`, `context.DeadlineExceeded`) rather than a boolean; `Tell` and `Ask` propagate this directly to callers. - **Never `Tell` from inside a receive goroutine without a bound.** A blocking diff --git a/baselib/actor/CLAUDE.md b/baselib/actor/CLAUDE.md index cb4a2129c..ab33b18ce 100644 --- a/baselib/actor/CLAUDE.md +++ b/baselib/actor/CLAUDE.md @@ -30,6 +30,11 @@ crash-safe at-least-once delivery with exactly-once deduplication. - `DurableActor` — Actor variant with crash-safe mailbox backed by SQL persistence. Provides `Wait(ctx)` to block until the actor stops and `StopAndWait(ctx)` to request a graceful shutdown and then wait. - `DurableActorConfig[M, R]` — Configuration struct for `DurableActor`: behavior, store, codec, clock, DLO, WaitGroup, `TellRetryPolicy`, lease/heartbeat/poll durations, max attempts, cleanup timeout, deduplication TTL, and `NumWorkers`. - `DurableActorConfig.NumWorkers` — How many concurrent worker loops drain the actor's single mailbox. Default and any value `<= 1` is one worker (strictly-sequential processing). A value `> 1` turns the actor into a competing-consumer pool: that many goroutines each lease distinct messages via `LeaseNextMailboxMessage`, so independent messages run in parallel while per-correlation-key FIFO still keeps same-key messages ordered. Only for behaviors whose handlers are concurrency-safe and hold no writer across their side effects (e.g. the serverconn egress sender on the Read/Commit path). `NewDurableActor` **fails closed** with `ErrConcurrentClassicBehavior` when `NumWorkers > 1` is paired with a classic (`Left`) `ActorBehavior`, since the classic path wraps the whole `Receive` in one write transaction and assumes sequential delivery; pools are only valid on the Read/Commit (`TxBehavior`) path. The test-only `DurableActorConfig.AllowConcurrentClassicBehavior()` escape hatch bypasses the guard for the egress benchmark that measures the forbidden config; production code must never call it. +- `DurableActorConfig.MaxRestarts` / `RestartWindow` — BEAM-style restart intensity budget for the supervision kernel. It defaults OFF: `DefaultMaxRestarts` is `UnlimitedRestarts` (-1) and a zero `MaxRestarts` normalizes to it, so a panicking actor restarts for as long as it keeps panicking. That is deliberate. Restarting forever is no worse than the nack-and-continue loop supervision replaces (both are rate-limited by the nack backoff), whereas a finite budget adds a failure mode the runtime did not have: the actor dies permanently and keeps looking alive to anyone who is not watching. Set a finite budget ONLY where the owner wires `Watch` and reacts to `TerminationRestartIntensityExceeded`; `RecommendedMaxRestarts` (5) over `DefaultRestartWindow` (60s) is the value to reach for when you do. Restart timestamps are tracked in a sliding window off the config's injected clock. +- `(*DurableActor).Watch(ctx) <-chan TerminationInfo` — Registers a terminal lifecycle watcher. The returned channel receives exactly one `TerminationInfo` and is then closed. Delivery is non-blocking by construction (a single-use buffer-of-one channel written once), so a slow or absent watcher can never park the actor's shutdown path (#1093 invariant). Registering after termination returns a channel already loaded with the notification, so there is no race with a stopping actor; cancelling `ctx` deregisters the watcher and closes its channel with no notification. The notification is published when the supervision loop exits, or by `Stop` for an actor that was never started; an actor that is neither started nor stopped never publishes one. +- `TerminationInfo` / `TerminationReason` — What a watcher observes: `TerminationStopped` (Stop/StopAndWait), `TerminationContextCancelled` (lifetime context died without a Stop; reserved for a future externally-owned-context constructor), `TerminationRestartIntensityExceeded` (restart budget spent, `Err` carries the panic, `RestartsExhausted` is true), `TerminationRestartFailed` (checkpoint reload or RestartMessage enqueue failed). `Restarts` counts restarts over the actor's whole lifetime. +- `MessageCodec.Supports(typeID)` — Reports whether the codec has a constructor registered for a TLV type. The supervision path uses it to decide, BEFORE tearing a generation down, whether the checkpoint hand-off is even possible; an actor whose codec never registered a `RestartMessage` degrades to cycling its worker generation with no `OnStop` and no restore, since a teardown it cannot be rebuilt from is strictly worse than leaving the behavior running. +- `PrependRestartMessageWithID` — `PrependRestartMessage` with the enqueued row ID returned. Supervision deletes the row it enqueued last time before writing the next, so a run of restarts leaves at most one restart row in the mailbox. A restart row carries `MaxAttempts` 1 and the runtime never nacks one (a nacked row at `attempts == max_attempts` is neither leasable nor reapable, so it would strand): a failed restart turn dead-letters instead, which makes restore handlers responsible for their own idempotency. - `DefaultDurableActorConfig[M, R]()` — Constructor returning a `DurableActorConfig` with safe defaults (30s lease, 10 max attempts, 1s poll floor / 30s poll ceiling, DefaultTellRetryPolicy). - `DurableActorConfig.PollInterval` / `MaxPollInterval` — Floor and ceiling of the idle mailbox poll backoff (defaults 1s / 30s). The fallback poll is NOT the delivery path: a same-process enqueue signals the mailbox's wake channel, and the store's post-commit `RegisterMailboxWake` callback rouses the exact mailbox a committed transaction enqueued into, so delivery latency is unaffected by how far the backoff has decayed. Each consecutive empty poll roughly doubles the wait from `PollInterval` up to `MaxPollInterval`; any wake or successfully claimed message snaps it back to the floor. This matters at scale because on a Postgres-backed store every empty poll is a full SERIALIZABLE write transaction that updates no rows, so thousands of resident-but-idle actors polling at a fixed 1Hz become a pure transaction tax. The timer is never stopped: since `RegisterMailboxWake` is same-process only, the poll remains the sole discovery mechanism for a row enqueued by another process or replica, which makes `MaxPollInterval` the worst-case cross-process/cross-replica delivery latency. A zero value normalizes to the default and a ceiling below the floor is raised to the floor (constant cadence, never a shrinking wait). - `TellRetryPolicy` — Function type `func(attempts int, lastErr error) (bool, time.Duration)` determining retry behavior for failed Tell messages. Return `(false, _)` to dead-letter immediately. @@ -93,6 +98,14 @@ crash-safe at-least-once delivery with exactly-once deduplication. decisions must use `Delivery.EffectiveAttempts()` so the in-flight peeked attempt is counted before a nack can raise the row to `max_attempts`. - `Tell` with a `DurableActor` persists the message before returning (crash-safe enqueue). +- **Panic means restart, not redeliver.** A behavior that *returns* an error is an ordinary message failure: it nacks and retries per `TellRetryPolicy`. A behavior that *panics* is treated as corrupted in-memory state: the recovered value becomes a `behaviorPanic`, the delivery's normal ack/nack/dead-letter bookkeeping runs first (so the poison message burns its attempt), and only then does the worker hand the panic to supervision. Supervision cancels the current worker generation (draining ALL workers, not just the panicking one), runs the behavior's `OnStop` bounded by `CleanupTimeout`, reloads the persisted FSM checkpoint, prepends a `RestartMessage` at `RestartPriority`, and starts a fresh generation. The nack-before-restart ordering is load-bearing: it is what makes a deterministic poison message climb to `max_attempts` and dead-letter instead of crash-looping the actor forever. +- **A restart reuses the behavior INSTANCE; the clean slate is the handler's job.** The framework does not rebuild the behavior. It stops the workers, optionally calls `OnStop`, and redelivers a `RestartMessage`; the same Go value keeps serving afterwards with whatever fields the panic left behind. The actor is therefore clean exactly when its `RestartMessage` handler rebuilds every piece of in-memory state from the durable row, and stale otherwise. This is not hypothetical: the behaviors that adopt supervision (`credit.opBehavior`, `oor.sessionBehavior`, `oor.oorRegistryBehavior`, `unroll.behavior`) all carry a reload seam, and a handler that returns Ok without using it leaves the actor exactly as far ahead of durable truth as the panic left it. A behavior with no in-memory turn state (the `serverconn` egress sender) may consume the message as a no-op, but it should say so and say why. +- **A restart message is not retried.** It is enqueued with `MaxAttempts` 1, and the runtime dead-letters (rather than nacks) a restart turn that fails, because a nacked row at `attempts == max_attempts` strands forever. Restore handlers get exactly one shot per restart and must be idempotent. +- **`OnStop` may run mid-life and more than once.** A supervised restart calls it before the rebuild, so implementations must be idempotent and must leave the behavior able to serve a new generation rather than assuming it is being discarded. A panic escaping `OnStop` is recovered (it is invoked precisely when the behavior's invariants are broken) and terminates the actor with `TerminationRestartFailed` rather than taking the process down. +- **A panicking turn's own writes are rolled back.** On the classic path the whole `Receive` runs inside one framework transaction, so supervision returns the panic from that transaction to force a rollback and redoes the message's ack/nack bookkeeping outside it. Committing the partial writes alongside the nack would persist exactly the torn state the restart exists to escape, and the checkpoint reload would then hand it straight back. +- **Restart preserves public identity.** The restart runs on an internal generation context derived from the actor's lifetime context, deliberately bypassing the `Once`-guarded `Start`/`Stop`. The actor keeps its ID, its `DurableMailbox` (so senders keep enqueueing across the restart gap, and the mailbox's promise registry survives), and its cached `Ref`. Callers holding an `ActorRef` observe nothing beyond a pause in processing. +- **In-flight Ask promises across a restart.** The panicking turn's promise is completed with the panic error by the normal result handling. A sibling worker's turn sees its generation context cancelled, returns a context error, and has its promise completed with that error; its durable bookkeeping still runs on a detached context. A message not yet handed to the behavior is simply redelivered afterwards and its caller still gets the eventual result. `DurableAsk` responses travel through the outbox, so a restart only delays them. +- **Exceeding the restart budget is terminal, which is why it is off by default.** Once a finite `MaxRestarts` is exhausted inside `RestartWindow`, supervision logs at error level (an internal-bug class, so the level rule allows it), cancels the actor's lifetime context so further sends fail fast rather than piling into a mailbox nothing will drain, tears the actor down, and publishes `TerminationRestartIntensityExceeded` to watchers. Nothing else notices: an actor that dies this way still holds its ID and its mailbox rows, so a finite budget without a `Watch` observer converts a visible crash loop into invisible permanent death. - Outbox messages are dispatched only after state is persisted (outbox pattern). - **Outbox fold p-model.** For tx-aware stores, outbox delivery is `claim -> (target mailbox enqueue + CompleteOutbox) in one write tx`. If the @@ -101,7 +114,7 @@ crash-safe at-least-once delivery with exactly-once deduplication. transaction failure even when the inner Tell/Complete operations returned nil, because begin/commit failures happen outside those operation-level logs. - `ServiceKey` lookup via `Receptionist` is type-safe: mismatched types return `ErrServiceKeyTypeMismatch`. -- `RestartMessage` has `RestartPriority` (MaxInt32) ensuring it is processed before all other messages on recovery. +- **`RestartMessage` ordering, and how it holds under a pool.** `RestartMessage` carries `RestartPriority` (MaxInt32), which makes the claim query hand it out before every other row. Under `NumWorkers > 1` that orders the CLAIMS but not the TURNS: launching the whole pool at once lets one worker take the restart while a sibling takes the row behind it, so a normal turn runs against a behavior instance that is still rebuilding itself from the checkpoint. The guarantee is therefore enforced by a **single-worker warm-up barrier**: a generation launches one worker first, and the rest of the pool waits until that worker has resolved the restart hand-off. The barrier holds unconditionally for a row supervision enqueued itself (a supervised restart), and for the boot hand-off, which an owner prepends before `Start` and which the actor cannot see, it orders the first claim and releases on an idle tick. It cannot wedge a pool: a first claim of anything other than a restart releases it before that message is processed, a restore that fails or panics releases it, whatever ends the warm-up worker releases it, and a `Stop` mid-barrier releases it through the generation context. A single-worker actor is already strictly sequential and gets no barrier at all. - Transaction context (`WithTx`/`RequireTx`) enables same-DB-transaction joining between actors and their callers. - `Mailbox.Send` returns the exact failure error (`ErrMailboxClosed`, `ErrActorTerminated`, `context.Canceled`, `context.DeadlineExceeded`) rather than a boolean; `Tell` and `Ask` propagate this directly to callers. - **Never `Tell` from inside a receive goroutine without a bound.** A blocking diff --git a/baselib/actor/delivery_test.go b/baselib/actor/delivery_test.go index d0bf01374..cf26525fb 100644 --- a/baselib/actor/delivery_test.go +++ b/baselib/actor/delivery_test.go @@ -3,6 +3,7 @@ package actor import ( "context" "errors" + "sort" "sync" "sync/atomic" "testing" @@ -62,6 +63,11 @@ type mockDeliveryStore struct { // edge while leaving the message peek-eligible. injectNackError error + // injectCheckpointError causes only LoadCheckpoint to fail, which is + // how the supervised restart's TerminationRestartFailed path is + // reached without breaking the rest of the store. + injectCheckpointError error + // peekCount counts PeekNextMessage calls. Used to assert that the // receive loop backs off after a failed leaseless nack instead of // tight-spinning re-peeks of the same eligible row. @@ -118,6 +124,42 @@ func (m *mockDeliveryStore) EnqueueMessage(ctx context.Context, return nil } +// claimOrder returns the mailbox's claim-eligible messages in the order the +// real SQL hands them out: highest priority first, then oldest, using the +// UUIDv7 message ID as the age tiebreak the way the query's id ordering does. +// +// Iterating the map directly (as this mock used to) picks an arbitrary +// message, which is fine while a test has one message in flight and quietly +// wrong as soon as ordering is the thing under test. RestartPriority only +// means anything if the claim honours it. +func (m *mockDeliveryStore) claimOrder(mailboxID string, + now time.Time) []*LeasedMessage { + + var eligible []*LeasedMessage + for _, msg := range m.messages { + if msg.MailboxID != mailboxID { + continue + } + + // Skip if already leased and not expired. + if msg.LeaseToken != "" && msg.LeaseUntil.After(now) { + continue + } + + eligible = append(eligible, msg) + } + + sort.Slice(eligible, func(i, j int) bool { + if eligible[i].Priority != eligible[j].Priority { + return eligible[i].Priority > eligible[j].Priority + } + + return eligible[i].ID < eligible[j].ID + }) + + return eligible +} + func (m *mockDeliveryStore) LeaseNextMessage(ctx context.Context, mailboxID string, leaseToken string, leaseDuration time.Duration) ( *LeasedMessage, error) { @@ -131,16 +173,7 @@ func (m *mockDeliveryStore) LeaseNextMessage(ctx context.Context, now := time.Now() - for _, msg := range m.messages { - if msg.MailboxID != mailboxID { - continue - } - - // Skip if already leased and not expired. - if msg.LeaseToken != "" && msg.LeaseUntil.After(now) { - continue - } - + for _, msg := range m.claimOrder(mailboxID, now) { // Lease this message. msg.LeaseToken = leaseToken msg.LeaseUntil = now.Add(leaseDuration) @@ -170,16 +203,7 @@ func (m *mockDeliveryStore) PeekNextMessage(ctx context.Context, now := time.Now() - for _, msg := range m.messages { - if msg.MailboxID != mailboxID { - continue - } - - // Skip if leased and not expired. - if msg.LeaseToken != "" && msg.LeaseUntil.After(now) { - continue - } - + for _, msg := range m.claimOrder(mailboxID, now) { // Skip if attempts exhausted, matching the SQL eligibility. if msg.Attempts >= msg.MaxAttempts { continue @@ -615,6 +639,10 @@ func (m *mockDeliveryStore) LoadCheckpoint(ctx context.Context, m.mu.Lock() defer m.mu.Unlock() + if m.injectCheckpointError != nil { + return nil, m.injectCheckpointError + } + return m.checkpoints[actorID], nil } diff --git a/baselib/actor/durable_actor.go b/baselib/actor/durable_actor.go index bee82ec62..8c395d9dc 100644 --- a/baselib/actor/durable_actor.go +++ b/baselib/actor/durable_actor.go @@ -116,9 +116,39 @@ type DurableActorConfig[M TLVMessage, R any] struct { MaxAttempts int // CleanupTimeout specifies the maximum duration for OnStop cleanup. + // It also bounds the checkpoint reload and RestartMessage enqueue that + // a supervised restart performs. // Default: 5 seconds. CleanupTimeout time.Duration + // MaxRestarts is how many times the actor may be restarted from its + // checkpoint inside RestartWindow after its behavior panics. Once the + // budget is broken the actor terminates PERMANENTLY instead of + // restarting again. + // + // Zero normalizes to DefaultMaxRestarts, which is UnlimitedRestarts: + // by default a panicking actor restarts for as long as it keeps + // panicking. That is deliberate. Restarting forever is strictly no + // worse than the nack-and-continue loop supervision replaces, and both + // are rate-limited by the nack backoff, whereas a finite budget adds a + // failure mode the runtime did not have: the actor dies for good and + // keeps looking alive to anyone who is not watching. + // + // Setting a finite budget WITHOUT registering a Watch observer trades + // a crash loop for unobserved permanent death, which is usually the + // worse of the two. Set it only where the owner reacts to the + // TerminationRestartIntensityExceeded notification, and reach for + // RecommendedMaxRestarts when you do. + // Default: UnlimitedRestarts. + MaxRestarts int + + // RestartWindow is the width of the sliding window over which + // MaxRestarts is counted. Restarts older than this age out, so an + // actor that panics once an hour restarts forever while one that + // panics five times in a minute is put down. + // Default: 60s. + RestartWindow time.Duration + // DeduplicationTTL is how long to keep processed message IDs for // deduplication. Should exceed the maximum possible redelivery window. // Default: 24 hours. @@ -215,6 +245,8 @@ func DefaultDurableActorConfig[M TLVMessage, R any]( CleanupTimeout: 5 * time.Second, DeduplicationTTL: 24 * time.Hour, NumWorkers: 1, + MaxRestarts: DefaultMaxRestarts, + RestartWindow: DefaultRestartWindow, } } @@ -305,6 +337,54 @@ type DurableActor[M TLVMessage, R any] struct { // at construction. numWorkers int + // codec serializes messages. The supervision path needs it to encode + // the RestartMessage it enqueues when restarting from a checkpoint. + codec *MessageCodec + + // restarts enforces the restart intensity budget. It is owned by the + // supervision goroutine. + restarts *restartTracker + + // stopHookRun records that the behavior's OnStop hook has already run + // for the current behavior generation, so a restart that stops the + // behavior and then fails does not stop it a second time on the way + // out. Supervision clears it when a new generation starts. It is owned + // by the supervision goroutine. + stopHookRun bool + + // lastRestartMsgID is the mailbox row ID of the RestartMessage this + // supervisor enqueued most recently. The next restart deletes it + // before enqueueing a fresh one, so a run of restarts leaves at most + // one restart row behind rather than one per restart. It is owned by + // the supervision goroutine. + lastRestartMsgID string + + // restartPending records that the restart path enqueued a + // RestartMessage for the generation about to start, which lets a + // worker pool hold its warm-up barrier for that row unconditionally + // rather than guessing at it. It is owned by the supervision + // goroutine and consumed when the generation starts. + restartPending bool + + // supervisionMu guards runCancel and pendingPanic, both of which are + // written by a worker goroutine and read by the supervision goroutine. + supervisionMu sync.Mutex + + // runCancel cancels the current generation of worker loops. Restart + // works by cancelling it (which drains every worker) rather than by + // cancelling the actor's lifetime context, so the mailbox, ID and Ref + // survive untouched across a restart. + runCancel context.CancelFunc + + // pendingPanic holds the panic that a worker recovered and that + // supervision has not yet acted on. Only the first panic of a + // generation is retained: the rest of the workers are being torn down + // anyway, and one restart answers all of them. + pendingPanic error + + // watchers holds the registered termination watchers. + watchers *watcherRegistry + // startOnce ensures the actor's processing loop starts only once. startOnce sync.Once @@ -314,6 +394,11 @@ type DurableActor[M TLVMessage, R any] struct { // started records whether Start has launched the processing loop. started atomic.Bool + // stopRequested records whether Stop was called, which is what lets + // supervision report a graceful termination apart from a lifetime + // context that was cancelled from elsewhere. + stopRequested atomic.Bool + // done closes once the processing loop has exited. done chan struct{} @@ -473,6 +558,16 @@ func NewDurableActor[M TLVMessage, R any]( mailboxCfg.SingleWorkerLeaseless = numWorkers == 1 && cfg.Behavior.IsRight() + // Resolve the restart intensity budget. A zero MaxRestarts is "unset" + // and normalizes to DefaultMaxRestarts, which is UnlimitedRestarts: a + // finite budget kills the actor permanently, so it is opt-in for + // owners that watch for the event rather than something a config + // inherits by omission. + maxRestarts := cfg.MaxRestarts + if maxRestarts == 0 { + maxRestarts = DefaultMaxRestarts + } + actor := &DurableActor[M, R]{ id: cfg.ID, behavior: cfg.Behavior, @@ -489,7 +584,15 @@ func NewDurableActor[M TLVMessage, R any]( cleanupTimeout: cfg.CleanupTimeout, deduplicationTTL: deduplicationTTL, numWorkers: numWorkers, - done: make(chan struct{}), + codec: cfg.Codec, + restarts: newRestartTracker( + maxRestarts, cfg.RestartWindow, + cfg.Clock.UnwrapOr( + clock.NewDefaultClock(), + ), + ), + watchers: newWatcherRegistry(), + done: make(chan struct{}), } // Create and cache the actor's reference. @@ -500,7 +603,9 @@ func NewDurableActor[M TLVMessage, R any]( return fn.Ok(actor) } -// Start initiates the actor's message processing loops. +// Start initiates the actor's message processing loops. It is idempotent: only +// the first call launches the supervision goroutine, and a supervised restart +// deliberately does not go back through it. func (a *DurableActor[M, R]) Start() { a.startOnce.Do(func() { a.started.Store(true) @@ -514,41 +619,374 @@ func (a *DurableActor[M, R]) Start() { a.wg.Add(1) } - // Launch numWorkers competing lease loops over the one shared - // mailbox. With numWorkers == 1 this is the historical - // single-loop behavior. A supervisor goroutine joins them and - // runs teardown once, so the actor's done / Wg / Stoppable - // semantics are unchanged regardless of the worker count. + go a.supervise() + }) +} + +// supervise owns the actor's worker generations. It runs one generation of +// numWorkers lease loops at a time and joins them; when a generation ends +// because the behavior panicked, it restarts the actor from its persisted +// checkpoint and runs a fresh generation, and when a generation ends for any +// other reason (or the restart budget is spent) it tears the actor down and +// publishes the termination notification. +// +// The restart deliberately bypasses the Once-guarded Start and Stop. Those +// guard the actor's public lifecycle, which a restart does not touch: the ID, +// the mailbox, and the Ref are the same objects afterwards, so callers holding +// an ActorRef never observe the restart beyond a pause in processing. +// +// In-flight Ask promises do not survive a restart as pending work, and they +// are not silently dropped either. The panicking turn's own promise is +// completed with the panic error by the normal result handling before the +// restart is requested. A sibling worker's turn sees its generation context +// cancelled, returns a context error, and has its promise completed with that +// error; its durable bookkeeping still runs on a detached context. A message +// that had not yet been handed to the behavior is simply redelivered after the +// restart, and because the mailbox's promise registry lives on the mailbox +// (which the restart does not touch), its caller still gets the eventual +// result. DurableAsk responses are unaffected: they travel through the outbox, +// so a restart just delays them. +func (a *DurableActor[M, R]) supervise() { + var info TerminationInfo + + for { + // Each generation gets its own context, derived from the + // actor's lifetime context. Cancelling it drains every worker + // without terminating the mailbox, which is what lets senders + // keep enqueueing across the restart gap. + runCtx, runCancel := context.WithCancel(a.ctx) + a.setRunCancel(runCancel) + + // A new generation gets a fresh behavior teardown budget: the + // hook may run once more before this generation is finished + // with, either for a restart or for the terminal teardown. + a.stopHookRun = false + + // A pool warms up one worker at a time. Launching the whole + // pool at once would let one worker take the generation's + // RestartMessage while a sibling takes the row behind it, so a + // normal turn could run against the behavior while the restart + // handler was still rebuilding it. RestartPriority orders the + // claims but not the turns, so the ordering guarantee needs + // this barrier to actually hold. A single-worker actor is + // already strictly sequential, so it gets a nil barrier and + // the identical code path with nothing to pay for. var workers sync.WaitGroup - for i := 0; i < a.numWorkers; i++ { - workers.Add(1) + workers.Add(a.numWorkers) + + // restartPending is set by the restart path, which enqueued the + // row itself and can therefore hold the barrier for it + // unconditionally. A boot generation cannot: the owner + // prepends that row before Start, so the actor never sees it + // and the barrier falls back to ordering the first claim. + barrier := newWarmupBarrier( + a.numWorkers > 1, a.restartPending, + ) + a.restartPending = false + + go a.worker(runCtx, &workers, barrier) + + // The idle window is the mailbox's own poll floor, so a + // generation that starts against an empty mailbox fans out on + // the same cadence the mailbox already polls at rather than + // introducing a second timing knob. + barrier.wait(runCtx, a.mailbox.cfg.PollInterval) + + // The rest of the pool always launches, even when the barrier + // was released by cancellation: they exit immediately against + // the done context, and launching unconditionally keeps the + // WaitGroup count honest. + for i := 1; i < a.numWorkers; i++ { + go a.worker(runCtx, &workers, nil) + } + + workers.Wait() - go a.worker(&workers) + // Every worker of this generation is gone. Release the + // generation context before deciding what happens next. + runCancel() + a.setRunCancel(nil) + + done, terminal := a.superviseGeneration() + if done { + info = terminal + + break } + } - go func() { - workers.Wait() - a.teardown() + // The actor is finished either way, so cancel the lifetime context + // before tearing down. On the graceful path Stop already did this; on + // a terminal failure it is what makes further sends fail fast instead + // of piling up in a mailbox nothing will ever drain. + a.cancel() - if a.wg != nil { - a.wg.Done() - } + a.teardown() + a.publishTermination(info) - close(a.done) - }() - }) + if a.wg != nil { + a.wg.Done() + } + + close(a.done) +} + +// superviseGeneration decides what happens after a worker generation has +// fully drained. It reports whether the actor is finished (along with the +// termination info to publish) or whether supervision should run another +// generation. +func (a *DurableActor[M, R]) superviseGeneration() (bool, TerminationInfo) { + panicErr := a.takePendingPanic() + + // A generation that ended without a panic ended because the actor's + // lifetime context was cancelled, which is the graceful path. The same + // holds for a panic that raced a Stop: the actor is going away, so + // there is nothing to restart into. + if panicErr == nil || a.ctx.Err() != nil { + return true, a.terminationInfo(TerminationStopped, nil) + } + + // The behavior panicked. Spend a unit of restart budget before doing + // any restart work, so a finite budget bounds every flavour of restart + // below, including the degraded one. + if !a.restarts.record() { + logger(a.ctx).ErrorS(a.ctx, "Durable actor exceeded restart "+ + "intensity, terminating", + panicErr, + "actor_id", a.id, + "restarts", a.restarts.count(), + ) + + return true, a.terminationInfo( + TerminationRestartIntensityExceeded, panicErr, + ) + } + + // An actor whose codec never registered the RestartMessage cannot be + // handed its checkpoint back, and the full restart would then be + // strictly WORSE than what it replaces: a mid-life OnStop against a + // behavior instance that is reused anyway, with no state rebuild to + // show for it. Degrade to cycling the worker generation and leave the + // behavior alone. We check this before the teardown, not after, so the + // degraded path never pays the OnStop it cannot benefit from. + if !a.codec.Supports(RestartTLVType) { + logger(a.ctx).WarnS(a.ctx, "Cycling durable actor workers "+ + "without checkpoint restore: codec has no "+ + "RestartMessage", + panicErr, + "actor_id", a.id, + "restarts", a.restarts.count(), + ) + + return false, TerminationInfo{} + } + + // Tear the panicking behavior down so it can release whatever it was + // holding, then re-run the startup path: the persisted checkpoint is + // reloaded and a RestartMessage is enqueued at RestartPriority, + // exactly as a process restart would do. + if err := a.runStopHook(); err != nil { + logger(a.ctx).ErrorS(a.ctx, "Durable actor cleanup panicked "+ + "during restart, terminating", + err, + "actor_id", a.id, + "restarts", a.restarts.count(), + ) + + return true, a.terminationInfo(TerminationRestartFailed, err) + } + + if err := a.restartFromCheckpoint(); err != nil { + // A restart that races Stop fails here on a cancelled or closed + // store. That is the graceful path, not a supervision failure. + if a.ctx.Err() != nil { + return true, a.terminationInfo( + TerminationStopped, nil, + ) + } + + logger(a.ctx).ErrorS(a.ctx, "Durable actor restart failed, "+ + "terminating", + err, + "actor_id", a.id, + "restarts", a.restarts.count(), + ) + + return true, a.terminationInfo(TerminationRestartFailed, err) + } + + logger(a.ctx).InfoS(a.ctx, "Restarted durable actor from checkpoint", + "actor_id", a.id, + "restarts", a.restarts.count(), + "reason", panicErr.Error(), + ) + + return false, TerminationInfo{} +} + +// terminationInfo builds the notification published to watchers for the given +// reason and failure. +func (a *DurableActor[M, R]) terminationInfo(reason TerminationReason, + err error) TerminationInfo { + + // Stop is what normally ends the actor. A lifetime context that went + // away without a Stop call is reported separately so a watcher can + // tell an orderly shutdown from one imposed from outside. + // + // Two caveats worth stating rather than pretending away. First, + // TerminationContextCancelled is currently unreachable: the lifetime + // context is rooted at context.Background and Stop is the only thing + // that cancels it, so the reason exists for the construction path that + // takes an externally owned context and would otherwise have no way to + // report itself. Second, this read of stopRequested is not ordered + // against a concurrent Stop, so a Stop landing in the same instant as + // an unrelated exit can be reported either way. Both readings describe + // the same event (the actor was shut down rather than failed), so the + // ambiguity costs a watcher nothing. + if reason == TerminationStopped && !a.stopRequested.Load() { + reason = TerminationContextCancelled + } + + exhausted := reason == TerminationRestartIntensityExceeded + + return TerminationInfo{ + ActorID: a.id, + Reason: reason, + Err: err, + Restarts: a.restarts.count(), + RestartsExhausted: exhausted, + } +} + +// setRunCancel publishes the current generation's cancel function so a worker +// that recovers a panic can drain the whole generation. +func (a *DurableActor[M, R]) setRunCancel(cancel context.CancelFunc) { + a.supervisionMu.Lock() + defer a.supervisionMu.Unlock() + + a.runCancel = cancel +} + +// requestRestart records a recovered behavior panic and cancels the current +// worker generation so supervision can restart the actor from its checkpoint. +// It never blocks: the only work it does is a short critical section plus a +// context cancellation, so a worker calling it on its way out cannot park. +func (a *DurableActor[M, R]) requestRestart(panicErr error) { + a.supervisionMu.Lock() + + // Keep only the first panic of the generation. The other workers are + // being torn down regardless, and one restart answers all of them. + if a.pendingPanic == nil { + a.pendingPanic = panicErr + } + cancel := a.runCancel + + a.supervisionMu.Unlock() + + if cancel != nil { + cancel() + } +} + +// takePendingPanic returns and clears the panic recorded for the generation +// that just ended, or nil when the generation ended for another reason. +func (a *DurableActor[M, R]) takePendingPanic() error { + a.supervisionMu.Lock() + defer a.supervisionMu.Unlock() + + panicErr := a.pendingPanic + a.pendingPanic = nil + + return panicErr +} + +// restartFromCheckpoint re-runs the durable actor's startup path against the +// persisted state: it reloads the FSM checkpoint and prepends a RestartMessage +// at RestartPriority so the behavior rebuilds its in-memory state from the +// checkpoint before it sees any other message. This is the same pair of steps +// an owner performs when booting the actor for the first time, which is why +// the restart needs no cooperation from the behavior beyond its existing +// RestartMessage handling. +// +// The work runs on a detached, bounded context so a Stop landing mid-restart +// cannot leave the mailbox without its restart message; the next generation +// notices the cancelled lifetime context and exits gracefully instead. +func (a *DurableActor[M, R]) restartFromCheckpoint() error { + ctx, cancel := context.WithTimeout( + context.WithoutCancel(a.ctx), a.cleanupTimeout, + ) + defer cancel() + + // Drop the restart row this supervisor enqueued last time round if it + // is somehow still pending, so a run of restarts leaves at most one + // restart row in the mailbox rather than one per restart. A row that + // was already consumed makes this a harmless no-op, and a failure here + // is not worth failing the restart over: the worst case is the extra + // row we were trying to avoid. + if a.lastRestartMsgID != "" { + if err := a.store.DeleteMessage( + ctx, a.lastRestartMsgID, + ); err != nil { + + logger(a.ctx).WarnS(a.ctx, "Failed to drop stale "+ + "restart message", + err, + "actor_id", a.id, + "delivery_id", a.lastRestartMsgID) + } + + a.lastRestartMsgID = "" + } + + checkpoint, err := a.store.LoadCheckpoint(ctx, a.id) + if err != nil { + return fmt.Errorf("load checkpoint: %w", err) + } + + id, err := PrependRestartMessageWithID( + ctx, a.store, a.codec, a.id, checkpoint, + ) + if err != nil { + return fmt.Errorf("prepend restart message: %w", err) + } + a.lastRestartMsgID = id + + // The next generation now has a restart row waiting for it, which is + // what lets a pool hold its warm-up barrier for that row rather than + // inferring one from the first claim. + a.restartPending = true + + return nil } // worker runs a single lease loop, draining deliveries from the shared mailbox -// until the actor context is cancelled. When the actor runs more than one +// until the generation context is cancelled. When the actor runs more than one // worker they compete for distinct messages via the store's lease, so // independent messages process in parallel; the per-correlation-key FIFO claim // keeps same-key messages ordered across workers. -func (a *DurableActor[M, R]) worker(wg *sync.WaitGroup) { +// +// A behavior panic ends the worker: the delivery's ack/nack bookkeeping has +// already run by then, and the worker hands the panic to supervision, which +// drains its siblings and restarts the actor from its checkpoint rather than +// letting further messages run against in-memory state the panic may have +// corrupted. +// +// A non-nil warmup marks this as the generation's warm-up worker, the one the +// rest of a pool waits behind. It holds the barrier across a RestartMessage +// turn and opens it for anything else, so a checkpoint hand-off completes +// before any sibling can run a normal turn against the same behavior. +func (a *DurableActor[M, R]) worker(ctx context.Context, wg *sync.WaitGroup, + warmup *warmupBarrier) { + defer wg.Done() + // Whatever ends this worker, the rest of the pool must not stay parked + // behind it. A panic, a closed mailbox, and a cancelled generation all + // land here. + defer warmup.open() + // Process messages from the durable mailbox. - for env := range a.mailbox.Receive(a.ctx) { + for env := range a.mailbox.Receive(ctx) { // Extract the Delivery from the envelope. For DurableMailbox, // the delivery is passed directly in env.delivery, eliminating // the need for a global map lookup. @@ -556,7 +994,7 @@ func (a *DurableActor[M, R]) worker(wg *sync.WaitGroup) { if !ok || delivery == nil { // This shouldn't happen for properly configured durable // actors, but handle gracefully. - logger(a.ctx).WarnS(a.ctx, "No delivery found in "+ + logger(ctx).WarnS(ctx, "No delivery found in "+ "envelope", nil, "actor_id", a.id, "msg_type", env.message.MessageType()) @@ -564,13 +1002,40 @@ func (a *DurableActor[M, R]) worker(wg *sync.WaitGroup) { continue } - a.processDelivery(delivery) + // Recording the claim is what tells the barrier's idle tick + // that a hand-off is in progress rather than absent. + warmup.noteClaim() + + // The pool waits behind a restart turn and nothing else. A + // first claim of any other message proves this generation had + // no hand-off waiting, since RestartPriority would have won + // the claim, so the pool fans out before that message is even + // processed rather than serializing behind it. + restart := isRestartDelivery(delivery) + if !restart { + warmup.open() + } + + panicErr := a.processDelivery(ctx, delivery) + + // The hand-off is resolved either way: the restore committed, + // failed and dead-lettered, or panicked. All three release the + // pool. + if restart { + warmup.open() + } + + if panicErr != nil { + a.requestRestart(panicErr) + + return + } } } // teardown closes the mailbox and runs the Stoppable cleanup hook exactly once, -// after every worker loop has exited. The supervisor goroutine started in Start -// invokes it before signaling done. +// after the last worker generation has exited. The supervision goroutine +// invokes it before publishing the termination notification. func (a *DurableActor[M, R]) teardown() { // The actor's context has been cancelled and all workers have exited. // Close the mailbox. @@ -579,9 +1044,54 @@ func (a *DurableActor[M, R]) teardown() { // For durable mailboxes, we don't drain to DLO since messages persist // in the database and will be picked up on restart. - // If a classic behavior implements Stoppable, call OnStop. The - // Read/Commit (Right) path has no Stoppable hook of its own; its owner - // manages cleanup. + // A restart that tore the behavior down and then failed already ran the + // hook for this generation, so runStopHook is a no-op there rather than + // a second OnStop for one teardown. + _ = a.runStopHook() + + logger(a.ctx).DebugS(a.ctx, "Durable actor terminated", + "actor_id", a.id, + ) +} + +// runStopHook calls the behavior's OnStop cleanup hook when it implements +// Stoppable, bounded by the configured cleanup timeout. Both the final +// teardown and a supervised restart run it: a restart is a behavior teardown +// followed by a checkpoint-driven rebuild, so the behavior gets the same +// chance to release resources it would get on a real stop. +// +// It is idempotent per behavior generation. A restart that runs the hook and +// then fails to carry the restart out falls through to the terminal teardown, +// and the behavior must not be stopped twice for the one teardown; supervision +// clears the flag when it starts a new generation. +// +// The hook runs with panic recovery, and returns the recovered panic when it +// panics. That matters more here than it looks: on the restart path OnStop is +// invoked precisely when the behavior's invariants are known to be broken, so +// a cleanup that trips over the same corrupt state is a realistic outcome and +// must not take the process down with it. +func (a *DurableActor[M, R]) runStopHook() (hookErr error) { + if a.stopHookRun { + return nil + } + a.stopHookRun = true + + defer func() { + if r := recover(); r != nil { + err := newBehaviorPanic(r) + + logger(a.ctx).ErrorS(a.ctx, "Panic during durable "+ + "actor cleanup", + err, + "actor_id", a.id, + "stack", string(err.Stack())) + + hookErr = err + } + }() + + // The Read/Commit (Right) path has no Stoppable hook of its own; its + // owner manages cleanup. a.behavior.WhenLeft(func(b ActorBehavior[M, R]) { stoppable, ok := b.(Stoppable) if !ok { @@ -599,18 +1109,84 @@ func (a *DurableActor[M, R]) teardown() { } }) - logger(a.ctx).DebugS(a.ctx, "Durable actor terminated", + return nil +} + +// publishTermination delivers the terminal notification to every registered +// watcher. Delivery is non-blocking by construction (one buffered value per +// single-use channel), so a watcher that never reads cannot hold up the +// actor's shutdown. +func (a *DurableActor[M, R]) publishTermination(info TerminationInfo) { + if !a.watchers.publish(info) { + return + } + + logger(a.ctx).DebugS(a.ctx, "Published durable actor termination", "actor_id", a.id, + "reason", info.Reason.String(), + "restarts", info.Restarts, ) } +// Watch registers interest in the actor's terminal lifecycle event and returns +// a channel that receives exactly one TerminationInfo and is then closed. The +// channel is buffered and written once, so the actor's shutdown path never +// blocks on a watcher that is slow or gone. +// +// Registering after the actor has already terminated returns a channel that is +// already loaded with the notification, so there is no race between Watch and +// the actor stopping. Cancelling ctx deregisters the watcher and closes the +// channel without a notification, which is how a caller that lost interest +// releases its registration. +// +// The notification is published when the supervision loop exits, or by Stop +// for an actor that was never started. An actor that is neither started nor +// stopped never publishes one and a watcher on it waits forever, so pass a +// cancellable ctx if that is a state your caller can reach. +func (a *DurableActor[M, R]) Watch(ctx context.Context) <-chan TerminationInfo { + ch, id, terminated := a.watchers.add() + if terminated { + return ch + } + + // Deregister the watcher if the caller's context goes away first. The + // goroutine retires on the publish rather than on the actor's done + // channel, so it does not outlive an actor that is stopped without + // ever having been started, whose done channel never closes. + if cancelled := ctx.Done(); cancelled != nil { + published := a.watchers.done() + + go func() { + select { + case <-cancelled: + a.watchers.remove(id) + + case <-published: + } + }() + } + + return ch +} + // processDelivery handles a single message delivery with deduplication, // transaction wrapping, panic recovery, lease heartbeating, and automatic -// ack/nack based on result. -func (a *DurableActor[M, R]) processDelivery(delivery *Delivery[M, R]) { +// ack/nack based on result. The ctx is the calling worker's generation +// context, so a supervised restart interrupts in-flight processing. +// +// It returns the recovered panic when the behavior panicked, and nil in every +// other case, including a behavior that merely returned a failed result. All +// of the delivery's ack, nack, and dead-letter bookkeeping has already run by +// the time a panic is returned, so the poison message has burned an attempt +// before supervision restarts the actor. That ordering is what keeps a +// deterministic poison message climbing toward max_attempts and the dead +// letter queue instead of restarting the actor forever. +func (a *DurableActor[M, R]) processDelivery(ctx context.Context, + delivery *Delivery[M, R]) error { + // Create a context for processing. Ask/DurableAsk messages merge the - // actor and caller contexts so request deadlines can still interrupt - // synchronous work. Tell messages use only the actor context, matching + // worker and caller contexts so request deadlines can still interrupt + // synchronous work. Tell messages use only the worker context, matching // non-durable actor semantics: once a fire-and-forget message is // durably enqueued, later caller cancellation must not cancel // processing. @@ -620,9 +1196,9 @@ func (a *DurableActor[M, R]) processDelivery(delivery *Delivery[M, R]) { if delivery.CallerCtx != nil && (delivery.IsAsk() || delivery.IsDurableAsk()) { - processCtx, cancel = mergeContexts(a.ctx, delivery.CallerCtx) + processCtx, cancel = mergeContexts(ctx, delivery.CallerCtx) } else { - processCtx = a.ctx + processCtx = ctx cancel = func() {} } defer cancel() @@ -664,7 +1240,7 @@ func (a *DurableActor[M, R]) processDelivery(delivery *Delivery[M, R]) { "duplicate", err, "delivery_id", delivery.ID) - return + return nil } if rows == 0 { // A zero-row ack means the row was not deleted. On the @@ -688,35 +1264,62 @@ func (a *DurableActor[M, R]) processDelivery(delivery *Delivery[M, R]) { } } - return + return nil } // If the actor opted into the Read/Commit execution path (a Right // behavior), drive it through the Exec handle. Construction guarantees // a tx-aware store is present in this case. if a.behavior.IsRight() { - a.processWithExec( + return a.processWithExec( processCtx, delivery, a.behavior.RightToSome().UnsafeFromSome(), ) - - return } // If we have a transaction-aware store, wrap processing in a // transaction. if a.txAwareStore != nil { - a.processInTransaction(processCtx, delivery) - } else { - a.processWithoutTransaction(processCtx, delivery) + return a.processInTransaction(processCtx, delivery) + } + + return a.processWithoutTransaction(processCtx, delivery) +} + +// isRestartDelivery reports whether the delivery carries a RestartMessage, +// which the runtime must never nack for retry. +// +// A restart message is enqueued with MaxAttempts 1 so it is delivered exactly +// once. Nacking one would leave a row whose attempts already equal its +// max_attempts: the claim query will not lease it again, and nothing will ever +// dead-letter it either, so it strands in the mailbox forever. Under a +// restart-forever budget a behavior that keeps failing its restore would +// accumulate one stranded row per restart. Dead-lettering a failed restart +// instead is both terminal and visible, at the price of making restore +// handlers responsible for their own idempotency, which +// PrependRestartMessageWithID documents. +func isRestartDelivery[M TLVMessage, R any](delivery *Delivery[M, R]) bool { + return IsRestartMessage(delivery.Message) +} + +// panicFrom extracts the recovered behavior panic from a result, or nil when +// the result did not come from a panic. It is the single place the runtime +// decides "this failure means the behavior's in-memory state is suspect". +func panicFrom[R any](result fn.Result[R]) error { + err := result.Err() + if err == nil || !isBehaviorPanic(err) { + return nil } + + return err } // processInTransaction wraps message processing in a database transaction. // All FSM state changes, outbox writes, and deduplication marks happen -// atomically within this transaction. +// atomically within this transaction. It returns the recovered panic when the +// behavior panicked, after the transaction's ack/nack bookkeeping has run. func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, - delivery *Delivery[M, R]) { + delivery *Delivery[M, R]) error { // Capture the behavior result so we can complete the in-memory // promise only after the transaction commits successfully. This @@ -735,6 +1338,18 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, // Execute behavior with panic recovery. behaviorResult = a.executeBehaviorSafely(txCtx, delivery) + // The classic path wraps the WHOLE Receive in this one + // transaction, so a behavior that panicked half way through + // left its own partial writes sitting in it. Committing those + // alongside the nack would persist exactly the torn state the + // restart is supposed to escape, and the checkpoint reload + // would hand it straight back. Return the panic instead, which + // rolls the transaction back; the message's own ack/nack + // bookkeeping is redone below outside it. + if panicErr := panicFrom(behaviorResult); panicErr != nil { + return panicErr + } + // Handle the result within the transaction. This determines // whether to ack, nack for retry, or dead-letter. We only mark // as processed if we're not going to retry - otherwise the @@ -744,6 +1359,28 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, ) }) + // The behavior panicked and its writes rolled back with the + // transaction, so nothing was acked, nacked, or dead-lettered inside + // it. Run that bookkeeping now on finishNonTx's detached, bounded + // context, which is what keeps the poison message burning an attempt + // (and eventually dead-lettering) even though the turn persisted + // nothing. The promise is no longer deferred because there is no + // commit left to wait for: the result is an error either way. + if panicErr := panicFrom(behaviorResult); panicErr != nil { + logger(ctx).WarnS(ctx, + "Rolled back a panicking turn, nacking message", + panicErr, + "actor_id", a.id, + "delivery_id", delivery.ID, + "msg_type", delivery.Message.MessageType(), + ) + + delivery.deferPromise = false + a.finishNonTx(ctx, delivery, behaviorResult) + + return panicErr + } + if err != nil { logger(ctx).WarnS(ctx, "Transaction failed, nacking message", @@ -753,9 +1390,17 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, "msg_type", delivery.Message.MessageType(), ) - // Transaction failed - Nack for retry. + // Transaction failed - Nack for retry. The nack is a durable + // write that must land even if the actor context is cancelled + // mid-failure, so it runs detached and bounded exactly as + // finishNonTx's bookkeeping does. + nackCtx, cancelNack := context.WithTimeout( + context.WithoutCancel(ctx), a.cleanupTimeout, + ) + defer cancelNack() + if nackErr := delivery.Nack( - ctx, err, 10*time.Second, + nackCtx, err, 10*time.Second, ); nackErr != nil { logger(ctx).WarnS(ctx, "Failed to nack after tx failure", @@ -763,7 +1408,7 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, "delivery_id", delivery.ID) } - return + return nil } // Transaction committed -- now it is safe to complete the @@ -771,12 +1416,15 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, if delivery.IsAsk() && delivery.Promise != nil { delivery.Promise.Complete(behaviorResult) } + + return panicFrom(behaviorResult) } // processWithoutTransaction handles message processing when no transaction -// support is available. +// support is available. It returns the recovered panic when the behavior +// panicked, after the ack/nack bookkeeping has run. func (a *DurableActor[M, R]) processWithoutTransaction(ctx context.Context, - delivery *Delivery[M, R]) { + delivery *Delivery[M, R]) error { // Start the heartbeat goroutine for lease extension. heartbeatDone := make(chan struct{}) @@ -788,6 +1436,8 @@ func (a *DurableActor[M, R]) processWithoutTransaction(ctx context.Context, // Hand the result to the shared non-transactional ack/nack bookkeeping. a.finishNonTx(ctx, delivery, result) + + return panicFrom(result) } // finishNonTx applies ack/nack/dead-letter bookkeeping for a result that was @@ -873,9 +1523,10 @@ func (a *DurableActor[M, R]) finishNonTx(ctx context.Context, // The behavior does any slow side-effect IO without holding the writer, then // commits state plus the lease-fenced ack in one short transaction. A lease // heartbeat runs for the duration so a long IO middle does not let the lease -// expire underneath an in-progress Commit. +// expire underneath an in-progress Commit. It returns the recovered panic when +// the behavior panicked, after the ack/nack bookkeeping has run. func (a *DurableActor[M, R]) processWithExec(ctx context.Context, - delivery *Delivery[M, R], tb BoundTxBehavior[M, R]) { + delivery *Delivery[M, R], tb BoundTxBehavior[M, R]) error { // The Read/Commit execution path does not yet support DurableAsk. On // this path the message is acked inside the behavior's own Commit, so @@ -887,7 +1538,7 @@ func (a *DurableActor[M, R]) processWithExec(ctx context.Context, if delivery.IsDurableAsk() { a.rejectDurableAskOnExecPath(ctx, delivery) - return + return nil } // Extend the lease while the behavior does IO outside the writer tx. @@ -949,7 +1600,7 @@ func (a *DurableActor[M, R]) processWithExec(ctx context.Context, delivery.Promise.Complete(result) } - return + return panicFrom(result) } // The behavior returned without committing: it either failed before @@ -969,6 +1620,8 @@ func (a *DurableActor[M, R]) processWithExec(ctx context.Context, // the success path you MUST call ax.Commit (even with an empty closure, // as the serverconn egress sender does) to get the lease fence. a.finishNonTx(ctx, delivery, result) + + return panicFrom(result) } // rejectDurableAskOnExecPath fails a DurableAsk delivered to a Read/Commit @@ -1045,20 +1698,24 @@ func (a *DurableActor[M, R]) rejectDurableAskOnExecPath(ctx context.Context, } // runExecSafely runs a TxBehavior with panic recovery, converting a panic into -// an error result so the caller treats it as a non-committed failure. +// an error result so the caller treats it as a non-committed failure. The +// error is a behaviorPanic rather than a plain error, which is what tells the +// worker to hand the failure to supervision for a restart instead of letting +// the next message run against state the panic may have corrupted. func (a *DurableActor[M, R]) runExecSafely(ctx context.Context, delivery *Delivery[M, R], tb BoundTxBehavior[M, R], core *execCore) ( result fn.Result[R]) { defer func() { if r := recover(); r != nil { - err := fmt.Errorf("panic: %v", r) + err := newBehaviorPanic(r) logger(ctx).ErrorS(ctx, "Panic during tx message "+ "processing", err, "actor_id", a.id, - "delivery_id", delivery.ID) + "delivery_id", delivery.ID, + "stack", string(err.Stack())) result = fn.Err[R](err) } @@ -1067,19 +1724,22 @@ func (a *DurableActor[M, R]) runExecSafely(ctx context.Context, return tb.run(ctx, core, delivery.Message) } -// executeBehaviorSafely runs the behavior with panic recovery. +// executeBehaviorSafely runs the behavior with panic recovery. As in +// runExecSafely, the recovered panic becomes a behaviorPanic so supervision +// can tell it apart from a behavior that simply returned an error. func (a *DurableActor[M, R]) executeBehaviorSafely(ctx context.Context, delivery *Delivery[M, R]) (result fn.Result[R]) { defer func() { if r := recover(); r != nil { - err := fmt.Errorf("panic: %v", r) + err := newBehaviorPanic(r) logger(ctx).ErrorS(ctx, "Panic during message "+ "processing", err, "actor_id", a.id, - "delivery_id", delivery.ID) + "delivery_id", delivery.ID, + "stack", string(err.Stack())) result = fn.Err[R](err) } @@ -1170,6 +1830,9 @@ func (a *DurableActor[M, R]) handleResultInTx( // Apply Tell retry policy. retry, delay := a.tellRetryPolicy(err, effectiveAttempts) + if retry && isRestartDelivery(delivery) { + retry = false + } if retry { // Don't mark as processed - we want retry to work. // nackMessage routes a leaseless (empty-token) delivery @@ -1288,6 +1951,9 @@ func (a *DurableActor[M, R]) handleResult(ctx context.Context, // Apply Tell retry policy. retry, delay := a.tellRetryPolicy(err, effectiveAttempts) + if retry && isRestartDelivery(delivery) { + retry = false + } if retry { if nackErr := delivery.Nack( ctx, err, delay, @@ -1453,10 +2119,28 @@ func (a *DurableActor[M, R]) writeAskResponseToOutbox( return nil } -// Stop signals the actor to terminate. +// Stop signals the actor to terminate. Cancelling the lifetime context ends +// the current worker generation and, because supervision only restarts while +// that context is live, ends the actor for good rather than triggering another +// restart. func (a *DurableActor[M, R]) Stop() { a.stopOnce.Do(func() { + a.stopRequested.Store(true) + a.cancel() + + // An actor that was never started has no supervision loop to + // publish its termination, and a watcher registered against it + // would otherwise wait for a notification nobody will ever + // send. Publish it here instead. A Start that races this loses + // harmlessly: publishing is first-wins, and the supervision + // loop it launches exits immediately against the cancelled + // lifetime context. + if !a.started.Load() { + a.publishTermination( + a.terminationInfo(TerminationStopped, nil), + ) + } }) } diff --git a/baselib/actor/durable_actor_test.go b/baselib/actor/durable_actor_test.go index b512f27ed..ed2ae8685 100644 --- a/baselib/actor/durable_actor_test.go +++ b/baselib/actor/durable_actor_test.go @@ -146,6 +146,7 @@ func (b *mockBehavior) setDelay(d time.Duration) { type stoppableMockBehavior struct { *mockBehavior stopCalled atomic.Bool + stopCount atomic.Int32 stopErr error } @@ -157,6 +158,7 @@ func newStoppableMockBehavior(result fn.Result[int]) *stoppableMockBehavior { func (b *stoppableMockBehavior) OnStop(ctx context.Context) error { b.stopCalled.Store(true) + b.stopCount.Add(1) return b.stopErr } @@ -183,6 +185,25 @@ type mockTxAwareStore struct { // completing and the transaction committing, and is used to // verify that promises are not completed prematurely. txPostCallbackHook func() + + // txErr records the FIRST error an ExecTx callback returned, which is + // what a real store would roll the transaction back on. It is + // first-wins rather than last-wins because a later successful + // transaction (a restart turn, say) would otherwise erase the evidence + // a test is waiting for. + txErr atomic.Pointer[error] +} + +// firstTxErr returns the first error an ExecTx callback returned, or nil when +// none has failed. Tests use it to assert that a panicking turn asked the +// store for a rollback rather than committing its partial writes. +func (m *mockTxAwareStore) firstTxErr() error { + stored := m.txErr.Load() + if stored == nil { + return nil + } + + return *stored } func newMockTxAwareStore() *mockTxAwareStore { @@ -205,7 +226,13 @@ func (m *mockTxAwareStore) ExecTx( } // Execute the function with the same store (simulating a transaction). - if err := fn(ctx, m.mockDeliveryStore); err != nil { + // The returned error is recorded because it is what a real store rolls + // the transaction back on, which is the only observable difference + // between committing a panicking turn's writes and discarding them. + err := fn(ctx, m.mockDeliveryStore) + if err != nil { + m.txErr.CompareAndSwap(nil, &err) + return err } @@ -407,7 +434,7 @@ func TestDurableActorAskRespectsCallerContextAfterEnqueue(t *testing.T) { MaxAttempts: 3, } - actor.processDelivery(&Delivery[*actorTestMsg, int]{ + actor.processDelivery(actor.ctx, &Delivery[*actorTestMsg, int]{ ID: deliveryID, Message: msg, Promise: NewPromise[int](), diff --git a/baselib/actor/interface.go b/baselib/actor/interface.go index 71ad17199..8eda1ef62 100644 --- a/baselib/actor/interface.go +++ b/baselib/actor/interface.go @@ -229,6 +229,20 @@ type Stoppable interface { // context has a deadline for cleanup operations. Implementations should // release resources and return promptly, respecting the context // deadline to avoid blocking system shutdown. + // + // On a DurableActor it is ALSO called mid-life, once per supervised + // restart, and so may run more than once over the actor's lifetime. + // Two consequences follow. It must be idempotent, because a restart + // that then fails to carry itself out is a real path. And it must + // leave the behavior able to serve a new generation of messages: the + // restart reuses the same behavior instance and rebuilds its state + // from the checkpoint via the RestartMessage, so releasing a resource + // here means the behavior has to be willing to reacquire it, not + // assume it is being thrown away. + // + // A panic escaping OnStop is recovered rather than allowed to take the + // process down, since a restart calls it precisely when the behavior's + // invariants are known to be broken. It terminates the actor. OnStop(ctx context.Context) error } diff --git a/baselib/actor/restart.go b/baselib/actor/restart.go index 3bdc5f9d2..39290dc44 100644 --- a/baselib/actor/restart.go +++ b/baselib/actor/restart.go @@ -157,6 +157,28 @@ func PrependRestartMessage( checkpoint *Checkpoint, ) error { + _, err := PrependRestartMessageWithID( + ctx, store, codec, mailboxID, checkpoint, + ) + + return err +} + +// PrependRestartMessageWithID is PrependRestartMessage with the enqueued row's +// ID returned. A caller that prepends repeatedly over an actor's lifetime (the +// supervision kernel, which prepends one per restart) uses the ID to delete its +// previous row before writing the next, so a run of restarts leaves at most one +// restart row in the mailbox rather than one per restart. +// +// Note that the row is enqueued with MaxAttempts 1 because it must be +// delivered exactly once. The runtime therefore never retries a restart +// message whose turn failed, and sends it straight to the dead letter queue +// instead, so a handler that rebuilds state from the checkpoint gets one shot +// and must be idempotent. +func PrependRestartMessageWithID(ctx context.Context, store DeliveryStore, + codec *MessageCodec, mailboxID string, + checkpoint *Checkpoint) (string, error) { + msg := &RestartMessage{ Checkpoint: fn.OptionFromPtr(checkpoint), } @@ -164,14 +186,14 @@ func PrependRestartMessage( // Encode the message. payload, err := codec.Encode(msg) if err != nil { - return err + return "", err } // Generate a UUID v7 for the message (time-ordered, RFC 9562). id := uuid.Must(uuid.NewV7()).String() // Enqueue with highest priority to ensure front-of-queue processing. - return store.EnqueueMessage(ctx, EnqueueParams{ + err = store.EnqueueMessage(ctx, EnqueueParams{ ID: id, MailboxID: mailboxID, MessageType: msg.MessageType(), @@ -183,6 +205,11 @@ func PrependRestartMessage( // Restart message should only be delivered once. MaxAttempts: 1, }) + if err != nil { + return "", err + } + + return id, nil } // IsRestartMessage returns true if the message is a RestartMessage. diff --git a/baselib/actor/supervision.go b/baselib/actor/supervision.go new file mode 100644 index 000000000..8e707e8ac --- /dev/null +++ b/baselib/actor/supervision.go @@ -0,0 +1,493 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "runtime/debug" + "sync" + "sync/atomic" + "time" + + "github.com/lightningnetwork/lnd/clock" +) + +const ( + // UnlimitedRestarts is the DurableActorConfig.MaxRestarts value that + // disables the intensity budget entirely, letting the actor restart + // from its checkpoint for as long as it keeps panicking. It is the + // default, because restarting forever is strictly no worse than the + // nack-and-continue loop supervision replaces (both are rate-limited + // by the nack backoff), while a finite budget introduces a genuinely + // new failure mode: silent permanent death. + UnlimitedRestarts = -1 + + // DefaultMaxRestarts is the restart budget a durable actor gets when + // its config does not choose one. It is UnlimitedRestarts: a finite + // intensity budget kills the actor for good, which is only a safe + // trade where the actor's owner is watching for that event, so it has + // to be chosen rather than inherited. See + // DurableActorConfig.MaxRestarts. + DefaultMaxRestarts = UnlimitedRestarts + + // RecommendedMaxRestarts is the intensity an owner that does wire a + // Watch observer should reach for. It matches the BEAM's default + // one_for_one supervisor intensity, counted over + // DefaultRestartWindow. + RecommendedMaxRestarts = 5 + + // DefaultRestartWindow is the width of the sliding window a finite + // MaxRestarts is counted over when the config does not set one. + DefaultRestartWindow = 60 * time.Second +) + +// TerminationReason classifies why a durable actor's supervision loop exited. +// Exactly one reason is reported per actor, carried on the single +// TerminationInfo delivered to every registered watcher. +type TerminationReason uint8 + +const ( + // TerminationStopped means the actor exited because Stop (or + // StopAndWait) was called. This is the graceful path: every worker + // drained out, the mailbox was closed, and the behavior's OnStop hook + // ran. + TerminationStopped TerminationReason = iota + + // TerminationContextCancelled means the actor's lifetime context was + // cancelled without Stop having been called. The actor's context is + // currently rooted at context.Background, so this reason is reserved + // for a future construction path that accepts an externally owned + // lifetime context. + TerminationContextCancelled + + // TerminationRestartIntensityExceeded means the behavior panicked more + // often than the configured MaxRestarts / RestartWindow budget allows, + // so supervision gave up rather than restarting the actor again. Err + // carries the panic that broke the budget. + TerminationRestartIntensityExceeded + + // TerminationRestartFailed means a restart was within budget but could + // not be carried out: the FSM checkpoint would not load, or the + // RestartMessage would not enqueue. Restarting anyway would hand the + // behavior a blank slate in place of its persisted state, so the actor + // terminates instead. Err carries the failure. + TerminationRestartFailed +) + +// String returns a human readable name for the termination reason. +func (r TerminationReason) String() string { + switch r { + case TerminationStopped: + return "stopped" + + case TerminationContextCancelled: + return "context_cancelled" + + case TerminationRestartIntensityExceeded: + return "restart_intensity_exceeded" + + case TerminationRestartFailed: + return "restart_failed" + + default: + return fmt.Sprintf("unknown(%d)", uint8(r)) + } +} + +// TerminationInfo describes how and why a durable actor stopped. It is the +// single value delivered on every channel handed out by +// (*DurableActor).Watch. +type TerminationInfo struct { + // ActorID is the ID of the actor that terminated. + ActorID string + + // Reason classifies the termination. + Reason TerminationReason + + // Err carries the failure behind a terminal-failure reason: the panic + // for TerminationRestartIntensityExceeded, the bookkeeping error for + // TerminationRestartFailed. It is nil for the graceful reasons. + Err error + + // Restarts is how many times the actor was restarted from its + // checkpoint over its whole lifetime, counting the restart that broke + // the intensity budget. + Restarts int + + // RestartsExhausted reports whether the actor died because it ran out + // of restart budget, as opposed to being stopped or failing to + // restart. + RestartsExhausted bool +} + +// behaviorPanic is the error a recovered behavior panic is converted into. It +// is what separates "the behavior returned an error" (an ordinary, retryable +// message failure) from "the behavior panicked" (its in-memory state is now +// suspect, so the actor must be restarted from its checkpoint). Both the +// recovered value and the stack captured at the recover site are retained so +// the termination notification carries something an operator can act on. +type behaviorPanic struct { + // value is the value that was passed to panic. + value any + + // stack is the goroutine stack captured at the recover site. + stack []byte +} + +// newBehaviorPanic wraps a recovered panic value along with the current stack. +func newBehaviorPanic(value any) *behaviorPanic { + return &behaviorPanic{ + value: value, + stack: debug.Stack(), + } +} + +// Error implements the error interface. The rendering matches the message the +// runtime produced before supervision existed ("panic: "), so the +// nack, dead-letter reason, and log strings a panicking behavior generates are +// unchanged. +func (p *behaviorPanic) Error() string { + return fmt.Sprintf("panic: %v", p.value) +} + +// Stack returns the goroutine stack captured where the panic was recovered. +func (p *behaviorPanic) Stack() []byte { + return p.stack +} + +// isBehaviorPanic reports whether err came from a panicking behavior rather +// than from a behavior that returned a failed result. +func isBehaviorPanic(err error) bool { + var bp *behaviorPanic + + return errors.As(err, &bp) +} + +// restartTracker enforces a BEAM-style restart intensity budget: at most max +// restarts inside a sliding window of the configured width. It is only ever +// touched from the supervision goroutine, so it carries no lock of its own. +type restartTracker struct { + // max is how many restarts are allowed inside window. A negative value + // disables the budget. + max int + + // window is the width of the sliding window. + window time.Duration + + // clock supplies the current time so tests can drive the window + // deterministically. + clock clock.Clock + + // stamps holds the times of the restarts still inside the window, in + // ascending order. It is written and read only by the supervision + // goroutine. + stamps []time.Time + + // total counts every restart the actor has ever taken, including the + // ones that have since aged out of the window. It is atomic because it + // is the one field observers outside supervision read. + total atomic.Int64 +} + +// newRestartTracker builds a tracker over the given budget. A non-positive +// window falls back to DefaultRestartWindow. +func newRestartTracker(max int, window time.Duration, + clk clock.Clock) *restartTracker { + + if window <= 0 { + window = DefaultRestartWindow + } + + return &restartTracker{ + max: max, + window: window, + clock: clk, + } +} + +// record registers one restart at the current time and reports whether the +// actor is still inside its intensity budget. It returns false when the +// restart being recorded is the one that breaks the budget, which is the +// signal for supervision to terminate the actor permanently. +func (r *restartTracker) record() bool { + now := r.clock.Now() + r.total.Add(1) + + // Drop the restarts that have aged out of the sliding window, reusing + // the backing array so a long-lived actor that restarts occasionally + // does not grow the slice without bound. + cutoff := now.Add(-r.window) + kept := r.stamps[:0] + for _, ts := range r.stamps { + if ts.After(cutoff) { + kept = append(kept, ts) + } + } + r.stamps = append(kept, now) + + // A negative budget is the explicit "restart forever" opt-in. + if r.max < 0 { + return true + } + + return len(r.stamps) <= r.max +} + +// count returns how many restarts the actor has taken over its whole lifetime. +// It is safe to call from any goroutine. +func (r *restartTracker) count() int { + return int(r.total.Load()) +} + +// watcherRegistry holds the termination watchers registered against a durable +// actor. Delivery is one buffered value per watcher followed by a close, so +// notifying watchers can never park the actor's shutdown path no matter how +// slowly a watcher reads. +type watcherRegistry struct { + // mu guards every field below. It is held across the notification + // sends, which is safe precisely because those sends cannot block. + mu sync.Mutex + + // watchers maps a registration handle to the channel to notify. A + // handle is removed as soon as it has been notified or its watching + // context was cancelled. + watchers map[uint64]chan TerminationInfo + + // nextID is the next registration handle to hand out. + nextID uint64 + + // terminated records whether the termination notification has already + // been published, so a watcher that registers afterwards is served + // immediately from info instead of waiting forever. + terminated bool + + // info is the published termination notification. It is only + // meaningful once terminated is set. + info TerminationInfo + + // published closes once the terminal notification has been delivered. + // A watcher's cleanup goroutine parks on it rather than on the actor's + // done channel, which never closes for an actor that was stopped + // without ever being started. + published chan struct{} +} + +// newWatcherRegistry builds an empty registry. +func newWatcherRegistry() *watcherRegistry { + return &watcherRegistry{ + watchers: make(map[uint64]chan TerminationInfo), + published: make(chan struct{}), + } +} + +// done returns a channel that closes once the terminal notification has been +// published. +func (w *watcherRegistry) done() <-chan struct{} { + return w.published +} + +// add registers a new watcher. It returns the channel to hand to the caller, +// the registration handle, and whether the actor had already terminated. In +// that last case the channel comes back already loaded with the notification +// and closed, and the handle is not registered. +func (w *watcherRegistry) add() (chan TerminationInfo, uint64, bool) { + w.mu.Lock() + defer w.mu.Unlock() + + // A buffer of one is what makes the eventual send unconditionally + // non-blocking: exactly one value is ever sent per channel. + ch := make(chan TerminationInfo, 1) + + if w.terminated { + ch <- w.info + close(ch) + + return ch, 0, true + } + + id := w.nextID + w.nextID++ + w.watchers[id] = ch + + return ch, id, false +} + +// remove deregisters a watcher that is no longer interested and closes its +// channel, so a caller ranging over it observes the end of the stream. It is a +// no-op once the watcher has been notified. +func (w *watcherRegistry) remove(id uint64) { + w.mu.Lock() + defer w.mu.Unlock() + + ch, ok := w.watchers[id] + if !ok { + return + } + + delete(w.watchers, id) + close(ch) +} + +// publish records the terminal notification and delivers it to every +// registered watcher exactly once. Each send targets a single-use buffered +// channel, so no watcher can park the actor's shutdown path. It reports +// whether this call was the one that published; later calls are no-ops. +func (w *watcherRegistry) publish(info TerminationInfo) bool { + w.mu.Lock() + defer w.mu.Unlock() + + if w.terminated { + return false + } + + w.terminated = true + w.info = info + + for id, ch := range w.watchers { + // The channel has a buffer of one and is written exactly once, + // so this send always succeeds. The default arm exists so a + // future change cannot quietly reintroduce a shutdown path + // that blocks on a watcher. + select { + case ch <- info: + default: + } + + close(ch) + delete(w.watchers, id) + } + + close(w.published) + + return true +} + +// warmupBarrier holds the rest of a competing-consumer pool behind the first +// message of a worker generation, so a RestartMessage hand-off is not raced by +// a sibling worker claiming the next row. +// +// The problem it solves is specific to NumWorkers > 1. A RestartMessage +// carries RestartPriority so the claim query hands it out first, but "first" +// only orders the claims, not the turns: launching the whole pool at once lets +// one worker take the restart while a sibling immediately takes the row behind +// it, and a normal turn then runs against the same behavior instance while the +// restart handler is still rebuilding it from the checkpoint. The documented +// guarantee that a restart message is processed before all other messages +// needs the pool to warm up one worker at a time to actually hold. +// +// The release rule is deliberately shaped so that it cannot wedge a pool that +// has no hand-off waiting for it. The warm-up worker holds the barrier only +// across a restart turn; the first claim of anything else opens it before that +// message is even processed, and a generation whose mailbox turns out to be +// empty opens it on the first idle tick. Whatever ends the warm-up worker +// opens it too, so a panic, a dead-lettered restore, or a closed mailbox all +// release the pool rather than stranding it at one worker. +type warmupBarrier struct { + // released closes when the pool may fan out. + released chan struct{} + + // openOnce keeps the close idempotent, since several paths race to + // open the barrier (the warm-up worker's claim, its exit, and the + // idle tick). + openOnce sync.Once + + // claimed records that the warm-up worker has taken at least one + // message. It is what separates "the restart turn is still running" + // from "there was never a hand-off here", which the idle tick would + // otherwise be unable to tell apart. + claimed atomic.Bool + + // required records that supervision KNOWS a restart row is waiting for + // this generation, because it enqueued that row itself. It disables + // the idle tick, which is what makes the guarantee exact rather than + // timing-dependent on the path this kernel creates: without it a tick + // that fired before the warm-up worker got its first claim would fan + // the pool out into the very race the barrier exists to prevent. + required atomic.Bool +} + +// newWarmupBarrier returns a barrier when one is wanted, and nil otherwise. A +// nil barrier is a working no-op through every method below, so a +// single-worker actor (which is already strictly sequential and needs no +// barrier at all) runs the identical code path with nothing to pay for. +// +// A required barrier is one supervision enqueued the restart row for, so it +// waits for that row unconditionally. A barrier that is merely wanted covers +// the boot hand-off, which an owner enqueues before Start and which the actor +// therefore cannot see: there the barrier waits for the first claim and lets +// an idle tick release it, which orders the common case without being able to +// prove a row was ever there. +func newWarmupBarrier(wanted, required bool) *warmupBarrier { + if !wanted { + return nil + } + + b := &warmupBarrier{ + released: make(chan struct{}), + } + b.required.Store(required) + + return b +} + +// noteClaim records that the warm-up worker has taken a message. +func (b *warmupBarrier) noteClaim() { + if b == nil { + return + } + + b.claimed.Store(true) +} + +// open releases the pool. It is safe to call from any goroutine and any number +// of times. +func (b *warmupBarrier) open() { + if b == nil { + return + } + + b.openOnce.Do(func() { + close(b.released) + }) +} + +// wait blocks until the pool may fan out: until the warm-up worker resolves +// the generation's restart hand-off, until the generation is cancelled, or +// until an idle tick proves there was no hand-off waiting in the first place. +// +// The idle tick only opens the barrier while nothing has been claimed AND the +// barrier is not required. Once the warm-up worker has taken a message the +// tick is inert, so a restore that takes longer than one idle period is waited +// out rather than raced; and a required barrier ignores the tick entirely, +// because supervision knows the row is there and a tick that beat the worker +// to its first claim would fan the pool out into the race the barrier exists +// to prevent. +func (b *warmupBarrier) wait(ctx context.Context, idle time.Duration) { + if b == nil { + return + } + + if idle <= 0 { + idle = defaultPollInterval + } + + ticker := time.NewTicker(idle) + defer ticker.Stop() + + for { + select { + case <-b.released: + return + + case <-ctx.Done(): + return + + case <-ticker.C: + if !b.required.Load() && !b.claimed.Load() { + b.open() + + return + } + } + } +} diff --git a/baselib/actor/supervision_test.go b/baselib/actor/supervision_test.go new file mode 100644 index 000000000..ec998a2d8 --- /dev/null +++ b/baselib/actor/supervision_test.go @@ -0,0 +1,1781 @@ +package actor + +import ( + "context" + "encoding/binary" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/lightningnetwork/lnd/tlv" + "github.com/stretchr/testify/require" +) + +// newSupervisedCodec builds a codec carrying both the actor test message and +// the framework's RestartMessage. A supervised restart enqueues the latter, so +// an actor that can restart must be able to decode it. +func newSupervisedCodec() *MessageCodec { + codec := newActorTestCodec() + codec.MustRegister(RestartTLVType, func() TLVMessage { + return &RestartMessage{} + }) + + return codec +} + +// supervisedBehavior is a classic behavior typed over the generic TLVMessage +// so it receives both the test message and the RestartMessage a supervised +// restart prepends. It panics on the test message for as long as the injected +// predicate says to, and records every restart checkpoint it is handed. +type supervisedBehavior struct { + mu sync.Mutex + + // restarts records the checkpoint carried by each RestartMessage the + // behavior has seen, in delivery order. + restarts []fn.Option[Checkpoint] + + // values records the payload of each non-restart message received. + values []uint64 + + // shouldPanic decides whether the given test message panics. A nil + // predicate never panics. + shouldPanic func(value uint64) bool + + // onReceive runs before the panic decision, for tests that need to + // observe or block inside the turn. + onReceive func(ctx context.Context, value uint64) + + // failRestarts makes the RestartMessage handler fail, which is the + // shape that would otherwise strand or pile up restart rows. + failRestarts bool + + // stopCalls counts OnStop invocations, which supervision runs once per + // restart plus once at final teardown. + stopCalls atomic.Int32 + + // stopPanics makes OnStop panic. Supervision must recover it rather + // than let a cleanup that tripped over the same corrupt state take the + // process down. + stopPanics atomic.Bool +} + +// Receive implements ActorBehavior over the generic TLVMessage type. +func (b *supervisedBehavior) Receive(ctx context.Context, + msg TLVMessage) fn.Result[int] { + + if restart, ok := msg.(*RestartMessage); ok { + b.mu.Lock() + b.restarts = append(b.restarts, restart.Checkpoint) + failRestarts := b.failRestarts + b.mu.Unlock() + + if failRestarts { + return fn.Err[int](errors.New("restore failed")) + } + + return fn.Ok(0) + } + + test, ok := msg.(*actorTestMsg) + if !ok { + return fn.Err[int](errors.New("unexpected message type")) + } + + value := test.Value.Val + + b.mu.Lock() + b.values = append(b.values, value) + onReceive := b.onReceive + shouldPanic := b.shouldPanic + b.mu.Unlock() + + if onReceive != nil { + onReceive(ctx, value) + } + + if shouldPanic != nil && shouldPanic(value) { + panic("supervised behavior panic") + } + + return fn.Ok(int(value)) +} + +// OnStop implements Stoppable so the tests can observe that supervision tears +// the behavior down before restarting it. +func (b *supervisedBehavior) OnStop(context.Context) error { + b.stopCalls.Add(1) + + if b.stopPanics.Load() { + panic("supervised behavior cleanup panic") + } + + return nil +} + +// restartCount returns how many RestartMessages the behavior has seen. +func (b *supervisedBehavior) restartCount() int { + b.mu.Lock() + defer b.mu.Unlock() + + return len(b.restarts) +} + +// lastRestart returns the checkpoint carried by the most recent +// RestartMessage. +func (b *supervisedBehavior) lastRestart() fn.Option[Checkpoint] { + b.mu.Lock() + defer b.mu.Unlock() + + if len(b.restarts) == 0 { + return fn.None[Checkpoint]() + } + + return b.restarts[len(b.restarts)-1] +} + +// valueCount returns how many non-restart messages the behavior has seen. +func (b *supervisedBehavior) valueCount() int { + b.mu.Lock() + defer b.mu.Unlock() + + return len(b.values) +} + +// newSupervisedActor builds a single-worker durable actor over a +// supervisedBehavior, with a fast poll so restarts are observable inside a +// test's patience. +func newSupervisedActor(t *testing.T, store DeliveryStore, + behavior *supervisedBehavior, + tweak func(*DurableActorConfig[TLVMessage, int]), +) *DurableActor[TLVMessage, int] { + + t.Helper() + + cfg := DefaultDurableActorConfig[TLVMessage, int]( + "supervised-actor", behavior, store, newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.CleanupTimeout = time.Second + + if tweak != nil { + tweak(&cfg) + } + + return NewDurableActor(cfg).UnwrapOrFail(t) +} + +// tellSupervised enqueues a value-carrying test message. +func tellSupervised(t *testing.T, a *DurableActor[TLVMessage, int], + value uint64) { + + t.Helper() + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](value), + } + + require.NoError(t, a.Ref().Tell(context.Background(), msg)) +} + +// TestDurableActorPanicRestartsFromCheckpoint verifies that a panicking +// behavior is not merely nacked and re-fed: the actor tears the behavior down, +// reloads its persisted FSM checkpoint, and hands it back through a +// RestartMessage exactly as a process restart would. +func TestDurableActorPanicRestartsFromCheckpoint(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(value uint64) bool { + return value == 1 + }, + } + + // Persist a checkpoint so the restart has real state to restore, which + // is what a supervised restart must feed back to the behavior. + require.NoError( + t, + store.SaveCheckpoint( + context.Background(), CheckpointParams{ + ActorID: "supervised-actor", + StateType: "SupervisedState", + StateData: []byte{0xDE, 0xAD}, + Version: 7, + }, + ), + ) + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + // Give up on the poison message immediately so the + // restart is the only thing left to observe. + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return false, 0 + } + }, + ) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 5*time.Second, 10*time.Millisecond) + + // The behavior was handed the persisted checkpoint, not a blank slate. + checkpoint := behavior.lastRestart().UnwrapOrFail(t) + require.Equal(t, "supervised-actor", checkpoint.ActorID) + require.Equal(t, "SupervisedState", checkpoint.StateType) + require.Equal(t, []byte{0xDE, 0xAD}, checkpoint.StateData) + require.EqualValues(t, 7, checkpoint.Version) + + // The behavior was torn down before the rebuild. + require.GreaterOrEqual(t, int(behavior.stopCalls.Load()), 1) + + // The actor is alive on the far side of the restart and still serving + // its original identity: the same Ref reaches it, and a new message is + // processed. + tellSupervised(t, a, 2) + + require.Eventually(t, func() bool { + return behavior.valueCount() >= 2 + }, 5*time.Second, 10*time.Millisecond) + + require.NoError(t, a.ctx.Err()) +} + +// TestDurableActorPanicKeepsIdentityStable verifies the actor's public +// identity survives a restart: the Ref handed out before the panic is the same +// object afterwards, and it still reaches the actor. +func TestDurableActorPanicKeepsIdentityStable(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(value uint64) bool { + return value == 1 + }, + } + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return false, 0 + } + }, + ) + + ref := a.Ref() + mailbox := a.mailbox + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 5*time.Second, 10*time.Millisecond) + + require.Same(t, mailbox, a.mailbox) + require.Equal(t, ref, a.Ref()) + require.Equal(t, "supervised-actor", ref.ID()) + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(9)), + } + require.NoError(t, ref.Tell(context.Background(), msg)) + + require.Eventually(t, func() bool { + return behavior.valueCount() >= 2 + }, 5*time.Second, 10*time.Millisecond) +} + +// TestDurableActorBehaviorErrorDoesNotRestart verifies supervision only fires +// on a panic. A behavior that returns a failed result is an ordinary message +// failure, so the message retries and the actor is left alone. +func TestDurableActorBehaviorErrorDoesNotRestart(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{} + behavior.onReceive = func(context.Context, uint64) {} + + failing := &failingSupervisedBehavior{supervisedBehavior: behavior} + + cfg := DefaultDurableActorConfig[TLVMessage, int]( + "supervised-actor", failing, store, newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.TellRetryPolicy = func(_ error, attempts int) (bool, + time.Duration) { + + return attempts < 3, time.Millisecond + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + require.Eventually(t, func() bool { + return behavior.valueCount() >= 3 + }, 5*time.Second, 10*time.Millisecond) + + // Retries happened, but no restart: the behavior never saw a + // RestartMessage and the tracker stayed at zero. + require.Zero(t, behavior.restartCount()) + require.Zero(t, a.restarts.count()) +} + +// failingSupervisedBehavior wraps supervisedBehavior and turns every test +// message into a failed result instead of a panic. +type failingSupervisedBehavior struct { + *supervisedBehavior +} + +// Receive records the message through the embedded behavior and then fails. +func (b *failingSupervisedBehavior) Receive(ctx context.Context, + msg TLVMessage) fn.Result[int] { + + if _, ok := msg.(*RestartMessage); ok { + return b.supervisedBehavior.Receive(ctx, msg) + } + + b.supervisedBehavior.Receive(ctx, msg) + + return fn.Err[int](errors.New("behavior failed")) +} + +// TestDurableActorPoisonMessageDeadLettersAcrossRestarts verifies the +// nack-before-restart ordering. A deterministically panicking message burns an +// attempt on every pass, so it climbs to max_attempts and dead-letters instead +// of restarting the actor forever. +func TestDurableActorPoisonMessageDeadLettersAcrossRestarts(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.MaxAttempts = 3 + cfg.MaxRestarts = 10 + cfg.TellRetryPolicy = func(_ error, attempts int) (bool, + time.Duration) { + + return attempts < 3, time.Millisecond + } + }, + ) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + // The poison message ends up in the dead letter queue rather than + // crash-looping the actor. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + return len(store.deadLetters) >= 1 + }, 10*time.Second, 10*time.Millisecond) + + // Each pass panicked, so each pass restarted the actor: the attempts + // budget, not the restart budget, is what stopped the loop. + require.Equal(t, 3, behavior.valueCount()) + require.Equal(t, 3, a.restarts.count()) + + // The actor survived, is still inside its restart budget, and keeps + // serving traffic. + require.NoError(t, a.ctx.Err()) + + behavior.mu.Lock() + behavior.shouldPanic = nil + behavior.mu.Unlock() + + tellSupervised(t, a, 2) + + require.Eventually(t, func() bool { + return behavior.valueCount() >= 4 + }, 5*time.Second, 10*time.Millisecond) +} + +// TestDurableActorRestartIntensityTerminates verifies that a behavior which +// keeps panicking eventually exhausts its restart budget, at which point the +// actor is stopped permanently and its watchers are told why. +func TestDurableActorRestartIntensityTerminates(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.MaxRestarts = 2 + cfg.RestartWindow = time.Hour + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return true, time.Millisecond + } + }, + ) + + watch := a.Watch(context.Background()) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + var info TerminationInfo + select { + case info = <-watch: + case <-time.After(15 * time.Second): + t.Fatal("timed out waiting for termination") + } + + require.Equal(t, "supervised-actor", info.ActorID) + require.Equal( + t, TerminationRestartIntensityExceeded, info.Reason, + ) + require.True(t, info.RestartsExhausted) + require.Equal(t, 3, info.Restarts) + require.Error(t, info.Err) + require.True(t, isBehaviorPanic(info.Err)) + + // The actor is terminal: it has finished shutting down and refuses + // further work rather than accumulating a backlog nothing will drain. + require.NoError(t, a.Wait(context.Background())) + require.Error(t, a.ctx.Err()) + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(2)), + } + require.Error(t, a.Ref().Tell(context.Background(), msg)) + + // The watch channel carries exactly one notification and is then + // closed. + _, ok := <-watch + require.False(t, ok) +} + +// TestDurableActorWatchReportsGracefulStop verifies that an ordinary Stop is +// reported as such, with no restarts and no exhausted budget. +func TestDurableActorWatchReportsGracefulStop(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{} + + a := newSupervisedActor(t, store, behavior, nil) + + watch := a.Watch(context.Background()) + + a.Start() + tellSupervised(t, a, 1) + + require.Eventually(t, func() bool { + return behavior.valueCount() >= 1 + }, 5*time.Second, 10*time.Millisecond) + + require.NoError(t, a.StopAndWait(context.Background())) + + info := <-watch + require.Equal(t, TerminationStopped, info.Reason) + require.Equal(t, "stopped", info.Reason.String()) + require.NoError(t, info.Err) + require.Zero(t, info.Restarts) + require.False(t, info.RestartsExhausted) + + // Registering after the fact still yields the notification, so a + // watcher cannot lose the race against a stopping actor. + late := a.Watch(context.Background()) + require.Equal(t, info, <-late) + + _, ok := <-late + require.False(t, ok) +} + +// TestDurableActorWatchDoesNotBlockShutdown verifies the watch contract's +// never-park half: several watchers that never read must not hold up the +// actor's shutdown, and each still receives exactly one notification. +func TestDurableActorWatchDoesNotBlockShutdown(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{} + + a := newSupervisedActor(t, store, behavior, nil) + + // Deliberately never read from these before the shutdown. + watches := make([]<-chan TerminationInfo, 0, 8) + for i := 0; i < 8; i++ { + watches = append(watches, a.Watch(context.Background())) + } + + a.Start() + + stopped := make(chan struct{}) + go func() { + defer close(stopped) + + a.Stop() + _ = a.Wait(context.Background()) + }() + + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("shutdown parked on an unread watcher") + } + + // Every watcher gets exactly one notification, then a closed channel. + for _, watch := range watches { + info, ok := <-watch + require.True(t, ok) + require.Equal(t, TerminationStopped, info.Reason) + + _, ok = <-watch + require.False(t, ok) + } +} + +// TestDurableActorWatchContextCancelDeregisters verifies a watcher that loses +// interest releases its registration: the channel closes with no notification. +func TestDurableActorWatchContextCancelDeregisters(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{} + + a := newSupervisedActor(t, store, behavior, nil) + a.Start() + defer a.Stop() + + watchCtx, cancel := context.WithCancel(context.Background()) + watch := a.Watch(watchCtx) + cancel() + + select { + case info, ok := <-watch: + require.False(t, ok, "expected close, got %v", info) + + case <-time.After(5 * time.Second): + t.Fatal("cancelled watcher was never released") + } +} + +// supervisedExecBehavior is a Read/Commit behavior typed over the generic +// TLVMessage, so a multi-worker pool (which is only valid on that path) can +// still receive the RestartMessage a supervised restart prepends. +type supervisedExecBehavior struct { + mu sync.Mutex + + // restarts counts the RestartMessages the behavior has seen. + restarts int + + // parked counts the turns that are currently blocked waiting for their + // context to be cancelled. + parked atomic.Int32 + + // cancelled counts the parked turns that observed cancellation, which + // is the evidence that a restart drained every worker. + cancelled atomic.Int32 + + // parking gates the parking behavior. It is cleared by the panicking + // turn so the parked messages simply commit when they are redelivered. + parking atomic.Bool + + // panics counts how many times the panic message has been delivered. + // Only the first delivery panics, so the redelivered message does not + // crash-loop the actor out of its restart budget. + panics atomic.Int32 + + // panicBarrier, when set, holds each panicking turn until every + // participant has arrived, so several workers panic inside the SAME + // generation rather than one per restart. + panicBarrier *sync.WaitGroup + + // barrierArrivals counts turns that reached the barrier. Only the + // first barrierSize of them panic; the redelivered messages commit so + // the actor settles after exactly one restart. + barrierArrivals atomic.Int32 + + // barrierSize is how many turns the barrier waits for. + barrierSize int32 + + // values counts the committed non-restart turns. + values int +} + +// Receive implements TxBehavior over the generic TLVMessage type. A message +// with value 0 parks until its context is cancelled, a message with value 1 +// panics on its first delivery, and anything else commits straight away. +func (b *supervisedExecBehavior) Receive(ctx context.Context, msg TLVMessage, + ax Exec[DeliveryStore]) fn.Result[int] { + + if _, ok := msg.(*RestartMessage); ok { + b.mu.Lock() + b.restarts++ + b.mu.Unlock() + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + return fn.Ok(0) + } + + test, ok := msg.(*actorTestMsg) + if !ok { + return fn.Err[int](errors.New("unexpected message type")) + } + + switch { + case test.Value.Val == 0 && b.parking.Load(): + b.parked.Add(1) + <-ctx.Done() + b.cancelled.Add(1) + + return fn.Err[int](ctx.Err()) + + case test.Value.Val == 1 && b.panicBarrier != nil: + // Wait for every participant to arrive so the panics land in + // one generation rather than one per restart. Only the first + // pass panics; the redelivered messages fall through to the + // Commit below so the actor settles after one restart. + if b.barrierArrivals.Add(1) <= b.barrierSize { + b.panics.Add(1) + b.panicBarrier.Done() + b.panicBarrier.Wait() + + panic("supervised exec behavior concurrent panic") + } + + case test.Value.Val == 1 && b.panics.Add(1) == 1: + // Release the parked turns from the panic itself, so the + // restart is what cancels them rather than a test-side race. + b.parking.Store(false) + + panic("supervised exec behavior panic") + } + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + b.mu.Lock() + b.values++ + b.mu.Unlock() + + return fn.Ok(0) +} + +// restartCount returns how many RestartMessages the behavior has seen. +func (b *supervisedExecBehavior) restartCount() int { + b.mu.Lock() + defer b.mu.Unlock() + + return b.restarts +} + +// TestDurableActorMultiWorkerRestartDrainsPool verifies a restart stops every +// worker of a competing-consumer pool, not just the one that panicked, and +// brings the configured worker count back afterwards. +func TestDurableActorMultiWorkerRestartDrainsPool(t *testing.T) { + t.Parallel() + + const numWorkers = 4 + + store := newMockTxAwareStore() + behavior := &supervisedExecBehavior{} + behavior.parking.Store(true) + + cfg := DefaultDurableTxActorConfig[TLVMessage, int, DeliveryStore]( + "supervised-actor", behavior, identityStoreFactory, store, + newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.NumWorkers = numWorkers + cfg.MaxRestarts = 5 + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return true, time.Millisecond + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + // Park three of the four workers. + for i := 0; i < numWorkers-1; i++ { + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(0)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + } + + require.Eventually(t, func() bool { + return behavior.parked.Load() == numWorkers-1 + }, 5*time.Second, 10*time.Millisecond) + + // Panic on the fourth. + panicMsg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), panicMsg)) + + // Every parked worker observed cancellation, so the restart drained + // the whole pool rather than only the panicking worker. + require.Eventually(t, func() bool { + return behavior.cancelled.Load() == numWorkers-1 + }, 10*time.Second, 10*time.Millisecond) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 10*time.Second, 10*time.Millisecond) + + require.Equal(t, numWorkers, a.numWorkers) + require.NoError(t, a.ctx.Err()) +} + +// TestDurableActorRestartWithoutRestartCodec verifies the degraded restart for +// an actor whose codec never registered the RestartMessage. Such an actor +// cannot be handed its checkpoint back, so the restart cycles the worker +// generation and otherwise leaves the behavior alone: no undecodable message +// is enqueued behind the poison one, and crucially no mid-life OnStop is run +// against a behavior instance that is reused anyway and gets no state rebuild +// out of the deal. +func TestDurableActorRestartWithoutRestartCodec(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := newStoppableMockBehavior(fn.Ok(42)) + behavior.panicOnReceive = true + + // newActorTestCodec deliberately carries no RestartMessage. + cfg := DefaultDurableActorConfig[*actorTestMsg, int]( + "test-actor", behavior, store, newActorTestCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return false, 0 + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(42)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + + require.Eventually(t, func() bool { + return a.restarts.count() >= 1 + }, 5*time.Second, 10*time.Millisecond) + + // The only dead letter is the poison message itself: no undecodable + // restart message was enqueued behind it. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + return len(store.deadLetters) == 1 + }, 5*time.Second, 10*time.Millisecond) + + store.mu.Lock() + for _, m := range store.messages { + require.NotEqual(t, "actor.Restart", m.MessageType) + } + store.mu.Unlock() + + // The behavior was never torn down mid-life, because a teardown it + // cannot be rebuilt from is strictly worse than leaving it running. + require.False(t, behavior.stopCalled.Load()) + + require.NoError(t, a.ctx.Err()) + + // The terminal stop still runs the hook exactly once. + require.NoError(t, a.StopAndWait(context.Background())) + require.True(t, behavior.stopCalled.Load()) + require.Equal(t, int32(1), behavior.stopCount.Load()) +} + +// TestDurableActorRestartRacingStop verifies that a Stop landing while a +// restart is in flight ends the actor gracefully rather than being reported as +// a supervision failure, and that StopAndWait still returns. +func TestDurableActorRestartRacingStop(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return true, time.Millisecond + } + }, + ) + + watch := a.Watch(context.Background()) + + a.Start() + tellSupervised(t, a, 1) + + // Let the restart loop get going, then stop into the middle of it. + require.Eventually(t, func() bool { + return a.restarts.count() >= 1 + }, 5*time.Second, time.Millisecond) + + require.NoError(t, a.StopAndWait(context.Background())) + + info := <-watch + require.Equal(t, TerminationStopped, info.Reason) + require.False(t, info.RestartsExhausted) + require.NoError(t, info.Err) +} + +// TestDurableActorRestartFailedTerminates verifies the TerminationRestartFailed +// path: a restart that is within budget but cannot be carried out (here the +// checkpoint load fails) terminates the actor and reports the failure rather +// than restarting into a behavior with no state. +func TestDurableActorRestartFailedTerminates(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + + loadErr := errors.New("checkpoint store is wedged") + store.injectCheckpointError = loadErr + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return false, 0 + } + }, + ) + + watch := a.Watch(context.Background()) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + var info TerminationInfo + select { + case info = <-watch: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for termination") + } + + require.Equal(t, TerminationRestartFailed, info.Reason) + require.False(t, info.RestartsExhausted) + require.ErrorIs(t, info.Err, loadErr) + require.Equal(t, 1, info.Restarts) + + require.NoError(t, a.Wait(context.Background())) + require.Error(t, a.ctx.Err()) +} + +// TestDurableActorCleanupPanicTerminates verifies that a behavior whose OnStop +// panics during a restart is recovered rather than taking the process down, +// and is reported as a failed restart. +func TestDurableActorCleanupPanicTerminates(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + behavior.stopPanics.Store(true) + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return false, 0 + } + }, + ) + + watch := a.Watch(context.Background()) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + var info TerminationInfo + select { + case info = <-watch: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for termination") + } + + require.Equal(t, TerminationRestartFailed, info.Reason) + require.True(t, isBehaviorPanic(info.Err)) + + // The hook panicked on the restart path and is not run a second time + // by the terminal teardown, so one teardown means one OnStop. + require.NoError(t, a.Wait(context.Background())) + require.Equal(t, int32(1), behavior.stopCalls.Load()) +} + +// TestDurableActorConcurrentPanicsRecordOneRestart verifies that two workers +// panicking inside the SAME generation cost one unit of restart budget, not +// two. Charging per panicking worker would let a pool burn a finite budget +// N times faster than a single-worker actor for the same fault. +func TestDurableActorConcurrentPanicsRecordOneRestart(t *testing.T) { + t.Parallel() + + const numWorkers = 4 + + store := newMockTxAwareStore() + behavior := &supervisedExecBehavior{} + + // Two of the four workers panic together: each parks on the barrier + // until both have arrived, so both panics land in one generation. + var barrier sync.WaitGroup + barrier.Add(2) + behavior.panicBarrier = &barrier + behavior.barrierSize = 2 + + cfg := DefaultDurableTxActorConfig[TLVMessage, int, DeliveryStore]( + "supervised-actor", behavior, identityStoreFactory, store, + newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.NumWorkers = numWorkers + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return true, time.Millisecond + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + for i := 0; i < 2; i++ { + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + } + + // Both panics land, and the generation they shared is restarted once. + require.Eventually(t, func() bool { + return behavior.panics.Load() >= 2 + }, 10*time.Second, 10*time.Millisecond) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 10*time.Second, 10*time.Millisecond) + + require.Equal(t, 1, a.restarts.count()) +} + +// TestDurableActorPanicRollsBackBehaviorWrites verifies that the classic +// transactional path rolls the panicking turn's own writes back rather than +// committing them alongside the nack. The whole Receive runs inside one +// transaction there, so committing would persist exactly the torn state the +// restart exists to escape. +func TestDurableActorPanicRollsBackBehaviorWrites(t *testing.T) { + t.Parallel() + + store := newMockTxAwareStore() + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + } + + cfg := DefaultDurableActorConfig[TLVMessage, int]( + "supervised-actor", behavior, store, newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return false, 0 + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + // The panic reached ExecTx as the transaction's error, which is what + // makes the store roll the behavior's writes back. + require.Eventually(t, func() bool { + return store.firstTxErr() != nil + }, 5*time.Second, 10*time.Millisecond) + + require.True(t, isBehaviorPanic(store.firstTxErr())) + + // The bookkeeping still ran outside the rolled-back transaction, so + // the poison message was dead-lettered rather than left in place. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + return len(store.deadLetters) == 1 + }, 5*time.Second, 10*time.Millisecond) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 5*time.Second, 10*time.Millisecond) +} + +// TestDurableActorKeepsOneRestartRow verifies the restart row hygiene: a run +// of restarts leaves at most one restart message in the mailbox, and a restart +// turn that fails dead-letters rather than stranding a row that nothing will +// ever lease or reap again. +func TestDurableActorKeepsOneRestartRow(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + + // The restore handler fails every time, which is the shape that piles + // rows up: each restart enqueues one and the handler never consumes it + // cleanly. + behavior := &supervisedBehavior{ + shouldPanic: func(uint64) bool { return true }, + failRestarts: true, + } + + a := newSupervisedActor( + t, store, behavior, + func(cfg *DurableActorConfig[TLVMessage, int]) { + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, + time.Duration) { + + return true, time.Millisecond + } + }, + ) + + a.Start() + defer a.Stop() + + tellSupervised(t, a, 1) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 3 + }, 10*time.Second, time.Millisecond) + + // Never more than one restart row pending at a time, and a failed + // restart turn ends up in the dead letter queue rather than stranded + // at attempts == max_attempts. + store.mu.Lock() + pending := 0 + for _, m := range store.messages { + if m.MessageType == "actor.Restart" { + pending++ + } + } + deadRestarts := 0 + for _, dl := range store.deadLetters { + if dl.MessageType == "actor.Restart" { + deadRestarts++ + } + } + store.mu.Unlock() + + require.LessOrEqual(t, pending, 1) + require.Positive(t, deadRestarts) +} + +// TestDurableActorWatchOnNeverStartedActor verifies that stopping an actor +// that was never started still publishes a termination, so a watcher on it +// does not wait forever for a supervision loop that will never run. +func TestDurableActorWatchOnNeverStartedActor(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + behavior := &supervisedBehavior{} + + a := newSupervisedActor(t, store, behavior, nil) + + watch := a.Watch(context.Background()) + a.Stop() + + select { + case info := <-watch: + require.Equal(t, TerminationStopped, info.Reason) + require.Zero(t, info.Restarts) + + case <-time.After(5 * time.Second): + t.Fatal("never-started actor published no termination") + } + + _, ok := <-watch + require.False(t, ok) + + // A watcher that registers afterwards is served from the recorded + // notification just as it is for a started actor. + late := <-a.Watch(context.Background()) + require.Equal(t, TerminationStopped, late.Reason) +} + +// restoringExecBehavior models the shape every durable adopter in this repo +// actually has: a Read/Commit behavior holding in-memory state that mirrors a +// durable row, plus a reload guard it arms whenever that mirror might have run +// ahead of the row. credit.opBehavior and oor.sessionBehavior arm exactly this +// guard on a rolled-back Commit, and (since supervision landed) on a restart +// message too. It exists so the restart contract is tested against a behavior +// that can actually diverge, rather than one for which any handler would pass. +type restoringExecBehavior struct { + mu sync.Mutex + + // store is the durable row this behavior mirrors. + store *mockTxAwareStore + + // actorID keys the checkpoint that holds the durable value. + actorID string + + // value is the in-memory mirror of the durable row: the analogue of + // credit's rec or oor's fsm. + value int64 + + // commitFailed is the reload guard. When set, the next turn rebuilds + // value from the durable row before it does anything else. + commitFailed bool + + // observed records the value each observe turn saw AFTER any reload, + // which is what the test asserts against. + observed []int64 + + // restarts counts the restart messages seen. + restarts int +} + +// restore rebuilds the in-memory value from the durable checkpoint. +func (b *restoringExecBehavior) restore(ctx context.Context) error { + checkpoint, err := b.store.LoadCheckpoint(ctx, b.actorID) + if err != nil { + return err + } + + if checkpoint == nil || len(checkpoint.StateData) != 8 { + b.value = 0 + + return nil + } + + b.value = int64(binary.BigEndian.Uint64(checkpoint.StateData)) + + return nil +} + +// Receive implements TxBehavior. Value 1 advances the in-memory mirror past +// the durable row and then panics, which is the divergence a restart has to +// undo. Value 2 observes the mirror after any pending reload. +func (b *restoringExecBehavior) Receive(ctx context.Context, msg TLVMessage, + ax Exec[DeliveryStore]) fn.Result[int] { + + b.mu.Lock() + defer b.mu.Unlock() + + if _, ok := msg.(*RestartMessage); ok { + b.restarts++ + + // The seam: the framework reuses this instance across the + // restart, so arm the reload rather than treating the message + // as a no-op. + b.commitFailed = true + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + return fn.Ok(0) + } + + test, ok := msg.(*actorTestMsg) + if !ok { + return fn.Err[int](errors.New("unexpected message type")) + } + + if test.Value.Val == 1 { + // Advance the mirror past the durable row, then die before + // anything could persist it. + b.value += 100 + + panic("restoring exec behavior panic") + } + + if b.commitFailed { + if err := b.restore(ctx); err != nil { + return fn.Err[int](err) + } + b.commitFailed = false + } + + b.observed = append(b.observed, b.value) + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + return fn.Ok(0) +} + +// observations returns the values the observe turns saw. +func (b *restoringExecBehavior) observations() []int64 { + b.mu.Lock() + defer b.mu.Unlock() + + return append([]int64(nil), b.observed...) +} + +// restartCount returns how many restart messages the behavior has seen. +func (b *restoringExecBehavior) restartCount() int { + b.mu.Lock() + defer b.mu.Unlock() + + return b.restarts +} + +// TestDurableActorRestartReloadsDivergedState verifies the restart contract +// end to end against a behavior that can actually diverge. The behavior +// advances its in-memory mirror past the durable row and then panics, and the +// next message must see the durable value rather than the stale advance. +// +// This is the test that fails if a RestartMessage handler treats the message +// as a no-op, which is exactly what every Read/Commit adopter did before +// supervision existed: the framework reuses the behavior INSTANCE across a +// restart, so the reload has to come from the handler. +func TestDurableActorRestartReloadsDivergedState(t *testing.T) { + t.Parallel() + + const durableValue = int64(7) + + store := newMockTxAwareStore() + + // The durable row the behavior mirrors. + var stateData [8]byte + binary.BigEndian.PutUint64(stateData[:], uint64(durableValue)) + require.NoError( + t, + store.SaveCheckpoint( + context.Background(), CheckpointParams{ + ActorID: "supervised-actor", + StateType: "MirrorState", + StateData: stateData[:], + Version: 1, + }, + ), + ) + + behavior := &restoringExecBehavior{ + store: store, + actorID: "supervised-actor", + value: durableValue, + } + + cfg := DefaultDurableTxActorConfig[TLVMessage, int, DeliveryStore]( + "supervised-actor", behavior, identityStoreFactory, store, + newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + + // Give up on the poison message immediately so it does not keep + // re-panicking behind the observation. + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return false, 0 + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + // Diverge and die. + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + + require.Eventually(t, func() bool { + return behavior.restartCount() >= 1 + }, 10*time.Second, 10*time.Millisecond) + + // Observe after the restart. + observe := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(2)), + } + require.NoError(t, a.Ref().Tell(context.Background(), observe)) + + require.Eventually(t, func() bool { + return len(behavior.observations()) >= 1 + }, 10*time.Second, 10*time.Millisecond) + + // The stale in-memory advance did not survive the restart: the turn + // after it saw durable truth. Without the handler arming its reload + // guard this would be durableValue + 100. + require.Equal(t, []int64{durableValue}, behavior.observations()) +} + +// gatedExecBehavior is a Read/Commit behavior whose restart handler parks on a +// gate, so a test can hold a generation's checkpoint hand-off open and watch +// what the rest of a worker pool does meanwhile. +type gatedExecBehavior struct { + // gate releases the restore turn when closed. + gate chan struct{} + + // entered closes when a restore turn has begun, which is the moment + // the warm-up barrier is provably holding the pool. + entered chan struct{} + enterOnce sync.Once + + // normals counts committed non-restart turns. Nothing may increment it + // while the gate is shut. + normals atomic.Int32 + + // panics makes the first delivery of value 1 panic, which is how the + // test provokes the supervised restart it wants to observe. + panics atomic.Int32 + + // failRestore makes the restore turn fail rather than park, which must + // still release the pool. + failRestore bool +} + +// Receive implements TxBehavior over the generic TLVMessage type. +func (b *gatedExecBehavior) Receive(ctx context.Context, msg TLVMessage, + ax Exec[DeliveryStore]) fn.Result[int] { + + if _, ok := msg.(*RestartMessage); ok { + b.enterOnce.Do(func() { close(b.entered) }) + + if b.failRestore { + return fn.Err[int](errors.New("restore failed")) + } + + // Park until the test opens the gate. The context arm is what + // lets a Stop landing mid-barrier unwedge this turn. + select { + case <-b.gate: + case <-ctx.Done(): + return fn.Err[int](ctx.Err()) + } + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + return fn.Ok(0) + } + + test, ok := msg.(*actorTestMsg) + if !ok { + return fn.Err[int](errors.New("unexpected message type")) + } + + if test.Value.Val == 1 && b.panics.Add(1) == 1 { + panic("gated exec behavior panic") + } + + if err := ax.Commit(ctx, noOpCommit); err != nil { + return fn.Err[int](err) + } + + // Only the backlog counts. The message that provoked the restart is + // redelivered afterwards and commits harmlessly, but counting it would + // make the backlog assertions read as an off-by-one rather than as the + // ordering property they are about. + if test.Value.Val != 1 { + b.normals.Add(1) + } + + return fn.Ok(0) +} + +// newGatedPoolActor builds a competing-consumer pool over a gatedExecBehavior. +func newGatedPoolActor(t *testing.T, store *mockTxAwareStore, + behavior *gatedExecBehavior, + numWorkers int) *DurableActor[TLVMessage, int] { + + t.Helper() + + cfg := DefaultDurableTxActorConfig[TLVMessage, int, DeliveryStore]( + "supervised-actor", behavior, identityStoreFactory, store, + newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.NumWorkers = numWorkers + cfg.MaxAttempts = 100 + cfg.TellRetryPolicy = func(error, int) (bool, time.Duration) { + return true, time.Millisecond + } + + return NewDurableActor(cfg).UnwrapOrFail(t) +} + +// newGatedBehavior builds a gated behavior with its channels wired. +func newGatedBehavior() *gatedExecBehavior { + return &gatedExecBehavior{ + gate: make(chan struct{}), + entered: make(chan struct{}), + } +} + +// TestDurableActorPoolWarmupBarrierHoldsRestart verifies the ordering +// guarantee under a competing-consumer pool: while a restart hand-off is being +// processed, no sibling worker may run a normal turn against the same behavior +// instance. +// +// RestartPriority orders the CLAIMS, not the turns. Launching a pool all at +// once lets one worker take the restart message while a sibling immediately +// takes the row behind it, so a normal turn runs against a behavior that is +// still rebuilding itself from the checkpoint. The warm-up barrier is what +// turns the documented "processed before all other messages" into something +// that actually holds. +func TestDurableActorPoolWarmupBarrierHoldsRestart(t *testing.T) { + t.Parallel() + + const ( + numWorkers = 4 + backlog = 6 + ) + + store := newMockTxAwareStore() + behavior := newGatedBehavior() + + a := newGatedPoolActor(t, store, behavior, numWorkers) + a.Start() + defer a.Stop() + + // Provoke a supervised restart. The framework enqueues the restart + // message itself, so the barrier holds for it unconditionally rather + // than inferring it from the first claim. + panicMsg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), panicMsg)) + + // Wait until the restore turn is parked in the gate. From here on the + // barrier is provably shut and only the warm-up worker exists. + select { + case <-behavior.entered: + case <-time.After(10 * time.Second): + t.Fatal("restore turn never started") + } + + // Queue a backlog behind the parked restore. Every one of these is + // claim-eligible, so a pool that had fanned out would drain them. + for i := 0; i < backlog; i++ { + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(2)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + } + + // Nothing may run while the hand-off is open. The window is many poll + // intervals wide, so a pool that fanned out early would be caught. + require.Never(t, func() bool { + return behavior.normals.Load() > 0 + }, 500*time.Millisecond, 10*time.Millisecond) + + // Release the restore and the pool fans out to drain the backlog. + close(behavior.gate) + + require.Eventually(t, func() bool { + return behavior.normals.Load() == int32(backlog) + }, 10*time.Second, 10*time.Millisecond) + + require.NoError(t, a.ctx.Err()) +} + +// TestDurableActorPoolWarmupBarrierReleasesOnFailedRestore verifies the +// barrier cannot wedge a pool when the restore turn fails. A failed restart +// turn is dead-lettered rather than retried, so the hand-off is resolved and +// the pool must fan out. +func TestDurableActorPoolWarmupBarrierReleasesOnFailedRestore(t *testing.T) { + t.Parallel() + + const ( + numWorkers = 4 + backlog = 6 + ) + + store := newMockTxAwareStore() + behavior := newGatedBehavior() + behavior.failRestore = true + + a := newGatedPoolActor(t, store, behavior, numWorkers) + a.Start() + defer a.Stop() + + panicMsg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), panicMsg)) + + select { + case <-behavior.entered: + case <-time.After(10 * time.Second): + t.Fatal("restore turn never started") + } + + for i := 0; i < backlog; i++ { + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(2)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + } + + // The restore failed, so the barrier releases and the backlog drains. + require.Eventually(t, func() bool { + return behavior.normals.Load() == int32(backlog) + }, 10*time.Second, 10*time.Millisecond) + + // The failed restart went to the dead letter queue rather than being + // retried into a second barrier. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + for _, dl := range store.deadLetters { + if dl.MessageType == "actor.Restart" { + return true + } + } + + return false + }, 10*time.Second, 10*time.Millisecond) +} + +// TestDurableActorPoolWarmupBarrierStopUnblocks verifies that a Stop landing +// while the barrier is shut terminates the actor cleanly rather than parking +// shutdown behind a restore that will never finish. +func TestDurableActorPoolWarmupBarrierStopUnblocks(t *testing.T) { + t.Parallel() + + store := newMockTxAwareStore() + behavior := newGatedBehavior() + + // The gate is never opened: only the generation context can end the + // restore turn. + a := newGatedPoolActor(t, store, behavior, 4) + + watch := a.Watch(context.Background()) + + a.Start() + + panicMsg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + } + require.NoError(t, a.Ref().Tell(context.Background(), panicMsg)) + + select { + case <-behavior.entered: + case <-time.After(10 * time.Second): + t.Fatal("restore turn never started") + } + + stopped := make(chan error, 1) + go func() { + stopCtx, cancel := context.WithTimeout( + context.Background(), 10*time.Second, + ) + defer cancel() + + stopped <- a.StopAndWait(stopCtx) + }() + + select { + case err := <-stopped: + require.NoError(t, err) + + case <-time.After(10 * time.Second): + t.Fatal("shutdown parked behind the warm-up barrier") + } + + info := <-watch + require.Equal(t, TerminationStopped, info.Reason) +} + +// TestDurableActorPoolWithoutRestartFansOut verifies the barrier costs a pool +// nothing when there is no hand-off to order: the first claim of a normal +// message releases it before that message is even processed, so the pool is at +// full width for the work behind it. +func TestDurableActorPoolWithoutRestartFansOut(t *testing.T) { + t.Parallel() + + const numWorkers = 4 + + store := newMockTxAwareStore() + behavior := &supervisedExecBehavior{} + behavior.parking.Store(true) + + cfg := DefaultDurableTxActorConfig[TLVMessage, int, DeliveryStore]( + "supervised-actor", behavior, identityStoreFactory, store, + newSupervisedCodec(), + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxPollInterval = 10 * time.Millisecond + cfg.NumWorkers = numWorkers + cfg.MaxAttempts = 100 + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + // Every worker parks inside its turn, so reaching numWorkers parked + // turns is only possible once the whole pool is running. + for i := 0; i < numWorkers; i++ { + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(0)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + } + + require.Eventually(t, func() bool { + return behavior.parked.Load() == numWorkers + }, 10*time.Second, 10*time.Millisecond) +} + +// TestRestartTrackerSlidingWindow verifies the intensity budget is a sliding +// window: restarts inside the window count against the budget, and restarts +// that have aged out do not. +func TestRestartTrackerSlidingWindow(t *testing.T) { + t.Parallel() + + clk := clock.NewTestClock(time.Unix(1000, 0)) + tracker := newRestartTracker(2, time.Minute, clk) + + require.True(t, tracker.record()) + require.True(t, tracker.record()) + + // The third restart inside the window breaks the budget. + require.False(t, tracker.record()) + require.Equal(t, 3, tracker.count()) + + // Once the window has slid past the earlier restarts, the budget is + // available again. + clk.SetTime(clk.Now().Add(2 * time.Minute)) + require.True(t, tracker.record()) + require.Equal(t, 4, tracker.count()) +} + +// TestRestartTrackerUnlimited verifies the explicit opt-out never runs out of +// budget. +func TestRestartTrackerUnlimited(t *testing.T) { + t.Parallel() + + clk := clock.NewTestClock(time.Unix(1000, 0)) + tracker := newRestartTracker(UnlimitedRestarts, time.Minute, clk) + + for i := 0; i < 100; i++ { + require.True(t, tracker.record()) + } + + require.Equal(t, 100, tracker.count()) +} + +// TestDurableActorRestartBudgetDefaults verifies the intensity budget defaults +// OFF: a finite budget kills the actor permanently, so both the default config +// and a hand-built config that never mentions MaxRestarts must land on +// unlimited rather than inheriting a kill switch. A finite budget is honored +// only when it is asked for. +func TestDurableActorRestartBudgetDefaults(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + codec := newSupervisedCodec() + behavior := &supervisedBehavior{} + + require.Equal(t, UnlimitedRestarts, DefaultMaxRestarts) + + cfg := DefaultDurableActorConfig[TLVMessage, int]( + "a", behavior, store, codec, + ) + require.Equal(t, UnlimitedRestarts, cfg.MaxRestarts) + require.Equal(t, DefaultRestartWindow, cfg.RestartWindow) + + defaulted := NewDurableActor(cfg).UnwrapOrFail(t) + require.Equal(t, UnlimitedRestarts, defaulted.restarts.max) + + // A hand-built config with a zero MaxRestarts normalizes to unlimited + // too, so a config that predates supervision cannot acquire a silent + // kill switch by omission. + bare := DurableActorConfig[TLVMessage, int]{ + ID: "a", + Behavior: NewClassicBehavior[TLVMessage, int](behavior), + Store: store, + Codec: codec, + } + bareActor := NewDurableActor(bare).UnwrapOrFail(t) + require.Equal(t, UnlimitedRestarts, bareActor.restarts.max) + require.Equal(t, DefaultRestartWindow, bareActor.restarts.window) + + // An explicitly chosen finite budget is honored verbatim. + cfg.MaxRestarts = RecommendedMaxRestarts + finite := NewDurableActor(cfg).UnwrapOrFail(t) + require.Equal(t, RecommendedMaxRestarts, finite.restarts.max) +} + +// TestTerminationReasonString verifies every reason renders a stable name. +func TestTerminationReasonString(t *testing.T) { + t.Parallel() + + require.Equal(t, "stopped", TerminationStopped.String()) + require.Equal( + t, "context_cancelled", TerminationContextCancelled.String(), + ) + require.Equal( + t, "restart_intensity_exceeded", + TerminationRestartIntensityExceeded.String(), + ) + require.Equal( + t, "restart_failed", TerminationRestartFailed.String(), + ) + require.Equal(t, "unknown(9)", TerminationReason(9).String()) +} diff --git a/baselib/actor/tlv_message.go b/baselib/actor/tlv_message.go index 4da1d047b..f5265163a 100644 --- a/baselib/actor/tlv_message.go +++ b/baselib/actor/tlv_message.go @@ -73,6 +73,20 @@ func (c *MessageCodec) Register(typeID tlv.Type, return nil } +// Supports reports whether the codec can decode the given TLV type. It lets a +// caller decide whether a message is worth enqueueing at all rather than +// discovering at delivery time that the consumer cannot decode it. The +// supervision path uses it to avoid prepending a RestartMessage to an actor +// whose codec never registered one. +func (c *MessageCodec) Supports(typeID tlv.Type) bool { + c.mu.RLock() + defer c.mu.RUnlock() + + _, exists := c.registry[typeID] + + return exists +} + // MustRegister is like Register but panics on error. Useful for init-time // registration where errors should be caught early. func (c *MessageCodec) MustRegister(typeID tlv.Type, diff --git a/credit/op_actor.go b/credit/op_actor.go index c1245ad97..1437a0709 100644 --- a/credit/op_actor.go +++ b/credit/op_actor.go @@ -266,7 +266,23 @@ func (b *opBehavior) Receive(ctx context.Context, msg CreditDurableMsg, if _, ok := msg.(*actor.RestartMessage); ok { - // Restore already ran at construction; nothing to persist. + // A restart message reaches a live behavior in exactly one + // case: the actor's supervisor restarted it after this + // behavior panicked. The panic may have left rec advanced in + // memory past the last durable checkpoint, and the framework + // reuses this same behavior instance across the restart, so + // treating the message as a no-op would hand the stale advance + // straight back to the next turn. Arm the same reload guard the + // failed-commit path uses: the next turn rebuilds rec from the + // durable row before it dispatches. + // + // The reload is deferred to that turn rather than run here so + // the restart message stays a pure control message with no IO + // of its own. That matters because the framework delivers it + // with max_attempts 1 and dead-letters (rather than retries) a + // restart turn that fails. + b.commitFailed = true + return fn.Ok[CreditResp](&AckResponse{}) } diff --git a/docs/durable_actor_architecture.md b/docs/durable_actor_architecture.md index 811aa4e26..b3c568d1e 100644 --- a/docs/durable_actor_architecture.md +++ b/docs/durable_actor_architecture.md @@ -635,6 +635,151 @@ flowchart LR C -.->|Time passes| D ``` +### In-Process Supervision: Panic-Driven Restart + +A crash is not the only way an actor's in-memory state goes bad. A behavior +that panics mid-turn leaves whatever it was mutating half-updated, and the +runtime used to recover the panic, nack the message, and feed the next one to +that same, now-suspect behavior instance. The supervision kernel closes that +gap by turning a panic into the in-process equivalent of the crash recovery +above. + +The runtime distinguishes the two failure shapes. A behavior that *returns* an +error is an ordinary message failure and retries per the `TellRetryPolicy`. A +behavior that *panics* is converted into a `behaviorPanic`, which the worker +recognises and hands to the actor's supervision loop. + +The restart sequence is: + +1. The delivery's normal ack/nack/dead-letter bookkeeping runs first, so the + message that triggered the panic burns an attempt exactly as it does today. + This ordering is what keeps a deterministic poison message climbing toward + `max_attempts` and the dead letter queue instead of restarting the actor + forever. On the classic path, where the whole `Receive` runs inside one + framework transaction, that bookkeeping is deliberately redone OUTSIDE the + transaction: supervision returns the panic from the transaction to force a + rollback of the behavior's own partial writes, because committing them + alongside the nack would persist the torn state the restart exists to escape + and the checkpoint reload would hand it straight back. +2. Supervision cancels the current worker *generation*, which drains every + worker of a `NumWorkers > 1` pool, not just the one that panicked. Several + workers panicking together cost one restart between them, not one each. +3. The behavior's `OnStop` hook runs (bounded by `CleanupTimeout`) if it + implements `Stoppable`. A panic escaping that hook is recovered rather than + allowed to take the process down, and terminates the actor. +4. The startup path re-runs: `LoadCheckpoint` followed by + `PrependRestartMessageWithID`, so the behavior can rebuild from its + persisted FSM state before it sees any other message. +5. A fresh generation of `NumWorkers` loops starts on the same mailbox, one + worker at a time behind a warm-up barrier (see below). + +#### Ordering the hand-off under a worker pool + +`RestartMessage` carries `RestartPriority`, which makes the claim query hand it +out before every other row. Under `NumWorkers > 1` that orders the *claims* and +not the *turns*, which is a weaker thing than it sounds: launch the whole pool +at once and one worker takes the restart while a sibling immediately takes the +row behind it, so a normal turn runs against a behavior instance that is still +rebuilding itself from the checkpoint. The documented guarantee that a restart +message is processed before all other messages needs more than a priority to +hold. + +So a generation warms up one worker at a time. The supervisor launches a single +worker, and the rest of the pool waits until that worker has resolved the +generation's restart hand-off. For a row supervision enqueued itself (any +supervised restart) the barrier waits for that row unconditionally. For the +boot hand-off, which an owner prepends before `Start` and which the actor +therefore never sees, the barrier orders the first claim instead and releases on +an idle tick, which orders the common case without being able to prove a row was +ever there. + +The release rule is shaped so the barrier cannot wedge a pool that has no +hand-off waiting for it: + +- A first claim of anything other than a restart message releases the pool + *before* that message is processed, so a generation with no hand-off pays + nothing and never serializes behind its first turn. +- A restore that fails is dead-lettered, and a restore that panics tears the + generation down. Both resolve the hand-off, and both release the pool. +- Whatever ends the warm-up worker releases the pool, including a closed + mailbox. +- A `Stop` landing mid-barrier releases it through the generation context, so + shutdown never parks behind a restore. + +A single-worker actor is already strictly sequential, so it gets no barrier and +runs the identical code path with nothing to pay for. + +#### What "restart" does and does not mean + +This is the part worth being precise about, because the name oversells it. The +framework does **not** rebuild the behavior. It stops the workers, optionally +calls `OnStop`, and redelivers a `RestartMessage`; the same Go value keeps +serving afterwards with whatever fields the panic left behind. The clean slate +therefore exists exactly when the actor's `RestartMessage` handler rebuilds +every piece of in-memory state from the durable row, and not otherwise. + +Every durable behavior in this repo that carries in-memory state carries a +reload seam for this already, because a rolled-back `Commit` poses the same +problem: `credit.opBehavior` and `oor.sessionBehavior` arm a `commitFailed` +guard that reloads before the next dispatch, `oor.oorRegistryBehavior` re-runs +its non-terminal restore, and `unroll.behavior` re-runs `restoreCheckpoint`. +Their `RestartMessage` handlers use those seams. A behavior with no in-memory +turn state, such as the `serverconn` egress sender, may consume the message as +a no-op, but it should say so and say why rather than leave the reader +guessing. + +Two constraints follow for handler authors. The restart message is enqueued +with `MaxAttempts` 1 and the runtime dead-letters rather than nacks a restart +turn that fails (a nacked row at `attempts == max_attempts` is neither +leasable nor reapable, so it would strand in the mailbox forever), which makes +restore handlers responsible for their own idempotency. And `OnStop` now runs +mid-life, once per restart, so it must be idempotent and must leave the +behavior able to serve a new generation rather than assuming it is being +discarded. + +An actor whose codec never registered the `RestartMessage` cannot be handed its +checkpoint at all. Supervision checks that up front, before tearing anything +down, and degrades to cycling the worker generation with no `OnStop` and no +restore: a teardown the behavior cannot be rebuilt from is strictly worse than +leaving it running. + +#### Identity, promises, and intensity + +The actor's public identity is untouched: same ID, same `DurableMailbox`, same +`Ref`. Senders keep enqueueing across the restart gap, and the mailbox's +promise registry survives, so a message that had not yet reached the behavior +is redelivered afterwards and its caller still gets a result. The turns that +were in flight do not survive: the panicking turn's promise is completed with +the panic error, and a sibling worker's turn sees its generation context +cancelled and completes its promise with that context error. + +Restarts can be bounded by a BEAM-style intensity budget, +`DurableActorConfig.MaxRestarts` restarts inside `RestartWindow`, tracked in a +sliding window. It defaults **off** (`DefaultMaxRestarts` is +`UnlimitedRestarts`). Restarting forever is no worse than the nack-and-continue +loop supervision replaces, since both are rate-limited by the nack backoff, +whereas a finite budget introduces a failure mode the runtime did not have: +breaking it is terminal, and a terminated actor still holds its ID and its +mailbox rows, so it keeps looking alive to anything that is not watching. Set a +finite budget only where the owner wires `Watch` and reacts to it; +`RecommendedMaxRestarts` (5 per 60s) is the value to reach for when you do. +Breaking the budget makes the actor log at error level, cancel its lifetime +context so further sends fail fast, tear down, and publish its termination. + +Supervision also keeps the mailbox tidy across a long run of restarts: each +restart deletes the restart row the previous one enqueued before writing its +own, so at most one restart row is pending at a time rather than one per +restart. + +`(*DurableActor).Watch(ctx)` is how another component observes the terminal +event. It returns a channel that receives exactly one `TerminationInfo` and is +then closed, carrying the reason (stopped, context cancelled, restart intensity +exceeded, restart failed), the failure behind it, the lifetime restart count, +and whether the restart budget was exhausted. Delivery is non-blocking by +construction, so a slow watcher can never park the actor's shutdown path. +`Stop` publishes the notification for an actor that was never started, so a +watcher on one does not wait forever. + --- ## TypeAssertingRef and MapRef Pattern diff --git a/oor/registry.go b/oor/registry.go index aae5f23b9..e5f639c4b 100644 --- a/oor/registry.go +++ b/oor/registry.go @@ -464,9 +464,22 @@ func (r *oorRegistryBehavior) Receive(ctx context.Context, msg OORDurableMsg, switch m := msg.(type) { case *actor.RestartMessage: - // The active set is rebuilt by RestoreNonTerminal; the restart - // message carries no state to persist. - return fn.Ok[ActorResp](&DriveEventResponse{}) + // A restart message reaches a live registry in exactly one + // case: the actor's supervisor restarted it after this + // behavior panicked. The framework reuses the same behavior + // instance across the restart, so the goroutine-owned routing + // state (the active child set and any staged handoff) is + // whatever the panic left behind. Drop the staged handoff, + // which belonged to the turn that died, and re-run the same + // non-terminal restore the boot path uses so the active set is + // reconciled against the durable rows again. The restore skips + // sessions that are already resident, so re-running it against + // a mostly-intact active set is cheap and idempotent. + r.pendingHandoff = nil + + return r.handleRestoreNonTerminal( + ctx, &RestoreNonTerminalRequest{}, + ) case *GetStateRequest: return r.routeAsk(ctx, m.SessionID, m) diff --git a/oor/session_actor.go b/oor/session_actor.go index 7c8bdbcb9..7ece1df94 100644 --- a/oor/session_actor.go +++ b/oor/session_actor.go @@ -383,9 +383,24 @@ func (b *sessionBehavior) Receive(ctx context.Context, msg OORDurableMsg, switch msg.(type) { case *actor.RestartMessage: - // Restore already ran at construction; the restart message - // carries no state to persist, so the framework consumes it via - // the non-transactional ack path. + // A restart message reaches a live behavior in exactly one + // case: the actor's supervisor restarted it after this + // behavior panicked. The panic may have left b.fsm advanced in + // memory past the last durably-committed snapshot, and the + // framework reuses this same behavior instance across the + // restart, so consuming the message as a no-op would hand that + // stale advance to the next driving event. Arm the same reload + // guard the failed-commit path uses: the next turn stops the + // stale FSM and rebuilds it from the registry row before it + // dispatches. + // + // The rebuild is deferred to that turn rather than run here + // because restore() starts an FSM goroutine that must outlive + // the turn, and because the framework delivers the restart + // message with max_attempts 1 and dead-letters (rather than + // retries) a restart turn that fails. + b.commitFailed = true + return fn.Ok[ActorResp](&DriveEventResponse{}) case *GetStateRequest: diff --git a/serverconn/actor.go b/serverconn/actor.go index 765bd4ba2..52ad4607c 100644 --- a/serverconn/actor.go +++ b/serverconn/actor.go @@ -773,6 +773,23 @@ func (a *ServerConnectionActor) Receive(ctx context.Context, msg ServerConnMsg, ax actor.Exec[egressTx]) fn.Result[ServerConnResp] { switch m := msg.(type) { + case *restartMsg: + // The egress sender keeps no per-turn state that a checkpoint + // could restore, so a supervised restart has nothing to rebuild + // here. Each turn reads its whole input from the durable + // message, converts it, sends it over the edge, and consumes + // the message in one Commit; nothing carries over between + // turns. What in-memory state the connector does hold belongs + // to the CONNECTION, not to a turn: the unary response + // registry, the last-send timestamp, the cached + // incompatibility, and the ingress cancel are all owned by the + // ingress loop and its callers, and a panicking egress turn + // cannot leave any of them half-written. Consuming the restart + // as an explicit no-op is therefore the whole restore, and + // saying so here is what keeps it from looking like an + // oversight. + return fn.Ok[ServerConnResp](&SendClientEventResponse{}) + case *SendClientEventRequest: return a.handleSendClientEvent(ctx, m, ax) @@ -1314,9 +1331,31 @@ func NewServerConnCodec() *actor.MessageCodec { }, ) + // The actor framework prepends its own RestartMessage when it restarts + // the egress actor from its checkpoint. Register the adapter rather + // than the framework type directly: the durable mailbox casts every + // decoded message to ServerConnMsg, and the bare framework type does + // not satisfy this package's seal, so an unadapted restart would fail + // that cast and dead-letter instead of being handled. + codec.MustRegister( + actor.RestartTLVType, + func() actor.TLVMessage { return &restartMsg{} }, + ) + return codec } +// restartMsg adapts the actor framework's RestartMessage into this package's +// sealed message surface. It adds nothing but the seal: encoding, decoding, +// the TLV type, and the restart priority are all the embedded framework +// message's. +type restartMsg struct { + actor.RestartMessage +} + +// serverConnMsgSealed implements the ServerConnMsg interface seal. +func (m *restartMsg) serverConnMsgSealed() {} + // Compile-time interface checks. var ( _ ServerConnMsg = (*SendClientEventRequest)(nil) diff --git a/unroll/actor.go b/unroll/actor.go index 13f6c30d4..7f03df79f 100644 --- a/unroll/actor.go +++ b/unroll/actor.go @@ -245,6 +245,25 @@ func (b *behavior) Receive(ctx context.Context, msg Msg, return fn.Ok[Resp](b.stateResponse(ctx)) } + // A restart message reaches a live behavior in exactly one case: the + // actor's supervisor restarted it after this behavior panicked. The + // framework reuses the same behavior instance across the restart, so a + // panic that advanced b.pending or b.sweepTx in memory past the last + // Staged checkpoint would otherwise survive it. Re-run the same + // checkpoint restore the constructor runs, which overwrites both from + // the durable row. It drives no FSM transition and writes nothing, so + // it returns before the Commit like the status probe above. + if _, ok := msg.(*restartMsg); ok { + if err := b.restoreCheckpoint(ctx); err != nil { + return fn.Err[Resp]( + fmt.Errorf("restore checkpoint on restart: %w", + err), + ) + } + + return fn.Ok[Resp](b.stateResponse(ctx)) + } + // Run the FSM pipeline. Every checkpoint write inside is a short, // lock-releasing Stage and the slow txconfirm IO runs with no writer // transaction held; dispatch never commits. diff --git a/unroll/messages.go b/unroll/messages.go index 09e3bd00d..464e2fe50 100644 --- a/unroll/messages.go +++ b/unroll/messages.go @@ -801,5 +801,30 @@ func newCodec() *actor.MessageCodec { func() actor.TLVMessage { return &SpendObservedMsg{} }, ) + // The actor framework prepends its own RestartMessage when it restarts + // this actor from its checkpoint. Register the adapter rather than the + // framework type directly: the durable mailbox casts every decoded + // message to Msg, and the bare framework type does not satisfy this + // package's seal, so an unadapted restart would fail that cast and + // dead-letter instead of restoring. + codec.MustRegister( + actor.RestartTLVType, + func() actor.TLVMessage { return &restartMsg{} }, + ) + return codec } + +// restartMsg adapts the actor framework's RestartMessage into this package's +// sealed message surface. It adds nothing but the seal: encoding, decoding, +// the TLV type, and the restart priority are all the embedded framework +// message's. +type restartMsg struct { + actor.RestartMessage +} + +// unrollMsgSealed implements the Msg interface seal. +func (m *restartMsg) unrollMsgSealed() {} + +// Compile-time check that the restart adapter is a durable unroll message. +var _ Msg = (*restartMsg)(nil)