Merge pull request #11023 from ellemouton/update-fee-log

lnwallet+htlcswitch: align update_fee handling with BOLT 2

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:
 	}
 
-	err := l.mailBox.AddMessage(message)
-	if err != nil {
-		l.log.Errorf("failed to add Message to mailbox: %v", err)
+	l.mailBoxIngressMtx.Lock()
+	if l.mailBoxIngressFailed {
+		l.mailBoxIngressMtx.Unlock()
+		return
 	}
+
+	err := l.mailBox.AddMessage(message)
+	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) {

lnwallet/channel.go

@@ -9431,7 +9431,7 @@ func (lc *LightningChannel) UpdateFee(feePerKw chainfee.SatPerKWeight) error {
 		EntryType: FeeUpdate,
 	}
 
-	lc.updateLogs.Local.appendUpdate(pd)
+	lc.updateLogs.Local.appendFeeUpdate(pd)
 
 	return nil
 }
@@ -9504,7 +9504,7 @@ func (lc *LightningChannel) ReceiveUpdateFee(feePerKw chainfee.SatPerKWeight) er
 		EntryType: FeeUpdate,
 	}
 
-	lc.updateLogs.Remote.appendUpdate(pd)
+	lc.updateLogs.Remote.appendFeeUpdate(pd)
 
 	return nil
 }

lnwallet/channel_test.go

@@ -5375,8 +5375,9 @@ func TestFeeUpdateOldDiskFormat(t *testing.T) {
 				err)
 		}
 	}
-	// Check that the expected number of items is found in the logs.
-	expFee := numHTLCs / 5
+	// Replacement semantics retain the final pending fee value alongside
+	// all of the HTLCs.
+	expFee := 1
 	assertLogItems(expFee, numHTLCs)
 
 	// Now, Alice will send a new commitment to Bob, but we'll simulate a

lnwallet/update_log.go

@@ -95,6 +95,30 @@ func (u *updateLog) appendHtlc(pd *paymentDescriptor) {
 	u.logIndex++
 }
 
+// appendFeeUpdate appends a fee update unless the newest fee update hasn't yet
+// been committed to either commitment chain. In that case, only its fee is
+// replaced. Keeping the original descriptor and log index preserves a
+// contiguous update stream for persistence while avoiding redundant entries.
+func (u *updateLog) appendFeeUpdate(pd *paymentDescriptor) {
+	for entry := u.Back(); entry != nil; entry = entry.Prev() {
+		update := entry.Value
+		if update.EntryType != FeeUpdate {
+			continue
+		}
+
+		if update.addCommitHeights.Local == 0 &&
+			update.addCommitHeights.Remote == 0 {
+
+			update.Amount = pd.Amount
+			return
+		}
+
+		break
+	}
+
+	u.appendUpdate(pd)
+}
+
 // lookupHtlc attempts to look up an offered HTLC according to its offer
 // index. If the entry isn't found, then a nil pointer is returned.
 func (u *updateLog) lookupHtlc(i uint64) *paymentDescriptor {

lnwallet/update_log_test.go

@@ -0,0 +1,214 @@
+package lnwallet
+
+import (
+	"testing"
+
+	"github.com/btcsuite/btcd/btcutil/v2"
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/stretchr/testify/require"
+	"pgregory.net/rapid"
+)
+
+// TestAppendFeeUpdateReplacementSequences checks replacement behavior across
+// generated sequences of fee and commitment state transitions.
+func TestAppendFeeUpdateReplacementSequences(t *testing.T) {
+	t.Parallel()
+
+	type feeAction struct {
+		fee          uint32
+		commitLocal  bool
+		commitRemote bool
+		interleave   bool
+	}
+
+	actions := rapid.SliceOfN(
+		rapid.Custom(func(t *rapid.T) feeAction {
+			return feeAction{
+				fee: rapid.Uint32Range(1, 10_000_000).Draw(
+					t, "fee",
+				),
+				commitLocal: rapid.Bool().Draw(
+					t, "commit_local",
+				),
+				commitRemote: rapid.Bool().Draw(
+					t, "commit_remote",
+				),
+				interleave: rapid.Bool().Draw(
+					t, "interleave",
+				),
+			}
+		}), 1, 200,
+	)
+
+	rapid.Check(t, func(t *rapid.T) {
+		log := newUpdateLog(0, 0)
+		committed := make(map[*paymentDescriptor]struct{})
+
+		for i, action := range actions.Draw(t, "actions") {
+			if action.interleave {
+				log.appendUpdate(&paymentDescriptor{
+					LogIndex:  log.logIndex,
+					EntryType: Settle,
+				})
+			}
+
+			feeUpdate := &paymentDescriptor{
+				LogIndex: log.logIndex,
+				Amount: lnwire.NewMSatFromSatoshis(
+					btcutil.Amount(action.fee),
+				),
+				EntryType: FeeUpdate,
+			}
+			log.appendFeeUpdate(feeUpdate)
+
+			var currentFee *paymentDescriptor
+			entry := log.Back()
+			for entry != nil {
+				if entry.Value.EntryType == FeeUpdate {
+					currentFee = entry.Value
+					break
+				}
+
+				entry = entry.Prev()
+			}
+			if currentFee == nil {
+				t.Fatal("fee update not retained")
+			}
+			if currentFee.Amount != feeUpdate.Amount {
+				t.Fatalf("latest fee is %v, expected %v",
+					currentFee.Amount, feeUpdate.Amount)
+			}
+
+			if action.commitLocal {
+				currentFee.setCommitHeight(
+					lntypes.Local, uint64(i+1),
+				)
+			}
+			if action.commitRemote {
+				currentFee.setCommitHeight(
+					lntypes.Remote, uint64(i+1),
+				)
+			}
+			if action.commitLocal || action.commitRemote {
+				committed[currentFee] = struct{}{}
+			}
+
+			var uncommitted int
+			resident := make(map[*paymentDescriptor]struct{})
+			var nextLogIndex uint64
+			entry = log.Front()
+			for entry != nil {
+				update := entry.Value
+				resident[update] = struct{}{}
+				if update.LogIndex != nextLogIndex {
+					t.Fatalf(
+						"non-contiguous log index: "+
+							"got %d, want %d",
+						update.LogIndex, nextLogIndex,
+					)
+				}
+				nextLogIndex++
+
+				if update.EntryType == FeeUpdate &&
+					update.addCommitHeights.Local == 0 &&
+					update.addCommitHeights.Remote == 0 {
+
+					uncommitted++
+				}
+
+				entry = entry.Next()
+			}
+			if log.logIndex != nextLogIndex {
+				t.Fatalf("log index is %d, expected %d",
+					log.logIndex, nextLogIndex)
+			}
+
+			if uncommitted > 1 {
+				t.Fatalf("retained %d uncommitted fee updates",
+					uncommitted)
+			}
+			for update := range committed {
+				if _, ok := resident[update]; !ok {
+					t.Fatal("committed fee update removed")
+				}
+			}
+		}
+	})
+}
+
+// TestReceiveUpdateFeeReplacement checks that consecutive fee updates retain
+// the latest value until a commitment chain observes the update.
+func TestReceiveUpdateFeeReplacement(t *testing.T) {
+	t.Parallel()
+
+	_, bobChannel, err := CreateTestChannels(
+		t, channeldb.SingleFunderTweaklessBit,
+	)
+	require.NoError(t, err)
+
+	const numUpdates = 10_000
+	for i := 1; i <= numUpdates; i++ {
+		require.NoError(
+			t, bobChannel.ReceiveUpdateFee(
+				chainfee.SatPerKWeight(i),
+			),
+		)
+	}
+
+	require.Equal(t, uint64(1), bobChannel.updateLogs.Remote.logIndex)
+
+	feeUpdates := make([]*paymentDescriptor, 0, 1)
+	entry := bobChannel.updateLogs.Remote.Front()
+	for entry != nil {
+		if entry.Value.EntryType == FeeUpdate {
+			feeUpdates = append(feeUpdates, entry.Value)
+		}
+
+		entry = entry.Next()
+	}
+
+	require.Len(t, feeUpdates, 1)
+	require.Equal(
+		t, int64(numUpdates), int64(feeUpdates[0].Amount.ToSatoshis()),
+	)
+	require.Zero(t, feeUpdates[0].LogIndex)
+}
+
+// TestAppendFeeUpdatePreservesCommitted checks that a fee update observed by
+// either commitment chain isn't replaced by a later update.
+func TestAppendFeeUpdatePreservesCommitted(t *testing.T) {
+	t.Parallel()
+
+	log := newUpdateLog(0, 0)
+	first := &paymentDescriptor{
+		LogIndex:  log.logIndex,
+		EntryType: FeeUpdate,
+	}
+	log.appendFeeUpdate(first)
+	first.setCommitHeight(lntypes.Remote, 1)
+
+	second := &paymentDescriptor{
+		LogIndex:  log.logIndex,
+		Amount:    2,
+		EntryType: FeeUpdate,
+	}
+	log.appendFeeUpdate(second)
+
+	third := &paymentDescriptor{
+		LogIndex:  log.logIndex,
+		Amount:    3,
+		EntryType: FeeUpdate,
+	}
+	log.appendFeeUpdate(third)
+
+	require.Same(t, first, log.Front().Value)
+	require.Same(t, second, log.Back().Value)
+	require.Equal(t, third.Amount, second.Amount)
+	require.Equal(t, uint64(2), log.logIndex)
+	require.Contains(t, log.updateIndex, first.LogIndex)
+	require.Contains(t, log.updateIndex, second.LogIndex)
+	require.NotContains(t, log.updateIndex, third.LogIndex)
+}