Send batch of blocks from the main loop #37

vegarsti · 2024-06-25T12:31:54Z

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.

vegarsti · 2024-06-25T12:32:11Z

Send batch of blocks from the main loop #37 👈
Dune client: SendBlock -> SendBlocks #36
main

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

Join @vegarsti and the rest of your teammates on Graphite

msf

this code is super clean & nice.
I have nitpicks on the log lines and how to name the env/config variable for defining the "block submit interval".
I also think we need to have a test that shows we sending a batched request instead of a single block..

config/config.go

ingester/mainloop.go

ingester/mainloop_test.go

vegarsti · 2024-06-27T07:10:52Z

client/duneapi/client.go

+			if resp == nil {
+				log.Warn("Retrying request", "error", err)


I saw a panic once on this, so safe guarding against that

vegarsti · 2024-06-27T07:37:46Z

ingester/mainloop.go

+				if !i.cfg.SkipFailedBlocks {
+					return err
+				}
+				// We need to send an empty block downstream to indicate that this failed
+				blocks <- models.RPCBlock{BlockNumber: blockNumber}
 				continue


I realized we need to handle the case of failing to get a block from the RPC node. If we just proceed without sending on the channel we will never proceed, since the SendBlocks channel relies on the block numbers. So we need to pass it downstream

okay, we need to change the channels to be "blockRPCResp" and that can be RPCBlock OR an error + blockNumber

vegarsti · 2024-06-27T07:39:57Z

ingester/mainloop.go

+			if block.Empty() {
+				// We got an empty block from the RPC client goroutine, either fail or send an empty block downstream
+				if !i.cfg.SkipFailedBlocks {
+					i.log.Info("Received empty block, exiting", "number", block.BlockNumber)
+					return errors.Errorf("empty block received")
+				}
+				i.log.Info("Received empty block", "number", block.BlockNumber)
+			}

-			nextNumberToSend = i.trySendCompletedBlocks(ctx, blocks, nextNumberToSend)
-			i.log.Info("SendBlocks: Sent any completed blocks to DuneAPI", "nextNumberToSend", nextNumberToSend)
+			collectedBlocks[block.BlockNumber] = block


I realized we need to handle the case of failing to get a block from the RPC node. If we just proceed without sending on the channel we will never proceed, since the SendBlocks channel relies on the block numbers. So we need to pass it downstream

wait, why don't we have:

if nextNumberToSend == block.Number && block.Empty() { log.WARN... if !skip { return .. } nextNumberToSend +=1 // skip it } .. normal logic here, the trySendBlocks will skip that block..

vegarsti · 2024-06-27T07:40:46Z

ingester/mainloop.go

+			// Skip block if it's empty and we're configured to skip empty blocks
+			if i.cfg.SkipFailedBlocks && block.Empty() {
+				nextNumberToSend++
+				continue
+			}


Only here do we properly skip the empty block

msf

approving with comments. :-)

msf · 2024-06-27T09:23:38Z

ingester/mainloop.go

+				if !i.cfg.SkipFailedBlocks {
+					return err
+				}
+				// We need to send an empty block downstream to indicate that this failed
+				blocks <- models.RPCBlock{BlockNumber: blockNumber}
 				continue


okay, we need to change the channels to be "blockRPCResp" and that can be RPCBlock OR an error + blockNumber

msf · 2024-06-27T09:24:40Z

ingester/mainloop.go

 		select {
 		case <-ctx.Done():
 			i.log.Info("SendBlocks: Context canceled, stopping")
 			return ctx.Err()
-		case block, ok := <-blocksCh:
+		case block, ok := <-blocks:


liked the old name more :)

msf · 2024-06-27T09:25:13Z

ingester/mainloop.go

+			if block.Empty() {
+				// We got an empty block from the RPC client goroutine, either fail or send an empty block downstream
+				if !i.cfg.SkipFailedBlocks {
+					i.log.Info("Received empty block, exiting", "number", block.BlockNumber)


duplicate log line, move it above the if

msf · 2024-06-27T09:28:06Z

ingester/mainloop.go

+			if block.Empty() {
+				// We got an empty block from the RPC client goroutine, either fail or send an empty block downstream
+				if !i.cfg.SkipFailedBlocks {
+					i.log.Info("Received empty block, exiting", "number", block.BlockNumber)
+					return errors.Errorf("empty block received")
+				}
+				i.log.Info("Received empty block", "number", block.BlockNumber)
+			}

-			nextNumberToSend = i.trySendCompletedBlocks(ctx, blocks, nextNumberToSend)
-			i.log.Info("SendBlocks: Sent any completed blocks to DuneAPI", "nextNumberToSend", nextNumberToSend)
+			collectedBlocks[block.BlockNumber] = block


wait, why don't we have:

if nextNumberToSend == block.Number && block.Empty() { log.WARN... if !skip { return .. } nextNumberToSend +=1 // skip it } .. normal logic here, the trySendBlocks will skip that block..

msf · 2024-06-27T09:32:30Z

ingester/mainloop.go

+	// Outer loop: We might need to send multiple batch requests if our buffer is too big
+	for _, ok := collectedBlocks[nextNumberToSend]; ok; _, ok = collectedBlocks[nextNumberToSend] {
+		// Collect a blocks of blocks to send, only send those which are in order
+		blocks := make([]models.RPCBlock, 0, len(collectedBlocks))


Suggested change

blocks := make([]models.RPCBlock, 0, len(collectedBlocks))

blocks := make([]models.RPCBlock, 0, maxBatchSize)

msf · 2024-06-27T10:10:51Z

ingester/mainloop_test.go

+			for _, block := range blocks {
+				// We cannot send blocks out of order to DuneAPI
+				if block.BlockNumber != next {
+					panic(fmt.Sprintf("expected block %d, got %d", next, block.BlockNumber))


use require.Equal(t, blockNumber, next)

msf · 2024-06-27T10:12:23Z

ingester/mainloop_test.go


-			// Test must fail if DuneAPI receives blocks out of order
-			require.Equal(t, block.BlockNumber, sentBlockNumber+1)
+			// Fail if we're not sending a batch of blocks


this is function screaming to get out.
(aka, this helper function is useful in many tests where you wanna validate what we pass to SendBlocks()

msf · 2024-06-27T10:12:57Z

ingester/mainloop_test.go

+	require.ErrorIs(t, err, someRPCError)
+}
+
+// TestRunRPCNodeFails shows that we crash if the RPC client fails to fetch a block


outdated comment

msf · 2024-06-27T10:13:57Z

ingester/mainloop_test.go

+	require.GreaterOrEqual(t, sentBlockNumber, maxBlockNumber)
+}
+
+// TestRunRPCNodeFails shows that we crash if the RPC client fails to fetch a block


this is showing we fail if DuneAPI.SendBlocks() fails right? so we shouldchange the name and comment?

msf · 2024-06-27T10:15:27Z

models/block.go

@@ -5,3 +5,7 @@ type RPCBlock struct {
 	// agnostic blob of data that is the block
 	Payload []byte


suggest we add a Error field and use Errored() bool instead of Empty()

This was referenced Jun 25, 2024

Send a batch of blocks #35

Closed

Dune client: SendBlock -> SendBlocks #36

Merged

vegarsti force-pushed the main-loop-batch-blocks branch 4 times, most recently from bfb9cb2 to d17408e Compare June 25, 2024 12:57

vegarsti requested a review from msf June 25, 2024 12:59

msf reviewed Jun 26, 2024

View reviewed changes

vegarsti force-pushed the dune-client-batch-blocks branch from 3c2104d to 75cb211 Compare June 27, 2024 05:45

Base automatically changed from dune-client-batch-blocks to main June 27, 2024 05:48

vegarsti force-pushed the main-loop-batch-blocks branch 6 times, most recently from 6399e35 to f0e0be9 Compare June 27, 2024 07:10

vegarsti commented Jun 27, 2024

View reviewed changes

vegarsti force-pushed the main-loop-batch-blocks branch 2 times, most recently from 2e022fa to 8d7e472 Compare June 27, 2024 07:13

vegarsti requested a review from msf June 27, 2024 07:14

vegarsti force-pushed the main-loop-batch-blocks branch 4 times, most recently from db35c5d to f00a5d4 Compare June 27, 2024 07:37

vegarsti commented Jun 27, 2024

View reviewed changes

vegarsti force-pushed the main-loop-batch-blocks branch from f00a5d4 to 11d3299 Compare June 27, 2024 07:39

vegarsti commented Jun 27, 2024

View reviewed changes

vegarsti force-pushed the main-loop-batch-blocks branch from 11d3299 to 3a457e0 Compare June 27, 2024 07:41

Send batch of blocks from the main loop

6730e76

vegarsti force-pushed the main-loop-batch-blocks branch from 3a457e0 to 6730e76 Compare June 27, 2024 09:21

msf approved these changes Jun 27, 2024

View reviewed changes

msf added 2 commits June 27, 2024 13:22

improve some tests

7ff628a

mainloop: don't handle duneapi errors yet, simplify around SkipFailed

98ac082

msf approved these changes Jun 27, 2024

View reviewed changes

allow for maxBatchSize blocks in flight

f2249f8

vegarsti merged commit a4ba018 into main Jun 27, 2024
1 check passed

vegarsti deleted the main-loop-batch-blocks branch June 27, 2024 12:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Send batch of blocks from the main loop #37

Send batch of blocks from the main loop #37

vegarsti commented Jun 25, 2024 •

edited

Loading

vegarsti commented Jun 25, 2024 •

edited

Loading

msf left a comment

vegarsti Jun 27, 2024

msf Jun 27, 2024

vegarsti Jun 27, 2024

msf Jun 27, 2024

vegarsti Jun 27, 2024

msf Jun 27, 2024

vegarsti Jun 27, 2024

msf left a comment

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

msf Jun 27, 2024

	blocks := make([]models.RPCBlock, 0, len(collectedBlocks))
	blocks := make([]models.RPCBlock, 0, maxBatchSize)

		@@ -5,3 +5,7 @@ type RPCBlock struct {
		// agnostic blob of data that is the block
		Payload []byte

Send batch of blocks from the main loop #37

Send batch of blocks from the main loop #37

Conversation

vegarsti commented Jun 25, 2024 • edited Loading

vegarsti commented Jun 25, 2024 • edited Loading

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

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

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

vegarsti commented Jun 25, 2024 •

edited

Loading

vegarsti commented Jun 25, 2024 •

edited

Loading