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

lightningnetwork / lnd / 16005100421

01 Jul 2025 04:35PM UTC coverage: 57.772% (+0.6%) from 57.216%
16005100421

Pull #10018

github

web-flow
Merge 0b73fe73c into d8a12a5e5
Pull Request #10018: Refactor link's long methods

390 of 746 new or added lines in 1 file covered. (52.28%)

41 existing lines in 11 files now uncovered.

98433 of 170383 relevant lines covered (57.77%)

1.79 hits per line

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

80.19
/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 {
3✔
69
        return &query{
3✔
70
                respChan: make(chan int32, 1),
3✔
71
        }
3✔
72
}
3✔
73

74
// NewBlockbeatDispatcher returns a new blockbeat dispatcher instance.
75
func NewBlockbeatDispatcher(n chainntnfs.ChainNotifier) *BlockbeatDispatcher {
3✔
76
        return &BlockbeatDispatcher{
3✔
77
                notifier:        n,
3✔
78
                quit:            make(chan struct{}),
3✔
79
                consumerQueues:  make(map[uint32][]Consumer),
3✔
80
                queryHeightChan: make(chan *query, 1),
3✔
81
        }
3✔
82
}
3✔
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) {
3✔
89
        qid := b.counter.Add(1)
3✔
90

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

3✔
95
        for _, c := range consumers {
6✔
96
                clog.Debugf("Consumer [%s] registered in queue %d", c.Name(),
3✔
97
                        qid)
3✔
98
        }
3✔
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 {
3✔
105
        // Make sure consumers are registered.
3✔
106
        if len(b.consumerQueues) == 0 {
3✔
107
                return fmt.Errorf("no consumers registered")
×
108
        }
×
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 {
3✔
114
                return fmt.Errorf("register block epoch ntfn: %w", err)
×
115
        }
×
116

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

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

3✔
124
        return nil
3✔
125
}
126

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

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

137
func (b *BlockbeatDispatcher) log() btclog.Logger {
3✔
138
        // There's no guarantee that the `b.beat` is initialized when the
3✔
139
        // dispatcher shuts down, especially in the case where the node is
3✔
140
        // running as a remote signer, which doesn't have a chainbackend. In
3✔
141
        // that case we will use the package logger.
3✔
142
        if b.beat == nil {
3✔
143
                return clog
×
144
        }
×
145

146
        return b.beat.logger()
3✔
147
}
148

149
// dispatchBlocks listens to new block epoch and dispatches it to all the
150
// consumers. Each queue is notified concurrently, and the consumers in the
151
// same queue are notified sequentially.
152
//
153
// NOTE: Must be run as a goroutine.
154
func (b *BlockbeatDispatcher) dispatchBlocks(
155
        blockEpochs *chainntnfs.BlockEpochEvent) {
3✔
156

3✔
157
        defer b.wg.Done()
3✔
158
        defer blockEpochs.Cancel()
3✔
159

3✔
160
        for {
6✔
161
                select {
3✔
162
                case blockEpoch, ok := <-blockEpochs.Epochs:
3✔
163
                        if !ok {
3✔
164
                                clog.Debugf("Block epoch channel closed")
×
165

×
166
                                return
×
167
                        }
×
168

169
                        // Log a separator so it's easier to identify when a
170
                        // new block arrives for subsystems.
171
                        clog.Debugf("%v", lnutils.NewSeparatorClosure())
3✔
172

3✔
173
                        clog.Debugf("Received new block %v at height %d, "+
3✔
174
                                "notifying consumers...", blockEpoch.Hash,
3✔
175
                                blockEpoch.Height)
3✔
176

3✔
177
                        // Record the time it takes the consumer to process
3✔
178
                        // this block.
3✔
179
                        start := time.Now()
3✔
180

3✔
181
                        // Update the current block epoch.
3✔
182
                        b.beat = NewBeat(*blockEpoch)
3✔
183

3✔
184
                        // Notify all consumers.
3✔
185
                        err := b.notifyQueues()
3✔
186
                        if err != nil {
3✔
187
                                b.log().Errorf("Notify block failed: %v", err)
×
188
                        }
×
189

190
                        b.log().Debugf("Notified all consumers on new block "+
3✔
191
                                "in %v", time.Since(start))
3✔
192

193
                // A query has been made to fetch the current height, we now
194
                // send the height from its current beat.
195
                case query := <-b.queryHeightChan:
3✔
196
                        // The beat may not be set yet, e.g., during the startup
3✔
197
                        // the query is made before the block epoch being sent.
3✔
198
                        height := int32(0)
3✔
199
                        if b.beat != nil {
6✔
200
                                height = b.beat.Height()
3✔
201
                        }
3✔
202

203
                        query.respChan <- height
3✔
204

205
                case <-b.quit:
3✔
206
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
3✔
207
                                "received")
3✔
208

3✔
209
                        return
3✔
210
                }
211
        }
212
}
213

214
// CurrentHeight returns the current best height known to the dispatcher. 0 is
215
// returned if the dispatcher is shutting down.
216
func (b *BlockbeatDispatcher) CurrentHeight() int32 {
3✔
217
        query := newQuery()
3✔
218

3✔
219
        select {
3✔
220
        case b.queryHeightChan <- query:
3✔
221

222
        case <-b.quit:
×
223
                clog.Debugf("BlockbeatDispatcher quit before query")
×
224
                return 0
×
225
        }
226

227
        select {
3✔
228
        case height := <-query.respChan:
3✔
229
                clog.Debugf("Responded current height: %v", height)
3✔
230
                return height
3✔
231

232
        case <-b.quit:
×
233
                clog.Debugf("BlockbeatDispatcher quit before response")
×
234
                return 0
×
235
        }
236
}
237

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

3✔
244
        // Notify each queue in goroutines.
3✔
245
        for qid, consumers := range b.consumerQueues {
6✔
246
                b.log().Debugf("Notifying queue=%d with %d consumers", qid,
3✔
247
                        len(consumers))
3✔
248

3✔
249
                // Create a signal chan.
3✔
250
                errChan := make(chan error, 1)
3✔
251
                errChans[qid] = errChan
3✔
252

3✔
253
                // Notify each queue concurrently.
3✔
254
                go func(qid uint32, c []Consumer, beat Blockbeat) {
6✔
255
                        // Notify each consumer in this queue sequentially.
3✔
256
                        errChan <- DispatchSequential(beat, c)
3✔
257
                }(qid, consumers, b.beat)
3✔
258
        }
259

260
        // Wait for all consumers in each queue to finish.
261
        for qid, errChan := range errChans {
6✔
262
                select {
3✔
263
                case err := <-errChan:
3✔
264
                        if err != nil {
3✔
265
                                return fmt.Errorf("queue=%d got err: %w", qid,
×
266
                                        err)
×
267
                        }
×
268

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

UNCOV
271
                case <-b.quit:
×
UNCOV
272
                        b.log().Debugf("BlockbeatDispatcher quit signal " +
×
UNCOV
273
                                "received, exit notifyQueues")
×
UNCOV
274

×
UNCOV
275
                        return nil
×
276
                }
277
        }
278

279
        return nil
3✔
280
}
281

282
// DispatchSequential takes a list of consumers and notify them about the new
283
// epoch sequentially. It requires the consumer to finish processing the block
284
// within the specified time, otherwise a timeout error is returned.
285
func DispatchSequential(b Blockbeat, consumers []Consumer) error {
3✔
286
        for _, c := range consumers {
6✔
287
                // Send the beat to the consumer.
3✔
288
                err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
3✔
289
                if err != nil {
3✔
290
                        b.logger().Errorf("Failed to process block: %v", err)
×
291

×
292
                        return err
×
293
                }
×
294
        }
295

296
        return nil
3✔
297
}
298

299
// DispatchConcurrent notifies each consumer concurrently about the blockbeat.
300
// It requires the consumer to finish processing the block within the specified
301
// time, otherwise a timeout error is returned.
302
func DispatchConcurrent(b Blockbeat, consumers []Consumer) error {
3✔
303
        eg := &errgroup.Group{}
3✔
304

3✔
305
        // Notify each queue in goroutines.
3✔
306
        for _, c := range consumers {
6✔
307
                // Notify each consumer concurrently.
3✔
308
                eg.Go(func() error {
6✔
309
                        // Send the beat to the consumer.
3✔
310
                        err := notifyAndWait(b, c, DefaultProcessBlockTimeout)
3✔
311

3✔
312
                        // Exit early if there's no error.
3✔
313
                        if err == nil {
6✔
314
                                return nil
3✔
315
                        }
3✔
316

317
                        b.logger().Errorf("Consumer=%v failed to process "+
×
318
                                "block: %v", c.Name(), err)
×
319

×
320
                        return err
×
321
                })
322
        }
323

324
        // Wait for all consumers in each queue to finish.
325
        if err := eg.Wait(); err != nil {
3✔
326
                return err
×
327
        }
×
328

329
        return nil
3✔
330
}
331

332
// notifyAndWait sends the blockbeat to the specified consumer. It requires the
333
// consumer to finish processing the block within the specified time, otherwise
334
// a timeout error is returned.
335
func notifyAndWait(b Blockbeat, c Consumer, timeout time.Duration) error {
3✔
336
        b.logger().Debugf("Waiting for consumer[%s] to process it", c.Name())
3✔
337

3✔
338
        // Record the time it takes the consumer to process this block.
3✔
339
        start := time.Now()
3✔
340

3✔
341
        errChan := make(chan error, 1)
3✔
342
        go func() {
6✔
343
                errChan <- c.ProcessBlock(b)
3✔
344
        }()
3✔
345

346
        // We expect the consumer to finish processing this block under 30s,
347
        // otherwise a timeout error is returned.
348
        select {
3✔
349
        case err := <-errChan:
3✔
350
                if err == nil {
6✔
351
                        break
3✔
352
                }
353

354
                return fmt.Errorf("%s got err in ProcessBlock: %w", c.Name(),
×
355
                        err)
×
356

357
        case <-time.After(timeout):
×
358
                return fmt.Errorf("consumer %s: %w", c.Name(),
×
359
                        ErrProcessBlockTimeout)
×
360
        }
361

362
        b.logger().Debugf("Consumer[%s] processed block in %v", c.Name(),
3✔
363
                time.Since(start))
3✔
364

3✔
365
        return nil
3✔
366
}
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