logs are forwarded to a processor in slot and trx order

[EasterTheBunny]

Description

Log ingestion heavily leverages async go-routines but LogPoller will rely on logs delivered in slot/block order. This commit adds a new component to the log ingestion to order logs before forwarding them to LogPoller.

[reductionista]

Not sure I understand this comment. Which function is it referring to that's preceded by cleanEmptyBlocks?

[EasterTheBunny]

I think this was just a comment from a previous iteration. I'll remove it.

[reductionista]

Could use slices.Delete:
https://pkg.go.dev/slices#Delete

[EasterTheBunny]

Good to know. I have not used the slices package yet.

[reductionista]

Could use slices.Index instead of defining are own:
https://pkg.go.dev/slices#Index

[EasterTheBunny]

Same here on the slices package.

[reductionista]

This is a perfect use case for a heap:
https://pkg.go.dev/container/heap#pkg-index

Instead of re-sorting the list each time a new block is added, you just call h.Push() to add each block and it inserts it efficiently into the tree. Then calling h.Pop() will remove them in order.

[cl-sonarqube-production]

Quality Gate passed

Issues
2 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
87.7% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube

[reductionista]

I don't think you can use slices.Delete with a heap, because it has to do do some fancy rebalancing while removing an element to preserve the heap invariant.

I think we just need to call:

heap.Remove(h, idx)

and then it does it all for you. No need to define a custom Delete method

[reductionista]

Update: you shouldn't need either this or heap.Remove, just heap.Pop--see other comments

[reductionista]

Iterating over them directly like this won't work right. The order is of the elements of a heap is well-defined, but it's not stored in the same order as the underlying elements of the backing slice (and I think some elements might be empty?)

To get the right order, you pop them one at a time. Something like this:

for p.blocks.Len() > 0 {
	block := heap.Pop(p.blocks)
	...
}

I left another comment somewhere else suggesting to replace the Delete method with a call to heap.Remove, but now that I'm looking at this part of the code I realize you shouldn't need either of those, I think Pop is all we need?

[reductionista]

Update: replace heap.Pop line above with block := (*p.blocks)[0], to delay the actual Pop for later when you are sure it can be removed.

[reductionista]

I see... you'll probably want to delay the heap.Pop call I have above till here, so that if you break or return early it's not already removed. Instead of the initial heap.Pop you can "peak" at the minimum block number on the heap with block := (*p.blocks)[0]. Even though the rest may be out of order, this should always be guaranteed to be the min element.

I think the two places you'll need the heap.Pop are in place of the 2 instances of rmvIdx = append(rmvIndex, idx), does that sound right?

[reductionista]

Suggested change

	rmvIdx = append(rmvIdx, idx)
	_ = heap.Pop(p.blocks)

[reductionista]

Almost there, I think this is coming together. Sorry for updating a couple of the comments after writing them hopefully it's clear what needs to be done. I think once you've got the iteration replaced with peaking and popping the heap it should simply things a bit.