From a7e3ae1a8f9ea876a7705bbe4265aaa00ab8ecae Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:49:37 -0700 Subject: [PATCH 01/12] actor: add durable mailbox backpressure watermarks In this commit, we give the durable mailbox the capacity signal it never had. A channel mailbox bounds its queue with ErrMailboxFull, but a durable mailbox lands every Tell as a database row, so a consumer that stops draining grows its backlog for as long as the outage lasts with nothing pushing back on producers. Two thresholds on DurableMailboxConfig (flowing through DurableActorConfig) now bound the persistent backlog. Past the soft watermark the mailbox logs one warning per breach episode; at or past the hard watermark, Send refuses the message with the new ErrMailboxSaturated sentinel. The check runs before encoding or promise registration, so a refusal needs no cleanup and TrySend/TryTell inherit it for free. Messages with priority >= RestartPriority are always exempt: the RestartMessage that would un-wedge a stuck actor must not be refused by the very backlog it exists to drain. Depth is read through the new MailboxDepthStore surface, a narrow optional interface discovered by type assertion on the DeliveryStore (the same pattern keeps test doubles small). The read is TTL-cached at 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 deliberately one-sided: local sends push it up immediately while acks only surface at the next probe, and overshooting is the safe direction for an admission check. A failed probe fails OPEN, since a broken monitoring read must not become message loss. Both thresholds default to zero (disabled), so existing actors are byte-for-byte unchanged until a site opts in. --- baselib/actor/bounded_delivery.go | 13 +- baselib/actor/durable_actor.go | 30 +- baselib/actor/durable_mailbox.go | 59 +++- baselib/actor/interface.go | 31 ++- baselib/actor/mailbox_watermarks.go | 210 ++++++++++++++ baselib/actor/mailbox_watermarks_test.go | 339 +++++++++++++++++++++++ 6 files changed, 657 insertions(+), 25 deletions(-) create mode 100644 baselib/actor/mailbox_watermarks.go create mode 100644 baselib/actor/mailbox_watermarks_test.go 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..c87dba0e3 100644 --- a/baselib/actor/durable_actor.go +++ b/baselib/actor/durable_actor.go @@ -115,6 +115,18 @@ 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; restart-priority messages are always + // exempt. See DurableMailboxConfig.HardHighWatermark. + HardHighWatermark int + // CleanupTimeout specifies the maximum duration for OnStop cleanup. // Default: 5 seconds. CleanupTimeout time.Duration @@ -385,14 +397,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..4822b3ced 100644 --- a/baselib/actor/durable_mailbox.go +++ b/baselib/actor/durable_mailbox.go @@ -147,6 +147,23 @@ 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 >= + // RestartPriority are exempt so recovery always lands. 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 +332,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 +367,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 +397,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 +459,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 +501,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..6f8501e12 --- /dev/null +++ b/baselib/actor/mailbox_watermarks.go @@ -0,0 +1,210 @@ +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. Restart-priority messages are exempt so recovery always lands, 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, so the estimate only ever overshoots (sends from other + // processes are missed until the next probe, but so are acks, and + // acks are what shrink the backlog). 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: local sends push the estimate up +// immediately, while acks (and remote sends) only surface at the next probe. +// Overshooting is the safe direction for an admission check. +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 + + // 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 carries restart priority, 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 + } + + // Recovery and other framework-priority messages always land: a + // saturated mailbox usually means the actor is wedged or gone, and the + // RestartMessage that would un-wedge it must not be refused by the + // very backlog it exists to drain. + if priority >= RestartPriority { + return nil + } + + m.depth.mu.Lock() + defer m.depth.mu.Unlock() + + now := m.clock.Now() + if m.depth.probedAt.IsZero() || + now.Sub(m.depth.probedAt) >= depthProbeTTL { + + depth, err := m.depthStore.MailboxDepth(ctx, m.cfg.MailboxID) + if err != nil { + log := logger(m.actorCtx) + log.WarnS(ctx, "Mailbox depth probe failed, "+ + "admitting send unchecked", err, + slog.String("mailbox_id", m.cfg.MailboxID), + ) + + return nil + } + + m.depth.probedAt = now + m.depth.depth = depth + m.depth.sentSinceProbe = 0 + } + + estimate := m.depth.depth + m.depth.sentSinceProbe + + if hard > 0 && estimate >= int64(hard) { + return fmt.Errorf("mailbox %s backlog %d at hard watermark "+ + "%d: %w", m.cfg.MailboxID, estimate, hard, + ErrMailboxSaturated) + } + + // The soft watermark only logs, once per breach episode: the first + // send that pushes the estimate over it opens the episode, and the + // first checked send after the estimate falls back under it closes + // the episode. + 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 + } + + // 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..ccde7c83b --- /dev/null +++ b/baselib/actor/mailbox_watermarks_test.go @@ -0,0 +1,339 @@ +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 +} + +// MailboxDepth reports the configured depth and counts the probe. +func (d *depthReportingStore) MailboxDepth(_ context.Context, _ string) (int64, + error) { + + d.probes++ + 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) +} + +// 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) + }) + } +} From e0711c7fd2cfcf62902346e16d6ec8039230054b Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:49:50 -0700 Subject: [PATCH 02/12] db: add mailbox depth count queries In this commit, we add the two backlog reads behind the new watermark admission check and the depth scrape gauges. CountMailboxMessages is a COUNT(*) over one mailbox's rows, leased or not: rows are deleted on ack, so the count is exactly the undelivered backlog, and the prefix of idx_mailbox_messages_available covers the equality scan. CountMailboxMessagesByMailbox is the GROUP BY variant for scrape time; mailboxes with an empty backlog produce no row, so the result stays bounded by the number of backed-up actors rather than the number of actors that have ever existed. --- db/actordelivery/queries/mailbox.sql | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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. From dbde8f3092cde606cb8fe2feecfbd21a33f2a38e Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:49:50 -0700 Subject: [PATCH 03/12] db: regenerate sqlc stubs for mailbox depth queries In this commit, we regenerate the actor delivery query layer via make sqlc to pick up CountMailboxMessages and CountMailboxMessagesByMailbox. No handwritten changes. --- db/actordelivery/sqlc/mailbox.sql.go | 55 ++++++++++++++++++++++++++++ db/actordelivery/sqlc/querier.go | 10 +++++ 2 files changed, 65 insertions(+) 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. From b4c0a3f7425743745e40080bf7931869a891c8ab Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:49:50 -0700 Subject: [PATCH 04/12] db: implement actor.MailboxDepthStore on the delivery store In this commit, we wire the depth queries into the Store as the actor.MailboxDepthStore surface: MailboxDepth for the watermark admission check and MailboxDepths for the scrape gauges, both plain read transactions mapped through the widened ActorDeliveryQueries interface. A compile-time assertion pins the implementation so the durable mailbox's type assertion can never silently stop matching. The tests pin the two facts the watermark check depends on: a leased (in-flight) row still counts toward depth until its ack deletes it, and a fully drained mailbox drops out of the grouped listing entirely. --- db/actordelivery/mailbox_depth_test.go | 124 +++++++++++++++++++++++++ db/actordelivery/store_impl.go | 60 ++++++++++++ 2 files changed, 184 insertions(+) create mode 100644 db/actordelivery/mailbox_depth_test.go 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/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) From c7b7bc07fe1da7542614a9f4f7147af8fb5f8ce6 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:50:03 -0700 Subject: [PATCH 05/12] metrics+waved: export durable mailbox depth scrape gauges In this commit, we surface the backlog the watermarks bound. The SystemCollector gains two scrape-time gauges read off the delivery store's MailboxDepthStore surface: waved_mailbox_backlog, an unlabelled total that emits an explicit zero when every mailbox is drained (so "all clear" and "scrape broke" stay distinguishable), and waved_mailbox_depth{mailbox_id}, one series per mailbox currently holding messages. Reporting only backed-up mailboxes keeps cardinality proportional to live trouble: per-session actor IDs never accumulate as permanent series, the same posture the OOR/round gauges take. The waved systemStatsAdapter resolves the depth surface with a type assertion, so a delivery store without it skips the gauges rather than failing the scrape. The README gains the reference rows plus an alerting recipe keyed to the soft/hard watermark defaults, and the ingress deferral counter's description now points at waved_mailbox_depth to tell a durable saturation refusal apart from a full in-memory mailbox. --- metrics/README.md | 19 +++++++- metrics/collector.go | 64 ++++++++++++++++++++++++++ metrics/collector_test.go | 95 +++++++++++++++++++++++++++++++++++++++ waved/metrics.go | 29 ++++++++++++ 4 files changed, 206 insertions(+), 1 deletion(-) 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/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 +} From 5d0e7c0962d0a6459da3c9fcdb94b62203b1eed8 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:50:03 -0700 Subject: [PATCH 06/12] serverconn: classify mailbox saturation as dispatch deferral In this commit, we teach the ingress dispatch path that a durable target can refuse for want of room. deliverToActor now classifies actor.ErrMailboxSaturated exactly like actor.ErrMailboxFull: the envelope comes back as ErrDispatchDeferred, unacknowledged, and the loop re-pulls it after a backoff. A durable actor whose backlog crossed its hard watermark therefore stalls the cursor and exerts backpressure on the operator stream instead of deepening its own backlog, which is precisely the behavior the watermark asks for. The deferral machinery downstream needs no change: the episode logging, the deferred counter, and the redrive loop all key off the ErrDispatchDeferred sentinel. We update the dispatch contract docs (doc.go, EnvelopeDispatcher) to name the second refusal shape. --- serverconn/dispatch_deferral.go | 26 +++++++++++++++++--------- serverconn/doc.go | 8 ++++++++ serverconn/types.go | 9 +++++---- 3 files changed, 30 insertions(+), 13 deletions(-) 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/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 From 58a4aa864c58b63eb142f5c57c702b357e2b265c Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:50:14 -0700 Subject: [PATCH 07/12] multi: enable backlog watermarks on the durable actor fleet In this commit, we opt the six durable actor construction sites into the shared default watermarks (soft 1000, hard 10000): the serverconn egress sender, the OOR registry and per-session actors, and the ledger, credit, and unroll actors. The framework default stays disabled, so the diff is the complete inventory of which mailboxes now shed load. The consequences differ by who produces into each mailbox. Ingress-fed actors (the OOR registry and sessions) get deferral semantics: the serverconn cursor stalls at the refused envelope and re-pulls, so saturation becomes backpressure on the operator stream. Locally-fed actors surface ErrMailboxSaturated to their producers, which at ten thousand parked messages is the only honest answer. For egress, an operator unreachable long enough to park that many events now fails producers loudly instead of accreting an unbounded replay queue. --- credit/op_actor.go | 6 ++++++ ledger/actor.go | 7 +++++++ oor/registry.go | 7 +++++++ oor/session_actor.go | 7 +++++++ serverconn/runtime.go | 8 ++++++++ unroll/actor.go | 6 ++++++ 6 files changed, 41 insertions(+) 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/ledger/actor.go b/ledger/actor.go index 1bdb7016a..9b2f5bc8f 100644 --- a/ledger/actor.go +++ b/ledger/actor.go @@ -384,6 +384,13 @@ func (a *LedgerActor) Start(ctx context.Context) error { ]( a.actorID, a, a.bindStores, a.cfg.DeliveryStore, codec, ) + + // Bound the ledger'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 fmt.Errorf("build ledger durable actor: %w", err) 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/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/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 From 765d424678e4ea195111c679928084251d738576 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 14:50:14 -0700 Subject: [PATCH 08/12] docs: document the backpressure watermark contract In this commit, we add a Backpressure Watermarks section to the durable actor architecture doc covering the semantics (soft warns, hard refuses with ErrMailboxSaturated, restart priority exempt), the TTL-cached probe and its one-sided estimate, who refuses and what each caller class does with it, and the deliberate exemption of the OutboxPublisher folded delivery path. The per-package CLAUDE/AGENTS pairs for baselib/actor, serverconn, metrics, and db/actordelivery pick up the new surfaces. --- baselib/actor/AGENTS.md | 21 ++++++++ baselib/actor/CLAUDE.md | 21 ++++++++ db/actordelivery/AGENTS.md | 7 ++- db/actordelivery/CLAUDE.md | 7 ++- docs/durable_actor_architecture.md | 79 ++++++++++++++++++++++++++++-- metrics/AGENTS.md | 6 ++- metrics/CLAUDE.md | 6 ++- serverconn/AGENTS.md | 11 +++++ serverconn/CLAUDE.md | 11 +++++ 9 files changed, 160 insertions(+), 9 deletions(-) diff --git a/baselib/actor/AGENTS.md b/baselib/actor/AGENTS.md index cb4a2129c..276ffae80 100644 --- a/baselib/actor/AGENTS.md +++ b/baselib/actor/AGENTS.md @@ -58,6 +58,27 @@ 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). + Priority `>= RestartPriority` is always exempt so recovery lands. Depth is + read via a 1s TTL-cached probe plus a local sent-since-probe delta + (one-sided: overshoots, never undershoots), and a probe failure fails + OPEN. The OutboxPublisher's folded delivery path bypasses `Send` and is + deliberately unthrottled. +- `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..276ffae80 100644 --- a/baselib/actor/CLAUDE.md +++ b/baselib/actor/CLAUDE.md @@ -58,6 +58,27 @@ 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). + Priority `>= RestartPriority` is always exempt so recovery lands. Depth is + read via a 1s TTL-cached probe plus a local sent-since-probe delta + (one-sided: overshoots, never undershoots), and a probe failure fails + OPEN. The OutboxPublisher's folded delivery path bypasses `Send` and is + deliberately unthrottled. +- `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/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/docs/durable_actor_architecture.md b/docs/durable_actor_architecture.md index 811aa4e26..70e139cf5 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,74 @@ 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. Messages with priority `>= RestartPriority` are always +exempt, because the `RestartMessage` that would un-wedge a stuck actor must +not be refused by the very backlog it exists to drain. 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 deliberately one-sided: local sends push it up +immediately, while acks and remote sends only surface at the next probe. +Overshooting is the safe direction for an admission check, and the +enforcement error is bounded by one probe window. + +### 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 deliberately unthrottled**: + it enqueues into the target mailbox inside the publisher's own write + transaction, bypassing `DurableMailbox.Send`, so CDC delivery is never + refused. Throttling it would only move the backlog from the target mailbox + to the outbox table while breaking the claim-expiry retry contract. + +### 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/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/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. From 82e5bb27e3fbfa7b5a2db1d0c9025b0b7fe22c1b Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 15:12:26 -0700 Subject: [PATCH 09/12] actor: exempt outbox hand-offs from watermarks, harden depth probe In this commit, we address three findings from the adversarial review of the watermark admission check. First, outbox-propagated deliveries are now exempt. The CDC hand-off was NOT bypassing DurableMailbox.Send as the original docs claimed: the publisher's deliver path goes ref.Tell -> mailbox.Send, so a saturated target refused committed outbox rows. Because ClaimOutboxBatch bumps delivery attempts in its own transaction, roughly ten refused claims would dead-letter the row (and any DurableAsk response it carries) instead of exerting backpressure. The publisher stamps the outbox ID into the context on every folded delivery, so the check keys off exactly that marker: an outbox-propagated send is admitted uncheck. Second, the exemption threshold drops from RestartPriority to the new ControlPriority (MaxInt32 - 1), making room for domain-level control messages, boot restores and resumes, that the daemon treats as fatal on failure. Without this, a backlog pinned at the hard mark at boot would refuse the restore Ask and turn a wedged consumer into a daemon-wide restart crash loop, with the durable backlog guaranteeing every subsequent boot fails the same way. Third, the probe itself is hardened: it now runs single-flighted outside the mutex (concurrent senders use the cached estimate instead of stacking behind one COUNT), and outside the sender's ambient transaction via WithoutTx. TransactionExecutor.ExecTx joins any ambient tx and ignores the read-only option, so the old probe executed its whole-mailbox COUNT inside the sender's SERIALIZABLE writer, taking predicate locks that manufacture rw-conflicts with the consumer's acks at precisely the moment the system is contended. A probe failure now falls back to the cached estimate rather than admitting unchecked, and only fails open when no baseline exists at all. The soft-watermark episode is also evaluated before the hard refusal, so a backlog that enters saturation within one probe window still fires the operator's early-warning line instead of failing sends silently first. --- baselib/actor/durable_actor.go | 5 +- baselib/actor/durable_mailbox.go | 12 ++- baselib/actor/mailbox_watermarks.go | 121 ++++++++++++++++------- baselib/actor/mailbox_watermarks_test.go | 82 ++++++++++++++- baselib/actor/restart.go | 10 ++ 5 files changed, 187 insertions(+), 43 deletions(-) diff --git a/baselib/actor/durable_actor.go b/baselib/actor/durable_actor.go index c87dba0e3..e49905430 100644 --- a/baselib/actor/durable_actor.go +++ b/baselib/actor/durable_actor.go @@ -123,8 +123,9 @@ type DurableActorConfig[M TLVMessage, R any] struct { // HardHighWatermark is the persistent backlog depth at which sends to // the actor's mailbox are refused with ErrMailboxSaturated. Zero (the - // default) disables the bound; restart-priority messages are always - // exempt. See DurableMailboxConfig.HardHighWatermark. + // 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. diff --git a/baselib/actor/durable_mailbox.go b/baselib/actor/durable_mailbox.go index 4822b3ced..9d5ef0bb4 100644 --- a/baselib/actor/durable_mailbox.go +++ b/baselib/actor/durable_mailbox.go @@ -158,10 +158,14 @@ type DurableMailboxConfig struct { // 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 >= - // RestartPriority are exempt so recovery always lands. 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. + // 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 diff --git a/baselib/actor/mailbox_watermarks.go b/baselib/actor/mailbox_watermarks.go index 6f8501e12..21f21cac2 100644 --- a/baselib/actor/mailbox_watermarks.go +++ b/baselib/actor/mailbox_watermarks.go @@ -15,8 +15,10 @@ import ( // 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. Restart-priority messages are exempt so recovery always lands, and a -// mailbox with no watermarks configured behaves exactly as before. +// 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 @@ -36,9 +38,11 @@ const ( // 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, so the estimate only ever overshoots (sends from other - // processes are missed until the next probe, but so are acks, and - // acks are what shrink the backlog). A one-second window keeps 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 ) @@ -75,9 +79,10 @@ type MailboxDepthStore interface { // 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: local sends push the estimate up -// immediately, while acks (and remote sends) only surface at the next probe. -// Overshooting is the safe direction for an admission check. +// 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 @@ -85,6 +90,12 @@ type depthProbe struct { // 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 @@ -99,9 +110,10 @@ type depthProbe struct { // 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 carries restart priority, 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. +// 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 @@ -114,11 +126,24 @@ func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, return nil } - // Recovery and other framework-priority messages always land: a - // saturated mailbox usually means the actor is wedged or gone, and the - // RestartMessage that would un-wedge it must not be refused by the - // very backlog it exists to drain. - if priority >= RestartPriority { + // 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 } @@ -126,37 +151,57 @@ func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, defer m.depth.mu.Unlock() now := m.clock.Now() - if m.depth.probedAt.IsZero() || - now.Sub(m.depth.probedAt) >= depthProbeTTL { + 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 + m.depth.mu.Unlock() + + depth, err := m.depthStore.MailboxDepth( + WithoutTx(ctx), m.cfg.MailboxID, + ) + + m.depth.mu.Lock() + m.depth.probing = false - depth, err := m.depthStore.MailboxDepth(ctx, m.cfg.MailboxID) if err != nil { log := logger(m.actorCtx) log.WarnS(ctx, "Mailbox depth probe failed, "+ - "admitting send unchecked", err, + "falling back to cached estimate", err, slog.String("mailbox_id", m.cfg.MailboxID), ) - - return nil + } else { + m.depth.probedAt = m.clock.Now() + m.depth.depth = depth + m.depth.sentSinceProbe = 0 } + } - m.depth.probedAt = now - m.depth.depth = depth - m.depth.sentSinceProbe = 0 + // 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 - if hard > 0 && estimate >= int64(hard) { - return fmt.Errorf("mailbox %s backlog %d at hard watermark "+ - "%d: %w", m.cfg.MailboxID, estimate, hard, - ErrMailboxSaturated) - } - - // The soft watermark only logs, once per breach episode: the first - // send that pushes the estimate over it opens the episode, and the - // first checked send after the estimate falls back under it closes - // the episode. + // 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 { @@ -182,6 +227,12 @@ func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, 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. diff --git a/baselib/actor/mailbox_watermarks_test.go b/baselib/actor/mailbox_watermarks_test.go index ccde7c83b..9a31a8051 100644 --- a/baselib/actor/mailbox_watermarks_test.go +++ b/baselib/actor/mailbox_watermarks_test.go @@ -27,13 +27,21 @@ type depthReportingStore struct { // 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(_ context.Context, _ string) (int64, - error) { +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 } @@ -151,6 +159,76 @@ func TestHardWatermarkRestartPriorityExempt(t *testing.T) { 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 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. From 95e773a01c95c38b96f28e715e4e8c2fc3811fdc Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 15:12:26 -0700 Subject: [PATCH 10/12] multi: mark boot restore and resume messages as control priority In this commit, we stamp actor.ControlPriority onto the three boot-path control messages that flow into watermarked durable mailboxes: RestoreNonTerminalRequest (OOR registry), ResumeUnrollRequest (per-target unroll actors), and ResumeCreditOpRequest (per-operation credit actors). All three were priority 0, and the first two are delivered as Asks whose failure waved treats as fatal at startup, so a backlog past the hard watermark at boot would have refused the very message that exists to work through that backlog and crash-looped the daemon. Control priority both bypasses the admission check and claims ahead of the redelivered backlog, mirroring RestartMessage's restart-first semantics. For OOR, restore-before-backlog is safe because routing and restore converge on ensureChild's dedup either way. --- credit/messages.go | 10 ++++++++++ oor/actor_messages.go | 12 ++++++++++++ unroll/messages.go | 10 ++++++++++ 3 files changed, 32 insertions(+) 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/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/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 From 14c26c62c58b73c874d720190faa30ddbbb219b6 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 15:12:26 -0700 Subject: [PATCH 11/12] docs: correct the watermark exemption contract In this commit, we fix the architecture doc's false claim that the OutboxPublisher folded path bypasses DurableMailbox.Send (it does not; the exemption is now real and implemented via the outbox-ID context marker), document the control-priority exemption and the boot messages that carry it, describe the hardened probe (single-flight, ambient-tx stripping), and state the known residual honestly: an in-turn Tell into a saturated peer fails the sender's turn and burns the inbound message's delivery attempts, with postpone semantics as the planned structural fix and the dead-letter tooling from #1119 as the interim recovery path. --- baselib/actor/AGENTS.md | 21 ++++++++--- baselib/actor/CLAUDE.md | 21 ++++++++--- docs/durable_actor_architecture.md | 57 ++++++++++++++++++++++-------- 3 files changed, 74 insertions(+), 25 deletions(-) diff --git a/baselib/actor/AGENTS.md b/baselib/actor/AGENTS.md index 276ffae80..2ada580bb 100644 --- a/baselib/actor/AGENTS.md +++ b/baselib/actor/AGENTS.md @@ -67,11 +67,22 @@ crash-safe at-least-once delivery with exactly-once deduplication. `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). - Priority `>= RestartPriority` is always exempt so recovery lands. Depth is - read via a 1s TTL-cached probe plus a local sent-since-probe delta - (one-sided: overshoots, never undershoots), and a probe failure fails - OPEN. The OutboxPublisher's folded delivery path bypasses `Send` and is - deliberately unthrottled. + 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 diff --git a/baselib/actor/CLAUDE.md b/baselib/actor/CLAUDE.md index 276ffae80..2ada580bb 100644 --- a/baselib/actor/CLAUDE.md +++ b/baselib/actor/CLAUDE.md @@ -67,11 +67,22 @@ crash-safe at-least-once delivery with exactly-once deduplication. `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). - Priority `>= RestartPriority` is always exempt so recovery lands. Depth is - read via a 1s TTL-cached probe plus a local sent-since-probe delta - (one-sided: overshoots, never undershoots), and a probe failure fails - OPEN. The OutboxPublisher's folded delivery path bypasses `Send` and is - deliberately unthrottled. + 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 diff --git a/docs/durable_actor_architecture.md b/docs/durable_actor_architecture.md index 70e139cf5..cb395a5c2 100644 --- a/docs/durable_actor_architecture.md +++ b/docs/durable_actor_architecture.md @@ -299,21 +299,41 @@ on ack, so the count is exactly the undelivered backlog): drains. Both default to zero (disabled) on the config structs; the bound is opted -into per actor. Messages with priority `>= RestartPriority` are always -exempt, because the `RestartMessage` that would un-wedge a stuck actor must -not be refused by the very backlog it exists to drain. 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. +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 deliberately one-sided: local sends push it up -immediately, while acks and remote sends only surface at the next probe. -Overshooting is the safe direction for an admission check, and the -enforcement error is bounded by one probe window. +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 @@ -326,11 +346,18 @@ enforcement error is bounded by one probe window. - **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 deliberately unthrottled**: - it enqueues into the target mailbox inside the publisher's own write - transaction, bypassing `DurableMailbox.Send`, so CDC delivery is never - refused. Throttling it would only move the backlog from the target mailbox - to the outbox table while breaking the claim-expiry retry contract. +- **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 From 8c98bcee36ecbc41db928520b12dc80602fd96b0 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Fri, 7 Aug 2026 16:03:46 -0700 Subject: [PATCH 12/12] ledger+actor: soft-only ledger watermark, keep probe delta on reset In this commit, we address the two findings from the bot review pass. The ledger actor drops its hard watermark and keeps only the soft one. Its producers are fire-and-forget Tells that log-and-continue on failure with no redrive path, so a hard ErrMailboxSaturated refusal there is not backpressure but a permanently missing accounting leg: unlike the in-turn transport sends (which nack and eventually surface in the dead-letter table), a refused ledger emission leaves no trace beyond one warning line, and the double-entry balance stays 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 depth gauge keep it observable. The probe reset also stops discarding racing sends: senders that increment the delta while the single-flighted COUNT runs enqueue their rows after the count, so zeroing sentSinceProbe on completion dropped them from the estimate, an undershoot in the unsafe direction. The reset now subtracts a pre-probe snapshot instead, keeping the racing increments and preserving the local no-undershoot property across the probe window. --- baselib/actor/mailbox_watermarks.go | 11 ++++++++++- ledger/AGENTS.md | 7 +++++++ ledger/CLAUDE.md | 7 +++++++ ledger/actor.go | 13 +++++++++---- 4 files changed, 33 insertions(+), 5 deletions(-) diff --git a/baselib/actor/mailbox_watermarks.go b/baselib/actor/mailbox_watermarks.go index 21f21cac2..a08fd192d 100644 --- a/baselib/actor/mailbox_watermarks.go +++ b/baselib/actor/mailbox_watermarks.go @@ -165,6 +165,15 @@ func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, // 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( @@ -183,7 +192,7 @@ func (m *DurableMailbox[M, R]) checkWatermarks(ctx context.Context, } else { m.depth.probedAt = m.clock.Now() m.depth.depth = depth - m.depth.sentSinceProbe = 0 + m.depth.sentSinceProbe -= preProbeSent } } 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 9b2f5bc8f..f8aef99c1 100644 --- a/ledger/actor.go +++ b/ledger/actor.go @@ -385,11 +385,16 @@ func (a *LedgerActor) Start(ctx context.Context) error { a.actorID, a, a.bindStores, a.cfg.DeliveryStore, codec, ) - // Bound the ledger's durable backlog with the shared default - // watermarks so a wedged consumer sheds load at the producer instead - // of growing its backlog without bound. + // 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 - durableCfg.HardHighWatermark = actor.DefaultHardHighWatermark durable, err := actor.NewDurableActor(durableCfg).Unpack() if err != nil {