From 6711f758fd81296f443f37b6dcfd89a0d0a6eca2 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 4 Aug 2026 12:40:13 -0700 Subject: [PATCH 1/2] lnwallet: coalesce uncommitted fee updates In this commit, we bound the resident fee update log without imposing a fixed update limit. We replace the value of the newest fee update while it remains uncommitted on both commitment chains. Once either chain observes the update, we retain it so retransmission and intermediate commitment states keep their existing semantics. We also cover generated interleavings, sustained remote update streams, committed-state preservation, and the legacy fee-update restore path. --- lnwallet/channel.go | 4 +- lnwallet/channel_test.go | 5 +- lnwallet/update_log.go | 24 ++++ lnwallet/update_log_test.go | 214 ++++++++++++++++++++++++++++++++++++ 4 files changed, 243 insertions(+), 4 deletions(-) create mode 100644 lnwallet/update_log_test.go diff --git a/lnwallet/channel.go b/lnwallet/channel.go index ea243dd86..436bff1e4 100644 --- a/lnwallet/channel.go +++ b/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 } diff --git a/lnwallet/channel_test.go b/lnwallet/channel_test.go index 5fd8cc08e..6a6e12079 100644 --- a/lnwallet/channel_test.go +++ b/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 diff --git a/lnwallet/update_log.go b/lnwallet/update_log.go index b2b8af58d..4e4f736a7 100644 --- a/lnwallet/update_log.go +++ b/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 { diff --git a/lnwallet/update_log_test.go b/lnwallet/update_log_test.go new file mode 100644 index 000000000..996590b4b --- /dev/null +++ b/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) +} From ef24f2c5c3e7371a0dd241d625447e9158b4fb6d Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 4 Aug 2026 12:40:35 -0700 Subject: [PATCH 2/2] htlcswitch: bound peer-controlled channel ingress In this commit, we bound the channel mailbox by message count and by the encoded size of non-commitment control messages. Commitment updates retain their full custom-record allowance and remain protected by the count bound. If either budget fills, we disconnect the peer instead of silently dropping an ordered channel message. We also reject unauthorized fee updates before fee-exposure evaluation, return the exposure error used to fail the link, and emit peer-controlled warning classes only once per link lifetime. --- htlcswitch/link.go | 60 +++++- htlcswitch/link_fee_update_test.go | 306 +++++++++++++++++++++++++++++ htlcswitch/mailbox.go | 60 +++++- htlcswitch/mailbox_test.go | 92 +++++++++ 4 files changed, 508 insertions(+), 10 deletions(-) create mode 100644 htlcswitch/link_fee_update_test.go diff --git a/htlcswitch/link.go b/htlcswitch/link.go index 1439171c8..1fbb4c2f8 100644 --- a/htlcswitch/link.go +++ b/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 } diff --git a/htlcswitch/link_fee_update_test.go b/htlcswitch/link_fee_update_test.go new file mode 100644 index 000000000..d8daafa8e --- /dev/null +++ b/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") + } +} diff --git a/htlcswitch/mailbox.go b/htlcswitch/mailbox.go index 2a0796855..e52daca56 100644 --- a/htlcswitch/mailbox.go +++ b/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 diff --git a/htlcswitch/mailbox_test.go b/htlcswitch/mailbox_test.go index df746c186..254a16469 100644 --- a/htlcswitch/mailbox_test.go +++ b/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) {