15869455718

Committed 25 Jun 2025 06:53AM UTC coverage: 55.806% (-12.2%) from 67.978%

Build # 15869455718

Build Type

Pull #9148

github

Committed by

web-flow

Commit Message

Merge c64e3a6c3 into 4335d9cfb

Pull Request Pull Request #9148: DynComms [2/n]: lnwire: add authenticated wire messages for Dyn*

Run Details

232 of 270 new or added lines in 5 files covered. (85.93%)

23628 existing lines in 289 files now uncovered.

108377 of 194204 relevant lines covered (55.81%)

22552.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/channel_notifier.go

package lnd

import (
        "context"
        "fmt"

        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/chanbackup"
        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/channelnotifier"
)

// channelNotifier is an implementation of the chanbackup.ChannelNotifier
// interface using the existing channelnotifier.ChannelNotifier struct. This
// implementation allows us to satisfy all the dependencies of the
// chanbackup.SubSwapper struct.
type channelNotifier struct {
        // chanNotifier is the based channel notifier that we'll proxy requests
        // from.
        chanNotifier *channelnotifier.ChannelNotifier

        // addrs is an implementation of the addrSource interface that allows
        // us to get the latest set of addresses for a given node. We'll need
        // this to be able to create an SCB for new channels.
        addrs channeldb.AddrSource
}

// SubscribeChans requests a new channel subscription relative to the initial
// set of known channels. We use the knownChans as a synchronization point to
// ensure that the chanbackup.SubSwapper does not miss any channel open or
// close events in the period between when it's created, and when it requests
// the channel subscription.
//
// NOTE: This is part of the chanbackup.ChannelNotifier interface.
func (c *channelNotifier) SubscribeChans(ctx context.Context,
        startingChans map[wire.OutPoint]struct{}) (
        *chanbackup.ChannelSubscription, error) {

        ltndLog.Infof("Channel backup proxy channel notifier starting")

        // TODO(roasbeef): read existing set of chans and diff

        quit := make(chan struct{})
        chanUpdates := make(chan chanbackup.ChannelEvent, 1)

        // sendChanOpenUpdate is a closure that sends a ChannelEvent to the
        // chanUpdates channel to inform subscribers about new pending or
        // confirmed channels.
        sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) {
                _, nodeAddrs, err := c.addrs.AddrsForNode(
                        ctx, newOrPendingChan.IdentityPub,
                )
                if err != nil {
                        pub := newOrPendingChan.IdentityPub
                        ltndLog.Errorf("unable to fetch addrs for %x: %v",
                                pub.SerializeCompressed(), err)
                }

                chanEvent := chanbackup.ChannelEvent{
                        NewChans: []chanbackup.ChannelWithAddrs{
                                {
                                        OpenChannel: newOrPendingChan,
                                        Addrs:       nodeAddrs,
                                },
                        },
                }

                select {
                case chanUpdates <- chanEvent:
                case <-quit:
                        return
                }
        }

        // In order to adhere to the interface, we'll proxy the events from the
        // channel notifier to the sub-swapper in a format it understands.
        go func() {
                // First, we'll subscribe to the primary channel notifier so we can
                // obtain events for new opened/closed channels.
                chanSubscription, err := c.chanNotifier.SubscribeChannelEvents()
                if err != nil {
                        panic(fmt.Sprintf("unable to subscribe to chans: %v",
                                err))
                }

                defer chanSubscription.Cancel()

                for {
                        select {

                        // A new event has been sent by the chanNotifier, we'll
                        // filter out the events we actually care about and
                        // send them to the sub-swapper.
                        case e := <-chanSubscription.Updates():
                                // TODO(roasbeef): batch dispatch ntnfs

                                switch event := e.(type) {
                                // A new channel has been opened and is still
                                // pending. We can still create a backup, even
                                // if the final channel ID is not yet available.
                                case channelnotifier.PendingOpenChannelEvent:
                                        pendingChan := event.PendingChannel
                                        sendChanOpenUpdate(pendingChan)

                                // A new channel has been confirmed, we'll
                                // obtain the node address, then send to the
                                // sub-swapper.
                                case channelnotifier.OpenChannelEvent:
                                        sendChanOpenUpdate(event.Channel)

                                // An existing channel has been closed, we'll
                                // send only the chanPoint of the closed
                                // channel to the sub-swapper.
                                case channelnotifier.ClosedChannelEvent:
                                        chanPoint := event.CloseSummary.ChanPoint
                                        closeType := event.CloseSummary.CloseType

                                        // Because we see the contract as closed
                                        // once our local force close TX
                                        // confirms, the channel arbitrator
                                        // already fires on this event. But
                                        // because our funds can be in limbo for
                                        // up to 2 weeks worst case we don't
                                        // want to remove the crucial info we
                                        // need for sweeping that time locked
                                        // output before we've actually done so.
                                        if closeType == channeldb.LocalForceClose {
                                                ltndLog.Debugf("Channel %v "+
                                                        "was force closed by "+
                                                        "us, not removing "+
                                                        "from channel backup "+
                                                        "until fully resolved",
                                                        chanPoint)

                                                continue
                                        }

                                        chanEvent := chanbackup.ChannelEvent{
                                                ClosedChans: []wire.OutPoint{
                                                        chanPoint,
                                                },
                                        }

                                        select {
                                        case chanUpdates <- chanEvent:
                                        case <-quit:
                                                return
                                        }

                                // A channel was fully resolved on chain. This
                                // should only really interest us if it was a
                                // locally force closed channel where we didn't
                                // remove the channel already when the close
                                // event was fired.
                                case channelnotifier.FullyResolvedChannelEvent:
                                        chanEvent := chanbackup.ChannelEvent{
                                                ClosedChans: []wire.OutPoint{
                                                        *event.ChannelPoint,
                                                },
                                        }

                                        select {
                                        case chanUpdates <- chanEvent:
                                        case <-quit:
                                                return
                                        }
                                }

                        // The cancel method has been called, signalling us to
                        // exit
                        case <-quit:
                                return
                        }
                }
        }()

        return &chanbackup.ChannelSubscription{
                ChanUpdates: chanUpdates,
                Cancel: func() {
                        close(quit)
                },
        }, nil
}

// A compile-time constraint to ensure channelNotifier implements
// chanbackup.ChannelNotifier.
var _ chanbackup.ChannelNotifier = (*channelNotifier)(nil)

1	package lnd
2
3	import (
4	"context"
5	"fmt"
6
7	"github.com/btcsuite/btcd/wire"
8	"github.com/lightningnetwork/lnd/chanbackup"
9	"github.com/lightningnetwork/lnd/channeldb"
10	"github.com/lightningnetwork/lnd/channelnotifier"
11	)
12
13	// channelNotifier is an implementation of the chanbackup.ChannelNotifier
14	// interface using the existing channelnotifier.ChannelNotifier struct. This
15	// implementation allows us to satisfy all the dependencies of the
16	// chanbackup.SubSwapper struct.
17	type channelNotifier struct {
18	// chanNotifier is the based channel notifier that we'll proxy requests
19	// from.
20	chanNotifier *channelnotifier.ChannelNotifier
21
22	// addrs is an implementation of the addrSource interface that allows
23	// us to get the latest set of addresses for a given node. We'll need
24	// this to be able to create an SCB for new channels.
25	addrs channeldb.AddrSource
26	}
27
28	// SubscribeChans requests a new channel subscription relative to the initial
29	// set of known channels. We use the knownChans as a synchronization point to
30	// ensure that the chanbackup.SubSwapper does not miss any channel open or
31	// close events in the period between when it's created, and when it requests
32	// the channel subscription.
33	//
34	// NOTE: This is part of the chanbackup.ChannelNotifier interface.
35	func (c *channelNotifier) SubscribeChans(ctx context.Context,
36	startingChans map[wire.OutPoint]struct{}) (
UNCOV 37	*chanbackup.ChannelSubscription, error) {	×
UNCOV 38		×
UNCOV 39	ltndLog.Infof("Channel backup proxy channel notifier starting")	×
UNCOV 40		×
UNCOV 41	// TODO(roasbeef): read existing set of chans and diff	×
UNCOV 42		×
UNCOV 43	quit := make(chan struct{})	×
UNCOV 44	chanUpdates := make(chan chanbackup.ChannelEvent, 1)	×
UNCOV 45		×
UNCOV 46	// sendChanOpenUpdate is a closure that sends a ChannelEvent to the	×
UNCOV 47	// chanUpdates channel to inform subscribers about new pending or	×
UNCOV 48	// confirmed channels.	×
UNCOV 49	sendChanOpenUpdate := func(newOrPendingChan *channeldb.OpenChannel) {	×
UNCOV 50	_, nodeAddrs, err := c.addrs.AddrsForNode(	×
UNCOV 51	ctx, newOrPendingChan.IdentityPub,	×
UNCOV 52	)	×
UNCOV 53	if err != nil {	×
54	pub := newOrPendingChan.IdentityPub	×
55	ltndLog.Errorf("unable to fetch addrs for %x: %v",	×
56	pub.SerializeCompressed(), err)	×
57	}	×
58
UNCOV 59	chanEvent := chanbackup.ChannelEvent{	×
UNCOV 60	NewChans: []chanbackup.ChannelWithAddrs{	×
UNCOV 61	{	×
UNCOV 62	OpenChannel: newOrPendingChan,	×
UNCOV 63	Addrs: nodeAddrs,	×
UNCOV 64	},	×
UNCOV 65	},	×
UNCOV 66	}	×
UNCOV 67		×
UNCOV 68	select {	×
UNCOV 69	case chanUpdates <- chanEvent:	×
70	case <-quit:	×
71	return	×
72	}
73	}
74
75	// In order to adhere to the interface, we'll proxy the events from the
76	// channel notifier to the sub-swapper in a format it understands.
UNCOV 77	go func() {	×
UNCOV 78	// First, we'll subscribe to the primary channel notifier so we can	×
UNCOV 79	// obtain events for new opened/closed channels.	×
UNCOV 80	chanSubscription, err := c.chanNotifier.SubscribeChannelEvents()	×
UNCOV 81	if err != nil {	×
82	panic(fmt.Sprintf("unable to subscribe to chans: %v",	×
83	err))	×
84	}
85
UNCOV 86	defer chanSubscription.Cancel()	×
UNCOV 87		×
UNCOV 88	for {	×
UNCOV 89	select {	×
90
91	// A new event has been sent by the chanNotifier, we'll
92	// filter out the events we actually care about and
93	// send them to the sub-swapper.
UNCOV 94	case e := <-chanSubscription.Updates():	×
UNCOV 95	// TODO(roasbeef): batch dispatch ntnfs	×
UNCOV 96		×
UNCOV 97	switch event := e.(type) {	×
98	// A new channel has been opened and is still
99	// pending. We can still create a backup, even
100	// if the final channel ID is not yet available.
UNCOV 101	case channelnotifier.PendingOpenChannelEvent:	×
UNCOV 102	pendingChan := event.PendingChannel	×
UNCOV 103	sendChanOpenUpdate(pendingChan)	×
104
105	// A new channel has been confirmed, we'll
106	// obtain the node address, then send to the
107	// sub-swapper.
UNCOV 108	case channelnotifier.OpenChannelEvent:	×
UNCOV 109	sendChanOpenUpdate(event.Channel)	×
110
111	// An existing channel has been closed, we'll
112	// send only the chanPoint of the closed
113	// channel to the sub-swapper.
UNCOV 114	case channelnotifier.ClosedChannelEvent:	×
UNCOV 115	chanPoint := event.CloseSummary.ChanPoint	×
UNCOV 116	closeType := event.CloseSummary.CloseType	×
UNCOV 117		×
UNCOV 118	// Because we see the contract as closed	×
UNCOV 119	// once our local force close TX	×
UNCOV 120	// confirms, the channel arbitrator	×
UNCOV 121	// already fires on this event. But	×
UNCOV 122	// because our funds can be in limbo for	×
UNCOV 123	// up to 2 weeks worst case we don't	×
UNCOV 124	// want to remove the crucial info we	×
UNCOV 125	// need for sweeping that time locked	×
UNCOV 126	// output before we've actually done so.	×
UNCOV 127	if closeType == channeldb.LocalForceClose {	×
UNCOV 128	ltndLog.Debugf("Channel %v "+	×
UNCOV 129	"was force closed by "+	×
UNCOV 130	"us, not removing "+	×
UNCOV 131	"from channel backup "+	×
UNCOV 132	"until fully resolved",	×
UNCOV 133	chanPoint)	×
UNCOV 134		×
UNCOV 135	continue	×
136	}
137
UNCOV 138	chanEvent := chanbackup.ChannelEvent{	×
UNCOV 139	ClosedChans: []wire.OutPoint{	×
UNCOV 140	chanPoint,	×
UNCOV 141	},	×
UNCOV 142	}	×
UNCOV 143		×
UNCOV 144	select {	×
UNCOV 145	case chanUpdates <- chanEvent:	×
146	case <-quit:	×
147	return	×
148	}
149
150	// A channel was fully resolved on chain. This
151	// should only really interest us if it was a
152	// locally force closed channel where we didn't
153	// remove the channel already when the close
154	// event was fired.
UNCOV 155	case channelnotifier.FullyResolvedChannelEvent:	×
UNCOV 156	chanEvent := chanbackup.ChannelEvent{	×
UNCOV 157	ClosedChans: []wire.OutPoint{	×
UNCOV 158	*event.ChannelPoint,	×
UNCOV 159	},	×
UNCOV 160	}	×
UNCOV 161		×
UNCOV 162	select {	×
UNCOV 163	case chanUpdates <- chanEvent:	×
164	case <-quit:	×
165	return	×
166	}
167	}
168
169	// The cancel method has been called, signalling us to
170	// exit
UNCOV 171	case <-quit:	×
UNCOV 172	return	×
173	}
174	}
175	}()
176
UNCOV 177	return &chanbackup.ChannelSubscription{	×
UNCOV 178	ChanUpdates: chanUpdates,	×
UNCOV 179	Cancel: func() {	×
UNCOV 180	close(quit)	×
UNCOV 181	},	×
182	}, nil
183	}
184
185	// A compile-time constraint to ensure channelNotifier implements
186	// chanbackup.ChannelNotifier.
187	var _ chanbackup.ChannelNotifier = (*channelNotifier)(nil)

lightningnetwork / lnd / 15869455718

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous