-
Notifications
You must be signed in to change notification settings - Fork 175
/
queue.go
449 lines (384 loc) · 15.8 KB
/
queue.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
package block_queue
import (
"fmt"
"sync"
"github.com/rs/zerolog"
"github.com/onflow/flow-go/model/flow"
"github.com/onflow/flow-go/module/mempool/entity"
)
var ErrMissingParent = fmt.Errorf("missing parent block")
// BlockQueue keeps track of state of blocks and determines which blocks are executable
// A block becomes executable when all the following conditions are met:
// 1. the block has been validated by consensus algorithm
// 2. the block's parent has been executed
// 3. all the collections included in the block have been received
type BlockQueue struct {
sync.Mutex
log zerolog.Logger
// if a block still exists in this map, it means the block has not been executed.
// it could either be one of the following cases:
// 1) block is not executed due to some of its collection is missing
// 2) block is not executed due to its parent block has not been executed
// 3) block is ready to execute, but the execution has not been finished yet.
// some consistency checks:
// 1) since an executed block must have been removed from this map, if a block's
// parent block has been executed, then its parent block must have been removed
// from this map
// 2) if a block's parent block has not been executed, then its parent block must still
// exist in this map
blocks map[flow.Identifier]*entity.ExecutableBlock
// a collection could be included in multiple blocks,
// when a missing block is received, it might trigger multiple blocks to be executable, which
// can be looked up by the map
// when a block is executed, its collections should be removed from this map unless a collection
// is still referenced by other blocks, which will eventually be removed when those blocks are
// executed.
collections map[flow.Identifier]*collectionInfo
// blockIDsByHeight is used to find next executable block.
// when a block is executed, the next executable block must be a block with height = current block height + 1
// the following map allows us to find the next executable block by height and their parent block ID
blockIDsByHeight map[uint64]map[flow.Identifier]*entity.ExecutableBlock
}
type MissingCollection struct {
BlockID flow.Identifier
Height uint64
Guarantee *flow.CollectionGuarantee
}
func (m *MissingCollection) ID() flow.Identifier {
return m.Guarantee.ID()
}
// collectionInfo is an internal struct used to keep track of the state of a collection,
// and the blocks that include the collection
type collectionInfo struct {
Collection *entity.CompleteCollection
IncludedIn map[flow.Identifier]*entity.ExecutableBlock
}
func NewBlockQueue(logger zerolog.Logger) *BlockQueue {
log := logger.With().Str("module", "block_queue").Logger()
return &BlockQueue{
log: log,
blocks: make(map[flow.Identifier]*entity.ExecutableBlock),
collections: make(map[flow.Identifier]*collectionInfo),
blockIDsByHeight: make(map[uint64]map[flow.Identifier]*entity.ExecutableBlock),
}
}
// HandleBlock is called when a new block is received, the parentFinalState indicates
// whether its parent block has been executed.
// Caller must ensure:
// 1. blocks are passsed in order, i.e. parent block is passed in before its child block
// 2. if a block's parent is not executed, then the parent block must be passed in first
// 3. if a block's parent is executed, then the parent's finalState must be passed in
// It returns (nil, nil, nil) if this block is a duplication
func (q *BlockQueue) HandleBlock(block *flow.Block, parentFinalState *flow.StateCommitment) (
[]*MissingCollection, // missing collections
[]*entity.ExecutableBlock, // blocks ready to execute
error, // exceptions
) {
q.Lock()
defer q.Unlock()
// check if the block already exists
blockID := block.ID()
executable, ok := q.blocks[blockID]
q.log.Debug().
Str("blockID", blockID.String()).
Uint64("height", block.Header.Height).
Bool("parent executed", parentFinalState != nil).
Msg("handle block")
if ok {
// handle the case where the block has seen before
return q.handleKnownBlock(executable, parentFinalState)
}
// handling a new block
_, parentExists := q.blocks[block.Header.ParentID]
// if parentFinalState is not provided, then its parent block must exists in the queue
// otherwise it's an exception
if parentFinalState == nil {
if !parentExists {
return nil, nil,
fmt.Errorf("block %s has no parent commitment, but its parent block %s does not exist in the queue: %w",
blockID, block.Header.ParentID, ErrMissingParent)
}
} else {
if parentExists {
// this is an edge case where A <- B, and B is received with A's final state, however,
// A is not executed yet.
// the reason this could happen is that there is a race condition in `OnBlockExecuted` and
// `HandleBlock`, when A's execution result (which contains the final state) has been
// saved to database, and B is received before `blockQueue.OnBlockExecuted(A)` is called, so
// `blockQueue.HandleBlock(B, A's final state)` will be called, which run into this case.
// In this case, if we consider A is executed, then return B as executables, and later
// when `OnBlockExecuted(A)` is called, it will return B as executables again, which
// will cause B to be executed twice.
// In order to prevent B to be executed twice, we will simply ignore A's final state,
// as if its parent has not been executed yet, then B is not executable. And when `OnBlockExecuted(A)`
// is called, it will return B as executables, so that both A and B will be executed only once.
// See test case: TestHandleBlockChildCalledBeforeOnBlockExecutedParent
q.log.Warn().
Str("blockID", blockID.String()).
Uint64("height", block.Header.Height).
Msgf("edge case: receiving block with parent commitment, but its parent block %s still exists",
block.Header.ParentID)
parentFinalState = nil
}
}
executable = &entity.ExecutableBlock{
Block: block,
StartState: parentFinalState,
}
// add block to blocks
q.blocks[blockID] = executable
// update collection
colls := make(map[flow.Identifier]*entity.CompleteCollection, len(block.Payload.Guarantees))
executable.CompleteCollections = colls
// find missing collections and update collection index
missingCollections := make([]*MissingCollection, 0, len(block.Payload.Guarantees))
for _, guarantee := range block.Payload.Guarantees {
colID := guarantee.ID()
colInfo, ok := q.collections[colID]
if ok {
// some other block also includes this collection
colInfo.IncludedIn[blockID] = executable
colls[colID] = colInfo.Collection
} else {
col := &entity.CompleteCollection{
Guarantee: guarantee,
}
colls[colID] = col
// add new collection to collections
q.collections[colID] = &collectionInfo{
Collection: col,
IncludedIn: map[flow.Identifier]*entity.ExecutableBlock{
blockID: executable,
},
}
missingCollections = append(missingCollections, missingCollectionForBlock(executable, guarantee))
}
}
// index height
blocksAtSameHeight, ok := q.blockIDsByHeight[block.Header.Height]
if !ok {
blocksAtSameHeight = make(map[flow.Identifier]*entity.ExecutableBlock)
q.blockIDsByHeight[block.Header.Height] = blocksAtSameHeight
}
blocksAtSameHeight[blockID] = executable
// check if the block is executable
var executables []*entity.ExecutableBlock
if executable.IsComplete() {
// executables might contain other siblings, but won't contain "executable",
// which is the block itself, that's because executables are created
// from OnBlockExecuted(
executables = []*entity.ExecutableBlock{executable}
}
return missingCollections, executables, nil
}
// HandleCollection is called when a new collection is received
// It returns a list of executable blocks that contains the collection
func (q *BlockQueue) HandleCollection(collection *flow.Collection) ([]*entity.ExecutableBlock, error) {
q.Lock()
defer q.Unlock()
// when a collection is received, we find the blocks the collection is included in,
// and check if the blocks become executable.
// Note a collection could be included in multiple blocks, so receiving a collection
// might trigger multiple blocks to be executable.
// check if the collection is for any block in the queue
colID := collection.ID()
colInfo, ok := q.collections[colID]
if !ok {
// no block in the queue includes this collection
return nil, nil
}
if colInfo.Collection.IsCompleted() {
// the collection is already received, no action needed because an action must
// have been returned when the collection is first received.
return nil, nil
}
// update collection
colInfo.Collection.Transactions = collection.Transactions
// check if any block, which includes this collection, became executable
executables := make([]*entity.ExecutableBlock, 0, len(colInfo.IncludedIn))
for _, block := range colInfo.IncludedIn {
if !block.IsComplete() {
continue
}
executables = append(executables, block)
}
if len(executables) == 0 {
return nil, nil
}
return executables, nil
}
// OnBlockExecuted is called when a block is executed
// It returns a list of executable blocks (usually its child blocks)
// The caller has to ensure OnBlockExecuted is not called in a wrong order, such as
// OnBlockExecuted(childBlock) being called before OnBlockExecuted(parentBlock).
func (q *BlockQueue) OnBlockExecuted(
blockID flow.Identifier,
commit flow.StateCommitment,
) ([]*entity.ExecutableBlock, error) {
q.Lock()
defer q.Unlock()
q.log.Debug().
Str("blockID", blockID.String()).
Hex("commit", commit[:]).
Msg("block executed")
return q.onBlockExecuted(blockID, commit)
}
func (q *BlockQueue) handleKnownBlock(executable *entity.ExecutableBlock, parentFinalState *flow.StateCommitment) (
[]*MissingCollection, // missing collections
[]*entity.ExecutableBlock, // blocks ready to execute
error, // exceptions
) {
// we have already received this block, and its parent still has not been executed yet
if executable.StartState == nil && parentFinalState == nil {
return nil, nil, nil
}
// this is an edge case where parentFinalState is provided, and its parent block exists
// in the queue but has not been marked as executed yet (OnBlockExecuted(parent) is not called),
// in this case, we will internally call OnBlockExecuted(parentBlockID, parentFinalState).
// there is no need to create the executable block again, since it's already created.
if executable.StartState == nil && parentFinalState != nil {
q.log.Warn().
Str("blockID", executable.ID().String()).
Uint64("height", executable.Block.Header.Height).
Hex("parentID", executable.Block.Header.ParentID[:]).
Msg("edge case: receiving block with no parent commitment, but its parent block actually has been executed")
executables, err := q.onBlockExecuted(executable.Block.Header.ParentID, *parentFinalState)
if err != nil {
return nil, nil, fmt.Errorf("receiving block %v with parent commitment %v, but parent block %v already exists with no commitment, fail to call mark parent as executed: %w",
executable.ID(), *parentFinalState, executable.Block.Header.ParentID, err)
}
// we already have this block, its collection must have been fetched, so we only return the
// executables from marking its parent as executed.
return nil, executables, nil
}
// this means the caller think it's parent has not been executed, but the queue's internal state
// shows the parent has been executed, then it's probably a race condition where the call to
// inform the parent block has been executed arrives earlier than this call, which is an edge case
// and we can simply ignore this call.
if executable.StartState != nil && parentFinalState == nil {
q.log.Warn().
Str("blockID", executable.ID().String()).
Uint64("height", executable.Block.Header.Height).
Hex("parentID", executable.Block.Header.ParentID[:]).
Msg("edge case: receiving block with no parent commitment, but its parent block actually has been executed")
return nil, nil, nil
}
// this is an exception that should not happen
if *executable.StartState != *parentFinalState {
return nil, nil,
fmt.Errorf("block %s has already been executed with a different parent final state, %v != %v",
executable.ID(), *executable.StartState, parentFinalState)
}
q.log.Warn().
Str("blockID", executable.ID().String()).
Uint64("height", executable.Block.Header.Height).
Msg("edge case: OnBlockExecuted is called with the same arguments again")
return nil, nil, nil
}
func (q *BlockQueue) onBlockExecuted(
blockID flow.Identifier,
commit flow.StateCommitment,
) ([]*entity.ExecutableBlock, error) {
// when a block is executed, the child block might become executable
// we also remove it from all the indexes
// remove block
block, ok := q.blocks[blockID]
if !ok {
return nil, nil
}
// sanity check
// if a block exists in the queue and is executed, then its parent block
// must not exist in the queue, otherwise the state is inconsistent
_, parentExists := q.blocks[block.Block.Header.ParentID]
if parentExists {
return nil, fmt.Errorf("parent block %s of block %s is in the queue",
block.Block.Header.ParentID, blockID)
}
delete(q.blocks, blockID)
// remove height index
height := block.Block.Header.Height
delete(q.blockIDsByHeight[height], blockID)
if len(q.blockIDsByHeight[height]) == 0 {
delete(q.blockIDsByHeight, height)
}
// remove colections if no other blocks include it
for colID := range block.CompleteCollections {
colInfo, ok := q.collections[colID]
if !ok {
return nil, fmt.Errorf("collection %s not found", colID)
}
delete(colInfo.IncludedIn, blockID)
if len(colInfo.IncludedIn) == 0 {
// no other blocks includes this collection,
// so this collection can be removed from the index
delete(q.collections, colID)
}
}
return q.checkIfChildBlockBecomeExecutable(block, commit)
}
func (q *BlockQueue) checkIfChildBlockBecomeExecutable(
block *entity.ExecutableBlock,
commit flow.StateCommitment,
) ([]*entity.ExecutableBlock, error) {
childHeight := block.Block.Header.Height + 1
blocksAtNextHeight, ok := q.blockIDsByHeight[childHeight]
if !ok {
// no block at next height
return nil, nil
}
// find children and update their start state
children := make([]*entity.ExecutableBlock, 0, len(blocksAtNextHeight))
for _, childBlock := range blocksAtNextHeight {
// a child block at the next height must have the same parent ID
// as the current block
isChild := childBlock.Block.Header.ParentID == block.ID()
if !isChild {
continue
}
// update child block's start state with current block's end state
childBlock.StartState = &commit
children = append(children, childBlock)
}
if len(children) == 0 {
return nil, nil
}
// check if children are executable
executables := make([]*entity.ExecutableBlock, 0, len(children))
for _, child := range children {
if child.IsComplete() {
executables = append(executables, child)
}
}
return executables, nil
}
// GetMissingCollections returns the missing collections and the start state for the given block
// Useful for debugging what is missing for the next unexecuted block to become executable.
// It returns an error if the block is not found
func (q *BlockQueue) GetMissingCollections(blockID flow.Identifier) (
[]*MissingCollection,
*flow.StateCommitment,
error,
) {
q.Lock()
defer q.Unlock()
block, ok := q.blocks[blockID]
if !ok {
return nil, nil, fmt.Errorf("block %s not found", blockID)
}
missingCollections := make([]*MissingCollection, 0, len(block.Block.Payload.Guarantees))
for _, col := range block.CompleteCollections {
// check if the collection is already received
if col.IsCompleted() {
continue
}
missingCollections = append(missingCollections, missingCollectionForBlock(block, col.Guarantee))
}
return missingCollections, block.StartState, nil
}
func missingCollectionForBlock(block *entity.ExecutableBlock, guarantee *flow.CollectionGuarantee) *MissingCollection {
return &MissingCollection{
BlockID: block.ID(),
Height: block.Block.Header.Height,
Guarantee: guarantee,
}
}