diff --git a/baselib/actor/AGENTS.md b/baselib/actor/AGENTS.md index cb4a2129c..642279272 100644 --- a/baselib/actor/AGENTS.md +++ b/baselib/actor/AGENTS.md @@ -26,13 +26,35 @@ crash-safe at-least-once delivery with exactly-once deduplication. - `Receptionist` — Service locator mapping `ServiceKey` → `ActorRef` for decoupled actor wiring. - `Message` — Sealed interface for all actor messages (must embed `BaseMessage`). - `MessageCodec` — TLV-based codec for message serialization/deserialization. -- `DeliveryStore` / `TxAwareDeliveryStore` — Interfaces for durable mailbox persistence (enqueue, claim, ack, dead-letter). The leaseless single-worker fast path adds `PeekNextMessage` (read-only claim, no lease, no attempts bump; yields an empty lease token), `AckMessageByID` (unfenced delete), and `NackMessageByID` (unfenced release that increments attempts). A `DurableActor` enables it (via `DurableMailboxConfig.SingleWorkerLeaseless`) strictly when `NumWorkers == 1` AND the behavior is the Read/Commit (Right/`TxBehavior`) path, eliminating the per-message lease write transaction. The multi-worker pool and the classic path are byte-for-byte unchanged: they keep `LeaseNextMessage` and the lease-fenced ack. Ack/nack route to the by-ID ops automatically whenever the delivery's lease token is empty; `Delivery.ShouldDeadLetter` counts the in-flight attempt as `Attempts + 1` on the leaseless path so the dead-letter boundary matches the leased path (where attempts is pre-incremented at lease). +- `DeliveryStore` / `TxAwareDeliveryStore` — Interfaces for durable mailbox persistence (enqueue, claim, ack, dead-letter). The leaseless single-worker fast path adds `PeekNextMessage` (read-only claim, no lease, no attempts bump; yields an empty lease token), `AckMessageByID` (unfenced delete), and `NackMessageByID` (unfenced release that increments attempts). Postpone adds the same fenced/unfenced pair: `PostponeMessage` (lease-fenced release that decrements attempts to compensate the lease-time bump) and `PostponeMessageByID` (unfenced release that leaves attempts untouched, since the peek never bumped them). A `DurableActor` enables it (via `DurableMailboxConfig.SingleWorkerLeaseless`) strictly when `NumWorkers == 1` AND the behavior is the Read/Commit (Right/`TxBehavior`) path, eliminating the per-message lease write transaction. The multi-worker pool and the classic path are byte-for-byte unchanged: they keep `LeaseNextMessage` and the lease-fenced ack. Ack/nack route to the by-ID ops automatically whenever the delivery's lease token is empty; `Delivery.ShouldDeadLetter` counts the in-flight attempt as `Attempts + 1` on the leaseless path so the dead-letter boundary matches the leased path (where attempts is pre-incremented at lease). - `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. - `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. +- `Postpone(delay) error` / `PostponeError` / `ErrPostponed` — The + attempt-preserving alternative to a nack, for a behavior that cannot handle + a message *yet* (a capacity cap, a peer still draining) as opposed to one + that failed. Returning `actor.Postpone(delay)` from a Tell turn releases the + message for redelivery after `delay` with `attempts` unchanged and without + marking it processed. Detection matches anywhere in the wrap chain, so a + behavior may annotate it (`fmt.Errorf("%w: %w", errCapped, + actor.Postpone(d))`). `ErrPostponed` is the `errors.Is` sentinel; + `*PostponeError.Delay` carries the backoff. Pass a real delay: zero or + negative makes the message immediately claim-eligible, which against an + unchanged condition is a busy loop against the database. +- `DeliveryEnqueuedAt(ctx) (time.Time, bool)` — When the message currently + being processed was first persisted, read from the durable row's `created_at` + and stamped onto the processing context by the consume path (once, above the + fork into the three execution paths, so all of them agree). Neither a nack + nor a postpone rewrites that column, so it survives every redelivery. This is + the intended horizon reference for a postponing behavior, and the reason it + is row-derived rather than behavior-derived: per-message state keyed on a + sender-chosen id is unbounded when the message stream is attacker-controlled. + The bool is false outside a delivery and for a store that reports no + timestamp. `Delivery.EnqueuedAt` is the same value on the delivery itself; + `WithDeliveryEnqueuedAtForTest` stamps a context for cross-package tests. +- `TellRetryPolicy` — Function type `func(attempts int, lastErr error) (bool, time.Duration)` determining retry behavior for failed Tell messages. Return `(false, _)` to dead-letter immediately. A postpone is detected *before* this policy is consulted and never reaches it. - `DefaultTellRetryPolicy` — Exponential backoff policy: up to 5 attempts, starting at 1s, capped at 60s. - `Checkpoint` — Serializable actor state snapshot for recovery. - `WithoutOutboxID` — Context helper that strips the propagated outbox ID so child operations do not inherit the parent's delivery tracking scope. @@ -111,6 +133,55 @@ crash-safe at-least-once delivery with exactly-once deduplication. the message. A `context.WithTimeout` around `Tell` is the weaker option: it burns the entire deadline against a peer that is already wedged. - During daemon teardown, the underlying DB is closed before every actor's lease loop has wound down. The lease loop uses `isExpectedShutdownErr` to demote these "database is closed" errors to debug level; real operational errors still surface as warnings because neither the actor context nor the outer context is done in those cases. +- **Postpone preserves the attempt budget; nack spends it.** A nack increments + `attempts` on every release, and both the claim and the peek queries filter + on `attempts < max_attempts`. A postpone leaves the budget exactly as it was: + the fenced `PostponeMessage` decrements to compensate the lease-time + increment, and the leaseless `PostponeMessageByID` leaves it untouched + because the peek never bumped it. An "always retry" `TellRetryPolicy` is + **not** a substitute, and what it actually does depends on the release path: + on the non-tx path (`handleResult` via `Delivery.Nack`, which the Read/Commit + `finishNonTx` tail also uses) `Nack` checks `ShouldDeadLetter` first, so the + message dead-letters at exhaustion regardless of the policy; on the tx path + (`handleResultInTx`) the retry branch calls the store nack with no + `ShouldDeadLetter` arm, so the row is nacked past `max_attempts` and goes + dark without ever reaching `dead_letters` (a pre-existing tx-path bug, not + something postpone introduces). Either way, a policy override cannot express + "wait indefinitely"; only a postpone can. +- **Postpone is Tell-only.** An Ask has a caller parked on the promise, so + postponing it would strand that caller for the length of the delay with + nothing to observe. An Ask behavior returning a `PostponeError` gets ordinary + error treatment (the promise completes with it) and the caller decides + whether to re-issue. A behavior serving the same condition over both a routed + Tell and an RPC Ask should postpone on the Tell path only. +- **A postponed message never auto-dead-letters, so behaviors bound their own + horizon.** This is the deliberate cost of the feature: postpone removes the + only mechanism that would eventually give up. A behavior that postpones + against a condition that never clears postpones forever. The framework cannot + bound this, because only the behavior knows when waiting stops making sense. + Track the wait and return a real error once it is no longer justified, so the + normal nack path can dead-letter it. Use `DeliveryEnqueuedAt(ctx) (time.Time, + bool)` for that: it reports the durable row's `created_at`, which neither a + nack nor a postpone rewrites, so it measures the true age of the wait across + every redelivery. Prefer it over behavior-side state whenever the message + stream is attacker-controlled, since a map keyed on a sender-chosen id is + itself unbounded. The bool distinguishes "no timestamp" from "age zero"; + treat absence as no horizon information. +- **A postponed head still blocks its correlation-key lane.** Postpone does not + exempt a message from per-key FIFO: the row is still in the mailbox, so no + later same-key message is claim-eligible until it drains, and every further + postpone extends the block. Blocking is bounded to the key, not the mailbox, + but a long backoff on a busy key is a throughput decision, not just a retry + decision. Unkeyed messages have no such interaction. **Scope:** per-key FIFO + holds for a postponing consumer only when the actor is single-worker OR the + lane's messages never reach their final attempt. A predecessor leased on its + final attempt is invisible to the claim anti-join (`m2.attempts < + m2.max_attempts`), so a pool worker can claim its successor; a postpone then + decrements the predecessor back below the cap and it reprocesses after the + successor, inverting order. No adopter combines keyed lanes with + `NumWorkers > 1` and postpone today, so the SQL is left alone; adding a + lease-liveness disjunct to the anti-join is the prerequisite for that + combination. - **Per-correlation-key FIFO claim.** Two messages in the same mailbox that share a non-empty `CorrelationKey()` are processed in emission order regardless of retry backoff. Without this invariant, a transient Tell diff --git a/baselib/actor/CLAUDE.md b/baselib/actor/CLAUDE.md index cb4a2129c..642279272 100644 --- a/baselib/actor/CLAUDE.md +++ b/baselib/actor/CLAUDE.md @@ -26,13 +26,35 @@ crash-safe at-least-once delivery with exactly-once deduplication. - `Receptionist` — Service locator mapping `ServiceKey` → `ActorRef` for decoupled actor wiring. - `Message` — Sealed interface for all actor messages (must embed `BaseMessage`). - `MessageCodec` — TLV-based codec for message serialization/deserialization. -- `DeliveryStore` / `TxAwareDeliveryStore` — Interfaces for durable mailbox persistence (enqueue, claim, ack, dead-letter). The leaseless single-worker fast path adds `PeekNextMessage` (read-only claim, no lease, no attempts bump; yields an empty lease token), `AckMessageByID` (unfenced delete), and `NackMessageByID` (unfenced release that increments attempts). A `DurableActor` enables it (via `DurableMailboxConfig.SingleWorkerLeaseless`) strictly when `NumWorkers == 1` AND the behavior is the Read/Commit (Right/`TxBehavior`) path, eliminating the per-message lease write transaction. The multi-worker pool and the classic path are byte-for-byte unchanged: they keep `LeaseNextMessage` and the lease-fenced ack. Ack/nack route to the by-ID ops automatically whenever the delivery's lease token is empty; `Delivery.ShouldDeadLetter` counts the in-flight attempt as `Attempts + 1` on the leaseless path so the dead-letter boundary matches the leased path (where attempts is pre-incremented at lease). +- `DeliveryStore` / `TxAwareDeliveryStore` — Interfaces for durable mailbox persistence (enqueue, claim, ack, dead-letter). The leaseless single-worker fast path adds `PeekNextMessage` (read-only claim, no lease, no attempts bump; yields an empty lease token), `AckMessageByID` (unfenced delete), and `NackMessageByID` (unfenced release that increments attempts). Postpone adds the same fenced/unfenced pair: `PostponeMessage` (lease-fenced release that decrements attempts to compensate the lease-time bump) and `PostponeMessageByID` (unfenced release that leaves attempts untouched, since the peek never bumped them). A `DurableActor` enables it (via `DurableMailboxConfig.SingleWorkerLeaseless`) strictly when `NumWorkers == 1` AND the behavior is the Read/Commit (Right/`TxBehavior`) path, eliminating the per-message lease write transaction. The multi-worker pool and the classic path are byte-for-byte unchanged: they keep `LeaseNextMessage` and the lease-fenced ack. Ack/nack route to the by-ID ops automatically whenever the delivery's lease token is empty; `Delivery.ShouldDeadLetter` counts the in-flight attempt as `Attempts + 1` on the leaseless path so the dead-letter boundary matches the leased path (where attempts is pre-incremented at lease). - `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. - `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. +- `Postpone(delay) error` / `PostponeError` / `ErrPostponed` — The + attempt-preserving alternative to a nack, for a behavior that cannot handle + a message *yet* (a capacity cap, a peer still draining) as opposed to one + that failed. Returning `actor.Postpone(delay)` from a Tell turn releases the + message for redelivery after `delay` with `attempts` unchanged and without + marking it processed. Detection matches anywhere in the wrap chain, so a + behavior may annotate it (`fmt.Errorf("%w: %w", errCapped, + actor.Postpone(d))`). `ErrPostponed` is the `errors.Is` sentinel; + `*PostponeError.Delay` carries the backoff. Pass a real delay: zero or + negative makes the message immediately claim-eligible, which against an + unchanged condition is a busy loop against the database. +- `DeliveryEnqueuedAt(ctx) (time.Time, bool)` — When the message currently + being processed was first persisted, read from the durable row's `created_at` + and stamped onto the processing context by the consume path (once, above the + fork into the three execution paths, so all of them agree). Neither a nack + nor a postpone rewrites that column, so it survives every redelivery. This is + the intended horizon reference for a postponing behavior, and the reason it + is row-derived rather than behavior-derived: per-message state keyed on a + sender-chosen id is unbounded when the message stream is attacker-controlled. + The bool is false outside a delivery and for a store that reports no + timestamp. `Delivery.EnqueuedAt` is the same value on the delivery itself; + `WithDeliveryEnqueuedAtForTest` stamps a context for cross-package tests. +- `TellRetryPolicy` — Function type `func(attempts int, lastErr error) (bool, time.Duration)` determining retry behavior for failed Tell messages. Return `(false, _)` to dead-letter immediately. A postpone is detected *before* this policy is consulted and never reaches it. - `DefaultTellRetryPolicy` — Exponential backoff policy: up to 5 attempts, starting at 1s, capped at 60s. - `Checkpoint` — Serializable actor state snapshot for recovery. - `WithoutOutboxID` — Context helper that strips the propagated outbox ID so child operations do not inherit the parent's delivery tracking scope. @@ -111,6 +133,55 @@ crash-safe at-least-once delivery with exactly-once deduplication. the message. A `context.WithTimeout` around `Tell` is the weaker option: it burns the entire deadline against a peer that is already wedged. - During daemon teardown, the underlying DB is closed before every actor's lease loop has wound down. The lease loop uses `isExpectedShutdownErr` to demote these "database is closed" errors to debug level; real operational errors still surface as warnings because neither the actor context nor the outer context is done in those cases. +- **Postpone preserves the attempt budget; nack spends it.** A nack increments + `attempts` on every release, and both the claim and the peek queries filter + on `attempts < max_attempts`. A postpone leaves the budget exactly as it was: + the fenced `PostponeMessage` decrements to compensate the lease-time + increment, and the leaseless `PostponeMessageByID` leaves it untouched + because the peek never bumped it. An "always retry" `TellRetryPolicy` is + **not** a substitute, and what it actually does depends on the release path: + on the non-tx path (`handleResult` via `Delivery.Nack`, which the Read/Commit + `finishNonTx` tail also uses) `Nack` checks `ShouldDeadLetter` first, so the + message dead-letters at exhaustion regardless of the policy; on the tx path + (`handleResultInTx`) the retry branch calls the store nack with no + `ShouldDeadLetter` arm, so the row is nacked past `max_attempts` and goes + dark without ever reaching `dead_letters` (a pre-existing tx-path bug, not + something postpone introduces). Either way, a policy override cannot express + "wait indefinitely"; only a postpone can. +- **Postpone is Tell-only.** An Ask has a caller parked on the promise, so + postponing it would strand that caller for the length of the delay with + nothing to observe. An Ask behavior returning a `PostponeError` gets ordinary + error treatment (the promise completes with it) and the caller decides + whether to re-issue. A behavior serving the same condition over both a routed + Tell and an RPC Ask should postpone on the Tell path only. +- **A postponed message never auto-dead-letters, so behaviors bound their own + horizon.** This is the deliberate cost of the feature: postpone removes the + only mechanism that would eventually give up. A behavior that postpones + against a condition that never clears postpones forever. The framework cannot + bound this, because only the behavior knows when waiting stops making sense. + Track the wait and return a real error once it is no longer justified, so the + normal nack path can dead-letter it. Use `DeliveryEnqueuedAt(ctx) (time.Time, + bool)` for that: it reports the durable row's `created_at`, which neither a + nack nor a postpone rewrites, so it measures the true age of the wait across + every redelivery. Prefer it over behavior-side state whenever the message + stream is attacker-controlled, since a map keyed on a sender-chosen id is + itself unbounded. The bool distinguishes "no timestamp" from "age zero"; + treat absence as no horizon information. +- **A postponed head still blocks its correlation-key lane.** Postpone does not + exempt a message from per-key FIFO: the row is still in the mailbox, so no + later same-key message is claim-eligible until it drains, and every further + postpone extends the block. Blocking is bounded to the key, not the mailbox, + but a long backoff on a busy key is a throughput decision, not just a retry + decision. Unkeyed messages have no such interaction. **Scope:** per-key FIFO + holds for a postponing consumer only when the actor is single-worker OR the + lane's messages never reach their final attempt. A predecessor leased on its + final attempt is invisible to the claim anti-join (`m2.attempts < + m2.max_attempts`), so a pool worker can claim its successor; a postpone then + decrements the predecessor back below the cap and it reprocesses after the + successor, inverting order. No adopter combines keyed lanes with + `NumWorkers > 1` and postpone today, so the SQL is left alone; adding a + lease-liveness disjunct to the anti-join is the prerequisite for that + combination. - **Per-correlation-key FIFO claim.** Two messages in the same mailbox that share a non-empty `CorrelationKey()` are processed in emission order regardless of retry backoff. Without this invariant, a transient Tell diff --git a/baselib/actor/delivery.go b/baselib/actor/delivery.go index 42e55bff2..889cbe5aa 100644 --- a/baselib/actor/delivery.go +++ b/baselib/actor/delivery.go @@ -61,6 +61,16 @@ type Delivery[M TLVMessage, R any] struct { // MaxAttempts is the maximum allowed attempts before dead-lettering. MaxAttempts int + // EnqueuedAt is when the message was first persisted to the mailbox. + // It is a property of the durable row, not of this delivery, so it + // survives every redelivery: neither a nack nor a postpone rewrites + // it. That makes it the one wall-clock reference a behavior can use to + // bound how long a message has been waiting without keeping per-message + // state of its own, which matters when the message stream is + // attacker-controlled and any in-memory map would be unbounded. See + // DeliveryEnqueuedAt for the behavior-facing accessor. + EnqueuedAt time.Time + // store is the backing store for persisting ack/nack operations. store DeliveryStore @@ -323,6 +333,44 @@ func (d *Delivery[M, R]) Nack( return nil } +// Postpone releases the message for redelivery after retryAfter WITHOUT +// burning a delivery attempt: unlike Nack, it never routes toward the +// dead-letter table and leaves the message's retry budget exactly as it was +// before this delivery. Use it for "not now" conditions (a capacity slot, a +// peer draining) rather than failures. Because attempts never advance, a +// perpetually-unmet condition postpones forever; the behavior owns bounding +// that horizon. +func (d *Delivery[M, R]) Postpone(ctx context.Context, + retryAfter time.Duration) error { + + d.mu.Lock() + if d.acked { + d.mu.Unlock() + + return ErrAlreadyAcked + } + d.mu.Unlock() + + rowsAffected, err := postponeMessage( + ctx, d.store, d.ID, d.LeaseToken, retryAfter, + ) + if err != nil { + d.setMutationFailed() + + return fmt.Errorf("postpone message: %w", err) + } + + if rowsAffected == 0 { + return ErrLeaseExpired + } + + d.mu.Lock() + d.acked = true + d.mu.Unlock() + + return nil +} + // Extend prolongs the lease for long-running message processing. This should // be called periodically for messages that take longer than the default lease // duration. Returns an error if the lease has already expired. @@ -377,6 +425,7 @@ func newDelivery[M TLVMessage, R any]( LeaseUntil: msg.LeaseUntil, Attempts: msg.Attempts, MaxAttempts: msg.MaxAttempts, + EnqueuedAt: msg.CreatedAt, store: store, acked: false, diff --git a/baselib/actor/delivery_store.go b/baselib/actor/delivery_store.go index 4e6d37ae6..c46bd50a8 100644 --- a/baselib/actor/delivery_store.go +++ b/baselib/actor/delivery_store.go @@ -60,6 +60,24 @@ type DeliveryStore interface { NackMessageByID(ctx context.Context, id string, retryAfter time.Duration) (int64, error) + // PostponeMessage releases a message for redelivery after the + // specified delay WITHOUT burning a delivery attempt: it decrements + // attempts to compensate the increment the lease took at claim, so + // the message's retry budget is exactly what it was before this + // delivery. Validates the lease token to prevent stale postpones. + // This is the attempt-preserving counterpart to NackMessage, used + // when a behavior returns a PostponeError ("not now") rather than a + // failure. + PostponeMessage(ctx context.Context, id, leaseToken string, + retryAfter time.Duration) (int64, error) + + // PostponeMessageByID is the leaseless single-worker counterpart to + // PostponeMessage: it releases the message by ID without validating + // a lease token and leaves attempts UNTOUCHED, because the leaseless + // peek never incremented it. Returns the number of rows affected. + PostponeMessageByID(ctx context.Context, id string, + retryAfter time.Duration) (int64, error) + // ExtendLease extends the lease for long-running message processing. // Validates the lease token to prevent stale extensions. ExtendLease(ctx context.Context, id, leaseToken string, diff --git a/baselib/actor/delivery_test.go b/baselib/actor/delivery_test.go index d0bf01374..1fd72bb74 100644 --- a/baselib/actor/delivery_test.go +++ b/baselib/actor/delivery_test.go @@ -297,6 +297,60 @@ func (m *mockDeliveryStore) NackMessageByID(ctx context.Context, id string, return 1, nil } +// PostponeMessage releases a leased message without burning an attempt, +// mirroring the real store: the decrement compensates the increment the +// lease took at claim, clamped at zero. +func (m *mockDeliveryStore) PostponeMessage(ctx context.Context, id, + leaseToken string, retryAfter time.Duration) (int64, error) { + + m.mu.Lock() + defer m.mu.Unlock() + + if m.injectError != nil { + return 0, m.injectError + } + + msg, ok := m.messages[id] + if !ok { + return 0, nil + } + + if msg.LeaseToken != leaseToken { + return 0, nil + } + + msg.LeaseToken = "" + msg.LeaseUntil = time.Time{} + if msg.Attempts > 0 { + msg.Attempts-- + } + + return 1, nil +} + +// PostponeMessageByID releases a message by ID without touching attempts, +// mirroring the unfenced leaseless postpone (the peek never bumped them). +func (m *mockDeliveryStore) PostponeMessageByID(ctx context.Context, id string, + retryAfter time.Duration) (int64, error) { + + m.mu.Lock() + defer m.mu.Unlock() + + if m.injectError != nil { + return 0, m.injectError + } + + msg, ok := m.messages[id] + if !ok { + return 0, nil + } + + msg.LeaseToken = "" + msg.LeaseUntil = time.Time{} + + return 1, nil +} + func (m *mockDeliveryStore) ExtendLease(ctx context.Context, id, leaseToken string, extension time.Duration) (int64, error) { diff --git a/baselib/actor/durable_actor.go b/baselib/actor/durable_actor.go index bee82ec62..8a6bd1b6d 100644 --- a/baselib/actor/durable_actor.go +++ b/baselib/actor/durable_actor.go @@ -14,6 +14,24 @@ import ( "github.com/lightningnetwork/lnd/fn/v2" ) +// errLostLeaseMidTurn reports that a lease-fenced release inside the +// transaction path matched no rows, which means this consumer no longer owns +// the message: the lease expired while the behavior was running and another +// consumer claimed it. +// +// It is returned from handleResultInTx for one reason only, to roll the +// transaction back, so the behavior's writes vanish rather than committing +// under a row somebody else now governs. processInTransaction recognizes it +// and returns without nacking, dead-lettering, or marking processed. +// +// The "without nacking" part is the load-bearing half. Delivery.Nack's +// dead-letter arm is unfenced: it calls MoveToDeadLetter and DeleteMessage by +// ID with no lease-token check, so routing a stale consumer through it would +// let us dead-letter (and delete) a message its legitimate owner is actively +// processing. Rolling back and walking away leaves the row entirely to that +// owner, whose own turn decides its fate. +var errLostLeaseMidTurn = errors.New("lease lost mid-turn; rolling back") + // TellRetryPolicy determines whether a failed Tell message should be retried // and how long to wait before the next attempt. type TellRetryPolicy func(err error, attempts int) (retry bool, delay time.Duration) @@ -627,6 +645,13 @@ func (a *DurableActor[M, R]) processDelivery(delivery *Delivery[M, R]) { } defer cancel() + // Expose the durable row's enqueue time to the behavior. This is the + // only wall-clock reference a postponing behavior can use to bound its + // own horizon without keeping per-message state, so it is stamped here, + // above the fork into the three execution paths, rather than in any one + // of them. + processCtx = withDeliveryEnqueuedAt(processCtx, delivery.EnqueuedAt) + logger(processCtx).TraceS(processCtx, "Durable actor processing message", "actor_id", a.id, "msg_type", delivery.Message.MessageType(), @@ -745,6 +770,31 @@ func (a *DurableActor[M, R]) processInTransaction(ctx context.Context, }) if err != nil { + // Losing the lease mid-turn is not a transaction failure to + // retry, it is a change of ownership. The rollback has already + // undone the behavior's writes, which is the correctness + // requirement, and the message now belongs to whichever + // consumer holds the lease. Do nothing further: no nack, no + // dead-letter, no processed mark. + // + // Nacking here would be actively harmful. Delivery.Nack + // dead-letters when the attempt budget is spent, and that arm + // is unfenced (MoveToDeadLetter and DeleteMessage run by ID + // with no lease check), so this stale consumer could delete a + // row its legitimate owner is in the middle of processing. + // Walking away leaves the row and its outcome to that owner. + if errors.Is(err, errLostLeaseMidTurn) { + logger(ctx).WarnS(ctx, + "Rolled back turn after losing lease", + err, + "actor_id", a.id, + "delivery_id", delivery.ID, + "msg_type", delivery.Message.MessageType(), + ) + + return + } + logger(ctx).WarnS(ctx, "Transaction failed, nacking message", err, @@ -845,10 +895,15 @@ func (a *DurableActor[M, R]) finishNonTx(ctx context.Context, // handleResult. shouldMarkProcessed = false } else if delivery.IsTell() && result.Err() != nil { - retry, _ := a.tellRetryPolicy( + // A postponed message always redelivers, so marking it + // processed here would dedup-skip the redelivery. Failures + // consult the retry policy as before. + if _, postponed := postponeDelay(result.Err()); postponed { + shouldMarkProcessed = false + } else if retry, _ := a.tellRetryPolicy( result.Err(), delivery.EffectiveAttempts(), - ) - if retry { + ); retry { + shouldMarkProcessed = false } } @@ -1092,6 +1147,49 @@ func (a *DurableActor[M, R]) executeBehaviorSafely(ctx context.Context, return classic.Receive(ctx, delivery.Message) } +// releasedRowsInTx interprets the row count from a release performed inside +// the transaction path, where a zero count has two very different meanings +// depending on whether the release was fenced. +// +// An unfenced (leaseless, empty-token) release that matched no rows just means +// the row is already gone, which is benign: the original processing consumed +// it and this is a duplicate. A fenced release that matched no rows means the +// lease-token comparison failed, so the lease expired mid-turn and another +// consumer owns the message now. That is not something to retry past, it is a +// reason to abandon the turn: returning errLostLeaseMidTurn rolls the +// transaction back so the behavior's writes never commit under a row this +// consumer no longer holds. +// +// The op argument names the release ("Postpone" or "Nack") for the log line. +func (a *DurableActor[M, R]) releasedRowsInTx(ctx context.Context, + delivery *Delivery[M, R], rows int64, op string) error { + + if rows != 0 { + return nil + } + + if delivery.LeaseToken == "" { + logger(ctx).DebugS(ctx, "Release found row already gone", + "actor_id", a.id, + "delivery_id", delivery.ID, + "op", op, + "msg_type", delivery.Message.MessageType(), + ) + + return nil + } + + logger(ctx).WarnS(ctx, "Release matched no rows; rolling back turn", + errLostLeaseMidTurn, + "actor_id", a.id, + "delivery_id", delivery.ID, + "op", op, + "msg_type", delivery.Message.MessageType(), + ) + + return errLostLeaseMidTurn +} + // handleResultInTx handles the result within a transaction. // It determines whether to ack, nack for retry, or dead-letter, and only // marks the message as processed when we won't retry (to avoid dedup issues). @@ -1117,6 +1215,7 @@ func (a *DurableActor[M, R]) handleResultInTx( LeaseUntil: delivery.LeaseUntil, Attempts: delivery.Attempts, MaxAttempts: delivery.MaxAttempts, + EnqueuedAt: delivery.EnqueuedAt, store: store, deferPromise: delivery.deferPromise, } @@ -1157,6 +1256,32 @@ func (a *DurableActor[M, R]) handleResultInTx( // For Tell messages, handle based on success/error. if err := result.Err(); err != nil { + // A postpone is control flow, not a failure: release the + // message without burning an attempt and without marking it + // processed, so it redelivers with its retry budget intact. + if delay, ok := postponeDelay(err); ok { + logger(ctx).DebugS( + ctx, + "Durable actor Tell message postponed", + "actor_id", a.id, + "delivery_id", delivery.ID, + "msg_type", delivery.Message.MessageType(), + "delay", delay, + ) + + rows, ppErr := postponeMessage( + ctx, store, delivery.ID, delivery.LeaseToken, + delay, + ) + if ppErr != nil { + return ppErr + } + + return a.releasedRowsInTx( + ctx, delivery, rows, "Postpone", + ) + } + effectiveAttempts := delivery.EffectiveAttempts() logger(ctx).WarnS(ctx, "Durable actor Tell message failed", @@ -1174,12 +1299,15 @@ func (a *DurableActor[M, R]) handleResultInTx( // Don't mark as processed - we want retry to work. // nackMessage routes a leaseless (empty-token) delivery // to the by-ID nack, which increments attempts. - _, nackErr := nackMessage( + rows, nackErr := nackMessage( ctx, store, delivery.ID, delivery.LeaseToken, delay, ) + if nackErr != nil { + return nackErr + } - return nackErr + return a.releasedRowsInTx(ctx, delivery, rows, "Nack") } // Max retries exceeded - dead letter. Mark as processed since @@ -1275,6 +1403,33 @@ func (a *DurableActor[M, R]) handleResult(ctx context.Context, // For Tell messages, handle based on success/error. if err := result.Err(); err != nil { + // A postpone is control flow, not a failure: release the + // message without burning an attempt so it redelivers with + // its retry budget intact. + if delay, ok := postponeDelay(err); ok { + logger(ctx).DebugS( + ctx, + "Durable actor Tell message postponed", + "actor_id", a.id, + "delivery_id", delivery.ID, + "msg_type", delivery.Message.MessageType(), + "delay", delay, + ) + + if ppErr := delivery.Postpone( + ctx, delay, + ); ppErr != nil { + + logger(ctx).WarnS(ctx, "Failed to postpone "+ + "Tell message", + ppErr, + "actor_id", a.id, + "delivery_id", delivery.ID) + } + + return + } + effectiveAttempts := delivery.EffectiveAttempts() logger(ctx).WarnS(ctx, "Durable actor Tell message failed", diff --git a/baselib/actor/durable_actor_test.go b/baselib/actor/durable_actor_test.go index b512f27ed..f2bf1477f 100644 --- a/baselib/actor/durable_actor_test.go +++ b/baselib/actor/durable_actor_test.go @@ -183,6 +183,32 @@ type mockTxAwareStore struct { // completing and the transaction committing, and is used to // verify that promises are not completed prematurely. txPostCallbackHook func() + + // txWrites models durable state a behavior writes inside the + // transaction, and txMu guards it. ExecTx snapshots its length before + // running fn and truncates back to that length when fn returns an + // error, giving the mock the rollback semantics a real store provides. + // Without this a test cannot tell a committed turn from a rolled-back + // one, since the mock otherwise mutates its maps in place. + txMu sync.Mutex + txWrites []string +} + +// recordTxWrite appends a write made by a behavior inside the transaction. +// ExecTx discards it again if the transaction rolls back. +func (m *mockTxAwareStore) recordTxWrite(write string) { + m.txMu.Lock() + defer m.txMu.Unlock() + + m.txWrites = append(m.txWrites, write) +} + +// committedTxWrites returns the writes that survived their transaction. +func (m *mockTxAwareStore) committedTxWrites() []string { + m.txMu.Lock() + defer m.txMu.Unlock() + + return append([]string(nil), m.txWrites...) } func newMockTxAwareStore() *mockTxAwareStore { @@ -204,8 +230,18 @@ func (m *mockTxAwareStore) ExecTx( return errors.New("simulated tx failure") } + // Snapshot the behavior-visible write log so a failed callback rolls + // back to exactly the state it found, the way a real transaction does. + m.txMu.Lock() + writesBefore := len(m.txWrites) + m.txMu.Unlock() + // Execute the function with the same store (simulating a transaction). if err := fn(ctx, m.mockDeliveryStore); err != nil { + m.txMu.Lock() + m.txWrites = m.txWrites[:writesBefore] + m.txMu.Unlock() + return err } diff --git a/baselib/actor/postpone.go b/baselib/actor/postpone.go new file mode 100644 index 000000000..c185b05cd --- /dev/null +++ b/baselib/actor/postpone.go @@ -0,0 +1,143 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "time" +) + +// A nack is the wrong tool for "not now": it burns one of the message's +// finite delivery attempts, so a consumer that is merely waiting on an +// external condition (a capacity slot, a peer draining, an operator catching +// up) walks an innocent message toward the dead-letter table one redelivery +// at a time. Postpone is the attempt-preserving alternative: the behavior +// returns a PostponeError, and the consume path releases the message for +// redelivery after the requested delay WITHOUT counting the attempt, so the +// message can wait out the condition indefinitely. +// +// The flip side is deliberate: a postponed message never climbs toward +// max_attempts, so nothing dead-letters it automatically. A behavior that +// postpones must bound its own horizon (give up with a real error once the +// wait stops making sense), or accept that the message waits forever. + +// deliveryEnqueuedAtKey keys the current delivery's enqueue timestamp in the +// processing context. +type deliveryEnqueuedAtKey struct{} + +// withDeliveryEnqueuedAt stamps the current delivery's enqueue timestamp onto +// the processing context. The consume path calls this once per delivery, +// before the behavior runs, so every execution path (tx fold, non-tx tail, and +// the Read/Commit Exec handle) exposes the same value. A zero timestamp is not +// stamped, so a store that does not report one leaves DeliveryEnqueuedAt +// reporting absence rather than a bogus epoch age. +func withDeliveryEnqueuedAt(ctx context.Context, + enqueuedAt time.Time) context.Context { + + if enqueuedAt.IsZero() { + return ctx + } + + return context.WithValue(ctx, deliveryEnqueuedAtKey{}, enqueuedAt) +} + +// DeliveryEnqueuedAt reports when the message currently being processed was +// first persisted to the mailbox, and whether that timestamp is available at +// all. The value comes from the durable row and survives every redelivery, +// since neither a nack nor a postpone rewrites it. +// +// This is the intended way for a postponing behavior to bound its own horizon. +// Postpone deliberately removes the attempt-based give-up mechanism, so a +// behavior that waits on an external condition needs some other reference to +// decide when waiting has stopped making sense. Deriving that from the row +// rather than from behavior-side state matters whenever the message stream is +// attacker-controlled: a per-message map keyed on anything the sender chooses +// is unbounded by construction, while the row's own age costs nothing to +// consult and cannot be inflated by fabricating new messages. +// +// The second return is false outside a delivery (so a behavior invoked +// directly in a test sees no timestamp) and for a store that does not report +// one. Treat absence as "no horizon information", not as "age zero". +func DeliveryEnqueuedAt(ctx context.Context) (time.Time, bool) { + enqueuedAt, ok := ctx.Value(deliveryEnqueuedAtKey{}).(time.Time) + + return enqueuedAt, ok +} + +// WithDeliveryEnqueuedAtForTest stamps a delivery enqueue timestamp onto ctx +// so a behavior in another package can be unit tested against its postpone +// horizon without standing up a durable actor and a real mailbox row. It is +// the exported form of what the consume path does for every delivery, and it +// exists only for tests. +func WithDeliveryEnqueuedAtForTest(ctx context.Context, + enqueuedAt time.Time) context.Context { + + return withDeliveryEnqueuedAt(ctx, enqueuedAt) +} + +// ErrPostponed is the sentinel matched by errors.Is for postpone requests. +// Behaviors construct one with Postpone; the consume path detects it and +// releases the delivery without burning an attempt. +var ErrPostponed = errors.New("delivery postponed") + +// PostponeError asks the consume path to re-enqueue the current message +// after Delay without counting the attempt. It is a control-flow signal, not +// a failure: the consume path logs it at debug level and never routes it +// toward the dead-letter table. +type PostponeError struct { + // Delay is how long the message stays invisible before it becomes + // claim-eligible again. A zero or negative value makes it eligible + // immediately, which on a still-unmet condition is a busy retry loop; + // callers should pass a real backoff. + Delay time.Duration +} + +// Error implements the error interface. +func (e *PostponeError) Error() string { + return fmt.Sprintf("delivery postponed for %v", e.Delay) +} + +// Unwrap exposes the ErrPostponed sentinel so errors.Is matches. +func (e *PostponeError) Unwrap() error { + return ErrPostponed +} + +// Postpone builds the error a behavior returns to re-enqueue the current +// Tell message after delay without burning a delivery attempt. Only Tell +// deliveries honor it: an Ask has a caller parked on the promise, so a +// postponed Ask would strand that caller for the length of the delay with +// nothing to observe. An Ask behavior that returns a PostponeError gets the +// ordinary error treatment (the promise completes with it) and the caller +// decides whether to re-issue. +func Postpone(delay time.Duration) error { + return &PostponeError{Delay: delay} +} + +// postponeDelay extracts the postpone request from a behavior error, if one +// is present. It matches anywhere in the wrap chain, so a behavior may +// annotate the postpone with context (fmt.Errorf("%w: over cap", ...)). +func postponeDelay(err error) (time.Duration, bool) { + var postpone *PostponeError + if errors.As(err, &postpone) { + return postpone.Delay, true + } + + return 0, false +} + +// postponeMessage releases a delivery for redelivery without burning an +// attempt, picking the fenced or unfenced store operation by whether a lease +// token is present, exactly as ackMessage/nackMessage do. The fenced variant +// decrements attempts to compensate the increment the lease took at claim; +// the by-ID variant leaves attempts untouched because the leaseless peek +// never bumped it. Either way the message's retry budget is exactly what it +// was before this delivery. +func postponeMessage(ctx context.Context, store DeliveryStore, id, + leaseToken string, retryAfter time.Duration) (int64, error) { + + if leaseToken == "" { + return store.PostponeMessageByID(ctx, id, retryAfter) + } + + return store.PostponeMessage(ctx, id, leaseToken, retryAfter) +} diff --git a/baselib/actor/postpone_test.go b/baselib/actor/postpone_test.go new file mode 100644 index 000000000..26564e85a --- /dev/null +++ b/baselib/actor/postpone_test.go @@ -0,0 +1,391 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/lightningnetwork/lnd/tlv" + "github.com/stretchr/testify/require" +) + +// TestPostponeDelayMatchesWrapChain pins the detection contract: a +// PostponeError is recognized bare, wrapped, and via errors.Is on the +// sentinel, and an ordinary error is not. +func TestPostponeDelayMatchesWrapChain(t *testing.T) { + t.Parallel() + + bare := Postpone(5 * time.Second) + delay, ok := postponeDelay(bare) + require.True(t, ok) + require.Equal(t, 5*time.Second, delay) + require.ErrorIs(t, bare, ErrPostponed) + + wrapped := fmt.Errorf("over cap: %w", Postpone(time.Minute)) + delay, ok = postponeDelay(wrapped) + require.True(t, ok) + require.Equal(t, time.Minute, delay) + + _, ok = postponeDelay(errors.New("real failure")) + require.False(t, ok) +} + +// TestDeliveryPostponeLeased verifies the fenced delivery-level postpone: +// the lease clears, the attempt the lease burned is restored, and a second +// postpone on the same delivery reports ErrAlreadyAcked. +func TestDeliveryPostponeLeased(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + ctx := context.Background() + + require.NoError( + t, + store.EnqueueMessage( + ctx, EnqueueParams{ + ID: "pp-1", + MailboxID: "mb", + MessageType: "t", + Payload: []byte{1}, + MaxAttempts: 3, + }, + ), + ) + + leased, err := store.LeaseNextMessage(ctx, "mb", "tok", time.Minute) + require.NoError(t, err) + require.Equal(t, 1, leased.Attempts) + + d := &Delivery[*durableTestMsg, int]{ + ID: "pp-1", + LeaseToken: "tok", + Attempts: leased.Attempts, + store: store, + } + + require.NoError(t, d.Postpone(ctx, time.Second)) + + store.mu.Lock() + msg := store.messages["pp-1"] + require.Zero(t, msg.Attempts) + require.Empty(t, msg.LeaseToken) + store.mu.Unlock() + + require.ErrorIs(t, d.Postpone(ctx, time.Second), ErrAlreadyAcked) +} + +// TestDeliveryPostponeLeaseless verifies the unfenced postpone leaves the +// attempts budget of a peeked (empty-token) delivery untouched. +func TestDeliveryPostponeLeaseless(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + ctx := context.Background() + + require.NoError( + t, + store.EnqueueMessage( + ctx, EnqueueParams{ + ID: "pp-2", + MailboxID: "mb", + MessageType: "t", + Payload: []byte{1}, + MaxAttempts: 3, + }, + ), + ) + + d := &Delivery[*durableTestMsg, int]{ + ID: "pp-2", + leaseless: true, + store: store, + } + + require.NoError(t, d.Postpone(ctx, time.Second)) + + store.mu.Lock() + require.Zero(t, store.messages["pp-2"].Attempts) + store.mu.Unlock() +} + +// TestDurableActorPostponeDoesNotBurnAttempts drives a full actor: the +// behavior postpones the first two deliveries and succeeds on the third. The +// message must redeliver past the point where the retry policy would have +// dead-lettered a nacked message, the retry policy must never be consulted, +// and the message must end processed rather than dead-lettered. +func TestDurableActorPostponeDoesNotBurnAttempts(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + codec := newActorTestCodec() + + var deliveries atomic.Int32 + behavior := newMockBehavior(fn.Err[int](Postpone(time.Millisecond))) + behavior.onReceive = func(ctx context.Context, msg *actorTestMsg) { + // Succeed on the third delivery. A nack-based retry with the + // policy below would have dead-lettered after the first. + if deliveries.Add(1) >= 3 { + behavior.setResult(fn.Ok(42)) + } + } + + cfg := DefaultDurableActorConfig("test-actor", behavior, store, codec) + cfg.PollInterval = 10 * time.Millisecond + + // A policy that never retries AND fails the test if consulted with a + // postpone: postpones are control flow and must bypass it entirely. + cfg.TellRetryPolicy = func(err error, attempts int) (bool, + time.Duration) { + + if _, ok := postponeDelay(err); ok { + t.Errorf("retry policy consulted for a postpone") + } + + return false, 0 + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(7)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + + // The message survives two postponed deliveries and completes on the + // third: consumed from the mailbox with nothing dead-lettered. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + return len(store.messages) == 0 && len(store.deadLetters) == 0 + }, 5*time.Second, 10*time.Millisecond) + + require.GreaterOrEqual(t, deliveries.Load(), int32(3)) + + // The attempts budget was never burned: every redelivery re-leased at + // attempt 1, so the message completed normally -- a dedup entry + // exists and no dead letter was written, even though a nack-based + // loop under this never-retry policy would have dead-lettered on the + // first failure. + store.mu.Lock() + require.Empty(t, store.deadLetters) + require.NotEmpty(t, store.processed) + store.mu.Unlock() +} + +// TestDurableActorPostponeInTransaction covers the postpone branch of +// handleResultInTx, the path a classic behavior takes when the actor has a +// tx-aware store. That branch is a separate implementation from the non-tx +// tail exercised above, so a regression in it would otherwise pass the suite: +// it must release the message without marking it processed, without burning an +// attempt, and without consulting the retry policy. +func TestDurableActorPostponeInTransaction(t *testing.T) { + t.Parallel() + + store := newMockTxAwareStore() + codec := newActorTestCodec() + + var deliveries atomic.Int32 + behavior := newMockBehavior(fn.Err[int](Postpone(time.Millisecond))) + behavior.onReceive = func(ctx context.Context, msg *actorTestMsg) { + // Succeed on the third delivery. Under the never-retry policy + // below a nack would have dead-lettered on the first. + if deliveries.Add(1) >= 3 { + behavior.setResult(fn.Ok(11)) + } + } + + // Passing the tx-aware store as the delivery store is what selects the + // transaction path: construction type-asserts it up to + // TxAwareDeliveryStore. + cfg := DefaultDurableActorConfig( + "tx-postpone-actor", behavior, store, codec, + ) + cfg.PollInterval = 10 * time.Millisecond + + // Fail the test if the retry policy is consulted for a postpone: the + // tx path must detect it first, exactly as the non-tx tail does. + cfg.TellRetryPolicy = func(err error, attempts int) (bool, + time.Duration) { + + if _, ok := postponeDelay(err); ok { + t.Errorf("retry policy consulted for a postpone") + } + + return false, 0 + } + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(11)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + + // The message survives two postponed deliveries on the transaction + // path and completes on the third. + require.Eventually(t, func() bool { + store.mu.Lock() + defer store.mu.Unlock() + + return len(store.messages) == 0 && len(store.deadLetters) == 0 + }, 5*time.Second, 10*time.Millisecond) + + require.GreaterOrEqual(t, deliveries.Load(), int32(3)) + require.True(t, store.txExecuted.Load()) + + // A postpone is not a nack: the tx path must never have taken the nack + // branch, and nothing dead-lettered despite a policy that would have + // dead-lettered a nack on the very first failure. + require.False(t, store.nackCalled.Load()) + + store.mu.Lock() + require.Empty(t, store.deadLetters) + require.NotEmpty(t, store.processed) + store.mu.Unlock() +} + +// TestPostponeRollsBackAfterLostLease verifies the transaction path abandons a +// turn whose lease was stolen while the behavior ran, rather than committing +// under a row it no longer owns. +// +// The fenced postpone matches no rows once the lease token has changed, which +// is the only signal that another consumer took over. The turn must then roll +// back so the behavior's writes vanish, and it must NOT fall through to +// Delivery.Nack: that path dead-letters once the attempt budget is spent, and +// its dead-letter arm runs MoveToDeadLetter and DeleteMessage by ID with no +// lease check, so a stale consumer could delete a message its legitimate owner +// is actively processing. +func TestPostponeRollsBackAfterLostLease(t *testing.T) { + t.Parallel() + + store := newMockTxAwareStore() + codec := newActorTestCodec() + + // Put the delivery on its final attempt so a nack would dead-letter + // immediately. That makes the "no nack" assertion meaningful: if the + // lost lease leaked into the nack path, the row would be gone. + const maxAttempts = 1 + + var deliveries atomic.Int32 + behavior := newMockBehavior(fn.Err[int](Postpone(time.Hour))) + behavior.onReceive = func(ctx context.Context, msg *actorTestMsg) { + deliveries.Add(1) + + // The behavior writes durable state, which must not survive a + // rolled-back turn. + store.recordTxWrite("write-from-stale-consumer") + + // Steal the lease mid-turn, exactly as an expiry followed by + // another consumer's claim would: the row is still there, but + // its token no longer matches the one this delivery holds. + store.mu.Lock() + for id := range store.messages { + store.messages[id].LeaseToken = "stolen-by-other" + } + store.mu.Unlock() + } + + cfg := DefaultDurableActorConfig( + "lost-lease-actor", behavior, store, codec, + ) + cfg.PollInterval = 10 * time.Millisecond + cfg.MaxAttempts = maxAttempts + + a := NewDurableActor(cfg).UnwrapOrFail(t) + a.Start() + defer a.Stop() + + msg := &actorTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(21)), + } + require.NoError(t, a.Ref().Tell(context.Background(), msg)) + + // Wait for the turn that loses the lease to run. + require.Eventually(t, func() bool { + return deliveries.Load() >= 1 + }, 5*time.Second, 10*time.Millisecond) + + // Give the consume path room to do the wrong thing if it is going to. + time.Sleep(200 * time.Millisecond) + + // The behavior's writes rolled back with the transaction: nothing it + // staged under the stolen row may survive. + require.Empty(t, store.committedTxWrites()) + + // The row still belongs to its new owner: not dead-lettered, not + // deleted, and still carrying the thief's lease token. + store.mu.Lock() + require.Len(t, store.messages, 1) + for _, queued := range store.messages { + require.Equal(t, "stolen-by-other", queued.LeaseToken) + } + require.Empty(t, store.deadLetters) + + // Nor was it marked processed: dedup must not swallow the redelivery + // the legitimate owner is about to perform. + require.Empty(t, store.processed) + store.mu.Unlock() + + // And the stale consumer never nacked, which is what would have + // dead-lettered the row at this attempt count. + require.False(t, store.nackCalled.Load()) +} + +// TestDeliveryEnqueuedAtFromContext verifies the enqueue timestamp a +// postponing behavior uses to bound its own horizon is plumbed from the +// durable row onto the processing context, and that absence is distinguishable +// from a zero time. A behavior that could not tell those apart would treat +// every store that omits the timestamp as reporting an infinitely old message. +func TestDeliveryEnqueuedAtFromContext(t *testing.T) { + t.Parallel() + + // No delivery in scope: no timestamp, and the zero value is not + // mistaken for a real one. + _, ok := DeliveryEnqueuedAt(context.Background()) + require.False(t, ok) + + // A zero enqueue time is not stamped, so it still reports absence + // rather than an epoch-aged message. + zeroCtx := withDeliveryEnqueuedAt(context.Background(), time.Time{}) + _, ok = DeliveryEnqueuedAt(zeroCtx) + require.False(t, ok) + + enqueuedAt := time.Date(2026, 8, 7, 12, 0, 0, 0, time.UTC) + ctx := withDeliveryEnqueuedAt(context.Background(), enqueuedAt) + + got, ok := DeliveryEnqueuedAt(ctx) + require.True(t, ok) + require.True(t, enqueuedAt.Equal(got)) +} + +// TestDeliveryCarriesEnqueuedAt verifies newDelivery copies the mailbox row's +// creation time onto the delivery. That timestamp is the horizon reference for +// every postponing behavior, and unlike attempts it must survive redelivery +// untouched, so it has to come from the row rather than from the claim. +func TestDeliveryCarriesEnqueuedAt(t *testing.T) { + t.Parallel() + + createdAt := time.Date(2026, 8, 7, 9, 30, 0, 0, time.UTC) + leased := &LeasedMessage{ + ID: "enq-1", + MailboxID: "actor-enq", + LeaseToken: "token-enq", + Attempts: 1, + MaxAttempts: 10, + CreatedAt: createdAt, + } + + delivery := newDelivery[*actorTestMsg, int]( + leased, &actorTestMsg{}, nil, nil, newMockDeliveryStore(), + ) + require.True(t, createdAt.Equal(delivery.EnqueuedAt)) +} diff --git a/db/actordelivery/AGENTS.md b/db/actordelivery/AGENTS.md index 14ca29801..501d6fdd4 100644 --- a/db/actordelivery/AGENTS.md +++ b/db/actordelivery/AGENTS.md @@ -48,6 +48,18 @@ other services can reuse durable actor storage without pulling unrelated tables. fence; the multi-worker pool keeps lease + fenced ack. The by-ID nack increments attempts because the peek does not, preserving dead-lettering on max attempts. +- **Attempt-preserving postpone** — `PostponeMailboxMessage` (lease-fenced: + validates `lease_token`, sets a new `available_at`, and *decrements* + attempts to compensate the increment the leased claim already applied, with + a `CASE WHEN attempts > 0` clamp so a corrupt row can never wrap negative) + and `PostponeMailboxMessageByID` (unfenced by-id counterpart that sets + `available_at` and leaves attempts untouched, because the leaseless peek + never incremented them). Exposed on both `Store` and + `TxActorDeliveryStore` as `PostponeMessage` and `PostponeMessageByID`. + These back `actor.Postpone(delay)`: a behavior that cannot handle a message + *yet* releases it for redelivery with its retry budget byte-identical to + what it was before the delivery, rather than spending an attempt on a + transient not-now condition. - `BatchedActorDeliveryQueries` — Batched transaction wrapper for `ActorDeliveryQueries`. - `MigrationOption` — Functional options for migration configuration @@ -86,6 +98,31 @@ other services can reuse durable actor storage without pulling unrelated tables. metadata from a previous leased claim. The by-ID nack path clears that stale metadata while incrementing attempts, preserving the leaseless p-model: `peek -> empty-token delivery -> by-ID ack/nack`. +- A postpone must leave the retry budget exactly as it found it, which is why + the fenced and by-id variants differ: the fenced query decrements (the lease + pre-incremented) and the by-id query does not (the peek never incremented). + Getting this backwards would either walk a postponing message toward + `max_attempts` or let it drift the counter downward on every release. Both + the claim and the peek queries filter on `attempts < max_attempts`, so a row + that reaches the cap does not merely become dead-letter-eligible, it leaves + the eligible set entirely; only a postpone keeps a waiting message + claim-eligible indefinitely. +- The keyed-lane anti-join in the claim/peek queries (`m2.attempts < + m2.max_attempts`) has a boundary case that only postpone can expose: a + predecessor leased on its **final** attempt already satisfies `attempts == + max_attempts`, so it is invisible to the anti-join while in flight and a + competing worker may claim its same-key successor. Harmless for ack and nack + (the predecessor can then only complete or dead-letter), but a postpone + decrements it back below the cap, so it reprocesses after the successor and + inverts per-key FIFO. No adopter combines keyed lanes, `NumWorkers > 1`, and + postpone today, so the SQL is deliberately unchanged; the prerequisite repair + is a lease-liveness disjunct so a currently-leased predecessor blocks its + successors regardless of attempts. See the CAVEAT comment on + `LeaseNextMailboxMessage` in `queries/mailbox.sql`. +- `mailbox_messages.created_at` is the durable enqueue time and is never + rewritten by a release: nack, postpone, and lease extension all leave it + alone. That is what makes it a trustworthy age reference, surfaced to + behaviors as `actor.DeliveryEnqueuedAt`, for bounding a postpone horizon. ## Deep Docs diff --git a/db/actordelivery/CLAUDE.md b/db/actordelivery/CLAUDE.md index 14ca29801..501d6fdd4 100644 --- a/db/actordelivery/CLAUDE.md +++ b/db/actordelivery/CLAUDE.md @@ -48,6 +48,18 @@ other services can reuse durable actor storage without pulling unrelated tables. fence; the multi-worker pool keeps lease + fenced ack. The by-ID nack increments attempts because the peek does not, preserving dead-lettering on max attempts. +- **Attempt-preserving postpone** — `PostponeMailboxMessage` (lease-fenced: + validates `lease_token`, sets a new `available_at`, and *decrements* + attempts to compensate the increment the leased claim already applied, with + a `CASE WHEN attempts > 0` clamp so a corrupt row can never wrap negative) + and `PostponeMailboxMessageByID` (unfenced by-id counterpart that sets + `available_at` and leaves attempts untouched, because the leaseless peek + never incremented them). Exposed on both `Store` and + `TxActorDeliveryStore` as `PostponeMessage` and `PostponeMessageByID`. + These back `actor.Postpone(delay)`: a behavior that cannot handle a message + *yet* releases it for redelivery with its retry budget byte-identical to + what it was before the delivery, rather than spending an attempt on a + transient not-now condition. - `BatchedActorDeliveryQueries` — Batched transaction wrapper for `ActorDeliveryQueries`. - `MigrationOption` — Functional options for migration configuration @@ -86,6 +98,31 @@ other services can reuse durable actor storage without pulling unrelated tables. metadata from a previous leased claim. The by-ID nack path clears that stale metadata while incrementing attempts, preserving the leaseless p-model: `peek -> empty-token delivery -> by-ID ack/nack`. +- A postpone must leave the retry budget exactly as it found it, which is why + the fenced and by-id variants differ: the fenced query decrements (the lease + pre-incremented) and the by-id query does not (the peek never incremented). + Getting this backwards would either walk a postponing message toward + `max_attempts` or let it drift the counter downward on every release. Both + the claim and the peek queries filter on `attempts < max_attempts`, so a row + that reaches the cap does not merely become dead-letter-eligible, it leaves + the eligible set entirely; only a postpone keeps a waiting message + claim-eligible indefinitely. +- The keyed-lane anti-join in the claim/peek queries (`m2.attempts < + m2.max_attempts`) has a boundary case that only postpone can expose: a + predecessor leased on its **final** attempt already satisfies `attempts == + max_attempts`, so it is invisible to the anti-join while in flight and a + competing worker may claim its same-key successor. Harmless for ack and nack + (the predecessor can then only complete or dead-letter), but a postpone + decrements it back below the cap, so it reprocesses after the successor and + inverts per-key FIFO. No adopter combines keyed lanes, `NumWorkers > 1`, and + postpone today, so the SQL is deliberately unchanged; the prerequisite repair + is a lease-liveness disjunct so a currently-leased predecessor blocks its + successors regardless of attempts. See the CAVEAT comment on + `LeaseNextMailboxMessage` in `queries/mailbox.sql`. +- `mailbox_messages.created_at` is the durable enqueue time and is never + rewritten by a release: nack, postpone, and lease extension all leave it + alone. That is what makes it a trustworthy age reference, surfaced to + behaviors as `actor.DeliveryEnqueuedAt`, for bounding a postpone horizon. ## Deep Docs diff --git a/db/actordelivery/postpone_test.go b/db/actordelivery/postpone_test.go new file mode 100644 index 000000000..fa5fad497 --- /dev/null +++ b/db/actordelivery/postpone_test.go @@ -0,0 +1,141 @@ +package actordelivery + +import ( + "testing" + "time" + + "github.com/lightninglabs/wavelength/baselib/actor" + "github.com/stretchr/testify/require" +) + +// enqueuePostponeTestMsg parks one immediately-available message. +func enqueuePostponeTestMsg(t *testing.T, store *testActorDeliveryStore, + id string) { + + t.Helper() + + err := store.EnqueueMessage(t.Context(), actor.EnqueueParams{ + ID: id, + MailboxID: "actor-pp", + MessageType: "test.Message", + Payload: []byte{1}, + AvailableAt: store.clock.Now().Add(-time.Minute), + MaxAttempts: 3, + }) + require.NoError(t, err) +} + +// TestPostponePreservesAttemptsLeased verifies the fenced postpone: the lease +// pre-increments attempts, the postpone decrements it back, so the message's +// retry budget after redelivery is exactly what it was before the claim. +func TestPostponePreservesAttemptsLeased(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newActorDeliveryStoreForTest(t) + + enqueuePostponeTestMsg(t, store, "msg-pp-1") + + leased, err := store.LeaseNextMessage( + ctx, "actor-pp", "token-pp", 30*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, leased) + require.Equal(t, 1, leased.Attempts) + + // A stale token must not postpone. + rows, err := store.PostponeMessage(ctx, "msg-pp-1", "wrong-token", 0) + require.NoError(t, err) + require.Zero(t, rows) + + rows, err = store.PostponeMessage(ctx, "msg-pp-1", "token-pp", 0) + require.NoError(t, err) + require.EqualValues(t, 1, rows) + + // The message is claimable again with its attempts budget restored: + // the fresh lease's pre-increment lands it back at 1, not 2. + released, err := store.LeaseNextMessage( + ctx, "actor-pp", "token-pp-2", 30*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, released) + require.Equal(t, "msg-pp-1", released.ID) + require.Equal(t, 1, released.Attempts) +} + +// TestPostponeByIDLeavesAttemptsUntouched verifies the leaseless postpone: +// the peek never bumped attempts, so the release leaves them exactly as +// stored. +func TestPostponeByIDLeavesAttemptsUntouched(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newActorDeliveryStoreForTest(t) + + enqueuePostponeTestMsg(t, store, "msg-pp-2") + + peeked, err := store.PeekNextMessage(ctx, "actor-pp") + require.NoError(t, err) + require.NotNil(t, peeked) + require.Zero(t, peeked.Attempts) + require.Empty(t, peeked.LeaseToken) + + rows, err := store.PostponeMessageByID(ctx, "msg-pp-2", time.Minute) + require.NoError(t, err) + require.EqualValues(t, 1, rows) + + // Not yet available: the postpone pushed available_at a minute out. + peeked, err = store.PeekNextMessage(ctx, "actor-pp") + require.NoError(t, err) + require.Nil(t, peeked) + + // After the delay elapses the message is re-peeked with attempts + // still untouched. + store.clock.SetTime(store.clock.Now().Add(2 * time.Minute)) + + peeked, err = store.PeekNextMessage(ctx, "actor-pp") + require.NoError(t, err) + require.NotNil(t, peeked) + require.Equal(t, "msg-pp-2", peeked.ID) + require.Zero(t, peeked.Attempts) +} + +// TestPostponeNeverWrapsAttemptsNegative pins the clamp: a fenced postpone on +// a row whose attempts is already zero (corrupt or hand-edited state) stays +// at zero instead of wrapping negative. +func TestPostponeNeverWrapsAttemptsNegative(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newActorDeliveryStoreForTest(t) + + enqueuePostponeTestMsg(t, store, "msg-pp-3") + + leased, err := store.LeaseNextMessage( + ctx, "actor-pp", "token-a", 30*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, leased) + + // First postpone: 1 -> 0. + rows, err := store.PostponeMessage(ctx, "msg-pp-3", "token-a", 0) + require.NoError(t, err) + require.EqualValues(t, 1, rows) + + // Re-lease (0 -> 1) and postpone twice more through fresh leases; the + // budget oscillates 1 -> 0 and never dips below. + for i := range 2 { + token := generateTestID() + + leased, err = store.LeaseNextMessage( + ctx, "actor-pp", token, 30*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, leased, "iteration %d", i) + require.Equal(t, 1, leased.Attempts) + + rows, err = store.PostponeMessage(ctx, "msg-pp-3", token, 0) + require.NoError(t, err) + require.EqualValues(t, 1, rows) + } +} diff --git a/db/actordelivery/queries/mailbox.sql b/db/actordelivery/queries/mailbox.sql index 785ca3070..3868c3bfc 100644 --- a/db/actordelivery/queries/mailbox.sql +++ b/db/actordelivery/queries/mailbox.sql @@ -62,6 +62,26 @@ ON CONFLICT (id) DO NOTHING; -- exhausted row is already filtered out of the outer candidate set by -- m.attempts < m.max_attempts, so this just brings the anti-join -- predicate into agreement with the eligibility predicate. +-- +-- CAVEAT (postpone + keyed lanes + multiple workers). That agreement has a +-- boundary case that postpone can turn into a FIFO inversion. A predecessor +-- leased on its FINAL attempt (attempts == max_attempts after the claim +-- pre-increment) is invisible to this anti-join while it is still being +-- processed, so a competing worker may claim its same-key successor. That is +-- harmless when the predecessor can only ack or dead-letter, which is true for +-- ack and nack. A postpone breaks it: the predecessor is released with attempts +-- DECREMENTED back below max_attempts, so it becomes claim-eligible again and +-- reprocesses AFTER the successor already ran, inverting per-key order. +-- +-- No adopter combines all three preconditions today (keyed correlation lanes, +-- NumWorkers > 1, and a postponing behavior), so the SQL is deliberately left +-- alone rather than complicated for a hypothetical. Per-key FIFO therefore +-- holds for a postponing consumer only when the actor is single-worker, or +-- when the lane's messages never reach their final attempt. Before wiring a +-- postponing behavior onto a keyed lane with a worker pool, fix this first: +-- the intended repair is to add a lease-liveness disjunct so a predecessor +-- that is currently leased still blocks its successors regardless of its +-- attempts, rather than relying on the attempts predicate alone. UPDATE mailbox_messages SET lease_token = $2, @@ -163,6 +183,37 @@ SET attempts = attempts + 1 WHERE id = $1; +-- name: PostponeMailboxMessage :execrows +-- Release a message for redelivery after a delay WITHOUT burning a delivery +-- attempt: the fenced (leased) claim pre-incremented attempts, so the +-- decrement here restores the retry budget to exactly what it was before +-- this delivery. The CASE clamp guards the invariant rather than trusting +-- it: attempts is always >= 1 under a valid lease, but a clamped decrement +-- can never wrap a corrupt row negative. Validates lease_token to prevent +-- stale postpones. This is the attempt-preserving counterpart to +-- NackMailboxMessage, used when a behavior reports "not now" (a postpone) +-- rather than a failure. +UPDATE mailbox_messages +SET + lease_token = NULL, + lease_until = NULL, + available_at = $3, + attempts = CASE WHEN attempts > 0 THEN attempts - 1 ELSE 0 END +WHERE id = $1 AND lease_token = $2; + +-- name: PostponeMailboxMessageByID :execrows +-- Leaseless single-worker counterpart to PostponeMailboxMessage: releases the +-- message by ID without validating a lease token and leaves attempts +-- UNTOUCHED, because the leaseless peek never incremented it. Stale expired +-- lease metadata is cleared so the persisted row matches the leaseless state +-- machine, mirroring NackMailboxMessageByID. +UPDATE mailbox_messages +SET + lease_token = NULL, + lease_until = NULL, + available_at = $2 +WHERE id = $1; + -- name: NackMailboxMessage :execrows -- Release message for redelivery after retry delay. -- Clears lease and sets new available_at. diff --git a/db/actordelivery/sqlc/AGENTS.md b/db/actordelivery/sqlc/AGENTS.md index 35036803a..b1f78ddbe 100644 --- a/db/actordelivery/sqlc/AGENTS.md +++ b/db/actordelivery/sqlc/AGENTS.md @@ -10,8 +10,19 @@ by `sqlc` from `db/actordelivery/queries/mailbox.sql`; regenerate with ## Key Types - `Queries` / `Querier` — generated query struct and interface (enqueue, - lease, peek, ack/nack, extend, expire, outbox claim/complete/fail, - dedup, FSM checkpoints, dead letters). + lease, peek, ack/nack, postpone, extend, expire, outbox + claim/complete/fail, dedup, FSM checkpoints, dead letters). +- `PostponeMailboxMessage` / `PostponeMailboxMessageByID` — the + attempt-preserving release pair behind `actor.Postpone(delay)`, for a + behavior that cannot handle a message *yet* rather than one that failed. + Both set a new `available_at`; they differ in how they treat `attempts`, + and the asymmetry is deliberate. The fenced variant validates + `lease_token` and *decrements* attempts, compensating the increment the + leased claim already applied, clamped by `CASE WHEN attempts > 0` so a + corrupt row cannot wrap negative. The by-id variant is unfenced and + leaves attempts untouched, because the leaseless peek never incremented + them. Either way the retry budget after the release is byte-identical to + what it was before the delivery. - `MailboxMessage`, `OutboxMessage`, `AskResult`, `FsmCheckpoint`, `DeadLetter`, `ProcessedMessage` — row models for the actor-delivery tables. diff --git a/db/actordelivery/sqlc/CLAUDE.md b/db/actordelivery/sqlc/CLAUDE.md index 35036803a..b1f78ddbe 100644 --- a/db/actordelivery/sqlc/CLAUDE.md +++ b/db/actordelivery/sqlc/CLAUDE.md @@ -10,8 +10,19 @@ by `sqlc` from `db/actordelivery/queries/mailbox.sql`; regenerate with ## Key Types - `Queries` / `Querier` — generated query struct and interface (enqueue, - lease, peek, ack/nack, extend, expire, outbox claim/complete/fail, - dedup, FSM checkpoints, dead letters). + lease, peek, ack/nack, postpone, extend, expire, outbox + claim/complete/fail, dedup, FSM checkpoints, dead letters). +- `PostponeMailboxMessage` / `PostponeMailboxMessageByID` — the + attempt-preserving release pair behind `actor.Postpone(delay)`, for a + behavior that cannot handle a message *yet* rather than one that failed. + Both set a new `available_at`; they differ in how they treat `attempts`, + and the asymmetry is deliberate. The fenced variant validates + `lease_token` and *decrements* attempts, compensating the increment the + leased claim already applied, clamped by `CASE WHEN attempts > 0` so a + corrupt row cannot wrap negative. The by-id variant is unfenced and + leaves attempts untouched, because the leaseless peek never incremented + them. Either way the retry budget after the release is byte-identical to + what it was before the delivery. - `MailboxMessage`, `OutboxMessage`, `AskResult`, `FsmCheckpoint`, `DeadLetter`, `ProcessedMessage` — row models for the actor-delivery tables. diff --git a/db/actordelivery/sqlc/mailbox.sql.go b/db/actordelivery/sqlc/mailbox.sql.go index 08acb7617..d1167a28c 100644 --- a/db/actordelivery/sqlc/mailbox.sql.go +++ b/db/actordelivery/sqlc/mailbox.sql.go @@ -637,6 +637,26 @@ type LeaseNextMailboxMessageParams struct { // exhausted row is already filtered out of the outer candidate set by // m.attempts < m.max_attempts, so this just brings the anti-join // predicate into agreement with the eligibility predicate. +// +// CAVEAT (postpone + keyed lanes + multiple workers). That agreement has a +// boundary case that postpone can turn into a FIFO inversion. A predecessor +// leased on its FINAL attempt (attempts == max_attempts after the claim +// pre-increment) is invisible to this anti-join while it is still being +// processed, so a competing worker may claim its same-key successor. That is +// harmless when the predecessor can only ack or dead-letter, which is true for +// ack and nack. A postpone breaks it: the predecessor is released with attempts +// DECREMENTED back below max_attempts, so it becomes claim-eligible again and +// reprocesses AFTER the successor already ran, inverting per-key order. +// +// No adopter combines all three preconditions today (keyed correlation lanes, +// NumWorkers > 1, and a postponing behavior), so the SQL is deliberately left +// alone rather than complicated for a hypothetical. Per-key FIFO therefore +// holds for a postponing consumer only when the actor is single-worker, or +// when the lane's messages never reach their final attempt. Before wiring a +// postponing behavior onto a keyed lane with a worker pool, fix this first: +// the intended repair is to add a lease-liveness disjunct so a predecessor +// that is currently leased still blocks its successors regardless of its +// attempts, rather than relying on the attempts predicate alone. func (q *Queries) LeaseNextMailboxMessage(ctx context.Context, arg LeaseNextMailboxMessageParams) (MailboxMessage, error) { row := q.db.QueryRowContext(ctx, LeaseNextMailboxMessage, arg.MailboxID, @@ -1078,6 +1098,66 @@ func (q *Queries) PeekNextMailboxMessage(ctx context.Context, arg PeekNextMailbo return i, err } +const PostponeMailboxMessage = `-- name: PostponeMailboxMessage :execrows +UPDATE mailbox_messages +SET + lease_token = NULL, + lease_until = NULL, + available_at = $3, + attempts = CASE WHEN attempts > 0 THEN attempts - 1 ELSE 0 END +WHERE id = $1 AND lease_token = $2 +` + +type PostponeMailboxMessageParams struct { + ID string + LeaseToken sql.NullString + AvailableAt int64 +} + +// Release a message for redelivery after a delay WITHOUT burning a delivery +// attempt: the fenced (leased) claim pre-incremented attempts, so the +// decrement here restores the retry budget to exactly what it was before +// this delivery. The CASE clamp guards the invariant rather than trusting +// it: attempts is always >= 1 under a valid lease, but a clamped decrement +// can never wrap a corrupt row negative. Validates lease_token to prevent +// stale postpones. This is the attempt-preserving counterpart to +// NackMailboxMessage, used when a behavior reports "not now" (a postpone) +// rather than a failure. +func (q *Queries) PostponeMailboxMessage(ctx context.Context, arg PostponeMailboxMessageParams) (int64, error) { + result, err := q.db.ExecContext(ctx, PostponeMailboxMessage, arg.ID, arg.LeaseToken, arg.AvailableAt) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +const PostponeMailboxMessageByID = `-- name: PostponeMailboxMessageByID :execrows +UPDATE mailbox_messages +SET + lease_token = NULL, + lease_until = NULL, + available_at = $2 +WHERE id = $1 +` + +type PostponeMailboxMessageByIDParams struct { + ID string + AvailableAt int64 +} + +// Leaseless single-worker counterpart to PostponeMailboxMessage: releases the +// message by ID without validating a lease token and leaves attempts +// UNTOUCHED, because the leaseless peek never incremented it. Stale expired +// lease metadata is cleared so the persisted row matches the leaseless state +// machine, mirroring NackMailboxMessageByID. +func (q *Queries) PostponeMailboxMessageByID(ctx context.Context, arg PostponeMailboxMessageByIDParams) (int64, error) { + result, err := q.db.ExecContext(ctx, PostponeMailboxMessageByID, arg.ID, arg.AvailableAt) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + const SaveFSMCheckpoint = `-- name: SaveFSMCheckpoint :exec INSERT INTO fsm_checkpoints (actor_id, state_type, state_data, version, updated_at) diff --git a/db/actordelivery/sqlc/querier.go b/db/actordelivery/sqlc/querier.go index ac53acad3..88a457dab 100644 --- a/db/actordelivery/sqlc/querier.go +++ b/db/actordelivery/sqlc/querier.go @@ -127,6 +127,26 @@ type Querier interface { // exhausted row is already filtered out of the outer candidate set by // m.attempts < m.max_attempts, so this just brings the anti-join // predicate into agreement with the eligibility predicate. + // + // CAVEAT (postpone + keyed lanes + multiple workers). That agreement has a + // boundary case that postpone can turn into a FIFO inversion. A predecessor + // leased on its FINAL attempt (attempts == max_attempts after the claim + // pre-increment) is invisible to this anti-join while it is still being + // processed, so a competing worker may claim its same-key successor. That is + // harmless when the predecessor can only ack or dead-letter, which is true for + // ack and nack. A postpone breaks it: the predecessor is released with attempts + // DECREMENTED back below max_attempts, so it becomes claim-eligible again and + // reprocesses AFTER the successor already ran, inverting per-key order. + // + // No adopter combines all three preconditions today (keyed correlation lanes, + // NumWorkers > 1, and a postponing behavior), so the SQL is deliberately left + // alone rather than complicated for a hypothetical. Per-key FIFO therefore + // holds for a postponing consumer only when the actor is single-worker, or + // when the lane's messages never reach their final attempt. Before wiring a + // postponing behavior onto a keyed lane with a worker pool, fix this first: + // the intended repair is to add a lease-liveness disjunct so a predecessor + // that is currently leased still blocks its successors regardless of its + // attempts, rather than relying on the attempts predicate alone. LeaseNextMailboxMessage(ctx context.Context, arg LeaseNextMailboxMessageParams) (MailboxMessage, error) // List dead letters for a specific actor. ListDeadLettersByActor(ctx context.Context, arg ListDeadLettersByActorParams) ([]DeadLetter, error) @@ -188,6 +208,22 @@ type Querier interface { // failure (nack) path via NackMailboxMessageByID so a repeatedly-failing // message still climbs to max_attempts and dead-letters. PeekNextMailboxMessage(ctx context.Context, arg PeekNextMailboxMessageParams) (MailboxMessage, error) + // Release a message for redelivery after a delay WITHOUT burning a delivery + // attempt: the fenced (leased) claim pre-incremented attempts, so the + // decrement here restores the retry budget to exactly what it was before + // this delivery. The CASE clamp guards the invariant rather than trusting + // it: attempts is always >= 1 under a valid lease, but a clamped decrement + // can never wrap a corrupt row negative. Validates lease_token to prevent + // stale postpones. This is the attempt-preserving counterpart to + // NackMailboxMessage, used when a behavior reports "not now" (a postpone) + // rather than a failure. + PostponeMailboxMessage(ctx context.Context, arg PostponeMailboxMessageParams) (int64, error) + // Leaseless single-worker counterpart to PostponeMailboxMessage: releases the + // message by ID without validating a lease token and leaves attempts + // UNTOUCHED, because the leaseless peek never incremented it. Stale expired + // lease metadata is cleared so the persisted row matches the leaseless state + // machine, mirroring NackMailboxMessageByID. + PostponeMailboxMessageByID(ctx context.Context, arg PostponeMailboxMessageByIDParams) (int64, error) // ============================================================================= // FSM Checkpoints // ============================================================================= diff --git a/db/actordelivery/store_impl.go b/db/actordelivery/store_impl.go index dc9186b9f..61c363e6f 100644 --- a/db/actordelivery/store_impl.go +++ b/db/actordelivery/store_impl.go @@ -16,18 +16,21 @@ import ( // Type aliases for SQLC-generated types to reduce import noise. type ( - MailboxMsgRow = adsqlc.MailboxMessage - OutboxMsgRow = adsqlc.OutboxMessage - AskResultRow = adsqlc.AskResult - FsmCheckpointRow = adsqlc.FsmCheckpoint - DeadLetterRow = adsqlc.DeadLetter - EnqueueMailboxParams = adsqlc.EnqueueMailboxMessageParams - EnqueueOutboxParams = adsqlc.EnqueueOutboxMessageParams - LeaseMailboxParams = adsqlc.LeaseNextMailboxMessageParams - PeekMailboxParams = adsqlc.PeekNextMailboxMessageParams - AckMailboxParams = adsqlc.AckMailboxMessageParams - NackMailboxParams = adsqlc.NackMailboxMessageParams - NackMailboxByIDParams = adsqlc.NackMailboxMessageByIDParams + MailboxMsgRow = adsqlc.MailboxMessage + OutboxMsgRow = adsqlc.OutboxMessage + AskResultRow = adsqlc.AskResult + FsmCheckpointRow = adsqlc.FsmCheckpoint + DeadLetterRow = adsqlc.DeadLetter + EnqueueMailboxParams = adsqlc.EnqueueMailboxMessageParams + EnqueueOutboxParams = adsqlc.EnqueueOutboxMessageParams + LeaseMailboxParams = adsqlc.LeaseNextMailboxMessageParams + PeekMailboxParams = adsqlc.PeekNextMailboxMessageParams + AckMailboxParams = adsqlc.AckMailboxMessageParams + NackMailboxParams = adsqlc.NackMailboxMessageParams + NackMailboxByIDParams = adsqlc.NackMailboxMessageByIDParams + PostponeMailboxParams = adsqlc.PostponeMailboxMessageParams + PostponeMailboxByIDParams = adsqlc. + PostponeMailboxMessageByIDParams ExtendMailboxParams = adsqlc.ExtendMailboxLeaseParams InsertAskResultParams = adsqlc.InsertAskResultParams ClaimOutboxBatchParams = adsqlc.ClaimOutboxBatchParams @@ -69,6 +72,12 @@ type ActorDeliveryQueries interface { NackMailboxMessageByID(ctx context.Context, arg NackMailboxByIDParams) (int64, error) + PostponeMailboxMessage(ctx context.Context, + arg PostponeMailboxParams) (int64, error) + + PostponeMailboxMessageByID(ctx context.Context, + arg PostponeMailboxByIDParams) (int64, error) + ExtendMailboxLease(ctx context.Context, arg ExtendMailboxParams) (int64, error) @@ -464,6 +473,73 @@ func (s *Store) NackMessageByID(ctx context.Context, id string, return rows, err } +// PostponeMessage releases a message for redelivery after the given delay +// WITHOUT burning a delivery attempt: the query decrements attempts to +// compensate the increment the lease took at claim. Validates the lease +// token to prevent stale postpones. +func (s *Store) PostponeMessage(ctx context.Context, id, leaseToken string, + retryAfter time.Duration) (int64, error) { + + writeTxOpts := db.WriteTxOption() + + var rows int64 + + err := s.db.ExecTx( + ctx, + writeTxOpts, + func(q ActorDeliveryQueries) error { + availableAt := s.clock.Now().Add(retryAfter) + + var err error + rows, err = q.PostponeMailboxMessage( + ctx, + PostponeMailboxParams{ + ID: id, + LeaseToken: toNullString(leaseToken), + AvailableAt: availableAt.Unix(), + }, + ) + + return err + }, + ) + + return rows, err +} + +// PostponeMessageByID releases a message for redelivery by ID without +// validating a lease token and without touching attempts, because the +// leaseless peek never incremented it. It is the leaseless single-worker +// counterpart to PostponeMessage. +func (s *Store) PostponeMessageByID(ctx context.Context, id string, + retryAfter time.Duration) (int64, error) { + + writeTxOpts := db.WriteTxOption() + + var rows int64 + + err := s.db.ExecTx( + ctx, + writeTxOpts, + func(q ActorDeliveryQueries) error { + availableAt := s.clock.Now().Add(retryAfter) + + var err error + rows, err = q.PostponeMailboxMessageByID( + ctx, + PostponeMailboxByIDParams{ + ID: id, + AvailableAt: availableAt.Unix(), + }, + ) + + return err + }, + ) + + return rows, err +} + // ExtendLease extends the lease for long-running message processing. func (s *Store) ExtendLease(ctx context.Context, id, leaseToken string, extension time.Duration) (int64, error) { @@ -1294,6 +1370,39 @@ func (s *TxActorDeliveryStore) NackMessageByID(ctx context.Context, id string, }) } +// PostponeMessage releases a message for redelivery WITHOUT burning a +// delivery attempt, within the current transaction. It is the +// attempt-preserving counterpart to NackMessage. +func (s *TxActorDeliveryStore) PostponeMessage(ctx context.Context, id, + leaseToken string, retryAfter time.Duration) (int64, error) { + + availableAt := s.clock.Now().Add(retryAfter) + + return s.querier.PostponeMailboxMessage(ctx, PostponeMailboxParams{ + ID: id, + LeaseToken: toNullString(leaseToken), + AvailableAt: availableAt.Unix(), + }) +} + +// PostponeMessageByID releases a message for redelivery by ID without +// validating a lease token and without touching attempts, within the current +// transaction. It is the leaseless single-worker counterpart to +// PostponeMessage. +func (s *TxActorDeliveryStore) PostponeMessageByID(ctx context.Context, + id string, retryAfter time.Duration) (int64, error) { + + availableAt := s.clock.Now().Add(retryAfter) + + return s.querier.PostponeMailboxMessageByID( + ctx, + PostponeMailboxByIDParams{ + ID: id, + AvailableAt: availableAt.Unix(), + }, + ) +} + // ExtendLease extends the lease for long-running message processing. func (s *TxActorDeliveryStore) ExtendLease(ctx context.Context, id, leaseToken string, extension time.Duration) (int64, error) { diff --git a/docs/durable_actor_architecture.md b/docs/durable_actor_architecture.md index 811aa4e26..d1cd23f16 100644 --- a/docs/durable_actor_architecture.md +++ b/docs/durable_actor_architecture.md @@ -9,11 +9,12 @@ semantics, recovery mechanisms, and type-erased actor discovery. 1. [Overview](#overview) 2. [OutboxPublisher CDC Pattern](#outboxpublisher-cdc-pattern) 3. [DurableMailbox Message Lifecycle](#durablemailbox-message-lifecycle) -4. [Actor System Architecture](#actor-system-architecture) -5. [Lease-Based Delivery Semantics](#lease-based-delivery-semantics) -6. [Recovery and Restart Flow](#recovery-and-restart-flow) -7. [TypeAssertingRef and MapRef Pattern](#typeassertingref-and-mapref-pattern) -8. [DurableAsk: Crash-Safe Request-Response](#durableask-crash-safe-request-response) +4. [Postpone Semantics](#postpone-semantics) +5. [Actor System Architecture](#actor-system-architecture) +6. [Lease-Based Delivery Semantics](#lease-based-delivery-semantics) +7. [Recovery and Restart Flow](#recovery-and-restart-flow) +8. [TypeAssertingRef and MapRef Pattern](#typeassertingref-and-mapref-pattern) +9. [DurableAsk: Crash-Safe Request-Response](#durableask-crash-safe-request-response) --- @@ -272,6 +273,237 @@ debug delivery issues and design retry strategies. --- +## Postpone Semantics + +A nack is the wrong tool for "not now". Sections 6 and 7 above describe the +nack path: a failed Tell is released with a retry delay, and every release +increments `attempts` until `max_attempts` sends the message to the +`dead_letters` table. That is the right shape for a message that failed. It is +the wrong shape for a message that a consumer merely cannot handle yet. + +Plenty of turns fail for reasons that have nothing to do with the message: a +concurrency cap is full, a peer is still draining, an operator has not caught +up. The condition clears on its own, and the message that happened to arrive +during the window did nothing to deserve a shortened life. Nacking it anyway +walks an innocent message toward the dead-letter table one redelivery at a +time, and it walks it there fastest exactly when the system is busiest, since +that is when the condition holds longest. + +Postpone is the attempt-preserving alternative. A behavior returns +`actor.Postpone(delay)` (or wraps it with context, since detection matches +anywhere in the wrap chain), and the consume path releases the message for +redelivery after the delay **without counting the attempt**: + +```go +if len(r.incoming) >= maxIncoming { + return fmt.Errorf("%w: %w", errAdmissionCapped, + actor.Postpone(5*time.Second)) +} +``` + +### Nack vs Postpone + +The two paths differ in exactly one dimension, the attempt budget, and that +one difference decides whether a message can outlive the condition blocking it. + +| | Nack | Postpone | +|---|---|---| +| `available_at` | pushed out by the retry delay | pushed out by the postpone delay | +| `attempts` | incremented | unchanged | +| Marked processed | no (so it redelivers) | no (so it redelivers) | +| Consults `TellRetryPolicy` | yes | no, it is detected first | +| Log level | warn (a failure) | debug (control flow) | +| Horizon | bounded by `max_attempts` | unbounded | + +`ErrPostponed` is the sentinel `errors.Is` matches, and `*PostponeError` +carries the delay. Pass a real backoff: a zero or negative delay makes the +message claim-eligible immediately, which against a condition that has not +moved is a busy retry loop against the database. + +### Fenced and Leaseless Mechanics + +The store operation mirrors the ack/nack pair, and which one runs depends on +whether the delivery holds a lease token, exactly as `ackMessage` and +`nackMessage` decide: + +- **Fenced** (`PostponeMessage`, non-empty lease token). The leased claim + pre-incremented `attempts`, so the postpone query decrements it to + compensate, restoring the retry budget to what it was before this delivery. + The decrement is clamped (`CASE WHEN attempts > 0`), so a corrupt row can + never wrap negative. +- **Leaseless** (`PostponeMessageByID`, empty lease token). The single-worker + peek takes no lease and never incremented `attempts`, so the query leaves it + alone. There is no lease to validate, so the release is by ID. + +Either way the message's retry budget after a postpone is byte-identical to +what it was before the delivery. + +That matters more than it looks, because an "always retry" `TellRetryPolicy` is +not a substitute. Such a policy cannot stop `attempts` from climbing, and what +happens when it reaches `max_attempts` depends on which release path the actor +uses: + +- **Non-transaction path** (`handleResult` via `Delivery.Nack`, used by the + classic non-tx path and by the Read/Commit path's `finishNonTx` tail). `Nack` + checks `ShouldDeadLetter` before releasing, so the message dead-letters at + exhaustion no matter what the policy said. The policy does not keep it alive; + it only chooses the delay right up until the boundary. +- **Transaction path** (`handleResultInTx`, a classic behavior on a tx-aware + store). The retry branch calls the store's nack directly with no + `ShouldDeadLetter` arm, so a policy that always answers "retry" nacks the row + past `max_attempts` and it simply goes dark: both the claim and the peek + queries filter on `attempts < max_attempts`, so the row leaves the eligible + set without ever reaching `dead_letters`. That asymmetry is a pre-existing + bug in the tx path, not something postpone introduces. + +Either way, a policy override is the wrong tool for "wait indefinitely": +depending on the path it either dead-letters anyway or strands the row +invisibly. Postpone is the only release that keeps a waiting message +claim-eligible without spending the budget. + +### Tell-Only Semantics + +Only Tell deliveries honor a postpone. An Ask has a caller parked on the +promise, so postponing it would strand that caller for the length of the delay +with nothing to observe and no way to learn why. An Ask behavior that returns a +`PostponeError` therefore gets ordinary error treatment: the promise completes +with the error, and the caller decides whether to re-issue. Behaviors that +serve both a routed Tell and an RPC-driven Ask over the same condition should +postpone on the Tell path only and hand the Ask a plain error. + +### The Livelock Tradeoff + +This is deliberate and it is the whole cost of the feature: **a postponed +message never climbs toward `max_attempts`, so nothing dead-letters it +automatically.** The framework has removed the only mechanism that would +eventually give up on it. A behavior that postpones against a condition that +never clears will postpone forever, and the message will sit in the mailbox +until an operator removes it. + +The framework cannot bound this for you, because only the behavior knows when +waiting stops making sense. A capacity cap that clears in seconds and an +operator response that may never come deserve different horizons, and a +generic attempt counter cannot tell them apart. **A behavior that postpones +must bound its own horizon**: track how long it has been waiting and return a +real error once the wait is no longer justified, so the normal nack path can +dead-letter it. If a behavior genuinely wants to wait forever, that is a +legitimate choice, but it must be a choice, not an oversight. + +The framework supplies the reference for that bound: + +```go +enqueuedAt, ok := actor.DeliveryEnqueuedAt(ctx) +if ok && time.Since(enqueuedAt) >= myHorizon { + return errCondition // Ordinary nack path, dead-letters. +} + +return fmt.Errorf("%w: %w", errCondition, actor.Postpone(backoff)) +``` + +`DeliveryEnqueuedAt` reports when the message was first persisted, taken from +the durable row's `created_at`. Neither a nack nor a postpone rewrites that +column, so it survives every redelivery and measures the true age of the wait. + +Prefer it over behavior-side state whenever the message stream is +attacker-controlled. A map keyed on anything the sender chooses (a session id, +a request id) is unbounded by construction, so the bookkeeping meant to protect +the actor becomes the resource the attacker grows. The row's own age costs +nothing to consult and cannot be inflated by fabricating new messages. Note +that the second return distinguishes "no timestamp available" from "age zero": +treat absence as no horizon information rather than as an infinitely old +message. + +### Correlation-Key Interaction + +A postpone does not exempt a message from per-correlation-key FIFO. The claim +path makes a keyed message eligible only when no earlier same-key message +exists in the mailbox, and a postponed message is still very much in the +mailbox. **A postponed head still blocks its key lane** for the length of the +delay, and every subsequent postpone extends the block. + +Head-of-line blocking is bounded to the key, not the mailbox, so unkeyed +traffic and other lanes drain normally. Still, a behavior that postpones a +keyed message is choosing to stall an ordered lane, and a long backoff on a +busy key is a throughput decision, not just a retry decision. Unkeyed messages +have no such interaction: the postponed row simply loses its place in the +global `available_at` order until it becomes eligible again. + +**Scope of the FIFO guarantee.** For a postponing consumer, per-key ordering +holds when **either** the actor is single-worker (`NumWorkers <= 1`) **or** the +lane's messages never reach their final attempt. Both hold for every adopter +today, so this is a constraint to respect rather than a bug to work around. + +The gap is a boundary case in the claim query's anti-join, which passes over a +predecessor that has exhausted its retry budget (`m2.attempts < +m2.max_attempts`, so a dead row cannot wedge its lane forever). A predecessor +leased on its *final* attempt already satisfies `attempts == max_attempts`, so +it is invisible to the anti-join while it is still being processed, and a +competing worker may claim its same-key successor. With only ack and nack that +is harmless, since the predecessor can then only complete or dead-letter. A +postpone breaks it: the release decrements `attempts` back below the cap, so +the predecessor becomes eligible again and reprocesses *after* the successor +already ran, inverting per-key order. + +No adopter combines all three preconditions (keyed lanes, `NumWorkers > 1`, and +a postponing behavior), so the SQL is deliberately left alone rather than +complicated for a hypothetical. Future work, required before wiring a +postponing behavior onto a keyed lane with a worker pool: add a lease-liveness +disjunct to the anti-join so a currently-leased predecessor blocks its +successors regardless of its attempts, instead of relying on the attempts +predicate alone. + +### Adopter: OOR Over-Cap Admission + +The `oor` registry bounds how many incoming receive sessions one daemon keeps +resident via `ReceiveLimits.MaxConcurrentIncomingSessions`, enforced at the +`ensureChild` choke point that every resident-making path funnels through. The +cap is a real defense: without it, an operator streaming unanswered hints over +an owned receive script could pin unbounded children, mailboxes, and rows. + +But the hint that arrives when the daemon is full is ordinary traffic, and the +cap clears as soon as an earlier session terminates and is reaped. Failing that +hint's turn nacked its durable message and spent one of its attempts on a +condition it did not cause, and a daemon that stayed full long enough would +dead-letter a perfectly valid incoming transfer. + +The registry now postpones instead, on the Tell-driven routed paths only: +`handleResolveIncoming` (the hint the event router pushes), `handleDriveEvent`'s +lazy restore of an already-admitted session, and `handleResumeSession` (the +retry callback's timer expiry). All three wrap the postpone alongside the +existing `errIncomingAdmissionCapped` sentinel with a double `%w`, so boot +restore's skip check keeps matching the sentinel while the consume path sees the +postpone. `StartTransferRequest` is untouched: it arrives as an Ask from the RPC +layer, and it is outgoing, so the incoming cap never applies to it. Boot restore +keeps its own treatment as well, skipping over-cap rows rather than aborting the +boot. + +The registry bounds the wait, as every postponing behavior must. +`overCapPostponeHorizon` (10 minutes) is measured against +`actor.DeliveryEnqueuedAt`, and past it the plain sentinel is returned so the +ordinary nack path dead-letters the hint into a table where it is visible and +requeue-able. The bound is load-bearing rather than decorative here, because the +hint stream is operator-controlled: every redelivery of a capped hint re-runs +the wallet-ownership query and the self-transfer row read before the cap +rejects it again, so an unbounded postpone would let an operator streaming +fabricated session ids build a permanent churn queue against the single-worker +registry. Deriving the age from the row rather than from a per-session map is +the same point in miniature: the map would be keyed on ids the operator picks. + +The deferred self-transfer hint in the same registry is the second adopter. It +previously carried a custom `TellRetryPolicy` that answered "retry in 30 +seconds, always", intending that the hint never dead-letter while its outgoing +session ran. That intent was never actually achieved. The registry runs the +Read/Commit path, whose `finishNonTx` tail releases through `Delivery.Nack`, +and `Nack` checks `ShouldDeadLetter` before the release: the tenth deferral +dead-lettered the hint regardless of what the policy answered. (Had the +registry been a classic behavior on the tx path, the same policy would instead +have stranded the row invisibly, per the asymmetry described above. Neither +outcome is the one the policy was written for.) Postponing on the same 30 +second backoff gives the intended semantics for real, and the registry went +back to the default Tell retry policy for everything else. + +--- + ## Actor System Architecture The durable actor system consists of several interconnected components organized diff --git a/oor/AGENTS.md b/oor/AGENTS.md index bb983c148..b02ade28a 100644 --- a/oor/AGENTS.md +++ b/oor/AGENTS.md @@ -83,6 +83,41 @@ For field-level detail, use `go doc github.com/lightninglabs/wavelength/oor. 0 { + return r.capBackoff + } + + return incomingCapBackoff +} + +// capPostponeHorizon returns the configured over-cap postpone horizon, falling +// back to the package default when unset. +func (r *oorRegistryBehavior) capPostponeHorizon() time.Duration { + if r.capHorizon > 0 { + return r.capHorizon + } + + return overCapPostponeHorizon +} + +// postponeOverCap converts an over-cap admission rejection into a postpone +// request, leaving every other error untouched. Being over the concurrency cap +// is a transient not-now condition that clears when a resident session +// terminates and frees a slot, so nacking the routed message that triggered it +// would walk an innocent hint toward the dead-letter table one redelivery at a +// time. A postpone releases the delivery on a backoff with its attempt budget +// fully intact instead. +// +// The postpone is bounded. Because a postponed message never climbs toward +// max_attempts, the framework has no give-up mechanism left, so this is the +// behavior discharging its own obligation to bound the wait: past +// capPostponeHorizon, measured from the durable row's own enqueue time, the +// plain capped error is returned instead and the ordinary nack path +// dead-letters the message. Without that bound an operator streaming +// fabricated session ids would build a churn queue on the single-worker +// registry that never drains, since every redelivery re-runs the ownership +// query and the self-transfer lookup before the cap rejects it again. +// +// The age reference is the delivery's enqueue timestamp, not behavior-side +// state, precisely because the session ids in that stream are operator-chosen: +// any map keyed on them would be the unbounded resource this is defending. A +// delivery with no timestamp (a directly-invoked behavior in a test, or a +// store that does not report one) postpones as before, since absence means "no +// horizon information", not "infinitely old". +// +// Callers must restrict this to the Tell-driven routed-message path. An +// Ask-driven admission has a caller parked on the promise, and the framework +// gives a postponed Ask ordinary error treatment, so the caller is better +// served by the bare sentinel. The double %w keeps errIncomingAdmissionCapped +// matchable (boot restore's skip check and the RPC surface both rely on it) +// while adding the postpone to the same wrap chain. +func (r *oorRegistryBehavior) postponeOverCap(ctx context.Context, + err error) error { + + if !errors.Is(err, errIncomingAdmissionCapped) { + return err + } + + horizon := r.capPostponeHorizon() + enqueuedAt, ok := actor.DeliveryEnqueuedAt(ctx) + if ok && r.registryNow().Sub(enqueuedAt) >= horizon { + r.logger(ctx).WarnS(ctx, "Over-cap incoming OOR hint exceeded "+ + "its postpone horizon; failing it to the dead-letter "+ + "path", err, + btclog.Fmt("enqueued_at", "%s", enqueuedAt), + btclog.Fmt("horizon", "%s", horizon), + ) + + return err + } + + return fmt.Errorf("%w: %w", err, actor.Postpone(r.capPostponeBackoff())) +} + // admissionHandoff records an outgoing admission whose caller promise must be // detached onto the child's future after the registry's consuming Commit // succeeds. @@ -929,7 +1042,13 @@ func (r *oorRegistryBehavior) handleDriveEvent(ctx context.Context, child, err := r.lookupOrRestore(ctx, req.SessionID) if err != nil { - return fn.Err[ActorResp](err) + + // A routed event for an already-admitted incoming session that + // cannot be made resident right now is over cap, not broken: + // postpone it so the event waits for a slot with its attempt + // budget intact rather than dead-lettering an event whose + // session row is still very much alive. + return fn.Err[ActorResp](r.postponeOverCap(ctx, err)) } if child == nil { // lookupOrRestore returns a nil child for both a truly-unknown @@ -991,6 +1110,18 @@ func (r *oorRegistryBehavior) handleResolveIncoming(ctx context.Context, // backoff as the crash-safety fallback. if errors.Is(err, errSelfTransferDeferred) { r.parkSelfHint(req) + + // A deferred hint is waiting on the outgoing session, + // not failing, so postpone rather than nack: the + // durable copy keeps its attempt budget and stays + // claim-eligible for as long as the wait lasts. + return fn.Err[ActorResp]( + fmt.Errorf( + "%w: %w", err, actor.Postpone( + selfHintRedeliveryBackoff, + ), + ), + ) } return fn.Err[ActorResp](err) @@ -1009,9 +1140,11 @@ func (r *oorRegistryBehavior) handleResolveIncoming(ctx context.Context, // unanswered hints (each a distinct fabricated session id over an owned // receive script) cannot pin unbounded goroutines, mailboxes, and rows. // A resident session forwarding a follow-up hint is exempt: it already - // counts against the cap. The hint is retried by transport, so an - // over-cap rejection is recoverable once earlier sessions terminate and - // are reaped. + // counts against the cap. The rejection is a postpone rather than a + // plain error, so the routed hint redelivers on a backoff with its + // attempt budget intact and admits as soon as an earlier session + // terminates and is reaped, instead of dead-lettering for a condition + // it did nothing to cause. if !existed { maxIncoming := r.cfg.Limits.MaxConcurrentIncomingSessions if _, counted := r.incoming[req.SessionID]; !counted && @@ -1032,7 +1165,11 @@ func (r *oorRegistryBehavior) handleResolveIncoming(ctx context.Context, slog.Uint64("cap", uint64(maxIncoming)), ) - return fn.Err[ActorResp](errIncomingAdmissionCapped) + return fn.Err[ActorResp]( + r.postponeOverCap( + ctx, errIncomingAdmissionCapped, + ), + ) } } @@ -1040,7 +1177,11 @@ func (r *oorRegistryBehavior) handleResolveIncoming(ctx context.Context, req.SessionID, clientdb.OORSessionDirectionIncoming, ) if err != nil { - return fn.Err[ActorResp](err) + + // ensureChild is the choke point behind the pre-check above, so + // it can still reject at the cap on a path the pre-check + // exempted; give that rejection the same postpone treatment. + return fn.Err[ActorResp](r.postponeOverCap(ctx, err)) } if !existed { @@ -1096,9 +1237,10 @@ func (r *oorRegistryBehavior) resolveSelfTransfer(ctx context.Context, return nil } - // Deferring is an error so the durable delivery is retained (nacked on - // the long self-transfer backoff); the caller parks the hint for the - // event-driven redrive at the outgoing session's terminal reap. + // Deferring is an error so the durable delivery is retained; the caller + // turns it into a postpone on the long self-transfer backoff and parks + // the hint for the event-driven redrive at the outgoing session's + // terminal reap. if !record.Status.IsTerminal() { r.logger(ctx).DebugS(ctx, "Deferring incoming self-transfer hint until "+ @@ -1292,7 +1434,13 @@ func (r *oorRegistryBehavior) handleResumeSession(ctx context.Context, child, err := r.lookupOrRestore(ctx, req.SessionID) if err != nil { - return fn.Err[ActorResp](err) + + // A resume arrives as a Tell from the retry callback, so it + // gets the same over-cap treatment as a routed drive-event: a + // session that cannot be made resident right now is over cap, + // not broken, and nacking its resume would dead-letter a real + // delivery for a condition that clears on its own. + return fn.Err[ActorResp](r.postponeOverCap(ctx, err)) } if child == nil { r.logger(ctx).DebugS( diff --git a/oor/registry_test.go b/oor/registry_test.go index 0e5f755ea..6d92c0265 100644 --- a/oor/registry_test.go +++ b/oor/registry_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -1147,6 +1148,14 @@ func TestOORRegistrySelfTransferParkAndRedrive(t *testing.T) { require.Len(t, b.parkedSelfHints, 1) require.Empty(t, rec.recorded()) + // The defer is a postpone on the long self-transfer backoff, not a + // nack: the durable copy is the crash-safety net for a wait with no + // bound, so it must keep its attempt budget for as long as the + // outgoing session runs. + var deferPostpone *actor.PostponeError + require.ErrorAs(t, res.Err(), &deferPostpone) + require.Equal(t, selfHintRedeliveryBackoff, deferPostpone.Delay) + // A terminal notification before the row flips is a no-op for the // park: admission re-checks the row on the redriven copy, so the // premature redrive simply re-parks. Flip the row terminal first to @@ -2173,3 +2182,304 @@ func TestNewOORRegistryActorValidatesRequiredDeps(t *testing.T) { }) } } + +// countingRecipientFilter owns every recipient it is shown and counts how many +// times the registry consulted it. Admission validation runs once per delivery +// of an incoming hint, before the concurrency cap is consulted, so the count is +// a delivery counter for a hint that never gets past the cap. +type countingRecipientFilter struct { + calls atomic.Int64 +} + +// Handle satisfies OutboxHandler; admission validation never invokes it. +func (f *countingRecipientFilter) Handle(context.Context, SessionID, + OutboxEvent) ([]Event, error) { + + return nil, nil +} + +// FilterIncomingMetadataRecipients records the call and owns every recipient. +func (f *countingRecipientFilter) FilterIncomingMetadataRecipients( + _ context.Context, recipients []ArkRecipientOutput) ( + []ArkRecipientOutput, error) { + + f.calls.Add(1) + + return recipients, nil +} + +// TestOORRegistryOverCapAdmissionPostpones verifies an over-cap incoming +// admission asks the durable consume path to postpone rather than fail. The +// rejection must still match errIncomingAdmissionCapped (boot restore's skip +// check keys off that sentinel), must carry the configured backoff, and must +// clear the moment a slot frees. +func TestOORRegistryOverCapAdmissionPostpones(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + b, rec := newTestRegistryBehavior(newFakeRegistryStore()) + b.cfg.IncomingHandler = &fakeRecipientFilter{owned: true} + b.cfg.Limits.MaxConcurrentIncomingSessions = 1 + b.capBackoff = 3 * time.Second + + ownedScript := []byte{0x51, 0x20, 0xaa, 0xbb} + admit := func(seed byte) fn.Result[ActorResp] { + return b.Receive(ctx, &ResolveIncomingTransferRequest{ + SessionID: oorSessionID(seed), + RecipientPkScript: ownedScript, + }, fakeExec{}) + } + + // The first hint takes the only incoming slot. + require.True(t, admit(0x01).IsOk()) + require.Equal(t, 1, rec.spawns) + + // The second is over cap: a postpone request, not a plain failure. + res := admit(0x02) + require.True(t, res.IsErr()) + require.Equal(t, 1, rec.spawns) + + err := res.Err() + require.ErrorIs(t, err, errIncomingAdmissionCapped) + require.ErrorIs(t, err, actor.ErrPostponed) + + var postpone *actor.PostponeError + require.ErrorAs(t, err, &postpone) + require.Equal(t, 3*time.Second, postpone.Delay) + + // Reaping the resident session frees the slot, so the same hint admits + // on its next redelivery. + resident := oorSessionID(0x01) + b.dropChild(resident, b.active[resident]) + + require.True(t, admit(0x02).IsOk()) + require.Equal(t, 2, rec.spawns) +} + +// TestOORRegistryDefaultCapBackoff verifies the over-cap postpone falls back to +// the package default when a behavior leaves the backoff unset, so production +// never postpones with a zero delay (which would busy-loop the registry against +// a cap that has not moved). +func TestOORRegistryDefaultCapBackoff(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + b, _ := newTestRegistryBehavior(newFakeRegistryStore()) + require.Equal(t, incomingCapBackoff, b.capPostponeBackoff()) + require.Positive(t, b.capPostponeBackoff()) + require.Equal(t, overCapPostponeHorizon, b.capPostponeHorizon()) + require.Positive(t, b.capPostponeHorizon()) + + // A non-cap error passes through untouched: only the cap is a not-now + // condition. + other := errors.New("boom") + require.Equal(t, other, b.postponeOverCap(ctx, other)) + require.NotErrorIs(t, b.postponeOverCap(ctx, other), actor.ErrPostponed) +} + +// TestOORRegistryOverCapPostponeHorizon verifies the registry bounds its own +// postpone horizon, which postpone requires of every adopter: because a +// postponed message never climbs toward max_attempts, nothing else will ever +// give up on it. A capped hint younger than the horizon postpones, one that has +// been waiting past it fails with the plain sentinel so the ordinary nack path +// dead-letters it. Without this bound an operator streaming fabricated session +// ids would pin a churn queue on the single-worker registry forever. +func TestOORRegistryOverCapPostponeHorizon(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 8, 7, 12, 0, 0, 0, time.UTC) + + const horizon = 10 * time.Minute + + b, rec := newTestRegistryBehavior(newFakeRegistryStore()) + b.cfg.IncomingHandler = &fakeRecipientFilter{owned: true} + b.cfg.Limits.MaxConcurrentIncomingSessions = 1 + b.cfg.Clock = clock.NewTestClock(now) + b.capHorizon = horizon + + ownedScript := []byte{0x51, 0x20, 0xaa, 0xbb} + admit := func(ctx context.Context, seed byte) fn.Result[ActorResp] { + return b.Receive(ctx, &ResolveIncomingTransferRequest{ + SessionID: oorSessionID(seed), + RecipientPkScript: ownedScript, + }, fakeExec{}) + } + + // Fill the only incoming slot so every later admission is over cap. + require.True(t, admit(t.Context(), 0x01).IsOk()) + require.Equal(t, 1, rec.spawns) + + // A hint enqueued one minute ago is still inside the horizon, so it + // postpones and keeps its attempt budget. + fresh := actor.WithDeliveryEnqueuedAtForTest( + t.Context(), now.Add(-time.Minute), + ) + res := admit(fresh, 0x02) + require.True(t, res.IsErr()) + require.ErrorIs(t, res.Err(), errIncomingAdmissionCapped) + require.ErrorIs(t, res.Err(), actor.ErrPostponed) + + // Exactly at the horizon the postpone stops: the hint has had its wait + // and now fails for real so the nack path can dead-letter it. + atHorizon := actor.WithDeliveryEnqueuedAtForTest( + t.Context(), now.Add(-horizon), + ) + res = admit(atHorizon, 0x03) + require.True(t, res.IsErr()) + require.ErrorIs(t, res.Err(), errIncomingAdmissionCapped) + require.NotErrorIs(t, res.Err(), actor.ErrPostponed) + + // Well past the horizon behaves the same way. + stale := actor.WithDeliveryEnqueuedAtForTest( + t.Context(), now.Add(-2*horizon), + ) + res = admit(stale, 0x04) + require.True(t, res.IsErr()) + require.NotErrorIs(t, res.Err(), actor.ErrPostponed) + + // A delivery with no enqueue timestamp carries no horizon information, + // so it postpones rather than being treated as infinitely old. + res = admit(t.Context(), 0x05) + require.True(t, res.IsErr()) + require.ErrorIs(t, res.Err(), actor.ErrPostponed) + + // None of the rejected hints spawned a child. + require.Equal(t, 1, rec.spawns) +} + +// TestOORRegistryOverCapHintKeepsAttempts drives an over-cap incoming hint +// through the registry's real durable mailbox and verifies the postpone +// contract end to end: the hint is redelivered repeatedly with its attempt +// budget untouched, nothing is dead-lettered, and it admits once the resident +// session terminates and frees a slot. A nack-based rejection would have +// counted every one of those redeliveries toward max_attempts instead. +func TestOORRegistryOverCapHintKeepsAttempts(t *testing.T) { + t.Parallel() + + ctx := t.Context() + + system := actor.NewActorSystem() + defer func() { + require.NoError(t, system.Shutdown(context.Background())) + }() + + filter := &countingRecipientFilter{} + store := newFakeRegistryStore() + deliveryStore := newTestDeliveryStore(t) + + registry, err := NewOORRegistryActor(OORRegistryConfig{ + RegistryStore: store, + DeliveryStore: deliveryStore, + ServerConn: fakeServerConnRef{}, + Signer: testSigner(t), + IncomingHandler: filter, + PackageStore: &fakePackageStore{}, + ReservationStore: &countingReservationStore{}, + ActorSystem: system, + Limits: ReceiveLimits{ + MaxConcurrentIncomingSessions: 1, + }, + }) + require.NoError(t, err) + defer registry.Stop() + + // Shrink the over-cap backoff so a capped hint becomes claim-eligible + // again well inside the mailbox's idle poll interval, which is what + // paces the redeliveries this test counts. + registry.behavior.capBackoff = 20 * time.Millisecond + + ownedScript := []byte{0x51, 0x20, 0xaa, 0xbb} + tellHint := func(sessionID SessionID) { + require.NoError( + t, + registry.Ref().Tell( + ctx, &ResolveIncomingTransferRequest{ + SessionID: sessionID, + RecipientPkScript: ownedScript, + RecipientEventID: 1, + }, + ), + ) + } + + incoming := clientdb.OORSessionDirectionIncoming + admitted := func(sessionID SessionID) bool { + record, gErr := store.GetSession(ctx, chainHashOf(sessionID)) + + return gErr == nil && record.Direction == incoming + } + + // The first hint takes the single incoming slot. + resident := oorSessionID(0x90) + tellHint(resident) + require.Eventually(t, func() bool { + return admitted(resident) + }, 10*time.Second, 20*time.Millisecond) + + // The second hint is over cap. It postpones, so it keeps coming back. + const wantRedeliveries = 3 + baseline := filter.calls.Load() + + capped := oorSessionID(0x91) + tellHint(capped) + + require.Eventually(t, func() bool { + return filter.calls.Load()-baseline >= wantRedeliveries + }, 60*time.Second, 20*time.Millisecond) + + // The hint is still queued, and its attempt budget is exactly what it + // was when it was enqueued. The registry is a single-worker leaseless + // consumer, so a nack would have run it up to wantRedeliveries here. + var queued *actor.LeasedMessage + require.Eventually(t, func() bool { + msg, pErr := deliveryStore.PeekNextMessage( + ctx, OORActorServiceKeyName, + ) + if pErr != nil || msg == nil { + return false + } + + queued = msg + + return true + }, 10*time.Second, 5*time.Millisecond) + require.Zero(t, queued.Attempts) + + // Nothing dead-lettered, and the capped session never became resident. + letters, err := deliveryStore.ListDeadLetters( + ctx, OORActorServiceKeyName, 10, + ) + require.NoError(t, err) + require.Empty(t, letters) + + _, err = store.GetSession(ctx, chainHashOf(capped)) + require.ErrorIs(t, err, clientdb.ErrOORSessionNotFound) + + // Free the slot: mark the resident session terminal and reap its child. + record, err := store.GetSession(ctx, chainHashOf(resident)) + require.NoError(t, err) + + record.Status = clientdb.OORSessionStatusCompleted + require.NoError(t, store.UpsertSession(ctx, *record)) + require.NoError( + t, + registry.Ref().Tell( + ctx, &SessionTerminalNotification{ + SessionID: resident, + }, + ), + ) + + // With a slot free the postponed hint admits on its next redelivery. + require.Eventually(t, func() bool { + return admitted(capped) + }, 60*time.Second, 20*time.Millisecond) + + letters, err = deliveryStore.ListDeadLetters( + ctx, OORActorServiceKeyName, 10, + ) + require.NoError(t, err) + require.Empty(t, letters) +} diff --git a/serverconn/testutil_test.go b/serverconn/testutil_test.go index d2cbe8128..b660885c5 100644 --- a/serverconn/testutil_test.go +++ b/serverconn/testutil_test.go @@ -564,6 +564,62 @@ func (s *memCheckpointStore) NackMessage(ctx context.Context, id, return 1, nil } +// PostponeMessage releases a message without burning an attempt, mirroring +// the fenced attempt-preserving postpone (decrement clamped at zero). +func (s *memCheckpointStore) PostponeMessage(ctx context.Context, id, + leaseToken string, retryAfter time.Duration) (int64, error) { + + _ = ctx + + s.mu.Lock() + defer s.mu.Unlock() + + msg, ok := s.messages[id] + if !ok { + return 0, nil + } + + // The fenced postpone validates the lease token exactly as the fenced + // nack does. Skipping the check here would let a stale consumer's + // postpone decrement the attempts of a row another consumer now owns, + // which is the corruption the fence exists to prevent, and the mock + // would report a zero-row release as success. + if msg.leased.LeaseToken != leaseToken { + return 0, nil + } + + msg.leased.LeaseToken = "" + msg.leased.LeaseUntil = time.Time{} + if msg.leased.Attempts > 0 { + msg.leased.Attempts-- + } + msg.availableAt = time.Now().Add(retryAfter) + + return 1, nil +} + +// PostponeMessageByID releases a message by ID without touching attempts, +// mirroring the unfenced leaseless postpone. +func (s *memCheckpointStore) PostponeMessageByID(ctx context.Context, id string, + retryAfter time.Duration) (int64, error) { + + _ = ctx + + s.mu.Lock() + defer s.mu.Unlock() + + msg, ok := s.messages[id] + if !ok { + return 0, nil + } + + msg.leased.LeaseToken = "" + msg.leased.LeaseUntil = time.Time{} + msg.availableAt = time.Now().Add(retryAfter) + + return 1, nil +} + // NackMessageByID releases a message by ID without lease-token validation and // increments attempts, mirroring the unfenced leaseless nack. func (s *memCheckpointStore) NackMessageByID(ctx context.Context, id string, diff --git a/unroll/actor_test.go b/unroll/actor_test.go index cd143c4d2..d67a7ca16 100644 --- a/unroll/actor_test.go +++ b/unroll/actor_test.go @@ -876,6 +876,20 @@ func (s *memCheckpointStore) NackMessageByID(context.Context, string, return 1, nil } +// PostponeMessage is unused in these tests. +func (s *memCheckpointStore) PostponeMessage(context.Context, string, string, + time.Duration) (int64, error) { + + return 1, nil +} + +// PostponeMessageByID is unused in these tests. +func (s *memCheckpointStore) PostponeMessageByID(context.Context, string, + time.Duration) (int64, error) { + + return 1, nil +} + // ExtendLease is unused in these tests. func (s *memCheckpointStore) ExtendLease(context.Context, string, string, time.Duration) (int64, error) {