diff --git a/baselib/actor/AGENTS.md b/baselib/actor/AGENTS.md index cb4a2129c..2ada580bb 100644 --- a/baselib/actor/AGENTS.md +++ b/baselib/actor/AGENTS.md @@ -58,6 +58,38 @@ crash-safe at-least-once delivery with exactly-once deduplication. false for Tells, DurableAsks, and redelivered asks whose caller is gone. - `ChannelMailbox[M, R]` — In-memory channel-based mailbox (non-durable, for lightweight actors). - `Mailbox[M, R]` — Interface for actor message queues: `Send(ctx, env) error` (blocking; returns `ErrMailboxClosed`, `ErrActorTerminated`, or a context error on failure), `TrySend(env) error` (non-blocking), `Receive(ctx) iter.Seq[envelope]`, `Close()`, `IsClosed() bool`, `Drain() iter.Seq[envelope]`. +- `ErrMailboxSaturated` / backlog watermarks — The durable analogue of + `ErrMailboxFull`. `DurableMailboxConfig.SoftHighWatermark` / + `HardHighWatermark` (flowing through `DurableActorConfig`) bound the + persistent backlog: past the soft mark the mailbox logs one warning per + breach episode, at or past the hard mark `Send` refuses with + `ErrMailboxSaturated` BEFORE encoding or promise registration (so + `TrySend`/`TryTell` inherit the check and a refusal needs no cleanup). + Both default to 0 (disabled); consumer sites opt in with + `DefaultSoftHighWatermark` (1000) / `DefaultHardHighWatermark` (10000). + Two exemptions: priority `>= ControlPriority` (restart plus the + boot-critical restore/resume messages — `RestoreNonTerminalRequest`, + `ResumeUnrollRequest`, `ResumeCreditOpRequest` — whose refusal would be a + fatal-boot crash loop), and outbox-propagated deliveries (detected via + the outbox-ID context marker; the publisher's claim path bumps attempts + in its own tx, so a refusal would dead-letter a committed CDC row rather + than shed load). Depth is read via a 1s TTL-cached, single-flighted probe + (run outside the mutex AND outside the sender's ambient tx via + `WithoutTx` — joining a SERIALIZABLE writer would take whole-partition + predicate locks) plus a local sent-since-probe delta (one-sided for local + traffic; remote sends invisible for up to one window). A probe failure + falls back to the cached estimate, or fails OPEN with no baseline. The + soft episode is evaluated before the hard refusal so the warning fires + even when the first probe already reads saturated. Known residual: an + in-turn Tell into a saturated peer fails the sender's turn and burns the + inbound message's attempts (postpone semantics are the planned fix). +- `MailboxDepthStore` / `MailboxDepthCount` — Narrow, optional read surface + (`MailboxDepth(ctx, mailboxID)`, `MailboxDepths(ctx)`) discovered by type + assertion on the `DeliveryStore`, deliberately NOT embedded in it (test + doubles stay small). Backing reads are `COUNT(*)` of `mailbox_messages` + rows — leased rows included, rows are deleted on ack — so depth is exactly + the undelivered backlog. Consumed by the watermark check and the + `waved_mailbox_depth`/`waved_mailbox_backlog` scrape gauges. - `isExpectedShutdownErr(err) bool` — Internal helper that classifies errors as expected during teardown: context cancellation/deadline, closed DB handle ("sql: database is closed", "sql: connection is already closed", "use of closed network connection"). Used by the lease loop to demote shutdown-path failures to debug instead of warn-flooding test artifacts at itest tail. - `Message.CorrelationKey() string` — Per-message FIFO key consumed by the durable mailbox's claim path. Non-empty keys participate in per-key FIFO: diff --git a/baselib/actor/CLAUDE.md b/baselib/actor/CLAUDE.md index cb4a2129c..2ada580bb 100644 --- a/baselib/actor/CLAUDE.md +++ b/baselib/actor/CLAUDE.md @@ -58,6 +58,38 @@ crash-safe at-least-once delivery with exactly-once deduplication. false for Tells, DurableAsks, and redelivered asks whose caller is gone. - `ChannelMailbox[M, R]` — In-memory channel-based mailbox (non-durable, for lightweight actors). - `Mailbox[M, R]` — Interface for actor message queues: `Send(ctx, env) error` (blocking; returns `ErrMailboxClosed`, `ErrActorTerminated`, or a context error on failure), `TrySend(env) error` (non-blocking), `Receive(ctx) iter.Seq[envelope]`, `Close()`, `IsClosed() bool`, `Drain() iter.Seq[envelope]`. +- `ErrMailboxSaturated` / backlog watermarks — The durable analogue of + `ErrMailboxFull`. `DurableMailboxConfig.SoftHighWatermark` / + `HardHighWatermark` (flowing through `DurableActorConfig`) bound the + persistent backlog: past the soft mark the mailbox logs one warning per + breach episode, at or past the hard mark `Send` refuses with + `ErrMailboxSaturated` BEFORE encoding or promise registration (so + `TrySend`/`TryTell` inherit the check and a refusal needs no cleanup). + Both default to 0 (disabled); consumer sites opt in with + `DefaultSoftHighWatermark` (1000) / `DefaultHardHighWatermark` (10000). + Two exemptions: priority `>= ControlPriority` (restart plus the + boot-critical restore/resume messages — `RestoreNonTerminalRequest`, + `ResumeUnrollRequest`, `ResumeCreditOpRequest` — whose refusal would be a + fatal-boot crash loop), and outbox-propagated deliveries (detected via + the outbox-ID context marker; the publisher's claim path bumps attempts + in its own tx, so a refusal would dead-letter a committed CDC row rather + than shed load). Depth is read via a 1s TTL-cached, single-flighted probe + (run outside the mutex AND outside the sender's ambient tx via + `WithoutTx` — joining a SERIALIZABLE writer would take whole-partition + predicate locks) plus a local sent-since-probe delta (one-sided for local + traffic; remote sends invisible for up to one window). A probe failure + falls back to the cached estimate, or fails OPEN with no baseline. The + soft episode is evaluated before the hard refusal so the warning fires + even when the first probe already reads saturated. Known residual: an + in-turn Tell into a saturated peer fails the sender's turn and burns the + inbound message's attempts (postpone semantics are the planned fix). +- `MailboxDepthStore` / `MailboxDepthCount` — Narrow, optional read surface + (`MailboxDepth(ctx, mailboxID)`, `MailboxDepths(ctx)`) discovered by type + assertion on the `DeliveryStore`, deliberately NOT embedded in it (test + doubles stay small). Backing reads are `COUNT(*)` of `mailbox_messages` + rows — leased rows included, rows are deleted on ack — so depth is exactly + the undelivered backlog. Consumed by the watermark check and the + `waved_mailbox_depth`/`waved_mailbox_backlog` scrape gauges. - `isExpectedShutdownErr(err) bool` — Internal helper that classifies errors as expected during teardown: context cancellation/deadline, closed DB handle ("sql: database is closed", "sql: connection is already closed", "use of closed network connection"). Used by the lease loop to demote shutdown-path failures to debug instead of warn-flooding test artifacts at itest tail. - `Message.CorrelationKey() string` — Per-message FIFO key consumed by the durable mailbox's claim path. Non-empty keys participate in per-key FIFO: diff --git a/baselib/actor/bounded_delivery.go b/baselib/actor/bounded_delivery.go index 0ba64a161..cee85fff4 100644 --- a/baselib/actor/bounded_delivery.go +++ b/baselib/actor/bounded_delivery.go @@ -56,7 +56,10 @@ type NonParkingTeller[M Message] interface { // enqueue into the caller's transaction, and guessing the other way would // silently break that atomicity for a reference implemented elsewhere. // -// Callers must handle ErrMailboxFull. Treating it as a delivery failure and +// Callers must handle ErrMailboxFull, and its durable analogue +// ErrMailboxSaturated when the target's mailbox carries backlog watermarks: +// both mean the target refused for want of room and expects the caller to +// stash, redrive, or shed. Treating either as a delivery failure and // dropping the message converts backpressure into silent message loss, which // for an at-least-once transport is worse than the stall it replaces. func TellWithoutParking[M Message](ctx context.Context, ref TellOnlyRef[M], @@ -84,8 +87,12 @@ func (ref *actorRefImpl[M, R]) TellWithoutParkingTo(ctx context.Context, } // TellWithoutParkingTo keeps the plain Tell: a durable mailbox has no -// capacity, so its enqueue waits on a database write inside the caller's -// transaction instead of on the receiving actor draining its queue. +// in-memory capacity, so its enqueue waits on a database write inside the +// caller's transaction instead of on the receiving actor draining its queue. +// The write itself never parks on the consumer; what it can do is refuse +// with ErrMailboxSaturated when the mailbox carries backlog watermarks and +// the backlog is past the hard one, which the caller handles like +// ErrMailboxFull. func (ref *durableActorRefImpl[M, R]) TellWithoutParkingTo(ctx context.Context, msg M) fn.Result[bool] { diff --git a/baselib/actor/durable_actor.go b/baselib/actor/durable_actor.go index bee82ec62..e49905430 100644 --- a/baselib/actor/durable_actor.go +++ b/baselib/actor/durable_actor.go @@ -115,6 +115,19 @@ type DurableActorConfig[M TLVMessage, R any] struct { // Default: 10. MaxAttempts int + // SoftHighWatermark is the persistent backlog depth at which the + // actor's mailbox starts logging that its consumer is falling behind. + // Zero (the default) disables the warning. See + // DurableMailboxConfig.SoftHighWatermark. + SoftHighWatermark int + + // HardHighWatermark is the persistent backlog depth at which sends to + // the actor's mailbox are refused with ErrMailboxSaturated. Zero (the + // default) disables the bound; control-priority messages and + // outbox-propagated deliveries are always exempt. See + // DurableMailboxConfig.HardHighWatermark. + HardHighWatermark int + // CleanupTimeout specifies the maximum duration for OnStop cleanup. // Default: 5 seconds. CleanupTimeout time.Duration @@ -385,14 +398,16 @@ func NewDurableActor[M TLVMessage, R any]( // normalizes zero and inverted values, so an actor config that predates // MaxPollInterval still lands on the default ceiling. mailboxCfg := DurableMailboxConfig{ - MailboxID: cfg.ID, - Store: cfg.Store, - Codec: cfg.Codec, - Clock: cfg.Clock, - LeaseDuration: cfg.LeaseDuration, - PollInterval: cfg.PollInterval, - MaxPollInterval: cfg.MaxPollInterval, - MaxAttempts: cfg.MaxAttempts, + MailboxID: cfg.ID, + Store: cfg.Store, + Codec: cfg.Codec, + Clock: cfg.Clock, + LeaseDuration: cfg.LeaseDuration, + PollInterval: cfg.PollInterval, + MaxPollInterval: cfg.MaxPollInterval, + MaxAttempts: cfg.MaxAttempts, + SoftHighWatermark: cfg.SoftHighWatermark, + HardHighWatermark: cfg.HardHighWatermark, // Size the wake channel to the worker count so a burst of // enqueues can rouse every idle worker at once. diff --git a/baselib/actor/durable_mailbox.go b/baselib/actor/durable_mailbox.go index b5fdf578a..9d5ef0bb4 100644 --- a/baselib/actor/durable_mailbox.go +++ b/baselib/actor/durable_mailbox.go @@ -147,6 +147,27 @@ type DurableMailboxConfig struct { // depends on a wake because the poll ticker is the fallback. WakeBuffer int + // SoftHighWatermark is the persistent backlog depth at which the + // mailbox starts logging that its consumer is falling behind. Sends + // still succeed. Zero (the default) disables the soft warning. The + // check only runs when Store implements MailboxDepthStore; a value + // above HardHighWatermark is lowered to it at construction. + SoftHighWatermark int + + // HardHighWatermark is the persistent backlog depth at which Send + // refuses new messages with ErrMailboxSaturated, shedding load at the + // producer instead of growing the backlog without bound. Zero (the + // default) disables the bound. Messages with priority >= + // ControlPriority (restart and boot restore/resume messages) are + // exempt so recovery always lands, as are outbox-propagated CDC + // deliveries (the message was already accepted at its true producer; + // refusing the hand-off would dead-letter a committed row instead of + // shedding load). Depth is read through a TTL-cached probe plus a + // local sent-since-probe delta, so enforcement is approximate within + // the probe window but the common send path never pays for an extra + // COUNT query. + HardHighWatermark int + // SingleWorkerLeaseless enables the leaseless peek consume path. When // set, Receive claims the next message with a READ-only PeekNextMessage // instead of the write-transaction LeaseNextMessage, and yields a @@ -315,6 +336,14 @@ type DurableMailbox[M TLVMessage, R any] struct { // actorCtx is the actor's lifecycle context. actorCtx context.Context + // depthStore is non-nil when cfg.Store can report mailbox backlog + // depth, which is what the watermark admission check reads. Resolved + // once at construction so the send path pays no type assertion. + depthStore MailboxDepthStore + + // depth caches the probed backlog depth for the watermark check. + depth depthProbe + // promiseRegistry maps message IDs to in-flight promises for Ask // messages. This allows the delivery to complete the promise after // processing. @@ -342,6 +371,13 @@ func NewDurableMailbox[M TLVMessage, R any]( cfg.PollInterval, cfg.MaxPollInterval, ) + // Resolve the watermark pair the same way: a soft warning above the + // hard refusal would make failing sends the operator's first signal, + // so it is lowered to the hard value instead. + cfg.SoftHighWatermark, cfg.HardHighWatermark = normalizeWatermarks( + cfg.SoftHighWatermark, cfg.HardHighWatermark, + ) + m := &DurableMailbox[M, R]{ cfg: cfg, clock: cfg.Clock.UnwrapOr(clock.NewDefaultClock()), @@ -365,6 +401,13 @@ func NewDurableMailbox[M TLVMessage, R any]( ) } + // Resolve the depth surface once so the watermark check on the send + // path is a nil test rather than a per-send type assertion. A store + // without the surface simply runs without watermarks. + if depthStore, ok := cfg.Store.(MailboxDepthStore); ok { + m.depthStore = depthStore + } + return m } @@ -420,6 +463,20 @@ func (m *DurableMailbox[M, R]) Send(ctx context.Context, return ErrMailboxClosed } + // Determine priority before the watermark check so restart-priority + // messages can be exempted from admission control. + priority := 0 + if pm, ok := any(env.message).(PriorityMessage); ok { + priority = pm.Priority() + } + + // Admit or refuse the send against the configured backlog watermarks + // BEFORE encoding or registering a promise, so a refusal needs no + // cleanup and TrySend/TryTell inherit the check for free. + if err := m.checkWatermarks(ctx, priority); err != nil { + return err + } + payload, err := m.cfg.Codec.Encode(env.message) if err != nil { return fmt.Errorf("encode mailbox message: %w", err) @@ -448,12 +505,6 @@ func (m *DurableMailbox[M, R]) Send(ctx context.Context, m.promiseRegistryMu.Unlock() } - // Determine priority. - priority := 0 - if pm, ok := any(env.message).(PriorityMessage); ok { - priority = pm.Priority() - } - // Enqueue the message. CorrelationKey opts the message into the // per-key FIFO claim lane in the durable mailbox; an empty key (the // default on BaseMessage) means the message uses the existing global diff --git a/baselib/actor/interface.go b/baselib/actor/interface.go index 71ad17199..cf35b0d86 100644 --- a/baselib/actor/interface.go +++ b/baselib/actor/interface.go @@ -134,6 +134,16 @@ var ErrMailboxFull = fmt.Errorf("mailbox full") // because it has been closed. var ErrMailboxClosed = fmt.Errorf("mailbox closed") +// ErrMailboxSaturated indicates that a durable mailbox refused a send because +// its persistent backlog is at or above the configured hard high watermark. +// It is the durable analogue of ErrMailboxFull: an in-memory mailbox bounds +// its queue with channel capacity, a durable mailbox with its watermarks. The +// message was NOT enqueued, so the caller may drop, stash, or retry it later, +// exactly as it would on ErrMailboxFull. The refusal is depth-based rather +// than transactional, so a retry after the consumer catches up succeeds. +// Mailboxes with no watermarks configured (the default) never return it. +var ErrMailboxSaturated = fmt.Errorf("mailbox saturated") + // TellOnlyRef is a reference to an actor that only supports "tell" operations. // This is useful for scenarios where only fire-and-forget message passing is // needed, or to restrict capabilities. @@ -161,18 +171,23 @@ type TellOnlyRef[M Message] interface { // Durable targets behave differently in three ways that callers must // plan for. // - // First, a durable queue has no capacity, so a durable ref never - // returns ErrMailboxFull. What it returns under load is the failure - // of a bounded database write, because the enqueue is a write that - // the mailbox bounds with a short internal deadline rather than - // completing instantaneously. That is usually a wrapped - // context.DeadlineExceeded, but the identity is the driver's to - // choose and some report a deadline as an error of their own. A + // First, a durable queue has no in-memory capacity, so a durable ref + // never returns ErrMailboxFull. Its capacity signal is + // ErrMailboxSaturated instead: a mailbox configured with backlog + // watermarks refuses the enqueue once its persistent backlog crosses + // the hard watermark, and a mailbox without watermarks (the default) + // never refuses for depth at all. What a durable ref returns under + // database load is the failure of a bounded write, because the + // enqueue is a write that the mailbox bounds with a short internal + // deadline rather than completing instantaneously. That is usually a + // wrapped context.DeadlineExceeded, but the identity is the driver's + // to choose and some report a deadline as an error of their own. A // caller that only retries on ErrMailboxFull therefore discards // durable messages the moment the database slows down, and one that // keys off the deadline instead is only slightly better off. Decide // by exclusion: retry on anything that is not ErrActorTerminated or - // ErrMailboxClosed. + // ErrMailboxClosed (ErrMailboxSaturated included: it clears once the + // consumer drains back under the watermark). // // Second, a durable TryTell drops the caller's database transaction: // the mailbox performs its bounded write on its own background diff --git a/baselib/actor/mailbox_watermarks.go b/baselib/actor/mailbox_watermarks.go new file mode 100644 index 000000000..a08fd192d --- /dev/null +++ b/baselib/actor/mailbox_watermarks.go @@ -0,0 +1,270 @@ +package actor + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" +) + +// A durable mailbox has no in-memory capacity, so the classic "mailbox full" +// backpressure signal never fires for it: every Tell lands as a database row +// and the backlog grows without bound while the consumer falls behind. The +// watermark check in this file is the missing bound. It reads the persistent +// backlog depth (with a TTL-cached probe so the common path costs nothing) +// and refuses new sends with ErrMailboxSaturated once the depth crosses a +// configured hard high watermark, with a soft watermark below it that only +// logs. Control-priority messages (restore/resume/restart) and +// outbox-propagated CDC deliveries are exempt so recovery and committed +// hand-offs always land, and a mailbox with no watermarks configured behaves +// exactly as before. + +const ( + // DefaultSoftHighWatermark is the backlog depth at which a durable + // mailbox that opts into watermarks starts logging that its consumer + // is falling behind. Sends still succeed; the soft watermark is purely + // an early-warning signal. + DefaultSoftHighWatermark = 1_000 + + // DefaultHardHighWatermark is the backlog depth at which a durable + // mailbox that opts into watermarks refuses new sends with + // ErrMailboxSaturated. At this depth the consumer is more than an + // order of magnitude behind the soft warning, so accepting more work + // only deepens the hole: shedding load at the producer is the only + // move that helps. + DefaultHardHighWatermark = 10_000 + + // depthProbeTTL bounds how often the watermark check issues a real + // COUNT query against the store. Between probes the check works off + // the cached depth plus a local count of sends accepted since the + // probe: within the window, LOCAL sends can only push the estimate up + // (acks that shrink the backlog surface at the next probe), which is + // the safe direction for an admission check. Sends from other + // processes or replicas are invisible for up to one window, so the + // bound is approximate, not exact. A one-second window keeps the + // probe cost negligible against the write each send already performs. + depthProbeTTL = time.Second +) + +// MailboxDepthCount reports the pending backlog of one durable mailbox. +type MailboxDepthCount struct { + // MailboxID identifies the mailbox. + MailboxID string + + // Depth is the number of rows currently parked in the mailbox, + // leased or not. Rows are deleted on ack, so this is exactly the + // undelivered backlog. + Depth int64 +} + +// MailboxDepthStore is the read-side surface for observing durable mailbox +// backlog depth. It is deliberately separate from DeliveryStore: the delivery +// pipeline never needs these reads, and folding them into DeliveryStore would +// force every test double through the depth surface. The watermark check and +// the metrics scrape both discover the capability with a type assertion, and +// a store that does not implement it simply runs without watermarks. +type MailboxDepthStore interface { + // MailboxDepth returns the number of messages currently parked in the + // given mailbox, including leased (in-flight) messages. Rows are + // deleted on ack, so the count is the undelivered backlog. + MailboxDepth(ctx context.Context, mailboxID string) (int64, error) + + // MailboxDepths returns the backlog of every mailbox that currently + // holds at least one message. Mailboxes with an empty backlog are + // absent from the result, which keeps a scrape cheap and bounded. + MailboxDepths(ctx context.Context) ([]MailboxDepthCount, error) +} + +// depthProbe caches the mailbox's probed backlog depth so the watermark check +// on the send path does not issue a COUNT query per send. Between probes the +// estimate is the probed depth plus the sends this mailbox accepted since, +// which is deliberately one-sided for LOCAL traffic: local sends push the +// estimate up immediately, while acks only surface at the next probe. +// Overshooting is the safe direction for an admission check; remote sends +// remain invisible for up to one probe window, so the bound is approximate. +type depthProbe struct { + mu sync.Mutex + + // probedAt is when the cached depth was last read from the store. The + // zero value forces a probe on the first checked send. + probedAt time.Time + + // probing is true while one sender runs the COUNT query outside the + // mutex. It single-flights the probe: concurrent senders keep using + // the cached estimate instead of stacking up behind the query or + // issuing duplicates. + probing bool + + // depth is the backlog depth reported by the last probe. + depth int64 + + // sentSinceProbe counts sends accepted since the last probe. + sentSinceProbe int64 + + // softBreached tracks whether the estimate was at or above the soft + // watermark after the last check, so the breach and the recovery are + // each logged once per episode instead of once per send. + softBreached bool +} + +// checkWatermarks admits or refuses a send against the mailbox's configured +// backlog watermarks. It returns nil when watermarks are disabled, the store +// cannot report depth, the message is exempt (control priority or an +// outbox-propagated delivery), or the estimated depth is below the hard +// watermark. It returns an error wrapping ErrMailboxSaturated when the +// estimate is at or above the hard watermark. +// +// A probe failure fails OPEN: refusing delivery because a monitoring read +// broke would convert an observability fault into message loss, which is +// strictly worse than a temporarily unenforced bound. +func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, + priority int) error { + + soft, hard := m.cfg.SoftHighWatermark, m.cfg.HardHighWatermark + if m.depthStore == nil || (soft <= 0 && hard <= 0) { + return nil + } + + // Control and restart messages always land: a saturated mailbox + // usually means the actor is wedged or gone, and the restore/resume + // message that would un-wedge it must not be refused by the very + // backlog it exists to drain. + if priority >= ControlPriority { + return nil + } + + // Outbox-propagated deliveries are exempt: the message was already + // accepted at its true producer and durably committed to the outbox, + // so refusing the CDC hand-off here sheds nothing — it only strands a + // committed message. Worse, the publisher's claim path bumps delivery + // attempts in its own transaction, so repeated refusals would + // dead-letter the outbox row (and any DurableAsk response it carries) + // instead of exerting backpressure. The publisher stamps the outbox ID + // into the context on every folded delivery, which is exactly the + // marker keyed off here. + if _, fromOutbox := OutboxIDFromContext(ctx); fromOutbox { + return nil + } + + m.depth.mu.Lock() + defer m.depth.mu.Unlock() + + now := m.clock.Now() + stale := m.depth.probedAt.IsZero() || + now.Sub(m.depth.probedAt) >= depthProbeTTL + + // Refresh the cached depth, single-flighted: only one sender runs the + // COUNT while concurrent senders keep using the cached estimate. The + // query runs OUTSIDE the mutex (so senders never park behind a slow + // read) and OUTSIDE the caller's ambient transaction: joining a + // sender's SERIALIZABLE write transaction would take predicate locks + // over the whole mailbox partition and manufacture rw-conflicts with + // the consumer's concurrent acks, exactly when the system is already + // contended. The local delta already accounts for this sender's own + // uncommitted enqueues, so the probe only needs committed state. + if stale && !m.depth.probing { + m.depth.probing = true + + // Snapshot the delta before releasing the mutex: sends that + // race the probe increment it while the COUNT runs, and their + // rows are enqueued after their check returns, so the COUNT + // cannot see them. Subtracting the snapshot (rather than + // zeroing) keeps those racing increments in the estimate, + // preserving the local no-undershoot property across the + // probe window. + preProbeSent := m.depth.sentSinceProbe + m.depth.mu.Unlock() + + depth, err := m.depthStore.MailboxDepth( + WithoutTx(ctx), m.cfg.MailboxID, + ) + + m.depth.mu.Lock() + m.depth.probing = false + + if err != nil { + log := logger(m.actorCtx) + log.WarnS(ctx, "Mailbox depth probe failed, "+ + "falling back to cached estimate", err, + slog.String("mailbox_id", m.cfg.MailboxID), + ) + } else { + m.depth.probedAt = m.clock.Now() + m.depth.depth = depth + m.depth.sentSinceProbe -= preProbeSent + } + } + + // No baseline at all -- the first probe failed or is still in flight + // on another sender -- fails open. + if m.depth.probedAt.IsZero() { + return nil + } + + estimate := m.depth.depth + m.depth.sentSinceProbe + + // Evaluate the soft-watermark episode BEFORE the hard refusal, so a + // backlog that enters saturation within one probe window still opens + // the episode: the soft warning must fire at or before the first + // refused send, or failing sends would be the operator's first signal. + // The episode logs once per breach: the first check that finds the + // estimate over the soft mark opens it, and the first check after it + // falls back under closes it. + if soft > 0 { + breached := estimate >= int64(soft) + switch { + case breached && !m.depth.softBreached: + log := logger(m.actorCtx) + log.WarnS(ctx, "Mailbox backlog crossed soft "+ + "watermark: consumer is falling behind", nil, + slog.String("mailbox_id", m.cfg.MailboxID), + slog.Int64("depth", estimate), + slog.Int("soft_watermark", soft), + slog.Int("hard_watermark", hard), + ) + + case !breached && m.depth.softBreached: + log := logger(m.actorCtx) + log.InfoS(ctx, "Mailbox backlog fell below soft "+ + "watermark", + slog.String("mailbox_id", m.cfg.MailboxID), + slog.Int64("depth", estimate), + slog.Int("soft_watermark", soft), + ) + } + m.depth.softBreached = breached + } + + if hard > 0 && estimate >= int64(hard) { + return fmt.Errorf("mailbox %s backlog %d at hard watermark "+ + "%d: %w", m.cfg.MailboxID, estimate, hard, + ErrMailboxSaturated) + } + + // Count this send into the estimate now, before the enqueue runs: a + // failed enqueue leaves the estimate one high until the next probe, + // which is the safe direction. + m.depth.sentSinceProbe++ + + return nil +} + +// normalizeWatermarks resolves the soft/hard watermark pair from raw config +// values. Negative values are treated as disabled, and a soft watermark above +// the hard one is lowered to it: the soft warning must fire at or before the +// hard refusal, or the operator's first signal of trouble would be failing +// sends. +func normalizeWatermarks(soft, hard int) (int, int) { + if soft < 0 { + soft = 0 + } + if hard < 0 { + hard = 0 + } + if hard > 0 && soft > hard { + soft = hard + } + + return soft, hard +} diff --git a/baselib/actor/mailbox_watermarks_test.go b/baselib/actor/mailbox_watermarks_test.go new file mode 100644 index 000000000..9a31a8051 --- /dev/null +++ b/baselib/actor/mailbox_watermarks_test.go @@ -0,0 +1,417 @@ +package actor + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/lightningnetwork/lnd/tlv" + "github.com/stretchr/testify/require" +) + +// depthReportingStore wraps the mock delivery store with a controllable +// MailboxDepthStore surface, so watermark tests can steer the probed depth +// independently of the rows actually enqueued. +type depthReportingStore struct { + *mockDeliveryStore + + // depth is the backlog depth every probe reports. + depth int64 + + // probeErr, when set, makes every probe fail. + probeErr error + + // probes counts MailboxDepth calls, so tests can assert the TTL cache + // is actually suppressing probes on the send path. + probes int + + // sawAmbientTx records whether any probe arrived with a database + // transaction still in its context. The watermark probe must strip + // the sender's ambient transaction, so tests assert this stays false. + sawAmbientTx bool +} + +// MailboxDepth reports the configured depth and counts the probe. +func (d *depthReportingStore) MailboxDepth(ctx context.Context, _ string) ( + int64, error) { + + d.probes++ + if HasTx(ctx) { + d.sawAmbientTx = true + } + if d.probeErr != nil { + return 0, d.probeErr + } + + return d.depth, nil +} + +// MailboxDepths reports a single-entry listing for the configured depth. +func (d *depthReportingStore) MailboxDepths(_ context.Context) ( + []MailboxDepthCount, error) { + + return []MailboxDepthCount{{MailboxID: "test-mailbox", Depth: d.depth}}, + nil +} + +// newWatermarkMailbox builds a durable mailbox over a depth-reporting store +// with the given watermarks and a test clock the caller controls. +func newWatermarkMailbox(t *testing.T, soft, hard int) (*DurableMailbox[ + *durablePriorityTestMsg, int], *depthReportingStore, *clock.TestClock) { + + t.Helper() + + store := &depthReportingStore{ + mockDeliveryStore: newMockDeliveryStore(), + } + testClock := clock.NewTestClock(time.Unix(1_000_000, 0)) + + cfg := DefaultDurableMailboxConfig( + "test-mailbox", store, newDurableTestCodec(), + ) + cfg.SoftHighWatermark = soft + cfg.HardHighWatermark = hard + cfg.Clock = fn.Some[clock.Clock](testClock) + + mailbox := NewDurableMailbox[*durablePriorityTestMsg, int]( + context.Background(), cfg, + ) + + return mailbox, store, testClock +} + +// watermarkTestEnv builds a sendable envelope with the given priority. +func watermarkTestEnv(priority int) envelope[*durablePriorityTestMsg, int] { + msg := &durablePriorityTestMsg{ + durableTestMsg: durableTestMsg{ + Value: tlv.NewPrimitiveRecord[tlv.TlvType1](uint64(1)), + Payload: tlv.NewPrimitiveRecord[tlv.TlvType2]( + []byte("wm"), + ), + }, + priority: priority, + } + + return envelope[*durablePriorityTestMsg, int]{ + message: msg, + callerCtx: context.Background(), + } +} + +// TestWatermarksDisabledByDefault asserts that a mailbox with no watermarks +// configured admits sends regardless of backlog depth, and never probes the +// depth surface at all. +func TestWatermarksDisabledByDefault(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 0) + store.depth = 1_000_000 + + require.NoError( + t, + mailbox.Send( + context.Background(), watermarkTestEnv(0), + ), + ) + require.Equal(t, 0, store.probes) +} + +// TestHardWatermarkRefusesSend asserts that a backlog at the hard watermark +// turns a send away with ErrMailboxSaturated and enqueues nothing. +func TestHardWatermarkRefusesSend(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 5 + + err := mailbox.Send(context.Background(), watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) + + store.mu.Lock() + require.Empty(t, store.messages) + store.mu.Unlock() +} + +// TestHardWatermarkRestartPriorityExempt asserts that a restart-priority +// message is admitted even when the backlog is far past the hard watermark: +// the RestartMessage that would un-wedge a stuck actor must not be refused by +// the backlog it exists to drain. +func TestHardWatermarkRestartPriorityExempt(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 500 + + err := mailbox.Send( + context.Background(), watermarkTestEnv(RestartPriority), + ) + require.NoError(t, err) + + store.mu.Lock() + require.Len(t, store.messages, 1) + store.mu.Unlock() + + // The exemption short-circuits before the probe, so the depth surface + // is never consulted for a restart message. + require.Equal(t, 0, store.probes) +} + +// TestControlPriorityExempt asserts that a control-priority message (the +// tier boot-time restore/resume messages sit at) is admitted at saturation, +// without ever consulting the depth surface. +func TestControlPriorityExempt(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 500 + + err := mailbox.Send( + context.Background(), watermarkTestEnv(ControlPriority), + ) + require.NoError(t, err) + require.Equal(t, 0, store.probes) +} + +// TestOutboxDeliveryExemptFromWatermarks asserts that an outbox-propagated +// send (the CDC hand-off, marked by the outbox ID the publisher stamps into +// the context) is admitted at saturation: the message was already accepted +// at its true producer, so refusing the hand-off would dead-letter a +// committed outbox row instead of shedding load. +func TestOutboxDeliveryExemptFromWatermarks(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 500 + + ctx := WithOutboxID(context.Background(), generateID()) + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + + store.mu.Lock() + require.Len(t, store.messages, 1) + store.mu.Unlock() + + require.Equal(t, 0, store.probes) +} + +// TestProbeStripsAmbientTx asserts that the depth probe never runs inside +// the sender's ambient database transaction: joining a writer's transaction +// would take predicate locks over the whole mailbox partition and +// manufacture serialization conflicts with the consumer's acks. +func TestProbeStripsAmbientTx(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 1 + + ctx := WithTx(context.Background(), nil) + require.True(t, HasTx(ctx)) + + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + require.Equal(t, 1, store.probes) + require.False(t, store.sawAmbientTx) +} + +// TestSoftEpisodeOpensAtSaturation asserts that a backlog that enters +// saturation within one probe window still opens the soft-watermark episode: +// the operator's designed first signal must fire at or before the first +// refused send. +func TestSoftEpisodeOpensAtSaturation(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 3, 5) + store.depth = 10 + + err := mailbox.Send(context.Background(), watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) + require.True(t, mailbox.depth.softBreached) +} + +// TestWatermarkLocalDeltaCrossesHard asserts that sends accepted inside one +// probe window count against the hard watermark: with a probed depth of 3 and +// a hard watermark of 5, the two sends that lift the estimate to 5 are +// admitted and the third is refused, all on a single probe. +func TestWatermarkLocalDeltaCrossesHard(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 3 + ctx := context.Background() + + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + + err := mailbox.Send(ctx, watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) + + require.Equal(t, 1, store.probes) +} + +// TestWatermarkProbeTTLRefresh asserts that a saturated mailbox recovers once +// the probe TTL elapses and the store reports a drained backlog. +func TestWatermarkProbeTTLRefresh(t *testing.T) { + t.Parallel() + + mailbox, store, testClock := newWatermarkMailbox(t, 0, 5) + store.depth = 5 + ctx := context.Background() + + err := mailbox.Send(ctx, watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) + + // Within the TTL the cached probe still refuses, without re-probing. + err = mailbox.Send(ctx, watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) + require.Equal(t, 1, store.probes) + + // The consumer drains the backlog; once the TTL elapses the next send + // re-probes and is admitted. + store.depth = 0 + testClock.SetTime(testClock.Now().Add(depthProbeTTL)) + + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + require.Equal(t, 2, store.probes) +} + +// TestWatermarkProbeFailureFailsOpen asserts that a failed depth probe admits +// the send: a broken monitoring read must not become message loss. +func TestWatermarkProbeFailureFailsOpen(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 1) + store.depth = 100 + store.probeErr = errors.New("depth read broke") + + require.NoError( + t, + mailbox.Send( + context.Background(), watermarkTestEnv(0), + ), + ) +} + +// TestSoftWatermarkWarnsWithoutRefusing asserts that a backlog past the soft +// watermark (with no hard watermark) never refuses a send, and that the +// breach episode opens and closes with the estimate. +func TestSoftWatermarkWarnsWithoutRefusing(t *testing.T) { + t.Parallel() + + mailbox, store, testClock := newWatermarkMailbox(t, 3, 0) + store.depth = 10 + ctx := context.Background() + + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + require.True(t, mailbox.depth.softBreached) + + // Drain the backlog: after the TTL elapses the next send closes the + // episode. + store.depth = 0 + testClock.SetTime(testClock.Now().Add(depthProbeTTL)) + + require.NoError(t, mailbox.Send(ctx, watermarkTestEnv(0))) + require.False(t, mailbox.depth.softBreached) +} + +// TestTrySendInheritsSaturation asserts that the non-blocking send path +// surfaces the same saturation refusal as Send. +func TestTrySendInheritsSaturation(t *testing.T) { + t.Parallel() + + mailbox, store, _ := newWatermarkMailbox(t, 0, 5) + store.depth = 5 + + err := mailbox.TrySend(watermarkTestEnv(0)) + require.ErrorIs(t, err, ErrMailboxSaturated) +} + +// TestWatermarksWithoutDepthStore asserts that configuring watermarks over a +// store without the depth surface is a harmless no-op. +func TestWatermarksWithoutDepthStore(t *testing.T) { + t.Parallel() + + store := newMockDeliveryStore() + cfg := DefaultDurableMailboxConfig( + "test-mailbox", store, newDurableTestCodec(), + ) + cfg.HardHighWatermark = 1 + + mailbox := NewDurableMailbox[*durablePriorityTestMsg, int]( + context.Background(), cfg, + ) + + require.NoError( + t, + mailbox.Send( + context.Background(), watermarkTestEnv(0), + ), + ) +} + +// TestNormalizeWatermarks pins the construction-time normalization: negatives +// disable, and a soft value above the hard value is lowered to it. +func TestNormalizeWatermarks(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + soft int + hard int + wantSoft int + wantHard int + }{ + { + "both disabled", + 0, + 0, + 0, + 0, + }, + { + "negatives disable", + -5, + -1, + 0, + 0, + }, + { + "soft only", + 100, + 0, + 100, + 0, + }, + { + "hard only", + 0, + 100, + 0, + 100, + }, + { + "ordered pair kept", + 100, + 1000, + 100, + 1000, + }, + { + "soft above hard lowered", + 5000, + 1000, + 1000, + 1000, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + soft, hard := normalizeWatermarks(tc.soft, tc.hard) + require.Equal(t, tc.wantSoft, soft) + require.Equal(t, tc.wantHard, hard) + }) + } +} diff --git a/baselib/actor/restart.go b/baselib/actor/restart.go index 3bdc5f9d2..44a700783 100644 --- a/baselib/actor/restart.go +++ b/baselib/actor/restart.go @@ -22,6 +22,16 @@ const RestartTLVType tlv.Type = 0xFFFE // Uses math.MaxInt32 to ensure restart messages are processed first. const RestartPriority = math.MaxInt32 +// ControlPriority is the priority level for domain-level control messages +// that recovery paths depend on: boot-time restore/resume requests and +// similar admissions that must land even when the target mailbox's backlog +// is past its hard watermark. Messages at or above this priority bypass the +// backlog admission check entirely (RestartPriority included), because the +// message that would un-wedge a backed-up actor must not be refused by the +// very backlog it exists to drain. It sits one below RestartPriority so a +// checkpoint restore still claims strictly first. +const ControlPriority = math.MaxInt32 - 1 + // RestartMessage is a special message sent to an actor when it starts up and // has a persisted checkpoint. This allows the actor to restore its FSM state // and continue processing from where it left off after a crash. diff --git a/credit/messages.go b/credit/messages.go index 80818545b..18da03a7a 100644 --- a/credit/messages.go +++ b/credit/messages.go @@ -164,6 +164,16 @@ type ResumeCreditOpRequest struct { FromRetryTimer bool } +// Priority marks the resume as a control message so it bypasses the durable +// mailbox's backlog admission check: a per-operation backlog past the hard +// watermark is exactly the condition the boot-time resume exists to work +// through, and refusing it would silently strand the in-flight operation. +// Control priority also claims the resume ahead of the redelivered backlog, +// mirroring RestartMessage. +func (m *ResumeCreditOpRequest) Priority() int { + return actor.ControlPriority +} + // MessageType returns the human-readable message type. func (m *ResumeCreditOpRequest) MessageType() string { return "ResumeCreditOpRequest" diff --git a/credit/op_actor.go b/credit/op_actor.go index c1245ad97..7b7cd3c4b 100644 --- a/credit/op_actor.go +++ b/credit/op_actor.go @@ -108,6 +108,12 @@ func NewOpActor(cfg OpActorConfig) (*OpActor, error) { ) durableCfg.Log = cfg.Log + // Bound the credit actor's durable backlog with the shared default + // watermarks so a wedged consumer sheds load at the producer instead + // of growing its backlog without bound. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark + durable, err := actor.NewDurableActor(durableCfg).Unpack() if err != nil { return nil, err diff --git a/db/actordelivery/AGENTS.md b/db/actordelivery/AGENTS.md index 14ca29801..90920b7a6 100644 --- a/db/actordelivery/AGENTS.md +++ b/db/actordelivery/AGENTS.md @@ -21,7 +21,12 @@ other services can reuse durable actor storage without pulling unrelated tables. `RegisterMailboxWake(mailboxID, wake func())` registers a targeted, per-mailbox wake: `ExecTx` tracks which mailbox IDs actually received an enqueue inside the transaction and, on commit, fires only those consumers' - callbacks instead of broadcasting to every registered mailbox. + callbacks instead of broadcasting to every registered mailbox. Also + implements `actor.MailboxDepthStore` (`MailboxDepth`, `MailboxDepths`): + `COUNT(*)` reads of `mailbox_messages` — leased rows included, rows are + deleted on ack — backing the durable mailbox's watermark admission check + and the `waved_mailbox_depth` scrape gauges. The prefix of + `idx_mailbox_messages_available` covers the single-mailbox count. - `TxActorDeliveryStore` — Transaction-scoped delivery store wrapping a live `*sql.Tx`. Implements `actor.DeliveryStore` directly against the transaction without additional `ExecTx` wrapping. `EnqueueOutbox` sets a shared diff --git a/db/actordelivery/CLAUDE.md b/db/actordelivery/CLAUDE.md index 14ca29801..90920b7a6 100644 --- a/db/actordelivery/CLAUDE.md +++ b/db/actordelivery/CLAUDE.md @@ -21,7 +21,12 @@ other services can reuse durable actor storage without pulling unrelated tables. `RegisterMailboxWake(mailboxID, wake func())` registers a targeted, per-mailbox wake: `ExecTx` tracks which mailbox IDs actually received an enqueue inside the transaction and, on commit, fires only those consumers' - callbacks instead of broadcasting to every registered mailbox. + callbacks instead of broadcasting to every registered mailbox. Also + implements `actor.MailboxDepthStore` (`MailboxDepth`, `MailboxDepths`): + `COUNT(*)` reads of `mailbox_messages` — leased rows included, rows are + deleted on ack — backing the durable mailbox's watermark admission check + and the `waved_mailbox_depth` scrape gauges. The prefix of + `idx_mailbox_messages_available` covers the single-mailbox count. - `TxActorDeliveryStore` — Transaction-scoped delivery store wrapping a live `*sql.Tx`. Implements `actor.DeliveryStore` directly against the transaction without additional `ExecTx` wrapping. `EnqueueOutbox` sets a shared diff --git a/db/actordelivery/mailbox_depth_test.go b/db/actordelivery/mailbox_depth_test.go new file mode 100644 index 000000000..b3149e41f --- /dev/null +++ b/db/actordelivery/mailbox_depth_test.go @@ -0,0 +1,124 @@ +package actordelivery + +import ( + "testing" + "time" + + "github.com/lightninglabs/wavelength/baselib/actor" + "github.com/stretchr/testify/require" +) + +// enqueueDepthTestMsg parks one message in the given mailbox. +func enqueueDepthTestMsg(t *testing.T, store *testActorDeliveryStore, + mailboxID string) string { + + t.Helper() + + id := generateTestID() + err := store.EnqueueMessage(t.Context(), actor.EnqueueParams{ + ID: id, + MailboxID: mailboxID, + MessageType: "test.Message", + Payload: []byte{1, 2, 3}, + AvailableAt: store.clock.Now().Add(-time.Minute), + MaxAttempts: 3, + }) + require.NoError(t, err) + + return id +} + +// TestMailboxDepthCountsBacklog verifies that MailboxDepth reports every row +// parked in a mailbox, that leasing a message does NOT shrink the depth (the +// row is still undelivered backlog until it acks), and that an ack does. +func TestMailboxDepthCountsBacklog(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newActorDeliveryStoreForTest(t) + + // An untouched mailbox has zero depth. + depth, err := store.MailboxDepth(ctx, "actor-1") + require.NoError(t, err) + require.Zero(t, depth) + + for range 3 { + enqueueDepthTestMsg(t, store, "actor-1") + } + + depth, err = store.MailboxDepth(ctx, "actor-1") + require.NoError(t, err) + require.EqualValues(t, 3, depth) + + // A leased (in-flight) message still counts: it has not been + // delivered until its ack deletes the row. + leased, err := store.LeaseNextMessage( + ctx, "actor-1", "token-1", 30*time.Second, + ) + require.NoError(t, err) + require.NotNil(t, leased) + + depth, err = store.MailboxDepth(ctx, "actor-1") + require.NoError(t, err) + require.EqualValues(t, 3, depth) + + // Acking deletes the row and the depth drops. + n, err := store.AckMessage(ctx, leased.ID, "token-1") + require.NoError(t, err) + require.EqualValues(t, 1, n) + + depth, err = store.MailboxDepth(ctx, "actor-1") + require.NoError(t, err) + require.EqualValues(t, 2, depth) +} + +// TestMailboxDepthsListsOnlyBackedUpMailboxes verifies that the grouped depth +// listing reports one entry per mailbox holding messages, and none for +// mailboxes that are empty or fully drained. +func TestMailboxDepthsListsOnlyBackedUpMailboxes(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := newActorDeliveryStoreForTest(t) + + // No backlog anywhere: the listing is empty. + depths, err := store.MailboxDepths(ctx) + require.NoError(t, err) + require.Empty(t, depths) + + for range 2 { + enqueueDepthTestMsg(t, store, "actor-a") + } + id := enqueueDepthTestMsg(t, store, "actor-b") + + depths, err = store.MailboxDepths(ctx) + require.NoError(t, err) + + expected := []actor.MailboxDepthCount{ + { + MailboxID: "actor-a", + Depth: 2, + }, + { + MailboxID: "actor-b", + Depth: 1, + }, + } + require.Equal(t, expected, depths) + + // Draining actor-b removes it from the listing entirely. + n, err := store.AckMessageByID(ctx, id) + require.NoError(t, err) + require.EqualValues(t, 1, n) + + depths, err = store.MailboxDepths(ctx) + require.NoError(t, err) + + expected = []actor.MailboxDepthCount{ + { + MailboxID: "actor-a", + Depth: 2, + }, + } + require.Equal(t, expected, depths) +} diff --git a/db/actordelivery/queries/mailbox.sql b/db/actordelivery/queries/mailbox.sql index 785ca3070..70e031b85 100644 --- a/db/actordelivery/queries/mailbox.sql +++ b/db/actordelivery/queries/mailbox.sql @@ -191,6 +191,24 @@ SELECT COUNT(*) FROM mailbox_messages WHERE mailbox_id = $1 AND (lease_until IS NULL OR lease_until < $2); +-- name: CountMailboxMessages :one +-- Count every row currently parked in one mailbox, leased or not. Rows are +-- deleted on ack, so COUNT(*) is exactly the undelivered backlog. This is the +-- depth the durable mailbox's watermark admission check reads; the prefix of +-- idx_mailbox_messages_available covers the mailbox_id equality scan. +SELECT COUNT(*) FROM mailbox_messages +WHERE mailbox_id = $1; + +-- name: CountMailboxMessagesByMailbox :many +-- Report the backlog of every mailbox currently holding at least one message, +-- for the scrape-time depth gauges. Mailboxes with an empty backlog produce +-- no row, which keeps the result bounded by the number of backed-up actors +-- rather than the number of actors that have ever existed. +SELECT mailbox_id, COUNT(*) AS depth +FROM mailbox_messages +GROUP BY mailbox_id +ORDER BY mailbox_id; + -- name: ExpireMailboxLeases :exec -- Release all expired leases so messages can be redelivered. -- Called periodically by a background cleanup task. diff --git a/db/actordelivery/sqlc/mailbox.sql.go b/db/actordelivery/sqlc/mailbox.sql.go index 08acb7617..95b5076bb 100644 --- a/db/actordelivery/sqlc/mailbox.sql.go +++ b/db/actordelivery/sqlc/mailbox.sql.go @@ -177,6 +177,61 @@ func (q *Queries) CountDeadLetters(ctx context.Context) (int64, error) { return count, err } +const CountMailboxMessages = `-- name: CountMailboxMessages :one +SELECT COUNT(*) FROM mailbox_messages +WHERE mailbox_id = $1 +` + +// Count every row currently parked in one mailbox, leased or not. Rows are +// deleted on ack, so COUNT(*) is exactly the undelivered backlog. This is the +// depth the durable mailbox's watermark admission check reads; the prefix of +// idx_mailbox_messages_available covers the mailbox_id equality scan. +func (q *Queries) CountMailboxMessages(ctx context.Context, mailboxID string) (int64, error) { + row := q.db.QueryRowContext(ctx, CountMailboxMessages, mailboxID) + var count int64 + err := row.Scan(&count) + return count, err +} + +const CountMailboxMessagesByMailbox = `-- name: CountMailboxMessagesByMailbox :many +SELECT mailbox_id, COUNT(*) AS depth +FROM mailbox_messages +GROUP BY mailbox_id +ORDER BY mailbox_id +` + +type CountMailboxMessagesByMailboxRow struct { + MailboxID string + Depth int64 +} + +// Report the backlog of every mailbox currently holding at least one message, +// for the scrape-time depth gauges. Mailboxes with an empty backlog produce +// no row, which keeps the result bounded by the number of backed-up actors +// rather than the number of actors that have ever existed. +func (q *Queries) CountMailboxMessagesByMailbox(ctx context.Context) ([]CountMailboxMessagesByMailboxRow, error) { + rows, err := q.db.QueryContext(ctx, CountMailboxMessagesByMailbox) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CountMailboxMessagesByMailboxRow + for rows.Next() { + var i CountMailboxMessagesByMailboxRow + if err := rows.Scan(&i.MailboxID, &i.Depth); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const CountPendingMailboxMessages = `-- name: CountPendingMailboxMessages :one SELECT COUNT(*) FROM mailbox_messages WHERE mailbox_id = $1 diff --git a/db/actordelivery/sqlc/querier.go b/db/actordelivery/sqlc/querier.go index ac53acad3..60854dcff 100644 --- a/db/actordelivery/sqlc/querier.go +++ b/db/actordelivery/sqlc/querier.go @@ -35,6 +35,16 @@ type Querier interface { CompleteOutboxMessage(ctx context.Context, arg CompleteOutboxMessageParams) error // Count total dead letters. CountDeadLetters(ctx context.Context) (int64, error) + // Count every row currently parked in one mailbox, leased or not. Rows are + // deleted on ack, so COUNT(*) is exactly the undelivered backlog. This is the + // depth the durable mailbox's watermark admission check reads; the prefix of + // idx_mailbox_messages_available covers the mailbox_id equality scan. + CountMailboxMessages(ctx context.Context, mailboxID string) (int64, error) + // Report the backlog of every mailbox currently holding at least one message, + // for the scrape-time depth gauges. Mailboxes with an empty backlog produce + // no row, which keeps the result bounded by the number of backed-up actors + // rather than the number of actors that have ever existed. + CountMailboxMessagesByMailbox(ctx context.Context) ([]CountMailboxMessagesByMailboxRow, error) // Count pending messages for an actor's mailbox. CountPendingMailboxMessages(ctx context.Context, arg CountPendingMailboxMessagesParams) (int64, error) // Count pending outbox messages. diff --git a/db/actordelivery/store_impl.go b/db/actordelivery/store_impl.go index dc9186b9f..e1808ddce 100644 --- a/db/actordelivery/store_impl.go +++ b/db/actordelivery/store_impl.go @@ -76,6 +76,12 @@ type ActorDeliveryQueries interface { ExpireMailboxLeases(ctx context.Context, leaseUntil sql.NullInt64) error + CountMailboxMessages(ctx context.Context, + mailboxID string) (int64, error) + + CountMailboxMessagesByMailbox(ctx context.Context) ( + []adsqlc.CountMailboxMessagesByMailboxRow, error) + // Ask result operations. InsertAskResult(ctx context.Context, arg InsertAskResultParams) error @@ -1090,6 +1096,56 @@ func (s *Store) CleanupExpired(ctx context.Context) error { ) } +// MailboxDepth returns the number of messages currently parked in the given +// mailbox, leased or not. Rows are deleted on ack, so the count is exactly the +// undelivered backlog; the prefix of idx_mailbox_messages_available covers the +// scan. This is the read behind the durable mailbox's watermark admission +// check, part of the actor.MailboxDepthStore surface. +func (s *Store) MailboxDepth(ctx context.Context, mailboxID string) (int64, + error) { + + readTxOpts := db.ReadTxOption() + + var depth int64 + err := s.db.ExecTx(ctx, readTxOpts, func(q ActorDeliveryQueries) error { + var err error + depth, err = q.CountMailboxMessages(ctx, mailboxID) + + return err + }) + + return depth, err +} + +// MailboxDepths returns the backlog of every mailbox currently holding at +// least one message, for the scrape-time depth gauges. Mailboxes with an +// empty backlog are absent from the result. +func (s *Store) MailboxDepths(ctx context.Context) ([]actor.MailboxDepthCount, + error) { + + readTxOpts := db.ReadTxOption() + + var result []actor.MailboxDepthCount + err := s.db.ExecTx(ctx, readTxOpts, func(q ActorDeliveryQueries) error { + rows, err := q.CountMailboxMessagesByMailbox(ctx) + if err != nil { + return err + } + + result = make([]actor.MailboxDepthCount, 0, len(rows)) + for _, row := range rows { + result = append(result, actor.MailboxDepthCount{ + MailboxID: row.MailboxID, + Depth: row.Depth, + }) + } + + return nil + }) + + return result, err +} + // Helper functions for SQL type conversions. // toNullString converts a string to sql.NullString. @@ -1859,6 +1915,10 @@ var _ actor.OutboxWakeRegistrar = (*Store)(nil) // after a folded outbox enqueue commits. var _ actor.MailboxWakeRegistrar = (*Store)(nil) +// Compile-time check that Store exposes the mailbox depth surface the durable +// mailbox's watermark check and the metrics scrape discover by assertion. +var _ actor.MailboxDepthStore = (*Store)(nil) + // Compile-time check that TxAwareActorDeliveryStore implements // actor.TxAwareDeliveryStore. var _ actor.TxAwareDeliveryStore = (*TxAwareActorDeliveryStore)(nil) diff --git a/docs/durable_actor_architecture.md b/docs/durable_actor_architecture.md index 811aa4e26..cb395a5c2 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. [Backpressure Watermarks](#backpressure-watermarks) +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,101 @@ debug delivery issues and design retry strategies. --- +## Backpressure Watermarks + +A durable mailbox has no in-memory capacity, so the `ErrMailboxFull` signal +that bounds a channel mailbox never fires for it: every `Tell` lands as a +database row and the backlog grows for as long as the consumer lags. Backlog +watermarks are the durable analogue of that bound. + +### Semantics + +Two thresholds on `DurableMailboxConfig` (flowing through +`DurableActorConfig`), both measured against the mailbox's persistent backlog +(`COUNT(*)` of its `mailbox_messages` rows, leased or not — rows are deleted +on ack, so the count is exactly the undelivered backlog): + +- **`SoftHighWatermark`** (site default `actor.DefaultSoftHighWatermark`, + 1000): crossing it logs a warning, once per breach episode. Sends still + succeed. This is the operator's early signal that a consumer is falling + behind. +- **`HardHighWatermark`** (site default `actor.DefaultHardHighWatermark`, + 10000): at or past it, `Send` refuses the message with + `ErrMailboxSaturated` before encoding anything, so a refusal needs no + cleanup and `TrySend`/`TryTell` inherit the check for free. The message was + not enqueued; the caller sheds, stashes, or retries after the consumer + drains. + +Both default to zero (disabled) on the config structs; the bound is opted +into per actor. Two message classes are always exempt: + +- **Control-priority messages** (`priority >= actor.ControlPriority`, which + includes `RestartPriority`): the restart, restore, and resume messages + that would un-wedge a stuck actor must not be refused by the very backlog + they exist to drain. Boot-time restores (`RestoreNonTerminalRequest` in + OOR, `ResumeUnrollRequest` in unroll, `ResumeCreditOpRequest` in credit) + carry this priority because the daemon treats their failure as fatal: a + refusal there would turn a backed-up mailbox into a restart crash loop. +- **Outbox-propagated deliveries** (detected via the outbox ID the + publisher stamps into the context): the message was already accepted at + its true producer and durably committed to the outbox, so refusing the + CDC hand-off sheds nothing. Worse, the publisher's claim path bumps + delivery attempts in its own transaction, so repeated refusals would + dead-letter the committed outbox row (and any DurableAsk response it + carries) instead of exerting backpressure. + +A store that does not implement `actor.MailboxDepthStore` runs without +watermarks entirely, and a failed depth probe fails OPEN (the send is +admitted): a broken monitoring read must not become message loss. + +### The probe + +The depth read is TTL-cached (one second) with a local count of sends +accepted since the last probe added on top, so the common send path pays no +extra query. The estimate is one-sided for local traffic (local sends push +it up immediately, acks surface at the next probe; overshooting is the safe +direction for an admission check), while sends from other processes stay +invisible for up to one window, so the bound is approximate rather than +exact. The probe is single-flighted (concurrent senders use the cached +estimate rather than stacking behind the query) and runs with the sender's +ambient transaction stripped (`WithoutTx`): joining a SERIALIZABLE writer +would take predicate locks over the whole mailbox partition and manufacture +serialization conflicts with the consumer's acks, precisely when the system +is already contended. + +### Who refuses, and what callers do + +- **serverconn ingress** classifies `ErrMailboxSaturated` from a durable + target exactly like `ErrMailboxFull` from a bounded in-memory one: the + envelope defers, the cursor stalls, and the loop re-pulls after a backoff + (see `serverconn/dispatch_deferral.go`). A backed-up durable actor + therefore exerts backpressure on the operator stream instead of deepening + its own backlog. +- **Local producers** (RPC handlers, other actors) see the error from + `Tell`/`Ask` and propagate it; at ten thousand parked messages, failing + loudly is the only move that helps. +- **The OutboxPublisher's folded delivery path is exempt** via the outbox-ID + context marker described above, so CDC delivery is never refused and the + claim-expiry retry contract is untouched. + +**Known residual**: an actor turn that Tells into a saturated peer inside +its own commit (e.g. an OOR session's transport send into serverconn +egress) fails the whole turn, which nacks the INBOUND message and burns one +of its finite delivery attempts; a long enough saturation episode +dead-letters it. The dead-letter tooling (#1119) makes those visible and +requeueable, and the planned postpone semantics (re-enqueue without burning +attempts) are the structural fix; until then, saturation-driven turn +failures ride the ordinary retry/dead-letter path. + +### Observability + +The scrape-time gauges `waved_mailbox_backlog` (unlabelled total, explicit +zero) and `waved_mailbox_depth{mailbox_id}` (one series per backed-up +mailbox) read the same store surface; see `metrics/README.md` for the +alerting recipes. + +--- + ## Actor System Architecture The durable actor system consists of several interconnected components organized diff --git a/ledger/AGENTS.md b/ledger/AGENTS.md index ea99f06a4..7ce80578f 100644 --- a/ledger/AGENTS.md +++ b/ledger/AGENTS.md @@ -182,6 +182,13 @@ or balance reconciliation. Required emission pairs: events. - Fire-and-forget: `LedgerResp` is always nil; callers `Tell`, never `Ask`. +- **Soft watermark only, never a hard one.** The actor opts into + `SoftHighWatermark` for backlog observability but deliberately leaves + `HardHighWatermark` disabled: producers log-and-continue on a failed + `Tell` with no redrive path, so a hard `ErrMailboxSaturated` refusal + would not be backpressure but a permanently missing accounting leg. A + deep-but-eventually-complete backlog is the better failure mode for an + audit trail. - TLV stream encoding. `decodeAmountSat` narrows `uint64` to `int64` (rejects values past `MaxInt64`); `decodeFixedBytes` enforces exact lengths (`RoundID=16`, `SessionID=32`, `OutpointHash=32`). diff --git a/ledger/CLAUDE.md b/ledger/CLAUDE.md index ea99f06a4..7ce80578f 100644 --- a/ledger/CLAUDE.md +++ b/ledger/CLAUDE.md @@ -182,6 +182,13 @@ or balance reconciliation. Required emission pairs: events. - Fire-and-forget: `LedgerResp` is always nil; callers `Tell`, never `Ask`. +- **Soft watermark only, never a hard one.** The actor opts into + `SoftHighWatermark` for backlog observability but deliberately leaves + `HardHighWatermark` disabled: producers log-and-continue on a failed + `Tell` with no redrive path, so a hard `ErrMailboxSaturated` refusal + would not be backpressure but a permanently missing accounting leg. A + deep-but-eventually-complete backlog is the better failure mode for an + audit trail. - TLV stream encoding. `decodeAmountSat` narrows `uint64` to `int64` (rejects values past `MaxInt64`); `decodeFixedBytes` enforces exact lengths (`RoundID=16`, `SessionID=32`, `OutpointHash=32`). diff --git a/ledger/actor.go b/ledger/actor.go index 1bdb7016a..f8aef99c1 100644 --- a/ledger/actor.go +++ b/ledger/actor.go @@ -384,6 +384,18 @@ func (a *LedgerActor) Start(ctx context.Context) error { ]( a.actorID, a, a.bindStores, a.cfg.DeliveryStore, codec, ) + + // The ledger gets a SOFT watermark only, never the hard refusal. Its + // producers are fire-and-forget Tells that log-and-continue on + // failure with no redrive path, so a refused send is not backpressure + // but a permanently missing accounting leg: the double-entry balance + // and the audit trail would stay silently incomplete after the + // backlog drains. A deep-but-eventually-complete backlog is the + // better failure mode for an audit trail; the soft warning (and the + // waved_mailbox_depth gauge) keeps the depth observable and + // alertable. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durable, err := actor.NewDurableActor(durableCfg).Unpack() if err != nil { return fmt.Errorf("build ledger durable actor: %w", err) diff --git a/metrics/AGENTS.md b/metrics/AGENTS.md index d691a0b9d..7b30c1e30 100644 --- a/metrics/AGENTS.md +++ b/metrics/AGENTS.md @@ -23,7 +23,11 @@ OOR/round state). across the worker pool. - `SystemCollector` / `SystemStatsQuerier` — `prometheus.Collector` that queries live client state on each scrape (VTXO inventory/value, wallet - balance, block height, `oor_sessions_by_state`, `rounds_by_status`). Each + balance, block height, `oor_sessions_by_state`, `rounds_by_status`, + and durable-mailbox backlog: `waved_mailbox_backlog` emits an explicit + zero when every mailbox is drained, `waved_mailbox_depth{mailbox_id}` + one series per mailbox currently holding messages, sourced from the + delivery store's `actor.MailboxDepthStore` surface). Each querier method is collected independently; an error only suppresses that method's gauges for the scrape. - `Server` / `ServerConfig` — opt-in HTTP `/metrics` endpoint; disabled diff --git a/metrics/CLAUDE.md b/metrics/CLAUDE.md index d691a0b9d..7b30c1e30 100644 --- a/metrics/CLAUDE.md +++ b/metrics/CLAUDE.md @@ -23,7 +23,11 @@ OOR/round state). across the worker pool. - `SystemCollector` / `SystemStatsQuerier` — `prometheus.Collector` that queries live client state on each scrape (VTXO inventory/value, wallet - balance, block height, `oor_sessions_by_state`, `rounds_by_status`). Each + balance, block height, `oor_sessions_by_state`, `rounds_by_status`, + and durable-mailbox backlog: `waved_mailbox_backlog` emits an explicit + zero when every mailbox is drained, `waved_mailbox_depth{mailbox_id}` + one series per mailbox currently holding messages, sourced from the + delivery store's `actor.MailboxDepthStore` surface). Each querier method is collected independently; an error only suppresses that method's gauges for the scrape. - `Server` / `ServerConfig` — opt-in HTTP `/metrics` endpoint; disabled diff --git a/metrics/README.md b/metrics/README.md index 1b96b9a44..366f35703 100644 --- a/metrics/README.md +++ b/metrics/README.md @@ -47,6 +47,8 @@ label cardinality tracks live inventory. | `waved_block_height` | gauge | — | scrape (chain backend) | Best block height seen by the client's chain backend. | | `waved_oor_sessions_by_state` | gauge | `state` | scrape (OOR actor) | Currently-tracked (live) OOR sessions by state, e.g. `pending`. Lifetime totals live in `oor_transfers_*_total`. | | `waved_rounds_by_status` | gauge | `status` | scrape (round actor) | Currently-live rounds by status, e.g. `joined`, `confirmed`. Lifetime totals live in `rounds_*_total`. | +| `waved_mailbox_backlog` | gauge | — | scrape (delivery store) | Total messages pending across all durable mailboxes. Emits an explicit `0` when every mailbox is drained, so "all clear" and "scrape broke" are distinguishable. | +| `waved_mailbox_depth` | gauge | `mailbox_id` | scrape (delivery store) | Pending messages in one durable mailbox, leased (in-flight) rows included. Only mailboxes currently holding messages are reported, so per-session actor IDs never accumulate as permanent series. A mailbox that sits near its hard watermark (default 10000) is refusing new sends with `ErrMailboxSaturated`; the soft watermark (default 1000) logs a warning first. | The on-chain wallet balance complements the off-chain VTXO value: together they give a full picture of client funds. The `*_by_state` / `*_by_status` gauges @@ -127,7 +129,7 @@ stall. |--------|------|--------|--------|-------------| | `waved_serverconn_last_ingress_poll_timestamp_seconds` | gauge | — | connector (ingress loop) | Unix timestamp of the last `Pull` that returned to the ingress loop, including an empty long-poll. Fresh at the long-poll cadence on an idle client, so staleness means the loop goroutine itself is gone. | | `waved_serverconn_last_ingress_event_timestamp_seconds` | gauge | — | connector (ingress loop) | Unix timestamp of the last pulled batch the loop delivered and committed, including a partial commit made while backpressure held the rest. Only advances on real traffic, so on its own it cannot tell an idle client from a wedged one; a fresh poll stamp with a stale event stamp says the loop is running but dispatch is not getting through. | -| `waved_serverconn_ingress_dispatch_deferred_total` | counter | `service`, `method` | connector (ingress loop) | Redrives a full target actor mailbox turned away, by the route of the envelope that was refused. One increment per re-pull that could not deliver, **not** one per queued envelope: the loop meets the full mailbox once and stops there, so the envelopes behind the first are never attempted. Nothing is lost — the refused envelope is unacknowledged and re-pulled after a short backoff — but a rate that does not fall back to zero means a local actor has stopped draining. | +| `waved_serverconn_ingress_dispatch_deferred_total` | counter | `service`, `method` | connector (ingress loop) | Redrives a target actor turned away for want of room — a full in-memory mailbox, or a durable mailbox past its hard backlog watermark — by the route of the envelope that was refused. One increment per re-pull that could not deliver, **not** one per queued envelope: the loop meets the refusal once and stops there, so the envelopes behind the first are never attempted. Nothing is lost — the refused envelope is unacknowledged and re-pulled after a short backoff — but a rate that does not fall back to zero means a local actor has stopped draining. Cross-reference `waved_mailbox_depth` to tell the two refusals apart: a durable refusal shows the target's mailbox pinned at its hard watermark. | ### Alerting @@ -166,6 +168,21 @@ Both gauges are unlabelled, which assumes one connector per process — true tod gauge the max of the two and the first alert blind; add a `mailbox_id` label before that happens. +**A durable consumer is falling behind.** The depth gauges see the backlog long +before the hard watermark starts refusing sends. A warning at the soft +watermark gives the operator the same head start the log line does: + +``` +waved_mailbox_depth > 1000 +``` + +for 10m, which filters the transient burst a healthy consumer drains on its +own. A mailbox pinned at the hard watermark (default 10000) is actively +shedding load — new sends fail with `ErrMailboxSaturated` — and warrants a +page, because at that depth something downstream has stopped, not slowed. + +## gRPC Client Metrics + ## gRPC Client Metrics Per-method **client-side** metrics for calls `waved` makes to the ark diff --git a/metrics/collector.go b/metrics/collector.go index 1a99e088a..d0f8a13b6 100644 --- a/metrics/collector.go +++ b/metrics/collector.go @@ -82,6 +82,25 @@ type SystemStatsQuerier interface { // GetRoundStatsByStatus returns a count of currently-live rounds // grouped by status label. GetRoundStatsByStatus(ctx context.Context) (map[string]int64, error) + + // GetMailboxDepths returns the pending backlog of every durable + // mailbox currently holding at least one message. Mailboxes with an + // empty backlog are absent, which keeps the scrape bounded by the + // number of backed-up actors. Implementations should return an error + // (so the scrape skips the gauges) when the delivery store cannot + // report depth. + GetMailboxDepths(ctx context.Context) ([]MailboxDepthRow, error) +} + +// MailboxDepthRow holds the pending backlog of one durable mailbox, as +// reported by the delivery store at scrape time. +type MailboxDepthRow struct { + // MailboxID identifies the mailbox (typically the actor ID). + MailboxID string + + // Depth is the number of messages parked in the mailbox, leased or + // not. + Depth int64 } // Metric descriptors for the scrape-driven gauges. @@ -134,6 +153,17 @@ var ( "Number of currently-live rounds by status.", []string{"status"}, nil, ) + mailboxBacklogDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, "", "mailbox_backlog"), + "Total number of messages pending across all durable "+ + "mailboxes.", nil, nil, + ) + mailboxDepthDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, "", "mailbox_depth"), + "Number of messages pending in a durable mailbox. Only "+ + "mailboxes currently holding messages are reported.", + []string{"mailbox_id"}, nil, + ) ) // liveStatus is the VTXO status label whose value is summed into the @@ -171,6 +201,8 @@ func (c *SystemCollector) Describe(ch chan<- *prometheus.Desc) { ch <- blockHeightDesc ch <- oorSessionsDesc ch <- roundsByStatusDesc + ch <- mailboxBacklogDesc + ch <- mailboxDepthDesc } // Collect queries the client's live system state and emits the scrape @@ -189,6 +221,7 @@ func (c *SystemCollector) Collect(ch chan<- prometheus.Metric) { c.collectBlockHeight(ctx, ch) c.collectOORSessions(ctx, ch) c.collectRounds(ctx, ch) + c.collectMailboxDepths(ctx, ch) } // collectVTXOStats emits the VTXO inventory gauges (count and value by @@ -305,3 +338,34 @@ func (c *SystemCollector) collectRounds(ctx context.Context, ) } } + +// collectMailboxDepths emits the durable-mailbox backlog gauges: the +// unlabelled total, and one labelled series per mailbox currently holding +// messages. The total is emitted even at zero, so dashboards and alert +// rules can tell "every mailbox drained" apart from "the scrape broke"; +// the per-mailbox series appear only while their mailbox is backed up, so +// per-session actor IDs never accumulate as permanent label children. +func (c *SystemCollector) collectMailboxDepths(ctx context.Context, + ch chan<- prometheus.Metric) { + + rows, err := c.querier.GetMailboxDepths(ctx) + if err != nil { + c.log.Debugf("Mailbox depth query skipped during scrape: %v", + err) + + return + } + + var total int64 + for _, row := range rows { + total += row.Depth + ch <- prometheus.MustNewConstMetric( + mailboxDepthDesc, prometheus.GaugeValue, + float64(row.Depth), row.MailboxID, + ) + } + + ch <- prometheus.MustNewConstMetric( + mailboxBacklogDesc, prometheus.GaugeValue, float64(total), + ) +} diff --git a/metrics/collector_test.go b/metrics/collector_test.go index b94e9d30c..db7ba322e 100644 --- a/metrics/collector_test.go +++ b/metrics/collector_test.go @@ -55,6 +55,14 @@ func (m *mockVTXOQuerier) GetRoundStatsByStatus(_ context.Context) ( return nil, nil } +// GetMailboxDepths implements SystemStatsQuerier; unavailable in this +// VTXO-focused double so the depth gauges stay out of the way. +func (m *mockVTXOQuerier) GetMailboxDepths(_ context.Context) ( + []MailboxDepthRow, error) { + + return nil, errors.New("not supported") +} + // TestSystemCollectorCollect verifies the scrape-driven collector emits // the expected count, value, and spendable-balance samples for a range // of VTXO inventories, and emits nothing on a query failure. @@ -191,8 +199,10 @@ type fullMockQuerier struct { height int64 oor map[string]int64 rounds map[string]int64 + depths []MailboxDepthRow balanceErr error heightErr error + depthsErr error } func (m *fullMockQuerier) GetVTXOStatsByStatus(_ context.Context) ( @@ -223,6 +233,12 @@ func (m *fullMockQuerier) GetRoundStatsByStatus(_ context.Context) ( return m.rounds, nil } +func (m *fullMockQuerier) GetMailboxDepths(_ context.Context) ( + []MailboxDepthRow, error) { + + return m.depths, m.depthsErr +} + // TestSystemCollectorExtendedGauges verifies the wallet-balance, // block-height, OOR-sessions, and rounds-by-status gauges are emitted // from the querier, and that a wallet/height query error suppresses only @@ -314,6 +330,85 @@ waved_rounds_by_status{status="joined"} 1 }) } +// TestSystemCollectorMailboxDepths verifies the durable-mailbox backlog +// gauges: one labelled series per backed-up mailbox, an unlabelled total +// that is emitted even when every mailbox is drained (explicit zero), and +// full suppression when the depth query fails. +func TestSystemCollectorMailboxDepths(t *testing.T) { + t.Parallel() + + t.Run("backed up mailboxes", func(t *testing.T) { + t.Parallel() + + q := &fullMockQuerier{ + depths: []MailboxDepthRow{ + { + MailboxID: "ledger", + Depth: 4, + }, + { + MailboxID: "serverconn-egress", + Depth: 2, + }, + }, + } + c := NewSystemCollector(q, fn.None[btclog.Logger]()) + + const want = ` +# HELP waved_mailbox_backlog Total number of messages pending across ` + + `all durable mailboxes. +# TYPE waved_mailbox_backlog gauge +waved_mailbox_backlog 6 +# HELP waved_mailbox_depth Number of messages pending in a durable ` + + `mailbox. Only mailboxes currently holding messages ` + + `are reported. +# TYPE waved_mailbox_depth gauge +waved_mailbox_depth{mailbox_id="ledger"} 4 +waved_mailbox_depth{mailbox_id="serverconn-egress"} 2 +` + err := testutil.CollectAndCompare( + c, strings.NewReader(want), + "waved_mailbox_backlog", "waved_mailbox_depth", + ) + require.NoError(t, err) + }) + + t.Run("drained emits explicit zero total", func(t *testing.T) { + t.Parallel() + + q := &fullMockQuerier{} + c := NewSystemCollector(q, fn.None[btclog.Logger]()) + + const want = ` +# HELP waved_mailbox_backlog Total number of messages pending across ` + + `all durable mailboxes. +# TYPE waved_mailbox_backlog gauge +waved_mailbox_backlog 0 +` + err := testutil.CollectAndCompare( + c, strings.NewReader(want), + "waved_mailbox_backlog", "waved_mailbox_depth", + ) + require.NoError(t, err) + }) + + t.Run("query error suppresses depth gauges", func(t *testing.T) { + t.Parallel() + + q := &fullMockQuerier{ + depthsErr: errors.New("store cannot report depth"), + } + c := NewSystemCollector(q, fn.None[btclog.Logger]()) + + require.Zero( + t, testutil.CollectAndCount( + c, "waved_mailbox_backlog", + "waved_mailbox_depth", + ), + ) + }) +} + // TestRegisterAllIdempotent verifies RegisterAll tolerates duplicate // registration on the same registry without panicking, matching the // multi-daemon test-process invariant. diff --git a/oor/actor_messages.go b/oor/actor_messages.go index ae83556aa..a646adbf6 100644 --- a/oor/actor_messages.go +++ b/oor/actor_messages.go @@ -889,6 +889,18 @@ func (m *RestoreNonTerminalRequest) MessageType() string { return "RestoreNonTerminalRequest" } +// Priority marks the restore as a control message. Boot treats a failed +// restore Ask as fatal, so the message must bypass the registry mailbox's +// backlog admission check: a backlog past the hard watermark is exactly the +// condition the restore exists to work through, and refusing it would turn +// a backed-up registry into a daemon-wide restart crash loop. Control +// priority also claims the restore ahead of the redelivered backlog; both +// orders are safe (routing and restore converge on ensureChild's dedup), +// and restore-first re-admits in-flight sessions sooner. +func (m *RestoreNonTerminalRequest) Priority() int { + return actor.ControlPriority +} + // actorMsgSealed marks this as implementing the sealed ActorMsg interface. func (m *RestoreNonTerminalRequest) actorMsgSealed() {} diff --git a/oor/registry.go b/oor/registry.go index aae5f23b9..04e98bbf5 100644 --- a/oor/registry.go +++ b/oor/registry.go @@ -265,6 +265,13 @@ func NewOORRegistryActor(cfg OORRegistryConfig) (*OORRegistryActor, error) { ) durableCfg.Log = cfg.Log + // Bound the registry's durable backlog with the shared default + // watermarks. The ingress dispatch path classifies the resulting + // ErrMailboxSaturated as a deferral, so a backed-up registry stalls + // the serverconn cursor instead of deepening its own backlog. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark + // A deferred self-transfer hint redelivers on a long flat backoff and // never dead-letters: the terminal-reap redrive is the fast path, so // the durable copy is purely the crash-safety net, and the default diff --git a/oor/session_actor.go b/oor/session_actor.go index 7c8bdbcb9..417ad4546 100644 --- a/oor/session_actor.go +++ b/oor/session_actor.go @@ -214,6 +214,13 @@ func NewOORSessionActor(cfg SessionActorConfig) (*OORSessionActor, error) { ) durableCfg.Log = cfg.Log + // Bound the session's durable backlog with the shared default + // watermarks; a single OOR session that parks this many undelivered + // events is wedged, and shedding at the producer is the only move + // that helps it. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark + durable, err := actor.NewDurableActor(durableCfg).Unpack() if err != nil { return nil, err diff --git a/serverconn/AGENTS.md b/serverconn/AGENTS.md index a88da6d0f..7be7e607b 100644 --- a/serverconn/AGENTS.md +++ b/serverconn/AGENTS.md @@ -66,6 +66,17 @@ background ingress polling with event routing. ## Invariants - Ack watermark only advances AFTER durable local dispatch commit (prevents message loss on crash). +- **Dispatch deferral covers both refusal shapes.** `deliverToActor` + classifies `actor.ErrMailboxFull` (bounded in-memory target) AND + `actor.ErrMailboxSaturated` (durable target past its hard backlog + watermark) as `ErrDispatchDeferred`: the envelope stays unacknowledged, + the cursor stalls, and the loop re-pulls after a backoff. A backed-up + durable actor therefore exerts backpressure on the operator stream + instead of deepening its own backlog. The egress `DurableActor` itself + carries the shared default watermarks (`actor.DefaultSoftHighWatermark` / + `DefaultHardHighWatermark`), so producers Telling into egress see + `ErrMailboxSaturated` once the operator has been unreachable long enough + to park that many undelivered events. - The ingress fold never holds the database writer across network IO. `runFoldedDispatch` runs waiter-backed responses and the `ConnectorConfig.NonTxRoutes` requests BEFORE opening the write transaction; only durable enqueues and the cursor checkpoint go inside it. A route is hoisted only when it is listed in `NonTxRoutes` AND the envelope is a `KIND_REQUEST`, so a durable actor `Tell` can never escape the fold. An envelope of any other kind arriving on a marked route is skip-warned by `dispatchBatch` rather than dispatched, because the mux bridge ignores `env.Rpc.Kind` and would otherwise serve a sender-mislabeled envelope over the network with the writer held. Any new dispatcher that terminates in `Edge.Send` rather than a durable enqueue MUST be added to `NonTxRoutes` at wiring time (see `waved.Server.buildRPCDispatchers`), otherwise it pins the SQLite global writer lock (production opens with `_txlock=immediate`) or a SERIALIZABLE Postgres snapshot for the length of a round trip to the operator. - Pre-transaction dispatch happens before the commit, never after. A crash in between re-pulls the batch and redelivers, which is the at-least-once contract; committing first would advance the cursor past a request that was never answered. - Unary RPC responses use in-memory registry first; if no waiter exists (crash replay), the ingress falls back to durable EventRouter dispatch. The ResponseRegistry returns a tri-state delivery result (waiter/buffered/dropped) so the ingress knows whether to route durably. diff --git a/serverconn/CLAUDE.md b/serverconn/CLAUDE.md index a88da6d0f..7be7e607b 100644 --- a/serverconn/CLAUDE.md +++ b/serverconn/CLAUDE.md @@ -66,6 +66,17 @@ background ingress polling with event routing. ## Invariants - Ack watermark only advances AFTER durable local dispatch commit (prevents message loss on crash). +- **Dispatch deferral covers both refusal shapes.** `deliverToActor` + classifies `actor.ErrMailboxFull` (bounded in-memory target) AND + `actor.ErrMailboxSaturated` (durable target past its hard backlog + watermark) as `ErrDispatchDeferred`: the envelope stays unacknowledged, + the cursor stalls, and the loop re-pulls after a backoff. A backed-up + durable actor therefore exerts backpressure on the operator stream + instead of deepening its own backlog. The egress `DurableActor` itself + carries the shared default watermarks (`actor.DefaultSoftHighWatermark` / + `DefaultHardHighWatermark`), so producers Telling into egress see + `ErrMailboxSaturated` once the operator has been unreachable long enough + to park that many undelivered events. - The ingress fold never holds the database writer across network IO. `runFoldedDispatch` runs waiter-backed responses and the `ConnectorConfig.NonTxRoutes` requests BEFORE opening the write transaction; only durable enqueues and the cursor checkpoint go inside it. A route is hoisted only when it is listed in `NonTxRoutes` AND the envelope is a `KIND_REQUEST`, so a durable actor `Tell` can never escape the fold. An envelope of any other kind arriving on a marked route is skip-warned by `dispatchBatch` rather than dispatched, because the mux bridge ignores `env.Rpc.Kind` and would otherwise serve a sender-mislabeled envelope over the network with the writer held. Any new dispatcher that terminates in `Edge.Send` rather than a durable enqueue MUST be added to `NonTxRoutes` at wiring time (see `waved.Server.buildRPCDispatchers`), otherwise it pins the SQLite global writer lock (production opens with `_txlock=immediate`) or a SERIALIZABLE Postgres snapshot for the length of a round trip to the operator. - Pre-transaction dispatch happens before the commit, never after. A crash in between re-pulls the batch and redelivers, which is the at-least-once contract; committing first would advance the cursor past a request that was never answered. - Unary RPC responses use in-memory registry first; if no waiter exists (crash replay), the ingress falls back to durable EventRouter dispatch. The ResponseRegistry returns a tri-state delivery result (waiter/buffered/dropped) so the ingress knows whether to route durably. diff --git a/serverconn/dispatch_deferral.go b/serverconn/dispatch_deferral.go index d1aa8083c..92c959d07 100644 --- a/serverconn/dispatch_deferral.go +++ b/serverconn/dispatch_deferral.go @@ -12,10 +12,11 @@ import ( ) // ErrDispatchDeferred marks an inbound envelope that could not be handed to its -// target actor because the target's in-memory mailbox was full. It is not a -// dispatch failure: the envelope is intact, unacknowledged, and the ingress -// loop re-pulls it after a backoff, so the cursor never passes an event that -// was not delivered. +// target actor because the target refused it for want of room: a full +// in-memory mailbox, or a durable mailbox whose backlog crossed its hard +// watermark. It is not a dispatch failure: the envelope is intact, +// unacknowledged, and the ingress loop re-pulls it after a backoff, so the +// cursor never passes an event that was not delivered. // // The error exists so the ingress loop can tell backpressure apart from a real // dispatch error. Both back off, but only backpressure is expected to clear on @@ -39,8 +40,9 @@ type deferredDispatchError struct { // also the cursor the next pull resumes from. eventSeq uint64 - // err is the underlying delivery error, always wrapping - // actor.ErrMailboxFull. + // err is the underlying delivery error, always wrapping either + // actor.ErrMailboxFull (bounded in-memory target) or + // actor.ErrMailboxSaturated (durable target past its hard watermark). err error } @@ -52,7 +54,8 @@ func (e *deferredDispatchError) Error() string { // Unwrap exposes both the sentinel and the underlying mailbox error, so // errors.Is matches ErrDispatchDeferred for the loop's classification and -// actor.ErrMailboxFull for anything that cares about the cause. +// the concrete cause (actor.ErrMailboxFull or actor.ErrMailboxSaturated) +// for anything that cares which bound turned the envelope away. func (e *deferredDispatchError) Unwrap() []error { return []error{ErrDispatchDeferred, e.err} } @@ -75,7 +78,11 @@ func deferDispatch(service, method, target string, eventSeq uint64, // parking the ingress goroutine. A bounded in-memory target is sent to with // TryTell and a full mailbox comes back as a deferral; a durable target keeps // the blocking Tell, which is bounded by its own write and has to stay inside -// the caller's transaction to commit atomically with the cursor. +// the caller's transaction to commit atomically with the cursor. A durable +// target with backlog watermarks configured can refuse that Tell with +// ErrMailboxSaturated, which classifies as the same deferral: the backlog is +// expected to drain on its own, and stalling the cursor at the undelivered +// envelope is exactly the backpressure the watermark asks for. // // This is the fix for the wedge that made a deployed client go deaf: the round // client and the incoming-VTXO handler are registered with the default @@ -99,7 +106,8 @@ func deliverToActor[M actor.Message, R any](ctx context.Context, } outsideTx, err := actor.TellWithoutParking[M](ctx, ref, msg).Unpack() - if errors.Is(err, actor.ErrMailboxFull) { + if errors.Is(err, actor.ErrMailboxFull) || + errors.Is(err, actor.ErrMailboxSaturated) { return deferDispatch(service, method, ref.ID(), eventSeq, err) } if err != nil { diff --git a/serverconn/doc.go b/serverconn/doc.go index 18cc5c3fe..79958643f 100644 --- a/serverconn/doc.go +++ b/serverconn/doc.go @@ -50,6 +50,14 @@ // only mailbox puller for as long as the target took to drain, with the write // transaction open. See deliverToActor. // +// Durable targets refuse for a different reason: a durable mailbox has no +// in-memory capacity, but one configured with backlog watermarks turns a Tell +// away with ErrMailboxSaturated once its persistent backlog crosses the hard +// watermark. That refusal classifies as the same deferral, so a durable actor +// whose consumer has fallen ten thousand messages behind stalls the cursor +// instead of deepening its backlog — the same backpressure shape as a full +// in-memory mailbox, at a bound measured in rows instead of channel slots. +// // Known residual: one strictly-ordered cursor feeds every inbound route, so an // actor that stops draining still stops delivery on ALL of them. What the // deferral changes is the blast radius and the diagnosis — the database writer diff --git a/serverconn/runtime.go b/serverconn/runtime.go index 267af6048..e80b3533a 100644 --- a/serverconn/runtime.go +++ b/serverconn/runtime.go @@ -79,6 +79,14 @@ func NewRuntime(cfg ConnectorConfig) (*Runtime, error) { // historical single-sender behavior for callers that leave it unset. durableCfg.NumWorkers = cfg.EgressWorkers + // Bound the egress backlog with the shared default watermarks: an + // operator that stays unreachable long enough to park ten thousand + // undelivered events is better served by producers failing loudly + // (and their retry policies backing off) than by a backlog that grows + // for as long as the outage lasts. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark + // A permanent version error is not retryable: dead-letter the failing // durable message immediately instead of retrying it forever. All other // (transient) failures keep the default exponential-backoff policy. diff --git a/serverconn/types.go b/serverconn/types.go index 7b8cb3987..930b24fe2 100644 --- a/serverconn/types.go +++ b/serverconn/types.go @@ -58,10 +58,11 @@ const ackStateType = mailboxconn.CheckpointStateType // that captures a ServiceKey reference for the target actor. // // An error wrapping ErrDispatchDeferred is the one non-failure a dispatcher may -// return: the target's in-memory mailbox had no room. The envelope is untouched -// and unacknowledged, and the ingress loop re-pulls it after a backoff. A -// dispatcher must never park waiting for room, because one goroutine dispatches -// every inbound route in the process. +// return: the target refused the envelope for want of room, either a full +// in-memory mailbox or a durable mailbox past its hard backlog watermark. The +// envelope is untouched and unacknowledged, and the ingress loop re-pulls it +// after a backoff. A dispatcher must never park waiting for room, because one +// goroutine dispatches every inbound route in the process. type EnvelopeDispatcher func( ctx context.Context, env *mailboxpb.Envelope, ) error diff --git a/unroll/actor.go b/unroll/actor.go index 13f6c30d4..d95eb8ae2 100644 --- a/unroll/actor.go +++ b/unroll/actor.go @@ -126,6 +126,12 @@ func NewVTXOUnrollActor(cfg Config) (*VTXOUnrollActor, error) { ) durableCfg.Log = cfg.Log + // Bound the unroll actor's durable backlog with the shared default + // watermarks so a wedged consumer sheds load at the producer instead + // of growing its backlog without bound. + durableCfg.SoftHighWatermark = actor.DefaultSoftHighWatermark + durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark + durable, err := actor.NewDurableActor(durableCfg).Unpack() if err != nil { return nil, err diff --git a/unroll/messages.go b/unroll/messages.go index 09e3bd00d..6d32e22ff 100644 --- a/unroll/messages.go +++ b/unroll/messages.go @@ -270,6 +270,16 @@ func (m *ResumeUnrollRequest) MessageType() string { return "ResumeUnrollRequest" } +// Priority marks the resume as a control message. Boot-time restore treats a +// failed resume Ask as fatal, so the message must bypass the mailbox's +// backlog admission check: a backlog past the hard watermark is exactly the +// condition the resume exists to drain, and refusing it would turn a wedged +// consumer into a daemon-wide restart crash loop. Control priority also +// claims ahead of the redelivered backlog, mirroring RestartMessage. +func (m *ResumeUnrollRequest) Priority() int { + return actor.ControlPriority +} + // TLVType returns the durable mailbox type ID. func (m *ResumeUnrollRequest) TLVType() tlv.Type { return resumeUnrollRequestTLVType diff --git a/waved/metrics.go b/waved/metrics.go index a0fe5f07c..c5af44bc3 100644 --- a/waved/metrics.go +++ b/waved/metrics.go @@ -255,3 +255,32 @@ func (a *systemStatsAdapter) GetRoundStatsByStatus(ctx context.Context) ( return a.srv.rpcServer.liveRoundsByStatus(ctx) } + +// GetMailboxDepths returns the pending backlog of every durable mailbox +// currently holding messages, straight off the delivery store. The depth +// surface is optional on the store interface, so a store without it skips +// the gauges rather than failing the scrape. +func (a *systemStatsAdapter) GetMailboxDepths(ctx context.Context) ( + []metrics.MailboxDepthRow, error) { + + depthStore, ok := a.srv.deliveryStore.(actor.MailboxDepthStore) + if !ok { + return nil, fmt.Errorf("delivery store cannot report mailbox " + + "depth") + } + + depths, err := depthStore.MailboxDepths(ctx) + if err != nil { + return nil, err + } + + rows := make([]metrics.MailboxDepthRow, 0, len(depths)) + for _, d := range depths { + rows = append(rows, metrics.MailboxDepthRow{ + MailboxID: d.MailboxID, + Depth: d.Depth, + }) + } + + return rows, nil +}