Fetch blocks concurrently but send in order #32

vegarsti · 2024-06-21T06:42:05Z

This PR rearchitects the main loop such that we can run multiple goroutines to fetch blocks concurrently, but send them in order to the API to ensure we don't get gaps. We produce block numbers to fetch on an unbuffered channel (ProduceBlockNumbers), and each concurrent ConsumeBlock goroutine gets a block number from that channel. The SendBlocks goroutine receives all blocks on an unbuffered channel, but buffers them in a map until they can be sent in order.

I deleted the test that asserted the exact number of blocks that were sent. I couldn't manage to get it working without off by one errors sometimes. Since the main use case is running it forever and not with a given blocks, I figured this was an OK decision. Let me know.

Possibly problematic scenario: If block N - and only block N - is somehow problematic from the RPC node, and must be retried ~forever, what will happen? MaxBatchSize - 1 goroutines continue to fetch blocks from the RPC node successfully, causing the buffer in SendBlocks to fill up. The mechanism in this PR doesn't have a way to backpressure there. Is that a problem? If memory footprint is a concern, we could possibly compress it here: At this point, currently, the block response is uncompressed. If we compress it first, it would take a lot more blocks for this to consume a lot of memory.

Otherwise, if there are general problems with the RPC node such that all requests take a long time or need to be retried, we'll get backpressured because we limit the concurrent number of calls to the node to MaxBatchSize.

FWIW I'm running this locally and getting 12 blocks per second with 5 workers!

vegarsti · 2024-06-21T06:42:21Z

Fetch blocks concurrently but send in order #32 👈
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @vegarsti and the rest of your teammates on Graphite

ingester/ingester.go

msf · 2024-06-21T08:12:13Z

ingester/mainloop.go

+
+			// Send this block only if we have sent all previous blocks
+			for block, ok := blockMap[next]; ok; block, ok = blockMap[next] {
+				i.log.Info("SendBlocks: Sending block to DuneAPI", "blockNumber", block.BlockNumber)


this should be an isolated function:
next = trySendCompletedBlocks(blockMap, next, ...) ?

yeah good call!

msf · 2024-06-21T08:12:53Z

ingester/mainloop.go

+func (i *ingester) SendBlocks(ctx context.Context, blocksCh <-chan models.RPCBlock, startBlockNumber int64) error {
+	i.log.Info("SendBlocks: Starting to receive blocks")
+	blockMap := make(map[int64]models.RPCBlock) // Buffer for temporarily storing blocks that have arrived out of order
+	next := startBlockNumber


nextNumberToSend ?

yes, agreed

msf · 2024-06-21T08:13:37Z

ingester/mainloop.go

+			i.log.Info("Received block", "blockNumber", block.BlockNumber)
+
+			// Send this block only if we have sent all previous blocks
+			for block, ok := blockMap[next]; ok; block, ok = blockMap[next] {


this is smart for loop :-)

ChatGPT showed me this 😄

msf

this is really cool, but we definitely need TESTS to prove it is correct.

msf · 2024-06-21T08:16:11Z

ingester/mainloop_test.go

+	// logOutput := os.Stderr
+	logOutput := io.Discard
+	ing := ingester.New(slog.New(slog.NewTextHandler(logOutput, nil)), rpcClient, duneapi, ingester.Config{
+		MaxBatchSize: 1,


we need a test with batchSize > 2 and where we assert that on duneclient (mock) we never receive blocks out of order...
(and the mocked rpcClient takes some random microsecond times on BlockByNumber)

oh, good catch that batch size is only 1 here, oops

This test doesn't use the RPC client at all, though, we are sending blocks out of order to the SendBlock goroutine.

But you're probably asking for an integration test with that scenario, though! I'll add it

msf · 2024-06-21T09:11:53Z

ingester/mainloop_test.go

 	sentBlockNumber := int64(0)
 	producedBlockNumber := int64(0)
 	duneapi := &duneapi_mock.BlockchainIngesterMock{
 		SendBlockFunc: func(_ context.Context, block models.RPCBlock) error {
+			// DuneAPI must fail if it receives blocks out of order
+			if block.BlockNumber != sentBlockNumber+1 {
+				return errors.Errorf("blocks out of order")


you can simply require.Equal() to abort the test

msf

SHIP IT

@msf

Implements some useful suggestion from @msf #32 - `ConsumeBlocks` -> `FetchBlockLoop` - Extract `trySendCompletedBlocks` from `SendBlocks`

This PR changes the node indexer from sending one block at a time to sending a batch of blocks. Earlier we implemented concurrent block fetching with buffering (#32). On a configurable interval (defaults to every second), we now check the buffer and send all possible blocks. We add a flag for specifying whether you wish to skip a failed block or not. It's off by default. This means if all the retries to the RPC node fails for a given block, we will crash. This ensures no block gaps. --------- Co-authored-by: Miguel Filipe <[email protected]>

vegarsti force-pushed the concurrent-block-fetch branch 3 times, most recently from fb06900 to 0b18551 Compare June 21, 2024 06:53

vegarsti requested a review from msf June 21, 2024 06:53

vegarsti force-pushed the concurrent-block-fetch branch 3 times, most recently from 04ee907 to aa11815 Compare June 21, 2024 07:39

msf reviewed Jun 21, 2024

View reviewed changes

ingester/ingester.go Show resolved Hide resolved

msf reviewed Jun 21, 2024

View reviewed changes

Fetch blocks concurrently but send in order

b0975b0

vegarsti force-pushed the concurrent-block-fetch branch from aa11815 to b0975b0 Compare June 21, 2024 09:03

msf reviewed Jun 21, 2024

View reviewed changes

msf approved these changes Jun 21, 2024

View reviewed changes

vegarsti merged commit 34cc3e7 into main Jun 21, 2024
1 check passed

vegarsti deleted the concurrent-block-fetch branch June 21, 2024 09:43

vegarsti mentioned this pull request Jun 22, 2024

Some improvements in mainloop.go #34

Merged

vegarsti added a commit that referenced this pull request Jun 24, 2024

Some improvements in mainloop.go (#34)

d83f853

Implements some useful suggestion from @msf #32 - `ConsumeBlocks` -> `FetchBlockLoop` - Extract `trySendCompletedBlocks` from `SendBlocks`

This was referenced Jun 24, 2024

Send a batch of blocks #35

Closed

Send batch of blocks from the main loop #37

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fetch blocks concurrently but send in order #32

Fetch blocks concurrently but send in order #32

vegarsti commented Jun 21, 2024 •

edited

Loading

vegarsti commented Jun 21, 2024

msf Jun 21, 2024

vegarsti Jun 21, 2024

msf Jun 21, 2024

vegarsti Jun 21, 2024

msf Jun 21, 2024

vegarsti Jun 24, 2024

msf left a comment

msf Jun 21, 2024

vegarsti Jun 21, 2024

vegarsti Jun 21, 2024

msf Jun 21, 2024

msf left a comment

Fetch blocks concurrently but send in order #32

Fetch blocks concurrently but send in order #32

Conversation

vegarsti commented Jun 21, 2024 • edited Loading

vegarsti commented Jun 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

msf left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

msf left a comment

Choose a reason for hiding this comment

vegarsti commented Jun 21, 2024 •

edited

Loading