Merge pull request #11090 from yyforyongyu/task-peer-manager

### docs/release-notes/release-notes-0.20.4.md
@@ -21,6 +21,15 @@
 
 # Bug Fixes
 
+* Peer connections [now rate limit inbound ping replies and bound outgoing
+  message queue growth](https://github.com/lightningnetwork/lnd/pull/11090),
+  preventing peer-controlled resource exhaustion.
+
+* Channel updates carrying [inbound fees now sign the same bytes that are
+  broadcast](https://github.com/lightningnetwork/lnd/pull/11090), preventing
+  remote signature failures. Forwarded updates also preserve unknown signed
+  TLV extensions.
+
 * Channel funding attempts [now return
   cleanly](https://github.com/lightningnetwork/lnd/pull/11035) when their
   pending wallet reservation is no longer present.

### docs/release-notes/release-notes-0.21.3.md
@@ -21,6 +21,15 @@
 
 # Bug Fixes
 
+* Peer connections [now rate limit inbound ping replies and bound outgoing
+  message queue growth](https://github.com/lightningnetwork/lnd/pull/11090),
+  preventing peer-controlled resource exhaustion.
+
+* Channel updates carrying [inbound fees now sign the same bytes that are
+  broadcast](https://github.com/lightningnetwork/lnd/pull/11090), preventing
+  remote signature failures. Forwarded updates also preserve unknown signed
+  TLV extensions.
+
 * Channel funding attempts [now return
   cleanly](https://github.com/lightningnetwork/lnd/pull/11035) when their
   pending wallet reservation is no longer present.

### lnwire/channel_update.go
@@ -177,13 +177,50 @@ func (a *ChannelUpdate1) Decode(r io.Reader, _ uint32) error {
 		a.InboundFee = tlv.SomeRecordT(inboundFee)
 	}
 
-	if len(tlvRecords) != 0 {
-		a.ExtraOpaqueData = tlvRecords
-	}
+	// Retain the complete stream, including its canonical empty-slice form,
+	// because legacy rows may store the inbound fee in both the opaque and
+	// typed fields.
+	a.ExtraOpaqueData = tlvRecords
 
 	return nil
 }
 
+// canonicalExtraData returns the encoded TLV stream without changing the
+// receiver. Decode deliberately retains the complete stream for compatibility
+// with persisted updates, so this method removes a retained inbound-fee record
+// before merging the typed field back into a fresh stream. The typed field wins
+// duplicate representations, while opaque-only bytes are cloned verbatim for
+// legacy compatibility.
+func (a *ChannelUpdate1) canonicalExtraData() ([]byte, error) {
+	// Only parse when replacing a retained fee. Cloning otherwise preserves
+	// arbitrary legacy extensions without sharing the receiver's storage.
+	if !a.InboundFee.IsSome() {
+		return bytes.Clone(a.ExtraOpaqueData), nil
+	}
+
+	inboundFee := a.InboundFee.Zero()
+	_, extraData, err := ParseAndExtractExtraData(
+		a.ExtraOpaqueData, &inboundFee,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("parse update extra data: %w", err)
+	}
+
+	recordProducers := make([]tlv.RecordProducer, 0, 1)
+	a.InboundFee.WhenSome(func(
+		fee tlv.RecordT[tlv.TlvType55555, Fee]) {
+
+		recordProducers = append(recordProducers, &fee)
+	})
+
+	encoded, err := MergeAndEncode(recordProducers, extraData, nil)
+	if err != nil {
+		return nil, fmt.Errorf("encode update extra data: %w", err)
+	}
+
+	return encoded, nil
+}
+
 // Encode serializes the target ChannelUpdate into the passed io.Writer
 // observing the protocol version specified.
 //
@@ -238,18 +275,13 @@ func (a *ChannelUpdate1) Encode(w *bytes.Buffer, pver uint32) error {
 		}
 	}
 
-	recordProducers := make([]tlv.RecordProducer, 0, 1)
-	a.InboundFee.WhenSome(func(fee tlv.RecordT[tlv.TlvType55555, Fee]) {
-		recordProducers = append(recordProducers, &fee)
-	})
-
-	err := EncodeMessageExtraData(&a.ExtraOpaqueData, recordProducers...)
+	extraData, err := a.canonicalExtraData()
 	if err != nil {
 		return err
 	}
 
 	// Finally, append any extra opaque data.
-	return WriteBytes(w, a.ExtraOpaqueData)
+	return WriteBytes(w, extraData)
 }
 
 // MsgType returns the integer uniquely identifying this message type on the
@@ -311,8 +343,15 @@ func (a *ChannelUpdate1) DataToSign() ([]byte, error) {
 		}
 	}
 
+	// Use Encode's canonical form so signatures cover the emitted bytes
+	// without modifying the caller's update.
+	extraData, err := a.canonicalExtraData()
+	if err != nil {
+		return nil, err
+	}
+
 	// Finally, append any extra opaque data.
-	if err := WriteBytes(buf, a.ExtraOpaqueData); err != nil {
+	if err := WriteBytes(buf, extraData); err != nil {
 		return nil, err
 	}
 

### lnwire/features.go
@@ -431,6 +431,13 @@ func (fv RawFeatureVector) IsEmpty() bool {
 	return len(fv.features) == 0
 }
 
+// NumFeatures returns the number of populated bits retained by the vector.
+// Callers that account for decoded memory can use the count without exposing
+// the internal map or allocating a separate bit slice.
+func (fv RawFeatureVector) NumFeatures() int {
+	return len(fv.features)
+}
+
 // OnlyContains determines whether only the specified feature bits are found.
 func (fv RawFeatureVector) OnlyContains(bits ...FeatureBit) bool {
 	if len(bits) != len(fv.features) {

### lnwire/message_test.go
@@ -183,6 +183,153 @@ func TestWriteMessage(t *testing.T) {
 	}
 }
 
+// makeChannelUpdateWithExtraData constructs an update whose known inbound-fee
+// record is represented by its typed field while an unknown record remains in
+// the opaque stream. Keeping the representations separate lets callers verify
+// that encoding merges them without taking ownership of the original bytes.
+func makeChannelUpdateWithExtraData(t *testing.T) *lnwire.ChannelUpdate1 {
+	t.Helper()
+
+	unknownValue := []byte{1, 2, 3}
+	unknownRecord := tlv.MakePrimitiveRecord(
+		tlv.Type(9), &unknownValue,
+	)
+	extraData, err := lnwire.EncodeRecords([]tlv.Record{unknownRecord})
+	require.NoError(t, err)
+
+	inboundFee := lnwire.Fee{
+		BaseFee: 11,
+		FeeRate: 22,
+	}
+
+	return &lnwire.ChannelUpdate1{
+		Signature: testNodeSig,
+		InboundFee: tlv.SomeRecordT(
+			tlv.NewRecordT[tlv.TlvType55555](inboundFee),
+		),
+		ExtraOpaqueData: extraData,
+	}
+}
+
+// TestChannelUpdateEncodePreservesReceiver verifies that encoding a channel
+// update does not replace its caller-owned opaque TLV bytes.
+func TestChannelUpdateEncodePreservesReceiver(t *testing.T) {
+	// Arrange an update with typed and opaque records, and snapshot the
+	// opaque bytes to detect both content and length changes.
+	update := makeChannelUpdateWithExtraData(t)
+	originalExtraData := bytes.Clone(update.ExtraOpaqueData)
+
+	// Act by encoding through the public wire-message method that
+	// previously rewrote ExtraOpaqueData in place.
+	var encoded bytes.Buffer
+	err := update.Encode(&encoded, 0)
+
+	// Assert that encoding succeeds and leaves the receiver-owned slice
+	// byte-for-byte unchanged for reuse by other peers or goroutines.
+	require.NoError(t, err)
+	require.Equal(t, originalExtraData, []byte(update.ExtraOpaqueData))
+}
+
+// TestChannelUpdateEncodePreservesUnknownTLV verifies that decoding and then
+// re-encoding an update retains unknown TLVs without duplicating known ones.
+func TestChannelUpdateEncodePreservesUnknownTLV(t *testing.T) {
+	// Arrange a canonical wire encoding containing a typed inbound fee and
+	// an unknown record, then decode it into the legacy typed-plus-opaque
+	// representation retained for persisted graph compatibility.
+	update := makeChannelUpdateWithExtraData(t)
+	var original bytes.Buffer
+	require.NoError(t, update.Encode(&original, 0))
+
+	var decoded lnwire.ChannelUpdate1
+	require.NoError(t, decoded.Decode(bytes.NewReader(original.Bytes()), 0))
+	require.True(t, decoded.InboundFee.IsSome())
+	require.NotEmpty(t, decoded.ExtraOpaqueData)
+
+	// Act by re-encoding the decoded update, which must reconcile the known
+	// record in both representations before constructing the TLV stream.
+	var reencoded bytes.Buffer
+	err := decoded.Encode(&reencoded, 0)
+
+	// Assert that reconciliation succeeds and produces the exact original
+	// wire bytes, proving the unknown record survived without duplication.
+	require.NoError(t, err)
+	require.Equal(t, original.Bytes(), reencoded.Bytes())
+}
+
+// TestChannelUpdateEncodePreservesOpaqueFee verifies compatibility with
+// callers that still represent the known inbound fee only as opaque data.
+func TestChannelUpdateEncodePreservesOpaqueFee(t *testing.T) {
+	// Arrange a canonical update, decode its complete stream, and clear the
+	// typed field to reproduce the legacy opaque-only construction shape.
+	update := makeChannelUpdateWithExtraData(t)
+	var original bytes.Buffer
+	require.NoError(t, update.Encode(&original, 0))
+
+	var opaqueOnly lnwire.ChannelUpdate1
+	require.NoError(t, opaqueOnly.Decode(
+		bytes.NewReader(original.Bytes()), 0,
+	))
+	opaqueOnly.InboundFee = tlv.OptionalRecordT[
+		tlv.TlvType55555, lnwire.Fee,
+	]{}
+
+	// Act by encoding without a typed fee, which clones the retained opaque
+	// stream unchanged so both record forms keep their original wire bytes.
+	var reencoded bytes.Buffer
+	err := opaqueOnly.Encode(&reencoded, 0)
+
+	// Assert the pass-through retains both the known fee and unknown record
+	// by producing the exact original wire bytes.
+	require.NoError(t, err)
+	require.Equal(t, original.Bytes(), reencoded.Bytes())
+}
+
+// TestChannelUpdateEncodeConcurrent verifies that one channel update can be
+// encoded concurrently without racing through receiver mutation.
+func TestChannelUpdateEncodeConcurrent(t *testing.T) {
+	// Arrange a shared update and canonical expected result. Each worker
+	// gets its own buffer so the only shared data is the message receiver.
+	update := makeChannelUpdateWithExtraData(t)
+	var expected bytes.Buffer
+	require.NoError(t, update.Encode(&expected, 0))
+
+	type encodeResult struct {
+		data []byte
+		err  error
+	}
+
+	const workerCount = 8
+	results := make(chan encodeResult, workerCount)
+	var workers sync.WaitGroup
+	workers.Add(workerCount)
+
+	// Act by encoding the shared receiver from independent goroutines. The
+	// buffered result channel gives every worker a shutdown path even if an
+	// encoding fails before the owner begins collecting results.
+	for range workerCount {
+		go func() {
+			defer workers.Done()
+
+			var encoded bytes.Buffer
+			err := update.Encode(&encoded, 0)
+			results <- encodeResult{
+				data: encoded.Bytes(),
+				err:  err,
+			}
+		}()
+	}
+
+	workers.Wait()
+	close(results)
+
+	// Assert in the owning test goroutine that every concurrent encoding
+	// succeeded and matched the same canonical bytes.
+	for result := range results {
+		require.NoError(t, result.err)
+		require.Equal(t, expected.Bytes(), result.data)
+	}
+}
+
 // BenchmarkWriteMessage benchmarks the performance of lnwire.WriteMessage. It
 // generates a test message for each of the lnwire.Message, calls the
 // WriteMessage method and benchmark it.

### lnwire/test_message.go
@@ -520,7 +520,8 @@ func (a *ChannelUpdate1) RandTestMessage(t *rapid.T) Message {
 		),
 		HtlcMaximumMsat: maxHtlc,
 		InboundFee:      inboundFee,
-		ExtraOpaqueData: extraBytes,
+		// Match Decode's empty shape for stable round-trip equality.
+		ExtraOpaqueData: append(ExtraOpaqueData{}, extraBytes...),
 	}
 }
 

### netann/channel_update_test.go
@@ -1,6 +1,7 @@
 package netann_test
 
 import (
+	"bytes"
 	"errors"
 	"testing"
 	"time"
@@ -11,6 +12,8 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/netann"
+	"github.com/lightningnetwork/lnd/tlv"
+	"github.com/stretchr/testify/require"
 )
 
 type mockSigner struct {
@@ -190,3 +193,48 @@ func TestUpdateDisableFlag(t *testing.T) {
 		})
 	}
 }
+
+// TestChannelUpdateSignaturePreservesExtraData verifies that signing and wire
+// encoding use the same non-mutating canonical TLV representation.
+func TestChannelUpdateSignaturePreservesExtraData(t *testing.T) {
+	// Arrange a typed inbound fee and a separate unknown TLV for signing.
+	// Signing this pre-encoding shape proves DataToSign includes the same
+	// canonical records that Encode will later put on the wire.
+	unknownValue := []byte{3, 2, 1}
+	unknownRecord := tlv.MakePrimitiveRecord(
+		tlv.Type(9), &unknownValue,
+	)
+	extraData, err := lnwire.EncodeRecords([]tlv.Record{unknownRecord})
+	require.NoError(t, err)
+
+	inboundFee := lnwire.Fee{
+		BaseFee: 33,
+		FeeRate: 44,
+	}
+	update := &lnwire.ChannelUpdate1{
+		InboundFee: tlv.SomeRecordT(
+			tlv.NewRecordT[tlv.TlvType55555](inboundFee),
+		),
+		ExtraOpaqueData: extraData,
+	}
+	require.NoError(t, netann.SignChannelUpdate(
+		netann.NewNodeSigner(privKeySigner), testKeyLoc, update,
+	))
+
+	var encoded bytes.Buffer
+	require.NoError(t, update.Encode(&encoded, 0))
+
+	var decoded lnwire.ChannelUpdate1
+	require.NoError(t, decoded.Decode(bytes.NewReader(encoded.Bytes()), 0))
+	require.True(t, decoded.InboundFee.IsSome())
+	require.NotEmpty(t, decoded.ExtraOpaqueData)
+
+	// Act by verifying the decoded wire representation. Decode retains the
+	// complete opaque stream, so verification must reconcile it with the
+	// typed fee exactly as signing did before encoding.
+	err = netann.VerifyChannelUpdateSignature(&decoded, pubKey)
+
+	// Assert that the signature remains valid across the full sign, encode,
+	// decode, and verify lifecycle while the unknown TLV is preserved.
+	require.NoError(t, err)
+}

### peer/brontide.go
@@ -2,7 +2,6 @@ package peer
 
 import (
 	"bytes"
-	"container/list"
 	"context"
 	"errors"
 	"fmt"
@@ -112,6 +111,16 @@ var (
 	// either the Brontide doesn't know of it, or the channel in question
 	// is pending.
 	ErrChannelNotFound = fmt.Errorf("channel not found")
+
+	// errPingFlood gives every flood-teardown path one stable identity. The
+	// peer still records the descriptive text, while callers and tests can
+	// match wrapped instances without depending on that text.
+	errPingFlood = errors.New("ping flood limit exceeded")
+
+	// errQueueOverflow identifies teardown caused by a bounded outgoing
+	// backlog. The detailed attempted totals wrap this stable identity so a
+	// synchronous sender can distinguish overload from generic peer exit.
+	errQueueOverflow = errors.New("outgoing message queue limit exceeded")
 )
 
 // outgoingMsg packages an lnwire.Message to be sent out on the wire, along with
@@ -121,6 +130,10 @@ type outgoingMsg struct {
 	priority bool
 	msg      lnwire.Message
 	errChan  chan error // MUST be buffered.
+
+	// queueCost is calculated before insertion so the generic queue can
+	// account for retained memory without interpreting wire message types.
+	queueCost int
 }
 
 // newChannelMsg packages a chanstate.OpenChannel with a channel that allows
@@ -585,6 +598,12 @@ type Brontide struct {
 
 	pingManager *PingManager
 
+	// pingLimits owns the two per-connection inbound Ping policies.
+	pingLimits pingLimits
+
+	// queueLimits supplies one accounting policy to the producer and queue.
+	queueLimits queueLimits
+
 	// lastPingPayload stores an unsafe pointer wrapped as an atomic
 	// variable which points to the last payload the remote party sent us
 	// as their ping.
@@ -746,6 +765,8 @@ func NewBrontide(cfg Config) *Brontide {
 		activeSignal:  make(chan struct{}),
 		sendQueue:     make(chan outgoingMsg),
 		outgoingQueue: make(chan outgoingMsg),
+		pingLimits:    defaultPingLimits(),
+		queueLimits:   defaultQueueLimits(),
 		addedChannels: &lnutils.SyncMap[lnwire.ChannelID, struct{}]{},
 		activeChannels: &lnutils.SyncMap[
 			lnwire.ChannelID, *lnwallet.LightningChannel,
@@ -2316,6 +2337,22 @@ out:
 			}
 		}
 
+		// Count before routing; consuming endpoints skip the switch.
+		// All Pings, including oversized ones, use the flood budget.
+		if _, ok := nextMsg.(*lnwire.Ping); ok &&
+			!p.pingLimits.pingLimiter.Allow() {
+
+			p.storeError(errPingFlood)
+			p.log.Warnf("%v", errPingFlood)
+
+			// Stop Ping management before peer cancellation.
+			// Keep queue handling active so a Ping send can finish
+			// through outgoingQueue without deadlock.
+			p.Disconnect(errPingFlood)
+
+			break out
+		}
+
 		// If a message router is active, then we'll try to have it
 		// handle this message. If it can, then we're able to skip the
 		// rest of the message handling logic.
@@ -2355,6 +2392,14 @@ out:
 				continue
 			}
 
+			// BOLT 1 requires a Pong for every Ping below the size
+			// ceiling. We limit reply frequency to guard against
+			// floods; normal keepalives remain below this limit.
+			if !p.pingLimits.pongLimiter.Allow() {
+				p.log.Debugf("Pong reply rate limited")
+				continue
+			}
+
 			// Next, we'll send over the amount of specified pong
 			// bytes.
 			pong := lnwire.NewPong(p.cfg.PongBuf[0:msg.NumPongBytes])
@@ -3084,62 +3129,84 @@ out:
 func (p *Brontide) queueHandler() {
 	defer p.cg.WgDone()
 
-	// priorityMsgs holds an in order list of messages deemed high-priority
-	// to be added to the sendQueue. This predominately includes messages
-	// from the funding manager and htlcswitch.
-	priorityMsgs := list.New()
-
-	// lazyMsgs holds an in order list of messages deemed low-priority to be
-	// added to the sendQueue only after all high-priority messages have
-	// been queued. This predominately includes messages from the gossiper.
-	lazyMsgs := list.New()
+	queue := newMsgQueue(p.queueLimits)
 
 	for {
-		// Examine the front of the priority queue, if it is empty check
-		// the low priority queue.
-		elem := priorityMsgs.Front()
-		if elem == nil {
-			elem = lazyMsgs.Front()
-		}
+		elem, next := queue.front()
 
+		// A nil channel disables this select case while the queue is
+		// empty. Incoming messages therefore use one generic path
+		// whether or not a message is ready for writeHandler.
+		var sendQueue chan outgoingMsg
 		if elem != nil {
-			front := elem.Value.(outgoingMsg)
+			sendQueue = p.sendQueue
+		}
 
-			// There's an element on the queue, try adding
-			// it to the sendQueue. We also watch for
-			// messages on the outgoingQueue, in case the
-			// writeHandler cannot accept messages on the
-			// sendQueue.
-			select {
-			case p.sendQueue <- front:
-				if front.priority {
-					priorityMsgs.Remove(elem)
-				} else {
-					lazyMsgs.Remove(elem)
-				}
-			case msg := <-p.outgoingQueue:
-				if msg.priority {
-					priorityMsgs.PushBack(msg)
-				} else {
-					lazyMsgs.PushBack(msg)
-				}
-			case <-p.cg.Done():
-				return
+		select {
+		case sendQueue <- next:
+			queue.pop(elem)
+
+		case msg := <-p.outgoingQueue:
+			if queue.push(msg) {
+				continue
 			}
-		} else {
-			// If there weren't any messages to send to the
-			// writeHandler, then we'll accept a new message
-			// into the queue from outside sub-systems.
-			select {
-			case msg := <-p.outgoingQueue:
-				if msg.priority {
-					priorityMsgs.PushBack(msg)
-				} else {
-					lazyMsgs.PushBack(msg)
-				}
-			case <-p.cg.Done():
-				return
+
+			// push leaves a rejected message unretained. Include it
+			// in the totals, then return its typed error before
+			// disconnecting.
+			p.failQueueOverflow(
+				msg, queue.numMsgs+1,
+				queue.numBytes+msg.queueCost,
+			)
+
+			return
+
+		case <-p.cg.Done():
+			return
+		}
+	}
+}
+
+// failQueueOverflow reports the rejected message's wrapped sentinel, tears the
+// connection down after the outgoing queue reaches a bound, and keeps that
+// queue serviced until teardown completes. Messages already admitted to the
+// local queue can be abandoned without individual replies; synchronous callers
+// then unblock through peer cancellation. This avoids blocking an arbitrary
+// producer and prevents subsystem-wide backpressure.
+//
+// NOTE: This blocks until the peer's context is cancelled, so it must be
+// called from the queueHandler goroutine itself.
+func (p *Brontide) failQueueOverflow(msg outgoingMsg, numQueued,
+	queuedBytes int) {
+
+	err := fmt.Errorf("%w: messages=%d, bytes=%d", errQueueOverflow,
+		numQueued, queuedBytes)
+
+	// The triggering message was never retained, so it cannot reach the
+	// write handler. A buffered response lets its synchronous SendMessage
+	// caller observe the precise overload error before teardown begins.
+	if msg.errChan != nil {
+		msg.errChan <- err
+	}
+
+	p.storeError(err)
+	p.log.Warnf("%v", err)
+
+	// Disconnect gets its own goroutine because we have to keep draining.
+	// Every message producer parks on outgoingQueue until the peer context
+	// is cancelled, and Disconnect waits on the ping manager before that
+	// cancellation. Walking away now could wedge both goroutines.
+	go p.Disconnect(err)
+
+	for {
+		select {
+		case msg := <-p.outgoingQueue:
+			if msg.errChan != nil {
+				msg.errChan <- lnpeer.ErrPeerExiting
 			}
+
+		case <-p.cg.Done():
+			return
 		}
 	}
 }
@@ -3169,8 +3236,17 @@ func (p *Brontide) queueMsgLazy(msg lnwire.Message, errChan chan error) {
 func (p *Brontide) queue(priority bool, msg lnwire.Message,
 	errChan chan error) {
 
+	// Compute retained-memory accounting at the producer boundary so the
+	// queue handles only generic cost metadata, never wire message types.
+	queuedMsg := outgoingMsg{
+		priority:  priority,
+		msg:       msg,
+		errChan:   errChan,
+		queueCost: p.queueLimits.msgCost(msg),
+	}
+
 	select {
-	case p.outgoingQueue <- outgoingMsg{priority, msg, errChan}:
+	case p.outgoingQueue <- queuedMsg:
 	case <-p.cg.Done():
 		p.log.Tracef("Peer shutting down, could not enqueue msg: %v.",
 			lnutils.SpewLogClosure(msg))
@@ -5117,39 +5193,58 @@ func (p *Brontide) SendMessageLazy(sync bool, msgs ...lnwire.Message) error {
 // messages have been sent to the remote peer or an error is returned, otherwise
 // it returns immediately after queueing.
 func (p *Brontide) sendMessage(sync, priority bool, msgs ...lnwire.Message) error {
-	// Add all incoming messages to the outgoing queue. A list of error
-	// chans is populated for each message if the caller requested a sync
-	// send.
-	var errChans []chan error
+	// One channel covers the whole synchronous batch so a later rejection
+	// cannot be hidden behind an earlier message that is still pending. A
+	// slot per message keeps producers non-blocking if this caller returns
+	// after the first error.
+	var errChan chan error
 	if sync {
-		errChans = make([]chan error, 0, len(msgs))
+		errChan = make(chan error, len(msgs))
 	}
 	for _, msg := range msgs {
-		// If a sync send was requested, create an error chan to listen
-		// for an ack from the writeHandler.
-		var errChan chan error
-		if sync {
-			errChan = make(chan error, 1)
-			errChans = append(errChans, errChan)
-		}
-
+		// Queue every message with the shared result destination. Async
+		// sends retain a nil channel and therefore require no replies.
 		if priority {
 			p.queueMsg(msg, errChan)
 		} else {
 			p.queueMsgLazy(msg, errChan)
 		}
 	}
 
-	// Wait for all replies from the writeHandler. For async sends, this
-	// will be a NOP as the list of error chans is nil.
-	for _, errChan := range errChans {
+	// Async callers are complete once every message reaches outgoingQueue;
+	// only synchronous batches have acknowledgements to collect below.
+	if !sync {
+		return nil
+	}
+
+	// peerExitError drains results already published before cancellation.
+	// Overflow reports its sentinel before initiating teardown, so scanning
+	// past successful replies preserves that more precise batch failure.
+	peerExitError := func() error {
+		for {
+			select {
+			case err := <-errChan:
+				if err != nil {
+					return err
+				}
+			default:
+				return lnpeer.ErrPeerExiting
+			}
+		}
+	}
+
+	// Wait for every synchronous reply so an accepted prefix cannot make a
+	// partially rejected variadic send appear successful.
+	for range msgs {
 		select {
 		case err := <-errChan:
-			return err
+			if err != nil {
+				return err
+			}
 		case <-p.cg.Done():
-			return lnpeer.ErrPeerExiting
+			return peerExitError()
 		case <-p.cfg.Quit:
-			return lnpeer.ErrPeerExiting
+			return peerExitError()
 		}
 	}
 

### peer/brontide_test.go
@@ -2,7 +2,10 @@ package peer
 
 import (
 	"bytes"
+	"context"
 	"fmt"
+	"net"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,14 +20,18 @@ import (
 	"github.com/lightningnetwork/lnd/contractcourt"
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/htlcswitch"
+	"github.com/lightningnetwork/lnd/lnpeer"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/msgmux"
 	"github.com/lightningnetwork/lnd/routing/route"
 	"github.com/lightningnetwork/lnd/tlv"
+	"github.com/lightningnetwork/lnd/tor"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -1213,6 +1220,8 @@ func TestPeerIgnoresPingWithoutPongReply(t *testing.T) {
 	_, err := fn.RecvOrTimeout(startPeerDone, 2*timeout)
 	require.NoError(t, err)
 
+	// writePing serializes each boundary request and injects it through the
+	// normal reader path so the assertions cover decoding and dispatch.
 	writePing := func(msg *lnwire.Ping) {
 		t.Helper()
 
@@ -1227,10 +1236,26 @@ func TestPeerIgnoresPingWithoutPongReply(t *testing.T) {
 		}
 	}
 
-	// Act: Deliver a ping in the BOLT 1 no-reply range.
+	// Act: Send the largest Ping BOLT 1 still requires us to answer,
+	// then read its response before exercising the adjacent no-reply value.
+	writePing(&lnwire.Ping{NumPongBytes: lnwire.MaxPongBytes})
+	rawMsg, err := fn.RecvOrTimeout(mockConn.writtenMessages, timeout)
+	require.NoError(t, err)
+
+	msg, err := lnwire.ReadMessage(bytes.NewReader(rawMsg), 0)
+	require.NoError(t, err)
+
+	// Assert: The inclusive boundary receives exactly the requested
+	// bytes, proving the implementation does not suppress one value early.
+	pong, ok := msg.(*lnwire.Pong)
+	require.True(t, ok)
+	require.Len(t, pong.PongBytes, int(lnwire.MaxPongBytes))
+
+	// Act: Send the first BOLT 1 no-reply value and retain a
+	// payload that shows when the read loop has processed it.
 	ignoredPayload := []byte{1, 2, 3}
 	writePing(&lnwire.Ping{
-		NumPongBytes: 65535,
+		NumPongBytes: lnwire.MaxPongBytes + 1,
 		PaddingBytes: ignoredPayload,
 	})
 
@@ -1253,18 +1278,973 @@ func TestPeerIgnoresPingWithoutPongReply(t *testing.T) {
 	// traffic.
 	writePing(&lnwire.Ping{NumPongBytes: 1})
 
-	rawMsg, err := fn.RecvOrTimeout(mockConn.writtenMessages, timeout)
+	rawMsg, err = fn.RecvOrTimeout(mockConn.writtenMessages, timeout)
 	require.NoError(t, err)
 
-	msg, err := lnwire.ReadMessage(bytes.NewReader(rawMsg), 0)
+	msg, err = lnwire.ReadMessage(bytes.NewReader(rawMsg), 0)
 	require.NoError(t, err)
 
 	// Assert: The follow-up ping receives the requested pong reply.
-	pong, ok := msg.(*lnwire.Pong)
+	pong, ok = msg.(*lnwire.Pong)
 	require.True(t, ok)
 	require.Len(t, pong.PongBytes, 1)
 }
 
+// TestPeerPingLimitsProductionBoundaries verifies the exact burst and refill
+// thresholds used by both production Ping policies.
+func TestPeerPingLimitsProductionBoundaries(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Use fresh production limiters and expected values
+	// so each subtest starts with a full, independent token bucket.
+	limits := defaultPingLimits()
+	tests := []struct {
+		name    string
+		limiter *rate.Limiter
+		limit   rate.Limit
+		burst   int
+	}{
+		{
+			name:    "Pong replies",
+			limiter: limits.pongLimiter,
+			limit:   pongReplyRate,
+			burst:   pongReplyBurst,
+		},
+		{
+			name:    "Ping floods",
+			limiter: limits.pingLimiter,
+			limit:   pingFloodRate,
+			burst:   pingFloodBurst,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Derive the one-token interval from the rate
+			// constant under test, then fix a synthetic timestamp.
+			// This makes both sides of the boundary deterministic.
+			// Two nanoseconds keep the pre-boundary deficit above
+			// rate's duration-truncation quantum.
+			now := time.Now()
+			refillTime := time.Duration(
+				float64(time.Second) / float64(test.limit),
+			)
+			const boundaryEpsilon = 2 * time.Nanosecond
+			require.Equal(t, test.limit, test.limiter.Limit())
+			require.Equal(t, test.burst, test.limiter.Burst())
+
+			// Act: Consume the burst, probe one token past it, and
+			// test just before and at the derived replacement time.
+			atBoundary := test.limiter.AllowN(now, test.burst)
+			pastBoundary := test.limiter.AllowN(now, 1)
+			beforeRefill := test.limiter.AllowN(
+				now.Add(refillTime-boundaryEpsilon), 1,
+			)
+			atRefill := test.limiter.AllowN(
+				now.Add(refillTime), 1,
+			)
+
+			// Assert: The burst boundary is inclusive, both probes
+			// before refill are rejected, and the derived boundary
+			// restores exactly one token without scheduler timing.
+			require.True(t, atBoundary)
+			require.False(t, pastBoundary)
+			require.False(t, beforeRefill)
+			require.True(t, atRefill)
+		})
+	}
+}
+
+// TestPeerPingLimitsAllowHonestCadence verifies that both inbound Ping
+// limiters admit realistic keepalive cadences for long-lived connections.
+func TestPeerPingLimitsAllowHonestCadence(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		cadence time.Duration
+	}{
+		{name: "lnd cadence", cadence: time.Minute},
+		{name: "aggressive cadence", cadence: 10 * time.Second},
+		{name: "five second cadence", cadence: 5 * time.Second},
+		{name: "pathological cadence", cadence: 2 * time.Second},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Construct the production Ping policy
+			// separately so token history cannot cross test cases.
+			limits := defaultPingLimits()
+			start := time.Now()
+
+			// Act: Advance a synthetic clock at the selected
+			// cadence, avoiding scheduler and wall-clock noise.
+			for i := 0; i < 5000; i++ {
+				elapsed := time.Duration(i) * test.cadence
+				now := start.Add(elapsed)
+
+				// Assert: Both budgets admit each ping, so this
+				// cadence reaches neither protection tier.
+				require.True(
+					t, limits.pongLimiter.AllowN(now, 1),
+				)
+				require.True(
+					t, limits.pingLimiter.AllowN(now, 1),
+				)
+			}
+		})
+	}
+}
+
+// TestPeerPongReplyRateLimited verifies that exhausting the reply budget
+// suppresses Pongs without disconnecting the peer.
+func TestPeerPongReplyRateLimited(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Start a peer whose reply limiter has one token, so the
+	// first valid ping replies and the second exhausts the budget.
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.pingLimits.pongLimiter = rate.NewLimiter(0, 1)
+
+	startDone := startPeer(t, params.mockConn, peer)
+	_, err := fn.RecvOrTimeout(startDone, 2*timeout)
+	require.NoError(t, err)
+
+	// writePing serializes a valid one-byte-reply ping with an observable
+	// payload and injects it through the mock connection's normal reader.
+	// Distinct payloads synchronize the assertion with each exact Ping.
+	writePing := func(payload []byte) {
+		var b bytes.Buffer
+		ping := lnwire.NewPing(1)
+		ping.PaddingBytes = payload
+		_, err := lnwire.WriteMessage(&b, ping, 0)
+		require.NoError(t, err)
+		select {
+		case params.mockConn.readMessages <- b.Bytes():
+		case <-peer.cg.Done():
+			t.Fatal("peer disconnected before Ping was delivered")
+		}
+	}
+
+	// Act: Deliver two unique Pings and consume the first Pong. Then inject
+	// the Ping that exhausts the reply budget.
+	firstPayload := []byte{1}
+	secondPayload := []byte{2}
+	writePing(firstPayload)
+	_, err = fn.RecvOrTimeout(params.mockConn.writtenMessages, timeout)
+	require.NoError(t, err)
+
+	writePing(secondPayload)
+
+	// Assert: Observe the second payload before checking the write channel.
+	// This proves the read loop processed the rate-limited Ping.
+	require.Eventually(t, func() bool {
+		return bytes.Equal(
+			peer.LastRemotePingPayload(), secondPayload,
+		)
+	}, timeout, 10*time.Millisecond)
+
+	select {
+	case msg := <-params.mockConn.writtenMessages:
+		t.Fatalf("unexpected Pong after reply budget: %x", msg)
+	case <-time.After(shortTimeout):
+	}
+
+	// Assert: The peer remains connected, proving reply exhaustion only
+	// suppresses amplification and does not trigger flood teardown.
+	require.Zero(t, atomic.LoadInt32(&peer.disconnect))
+}
+
+// mockMsgRouter records message-router calls while letting a test choose
+// whether a message would be consumed. Embedding mock.Mock keeps every
+// interface interaction explicit and independently assertable.
+type mockMsgRouter struct {
+	mock.Mock
+}
+
+// RegisterEndpoint returns the result configured for one endpoint so tests
+// can exercise router registration without adding a second fake.
+func (m *mockMsgRouter) RegisterEndpoint(endpoint msgmux.Endpoint) error {
+	args := m.Called(endpoint)
+
+	return args.Error(0)
+}
+
+// UnregisterEndpoint returns the configured removal result for the supplied
+// endpoint name.
+func (m *mockMsgRouter) UnregisterEndpoint(name msgmux.EndpointName) error {
+	args := m.Called(name)
+
+	return args.Error(0)
+}
+
+// RouteMsg returns the configured routing result while recording the complete
+// peer message that reached the generic routing boundary.
+func (m *mockMsgRouter) RouteMsg(msg msgmux.PeerMsg) error {
+	args := m.Called(msg)
+
+	return args.Error(0)
+}
+
+// Start records the lifecycle context so any test that starts the mock router
+// must declare that interaction explicitly.
+func (m *mockMsgRouter) Start(ctx context.Context) {
+	m.Called(ctx)
+}
+
+// Stop records shutdown so tests cannot accidentally rely on an unobserved
+// router lifecycle transition.
+func (m *mockMsgRouter) Stop() {
+	m.Called()
+}
+
+// Compile-time verification keeps the focused mock synchronized with the
+// production router interface used by Brontide.
+var _ msgmux.Router = (*mockMsgRouter)(nil)
+
+// TestPeerPingFloodDisconnects verifies flood accounting precedes a generic
+// router that would consume an oversized Ping.
+func TestPeerPingFloodDisconnects(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Empty the flood budget and retain errors through an active
+	// channel. Install a mock router prepared to consume any message;
+	// marking it global avoids unrelated lifecycle calls.
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.pingLimits.pingLimiter = rate.NewLimiter(0, 0)
+	peer.remoteFeatures = lnwire.EmptyFeatureVector()
+	peer.activeChannels.Store(
+		lnwire.ChannelID{1}, &lnwallet.LightningChannel{},
+	)
+
+	router := &mockMsgRouter{}
+	router.On("RouteMsg", mock.Anything).Return(nil).Maybe()
+	peer.msgRouter = fn.Some[msgmux.Router](router)
+	peer.globalMsgRouter = true
+
+	// Arrange: Encode the first oversized Pong request and register the
+	// focused reader with the control group so shutdown remains joinable.
+	var b bytes.Buffer
+	_, err := lnwire.WriteMessage(&b, &lnwire.Ping{
+		NumPongBytes: lnwire.MaxPongBytes + 1,
+	}, 0)
+	require.NoError(t, err)
+
+	peer.cg.WgAdd(1)
+	go peer.readHandler()
+
+	// Act: Send the oversized Ping through normal decoding, then wait for
+	// the empty flood budget to cancel and fully stop the focused reader.
+	select {
+	case params.mockConn.readMessages <- b.Bytes():
+	case <-peer.cg.Done():
+		t.Fatal("peer disconnected before Ping was delivered")
+	}
+
+	_, err = fn.RecvOrTimeout(peer.cg.Done(), timeout)
+	require.NoError(t, err)
+	peer.cg.WgWait()
+
+	// Assert: Teardown precedes generic routing, and the retained error
+	// matches the stable sentinel without depending on its display text.
+	require.EqualValues(t, 1, atomic.LoadInt32(&peer.disconnect))
+	router.AssertNotCalled(t, "RouteMsg", mock.Anything)
+
+	storedErrors := peer.ErrorBuffer().List()
+	require.NotEmpty(t, storedErrors)
+	storedErr, ok := storedErrors[0].(*TimestampedError)
+	require.True(t, ok)
+	require.ErrorIs(t, storedErr.Error, errPingFlood)
+}
+
+// startTestQueueHandler isolates queue ownership from unrelated peer loops so
+// focused tests can drive outgoingQueue directly; callers own cancellation
+// and join the registered goroutine before returning.
+func startTestQueueHandler(peer *Brontide) {
+	peer.cg.WgAdd(1)
+	go peer.queueHandler()
+}
+
+// mockQueueWriteConn adds testify-controlled write behavior to the shared
+// connection fixture while inheriting its safe address and close methods.
+type mockQueueWriteConn struct {
+	mock.Mock
+	*mockMessageConn
+}
+
+// Compile-time conformance keeps the focused mock aligned with MessageConn.
+var _ MessageConn = (*mockQueueWriteConn)(nil)
+
+// SetWriteDeadline records each pre-flush deadline for exact call assertions.
+func (m *mockQueueWriteConn) SetWriteDeadline(deadline time.Time) error {
+	return m.Called(deadline).Error(0)
+}
+
+// WriteMessage lets each test control when a serialized message is accepted.
+func (m *mockQueueWriteConn) WriteMessage(msg []byte) error {
+	return m.Called(msg).Error(0)
+}
+
+// Flush records the final wire flush and returns its configured outcome.
+func (m *mockQueueWriteConn) Flush() (int, error) {
+	args := m.Called()
+	return args.Int(0), args.Error(1)
+}
+
+// TestMsgQueueRejectsOverflowWithoutRetention verifies that prospective limit
+// checks leave both priority lists and their accounting at the accepted cap.
+func TestMsgQueueRejectsOverflowWithoutRetention(t *testing.T) {
+	// Arrange: Fill a one-message, one-byte queue with a priority item so a
+	// lazy item would cross both limits and expose either insertion path.
+	queue := newMsgQueue(queueLimits{maxMsgs: 1, maxBytes: 1})
+	require.True(t, queue.push(outgoingMsg{
+		priority: true, queueCost: 1,
+	}))
+
+	// Act: Attempt to append one excess lazy item through the normal push.
+	accepted := queue.push(outgoingMsg{queueCost: 1})
+
+	// Assert: Rejection preserves the exact accepted totals and leaves the
+	// lazy list empty, proving the excess object is no longer retained.
+	require.False(t, accepted)
+	require.Equal(t, 1, queue.numMsgs)
+	require.Equal(t, 1, queue.numBytes)
+	require.Equal(t, 1, queue.priorityMsgs.Len())
+	require.Zero(t, queue.lazyMsgs.Len())
+}
+
+// TestMsgQueueOrdersAndReleasesCapacity verifies strict priority selection,
+// FIFO ordering within each class, and exact accounting release on removal.
+func TestMsgQueueOrdersAndReleasesCapacity(t *testing.T) {
+	// Arrange: Interleave lazy and priority messages with distinct costs.
+	// The queue starts below both limits so every item is admitted and the
+	// expected removal totals can prove which element left at each step.
+	queue := newMsgQueue(queueLimits{maxMsgs: 4, maxBytes: 40})
+	lazyFirst := lnwire.NewPing(1)
+	priorityFirst := lnwire.NewPing(2)
+	lazySecond := lnwire.NewPing(3)
+	prioritySecond := lnwire.NewPing(4)
+
+	require.True(t, queue.push(outgoingMsg{
+		msg: lazyFirst, queueCost: 11,
+	}))
+	require.True(t, queue.push(outgoingMsg{
+		priority: true, msg: priorityFirst, queueCost: 7,
+	}))
+	require.True(t, queue.push(outgoingMsg{
+		msg: lazySecond, queueCost: 13,
+	}))
+	require.True(t, queue.push(outgoingMsg{
+		priority: true, msg: prioritySecond, queueCost: 5,
+	}))
+
+	// Act: Repeatedly select and remove the front item, recording both the
+	// chosen message and the shadow totals after each exact cost is
+	// released.
+	var (
+		orderedMsgs    []lnwire.Message
+		remainingMsgs  []int
+		remainingBytes []int
+	)
+	for {
+		elem, msg := queue.front()
+		if elem == nil {
+			break
+		}
+
+		orderedMsgs = append(orderedMsgs, msg.msg)
+		queue.pop(elem)
+		remainingMsgs = append(remainingMsgs, queue.numMsgs)
+		remainingBytes = append(remainingBytes, queue.numBytes)
+	}
+
+	acceptedAfterDrain := queue.push(outgoingMsg{queueCost: 40})
+
+	// Assert: Priority items lead in FIFO order, lazy items follow in FIFO
+	// order, each pop releases its own count and bytes, and the empty queue
+	// can admit an item that consumes the complete byte budget again.
+	require.Equal(t, []lnwire.Message{
+		priorityFirst, prioritySecond, lazyFirst, lazySecond,
+	}, orderedMsgs)
+	require.Equal(t, []int{3, 2, 1, 0}, remainingMsgs)
+	require.Equal(t, []int{29, 24, 13, 0}, remainingBytes)
+	require.True(t, acceptedAfterDrain)
+	require.Equal(t, 1, queue.numMsgs)
+	require.Equal(t, 40, queue.numBytes)
+}
+
+// TestPeerSendMessageQueueBounds verifies that the public sending API accepts
+// an exact queue boundary and returns a stable error for its first excess.
+func TestPeerSendMessageQueueBounds(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Derive exact data-only boundaries from the production
+	// limits.
+	// The onion size makes each charged message 64 KiB, so 256 distinct
+	// blobs fill the 16 MiB byte budget without approaching the count cap.
+	limits := defaultQueueLimits()
+	onionBlobSize := (1 << 16) - limits.msgOverhead
+	require.Zero(t, limits.maxBytes%(limits.msgOverhead+onionBlobSize))
+	tests := []struct {
+		name        string
+		msgType     lnwire.MessageType
+		payloadSize int
+		numAtLimit  int
+	}{
+		{
+			name:       "message count",
+			msgType:    lnwire.MsgPong,
+			numAtLimit: limits.maxMsgs,
+		},
+		{
+			name:        "message bytes",
+			msgType:     lnwire.MsgOnionMessage,
+			payloadSize: onionBlobSize,
+			numAtLimit: limits.maxBytes /
+				(limits.msgOverhead + onionBlobSize),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Start only the queue handler so no writer
+			// drains it. The row selects a builder that allocates a
+			// fresh peer-controlled blob for every message.
+			params := createTestPeer(t)
+			peer := params.peer
+			startTestQueueHandler(peer)
+			t.Cleanup(func() {
+				peer.cg.Quit()
+				peer.cg.WgWait()
+			})
+
+			// newMsg builds fresh values while keeping rows
+			// declarative, including distinct blobs for every send.
+			newMsg := func() lnwire.Message {
+				switch test.msgType {
+				case lnwire.MsgPong:
+					return lnwire.NewPong(nil)
+
+				case lnwire.MsgOnionMessage:
+					blob := make([]byte, test.payloadSize)
+					return &lnwire.OnionMessage{
+						OnionBlob: blob,
+					}
+
+				default:
+					t.Fatalf(
+						"unsupported queue message: %v",
+						test.msgType,
+					)
+				}
+
+				return nil
+			}
+
+			// Act: Queue the exact boundary through SendMessage.
+			// Its async path waits for queueHandler to
+			// receive its message without requiring a writer.
+			for i := 0; i < test.numAtLimit; i++ {
+				err := peer.SendMessage(false, newMsg())
+				require.NoError(t, err)
+			}
+
+			// Assert: The inclusive boundary remains connected. It
+			// proves the cap describes accepted traffic rather than
+			// the message that merely reaches it.
+			select {
+			case <-peer.cg.Done():
+				t.Fatal("peer disconnected at the inclusive " +
+					"queue limit")
+			default:
+			}
+
+			// Act: Send one additional message synchronously so its
+			// rejected caller receives the precise overflow cause.
+			err := peer.SendMessage(true, newMsg())
+
+			// Assert: The first excess send returns the wrapped
+			// sentinel. Cancellation and joining prove teardown.
+			require.ErrorIs(t, err, errQueueOverflow)
+			_, err = fn.RecvOrTimeout(peer.cg.Done(), timeout)
+			require.NoError(t, err)
+			peer.cg.WgWait()
+		})
+	}
+}
+
+// TestPeerSendMessageMixedWriteAndOverflow verifies that a live writer can
+// complete earlier public sends before a later message crosses the queue cap.
+func TestPeerSendMessageMixedWriteAndOverflow(t *testing.T) {
+	// Arrange: Allow one retained message, make the third wire write block,
+	// and run both queue stages. First two writes must fully acknowledge
+	// before the blocked write lets one message fill the local queue.
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.queueLimits = defaultQueueLimits()
+	peer.queueLimits.maxMsgs = 1
+	releaseWrite := make(chan struct{})
+	thirdWriteStarted := make(chan struct{})
+	var writeCount atomic.Int32
+	conn := &mockQueueWriteConn{mockMessageConn: params.mockConn}
+	conn.On("WriteMessage", mock.Anything).Run(func(mock.Arguments) {
+		if writeCount.Add(1) == 3 {
+			close(thirdWriteStarted)
+			<-releaseWrite
+		}
+	}).Return(nil).Times(3)
+	conn.On("SetWriteDeadline", mock.Anything).Return(nil).Times(3)
+	conn.On("Flush").Return(0, nil).Times(3)
+	peer.cfg.Conn = conn
+	peer.cg.WgAdd(2)
+	go peer.queueHandler()
+	go peer.writeHandler()
+	released := false
+	t.Cleanup(func() {
+		if !released {
+			close(releaseWrite)
+		}
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Complete two synchronous sends, block a third in the writer,
+	// admit a fourth asynchronously, then submit the fifth message that
+	// exceeds the one-message local backlog.
+	firstErr := peer.SendMessage(true, lnwire.NewPing(0))
+	secondErr := peer.SendMessage(true, lnwire.NewPing(0))
+	thirdResult := make(chan error, 1)
+	go func() {
+		thirdResult <- peer.SendMessage(true, lnwire.NewPing(0))
+	}()
+	_, err := fn.RecvOrTimeout(thirdWriteStarted, timeout)
+	require.NoError(t, err)
+	queuedErr := peer.SendMessage(false, lnwire.NewPing(0))
+	overflowErr := peer.SendMessage(true, lnwire.NewPing(0))
+
+	// Assert: The live writes and exact-limit admission succeed, while only
+	// the first excess public send receives the typed overflow. Releasing
+	// the in-flight write proves all participating goroutines terminate.
+	require.NoError(t, firstErr)
+	require.NoError(t, secondErr)
+	require.NoError(t, queuedErr)
+	require.ErrorIs(t, overflowErr, errQueueOverflow)
+	close(releaseWrite)
+	released = true
+	_, err = fn.RecvOrTimeout(thirdResult, timeout)
+	require.NoError(t, err)
+	peer.cg.WgWait()
+	conn.AssertExpectations(t)
+}
+
+// TestPeerConcurrentSenders verifies that synchronized public callers share
+// the bounded queue and single writer without racing or losing acknowledgments.
+func TestPeerConcurrentSenders(t *testing.T) {
+	// Arrange: Give each sender one queue slot and configure a testify mock
+	// for the exact write lifecycle, so the race run observes queue and
+	// writer coordination instead of stopping at outgoingQueue admission.
+	const numSenders = 16
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.queueLimits = defaultQueueLimits()
+	peer.queueLimits.maxMsgs = numSenders
+	conn := &mockQueueWriteConn{mockMessageConn: params.mockConn}
+	conn.On("WriteMessage", mock.Anything).Return(nil).Times(numSenders)
+	conn.On("SetWriteDeadline", mock.Anything).Return(nil).Times(numSenders)
+	conn.On("Flush").Return(0, nil).Times(numSenders)
+	peer.cfg.Conn = conn
+	peer.cg.WgAdd(2)
+	go peer.queueHandler()
+	go peer.writeHandler()
+	t.Cleanup(func() {
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Launch all synchronous senders concurrently and collect each
+	// public result through a buffered channel that cannot serialize them.
+	results := make(chan error, numSenders)
+	for i := 0; i < numSenders; i++ {
+		go func() {
+			results <- peer.SendMessage(true, lnwire.NewPing(0))
+		}()
+	}
+
+	// Assert: Every sender receives its successful writer acknowledgment,
+	// all expected wire operations occur, and both handlers join cleanly.
+	for i := 0; i < numSenders; i++ {
+		err, recvErr := fn.RecvOrTimeout(results, timeout)
+		require.NoError(t, recvErr)
+		require.NoError(t, err)
+	}
+	peer.cg.Quit()
+	peer.cg.WgWait()
+	conn.AssertExpectations(t)
+}
+
+// TestPeerSendMessageBatchReturnsQueueOverflow verifies that a synchronous
+// variadic send reports a later queue rejection even while its first message
+// remains accepted and pending.
+func TestPeerSendMessageBatchReturnsQueueOverflow(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Restrict the peer to one retained message and start only the
+	// queue handler. With no writer, the first batch item cannot
+	// acknowledge before the second item crosses the count limit.
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.queueLimits = defaultQueueLimits()
+	peer.queueLimits.maxMsgs = 1
+	startTestQueueHandler(peer)
+	t.Cleanup(func() {
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Submit both messages through the public synchronous API so they
+	// share one batch result path and the later rejection initiates
+	// teardown.
+	err := peer.SendMessage(
+		true, lnwire.NewPing(0), lnwire.NewPing(0),
+	)
+
+	// Assert: The overflow sentinel wins over the first message's missing
+	// acknowledgement and generic cancellation, then teardown completes.
+	require.ErrorIs(t, err, errQueueOverflow)
+	_, err = fn.RecvOrTimeout(peer.cg.Done(), timeout)
+	require.NoError(t, err)
+	peer.cg.WgWait()
+}
+
+// TestPeerMessageQueueCost verifies the non-serializing cost rules used by
+// the outgoing queue byte budget.
+func TestPeerMessageQueueCost(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Load the production fixed overhead and describe each
+	// retained payload declaratively so expectations follow the policy
+	// without serialization hiding whether the original slice is charged.
+	limits := defaultQueueLimits()
+	channelFeatures := lnwire.NewRawFeatureVector(1, 17)
+	nodeFeatures := lnwire.NewRawFeatureVector(3, 9)
+	// The highest defined feature produces a large, mostly empty serialized
+	// span, guarding against multiplication as if every bit were set.
+	sparseFeatures := lnwire.NewRawFeatureVector(
+		lnwire.SimpleTaprootOverlayChansRequired,
+	)
+	// A dense vector retains one map entry for every possible FeatureBit.
+	// Building the complete key space proves queue charging follows decoded
+	// population instead of only the much smaller serialized bit span.
+	denseFeatureBits := make([]lnwire.FeatureBit, 1<<16)
+	for i := range denseFeatureBits {
+		denseFeatureBits[i] = lnwire.FeatureBit(i)
+	}
+	denseFeatures := lnwire.NewRawFeatureVector(denseFeatureBits...)
+	tcpAddr := &net.TCPAddr{
+		IP:   net.IPv4(192, 0, 2, 1),
+		Port: 9735,
+		Zone: "test-zone",
+	}
+	onionAddr := &tor.OnionAddr{
+		OnionService: "abcdefghijklmnop.onion",
+		Port:         9735,
+	}
+	dnsAddr := &lnwire.DNSAddress{
+		Hostname: "node.example.com",
+		Port:     9735,
+	}
+	opaqueAddr := &lnwire.OpaqueAddrs{
+		Payload: make([]byte, 11),
+	}
+	nodeAddrs := []net.Addr{tcpAddr, onionAddr, dnsAddr, opaqueAddr}
+	addressCost := len(nodeAddrs)*queuedAddrOverhead + len(tcpAddr.IP) +
+		len(tcpAddr.Zone) + len(onionAddr.OnionService) +
+		len(dnsAddr.Hostname) + len(opaqueAddr.Payload)
+	tests := []struct {
+		name     string
+		msg      lnwire.Message
+		expected int
+	}{
+		{
+			name:     "fixed message",
+			msg:      lnwire.NewPing(0),
+			expected: limits.msgOverhead,
+		},
+		{
+			name:     "shared Pong payload",
+			msg:      lnwire.NewPong(make([]byte, 1000)),
+			expected: limits.msgOverhead,
+		},
+		{
+			name: "failure reason",
+			msg: &lnwire.UpdateFailHTLC{
+				Reason: make([]byte, 5),
+			},
+			expected: limits.msgOverhead + 5,
+		},
+		{
+			name: "add onion and extra data",
+			msg: &lnwire.UpdateAddHTLC{
+				ExtraData: make([]byte, 7),
+			},
+			expected: limits.msgOverhead +
+				lnwire.OnionPacketSize + 7,
+		},
+		{
+			name: "error data",
+			msg: &lnwire.Error{
+				Data: make([]byte, 3),
+			},
+			expected: limits.msgOverhead + 3,
+		},
+		{
+			name: "warning data",
+			msg: &lnwire.Warning{
+				Data: make([]byte, 4),
+			},
+			expected: limits.msgOverhead + 4,
+		},
+		{
+			name: "onion message blob",
+			msg: &lnwire.OnionMessage{
+				OnionBlob: make([]byte, 6),
+			},
+			expected: limits.msgOverhead + 6,
+		},
+		{
+			name: "channel announcement retained data",
+			msg: &lnwire.ChannelAnnouncement1{
+				Features:        channelFeatures,
+				ExtraOpaqueData: make([]byte, 8),
+			},
+			// Literal 107 pins 8 opaque bytes, 64 bytes of feature
+			// overhead, two map entries, and the three-byte span.
+			expected: limits.msgOverhead + 107,
+		},
+		{
+			name: "node announcement retained data",
+			msg: &lnwire.NodeAnnouncement1{
+				Features:        nodeFeatures,
+				Addresses:       nodeAddrs,
+				ExtraOpaqueData: make([]byte, 9),
+			},
+			// Literal 107 pins 9 opaque bytes, 64 bytes of feature
+			// overhead, two map entries, and the two-byte span.
+			expected: limits.msgOverhead + 107 + addressCost,
+		},
+		{
+			name: "sparse high feature",
+			msg: &lnwire.ChannelAnnouncement1{
+				Features: sparseFeatures,
+			},
+			// Literal 334 covers fixed, sparse-span, and one-entry
+			// costs without charging unset intervening bits.
+			expected: limits.msgOverhead + 334,
+		},
+		{
+			name: "dense feature entries",
+			msg: &lnwire.ChannelAnnouncement1{
+				Features: denseFeatures,
+			},
+			// The expected cost independently derives the retained
+			// entry charge from the full input. This catches a
+			// regression to serialized-span-only accounting.
+			expected: limits.msgOverhead + queuedFeatureOverhead +
+				denseFeatures.SerializeSize() +
+				len(denseFeatureBits)*
+					queuedFeatureEntryOverhead,
+		},
+		{
+			name: "channel update opaque data",
+			msg: &lnwire.ChannelUpdate1{
+				ExtraOpaqueData: make([]byte, 10),
+			},
+			expected: limits.msgOverhead + 10,
+		},
+		{
+			name: "channel range query retained data",
+			msg: &lnwire.QueryChannelRange{
+				QueryOptions: lnwire.NewTimestampQueryOption(),
+				ExtraData:    make([]byte, 11),
+			},
+			// The single query bit retains its vector object, map
+			// entry, and one-byte span in addition to opaque data.
+			expected: limits.msgOverhead + 11 +
+				queuedFeatureOverhead +
+				queuedFeatureEntryOverhead + 1,
+		},
+		{
+			name: "short channel ID query retained data",
+			msg: &lnwire.QueryShortChanIDs{
+				ShortChanIDs: make([]lnwire.ShortChannelID, 2),
+				ExtraData:    make([]byte, 7),
+			},
+			// Decoded SCIDs occupy their padded in-memory width,
+			// while the unknown TLV backing bytes remain retained.
+			expected: limits.msgOverhead +
+				2*queuedShortChanIDSize + 7,
+		},
+		{
+			name: "channel range reply retained data",
+			msg: &lnwire.ReplyChannelRange{
+				ShortChanIDs: make([]lnwire.ShortChannelID, 3),
+				Timestamps:   make(lnwire.Timestamps, 3),
+				ExtraData:    make([]byte, 5),
+			},
+			// Each decoded reply row retains one padded SCID and
+			// one pair of uint32 update timestamps.
+			expected: limits.msgOverhead +
+				3*queuedShortChanIDSize +
+				3*queuedTimestampPairSize + 5,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Select the row's concrete message and exact
+			// fixed-plus-dynamic expectation without an encoder.
+
+			// Act: Evaluate the immutable charge stored with the
+			// message when it enters the outgoing queue.
+			actual := limits.msgCost(test.msg)
+
+			// Assert: Exact equality proves the dynamic bytes are
+			// neither omitted nor counted more than once.
+			require.Equal(t, test.expected, actual)
+		})
+	}
+}
+
+// TestPeerQueueHandlerDrainsBacklog verifies that messages leaving the queue
+// decrement its shadow count for a long-lived peer.
+func TestPeerQueueHandlerDrainsBacklog(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Start the isolated handler and register cleanup that
+	// cancels and joins it so no goroutine survives this test.
+	params := createTestPeer(t)
+	peer := params.peer
+	startTestQueueHandler(peer)
+	t.Cleanup(func() {
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Exceed the lifetime count cap while draining each message
+	// immediately, forcing the shadow count back to zero each time.
+	for i := 0; i <= peer.queueLimits.maxMsgs; i++ {
+		peer.queueMsg(lnwire.NewPong(nil), nil)
+
+		select {
+		case <-peer.sendQueue:
+		case <-peer.cg.Done():
+			t.Fatal("healthy drained queue exceeded message bound")
+		case <-time.After(timeout):
+			t.Fatal("queued message was not drained")
+		}
+	}
+
+	// Assert: Every item drained and the peer remains live, proving
+	// only the concurrent backlog contributes to the queue bound.
+	select {
+	case <-peer.cg.Done():
+		t.Fatal("healthy drained queue disconnected")
+	default:
+	}
+}
+
+// TestPeerQueueHandlerServicesQueueDuringTeardown verifies that queue
+// producers are failed while Disconnect waits for peer startup to finish.
+func TestPeerQueueHandlerServicesQueueDuringTeardown(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Mark the peer started but hold startReady open, so
+	// overflow enters Disconnect without finishing; cleanup later
+	// releases that gate, cancels, and joins the queue goroutine.
+	params := createTestPeer(t)
+	peer := params.peer
+	atomic.StoreInt32(&peer.started, 1)
+	startTestQueueHandler(peer)
+	t.Cleanup(func() {
+		select {
+		case <-peer.startReady:
+		default:
+			close(peer.startReady)
+		}
+
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Cross the count cap, wait for Disconnect to block, then
+	// invoke a synchronous sender in a goroutine so the queue handler
+	// must return its result while teardown remains pending.
+	for i := 0; i <= peer.queueLimits.maxMsgs; i++ {
+		peer.queueMsg(lnwire.NewPong(nil), nil)
+	}
+
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt32(&peer.disconnect) == 1
+	}, timeout, 10*time.Millisecond)
+
+	errChan := make(chan error, 1)
+	go func() {
+		errChan <- peer.SendMessage(true, lnwire.NewPing(0))
+	}()
+
+	// Assert: The sender gets ErrPeerExiting while cg remains live,
+	// proving producers are serviced until startup teardown can finish.
+	err, recvErr := fn.RecvOrTimeout(errChan, timeout)
+	require.NoError(t, recvErr)
+	require.ErrorIs(t, err, lnpeer.ErrPeerExiting)
+
+	select {
+	case <-peer.cg.Done():
+		t.Fatal("Disconnect completed before startReady was signaled")
+	default:
+	}
+}
+
+// TestPeerPriorityMessageSharesQueueBudget verifies that lazy traffic can fill
+// the shared queue budget and cause the first later priority send to tear down
+// the peer rather than bypassing or displacing an already accepted message.
+func TestPeerPriorityMessageSharesQueueBudget(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Restrict the peer to one fixed-cost message and start only
+	// queue ownership, leaving no writer to drain the lazy message. Cleanup
+	// cancels and joins the handler even if an assertion stops the test
+	// early.
+	params := createTestPeer(t)
+	peer := params.peer
+	peer.queueLimits = queueLimits{
+		maxMsgs:     1,
+		maxBytes:    queuedMsgOverhead,
+		msgOverhead: queuedMsgOverhead,
+	}
+	startTestQueueHandler(peer)
+	t.Cleanup(func() {
+		peer.cg.Quit()
+		peer.cg.WgWait()
+	})
+
+	// Act: Fill the exact shared budget through the public lazy API, then
+	// send one synchronous priority message so its rejection is observable.
+	err := peer.SendMessageLazy(false, lnwire.NewPong(nil))
+	require.NoError(t, err)
+	err = peer.SendMessage(true, lnwire.NewPing(0))
+
+	// Assert: Priority has no reserved capacity: the first excess send gets
+	// the typed overflow cause, disconnects the peer, and finishes
+	// teardown.
+	require.ErrorIs(t, err, errQueueOverflow)
+	_, err = fn.RecvOrTimeout(peer.cg.Done(), timeout)
+	require.NoError(t, err)
+	peer.cg.WgWait()
+}
+
 // TestMessageSummaryPingIncludesNumPongBytes ensures the debug summary for a
 // ping exposes the requested pong size, which makes ignored no-reply pings
 // visible without requiring trace-level logging.

### peer/msg_queue.go
@@ -0,0 +1,98 @@
+package peer
+
+import "container/list"
+
+// msgQueue owns the two priority lists and their combined resource accounting.
+// Message-specific retained-memory estimates are supplied in outgoingMsg, so
+// this type remains independent of Ping, Pong, and other wire semantics.
+type msgQueue struct {
+	// priorityMsgs stores messages that must be selected before any lazy
+	// message while preserving insertion order within the priority class.
+	priorityMsgs list.List
+
+	// lazyMsgs stores deferrable messages in insertion order so front can
+	// service them only when the strict-priority list is empty.
+	lazyMsgs list.List
+
+	// limits is the immutable per-peer policy used by push to decide when
+	// retaining another message requires disconnecting the connection.
+	limits queueLimits
+
+	// numMsgs mirrors the combined list length, avoiding a traversal on
+	// every insertion while enforcing the total message-count bound.
+	numMsgs int
+
+	// numBytes mirrors the combined queueCost values, letting push and pop
+	// enforce retained-memory bounds without knowing wire message shapes.
+	numBytes int
+}
+
+// newMsgQueue constructs an empty queue with per-peer resource bounds. The
+// list zero values are ready for use, so only the immutable limits are stored.
+func newMsgQueue(limits queueLimits) *msgQueue {
+	return &msgQueue{limits: limits}
+}
+
+// front returns the next message using strict priority ordering. Returning a
+// nil element lets queueHandler disable its send case without a second select.
+func (q *msgQueue) front() (*list.Element, outgoingMsg) {
+	elem := q.priorityMsgs.Front()
+	if elem == nil {
+		elem = q.lazyMsgs.Front()
+	}
+	if elem == nil {
+		return nil, outgoingMsg{}
+	}
+
+	return elem, msgFromElement(elem)
+}
+
+// msgFromElement enforces msgQueue's internal list invariant. A panic denotes
+// a programming error because push is the only method that inserts elements.
+func msgFromElement(elem *list.Element) outgoingMsg {
+	msg, ok := elem.Value.(outgoingMsg)
+	if !ok {
+		panic("msgQueue element is not an outgoingMsg")
+	}
+
+	return msg
+}
+
+// push appends a message only when its prospective count and byte totals fit
+// the combined backlog limits. A false result leaves the queue unchanged so
+// the rejected message cannot make retained memory exceed the advertised cap;
+// the owner can then disconnect without dropping an already-accepted message.
+func (q *msgQueue) push(msg outgoingMsg) bool {
+	nextNumMsgs := q.numMsgs + 1
+	nextNumBytes := q.numBytes + msg.queueCost
+	if nextNumMsgs > q.limits.maxMsgs ||
+		nextNumBytes > q.limits.maxBytes {
+
+		return false
+	}
+
+	if msg.priority {
+		q.priorityMsgs.PushBack(msg)
+	} else {
+		q.lazyMsgs.PushBack(msg)
+	}
+
+	q.numMsgs = nextNumMsgs
+	q.numBytes = nextNumBytes
+
+	return true
+}
+
+// pop removes the selected front element and releases the exact cost charged
+// at insertion, avoiding both message-type knowledge and cost recomputation.
+func (q *msgQueue) pop(elem *list.Element) {
+	msg := msgFromElement(elem)
+	if msg.priority {
+		q.priorityMsgs.Remove(elem)
+	} else {
+		q.lazyMsgs.Remove(elem)
+	}
+
+	q.numMsgs--
+	q.numBytes -= msg.queueCost
+}

### peer/ping_limits.go
@@ -0,0 +1,53 @@
+package peer
+
+import "golang.org/x/time/rate"
+
+const (
+	// pongReplyRate refills the reply budget quickly enough for normal
+	// keepalives while bounding sustained amplification from remote Pings.
+	pongReplyRate rate.Limit = 1
+
+	// pongReplyBurst absorbs short keepalive bursts without suppressing a
+	// reply before the sustained-rate policy has time to take effect.
+	pongReplyBurst = 20
+
+	// pingFloodRate admits substantially more inbound Pings than an honest
+	// keepalive cadence while placing a finite bound on sustained floods.
+	pingFloodRate rate.Limit = 10
+
+	// pingFloodBurst tolerates transient bursts before the peer is treated
+	// as a flood source and disconnected by the read loop.
+	pingFloodBurst = 200
+)
+
+// pingLimits holds the stateful limiters for the two inbound Ping policies.
+// Keeping them together makes their different outcomes explicit without
+// exposing fixed denial-of-service thresholds as operator configuration.
+type pingLimits struct {
+	// pongLimiter controls whether a valid Ping receives a Pong. Exhausting
+	// this limiter suppresses the reply but leaves the connection active.
+	pongLimiter *rate.Limiter
+
+	// pingLimiter counts every inbound Ping. Exhausting this limiter
+	// disconnects the peer, including for Pings that request no reply.
+	pingLimiter *rate.Limiter
+}
+
+// defaultPingLimits constructs independent limiter state for a new peer. The
+// selected rates leave ample room above normal keepalive traffic while
+// separating reply suppression from flood teardown.
+func defaultPingLimits() pingLimits {
+	return pingLimits{
+		// Refill one Pong per second and absorb a 20-Ping burst,
+		// leaving wide headroom above honest keepalives.
+		pongLimiter: rate.NewLimiter(
+			pongReplyRate, pongReplyBurst,
+		),
+
+		// Permit ten Pings per second and a burst of 200 before
+		// treating the connection as a flood source.
+		pingLimiter: rate.NewLimiter(
+			pingFloodRate, pingFloodBurst,
+		),
+	}
+}

### peer/queue_cost.go
@@ -0,0 +1,143 @@
+package peer
+
+import (
+	"net"
+
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/tor"
+)
+
+const (
+	// queuedFeatureOverhead covers the retained feature-vector object and
+	// its map allocation independently of the encoded feature-bit span.
+	queuedFeatureOverhead = 64
+
+	// queuedFeatureEntryOverhead conservatively covers the map bucket,
+	// key, top-hash, and overflow storage retained for each populated bit.
+	queuedFeatureEntryOverhead = 16
+
+	// queuedAddrOverhead covers each address interface and decoded object;
+	// separately allocated fields are added by nodeAnnouncementAddrCost.
+	queuedAddrOverhead = 64
+
+	// queuedShortChanIDSize is the in-memory width of a decoded SCID,
+	// including the padding after its uint16 transaction position.
+	queuedShortChanIDSize = 12
+	// queuedTimestampPairSize covers both retained uint32 update times.
+	queuedTimestampPairSize = 8
+)
+
+// featureMapCost charges one fixed allocation allowance, the encoded bit span,
+// and every populated map entry. Separating span from population keeps sparse
+// high bits affordable while preventing dense vectors from hiding map memory.
+func featureMapCost(features *lnwire.RawFeatureVector) int {
+	if features == nil {
+		return 0
+	}
+
+	return queuedFeatureOverhead + features.SerializeSize() +
+		features.NumFeatures()*queuedFeatureEntryOverhead
+}
+
+// nodeAnnouncementAddrCost charges the retained address slice, concrete
+// objects, and their separately allocated fields. The type switch mirrors the
+// address shapes produced by lnwire decoding without reserializing them.
+func nodeAnnouncementAddrCost(addrs []net.Addr) int {
+	cost := len(addrs) * queuedAddrOverhead
+	for _, addr := range addrs {
+		switch addr := addr.(type) {
+		case *net.TCPAddr:
+			cost += len(addr.IP) + len(addr.Zone)
+		case *tor.OnionAddr:
+			cost += len(addr.OnionService)
+		case *lnwire.DNSAddress:
+			cost += len(addr.Hostname)
+		case *lnwire.OpaqueAddrs:
+			cost += len(addr.Payload)
+		}
+	}
+
+	return cost
+}
+
+// msgCost estimates memory retained by an outgoing message without
+// serializing it. The independent count limit still bounds message shapes
+// whose dynamic memory is not included in this targeted estimate. The cases
+// below enumerate peer-controlled payloads that can arrive in bulk; CommitSig
+// signatures are the known material undercount, but commitment flow control
+// bounds them by channel count rather than permitting a bulk peer flood.
+func (l queueLimits) msgCost(msg lnwire.Message) int {
+	switch msg := msg.(type) {
+	// Pong payloads alias one server-wide buffer, so only their wrapper and
+	// list storage contribute additional retained queue memory.
+	case *lnwire.Pong:
+		return l.msgOverhead
+
+	// Failure reasons are preserved byte-for-byte when forwarded upstream
+	// and are the largest variable payload a remote peer can drive in bulk.
+	case *lnwire.UpdateFailHTLC:
+		return l.msgOverhead + len(msg.Reason)
+
+	// The onion packet is inline rather than a slice, so charge it together
+	// with any separately retained extra data.
+	case *lnwire.UpdateAddHTLC:
+		return l.msgOverhead + lnwire.OnionPacketSize +
+			len(msg.ExtraData)
+
+	// Error and Warning retain peer-controlled diagnostic payloads, so
+	// charge their backing bytes against the queue memory limit.
+	case *lnwire.Error:
+		return l.msgOverhead + len(msg.Data)
+
+	case *lnwire.Warning:
+		return l.msgOverhead + len(msg.Data)
+
+	// Forwarded onion messages retain a fresh peer-controlled blob. Charge
+	// those backing bytes so many maximum-sized messages cannot outgrow the
+	// queue's retained-memory budget while paying only fixed overhead.
+	case *lnwire.OnionMessage:
+		return l.msgOverhead + len(msg.OnionBlob)
+
+	// V1 gossip forwarding preserves peer-authored opaque extensions in
+	// each queued message. Feature maps and node addresses are decoded into
+	// larger retained objects, so charge those allocations as well.
+	case *lnwire.ChannelAnnouncement1:
+		return l.msgOverhead + len(msg.ExtraOpaqueData) +
+			featureMapCost(msg.Features)
+
+	case *lnwire.NodeAnnouncement1:
+		return l.msgOverhead + len(msg.ExtraOpaqueData) +
+			featureMapCost(msg.Features) +
+			nodeAnnouncementAddrCost(msg.Addresses)
+
+	case *lnwire.ChannelUpdate1:
+		return l.msgOverhead + len(msg.ExtraOpaqueData)
+
+	// Gossip queries retain decoded slices that can be much larger than
+	// their compressed wire forms, along with any unknown TLV bytes.
+	case *lnwire.QueryChannelRange:
+		cost := l.msgOverhead + len(msg.ExtraData)
+		if msg.QueryOptions != nil {
+			features := lnwire.RawFeatureVector(*msg.QueryOptions)
+			cost += featureMapCost(&features)
+		}
+
+		return cost
+
+	case *lnwire.QueryShortChanIDs:
+		return l.msgOverhead +
+			len(msg.ShortChanIDs)*queuedShortChanIDSize +
+			len(msg.ExtraData)
+
+	case *lnwire.ReplyChannelRange:
+		return l.msgOverhead +
+			len(msg.ShortChanIDs)*queuedShortChanIDSize +
+			len(msg.Timestamps)*queuedTimestampPairSize +
+			len(msg.ExtraData)
+
+	// Other messages receive the fixed charge. Their total count is still
+	// bounded even if they retain dynamic data not enumerated above.
+	default:
+		return l.msgOverhead
+	}
+}

### peer/queue_limits.go
@@ -0,0 +1,48 @@
+package peer
+
+const (
+	// maxQueuedMsgs caps fixed-cost message floods even when their retained
+	// byte charge remains well below the independent memory budget.
+	maxQueuedMsgs = 10000
+
+	// maxQueuedBytes bounds explicitly charged per-peer backlog memory at
+	// approximately 16 MiB before the owning connection is disconnected.
+	maxQueuedBytes = 16 << 20
+
+	// queuedMsgOverhead conservatively charges the Go message wrapper and
+	// list element retained for every queued wire message.
+	queuedMsgOverhead = 128
+)
+
+// queueLimits groups the count and retained-memory bounds applied to one
+// peer's outgoing backlog. The values remain private because they protect
+// internal resource ownership rather than define user-facing behavior.
+type queueLimits struct {
+	// maxMsgs prevents fixed-size messages from growing the queue without
+	// bound even when their charged byte cost is small.
+	maxMsgs int
+
+	// maxBytes caps the explicitly charged retained memory. Message shapes
+	// not included in the estimate remain protected by maxMsgs.
+	maxBytes int
+
+	// msgOverhead charges the message wrapper and list element even when a
+	// payload aliases memory owned elsewhere, as Pongs do.
+	msgOverhead int
+}
+
+// defaultQueueLimits returns the private resource bounds applied to each
+// peer's outgoing backlog. Keeping them in one value gives the producer and
+// queue owner the same immutable accounting policy.
+func defaultQueueLimits() queueLimits {
+	return queueLimits{
+		// Bound both cheap-message floods and approximately 16 MiB
+		// of explicitly charged retained queue memory.
+		maxMsgs:  maxQueuedMsgs,
+		maxBytes: maxQueuedBytes,
+
+		// A retained Pong costs about 104 bytes across its wrapper
+		// and list element, rounded up for accounting.
+		msgOverhead: queuedMsgOverhead,
+	}
+}

### queue/circular_buf.go
@@ -2,6 +2,7 @@ package queue
 
 import (
 	"errors"
+	"sync"
 )
 
 // errInvalidSize is returned when an invalid size for a buffer is provided.
@@ -11,6 +12,10 @@ var errInvalidSize = errors.New("buffer size must be > 0")
 // overwrites the oldest item in the buffer when a new item needs to be
 // written.
 type CircularBuffer struct {
+	// mu protects total and items so peer error writers and RPC readers can
+	// safely share one buffer across their independent goroutines.
+	mu sync.RWMutex
+
 	// total is the total number of items that have been added to the
 	// buffer.
 	total int
@@ -36,14 +41,18 @@ func NewCircularBuffer(size int) (*CircularBuffer, error) {
 	}, nil
 }
 
-// index returns the index that should be written to next.
+// index returns the index that should be written to next. Callers must hold
+// either mu for writes or mu's read lock while inspecting the current index.
 func (c *CircularBuffer) index() int {
 	return c.total % len(c.items)
 }
 
 // Add adds an item to the buffer, overwriting the oldest item if the buffer
 // is full.
 func (c *CircularBuffer) Add(item interface{}) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	// Set the item in the next free index in the items array.
 	c.items[c.index()] = item
 
@@ -54,6 +63,9 @@ func (c *CircularBuffer) Add(item interface{}) {
 // List returns a copy of the items in the buffer ordered from the oldest to
 // newest item.
 func (c *CircularBuffer) List() []interface{} {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	size := cap(c.items)
 	index := c.index()
 
@@ -98,11 +110,17 @@ func (c *CircularBuffer) List() []interface{} {
 
 // Total returns the total number of items that have been added to the buffer.
 func (c *CircularBuffer) Total() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	return c.total
 }
 
 // Latest returns the item that was most recently added to the buffer.
 func (c *CircularBuffer) Latest() interface{} {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	// If no items have been added yet, return nil.
 	if c.total == 0 {
 		return nil

### queue/circular_buf_test.go
@@ -2,7 +2,10 @@ package queue
 
 import (
 	"reflect"
+	"sync"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 // TestNewCircularBuffer tests the size parameter check when creating a circular
@@ -196,3 +199,61 @@ func TestLatest(t *testing.T) {
 		})
 	}
 }
+
+// TestCircularBufferConcurrentAccess verifies that peer error writers and RPC
+// readers can safely share a buffer without corrupting its order or counters.
+func TestCircularBufferConcurrentAccess(t *testing.T) {
+	// Arrange: Create a small wrapping buffer and two finite workers behind
+	// one start gate. The writer repeatedly wraps the storage while the
+	// reader exercises every public observation method against those writes.
+	const (
+		bufferSize = 8
+		iterations = 1000
+	)
+	buffer, err := NewCircularBuffer(bufferSize)
+	require.NoError(t, err)
+
+	start := make(chan struct{})
+	var workers sync.WaitGroup
+	workers.Add(2)
+
+	go func() {
+		defer workers.Done()
+		<-start
+
+		for i := 0; i < iterations; i++ {
+			buffer.Add(i)
+		}
+	}()
+
+	go func() {
+		defer workers.Done()
+		<-start
+
+		for i := 0; i < iterations; i++ {
+			buffer.List()
+			buffer.Latest()
+			buffer.Total()
+		}
+	}()
+
+	// Act: Release both workers together and join them before inspecting
+	// results, keeping test assertions in the owning goroutine.
+	close(start)
+	workers.Wait()
+
+	// Assert: Every write is counted and the retained wrapping window is
+	// still ordered from oldest to newest after concurrent observations.
+	require.Equal(t, iterations, buffer.Total())
+	require.Equal(t, iterations-1, buffer.Latest())
+	require.Equal(t, []interface{}{
+		iterations - 8,
+		iterations - 7,
+		iterations - 6,
+		iterations - 5,
+		iterations - 4,
+		iterations - 3,
+		iterations - 2,
+		iterations - 1,
+	}, buffer.List())
+}