Merge pull request #11023 from ellemouton/update-fee-log
What changed, and why it matters
This commit hardens how LND handles Lightning channel fee updates and mailbox message admission. It enforces the protocol rule that only the channel opener can send fee updates, prevents uncommitted fee updates from piling up in the update log, limits how many queued messages a peer can stuff into a channel's mailbox, and makes the link disconnect if the mailbox is full so messages are not silently dropped. These changes reduce ways a misbehaving or buggy peer could disrupt a channel or waste resources.
Review and merge if tests pass. Monitor for any peer-disconnect regressions caused by the new mailbox limits, and consider whether maxWireMessages/maxWireBytes need tuning for high-traffic nodes.
Security signals we found
BOLT 2 role validation for UpdateFee sender
Replacement of uncommitted fee updates to prevent log bloat and index gaps
Bounded mailbox wire-message queue (count and bytes)
Disconnect peer on mailbox admission failure to preserve ordered message stream
Log deduplication for non-fatal/unknown messages
Evidence from the diff
The patch makes four related changes in htlcswitch and lnwallet: (1) processRemoteUpdateFee now rejects UpdateFee messages when the local node is the channel initiator, per BOLT 2; (2) update_log.go adds appendFeeUpdate, which replaces an existing uncommitted fee descriptor’s Amount instead of appending a new one, preserving contiguous log indices and avoiding redundant fee entries; (3) memoryMailBox adds count (maxWireMessages=1000) and byte (maxWireBytes=4 MiB) admission budgets for wire messages, returning errWireMessageQueueFull when exceeded; (4) HandleChannelUpdate now serializes admission with a mutex, latches on the first AddMessage failure, and disconnects the peer, preventing out-of-order or dropped messages. Tests cover role validation, fee exposure errors, log deduplication, mailbox budgets, and disconnect-on-full behavior.
Changed components
htlcswitch/link.gohtlcswitch/mailbox.golnwallet/channel.golnwallet/update_log.goInspect captured patch +750 / −13
### 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) {
### 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)
+}Why this scored 59/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.