12312390362

Committed 13 Dec 2024 08:44AM UTC coverage: 57.458% (+8.5%) from 48.92%

Build # 12312390362

Build Type

Pull #9343

github

Committed by

ellemouton

Commit Message

fn: rework the ContextGuard and add tests

In this commit, the ContextGuard struct is re-worked such that the
context that its new main WithCtx method provides is cancelled in sync
with a parent context being cancelled or with it's quit channel being
cancelled. Tests are added to assert the behaviour. In order for the
close of the quit channel to be consistent with the cancelling of the
derived context, the quit channel _must_ be contained internal to the
ContextGuard so that callers are only able to close the channel via the
exposed Quit method which will then take care to first cancel any
derived context that depend on the quit channel before returning.

Pull Request Pull Request #9343: fn: expand the ContextGuard and add tests

Run Details

101853 of 177264 relevant lines covered (57.46%)

24972.93 hits per line

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

30.49

/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
}

// 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)
        }
}

// 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	// New creates a new channel notifier. The ChannelNotifier gets channel
84	// events from peers and from the chain arbitrator, and dispatches them to
85	// its clients.
86	func New(chanDB channeldb.ChannelStateDB) ChannelNotifier {	19✔
87	return &ChannelNotifier{	19✔
88	ntfnServer: subscribe.NewServer(),	19✔
89	chanDB: chanDB,	19✔
90	}	19✔
91	}	19✔
92
93	// Start starts the ChannelNotifier and all goroutines it needs to carry out its task.
94	func (c *ChannelNotifier) Start() error {	19✔
95	var err error	19✔
96	c.started.Do(func() {	38✔
97	log.Info("ChannelNotifier starting")	19✔
98	err = c.ntfnServer.Start()	19✔
99	})	19✔
100	return err	19✔
101	}
102
103	// Stop signals the notifier for a graceful shutdown.
104	func (c *ChannelNotifier) Stop() error {	19✔
105	var err error	19✔
106	c.stopped.Do(func() {	38✔
107	log.Info("ChannelNotifier shutting down...")	19✔
108	defer log.Debug("ChannelNotifier shutdown complete")	19✔
109		19✔
110	err = c.ntfnServer.Stop()	19✔
111	})	19✔
112	return err	19✔
113	}
114
115	// SubscribeChannelEvents returns a subscribe.Client that will receive updates
116	// any time the Server is made aware of a new event. The subscription provides
117	// channel events from the point of subscription onwards.
118	//
119	// TODO(carlaKC): update to allow subscriptions to specify a block height from
120	// which we would like to subscribe to events.
121	func (c ChannelNotifier) SubscribeChannelEvents() (subscribe.Client, error) {	3✔
122	return c.ntfnServer.Subscribe()	3✔
123	}	3✔
124
125	// NotifyPendingOpenChannelEvent notifies the channelEventNotifier goroutine
126	// that a new channel is pending. The pending channel is passed as a parameter
127	// instead of read from the database because it might not yet have been
128	// persisted to the DB because we still wait for the final message from the
129	// remote peer.
130	func (c *ChannelNotifier) NotifyPendingOpenChannelEvent(chanPoint wire.OutPoint,
131	pendingChan *channeldb.OpenChannel) {	×
132		×
133	event := PendingOpenChannelEvent{	×
134	ChannelPoint: &chanPoint,	×
135	PendingChannel: pendingChan,	×
136	}	×
137		×
138	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
139	log.Warnf("Unable to send pending open channel update: %v", err)	×
140	}	×
141	}
142
143	// NotifyOpenChannelEvent notifies the channelEventNotifier goroutine that a
144	// channel has gone from pending open to open.
145	func (c *ChannelNotifier) NotifyOpenChannelEvent(chanPoint wire.OutPoint) {	×
146	// Fetch the relevant channel from the database.	×
147	channel, err := c.chanDB.FetchChannel(chanPoint)	×
148	if err != nil {	×
149	log.Warnf("Unable to fetch open channel from the db: %v", err)	×
150	}	×
151
152	// Send the open event to all channel event subscribers.
153	event := OpenChannelEvent{Channel: channel}	×
154	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
155	log.Warnf("Unable to send open channel update: %v", err)	×
156	}	×
157	}
158
159	// NotifyClosedChannelEvent notifies the channelEventNotifier goroutine that a
160	// channel has closed.
161	func (c *ChannelNotifier) NotifyClosedChannelEvent(chanPoint wire.OutPoint) {	×
162	// Fetch the relevant closed channel from the database.	×
163	closeSummary, err := c.chanDB.FetchClosedChannel(&chanPoint)	×
164	if err != nil {	×
165	log.Warnf("Unable to fetch closed channel summary from the db: %v", err)	×
166	}	×
167
168	// Send the closed event to all channel event subscribers.
169	event := ClosedChannelEvent{CloseSummary: closeSummary}	×
170	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
171	log.Warnf("Unable to send closed channel update: %v", err)	×
172	}	×
173	}
174
175	// NotifyFullyResolvedChannelEvent notifies the channelEventNotifier goroutine
176	// that a channel was fully resolved on chain.
177	func (c *ChannelNotifier) NotifyFullyResolvedChannelEvent(
178	chanPoint wire.OutPoint) {	×
179		×
180	// Send the resolved event to all channel event subscribers.	×
181	event := FullyResolvedChannelEvent{ChannelPoint: &chanPoint}	×
182	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
183	log.Warnf("Unable to send resolved channel update: %v", err)	×
184	}	×
185	}
186
187	// NotifyActiveLinkEvent notifies the channelEventNotifier goroutine that a
188	// link has been added to the switch.
189	func (c *ChannelNotifier) NotifyActiveLinkEvent(chanPoint wire.OutPoint) {	×
190	event := ActiveLinkEvent{ChannelPoint: &chanPoint}	×
191	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
192	log.Warnf("Unable to send active link update: %v", err)	×
193	}	×
194	}
195
196	// NotifyActiveChannelEvent notifies the channelEventNotifier goroutine that a
197	// channel is active.
198	func (c *ChannelNotifier) NotifyActiveChannelEvent(chanPoint wire.OutPoint) {	×
199	event := ActiveChannelEvent{ChannelPoint: &chanPoint}	×
200	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
201	log.Warnf("Unable to send active channel update: %v", err)	×
202	}	×
203	}
204
205	// NotifyInactiveLinkEvent notifies the channelEventNotifier goroutine that a
206	// link has been removed from the switch.
207	func (c *ChannelNotifier) NotifyInactiveLinkEvent(chanPoint wire.OutPoint) {	×
208	event := InactiveLinkEvent{ChannelPoint: &chanPoint}	×
209	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
210	log.Warnf("Unable to send inactive link update: %v", err)	×
211	}	×
212	}
213
214	// NotifyInactiveChannelEvent notifies the channelEventNotifier goroutine that a
215	// channel is inactive.
216	func (c *ChannelNotifier) NotifyInactiveChannelEvent(chanPoint wire.OutPoint) {	×
217	event := InactiveChannelEvent{ChannelPoint: &chanPoint}	×
218	if err := c.ntfnServer.SendUpdate(event); err != nil {	×
219	log.Warnf("Unable to send inactive channel update: %v", err)	×
220	}	×
221	}

lightningnetwork / lnd / 12312390362

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