onionmessage: use BackpressureMailbox for onion peer actors

diff --git a/onionmessage/actor.go b/onionmessage/actor.go
index e7e4357..bb5671f 100644
--- a/onionmessage/actor.go
+++ b/onionmessage/actor.go
@@ -10,9 +10,27 @@ import (
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/lnutils"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/queue"
 	"github.com/lightningnetwork/lnd/record"
 )
 
+const (
+	// DefaultOnionMailboxSize is the buffer capacity for per-peer onion
+	// message actor mailboxes.
+	DefaultOnionMailboxSize = 50
+
+	// DefaultMinREDThreshold is the queue depth at which Random Early
+	// Detection begins probabilistically dropping onion messages. Below
+	// this threshold no drops occur; above DefaultOnionMailboxSize all
+	// messages are dropped. Must be strictly less than
+	// DefaultOnionMailboxSize.
+	DefaultMinREDThreshold = 40
+)
+
+// Compile-time assertion: DefaultMinREDThreshold must be strictly less than
+// DefaultOnionMailboxSize. If this overflows, the constants are misconfigured.
+const _ = uint(DefaultOnionMailboxSize - DefaultMinREDThreshold - 1)
+
 // Request is a message sent to an OnionPeerActor when an onion message is
 // received from the peer. The actor processes the message through the full
 // onion message pipeline: decode, decrypt, route, and forward/deliver.
@@ -66,9 +84,12 @@ func NewOnionMessageServiceKey(
 // OnionActorFactory is a function that spawns a new OnionPeerActor for a
 // given peer within the actor system. The factory captures shared dependencies
 // (router, resolver, sender, dispatcher) and only requires per-peer parameters
-// at spawn time.
+// at spawn time. Callers may pass ActorOptions to customise the mailbox (size,
+// drop predicate, etc.) on a per-peer basis.
 type OnionActorFactory func(system *actor.ActorSystem,
-	peerPubKey [33]byte) (OnionPeerActorRef, error)
+	peerPubKey [33]byte,
+	opts ...actor.ActorOption[*Request, *Response]) (OnionPeerActorRef,
+	error)
 
 // OnionPeerActor handles the full onion message processing pipeline for a
 // specific peer connection. It decodes incoming onion messages, determines
@@ -202,13 +223,17 @@ func (a *OnionPeerActor) Receive(ctx context.Context,
 // NewOnionActorFactory creates a factory function that spawns OnionPeerActors
 // with shared dependencies. The returned factory captures the router,
 // resolver, peer sender, and update dispatcher, requiring only the actor
-// system and peer public key at spawn time.
+// system, peer public key, and optional per-peer ActorOptions at spawn time.
+//
+// Callers supply ActorOptions (mailbox factory, size overrides, etc.) via the
+// opts variadic so that backpressure policy can be customised per peer.
 func NewOnionActorFactory(router OnionRouter, resolver NodeIDResolver,
 	peerSender PeerMessageSender,
 	dispatcher OnionMessageUpdateDispatcher) OnionActorFactory {
 
-	return func(system *actor.ActorSystem,
-		peerPubKey [33]byte) (OnionPeerActorRef, error) {
+	return func(system *actor.ActorSystem, peerPubKey [33]byte,
+		opts ...actor.ActorOption[*Request, *Response],
+	) (OnionPeerActorRef, error) {
 
 		peerActor := &OnionPeerActor{
 			peerPubKey:       peerPubKey,
@@ -223,6 +248,7 @@ func NewOnionActorFactory(router OnionRouter, resolver NodeIDResolver,
 		)
 		actorRef, err := serviceKey.Spawn(
 			system, "onion-peer-actor-"+pubKeyHex, peerActor,
+			opts...,
 		)
 		if err != nil {
 			return nil, err
@@ -235,6 +261,55 @@ func NewOnionActorFactory(router OnionRouter, resolver NodeIDResolver,
 	}
 }
 
+// DefaultOnionActorOpts returns ActorOptions that configure a
+// BackpressureMailbox with a RED drop predicate and the default onion mailbox
+// size. The RED thresholds are derived from the mailbox capacity so that all
+// parameters are centralised and self-consistent.
+func DefaultOnionActorOpts() []actor.ActorOption[*Request, *Response] {
+	factory := func(ctx context.Context,
+		capacity int) actor.Mailbox[*Request, *Response] {
+
+		// Dynamically calculate the min threshold to be
+		// the same proportion (40/50 = 80%) of the actual
+		// capacity.
+		minThreshold := (capacity * DefaultMinREDThreshold) /
+			DefaultOnionMailboxSize
+
+		// Ensure minThreshold is strictly less than
+		// capacity for RED to work.
+		if minThreshold >= capacity {
+			minThreshold = capacity - 1
+		}
+		if minThreshold < 0 {
+			minThreshold = 0
+		}
+
+		shouldDrop, err := queue.RandomEarlyDrop(
+			minThreshold, capacity,
+		)
+		if err != nil {
+			// This should never happen given the
+			// threshold clamping above, but fall back to
+			// dropping all messages rather than risking
+			// a blocked readHandler.
+			shouldDrop = func(int) bool {
+				return true
+			}
+		}
+
+		return actor.NewBackpressureMailbox[*Request, *Response](
+			ctx, capacity, shouldDrop,
+		)
+	}
+
+	return []actor.ActorOption[*Request, *Response]{
+		actor.WithMailboxFactory(factory),
+		actor.WithMailboxSize[*Request, *Response](
+			DefaultOnionMailboxSize,
+		),
+	}
+}
+
 // StopOnionActor stops the onion peer actor for the given public key using the
 // provided actor reference. This should be called when a peer disconnects to
 // clean up the actor.
diff --git a/peer/brontide.go b/peer/brontide.go
index ae8c377..e97568b 100644
--- a/peer/brontide.go
+++ b/peer/brontide.go
@@ -70,6 +70,10 @@ const (
 	// This MUST be a smaller value than the pingInterval.
 	pingTimeout = 30 * time.Second
 
+	// tellTimeout is the amount of time we will wait for a response to a
+	// tell.
+	tellTimeout = 30 * time.Second
+
 	// idleTimeout is the duration of inactivity before we time out a peer.
 	idleTimeout = 5 * time.Minute
 
@@ -311,6 +315,13 @@ type Config struct {
 	// message actor. If nil, onion messaging is disabled.
 	SpawnOnionActor onionmessage.OnionActorFactory
 
+	// OnionActorOpts returns ActorOptions for the onion peer actor
+	// being spawned for the given peer. This allows per-peer
+	// customization of mailbox size, drop predicates, etc.
+	OnionActorOpts func(peerPubKey [33]byte) []actor.ActorOption[
+		*onionmessage.Request, *onionmessage.Response,
+	]
+
 	// ActorSystem is the actor system tasked with managing actors.
 	ActorSystem *actor.ActorSystem
 
@@ -930,8 +941,25 @@ func (p *Brontide) Start() error {
 		p.log.Infof("Remote peer supports onion messages, " +
 			"spawning onion message actor")
 
+		// Fetch per-peer actor options. The OnionActorOpts
+		// callback is the extension point for per-peer
+		// customization of the drop predicate (e.g. choosing
+		// RED thresholds based on channel capacity or routing
+		// importance). Today the callback returns identical
+		// defaults for every peer; to differentiate, supply a
+		// callback that inspects peerPubKey and returns
+		// tailored options via
+		// onionmessage.DefaultOnionActorOpts with a
+		// peer-specific DropCheckFunc.
+		var opts []actor.ActorOption[
+			*onionmessage.Request, *onionmessage.Response,
+		]
+		if p.cfg.OnionActorOpts != nil {
+			opts = p.cfg.OnionActorOpts(p.PubKey())
+		}
+
 		ref, spawnErr := p.cfg.SpawnOnionActor(
-			p.cfg.ActorSystem, p.PubKey(),
+			p.cfg.ActorSystem, p.PubKey(), opts...,
 		)
 		if spawnErr != nil {
 			return fmt.Errorf("unable to spawn onion peer "+
@@ -2310,7 +2338,16 @@ out:
 					// TODO(elle): thread contexts through
 					// the peer system properly so that a
 					// parent context can be passed in here.
-					ctx := context.TODO()
+
+					// Use a timeout context to prevent
+					// the readHandler from blocking
+					// indefinitely if the actor's mailbox
+					// is full.
+					ctx, cancel := context.WithTimeout(
+						context.Background(),
+						tellTimeout,
+					)
+					defer cancel()
 
 					req := onionmessage.NewRequest(*msg)
 					ref.Tell(ctx, req)
diff --git a/server.go b/server.go
index 0e7fe48..b0e1f02 100644
--- a/server.go
+++ b/server.go
@@ -434,6 +434,14 @@ type server struct {
 	// each peer connection.
 	onionActorFactory onionmessage.OnionActorFactory
 
+	// defaultOnionActorOpts holds the default ActorOptions (backpressure
+	// mailbox with RED) applied to every onion peer actor. These are
+	// computed once during server start and returned by the per-peer
+	// OnionActorOpts callback.
+	defaultOnionActorOpts []actor.ActorOption[
+		*onionmessage.Request, *onionmessage.Response,
+	]
+
 	// txPublisher is a publisher with fee-bumping capability.
 	txPublisher *sweep.TxPublisher
 
@@ -2381,6 +2389,9 @@ func (s *server) Start(ctx context.Context) error {
 				s.sphinxOnionMsg, resolver, s,
 				s.onionMessageServer,
 			)
+
+			s.defaultOnionActorOpts = onionmessage.
+				DefaultOnionActorOpts()
 		}
 
 		cleanup = cleanup.add(s.chanStatusMgr.Stop)
@@ -4450,13 +4461,19 @@ func (s *server) peerConnected(conn net.Conn, connReq *connmgr.ConnReq,
 		RoutingPolicy:           s.cc.RoutingPolicy,
 		SphinxPayment:           s.sphinxPayment,
 		SpawnOnionActor:         s.onionActorFactory,
-		ActorSystem:             s.actorSystem,
-		WitnessBeacon:           s.witnessBeacon,
-		Invoices:                s.invoices,
-		ChannelNotifier:         s.channelNotifier,
-		HtlcNotifier:            s.htlcNotifier,
-		TowerClient:             towerClient,
-		DisconnectPeer:          s.DisconnectPeer,
+		OnionActorOpts: func(_ [33]byte) []actor.ActorOption[
+			*onionmessage.Request, *onionmessage.Response,
+		] {
+
+			return s.defaultOnionActorOpts
+		},
+		ActorSystem:     s.actorSystem,
+		WitnessBeacon:   s.witnessBeacon,
+		Invoices:        s.invoices,
+		ChannelNotifier: s.channelNotifier,
+		HtlcNotifier:    s.htlcNotifier,
+		TowerClient:     towerClient,
+		DisconnectPeer:  s.DisconnectPeer,
 		GenNodeAnnouncement: func(...netann.NodeAnnModifier) (
 			lnwire.NodeAnnouncement1, error) {