Added first pass on integration tests

eth/tracers/firehose.go

@@ -61,17 +61,15 @@ func init() {
 }
 
 func newFirehoseTracer(cfg json.RawMessage) (*tracing.Hooks, error) {
-	var config FirehoseConfig
+	firehoseTracer, err := NewFirehoseFromRawJSON(cfg)
-	if len([]byte(cfg)) > 0 {
+	if err != nil {
-		if err := json.Unmarshal(cfg, &config); err != nil {
+		return nil, err
-			return nil, fmt.Errorf("failed to parse Firehose config: %w", err)
-		}
 	}
 
-	return newTracingHooksFromFirehose(NewFirehose(&config)), nil
+	return NewTracingHooksFromFirehose(firehoseTracer), nil
 }
 
-func newTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
+func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
 	return &tracing.Hooks{
 		OnBlockchainInit: tracer.OnBlockchainInit,
 		OnGenesisBlock:   tracer.OnGenesisBlock,
@@ -108,6 +106,14 @@ func newTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
 
 type FirehoseConfig struct {
 	ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"`
+
+	// Only used for testing, only possible through JSON configuration
+	private *privateFirehoseConfig
+}
+
+type privateFirehoseConfig struct {
+	FlushToTestBuffer  bool `json:"flushToTestBuffer"`
+	IgnoreGenesisBlock bool `json:"ignoreGenesisBlock"`
 }
 
 // LogKeValues returns a list of key-values to be logged when the config is printed.
@@ -163,14 +169,41 @@ type Firehose struct {
 	callStack               *CallStack
 	deferredCallState       *DeferredCallState
 	latestCallEnterSuicided bool
+
+	// Testing state, only used in tests and private configs
+	testingBuffer             *bytes.Buffer
+	testingIgnoreGenesisBlock bool
 }
 
 const FirehoseProtocolVersion = "3.0"
 
+func NewFirehoseFromRawJSON(cfg json.RawMessage) (*Firehose, error) {
+	var config FirehoseConfig
+	if len([]byte(cfg)) > 0 {
+		if err := json.Unmarshal(cfg, &config); err != nil {
+			return nil, fmt.Errorf("failed to parse Firehose config: %w", err)
+		}
+
+		// Special handling of some "private" fields
+		type privateConfigRoot struct {
+			Private *privateFirehoseConfig `json:"_private"`
+		}
+
+		var privateConfig privateConfigRoot
+		if err := json.Unmarshal(cfg, &privateConfig); err != nil {
+			log.Info("Firehose failed to parse private config, ignoring", "error", err)
+		} else {
+			config.private = privateConfig.Private
+		}
+	}
+
+	return NewFirehose(&config), nil
+}
+
 func NewFirehose(config *FirehoseConfig) *Firehose {
 	log.Info("Firehose tracer created", config.LogKeyValues()...)
 
-	return &Firehose{
+	firehose := &Firehose{
 		// Global state
 		outputBuffer:               bytes.NewBuffer(make([]byte, 0, 100*1024*1024)),
 		initSent:                   new(atomic.Bool),
@@ -192,6 +225,15 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		deferredCallState:       NewDeferredCallState(),
 		latestCallEnterSuicided: false,
 	}
+
+	if config.private != nil {
+		firehose.testingIgnoreGenesisBlock = config.private.IgnoreGenesisBlock
+		if config.private.FlushToTestBuffer {
+			firehose.testingBuffer = bytes.NewBuffer(nil)
+		}
+	}
+
+	return firehose
 }
 
 func (f *Firehose) newIsolatedTransactionTracer(tracerID string) *Firehose {
@@ -254,7 +296,7 @@ func (f *Firehose) OnBlockchainInit(chainConfig *params.ChainConfig) {
 	f.chainConfig = chainConfig
 
 	if wasNeverSent := f.initSent.CompareAndSwap(false, true); wasNeverSent {
-		printToFirehose("INIT", FirehoseProtocolVersion, "geth", params.Version)
+		f.printToFirehose("INIT", FirehoseProtocolVersion, "geth", params.Version)
 	} else {
 		f.panicInvalidState("The OnBlockchainInit callback was called more than once", 0)
 	}
@@ -530,7 +572,7 @@ func (f *Firehose) onTxStart(tx *types.Transaction, hash common.Hash, from, to c
 }
 
 func (f *Firehose) OnTxEnd(receipt *types.Receipt, err error) {
-	firehoseInfo("trx ending (tracer=%s)", f.tracerID)
+	firehoseInfo("trx ending (tracer=%s, isolated=%t, error=%s)", f.tracerID, f.transactionIsolated, errorView(err))
 	f.ensureInBlockAndInTrx()
 
 	trxTrace := f.completeTransaction(receipt)
@@ -1005,6 +1047,12 @@ func computeCallSource(depth int) string {
 }
 
 func (f *Firehose) OnGenesisBlock(b *types.Block, alloc types.GenesisAlloc) {
+	firehoseInfo("genesis block (number=%d hash=%s)", b.NumberU64(), b.Hash())
+	if f.testingIgnoreGenesisBlock {
+		firehoseInfo("genesis block ignored due to testing config")
+		return
+	}
+
 	f.ensureBlockChainInit()
 
 	f.OnBlockStart(tracing.BlockEvent{Block: b, TD: big.NewInt(0), Finalized: nil, Safe: nil})
@@ -1386,11 +1434,7 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string {
 	panic(fmt.Errorf("%s (caller=%s, init=%t, inBlock=%t, inTransaction=%t, inCall=%t)", msg, caller, f.chainConfig != nil, f.block != nil, f.transaction != nil, f.callStack.HasActiveCall()))
 }
 
-// printToFirehose is an easy way to print to Firehose format, it essentially
+// printBlockToFirehose is a helper function to print a block to Firehose protocl format.
-// adds the "FIRE" prefix to the input and joins the input with spaces as well
-// as adding a newline at the end.
-//
-// It flushes this through [flushToFirehose] to the `os.Stdout` writer.
 func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) {
 	marshalled, err := proto.Marshal(block)
 	if err != nil {
@@ -1430,7 +1474,7 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
 
 	f.outputBuffer.WriteString("\n")
 
-	flushToFirehose(f.outputBuffer.Bytes(), os.Stdout)
+	f.flushToFirehose(f.outputBuffer.Bytes())
 }
 
 // printToFirehose is an easy way to print to Firehose format, it essentially
@@ -1438,8 +1482,8 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
 // as adding a newline at the end.
 //
 // It flushes this through [flushToFirehose] to the `os.Stdout` writer.
-func printToFirehose(input ...string) {
+func (f *Firehose) printToFirehose(input ...string) {
-	flushToFirehose([]byte("FIRE "+strings.Join(input, " ")+"\n"), os.Stdout)
+	f.flushToFirehose([]byte("FIRE " + strings.Join(input, " ") + "\n"))
 }
 
 // flushToFirehose sends data to Firehose via `io.Writter` checking for errors
@@ -1448,7 +1492,12 @@ func printToFirehose(input ...string) {
 // If error is still present after 10 retries, prints an error message to `writer`
 // as well as writing file `/tmp/firehose_writer_failed_print.log` with the same
 // error message.
-func flushToFirehose(in []byte, writer io.Writer) {
+func (f *Firehose) flushToFirehose(in []byte) {
+	var writer io.Writer = os.Stdout
+	if f.testingBuffer != nil {
+		writer = f.testingBuffer
+	}
+
 	var written int
 	var err error
 	loops := 10
@@ -1470,6 +1519,14 @@ func flushToFirehose(in []byte, writer io.Writer) {
 	fmt.Fprint(writer, errstr)
 }
 
+// TestingBuffer is an internal method only used for testing purposes
+// that should never be used in production code.
+//
+// There is no public api guaranteed for this method.
+func (f *Firehose) InternalTestingBuffer() *bytes.Buffer {
+	return f.testingBuffer
+}
+
 // FIXME: Create a unit test that is going to fail as soon as any header is added in
 func newBlockHeaderFromChainHeader(h *types.Header, td *pbeth.BigInt) *pbeth.BlockHeader {
 	var withdrawalsHashBytes []byte

eth/tracers/tests/firehose/integration_test.go

@@ -0,0 +1,288 @@
+package firehose_test
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"hash"
+	"math/big"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/tracing"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/core/vm"
+	"github.com/ethereum/go-ethereum/eth/tracers"
+	"github.com/ethereum/go-ethereum/params"
+	pbeth "github.com/ethereum/go-ethereum/pb/sf/ethereum/type/v2"
+	"github.com/ethereum/go-ethereum/trie"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/crypto/sha3"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestFirehoseIntegrationTest(t *testing.T) {
+	context := vm.BlockContext{
+		CanTransfer: core.CanTransfer,
+		Transfer:    core.Transfer,
+		Coinbase:    common.Address{},
+		BlockNumber: new(big.Int).SetUint64(uint64(1)),
+		Time:        1,
+		Difficulty:  big.NewInt(2),
+		GasLimit:    uint64(1000000),
+		BaseFee:     big.NewInt(8),
+	}
+
+	tracer, err := tracers.NewFirehoseFromRawJSON(json.RawMessage(`{
+		"applyBackwardsCompatibility": true,
+		"_private": {
+			"flushToTestBuffer": true,
+			"ignoreGenesisBlock": true
+		}
+	}`))
+	if err != nil {
+		t.Fatal(fmt.Errorf("failed to create firehose tracer: %w", err))
+	}
+	hooks := tracers.NewTracingHooksFromFirehose(tracer)
+
+	genesis, blockchain := newCanonical(t, types.GenesisAlloc{}, context, hooks)
+
+	block := types.NewBlock(&types.Header{
+		ParentHash:       genesis.ToBlock().Hash(),
+		Number:           context.BlockNumber,
+		Difficulty:       context.Difficulty,
+		Coinbase:         context.Coinbase,
+		Time:             context.Time,
+		GasLimit:         context.GasLimit,
+		BaseFee:          context.BaseFee,
+		ParentBeaconRoot: ptr(common.Hash{}),
+	}, nil, nil, nil, trie.NewStackTrie(nil))
+
+	blockchain.SetBlockValidatorAndProcessorForTesting(
+		ignoreValidateStateValidator{core.NewBlockValidator(genesis.Config, blockchain, blockchain.Engine())},
+		core.NewStateProcessor(genesis.Config, blockchain, blockchain.Engine()),
+	)
+
+	n, err := blockchain.InsertChain(types.Blocks{block})
+	require.NoError(t, err)
+	require.Equal(t, 1, n)
+
+	genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer)
+	require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join(slicesMap(unknownLines, func(l unknwonLine) string { return "- '" + string(l) + "'" }), "\n"))
+	require.NotNil(t, genesisLine)
+	blockLines.assertEquals(t,
+		firehoseBlockLineParams{"1", "8e6ee4b1054d94df1d8a51fb983447dc2e27a854590c3ac0061f994284be8150", "0", "845bad515694a416bab4b8d44e22cf97a8c894a8502110ab807883940e185ce0", "0", "1000000000"},
+	)
+}
+
+type firehoseInitLine struct {
+	ProtocolVersion string
+	NodeName        string
+	NodeVersion     string
+}
+
+type firehoseBlockLines []firehoseBlockLine
+
+func (lines firehoseBlockLines) assertEquals(t *testing.T, expected ...firehoseBlockLineParams) {
+	actualParams := slicesMap(lines, func(l firehoseBlockLine) firehoseBlockLineParams { return l.Params })
+	require.Equal(t, expected, actualParams, "Actual lines block params do not match expected lines block params")
+
+	goldenUpdate := os.Getenv("GOLDEN_UPDATE") == "true"
+
+	for _, line := range lines {
+		goldenPath := fmt.Sprintf("testdata/%s/block.%d.golden.json", t.Name(), line.Block.Header.Number)
+		if !goldenUpdate && !fileExists(t, goldenPath) {
+			t.Fatalf("the golden file %q does not exist, re-run with 'GOLDEN_UPDATE=true go test ./... -run %q' to generate the intial version", goldenPath, t.Name())
+		}
+
+		content, err := protojson.MarshalOptions{Indent: "  "}.Marshal(line.Block)
+		require.NoError(t, err)
+
+		if goldenUpdate {
+			require.NoError(t, os.MkdirAll(filepath.Dir(goldenPath), 0755))
+			require.NoError(t, os.WriteFile(goldenPath, content, 0644))
+		}
+
+		expected, err := os.ReadFile(goldenPath)
+		require.NoError(t, err)
+
+		expectedBlock := &pbeth.Block{}
+		require.NoError(t, protojson.Unmarshal(expected, expectedBlock))
+
+		if !proto.Equal(expectedBlock, line.Block) {
+			assert.EqualExportedValues(t, expectedBlock, line.Block, "Run 'GOLDEN_UPDATE=true go test ./... -run %q' to update golden file", t.Name())
+		}
+	}
+}
+
+func fileExists(t *testing.T, path string) bool {
+	t.Helper()
+	stat, err := os.Stat(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return false
+		}
+
+		t.Fatal(err)
+	}
+
+	return !stat.IsDir()
+}
+
+func slicesMap[T any, U any](s []T, f func(T) U) []U {
+	result := make([]U, len(s))
+	for i, v := range s {
+		result[i] = f(v)
+	}
+	return result
+}
+
+type firehoseBlockLine struct {
+	// We split params and block to make it easier to compare stuff
+	Params firehoseBlockLineParams
+	Block  *pbeth.Block
+}
+
+type firehoseBlockLineParams struct {
+	Number       string
+	Hash         string
+	PreviousNum  string
+	PreviousHash string
+	LibNum       string
+	Time         string
+}
+
+type unknwonLine string
+
+func readTracerFirehoseLines(t *testing.T, tracer *tracers.Firehose) (genesisLine *firehoseInitLine, blockLines firehoseBlockLines, unknownLines []unknwonLine) {
+	t.Helper()
+
+	lines := bytes.Split(tracer.InternalTestingBuffer().Bytes(), []byte{'\n'})
+	for _, line := range lines {
+		if len(line) == 0 {
+			continue
+		}
+
+		parts := bytes.Split(line, []byte{' '})
+		if len(parts) == 0 || string(parts[0]) != "FIRE" {
+			unknownLines = append(unknownLines, unknwonLine(line))
+			continue
+		}
+
+		action := string(parts[1])
+		fireParts := parts[2:]
+		switch action {
+		case "INIT":
+			genesisLine = &firehoseInitLine{
+				ProtocolVersion: string(fireParts[0]),
+				NodeName:        string(fireParts[1]),
+				NodeVersion:     string(fireParts[2]),
+			}
+
+		case "BLOCK":
+			protoBytes, err := base64.StdEncoding.DecodeString(string(fireParts[6]))
+			require.NoError(t, err)
+
+			block := &pbeth.Block{}
+			require.NoError(t, proto.Unmarshal(protoBytes, block))
+
+			blockLines = append(blockLines, firehoseBlockLine{
+				Params: firehoseBlockLineParams{
+					Number:       string(fireParts[0]),
+					Hash:         string(fireParts[1]),
+					PreviousNum:  string(fireParts[2]),
+					PreviousHash: string(fireParts[3]),
+					LibNum:       string(fireParts[4]),
+					Time:         string(fireParts[5]),
+				},
+				Block: block,
+			})
+
+		default:
+			unknownLines = append(unknownLines, unknwonLine(line))
+		}
+	}
+
+	return
+}
+
+func newCanonical(t *testing.T, alloc types.GenesisAlloc, context vm.BlockContext, tracer *tracing.Hooks) (*core.Genesis, *core.BlockChain) {
+	t.Helper()
+
+	var (
+		engine  = ethash.NewFullFaker()
+		genesis = &core.Genesis{
+			Difficulty: new(big.Int).Sub(context.Difficulty, big.NewInt(1)),
+			Timestamp:  context.Time - 1,
+			Number:     new(big.Int).Sub(context.BlockNumber, big.NewInt(1)).Uint64(),
+			BaseFee:    big.NewInt(params.InitialBaseFee),
+			Coinbase:   context.Coinbase,
+			Config:     params.AllEthashProtocolChanges,
+			Alloc:      alloc,
+		}
+	)
+	// Initialize a fresh chain with only a genesis block
+	blockchain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), core.DefaultCacheConfigWithScheme(rawdb.HashScheme), genesis, nil, engine, vm.Config{
+		Tracer: tracer,
+	}, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return genesis, blockchain
+}
+
+// testHasher is the helper tool for transaction/receipt list hashing.
+// The original hasher is trie, in order to get rid of import cycle,
+// use the testing hasher instead.
+type testHasher struct {
+	hasher hash.Hash
+}
+
+// NewHasher returns a new testHasher instance.
+func NewHasher() *testHasher {
+	return &testHasher{hasher: sha3.NewLegacyKeccak256()}
+}
+
+// Reset resets the hash state.
+func (h *testHasher) Reset() {
+	h.hasher.Reset()
+}
+
+// Update updates the hash state with the given key and value.
+func (h *testHasher) Update(key, val []byte) error {
+	h.hasher.Write(key)
+	h.hasher.Write(val)
+	return nil
+}
+
+// Hash returns the hash value.
+func (h *testHasher) Hash() common.Hash {
+	return common.BytesToHash(h.hasher.Sum(nil))
+}
+
+type ignoreValidateStateValidator struct {
+	core.Validator
+}
+
+func (v ignoreValidateStateValidator) ValidateBody(block *types.Block) error {
+	return v.Validator.ValidateBody(block)
+}
+
+func (v ignoreValidateStateValidator) ValidateState(block *types.Block, statedb *state.StateDB, receipts types.Receipts, usedGas uint64) error {
+	return nil
+}
+
+func ptr[T any](v T) *T {
+	return &v
+}

eth/tracers/tests/firehose/testdata/TestFirehoseIntegrationTest/block.1.golden.json

@@ -0,0 +1,64 @@
+{
+  "hash": "jm7ksQVNlN8dilH7mDRH3C4nqFRZDDrABh+ZQoS+gVA=",
+  "number": "1",
+  "size": "541",
+  "header": {
+    "parentHash": "hFutUVaUpBa6tLjUTiLPl6jIlKhQIRCrgHiDlA4YXOA=",
+    "uncleHash": "HcxN6N7HXXqrhbVntszUGtMSRRuUinQT8KFC/UDUk0c=",
+    "coinbase": "AAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    "stateRoot": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    "transactionsRoot": "VugfFxvMVab/g0XmksD4bltI4BuZbK3AAWIvteNjtCE=",
+    "receiptRoot": "VugfFxvMVab/g0XmksD4bltI4BuZbK3AAWIvteNjtCE=",
+    "logsBloom": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+    "difficulty": {
+      "bytes": "Ag=="
+    },
+    "totalDifficulty": {
+      "bytes": "Aw=="
+    },
+    "number": "1",
+    "gasLimit": "1000000",
+    "timestamp": "1970-01-01T00:00:01Z",
+    "mixHash": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    "hash": "jm7ksQVNlN8dilH7mDRH3C4nqFRZDDrABh+ZQoS+gVA=",
+    "baseFeePerGas": {
+      "bytes": "CA=="
+    },
+    "parentBeaconRoot": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+  },
+  "balanceChanges": [
+    {
+      "address": "AAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+      "newValue": {
+        "bytes": "G8FtZ07IAAA="
+      },
+      "reason": "REASON_REWARD_MINE_BLOCK",
+      "ordinal": "5"
+    }
+  ],
+  "systemCalls": [
+    {
+      "index": 1,
+      "callType": "CALL",
+      "caller": "//////////////////////////4=",
+      "address": "AA899tcygH7xMZ+3uLuFItC+rAI=",
+      "gasLimit": "30000000",
+      "input": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+      "gasChanges": [
+        {
+          "newValue": "30000000",
+          "reason": "REASON_CALL_INITIAL_BALANCE",
+          "ordinal": "2"
+        },
+        {
+          "oldValue": "30000000",
+          "reason": "REASON_CALL_LEFT_OVER_RETURNED",
+          "ordinal": "3"
+        }
+      ],
+      "beginOrdinal": "1",
+      "endOrdinal": "4"
+    }
+  ],
+  "ver": 4
+}