actor: durable mailbox backpressure watermarks + depth metrics

baselib/actor/AGENTS.md

@@ -58,6 +58,38 @@ crash-safe at-least-once delivery with exactly-once deduplication.
   false for Tells, DurableAsks, and redelivered asks whose caller is gone.
 - `ChannelMailbox[M, R]` — In-memory channel-based mailbox (non-durable, for lightweight actors).
 - `Mailbox[M, R]` — Interface for actor message queues: `Send(ctx, env) error` (blocking; returns `ErrMailboxClosed`, `ErrActorTerminated`, or a context error on failure), `TrySend(env) error` (non-blocking), `Receive(ctx) iter.Seq[envelope]`, `Close()`, `IsClosed() bool`, `Drain() iter.Seq[envelope]`.
+- `ErrMailboxSaturated` / backlog watermarks — The durable analogue of
+  `ErrMailboxFull`. `DurableMailboxConfig.SoftHighWatermark` /
+  `HardHighWatermark` (flowing through `DurableActorConfig`) bound the
+  persistent backlog: past the soft mark the mailbox logs one warning per
+  breach episode, at or past the hard mark `Send` refuses with
+  `ErrMailboxSaturated` BEFORE encoding or promise registration (so
+  `TrySend`/`TryTell` inherit the check and a refusal needs no cleanup).
+  Both default to 0 (disabled); consumer sites opt in with
+  `DefaultSoftHighWatermark` (1000) / `DefaultHardHighWatermark` (10000).
+  Two exemptions: priority `>= ControlPriority` (restart plus the
+  boot-critical restore/resume messages — `RestoreNonTerminalRequest`,
+  `ResumeUnrollRequest`, `ResumeCreditOpRequest` — whose refusal would be a
+  fatal-boot crash loop), and outbox-propagated deliveries (detected via
+  the outbox-ID context marker; the publisher's claim path bumps attempts
+  in its own tx, so a refusal would dead-letter a committed CDC row rather
+  than shed load). Depth is read via a 1s TTL-cached, single-flighted probe
+  (run outside the mutex AND outside the sender's ambient tx via
+  `WithoutTx` — joining a SERIALIZABLE writer would take whole-partition
+  predicate locks) plus a local sent-since-probe delta (one-sided for local
+  traffic; remote sends invisible for up to one window). A probe failure
+  falls back to the cached estimate, or fails OPEN with no baseline. The
+  soft episode is evaluated before the hard refusal so the warning fires
+  even when the first probe already reads saturated. Known residual: an
+  in-turn Tell into a saturated peer fails the sender's turn and burns the
+  inbound message's attempts (postpone semantics are the planned fix).
+- `MailboxDepthStore` / `MailboxDepthCount` — Narrow, optional read surface
+  (`MailboxDepth(ctx, mailboxID)`, `MailboxDepths(ctx)`) discovered by type
+  assertion on the `DeliveryStore`, deliberately NOT embedded in it (test
+  doubles stay small). Backing reads are `COUNT(*)` of `mailbox_messages`
+  rows — leased rows included, rows are deleted on ack — so depth is exactly
+  the undelivered backlog. Consumed by the watermark check and the
+  `waved_mailbox_depth`/`waved_mailbox_backlog` scrape gauges.
 - `isExpectedShutdownErr(err) bool` — Internal helper that classifies errors as expected during teardown: context cancellation/deadline, closed DB handle ("sql: database is closed", "sql: connection is already closed", "use of closed network connection"). Used by the lease loop to demote shutdown-path failures to debug instead of warn-flooding test artifacts at itest tail.
 - `Message.CorrelationKey() string` — Per-message FIFO key consumed by the
   durable mailbox's claim path. Non-empty keys participate in per-key FIFO:

baselib/actor/CLAUDE.md

@@ -58,6 +58,38 @@ crash-safe at-least-once delivery with exactly-once deduplication.
   false for Tells, DurableAsks, and redelivered asks whose caller is gone.
 - `ChannelMailbox[M, R]` — In-memory channel-based mailbox (non-durable, for lightweight actors).
 - `Mailbox[M, R]` — Interface for actor message queues: `Send(ctx, env) error` (blocking; returns `ErrMailboxClosed`, `ErrActorTerminated`, or a context error on failure), `TrySend(env) error` (non-blocking), `Receive(ctx) iter.Seq[envelope]`, `Close()`, `IsClosed() bool`, `Drain() iter.Seq[envelope]`.
+- `ErrMailboxSaturated` / backlog watermarks — The durable analogue of
+  `ErrMailboxFull`. `DurableMailboxConfig.SoftHighWatermark` /
+  `HardHighWatermark` (flowing through `DurableActorConfig`) bound the
+  persistent backlog: past the soft mark the mailbox logs one warning per
+  breach episode, at or past the hard mark `Send` refuses with
+  `ErrMailboxSaturated` BEFORE encoding or promise registration (so
+  `TrySend`/`TryTell` inherit the check and a refusal needs no cleanup).
+  Both default to 0 (disabled); consumer sites opt in with
+  `DefaultSoftHighWatermark` (1000) / `DefaultHardHighWatermark` (10000).
+  Two exemptions: priority `>= ControlPriority` (restart plus the
+  boot-critical restore/resume messages — `RestoreNonTerminalRequest`,
+  `ResumeUnrollRequest`, `ResumeCreditOpRequest` — whose refusal would be a
+  fatal-boot crash loop), and outbox-propagated deliveries (detected via
+  the outbox-ID context marker; the publisher's claim path bumps attempts
+  in its own tx, so a refusal would dead-letter a committed CDC row rather
+  than shed load). Depth is read via a 1s TTL-cached, single-flighted probe
+  (run outside the mutex AND outside the sender's ambient tx via
+  `WithoutTx` — joining a SERIALIZABLE writer would take whole-partition
+  predicate locks) plus a local sent-since-probe delta (one-sided for local
+  traffic; remote sends invisible for up to one window). A probe failure
+  falls back to the cached estimate, or fails OPEN with no baseline. The
+  soft episode is evaluated before the hard refusal so the warning fires
+  even when the first probe already reads saturated. Known residual: an
+  in-turn Tell into a saturated peer fails the sender's turn and burns the
+  inbound message's attempts (postpone semantics are the planned fix).
+- `MailboxDepthStore` / `MailboxDepthCount` — Narrow, optional read surface
+  (`MailboxDepth(ctx, mailboxID)`, `MailboxDepths(ctx)`) discovered by type
+  assertion on the `DeliveryStore`, deliberately NOT embedded in it (test
+  doubles stay small). Backing reads are `COUNT(*)` of `mailbox_messages`
+  rows — leased rows included, rows are deleted on ack — so depth is exactly
+  the undelivered backlog. Consumed by the watermark check and the
+  `waved_mailbox_depth`/`waved_mailbox_backlog` scrape gauges.
 - `isExpectedShutdownErr(err) bool` — Internal helper that classifies errors as expected during teardown: context cancellation/deadline, closed DB handle ("sql: database is closed", "sql: connection is already closed", "use of closed network connection"). Used by the lease loop to demote shutdown-path failures to debug instead of warn-flooding test artifacts at itest tail.
 - `Message.CorrelationKey() string` — Per-message FIFO key consumed by the
   durable mailbox's claim path. Non-empty keys participate in per-key FIFO:

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] {
 

baselib/actor/durable_actor.go

@@ -115,6 +115,19 @@ type DurableActorConfig[M TLVMessage, R any] struct {
 	// Default: 10.
 	MaxAttempts int
 
+	// SoftHighWatermark is the persistent backlog depth at which the
+	// actor's mailbox starts logging that its consumer is falling behind.
+	// Zero (the default) disables the warning. See
+	// DurableMailboxConfig.SoftHighWatermark.
+	SoftHighWatermark int
+
+	// HardHighWatermark is the persistent backlog depth at which sends to
+	// the actor's mailbox are refused with ErrMailboxSaturated. Zero (the
+	// default) disables the bound; control-priority messages and
+	// outbox-propagated deliveries are always exempt. See
+	// DurableMailboxConfig.HardHighWatermark.
+	HardHighWatermark int
+
 	// CleanupTimeout specifies the maximum duration for OnStop cleanup.
 	// Default: 5 seconds.
 	CleanupTimeout time.Duration
@@ -385,14 +398,16 @@ func NewDurableActor[M TLVMessage, R any](
 	// normalizes zero and inverted values, so an actor config that predates
 	// MaxPollInterval still lands on the default ceiling.
 	mailboxCfg := DurableMailboxConfig{
-		MailboxID:       cfg.ID,
-		Store:           cfg.Store,
-		Codec:           cfg.Codec,
-		Clock:           cfg.Clock,
-		LeaseDuration:   cfg.LeaseDuration,
-		PollInterval:    cfg.PollInterval,
-		MaxPollInterval: cfg.MaxPollInterval,
-		MaxAttempts:     cfg.MaxAttempts,
+		MailboxID:         cfg.ID,
+		Store:             cfg.Store,
+		Codec:             cfg.Codec,
+		Clock:             cfg.Clock,
+		LeaseDuration:     cfg.LeaseDuration,
+		PollInterval:      cfg.PollInterval,
+		MaxPollInterval:   cfg.MaxPollInterval,
+		MaxAttempts:       cfg.MaxAttempts,
+		SoftHighWatermark: cfg.SoftHighWatermark,
+		HardHighWatermark: cfg.HardHighWatermark,
 
 		// Size the wake channel to the worker count so a burst of
 		// enqueues can rouse every idle worker at once.

baselib/actor/durable_mailbox.go

@@ -147,6 +147,27 @@ type DurableMailboxConfig struct {
 	// depends on a wake because the poll ticker is the fallback.
 	WakeBuffer int
 
+	// SoftHighWatermark is the persistent backlog depth at which the
+	// mailbox starts logging that its consumer is falling behind. Sends
+	// still succeed. Zero (the default) disables the soft warning. The
+	// check only runs when Store implements MailboxDepthStore; a value
+	// above HardHighWatermark is lowered to it at construction.
+	SoftHighWatermark int
+
+	// HardHighWatermark is the persistent backlog depth at which Send
+	// refuses new messages with ErrMailboxSaturated, shedding load at the
+	// producer instead of growing the backlog without bound. Zero (the
+	// default) disables the bound. Messages with priority >=
+	// ControlPriority (restart and boot restore/resume messages) are
+	// exempt so recovery always lands, as are outbox-propagated CDC
+	// deliveries (the message was already accepted at its true producer;
+	// refusing the hand-off would dead-letter a committed row instead of
+	// shedding load). Depth is read through a TTL-cached probe plus a
+	// local sent-since-probe delta, so enforcement is approximate within
+	// the probe window but the common send path never pays for an extra
+	// COUNT query.
+	HardHighWatermark int
+
 	// SingleWorkerLeaseless enables the leaseless peek consume path. When
 	// set, Receive claims the next message with a READ-only PeekNextMessage
 	// instead of the write-transaction LeaseNextMessage, and yields a
@@ -315,6 +336,14 @@ type DurableMailbox[M TLVMessage, R any] struct {
 	// actorCtx is the actor's lifecycle context.
 	actorCtx context.Context
 
+	// depthStore is non-nil when cfg.Store can report mailbox backlog
+	// depth, which is what the watermark admission check reads. Resolved
+	// once at construction so the send path pays no type assertion.
+	depthStore MailboxDepthStore
+
+	// depth caches the probed backlog depth for the watermark check.
+	depth depthProbe
+
 	// promiseRegistry maps message IDs to in-flight promises for Ask
 	// messages. This allows the delivery to complete the promise after
 	// processing.
@@ -342,6 +371,13 @@ func NewDurableMailbox[M TLVMessage, R any](
 		cfg.PollInterval, cfg.MaxPollInterval,
 	)
 
+	// Resolve the watermark pair the same way: a soft warning above the
+	// hard refusal would make failing sends the operator's first signal,
+	// so it is lowered to the hard value instead.
+	cfg.SoftHighWatermark, cfg.HardHighWatermark = normalizeWatermarks(
+		cfg.SoftHighWatermark, cfg.HardHighWatermark,
+	)
+
 	m := &DurableMailbox[M, R]{
 		cfg:             cfg,
 		clock:           cfg.Clock.UnwrapOr(clock.NewDefaultClock()),
@@ -365,6 +401,13 @@ func NewDurableMailbox[M TLVMessage, R any](
 		)
 	}
 
+	// Resolve the depth surface once so the watermark check on the send
+	// path is a nil test rather than a per-send type assertion. A store
+	// without the surface simply runs without watermarks.
+	if depthStore, ok := cfg.Store.(MailboxDepthStore); ok {
+		m.depthStore = depthStore
+	}
+
 	return m
 }
 
@@ -420,6 +463,20 @@ func (m *DurableMailbox[M, R]) Send(ctx context.Context,
 		return ErrMailboxClosed
 	}
 
+	// Determine priority before the watermark check so restart-priority
+	// messages can be exempted from admission control.
+	priority := 0
+	if pm, ok := any(env.message).(PriorityMessage); ok {
+		priority = pm.Priority()
+	}
+
+	// Admit or refuse the send against the configured backlog watermarks
+	// BEFORE encoding or registering a promise, so a refusal needs no
+	// cleanup and TrySend/TryTell inherit the check for free.
+	if err := m.checkWatermarks(ctx, priority); err != nil {
+		return err
+	}
+
 	payload, err := m.cfg.Codec.Encode(env.message)
 	if err != nil {
 		return fmt.Errorf("encode mailbox message: %w", err)
@@ -448,12 +505,6 @@ func (m *DurableMailbox[M, R]) Send(ctx context.Context,
 		m.promiseRegistryMu.Unlock()
 	}
 
-	// Determine priority.
-	priority := 0
-	if pm, ok := any(env.message).(PriorityMessage); ok {
-		priority = pm.Priority()
-	}
-
 	// Enqueue the message. CorrelationKey opts the message into the
 	// per-key FIFO claim lane in the durable mailbox; an empty key (the
 	// default on BaseMessage) means the message uses the existing global

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