Ingester main loop

[msf]

This is the main loop logic, it connects the:

• rpc node client that is getting blocks
• dune api client, that is sending blocks to dune.

it has a few responsabilities:

• track errors on either side (and later do something about it)
• track if it is behind or up to date
• try to go fast.
• produce good logs of whats going on

[vegarsti]

Nice!!

[vegarsti]

Something like this?

	// Every N seconds: Send the batched blocks
	buffer := make([]*Block, 0, concurrency)
	ticker := time.NewTicker(batchInterval)
	for {
		select {
		case block := <-blocks:
			buffer = append(buffer, block)
			if len(buffer) == cap(buffer) {
				sendBlocks(buffer)
				buffer = buffer[:0] // Clear the buffer after sending.
			}
		case <-ticker.C:
			if len(buffer) > 0 {
				sendBlocks(buffer)
				buffer = buffer[:0] // Clear the buffer after sending.
			}
		}
	}

[msf]

exactly, but batchInterval should be like 150millis.. so we don't cause delays, its only for the case where we can harvest fast (or forces us to harvest fast?)

[vegarsti]

Running this in a separate goroutine is such a great idea!

[vegarsti]

For the main loop, when we start trying to do reliable delivery, I'm thinking we might want to use more of a pipeline with channels

• one channel to send block numbers on. 2 goroutines to send block numbers into this channel: one for incrementing the next block number, another for checking Dune API for missing block numbers. Or it reads from the DLQ and sends those blocks on the channel
• one channel to send blocks on. Many goroutines fetch blocks and send into this channel.
• one goroutine reads blocks, buffers, and sends to the API

What do you think? Wrote a prototype here: https://gist.github.com/vegarsti/bc4d9dceec87666f456f3fbf6b213edb

[helanto]

As far as I can tell, LGTM :)

[helanto]

In the future do we consider parallelizing this part ? I mean to consume blocks in parallel ?

[msf]

agreed, like Vegard proposed above