-
Notifications
You must be signed in to change notification settings - Fork 176
/
consensus_committee.go
510 lines (458 loc) · 19.1 KB
/
consensus_committee.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
package committees
import (
"fmt"
"sync"
"go.uber.org/atomic"
"github.com/onflow/flow-go/consensus/hotstuff"
"github.com/onflow/flow-go/consensus/hotstuff/committees/leader"
"github.com/onflow/flow-go/consensus/hotstuff/model"
"github.com/onflow/flow-go/model/flow"
"github.com/onflow/flow-go/model/flow/filter"
"github.com/onflow/flow-go/module/component"
"github.com/onflow/flow-go/module/irrecoverable"
"github.com/onflow/flow-go/state/protocol"
"github.com/onflow/flow-go/state/protocol/events"
"github.com/onflow/flow-go/state/protocol/prg"
)
// staticEpochInfo contains leader selection and the initial committee for one epoch.
// This data structure must not be mutated after construction.
type staticEpochInfo struct {
firstView uint64 // first view of the epoch (inclusive)
finalView uint64 // final view of the epoch (inclusive)
randomSource []byte // random source of epoch
leaders *leader.LeaderSelection // pre-computed leader selection for the epoch
// TODO: should use identity skeleton https://github.com/dapperlabs/flow-go/issues/6232
initialCommittee flow.IdentitySkeletonList
initialCommitteeMap map[flow.Identifier]*flow.IdentitySkeleton
weightThresholdForQC uint64 // computed based on initial committee weights
weightThresholdForTO uint64 // computed based on initial committee weights
dkg hotstuff.DKG
}
// newStaticEpochInfo returns the static epoch information from the epoch.
// This can be cached and used for all by-view queries for this epoch.
func newStaticEpochInfo(epoch protocol.Epoch) (*staticEpochInfo, error) {
firstView, err := epoch.FirstView()
if err != nil {
return nil, fmt.Errorf("could not get first view: %w", err)
}
finalView, err := epoch.FinalView()
if err != nil {
return nil, fmt.Errorf("could not get final view: %w", err)
}
randomSource, err := epoch.RandomSource()
if err != nil {
return nil, fmt.Errorf("could not get random source: %w", err)
}
leaders, err := leader.SelectionForConsensus(epoch)
if err != nil {
return nil, fmt.Errorf("could not get leader selection: %w", err)
}
initialIdentities, err := epoch.InitialIdentities()
if err != nil {
return nil, fmt.Errorf("could not initial identities: %w", err)
}
initialCommittee := initialIdentities.Filter(filter.IsVotingConsensusCommitteeMember).ToSkeleton()
dkg, err := epoch.DKG()
if err != nil {
return nil, fmt.Errorf("could not get dkg: %w", err)
}
totalWeight := initialCommittee.TotalWeight()
epochInfo := &staticEpochInfo{
firstView: firstView,
finalView: finalView,
randomSource: randomSource,
leaders: leaders,
initialCommittee: initialCommittee,
initialCommitteeMap: initialCommittee.Lookup(),
weightThresholdForQC: WeightThresholdToBuildQC(totalWeight),
weightThresholdForTO: WeightThresholdToTimeout(totalWeight),
dkg: dkg,
}
return epochInfo, nil
}
// newEmergencyFallbackEpoch creates an artificial fallback epoch generated from
// the last committed epoch at the time epoch emergency fallback is triggered.
// The fallback epoch:
// * begins after the last committed epoch
// * lasts until the next spork (estimated 6 months)
// * has the same static committee as the last committed epoch
func newEmergencyFallbackEpoch(lastCommittedEpoch *staticEpochInfo) (*staticEpochInfo, error) {
rng, err := prg.New(lastCommittedEpoch.randomSource, prg.ConsensusLeaderSelection, nil)
if err != nil {
return nil, fmt.Errorf("could not create rng from seed: %w", err)
}
leaders, err := leader.ComputeLeaderSelection(
lastCommittedEpoch.finalView+1,
rng,
leader.EstimatedSixMonthOfViews,
lastCommittedEpoch.initialCommittee,
)
if err != nil {
return nil, fmt.Errorf("could not compute leader selection for fallback epoch: %w", err)
}
epochInfo := &staticEpochInfo{
firstView: lastCommittedEpoch.finalView + 1,
finalView: lastCommittedEpoch.finalView + leader.EstimatedSixMonthOfViews,
randomSource: lastCommittedEpoch.randomSource,
leaders: leaders,
initialCommittee: lastCommittedEpoch.initialCommittee,
initialCommitteeMap: lastCommittedEpoch.initialCommitteeMap,
weightThresholdForQC: lastCommittedEpoch.weightThresholdForQC,
weightThresholdForTO: lastCommittedEpoch.weightThresholdForTO,
dkg: lastCommittedEpoch.dkg,
}
return epochInfo, nil
}
// Consensus represents the main committee for consensus nodes. The consensus
// committee might be active for multiple successive epochs.
type Consensus struct {
state protocol.State // the protocol state
me flow.Identifier // the node ID of this node
mu sync.RWMutex // protects access to epochs
epochs map[uint64]*staticEpochInfo // cache of initial committee & leader selection per epoch
committedEpochsCh chan *flow.Header // protocol events for newly committed epochs (the first block of the epoch is passed over the channel)
epochEmergencyFallback chan struct{} // protocol event for epoch emergency fallback
isEpochFallbackHandled *atomic.Bool // ensure we only inject fallback epoch once
events.Noop // implements protocol.Consumer
component.Component
}
var _ protocol.Consumer = (*Consensus)(nil)
var _ hotstuff.Replicas = (*Consensus)(nil)
var _ hotstuff.DynamicCommittee = (*Consensus)(nil)
func NewConsensusCommittee(state protocol.State, me flow.Identifier) (*Consensus, error) {
com := &Consensus{
state: state,
me: me,
epochs: make(map[uint64]*staticEpochInfo),
committedEpochsCh: make(chan *flow.Header, 1),
epochEmergencyFallback: make(chan struct{}, 1),
isEpochFallbackHandled: atomic.NewBool(false),
}
com.Component = component.NewComponentManagerBuilder().
AddWorker(com.handleProtocolEvents).
Build()
final := state.Final()
// pre-compute leader selection for all presently relevant committed epochs
epochs := make([]protocol.Epoch, 0, 3)
// we always prepare the current epoch
epochs = append(epochs, final.Epochs().Current())
// we prepare the previous epoch, if one exists
exists, err := protocol.PreviousEpochExists(final)
if err != nil {
return nil, fmt.Errorf("could not check previous epoch exists: %w", err)
}
if exists {
epochs = append(epochs, final.Epochs().Previous())
}
// we prepare the next epoch, if it is committed
phase, err := final.Phase()
if err != nil {
return nil, fmt.Errorf("could not check epoch phase: %w", err)
}
if phase == flow.EpochPhaseCommitted {
epochs = append(epochs, final.Epochs().Next())
}
for _, epoch := range epochs {
_, err = com.prepareEpoch(epoch)
if err != nil {
return nil, fmt.Errorf("could not prepare initial epochs: %w", err)
}
}
// if epoch emergency fallback was triggered, inject the fallback epoch
triggered, err := state.Params().EpochFallbackTriggered()
if err != nil {
return nil, fmt.Errorf("could not check epoch fallback: %w", err)
}
if triggered {
err = com.onEpochEmergencyFallbackTriggered()
if err != nil {
return nil, fmt.Errorf("could not prepare emergency fallback epoch: %w", err)
}
}
return com, nil
}
// IdentitiesByBlock returns the identities of all authorized consensus participants at the given block.
// The order of the identities is the canonical order.
// ERROR conditions:
// - state.ErrUnknownSnapshotReference if the blockID is for an unknown block
func (c *Consensus) IdentitiesByBlock(blockID flow.Identifier) (flow.IdentityList, error) {
il, err := c.state.AtBlockID(blockID).Identities(filter.IsVotingConsensusCommitteeMember)
if err != nil {
return nil, fmt.Errorf("could not identities at block %x: %w", blockID, err) // state.ErrUnknownSnapshotReference or exception
}
return il, nil
}
// IdentityByBlock returns the identity of the node with the given node ID at the given block.
// ERROR conditions:
// - model.InvalidSignerError if participantID does NOT correspond to an authorized HotStuff participant at the specified block.
// - state.ErrUnknownSnapshotReference if the blockID is for an unknown block
func (c *Consensus) IdentityByBlock(blockID flow.Identifier, nodeID flow.Identifier) (*flow.Identity, error) {
identity, err := c.state.AtBlockID(blockID).Identity(nodeID)
if err != nil {
if protocol.IsIdentityNotFound(err) {
return nil, model.NewInvalidSignerErrorf("id %v is not a valid node id: %w", nodeID, err)
}
return nil, fmt.Errorf("could not get identity for node ID %x: %w", nodeID, err) // state.ErrUnknownSnapshotReference or exception
}
if !filter.IsVotingConsensusCommitteeMember(identity) {
return nil, model.NewInvalidSignerErrorf("node %v is not an authorized hotstuff voting participant", nodeID)
}
return identity, nil
}
// IdentitiesByEpoch returns the committee identities in the epoch which contains
// the given view.
// CAUTION: This method considers epochs outside of Previous, Current, Next, w.r.t. the
// finalized block, to be unknown. https://github.com/onflow/flow-go/issues/4085
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known.
// This is an expected error and must be handled.
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) IdentitiesByEpoch(view uint64) (flow.IdentitySkeletonList, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return nil, err
}
return epochInfo.initialCommittee, nil
}
// IdentityByEpoch returns the identity for the given node ID, in the epoch which
// contains the given view.
// CAUTION: This method considers epochs outside of Previous, Current, Next, w.r.t. the
// finalized block, to be unknown. https://github.com/onflow/flow-go/issues/4085
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known.
// This is an expected error and must be handled.
// - model.InvalidSignerError if nodeID was not listed by the Epoch Setup event as an
// authorized consensus participants.
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) IdentityByEpoch(view uint64, participantID flow.Identifier) (*flow.IdentitySkeleton, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return nil, err
}
identity, ok := epochInfo.initialCommitteeMap[participantID]
if !ok {
return nil, model.NewInvalidSignerErrorf("id %v is not a valid node id", participantID)
}
return identity, nil
}
// LeaderForView returns the node ID of the leader for the given view.
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known.
// This is an expected error and must be handled.
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) LeaderForView(view uint64) (flow.Identifier, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return flow.ZeroID, err
}
leaderID, err := epochInfo.leaders.LeaderForView(view)
if leader.IsInvalidViewError(err) {
// an invalid view error indicates that no leader was computed for this view
// this is a fatal internal error, because the view necessarily is within an
// epoch for which we have pre-computed leader selection
return flow.ZeroID, fmt.Errorf("unexpected inconsistency in epoch view spans for view %d: %v", view, err)
}
if err != nil {
return flow.ZeroID, err
}
return leaderID, nil
}
// QuorumThresholdForView returns the minimum weight required to build a valid
// QC in the given view. The weight threshold only changes at epoch boundaries
// and is computed based on the initial committee weights.
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known.
// This is an expected error and must be handled.
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) QuorumThresholdForView(view uint64) (uint64, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return 0, err
}
return epochInfo.weightThresholdForQC, nil
}
func (c *Consensus) Self() flow.Identifier {
return c.me
}
// TimeoutThresholdForView returns the minimum weight of observed timeout objects
// to safely immediately timeout for the current view. The weight threshold only
// changes at epoch boundaries and is computed based on the initial committee weights.
func (c *Consensus) TimeoutThresholdForView(view uint64) (uint64, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return 0, err
}
return epochInfo.weightThresholdForTO, nil
}
// DKG returns the DKG for epoch which includes the given view.
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known.
// This is an expected error and must be handled.
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) DKG(view uint64) (hotstuff.DKG, error) {
epochInfo, err := c.staticEpochInfoByView(view)
if err != nil {
return nil, err
}
return epochInfo.dkg, nil
}
// handleProtocolEvents processes queued Epoch events `EpochCommittedPhaseStarted`
// and `EpochEmergencyFallbackTriggered`. This function permanently utilizes a worker
// routine until the `Component` terminates.
// When we observe a new epoch being committed, we compute
// the leader selection and cache static info for the epoch. When we observe
// epoch emergency fallback being triggered, we inject a fallback epoch.
func (c *Consensus) handleProtocolEvents(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
ready()
for {
select {
case <-ctx.Done():
return
case block := <-c.committedEpochsCh:
epoch := c.state.AtBlockID(block.ID()).Epochs().Next()
_, err := c.prepareEpoch(epoch)
if err != nil {
ctx.Throw(err)
}
case <-c.epochEmergencyFallback:
err := c.onEpochEmergencyFallbackTriggered()
if err != nil {
ctx.Throw(err)
}
}
}
}
// EpochCommittedPhaseStarted informs the `committee.Consensus` that the block starting the Epoch Committed Phase has been finalized.
func (c *Consensus) EpochCommittedPhaseStarted(_ uint64, first *flow.Header) {
c.committedEpochsCh <- first
}
// EpochEmergencyFallbackTriggered passes the protocol event to the worker thread.
func (c *Consensus) EpochEmergencyFallbackTriggered() {
c.epochEmergencyFallback <- struct{}{}
}
// onEpochEmergencyFallbackTriggered handles the protocol event for emergency epoch
// fallback mode being triggered. When this occurs, we inject a fallback epoch
// to the committee which extends the current epoch.
// This method must also be called on initialization, if emergency fallback mode
// was triggered in the past.
// No errors are expected during normal operation.
func (c *Consensus) onEpochEmergencyFallbackTriggered() error {
// we respond to epoch fallback being triggered at most once, therefore
// the core logic is protected by an atomic bool.
// although it is only valid for epoch fallback to be triggered once per spork,
// we must account for repeated delivery of protocol events.
if !c.isEpochFallbackHandled.CompareAndSwap(false, true) {
return nil
}
currentEpochCounter, err := c.state.Final().Epochs().Current().Counter()
if err != nil {
return fmt.Errorf("could not get current epoch counter: %w", err)
}
c.mu.RLock()
// sanity check: current epoch must be cached already
currentEpoch, ok := c.epochs[currentEpochCounter]
if !ok {
c.mu.RUnlock()
return fmt.Errorf("epoch fallback: could not find current epoch (counter=%d) info", currentEpochCounter)
}
// sanity check: next epoch must never be committed, therefore must not be cached
_, ok = c.epochs[currentEpochCounter+1]
c.mu.RUnlock()
if ok {
return fmt.Errorf("epoch fallback: next epoch (counter=%d) is cached contrary to expectation", currentEpochCounter+1)
}
fallbackEpoch, err := newEmergencyFallbackEpoch(currentEpoch)
if err != nil {
return fmt.Errorf("could not construct fallback epoch: %w", err)
}
// cache the epoch info
c.mu.Lock()
c.epochs[currentEpochCounter+1] = fallbackEpoch
c.mu.Unlock()
return nil
}
// staticEpochInfoByView retrieves the previously cached static epoch info for
// the epoch which includes the given view. If no epoch is known for the given
// view, we will attempt to cache the next epoch.
//
// Error returns:
// - model.ErrViewForUnknownEpoch if no committed epoch containing the given view is known
// - unspecific error in case of unexpected problems and bugs
func (c *Consensus) staticEpochInfoByView(view uint64) (*staticEpochInfo, error) {
// look for an epoch matching this view for which we have already pre-computed
// leader selection. Epochs last ~500k views, so we find the epoch here 99.99%
// of the time. Since epochs are long-lived and we only cache the most recent 3,
// this linear map iteration is inexpensive.
c.mu.RLock()
for _, epoch := range c.epochs {
if epoch.firstView <= view && view <= epoch.finalView {
c.mu.RUnlock()
return epoch, nil
}
}
c.mu.RUnlock()
return nil, model.ErrViewForUnknownEpoch
}
// prepareEpoch pre-computes and stores the static epoch information for the
// given epoch, including leader selection. Calling prepareEpoch multiple times
// for the same epoch returns cached static epoch information.
// Input must be a committed epoch.
// No errors are expected during normal operation.
func (c *Consensus) prepareEpoch(epoch protocol.Epoch) (*staticEpochInfo, error) {
counter, err := epoch.Counter()
if err != nil {
return nil, fmt.Errorf("could not get counter for epoch to prepare: %w", err)
}
// this is a no-op if we have already computed static info for this epoch
c.mu.RLock()
epochInfo, exists := c.epochs[counter]
c.mu.RUnlock()
if exists {
return epochInfo, nil
}
epochInfo, err = newStaticEpochInfo(epoch)
if err != nil {
return nil, fmt.Errorf("could not create static epoch info for epch %d: %w", counter, err)
}
// sanity check: ensure new epoch has contiguous views with the prior epoch
c.mu.RLock()
prevEpochInfo, exists := c.epochs[counter-1]
c.mu.RUnlock()
if exists {
if epochInfo.firstView != prevEpochInfo.finalView+1 {
return nil, fmt.Errorf("non-contiguous view ranges between consecutive epochs (epoch_%d=[%d,%d], epoch_%d=[%d,%d])",
counter-1, prevEpochInfo.firstView, prevEpochInfo.finalView,
counter, epochInfo.firstView, epochInfo.finalView)
}
}
// cache the epoch info
c.mu.Lock()
defer c.mu.Unlock()
c.epochs[counter] = epochInfo
// now prune any old epochs, if we have exceeded our maximum of 3
// if we have fewer than 3 epochs, this is a no-op
c.pruneEpochInfo()
return epochInfo, nil
}
// pruneEpochInfo removes any epochs older than the most recent 3.
// NOTE: Not safe for concurrent use - the caller must first acquire the lock.
func (c *Consensus) pruneEpochInfo() {
// find the maximum counter, including the epoch we just computed
max := uint64(0)
for counter := range c.epochs {
if counter > max {
max = counter
}
}
// remove any epochs which aren't within the most recent 3
for counter := range c.epochs {
if counter+3 <= max {
delete(c.epochs, counter)
}
}
}