• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

lightningnetwork / lnd / 13536249039

26 Feb 2025 03:42AM UTC coverage: 57.462% (-1.4%) from 58.835%
13536249039

Pull #8453

github

Roasbeef
peer: update chooseDeliveryScript to gen script if needed

In this commit, we update `chooseDeliveryScript` to generate a new
script if needed. This allows us to fold in a few other lines that
always followed this function into this expanded function.

The tests have been updated accordingly.
Pull Request #8453: [4/4] - multi: integrate new rbf coop close FSM into the existing peer flow

275 of 1318 new or added lines in 22 files covered. (20.86%)

19521 existing lines in 257 files now uncovered.

103858 of 180741 relevant lines covered (57.46%)

24750.23 hits per line

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

82.09
/chainio/dispatcher.go
1
package chainio
2

3
import (
4
        "errors"
5
        "fmt"
6
        "sync"
7
        "sync/atomic"
8
        "time"
9

10
        "github.com/btcsuite/btclog/v2"
11
        "github.com/lightningnetwork/lnd/chainntnfs"
12
        "github.com/lightningnetwork/lnd/lnutils"
13
        "golang.org/x/sync/errgroup"
14
)
15

16
// DefaultProcessBlockTimeout is the timeout value used when waiting for one
17
// consumer to finish processing the new block epoch.
18
var DefaultProcessBlockTimeout = 60 * time.Second
19

20
// ErrProcessBlockTimeout is the error returned when a consumer takes too long
21
// to process the block.
22
var ErrProcessBlockTimeout = errors.New("process block timeout")
23

24
// BlockbeatDispatcher is a service that handles dispatching new blocks to
25
// `lnd`'s subsystems. During startup, subsystems that are block-driven should
26
// implement the `Consumer` interface and register themselves via
27
// `RegisterQueue`. When two subsystems are independent of each other, they
28
// should be registered in different queues so blocks are notified concurrently.
29
// Otherwise, when living in the same queue, the subsystems are notified of the
30
// new blocks sequentially, which means it's critical to understand the
31
// relationship of these systems to properly handle the order.
32
type BlockbeatDispatcher struct {
33
        wg sync.WaitGroup
34

35
        // notifier is used to receive new block epochs.
36
        notifier chainntnfs.ChainNotifier
37

38
        // beat is the latest blockbeat received.
39
        beat Blockbeat
40

41
        // consumerQueues is a map of consumers that will receive blocks. Its
42
        // key is a unique counter and its value is a queue of consumers. Each
43
        // queue is notified concurrently, and consumers in the same queue is
44
        // notified sequentially.
45
        consumerQueues map[uint32][]Consumer
46

47
        // counter is used to assign a unique id to each queue.
48
        counter atomic.Uint32
49

50
        // quit is used to signal the BlockbeatDispatcher to stop.
51
        quit chan struct{}
52

53
        // queryHeightChan is used to receive queries on the current height of
54
        // the dispatcher.
55
        queryHeightChan chan *query
56
}
57

58
// query is used to fetch the internal state of the dispatcher.
59
type query struct {
60
        // respChan is used to send back the current height back to the caller.
61
        //
62
        // NOTE: This channel must be buffered.
63
        respChan chan int32
64
}
65

66
// newQuery creates a query to be used to fetch the internal state of the
67
// dispatcher.
68
func newQuery() *query {
2✔
69
        return &query{
2✔
70
                respChan: make(chan int32, 1),
2✔
71
        }
2✔
72
}
2✔
73

74
// NewBlockbeatDispatcher returns a new blockbeat dispatcher instance.
75
func NewBlockbeatDispatcher(n chainntnfs.ChainNotifier) *BlockbeatDispatcher {
6✔
76
        return &BlockbeatDispatcher{
6✔
77
                notifier:        n,
6✔
78
                quit:            make(chan struct{}),
6✔
79
                consumerQueues:  make(map[uint32][]Consumer),
6✔
80
                queryHeightChan: make(chan *query, 1),
6✔
81
        }
6✔
82
}
6✔
83

84
// RegisterQueue takes a list of consumers and registers them in the same
85
// queue.
86
//
87
// NOTE: these consumers are notified sequentially.
88
func (b *BlockbeatDispatcher) RegisterQueue(consumers []Consumer) {
7✔
89
        qid := b.counter.Add(1)
7✔
90

7✔
91
        b.consumerQueues[qid] = append(b.consumerQueues[qid], consumers...)
7✔
92
        clog.Infof("Registered queue=%d with %d blockbeat consumers", qid,
7✔
93
                len(consumers))
7✔
94

7✔
95
        for _, c := range consumers {
15✔
96
                clog.Debugf("Consumer [%s] registered in queue %d", c.Name(),
8✔
97
                        qid)
8✔
98
        }
8✔
99
}
100

101
// Start starts the blockbeat dispatcher - it registers a block notification
102
// and monitors and dispatches new blocks in a goroutine. It will refuse to
103
// start if there are no registered consumers.
104
func (b *BlockbeatDispatcher) Start() error {
4✔
105
        // Make sure consumers are registered.
4✔
106
        if len(b.consumerQueues) == 0 {
5✔
107
                return fmt.Errorf("no consumers registered")
1✔
108
        }
1✔
109

110
        // Start listening to new block epochs. We should get a notification
111
        // with the current best block immediately.
112
        blockEpochs, err := b.notifier.RegisterBlockEpochNtfn(nil)
3✔
113
        if err != nil {
4✔
114
                return fmt.Errorf("register block epoch ntfn: %w", err)
1✔
115
        }
1✔
116

117
        clog.Infof("BlockbeatDispatcher is starting with %d consumer queues",
2✔
118
                len(b.consumerQueues))
2✔
119
        defer clog.Debug("BlockbeatDispatcher started")
2✔
120

2✔
121
        b.wg.Add(1)
2✔
122
        go b.dispatchBlocks(blockEpochs)
2✔
123

2✔
124
        return nil
2✔
125
}
126

127
// Stop shuts down the blockbeat dispatcher.
128
func (b *BlockbeatDispatcher) Stop() {
2✔
129
        clog.Info("BlockbeatDispatcher is stopping")
2✔
130
        defer clog.Debug("BlockbeatDispatcher stopped")
2✔
131

2✔
132
        // Signal the dispatchBlocks goroutine to stop.
2✔
133
        close(b.quit)
2✔
134
        b.wg.Wait()
2✔
135
}
2✔
136

137
func (b *BlockbeatDispatcher) log() btclog.Logger {
10✔
138
        return b.beat.logger()
10✔
139
}
10✔
140

141
// dispatchBlocks listens to new block epoch and dispatches it to all the
142
// consumers. Each queue is notified concurrently, and the consumers in the
143
// same queue are notified sequentially.
144
//
145
// NOTE: Must be run as a goroutine.
146
func (b *BlockbeatDispatcher) dispatchBlocks(
147
        blockEpochs *chainntnfs.BlockEpochEvent) {
3✔
148

3✔
149
        defer b.wg.Done()
3✔
150
        defer blockEpochs.Cancel()
3✔
151

3✔
152
        for {
8✔
153
                select {
5✔
154
                case blockEpoch, ok := <-blockEpochs.Epochs:
1✔
155
                        if !ok {
1✔
156
                                clog.Debugf("Block epoch channel closed")
×
157

×
158
                                return
×
159
                        }
×
160

161
                        // Log a separator so it's easier to identify when a
162
                        // new block arrives for subsystems.
163
                        clog.Debugf("%v", lnutils.NewSeparatorClosure())
1✔
164

1✔
165
                        clog.Infof("Received new block %v at height %d, "+
1✔
166
                                "notifying consumers...", blockEpoch.Hash,
1✔
167
                                blockEpoch.Height)
1✔
168

1✔
169
                        // Record the time it takes the consumer to process
1✔
170
                        // this block.
1✔
171
                        start := time.Now()
1✔
172

1✔
173
                        // Update the current block epoch.
1✔
174
                        b.beat = NewBeat(*blockEpoch)
1✔
175

1✔
176
                        // Notify all consumers.
1✔
177
                        err := b.notifyQueues()
1✔
178
                        if err != nil {
1✔
179
                                b.log().Errorf("Notify block failed: %v", err)
×
180
                        }
×
181

182
                        b.log().Infof("Notified all consumers on new block "+
1✔
183
                                "in %v", time.Since(start))
1✔
184

185
                // A query has been made to fetch the current height, we now
186
                // send the height from its current beat.
187
                case query := <-b.queryHeightChan:
1✔
188
                        // The beat may not be set yet, e.g., during the startup
1✔
189
                        // the query is made before the block epoch being sent.
1✔
190
                        height := int32(0)
1✔
191
                        if b.beat != nil {
2✔
192
                                height = b.beat.Height()
1✔
193
                        }
1✔
194

195
                        query.respChan <- height
1✔
196

197
                case <-b.quit:
2✔
198
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
2✔
199
                                "received")
2✔
200

2✔
201
                        return
2✔
202
                }
203
        }
204
}
205

206
// CurrentHeight returns the current best height known to the dispatcher. 0 is
207
// returned if the dispatcher is shutting down.
208
func (b *BlockbeatDispatcher) CurrentHeight() int32 {
2✔
209
        query := newQuery()
2✔
210

2✔
211
        select {
2✔
212
        case b.queryHeightChan <- query:
2✔
213

UNCOV
214
        case <-b.quit:
×
UNCOV
215
                clog.Debugf("BlockbeatDispatcher quit before query")
×
UNCOV
216
                return 0
×
217
        }
218

219
        select {
2✔
220
        case height := <-query.respChan:
1✔
221
                clog.Debugf("Responded current height: %v", height)
1✔
222
                return height
1✔
223

224
        case <-b.quit:
1✔
225
                clog.Debugf("BlockbeatDispatcher quit before response")
1✔
226
                return 0
1✔
227
        }
228
}
229

230
// notifyQueues notifies each queue concurrently about the latest block epoch.
231
func (b *BlockbeatDispatcher) notifyQueues() error {
3✔
232
        // errChans is a map of channels that will be used to receive errors
3✔
233
        // returned from notifying the consumers.
3✔
234
        errChans := make(map[uint32]chan error, len(b.consumerQueues))
3✔
235

3✔
236
        // Notify each queue in goroutines.
3✔
237
        for qid, consumers := range b.consumerQueues {
7✔
238
                b.log().Debugf("Notifying queue=%d with %d consumers", qid,
4✔
239
                        len(consumers))
4✔
240

4✔
241
                // Create a signal chan.
4✔
242
                errChan := make(chan error, 1)
4✔
243
                errChans[qid] = errChan
4✔
244

4✔
245
                // Notify each queue concurrently.
4✔
246
                go func(qid uint32, c []Consumer, beat Blockbeat) {
8✔
247
                        // Notify each consumer in this queue sequentially.
4✔
248
                        errChan <- DispatchSequential(beat, c)
4✔
249
                }(qid, consumers, b.beat)
4✔
250
        }
251

252
        // Wait for all consumers in each queue to finish.
253
        for qid, errChan := range errChans {
7✔
254
                select {
4✔
255
                case err := <-errChan:
4✔
256
                        if err != nil {
5✔
257
                                return fmt.Errorf("queue=%d got err: %w", qid,
1✔
258
                                        err)
1✔
259
                        }
1✔
260

261
                        b.log().Debugf("Notified queue=%d", qid)
3✔
262

263
                case <-b.quit:
×
264
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
×
265
                                "received, exit notifyQueues")
×
266

×
267
                        return nil
×
268
                }
269
        }
270

271
        return nil
2✔
272
}
273

274
// DispatchSequential takes a list of consumers and notify them about the new
275
// epoch sequentially. It requires the consumer to finish processing the block
276
// within the specified time, otherwise a timeout error is returned.
277
func DispatchSequential(b Blockbeat, consumers []Consumer) error {
5✔
278
        for _, c := range consumers {
12✔
279
                // Send the beat to the consumer.
7✔
280
                err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
7✔
281
                if err != nil {
8✔
282
                        b.logger().Errorf("Failed to process block: %v", err)
1✔
283

1✔
284
                        return err
1✔
285
                }
1✔
286
        }
287

288
        return nil
4✔
289
}
290

291
// DispatchConcurrent notifies each consumer concurrently about the blockbeat.
292
// It requires the consumer to finish processing the block within the specified
293
// time, otherwise a timeout error is returned.
UNCOV
294
func DispatchConcurrent(b Blockbeat, consumers []Consumer) error {
×
UNCOV
295
        eg := &errgroup.Group{}
×
UNCOV
296

×
UNCOV
297
        // Notify each queue in goroutines.
×
UNCOV
298
        for _, c := range consumers {
×
UNCOV
299
                // Notify each consumer concurrently.
×
UNCOV
300
                eg.Go(func() error {
×
UNCOV
301
                        // Send the beat to the consumer.
×
UNCOV
302
                        err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
×
UNCOV
303

×
UNCOV
304
                        // Exit early if there's no error.
×
UNCOV
305
                        if err == nil {
×
UNCOV
306
                                return nil
×
UNCOV
307
                        }
×
308

309
                        b.logger().Errorf("Consumer=%v failed to process "+
×
310
                                "block: %v", c.Name(), err)
×
311

×
312
                        return err
×
313
                })
314
        }
315

316
        // Wait for all consumers in each queue to finish.
UNCOV
317
        if err := eg.Wait(); err != nil {
×
318
                return err
×
319
        }
×
320

UNCOV
321
        return nil
×
322
}
323

324
// notifyAndWait sends the blockbeat to the specified consumer. It requires the
325
// consumer to finish processing the block within the specified time, otherwise
326
// a timeout error is returned.
327
func notifyAndWait(b Blockbeat, c Consumer, timeout time.Duration) error {
10✔
328
        b.logger().Debugf("Waiting for consumer[%s] to process it", c.Name())
10✔
329

10✔
330
        // Record the time it takes the consumer to process this block.
10✔
331
        start := time.Now()
10✔
332

10✔
333
        errChan := make(chan error, 1)
10✔
334
        go func() {
20✔
335
                errChan <- c.ProcessBlock(b)
10✔
336
        }()
10✔
337

338
        // We expect the consumer to finish processing this block under 30s,
339
        // otherwise a timeout error is returned.
340
        select {
10✔
341
        case err := <-errChan:
9✔
342
                if err == nil {
16✔
343
                        break
7✔
344
                }
345

346
                return fmt.Errorf("%s got err in ProcessBlock: %w", c.Name(),
2✔
347
                        err)
2✔
348

349
        case <-time.After(timeout):
1✔
350
                return fmt.Errorf("consumer %s: %w", c.Name(),
1✔
351
                        ErrProcessBlockTimeout)
1✔
352
        }
353

354
        b.logger().Debugf("Consumer[%s] processed block in %v", c.Name(),
7✔
355
                time.Since(start))
7✔
356

7✔
357
        return nil
7✔
358
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc