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) { 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) +}