fix field derivation for sending new log events (#1275)

core/blockchain.go

@@ -545,14 +545,16 @@ func (bc *BlockChain) warmAcceptedCaches() {
 		startIndex = lastAccepted - cacheDiff
 	}
 	for i := startIndex; i <= lastAccepted; i++ {
-		header := bc.GetHeaderByNumber(i)
-		if header == nil {
+		block := bc.GetBlockByNumber(i)
+		if block == nil {
 			// This could happen if a node state-synced
 			log.Info("Exiting accepted cache warming early because header is nil", "height", i, "t", time.Since(startTime))
 			break
 		}
-		bc.hc.acceptedNumberCache.Put(header.Number.Uint64(), header)
-		bc.acceptedLogsCache.Put(header.Hash(), rawdb.ReadLogs(bc.db, header.Hash(), header.Number.Uint64()))
+		// TODO: handle blocks written to disk during state sync
+		bc.hc.acceptedNumberCache.Put(block.NumberU64(), block.Header())
+		logs := bc.collectUnflattenedLogs(block, false)
+		bc.acceptedLogsCache.Put(block.Hash(), logs)
 	}
 	log.Info("Warmed accepted caches", "start", startIndex, "end", lastAccepted, "t", time.Since(startTime))
 }
@@ -579,7 +581,7 @@ func (bc *BlockChain) startAcceptor() {
 
 		// Ensure [hc.acceptedNumberCache] and [acceptedLogsCache] have latest content
 		bc.hc.acceptedNumberCache.Put(next.NumberU64(), next.Header())
-		logs := rawdb.ReadLogs(bc.db, next.Hash(), next.NumberU64())
+		logs := bc.collectUnflattenedLogs(next, false)
 		bc.acceptedLogsCache.Put(next.Hash(), logs)
 
 		// Update accepted feeds
@@ -1387,25 +1389,34 @@ func (bc *BlockChain) insertBlock(block *types.Block, writes bool) error {
 	return nil
 }
 
-// collectLogs collects the logs that were generated or removed during
-// the processing of a block. These logs are later announced as deleted or reborn.
-func (bc *BlockChain) collectLogs(b *types.Block, removed bool) []*types.Log {
+// collectUnflattenedLogs collects the logs that were generated or removed during
+// the processing of a block.
+func (bc *BlockChain) collectUnflattenedLogs(b *types.Block, removed bool) [][]*types.Log {
 	receipts := rawdb.ReadRawReceipts(bc.db, b.Hash(), b.NumberU64())
 	receipts.DeriveFields(bc.chainConfig, b.Hash(), b.NumberU64(), b.Time(), b.BaseFee(), b.Transactions())
 
-	var logs []*types.Log
+	var logs [][]*types.Log
 	for _, receipt := range receipts {
-		for _, log := range receipt.Logs {
+		receiptLogs := make([]*types.Log, len(receipt.Logs))
+		for i, log := range receipt.Logs {
 			l := *log
 			if removed {
 				l.Removed = true
 			}
-			logs = append(logs, &l)
+			receiptLogs[i] = &l
 		}
+		logs = append(logs, receiptLogs)
 	}
 	return logs
 }
 
+// collectLogs collects the logs that were generated or removed during
+// the processing of a block. These logs are later announced as deleted or reborn.
+func (bc *BlockChain) collectLogs(b *types.Block, removed bool) []*types.Log {
+	unflattenedLogs := bc.collectUnflattenedLogs(b, removed)
+	return types.FlattenLogs(unflattenedLogs)
+}
+
 // reorg takes two blocks, an old chain and a new chain and will reconstruct the
 // blocks and inserts them to be part of the new canonical chain and accumulates
 // potential missing transactions and post an event about them.

core/blockchain_log_test.go

@@ -0,0 +1,98 @@
+// (c) 2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package core
+
+import (
+	"math/big"
+	"strings"
+	"testing"
+
+	"github.com/ava-labs/coreth/accounts/abi"
+	"github.com/ava-labs/coreth/consensus/dummy"
+	"github.com/ava-labs/coreth/core/rawdb"
+	"github.com/ava-labs/coreth/core/types"
+	"github.com/ava-labs/coreth/core/vm"
+	"github.com/ava-labs/coreth/params"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/stretchr/testify/require"
+)
+
+/*
+Example contract to test event emission:
+
+	pragma solidity >=0.7.0 <0.9.0;
+	contract Callable {
+		event Called();
+		function Call() public { emit Called(); }
+	}
+*/
+const callableAbi = "[{\"anonymous\":false,\"inputs\":[],\"name\":\"Called\",\"type\":\"event\"},{\"inputs\":[],\"name\":\"Call\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"}]"
+
+const callableBin = "6080604052348015600f57600080fd5b5060998061001e6000396000f3fe6080604052348015600f57600080fd5b506004361060285760003560e01c806334e2292114602d575b600080fd5b60336035565b005b7f81fab7a4a0aa961db47eefc81f143a5220e8c8495260dd65b1356f1d19d3c7b860405160405180910390a156fea2646970667358221220029436d24f3ac598ceca41d4d712e13ced6d70727f4cdc580667de66d2f51d8b64736f6c63430008010033"
+
+func TestEmitLogsCorrectness(t *testing.T) {
+	var (
+		require = require.New(t)
+		engine  = dummy.NewFaker()
+		key1, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+		addr1   = crypto.PubkeyToAddress(key1.PublicKey)
+		funds   = new(big.Int).Mul(big.NewInt(100), big.NewInt(params.Ether))
+		gspec   = &Genesis{
+			Config:  params.TestChainConfig,
+			Alloc:   GenesisAlloc{addr1: {Balance: funds}},
+			BaseFee: big.NewInt(params.ApricotPhase3InitialBaseFee),
+		}
+		contractAddress = crypto.CreateAddress(addr1, 0)
+		signer          = types.LatestSigner(gspec.Config)
+	)
+
+	parsed, err := abi.JSON(strings.NewReader(callableAbi))
+	require.NoError(err)
+
+	packedFunction, err := parsed.Pack("Call")
+	require.NoError(err)
+
+	_, blocks, _, err := GenerateChainWithGenesis(gspec, engine, 2, 10, func(i int, b *BlockGen) {
+		switch i {
+		case 0:
+			// First, we deploy the contract
+			contractTx := types.NewContractCreation(0, common.Big0, 200000, big.NewInt(params.ApricotPhase3InitialBaseFee), common.FromHex(callableBin))
+			contractSignedTx, err := types.SignTx(contractTx, signer, key1)
+			require.NoError(err)
+			b.AddTx(contractSignedTx)
+		case 1:
+			// In the next block, we call the contract function
+			tx := types.NewTransaction(1, contractAddress, common.Big0, 23000, big.NewInt(params.ApricotPhase3InitialBaseFee), packedFunction)
+			tx, err := types.SignTx(tx, signer, key1)
+			require.NoError(err)
+			b.AddTx(tx)
+		}
+	})
+	require.NoError(err)
+
+	chain, err := NewBlockChain(rawdb.NewMemoryDatabase(), DefaultCacheConfig, gspec, engine, vm.Config{}, common.Hash{}, false)
+	require.NoError(err)
+
+	// Create Log Subscriber
+	logsCh := make(chan []*types.Log, 10)
+	defer close(logsCh)
+
+	sub := chain.SubscribeAcceptedLogsEvent(logsCh)
+	defer sub.Unsubscribe()
+
+	_, err = chain.InsertChain(blocks)
+	require.NoError(err)
+
+	for _, block := range blocks {
+		err := chain.Accept(block)
+		require.NoError(err)
+	}
+	chain.DrainAcceptorQueue()
+
+	logs := <-logsCh
+	require.Len(logs, 1)
+	require.Equal(blocks[1].Hash(), logs[0].BlockHash)
+	require.Equal(blocks[1].Number().Uint64(), logs[0].BlockNumber)
+}

core/blockchain_reader.go

@@ -346,5 +346,10 @@ func (bc *BlockChain) GetLogs(hash common.Hash, number uint64) [][]*types.Log {
 	if ok {
 		return logs
 	}
-	return rawdb.ReadLogs(bc.db, hash, number)
+	block := bc.GetBlockByHash(hash)
+	if block == nil {
+		return nil
+	}
+	logs = bc.collectUnflattenedLogs(block, false)
+	return logs
 }