diff --git a/docs/release-notes/release-notes-0.20.4.md b/docs/release-notes/release-notes-0.20.4.md
index cdef5fdbfe..c40a893464 100644
--- a/docs/release-notes/release-notes-0.20.4.md
+++ b/docs/release-notes/release-notes-0.20.4.md
@@ -21,6 +21,10 @@
 
 # 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 funding attempts [now return
   cleanly](https://github.com/lightningnetwork/lnd/pull/11035) when their
   pending wallet reservation is no longer present.
diff --git a/docs/release-notes/release-notes-0.21.3.md b/docs/release-notes/release-notes-0.21.3.md
index e56ef10f73..f26af10977 100644
--- a/docs/release-notes/release-notes-0.21.3.md
+++ b/docs/release-notes/release-notes-0.21.3.md
@@ -21,6 +21,10 @@
 
 # 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 funding attempts [now return
   cleanly](https://github.com/lightningnetwork/lnd/pull/11035) when their
   pending wallet reservation is no longer present.
diff --git a/peer/brontide.go b/peer/brontide.go
index 6b2a3ad332..5b42d90051 100644
--- a/peer/brontide.go
+++ b/peer/brontide.go
@@ -2,7 +2,6 @@ package peer
 
 import (
 	"bytes"
-	"container/list"
 	"context"
 	"errors"
 	"fmt"
@@ -112,6 +111,11 @@ 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")
 )
 
 // outgoingMsg packages an lnwire.Message to be sent out on the wire, along with
@@ -121,6 +125,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 +593,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 +760,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 +2332,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 +2387,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 +3124,67 @@ 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
+
+			p.failQueueOverflow(queue.numMsgs, queue.numBytes)
+
+			return
+
+		case <-p.cg.Done():
+			return
+		}
+	}
+}
+
+// failQueueOverflow tears the connection down after the peer's outgoing
+// message queue has grown past its bounds, then keeps that queue serviced
+// until teardown completes. We disconnect rather than drop or block: dropping
+// would punch a hole in an ordered protocol stream, while blocking would push
+// backpressure onto whichever subsystem happened to be sending.
+//
+// NOTE: This blocks until the peer's context is cancelled, so it must be
+// called from the queueHandler goroutine itself.
+func (p *Brontide) failQueueOverflow(numQueued, queuedBytes int) {
+	err := fmt.Errorf("outgoing message queue exceeded bounds: "+
+		"messages=%d, bytes=%d", numQueued, queuedBytes)
+	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 +3214,19 @@ 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: msgQueueCost(
+			msg, p.queueLimits.msgOverhead,
+		),
+	}
+
 	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))
diff --git a/peer/brontide_test.go b/peer/brontide_test.go
index c9dd25dbf7..47e0c8cd3a 100644
--- a/peer/brontide_test.go
+++ b/peer/brontide_test.go
@@ -2,7 +2,9 @@ package peer
 
 import (
 	"bytes"
+	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,14 +19,17 @@ 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/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -1213,6 +1218,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 +1234,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 +1276,512 @@ 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
+		refillTime time.Duration
+	}{
+		{
+			name:       "Pong replies",
+			limiter:    limits.pongLimiter,
+			limit:      1,
+			burst:      20,
+			refillTime: time.Second,
+		},
+		{
+			name:       "Ping floods",
+			limiter:    limits.pingLimiter,
+			limit:      10,
+			burst:      200,
+			refillTime: 100 * time.Millisecond,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Fix one synthetic timestamp and confirm the
+			// limiter uses the intended production configuration.
+			now := time.Now()
+			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,
+			// then advance exactly one production refill interval.
+			atBoundary := test.limiter.AllowN(now, test.burst)
+			pastBoundary := test.limiter.AllowN(now, 1)
+			afterRefill := test.limiter.AllowN(
+				now.Add(test.refillTime), 1,
+			)
+
+			// Assert: The boundary is inclusive, its successor is
+			// rejected, and one interval restores one token.
+			require.True(t, atBoundary)
+			require.False(t, pastBoundary)
+			require.True(t, afterRefill)
+		})
+	}
+}
+
+// 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 and injects it
+	// through the mock connection's normal reader path.
+	writePing := func() {
+		var b bytes.Buffer
+		_, err := lnwire.WriteMessage(&b, lnwire.NewPing(1), 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 pings, consuming the first expected pong before
+	// allowing a bounded window for an incorrect second reply.
+	writePing()
+	_, err = fn.RecvOrTimeout(params.mockConn.writtenMessages, timeout)
+	require.NoError(t, err)
+
+	writePing()
+	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()
+}
+
+// TestPeerQueueHandlerBoundsBacklog verifies that crossing a queue bound
+// tears down the peer connection.
+func TestPeerQueueHandlerBoundsBacklog(t *testing.T) {
+	t.Parallel()
+
+	// Arrange: Derive data-only workloads from the production limits. The
+	// count case uses fixed-cost Pongs. The byte case retains distinct
+	// maximum onion blobs and stays below the independent count cap.
+	const onionBlobSize = 65000
+	limits := defaultQueueLimits()
+	tests := []struct {
+		name        string
+		msgType     lnwire.MessageType
+		payloadSize int
+		numMsgs     int
+	}{
+		{
+			name:    "message count",
+			msgType: lnwire.MsgPong,
+			numMsgs: limits.maxMsgs + 1,
+		},
+		{
+			name:        "message bytes",
+			msgType:     lnwire.MsgOnionMessage,
+			payloadSize: onionBlobSize,
+			numMsgs: limits.maxBytes/
+				(limits.msgOverhead+onionBlobSize) + 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Start only the queue handler so no writer
+			// drains sendQueue or hides the staged backlog.
+			params := createTestPeer(t)
+			peer := params.peer
+			startTestQueueHandler(peer)
+
+			// Act: Build each row through one path. Onion messages
+			// use fresh slices, modeling memory a forwarding peer
+			// can make us retain.
+			for i := 0; i < test.numMsgs; i++ {
+				var msg lnwire.Message
+				switch test.msgType {
+				case lnwire.MsgPong:
+					msg = lnwire.NewPong(nil)
+
+				case lnwire.MsgOnionMessage:
+					onionBlob := make(
+						[]byte, test.payloadSize,
+					)
+					msg = &lnwire.OnionMessage{
+						OnionBlob: onionBlob,
+					}
+
+				default:
+					t.Fatalf(
+						"unsupported queue message: %v",
+						test.msgType,
+					)
+				}
+
+				peer.queueMsg(msg, nil)
+			}
+
+			// Assert: Cancellation proves overflow, and waiting
+			// proves the asynchronous handler exits cleanly.
+			_, 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 so every table
+	// expectation follows the queue policy without duplicating its value.
+	limits := defaultQueueLimits()
+	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,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Arrange: Select one message shape and its exact
+			// fixed-plus-dynamic cost from the test table.
+
+			// Act: Evaluate its non-serializing queue charge.
+			actual := msgQueueCost(test.msg, limits.msgOverhead)
+
+			// Assert: Equality proves this shape charges only
+			// the enumerated payload plus fixed overhead.
+			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:
+	}
+}
+
 // 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.
diff --git a/peer/msg_queue.go b/peer/msg_queue.go
new file mode 100644
index 0000000000..91e4f26af5
--- /dev/null
+++ b/peer/msg_queue.go
@@ -0,0 +1,76 @@
+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 list.List
+	lazyMsgs     list.List
+	limits       queueLimits
+	numMsgs      int
+	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 to its priority list and charges its immutable cost
+// to the combined backlog. A false result tells the owner to disconnect rather
+// than dropping an ordered protocol message or blocking an arbitrary producer.
+func (q *msgQueue) push(msg outgoingMsg) bool {
+	if msg.priority {
+		q.priorityMsgs.PushBack(msg)
+	} else {
+		q.lazyMsgs.PushBack(msg)
+	}
+
+	q.numMsgs++
+	q.numBytes += msg.queueCost
+
+	return q.numMsgs <= q.limits.maxMsgs &&
+		q.numBytes <= q.limits.maxBytes
+}
+
+// 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
+}
diff --git a/peer/ping_limits.go b/peer/ping_limits.go
new file mode 100644
index 0000000000..e32116f104
--- /dev/null
+++ b/peer/ping_limits.go
@@ -0,0 +1,31 @@
+package peer
+
+import "golang.org/x/time/rate"
+
+// 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(1, 20),
+
+		// Permit ten Pings per second and a burst of 200 before
+		// treating the connection as a flood source.
+		pingLimiter: rate.NewLimiter(10, 200),
+	}
+}
diff --git a/peer/queue_limits.go b/peer/queue_limits.go
new file mode 100644
index 0000000000..71eed3086b
--- /dev/null
+++ b/peer/queue_limits.go
@@ -0,0 +1,79 @@
+package peer
+
+import "github.com/lightningnetwork/lnd/lnwire"
+
+// 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:  10000,
+		maxBytes: 16 << 20,
+
+		// A retained Pong costs about 104 bytes across its wrapper
+		// and list element, rounded up for accounting.
+		msgOverhead: 128,
+	}
+}
+
+// msgQueueCost 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. CommitSig
+// signatures are the known material undercount, but commitment flow control
+// bounds them by channel count rather than permitting a bulk peer flood.
+func msgQueueCost(msg lnwire.Message, overhead int) 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 overhead
+
+	// 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 overhead + 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 overhead + 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 overhead + len(msg.Data)
+
+	case *lnwire.Warning:
+		return overhead + 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 overhead + len(msg.OnionBlob)
+
+	// Other messages receive the fixed charge. Their total count is still
+	// bounded even if they retain dynamic data not enumerated above.
+	default:
+		return overhead
+	}
+}