htlcswitch: bound peer-controlled channel ingress

### htlcswitch/link.go
@@ -363,6 +363,14 @@ type channelLink struct {
 	// forwarded sent by the switch.
 	mailBox MailBox
 
+	// mailBoxIngressMtx guards mailBoxIngressFailed and serializes peer
+	// message admission into the mailbox.
+	mailBoxIngressMtx sync.Mutex
+
+	// mailBoxIngressFailed is set after the first peer message admission
+	// failure so later messages cannot be processed across a gap.
+	mailBoxIngressFailed bool
+
 	// upstream is a channel that new messages sent from the remote peer to
 	// the local peer will be sent across.
 	upstream chan lnwire.Message
@@ -395,6 +403,11 @@ type channelLink struct {
 	// log is a link-specific logging instance.
 	log btclog.Logger
 
+	// warningLogged and unknownMessageLogged track whether each non-fatal
+	// message class has already been recorded for this link lifetime.
+	warningLogged        bool
+	unknownMessageLogged bool
+
 	// isOutgoingAddBlocked tracks whether the channelLink can send an
 	// UpdateAddHTLC.
 	isOutgoingAddBlocked atomic.Bool
@@ -1862,14 +1875,20 @@ func (l *channelLink) handleUpstreamMsg(ctx context.Context,
 	// log it and move on. We choose not to disconnect from our peer,
 	// although we "MAY" do so according to the specification.
 	case *lnwire.Warning:
-		l.log.Warnf("received warning message from peer: %v",
-			msg.Warning())
+		if !l.warningLogged {
+			l.log.Warnf("received warning message from peer: %v",
+				msg.Warning())
+			l.warningLogged = true
+		}
 
 	case *lnwire.Error:
 		l.processRemoteError(msg)
 
 	default:
-		l.log.Warnf("received unknown message of type %T", msg)
+		if !l.unknownMessageLogged {
+			l.log.Warnf("received unknown message of type %T", msg)
+			l.unknownMessageLogged = true
+		}
 	}
 
 	if err != nil {
@@ -2804,10 +2823,23 @@ func (l *channelLink) HandleChannelUpdate(message lnwire.Message) {
 	default:
 	}
 
+	l.mailBoxIngressMtx.Lock()
+	if l.mailBoxIngressFailed {
+		l.mailBoxIngressMtx.Unlock()
+		return
+	}
+
 	err := l.mailBox.AddMessage(message)
-	if err != nil {
-		l.log.Errorf("failed to add Message to mailbox: %v", err)
+	if err == nil {
+		l.mailBoxIngressMtx.Unlock()
+		return
 	}
+
+	l.mailBoxIngressFailed = true
+	l.mailBoxIngressMtx.Unlock()
+
+	l.log.Errorf("failed to add Message to mailbox: %v", err)
+	go l.cfg.Peer.Disconnect(err)
 }
 
 // updateChannelFee updates the commitment fee-per-kw on this channel by
@@ -4583,6 +4615,16 @@ func (l *channelLink) processRemoteRevokeAndAck(ctx context.Context,
 // processRemoteUpdateFee takes an `UpdateFee` msg sent from the remote and
 // processes it.
 func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
+	// BOLT 2 only permits the channel initiator to send fee updates.
+	// Validate the sender's role before applying message-specific
+	// calculations.
+	if l.channel.IsInitiator() {
+		err := fmt.Errorf("received fee update as initiator")
+		l.failf(LinkFailureError{code: ErrInvalidUpdate}, "%v", err)
+
+		return err
+	}
+
 	// Check and see if their proposed fee-rate would make us exceed the fee
 	// threshold.
 	fee := chainfee.SatPerKWeight(msg.FeePerKw)
@@ -4601,8 +4643,9 @@ func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
 
 	if isDust {
 		// The proposed fee-rate makes us exceed the fee threshold.
-		l.failf(LinkFailureError{code: ErrInternalError},
-			"fee threshold exceeded: %v", err)
+		err := fmt.Errorf("fee threshold exceeded")
+		l.failf(LinkFailureError{code: ErrInternalError}, "%v", err)
+
 		return err
 	}
 
@@ -4611,6 +4654,7 @@ func (l *channelLink) processRemoteUpdateFee(msg *lnwire.UpdateFee) error {
 	if err := l.channel.ReceiveUpdateFee(fee); err != nil {
 		l.failf(LinkFailureError{code: ErrInvalidUpdate},
 			"error receiving fee update: %v", err)
+
 		return err
 	}
 

### htlcswitch/link_fee_update_test.go
@@ -0,0 +1,306 @@
+package htlcswitch
+
+import (
+	"bytes"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btclog/v2"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lnpeer"
+	"github.com/lightningnetwork/lnd/lnwallet"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/require"
+)
+
+// mailboxAdmissionPeer records disconnect requests made by a channel link.
+type mailboxAdmissionPeer struct {
+	*lnpeer.MockPeer
+
+	disconnected chan error
+}
+
+// Disconnect records the error supplied by the channel link.
+func (p *mailboxAdmissionPeer) Disconnect(err error) {
+	p.disconnected <- err
+}
+
+// mailboxAdmissionTestBox fails its first message admission and records the
+// number of admission attempts.
+type mailboxAdmissionTestBox struct {
+	MailBox
+
+	mu       sync.Mutex
+	addCalls int
+}
+
+// AddMessage records an admission attempt and fails the first one.
+func (m *mailboxAdmissionTestBox) AddMessage(lnwire.Message) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.addCalls++
+	if m.addCalls == 1 {
+		return errWireMessageQueueFull
+	}
+
+	return nil
+}
+
+// calls returns the number of message admission attempts.
+func (m *mailboxAdmissionTestBox) calls() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	return m.addCalls
+}
+
+// newLinkCapturingLogger returns a logger backed by an in-memory buffer.
+func newLinkCapturingLogger() (btclog.Logger, *bytes.Buffer) {
+	buf := &bytes.Buffer{}
+	handler := btclog.NewDefaultHandler(buf, btclog.WithNoTimestamp())
+
+	return btclog.NewSLogger(handler), buf
+}
+
+// TestProcessRemoteUpdateFeeRoleValidation checks that fee update role
+// validation is performed at the link boundary.
+func TestProcessRemoteUpdateFeeRoleValidation(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, bobChannel, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	newLink := func(channel *lnwallet.LightningChannel) *channelLink {
+		link, ok := NewChannelLink(ChannelLinkConfig{
+			DisallowQuiescence: true,
+			OnChannelFailure: func(lnwire.ChannelID,
+				lnwire.ShortChannelID, LinkFailureError) {
+			},
+		}, channel).(*channelLink)
+		require.True(t, ok)
+
+		return link
+	}
+
+	t.Run("unauthorized sender", func(t *testing.T) {
+		link := newLink(aliceChannel)
+
+		err := link.processRemoteUpdateFee(&lnwire.UpdateFee{})
+		require.EqualError(t, err, "received fee update as initiator")
+		require.True(t, link.failed)
+	})
+
+	t.Run("authorized sender", func(t *testing.T) {
+		link := newLink(bobChannel)
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		link.mailBox = mailbox
+
+		feeRate := bobChannel.CommitFeeRate() + 1
+		err := link.processRemoteUpdateFee(&lnwire.UpdateFee{
+			FeePerKw: uint32(feeRate),
+		})
+		require.NoError(t, err)
+		require.False(t, link.failed)
+		require.True(t, bobChannel.NeedCommitment())
+		require.Equal(t, feeRate, mailbox.feeRate)
+	})
+}
+
+// TestProcessRemoteUpdateFeeExposureError checks that exceeding the fee
+// exposure limit returns the error used to fail the link.
+func TestProcessRemoteUpdateFeeExposureError(t *testing.T) {
+	t.Parallel()
+
+	_, bobChannel, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		DisallowQuiescence: true,
+		MaxFeeExposure:     1,
+		OnChannelFailure: func(lnwire.ChannelID,
+			lnwire.ShortChannelID, LinkFailureError) {
+		},
+	}, bobChannel).(*channelLink)
+	require.True(t, ok)
+
+	err = link.processRemoteUpdateFee(&lnwire.UpdateFee{
+		FeePerKw: 1000,
+	})
+	require.EqualError(t, err, "fee threshold exceeded")
+	require.True(t, link.failed)
+}
+
+// TestLinkLogDeduplication checks that repeated non-fatal message classes are
+// only recorded once during a link lifetime.
+func TestLinkLogDeduplication(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+	logger, logBuffer := newLinkCapturingLogger()
+	link.log = logger
+
+	for i := 0; i < 2; i++ {
+		link.handleUpstreamMsg(t.Context(), &lnwire.Warning{})
+		link.handleUpstreamMsg(
+			t.Context(), &lnwire.ChannelReestablish{},
+		)
+	}
+
+	warningCount := strings.Count(
+		logBuffer.String(), "received warning message from peer",
+	)
+	require.Equal(t, 1, warningCount)
+	require.Equal(
+		t, 1, strings.Count(
+			logBuffer.String(), "received unknown message of type",
+		),
+	)
+}
+
+// TestChannelMessageAdmissionError checks that an admission error reconnects
+// the ordered channel message stream instead of omitting a message.
+func TestChannelMessageAdmissionError(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 1),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := newMemoryMailBox(&mailBoxConfig{})
+	link.mailBox = mailbox
+	for i := 0; i < maxWireMessages; i++ {
+		require.NoError(t, mailbox.AddMessage(&lnwire.UpdateFee{}))
+	}
+
+	link.HandleChannelUpdate(&lnwire.UpdateFee{})
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("mailbox admission error did not disconnect peer")
+	}
+}
+
+// TestChannelMessageAdmissionFailureLatch checks that a link stops admitting
+// peer messages after its first mailbox admission failure.
+func TestChannelMessageAdmissionFailureLatch(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 2),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := &mailboxAdmissionTestBox{}
+	link.mailBox = mailbox
+	logger, logBuffer := newLinkCapturingLogger()
+	link.log = logger
+
+	link.HandleChannelUpdate(&lnwire.UpdateFee{})
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("mailbox admission error did not disconnect peer")
+	}
+
+	link.HandleChannelUpdate(&lnwire.CommitSig{})
+
+	require.Equal(t, 1, mailbox.calls())
+	require.Equal(
+		t, 1, strings.Count(
+			logBuffer.String(), "failed to add Message to mailbox",
+		),
+	)
+	select {
+	case err := <-peer.disconnected:
+		t.Fatalf("unexpected second disconnect: %v", err)
+
+	default:
+	}
+}
+
+// TestChannelMessageSizeAdmissionError checks that a message-size admission
+// error reconnects the ordered channel message stream.
+func TestChannelMessageSizeAdmissionError(t *testing.T) {
+	t.Parallel()
+
+	aliceChannel, _, err := lnwallet.CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	peer := &mailboxAdmissionPeer{
+		MockPeer:     &lnpeer.MockPeer{},
+		disconnected: make(chan error, 1),
+	}
+	link, ok := NewChannelLink(ChannelLinkConfig{
+		Peer:               peer,
+		DisallowQuiescence: true,
+	}, aliceChannel).(*channelLink)
+	require.True(t, ok)
+
+	mailbox := newMemoryMailBox(&mailBoxConfig{})
+	link.mailBox = mailbox
+	msg := &lnwire.Warning{
+		Data: make([]byte, lnwire.MaxMsgBody-40),
+	}
+	for {
+		err := mailbox.AddMessage(msg)
+		if errors.Is(err, errWireMessageQueueFull) {
+			break
+		}
+		require.NoError(t, err)
+	}
+
+	link.HandleChannelUpdate(msg)
+
+	select {
+	case err := <-peer.disconnected:
+		require.ErrorIs(t, err, errWireMessageQueueFull)
+
+	case <-time.After(time.Second):
+		t.Fatal("message-size admission error did not disconnect peer")
+	}
+}

### htlcswitch/mailbox.go
@@ -14,6 +14,16 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 )
 
+const (
+	// maxWireMessages is the maximum number of ordered messages that can
+	// wait for a channel link. It accommodates a full commitment batch.
+	maxWireMessages = 1000
+
+	// maxWireBytes bounds the encoded size of messages that can wait for a
+	// channel link.
+	maxWireBytes = 4 * 1024 * 1024
+)
+
 var (
 	// ErrMailBoxShuttingDown is returned when the mailbox is interrupted by
 	// a shutdown request.
@@ -22,6 +32,12 @@ var (
 	// ErrPacketAlreadyExists signals that an attempt to add a packet failed
 	// because it already exists in the mailbox.
 	ErrPacketAlreadyExists = errors.New("mailbox already has packet")
+
+	// errWireMessageQueueFull signals that the wire-message queue has
+	// reached one of its admission budgets.
+	errWireMessageQueueFull = errors.New(
+		"mailbox wire message queue is full",
+	)
 )
 
 // MailBox is an interface which represents a concurrent-safe, in-order
@@ -122,6 +138,7 @@ type memoryMailBox struct {
 	cfg *mailBoxConfig
 
 	wireMessages *list.List
+	wireBytes    uint32
 	wireMtx      sync.Mutex
 	wireCond     *sync.Cond
 
@@ -160,6 +177,13 @@ type memoryMailBox struct {
 	isDust dustClosure
 }
 
+// queuedWireMessage stores a wire message and its encoded size charged to the
+// wire-message budget.
+type queuedWireMessage struct {
+	msg  lnwire.Message
+	size uint32
+}
+
 // newMemoryMailBox creates a new instance of the memoryMailBox.
 func newMemoryMailBox(cfg *mailBoxConfig) *memoryMailBox {
 	box := &memoryMailBox{
@@ -383,6 +407,7 @@ func (m *memoryMailBox) wireMailCourier() {
 			select {
 			case msgDone := <-m.msgReset:
 				m.wireMessages.Init()
+				m.wireBytes = 0
 				close(msgDone)
 			case <-m.quit:
 				m.wireCond.L.Unlock()
@@ -397,7 +422,9 @@ func (m *memoryMailBox) wireMailCourier() {
 		entry := m.wireMessages.Front()
 
 		//nolint:forcetypeassert
-		nextMsg := m.wireMessages.Remove(entry).(lnwire.Message)
+		queuedMsg := m.wireMessages.Remove(entry).(*queuedWireMessage)
+		m.wireBytes -= queuedMsg.size
+		nextMsg := queuedMsg.msg
 
 		// Now that we're done with the condition, we can unlock it to
 		// allow any callers to append to the end of our target queue.
@@ -411,6 +438,7 @@ func (m *memoryMailBox) wireMailCourier() {
 		case msgDone := <-m.msgReset:
 			m.wireCond.L.Lock()
 			m.wireMessages.Init()
+			m.wireBytes = 0
 			m.wireCond.L.Unlock()
 
 			close(msgDone)
@@ -560,10 +588,28 @@ func (m *memoryMailBox) pktMailCourier() {
 // NOTE: This method is safe for concrete use and part of the MailBox
 // interface.
 func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
+	msgSize, err := wireMessageSize(msg)
+	if err != nil {
+		return fmt.Errorf(
+			"unable to determine wire message size: %w", err,
+		)
+	}
+
 	// First, we'll lock the condition, and add the message to the end of
 	// the wire message inbox.
 	m.wireCond.L.Lock()
-	m.wireMessages.PushBack(msg)
+	if m.wireMessages.Len() >= maxWireMessages ||
+		m.wireBytes+msgSize > maxWireBytes {
+
+		m.wireCond.L.Unlock()
+		return errWireMessageQueueFull
+	}
+
+	m.wireMessages.PushBack(&queuedWireMessage{
+		msg:  msg,
+		size: msgSize,
+	})
+	m.wireBytes += msgSize
 	m.wireCond.L.Unlock()
 
 	// With the message added, we signal to the mailCourier that there are
@@ -573,6 +619,16 @@ func (m *memoryMailBox) AddMessage(msg lnwire.Message) error {
 	return nil
 }
 
+// wireMessageSize returns the serialized bytes charged to the wire-message
+// budget.
+func wireMessageSize(msg lnwire.Message) (uint32, error) {
+	if sizeableMsg, ok := msg.(lnwire.SizeableMessage); ok {
+		return sizeableMsg.SerializedSize()
+	}
+
+	return lnwire.MessageSerializedSize(msg)
+}
+
 // AddPacket appends a new message to the end of the packet queue.
 //
 // NOTE: This method is safe for concrete use and part of the MailBox

### htlcswitch/mailbox_test.go
@@ -1,11 +1,13 @@
 package htlcswitch
 
 import (
+	"errors"
 	prand "math/rand"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcutil/v2"
 	"github.com/davecgh/go-spew/spew"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -168,6 +170,96 @@ func TestMailBoxCouriers(t *testing.T) {
 	}
 }
 
+// TestMailBoxAdmissionBudgets checks message-count and serialized-size
+// admission behavior for the wire-message queue.
+func TestMailBoxAdmissionBudgets(t *testing.T) {
+	t.Parallel()
+
+	t.Run("message count", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		msg := &lnwire.UpdateFee{}
+
+		for i := 0; i < maxWireMessages; i++ {
+			require.NoError(t, mailbox.AddMessage(msg))
+		}
+
+		require.ErrorIs(
+			t, mailbox.AddMessage(msg), errWireMessageQueueFull,
+		)
+		require.Equal(t, maxWireMessages, mailbox.wireMessages.Len())
+		require.LessOrEqual(
+			t, mailbox.wireBytes, uint32(maxWireBytes),
+		)
+	})
+
+	t.Run("encoded bytes", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		msg := &lnwire.Warning{
+			Data: make([]byte, lnwire.MaxMsgBody-40),
+		}
+
+		for {
+			err := mailbox.AddMessage(msg)
+			if errors.Is(err, errWireMessageQueueFull) {
+				break
+			}
+			require.NoError(t, err)
+		}
+
+		require.Less(t, mailbox.wireMessages.Len(), maxWireMessages)
+		require.LessOrEqual(
+			t, mailbox.wireBytes, uint32(maxWireBytes),
+		)
+	})
+
+	t.Run("commitment message sizes", func(t *testing.T) {
+		_, pubKey := btcec.PrivKeyFromBytes(make([]byte, 32))
+		extraData := lnwire.ExtraOpaqueData{
+			0xfe, 0x00, 0x01, 0x00, 0x00, 0x03, 0x01, 0x02, 0x03,
+		}
+
+		messages := []lnwire.Message{
+			&lnwire.CommitSig{ExtraData: extraData},
+			&lnwire.RevokeAndAck{
+				NextRevocationKey: pubKey,
+				ExtraData:         extraData,
+			},
+			&lnwire.Stfu{ExtraData: extraData},
+		}
+		for _, msg := range messages {
+			mailbox := newMemoryMailBox(&mailBoxConfig{})
+			sizeableMsg, ok := msg.(lnwire.SizeableMessage)
+			require.True(t, ok)
+
+			expectedSize, err := sizeableMsg.SerializedSize()
+			require.NoError(t, err)
+
+			require.NoError(t, mailbox.AddMessage(msg))
+			require.Equal(t, expectedSize, mailbox.wireBytes)
+		}
+	})
+
+	t.Run("reset restores byte budget", func(t *testing.T) {
+		mailbox := newMemoryMailBox(&mailBoxConfig{})
+		mailbox.Start()
+		t.Cleanup(mailbox.Stop)
+
+		msg := &lnwire.Warning{
+			Data: make([]byte, lnwire.MaxMsgBody-40),
+		}
+		for {
+			err := mailbox.AddMessage(msg)
+			if errors.Is(err, errWireMessageQueueFull) {
+				break
+			}
+			require.NoError(t, err)
+		}
+
+		require.NoError(t, mailbox.ResetMessages())
+		require.NoError(t, mailbox.AddMessage(msg))
+	})
+}
+
 // TestMailBoxResetAfterShutdown tests that ResetMessages and ResetPackets
 // return ErrMailBoxShuttingDown after the mailbox has been stopped.
 func TestMailBoxResetAfterShutdown(t *testing.T) {