BUG: Unclear block start

eth/tracers/firehose.go

@@ -193,6 +193,7 @@ type Firehose struct {
 
 	// Worker queue for block printing
 	blockPrintQueue  chan *blockPrintJob
+	blockOutputQueue chan *blockOutput
 	flushDone        sync.WaitGroup
 	closeOnce        sync.Once
 }
@@ -261,11 +262,32 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		log.Info("Concurrent block flushing enabled: starting goroutine...")
 		const numWorkers = 10                                     // TODO: This becomes a parameter
 		firehose.blockPrintQueue = make(chan *blockPrintJob, 100) // TODO: Optimal buffer size tbd
+		firehose.blockOutputQueue = make(chan *blockOutput, 100)
 
 		for i := 0; i < numWorkers; i++ {
 			firehose.flushDone.Add(1)
 			go firehose.blockPrintWorker()
 		}
+
+		// Output channel to order the flushing linearly
+		go func() {
+			expected := uint64(1000) // TODO: We will need to determine at which block it starts
+			buffer := make(map[uint64][]byte)
+
+			for result := range firehose.blockOutputQueue {
+				buffer[result.blockNum] = result.data
+
+				for {
+					data, ok := buffer[expected]
+					if !ok {
+						break
+					}
+					firehose.flushToFirehose(data)
+					delete(buffer, expected)
+					expected++ // TODO: Assuming block numbers are linear
+				}
+			}
+		}()
 	}
 
 	return firehose
@@ -1859,13 +1881,13 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string {
 
 // printBlockToFirehose is a helper function to print a block to Firehose protocl format.
 func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) {
+	var buf bytes.Buffer
+
 	marshalled, err := proto.Marshal(block)
 	if err != nil {
 		panic(fmt.Errorf("failed to marshal block: %w", err))
 	}
 
-	var buf bytes.Buffer
-
 	previousHash := block.PreviousID()
 	previousNum := 0
 	if block.Number > 0 {
@@ -1897,8 +1919,15 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
 
 	buf.WriteString("\n")
 
+	if f.concurrentBlockFlushing {
+		f.blockOutputQueue <- &blockOutput{
+			blockNum: block.Number,
+			data:     buf.Bytes(),
+		}
+	} else {
 		f.flushToFirehose(buf.Bytes())
 	}
+}
 
 // printToFirehose is an easy way to print to Firehose format, it essentially
 // adds the "FIRE" prefix to the input and joins the input with spaces as well
@@ -2852,3 +2881,8 @@ func (f *Firehose) CloseBlockPrintQueue() {
 		})
 	}
 }
+
+type blockOutput struct {
+	blockNum uint64
+	data     []byte
+}

eth/tracers/firehose_concurrency_test.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) {
@@ -92,6 +93,8 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) {
 		f.OnBlockEnd(nil)
 	}
 
+	time.Sleep(5 * time.Second)
+
 	f.OnClose()
 
 	output := f.InternalTestingBuffer().String()