Skip to content

Commit 3b16aff

Browse files
committed
Give blocks a chance to become sealed before calling NotifyBlock
1 parent 88e6d7e commit 3b16aff

1 file changed

Lines changed: 18 additions & 7 deletions

File tree

services/ingestion/block_tracking_subscriber.go

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,8 @@ func (r *RPCBlockTrackingSubscriber) subscribe(ctx context.Context, height uint6
166166
close(eventsChan)
167167
}()
168168

169+
blockHeadersQueue := []flow.BlockHeader{}
170+
169171
for ctx.Err() == nil {
170172
select {
171173
case <-ctx.Done():
@@ -219,13 +221,22 @@ func (r *RPCBlockTrackingSubscriber) subscribe(ctx context.Context, height uint6
219221
for _, evt := range blockEvents.Events {
220222
r.keyLock.NotifyTransaction(evt.TransactionID)
221223
}
222-
r.keyLock.NotifyBlock(
223-
flow.BlockHeader{
224-
ID: blockEvents.BlockID,
225-
Height: blockEvents.Height,
226-
},
227-
)
228-
lastReceivedHeight = blockHeader.Height
224+
225+
blockHeadersQueue = append(blockHeadersQueue, *blockHeader)
226+
227+
// The current `blockHeader` has a status of `BlockStatusFinalized`,
228+
// but calling `NotifyBlock` might fail if the AN has not actually
229+
// finished syncing all collections.
230+
// Hence, we keep a small queue of the incoming block headers, so
231+
// that we can call `NotifyBlock` on block N-5, where N is the
232+
// height of the current block header. This will give enough time
233+
// for the block to be sealed.
234+
if len(blockHeadersQueue) > 5 {
235+
earliestBlockHeader := blockHeadersQueue[0]
236+
r.keyLock.NotifyBlock(earliestBlockHeader)
237+
238+
blockHeadersQueue = blockHeadersQueue[1:]
239+
}
229240

230241
eventsChan <- evmEvents
231242

0 commit comments

Comments
 (0)