13593508312

Committed 28 Feb 2025 05:41PM UTC coverage: 58.287% (-10.4%) from 68.65%

Build # 13593508312

Build Type

Pull #9458

github

Committed by

web-flow

Commit Message

Merge d40067c0c into f1182e433

Pull Request Pull Request #9458: multi+server.go: add initial permissions for some peers

Run Details

346 of 548 new or added lines in 10 files covered. (63.14%)

27412 existing lines in 442 files now uncovered.

94709 of 162488 relevant lines covered (58.29%)

1.81 hits per line

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

70.45

/channelnotifier/channelnotifier.go

package channelnotifier

import (
        "sync"

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

// ChannelNotifier is a subsystem which all active, inactive, and closed channel
// events pipe through. It takes subscriptions for its events, and whenever
// it receives a new event it notifies its subscribers over the proper channel.
type ChannelNotifier struct {
        started sync.Once
        stopped sync.Once

        ntfnServer *subscribe.Server

        chanDB *channeldb.ChannelStateDB
}

// PendingOpenChannelEvent represents a new event where a new channel has
// entered a pending open state.
type PendingOpenChannelEvent struct {
        // ChannelPoint is the channel outpoint for the new channel.
        ChannelPoint *wire.OutPoint

        // PendingChannel is the channel configuration for the newly created
        // channel. This might not have been persisted to the channel DB yet
        // because we are still waiting for the final message from the remote
        // peer.
        PendingChannel *channeldb.OpenChannel
}

// OpenChannelEvent represents a new event where a channel goes from pending
// open to open.
type OpenChannelEvent struct {
        // Channel is the channel that has become open.
        Channel *channeldb.OpenChannel
}

// ActiveLinkEvent represents a new event where the link becomes active in the
// switch. This happens before the ActiveChannelEvent.
type ActiveLinkEvent struct {
        // ChannelPoint is the channel point for the newly active channel.
        ChannelPoint *wire.OutPoint
}

// InactiveLinkEvent represents a new event where the link becomes inactive in
// the switch.
type InactiveLinkEvent struct {
        // ChannelPoint is the channel point for the inactive channel.
        ChannelPoint *wire.OutPoint
}

// ActiveChannelEvent represents a new event where a channel becomes active.
type ActiveChannelEvent struct {
        // ChannelPoint is the channelpoint for the newly active channel.
        ChannelPoint *wire.OutPoint
}

// InactiveChannelEvent represents a new event where a channel becomes inactive.
type InactiveChannelEvent struct {
        // ChannelPoint is the channelpoint for the newly inactive channel.
        ChannelPoint *wire.OutPoint
}

// ClosedChannelEvent represents a new event where a channel becomes closed.
type ClosedChannelEvent struct {
        // CloseSummary is the summary of the channel close that has occurred.
        CloseSummary *channeldb.ChannelCloseSummary
}

// FullyResolvedChannelEvent represents a new event where a channel becomes
// fully resolved.
type FullyResolvedChannelEvent struct {
        // ChannelPoint is the channelpoint for the newly fully resolved
        // channel.
        ChannelPoint *wire.OutPoint
}

// FundingTimeoutEvent represents a new event where a pending-open channel has
// timed out from the PoV of the funding manager because the funding tx
// has not confirmed in the allotted time.
type FundingTimeoutEvent struct {
        // ChannelPoint is the channelpoint for the newly inactive channel.
        ChannelPoint *wire.OutPoint
}

// New creates a new channel notifier. The ChannelNotifier gets channel
// events from peers and from the chain arbitrator, and dispatches them to
// its clients.
func New(chanDB *channeldb.ChannelStateDB) *ChannelNotifier {
        return &ChannelNotifier{
                ntfnServer: subscribe.NewServer(),
                chanDB:     chanDB,
        }
}

// Start starts the ChannelNotifier and all goroutines it needs to carry out its task.
func (c *ChannelNotifier) Start() error {
        var err error
        c.started.Do(func() {
                log.Info("ChannelNotifier starting")
                err = c.ntfnServer.Start()
        })
        return err
}

// Stop signals the notifier for a graceful shutdown.
func (c *ChannelNotifier) Stop() error {
        var err error
        c.stopped.Do(func() {
                log.Info("ChannelNotifier shutting down...")
                defer log.Debug("ChannelNotifier shutdown complete")

                err = c.ntfnServer.Stop()
        })
        return err
}

// SubscribeChannelEvents returns a subscribe.Client that will receive updates
// any time the Server is made aware of a new event. The subscription provides
// channel events from the point of subscription onwards.
//
// TODO(carlaKC): update  to allow subscriptions to specify a block height from
// which we would like to subscribe to events.
func (c *ChannelNotifier) SubscribeChannelEvents() (*subscribe.Client, error) {
        return c.ntfnServer.Subscribe()
}

// NotifyPendingOpenChannelEvent notifies the channelEventNotifier goroutine
// that a new channel is pending. The pending channel is passed as a parameter
// instead of read from the database because it might not yet have been
// persisted to the DB because we still wait for the final message from the
// remote peer.
func (c *ChannelNotifier) NotifyPendingOpenChannelEvent(chanPoint wire.OutPoint,
        pendingChan *channeldb.OpenChannel) {

        event := PendingOpenChannelEvent{
                ChannelPoint:   &chanPoint,
                PendingChannel: pendingChan,
        }

        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send pending open channel update: %v", err)
        }
}

// NotifyOpenChannelEvent notifies the channelEventNotifier goroutine that a
// channel has gone from pending open to open.
func (c *ChannelNotifier) NotifyOpenChannelEvent(chanPoint wire.OutPoint) {
        // Fetch the relevant channel from the database.
        channel, err := c.chanDB.FetchChannel(chanPoint)
        if err != nil {
                log.Warnf("Unable to fetch open channel from the db: %v", err)
        }

        // Send the open event to all channel event subscribers.
        event := OpenChannelEvent{Channel: channel}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send open channel update: %v", err)
        }
}

// NotifyClosedChannelEvent notifies the channelEventNotifier goroutine that a
// channel has closed.
func (c *ChannelNotifier) NotifyClosedChannelEvent(chanPoint wire.OutPoint) {
        // Fetch the relevant closed channel from the database.
        closeSummary, err := c.chanDB.FetchClosedChannel(&chanPoint)
        if err != nil {
                log.Warnf("Unable to fetch closed channel summary from the db: %v", err)
        }

        // Send the closed event to all channel event subscribers.
        event := ClosedChannelEvent{CloseSummary: closeSummary}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send closed channel update: %v", err)
        }
}

// NotifyFullyResolvedChannelEvent notifies the channelEventNotifier goroutine
// that a channel was fully resolved on chain.
func (c *ChannelNotifier) NotifyFullyResolvedChannelEvent(
        chanPoint wire.OutPoint) {

        // Send the resolved event to all channel event subscribers.
        event := FullyResolvedChannelEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send resolved channel update: %v", err)
        }
}

// NotifyFundingTimeoutEvent notifies the channelEventNotifier goroutine that
// a funding timeout has occurred for a certain channel point.
func (c *ChannelNotifier) NotifyFundingTimeout(chanPoint wire.OutPoint) {
        // Send this event to all channel event subscribers.
        event := FundingTimeoutEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send funding timeout update: %v", err)
        }
}

// NotifyActiveLinkEvent notifies the channelEventNotifier goroutine that a
// link has been added to the switch.
func (c *ChannelNotifier) NotifyActiveLinkEvent(chanPoint wire.OutPoint) {
        event := ActiveLinkEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send active link update: %v", err)
        }
}

// NotifyActiveChannelEvent notifies the channelEventNotifier goroutine that a
// channel is active.
func (c *ChannelNotifier) NotifyActiveChannelEvent(chanPoint wire.OutPoint) {
        event := ActiveChannelEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send active channel update: %v", err)
        }
}

// NotifyInactiveLinkEvent notifies the channelEventNotifier goroutine that a
// link has been removed from the switch.
func (c *ChannelNotifier) NotifyInactiveLinkEvent(chanPoint wire.OutPoint) {
        event := InactiveLinkEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send inactive link update: %v", err)
        }
}

// NotifyInactiveChannelEvent notifies the channelEventNotifier goroutine that a
// channel is inactive.
func (c *ChannelNotifier) NotifyInactiveChannelEvent(chanPoint wire.OutPoint) {
        event := InactiveChannelEvent{ChannelPoint: &chanPoint}
        if err := c.ntfnServer.SendUpdate(event); err != nil {
                log.Warnf("Unable to send inactive channel update: %v", err)
        }
}

1	package channelnotifier
2
3	import (
4	"sync"
5
6	"github.com/btcsuite/btcd/wire"
7	"github.com/lightningnetwork/lnd/channeldb"
8	"github.com/lightningnetwork/lnd/subscribe"
9	)
10
11	// ChannelNotifier is a subsystem which all active, inactive, and closed channel
12	// events pipe through. It takes subscriptions for its events, and whenever
13	// it receives a new event it notifies its subscribers over the proper channel.
14	type ChannelNotifier struct {
15	started sync.Once
16	stopped sync.Once
17
18	ntfnServer *subscribe.Server
19
20	chanDB *channeldb.ChannelStateDB
21	}
22
23	// PendingOpenChannelEvent represents a new event where a new channel has
24	// entered a pending open state.
25	type PendingOpenChannelEvent struct {
26	// ChannelPoint is the channel outpoint for the new channel.
27	ChannelPoint *wire.OutPoint
28
29	// PendingChannel is the channel configuration for the newly created
30	// channel. This might not have been persisted to the channel DB yet
31	// because we are still waiting for the final message from the remote
32	// peer.
33	PendingChannel *channeldb.OpenChannel
34	}
35
36	// OpenChannelEvent represents a new event where a channel goes from pending
37	// open to open.
38	type OpenChannelEvent struct {
39	// Channel is the channel that has become open.
40	Channel *channeldb.OpenChannel
41	}
42
43	// ActiveLinkEvent represents a new event where the link becomes active in the
44	// switch. This happens before the ActiveChannelEvent.
45	type ActiveLinkEvent struct {
46	// ChannelPoint is the channel point for the newly active channel.
47	ChannelPoint *wire.OutPoint
48	}
49
50	// InactiveLinkEvent represents a new event where the link becomes inactive in
51	// the switch.
52	type InactiveLinkEvent struct {
53	// ChannelPoint is the channel point for the inactive channel.
54	ChannelPoint *wire.OutPoint
55	}
56
57	// ActiveChannelEvent represents a new event where a channel becomes active.
58	type ActiveChannelEvent struct {
59	// ChannelPoint is the channelpoint for the newly active channel.
60	ChannelPoint *wire.OutPoint
61	}
62
63	// InactiveChannelEvent represents a new event where a channel becomes inactive.
64	type InactiveChannelEvent struct {
65	// ChannelPoint is the channelpoint for the newly inactive channel.
66	ChannelPoint *wire.OutPoint
67	}
68
69	// ClosedChannelEvent represents a new event where a channel becomes closed.
70	type ClosedChannelEvent struct {
71	// CloseSummary is the summary of the channel close that has occurred.
72	CloseSummary *channeldb.ChannelCloseSummary
73	}
74
75	// FullyResolvedChannelEvent represents a new event where a channel becomes
76	// fully resolved.
77	type FullyResolvedChannelEvent struct {
78	// ChannelPoint is the channelpoint for the newly fully resolved
79	// channel.
80	ChannelPoint *wire.OutPoint
81	}
82
83	// FundingTimeoutEvent represents a new event where a pending-open channel has
84	// timed out from the PoV of the funding manager because the funding tx
85	// has not confirmed in the allotted time.
86	type FundingTimeoutEvent struct {
87	// ChannelPoint is the channelpoint for the newly inactive channel.
88	ChannelPoint *wire.OutPoint
89	}
90
91	// New creates a new channel notifier. The ChannelNotifier gets channel
92	// events from peers and from the chain arbitrator, and dispatches them to
93	// its clients.
94	func New(chanDB channeldb.ChannelStateDB) ChannelNotifier {	3✔
95	return &ChannelNotifier{	3✔
96	ntfnServer: subscribe.NewServer(),	3✔
97	chanDB: chanDB,	3✔
98	}	3✔
99	}	3✔
100
101	// Start starts the ChannelNotifier and all goroutines it needs to carry out its task.
102	func (c *ChannelNotifier) Start() error {	3✔
103	var err error	3✔
104	c.started.Do(func() {	6✔
105	log.Info("ChannelNotifier starting")	3✔
106	err = c.ntfnServer.Start()	3✔
107	})	3✔
108	return err	3✔
109	}
110
111	// Stop signals the notifier for a graceful shutdown.
112	func (c *ChannelNotifier) Stop() error {	3✔
113	var err error	3✔
114	c.stopped.Do(func() {	6✔
115	log.Info("ChannelNotifier shutting down...")	3✔
116	defer log.Debug("ChannelNotifier shutdown complete")	3✔
117		3✔
118	err = c.ntfnServer.Stop()	3✔
119	})	3✔
120	return err	3✔
121	}
122
123	// SubscribeChannelEvents returns a subscribe.Client that will receive updates
124	// any time the Server is made aware of a new event. The subscription provides
125	// channel events from the point of subscription onwards.
126	//
127	// TODO(carlaKC): update to allow subscriptions to specify a block height from
128	// which we would like to subscribe to events.
129	func (c ChannelNotifier) SubscribeChannelEvents() (subscribe.Client, error) {	3✔
130	return c.ntfnServer.Subscribe()	3✔
131	}	3✔
132
133	// NotifyPendingOpenChannelEvent notifies the channelEventNotifier goroutine
134	// that a new channel is pending. The pending channel is passed as a parameter
135	// instead of read from the database because it might not yet have been
136	// persisted to the DB because we still wait for the final message from the
137	// remote peer.
138	func (c *ChannelNotifier) NotifyPendingOpenChannelEvent(chanPoint wire.OutPoint,
139	pendingChan *channeldb.OpenChannel) {	3✔
140		3✔
141	event := PendingOpenChannelEvent{	3✔
142	ChannelPoint: &chanPoint,	3✔
143	PendingChannel: pendingChan,	3✔
144	}	3✔
145		3✔
146	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
147	log.Warnf("Unable to send pending open channel update: %v", err)	×
148	}	×
149	}
150
151	// NotifyOpenChannelEvent notifies the channelEventNotifier goroutine that a
152	// channel has gone from pending open to open.
153	func (c *ChannelNotifier) NotifyOpenChannelEvent(chanPoint wire.OutPoint) {	3✔
154	// Fetch the relevant channel from the database.	3✔
155	channel, err := c.chanDB.FetchChannel(chanPoint)	3✔
156	if err != nil {	3✔
157	log.Warnf("Unable to fetch open channel from the db: %v", err)	×
158	}	×
159
160	// Send the open event to all channel event subscribers.
161	event := OpenChannelEvent{Channel: channel}	3✔
162	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
163	log.Warnf("Unable to send open channel update: %v", err)	×
164	}	×
165	}
166
167	// NotifyClosedChannelEvent notifies the channelEventNotifier goroutine that a
168	// channel has closed.
169	func (c *ChannelNotifier) NotifyClosedChannelEvent(chanPoint wire.OutPoint) {	3✔
170	// Fetch the relevant closed channel from the database.	3✔
171	closeSummary, err := c.chanDB.FetchClosedChannel(&chanPoint)	3✔
172	if err != nil {	3✔
173	log.Warnf("Unable to fetch closed channel summary from the db: %v", err)	×
174	}	×
175
176	// Send the closed event to all channel event subscribers.
177	event := ClosedChannelEvent{CloseSummary: closeSummary}	3✔
178	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
179	log.Warnf("Unable to send closed channel update: %v", err)	×
180	}	×
181	}
182
183	// NotifyFullyResolvedChannelEvent notifies the channelEventNotifier goroutine
184	// that a channel was fully resolved on chain.
185	func (c *ChannelNotifier) NotifyFullyResolvedChannelEvent(
186	chanPoint wire.OutPoint) {	3✔
187		3✔
188	// Send the resolved event to all channel event subscribers.	3✔
189	event := FullyResolvedChannelEvent{ChannelPoint: &chanPoint}	3✔
190	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
191	log.Warnf("Unable to send resolved channel update: %v", err)	×
192	}	×
193	}
194
195	// NotifyFundingTimeoutEvent notifies the channelEventNotifier goroutine that
196	// a funding timeout has occurred for a certain channel point.
NEW 197	func (c *ChannelNotifier) NotifyFundingTimeout(chanPoint wire.OutPoint) {	×
NEW 198	// Send this event to all channel event subscribers.	×
NEW 199	event := FundingTimeoutEvent{ChannelPoint: &chanPoint}	×
NEW 200	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
NEW 201	log.Warnf("Unable to send funding timeout update: %v", err)	×
NEW 202	}	×
203	}
204
205	// NotifyActiveLinkEvent notifies the channelEventNotifier goroutine that a
206	// link has been added to the switch.
207	func (c *ChannelNotifier) NotifyActiveLinkEvent(chanPoint wire.OutPoint) {	3✔
208	event := ActiveLinkEvent{ChannelPoint: &chanPoint}	3✔
209	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
210	log.Warnf("Unable to send active link update: %v", err)	×
211	}	×
212	}
213
214	// NotifyActiveChannelEvent notifies the channelEventNotifier goroutine that a
215	// channel is active.
216	func (c *ChannelNotifier) NotifyActiveChannelEvent(chanPoint wire.OutPoint) {	3✔
217	event := ActiveChannelEvent{ChannelPoint: &chanPoint}	3✔
218	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
219	log.Warnf("Unable to send active channel update: %v", err)	×
220	}	×
221	}
222
223	// NotifyInactiveLinkEvent notifies the channelEventNotifier goroutine that a
224	// link has been removed from the switch.
225	func (c *ChannelNotifier) NotifyInactiveLinkEvent(chanPoint wire.OutPoint) {	3✔
226	event := InactiveLinkEvent{ChannelPoint: &chanPoint}	3✔
227	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
228	log.Warnf("Unable to send inactive link update: %v", err)	×
229	}	×
230	}
231
232	// NotifyInactiveChannelEvent notifies the channelEventNotifier goroutine that a
233	// channel is inactive.
234	func (c *ChannelNotifier) NotifyInactiveChannelEvent(chanPoint wire.OutPoint) {	3✔
235	event := InactiveChannelEvent{ChannelPoint: &chanPoint}	3✔
236	if err := c.ntfnServer.SendUpdate(event); err != nil {	3✔
237	log.Warnf("Unable to send inactive channel update: %v", err)	×
238	}	×
239	}

lightningnetwork / lnd / 13593508312

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