Mining: use noop writer when applyTx, pending logs subscription

cmd/integration/commands/stages.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ledgerwatch/turbo-geth/eth/stagedsync"
 	"github.com/ledgerwatch/turbo-geth/eth/stagedsync/stages"
 	"github.com/ledgerwatch/turbo-geth/ethdb"
+	"github.com/ledgerwatch/turbo-geth/ethdb/remote/remotedbserver"
 	"github.com/ledgerwatch/turbo-geth/log"
 	"github.com/ledgerwatch/turbo-geth/migrations"
 	"github.com/ledgerwatch/turbo-geth/params"
@@ -651,6 +652,7 @@ func newSync2(db ethdb.Database, tx ethdb.Database) (ethdb.StorageMode, *core.Ti
 
 	vmConfig := &vm.Config{NoReceipts: !sm.Receipts}
 	chainConfig := params.MainnetChainConfig
+	events := remotedbserver.NewEvents()
 
 	cc := &core.TinyChainContext{}
 	cc.SetDB(tx)
@@ -666,12 +668,12 @@ func newSync2(db ethdb.Database, tx ethdb.Database) (ethdb.StorageMode, *core.Ti
 	st := stagedsync.New(
 		stagedsync.DefaultStages(),
 		stagedsync.DefaultUnwindOrder(),
-		stagedsync.OptionalParameters{SilkwormExecutionFunc: silkwormExecutionFunc()},
+		stagedsync.OptionalParameters{SilkwormExecutionFunc: silkwormExecutionFunc(), Notifier: events},
 	)
 	stMining := stagedsync.New(
 		stagedsync.MiningStages(),
 		stagedsync.MiningUnwindOrder(),
-		stagedsync.OptionalParameters{SilkwormExecutionFunc: silkwormExecutionFunc()},
+		stagedsync.OptionalParameters{SilkwormExecutionFunc: silkwormExecutionFunc(), Notifier: events},
 	)
 	return sm, cc, chainConfig, vmConfig, nil, st, stMining, cache
 }

cmd/integration/commands/state_stages.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ledgerwatch/turbo-geth/common"
 	"github.com/ledgerwatch/turbo-geth/common/changeset"
 	"github.com/ledgerwatch/turbo-geth/common/dbutils"
+	"github.com/ledgerwatch/turbo-geth/common/debugprint"
 	"github.com/ledgerwatch/turbo-geth/common/etl"
 	"github.com/ledgerwatch/turbo-geth/core/rawdb"
 	"github.com/ledgerwatch/turbo-geth/core/state"
@@ -43,7 +44,6 @@ Examples:
 		`,
 	Example: "go run ./cmd/integration state_stages --datadir=... --verbosity=3 --unwind=100 --unwind.every=100000 --block=2000000",
 	RunE: func(cmd *cobra.Command, args []string) error {
-
 		ctx := utils.RootContext()
 		cfg := &node.DefaultConfig
 		utils.SetNodeConfigCobra(cmd, cfg)
@@ -63,13 +63,13 @@ Examples:
 		defer db.Close()
 		if err := syncBySmallSteps(db, miningConfig, ctx); err != nil {
 			log.Error("Error", "err", err)
-			return err
+			return nil
 		}
 
 		if referenceChaindata != "" {
 			if err := compareStates(ctx, chaindata, referenceChaindata); err != nil {
 				log.Error(err.Error())
-				return err
+				return nil
 			}
 
 		}
@@ -107,7 +107,7 @@ var loopExecCmd = &cobra.Command{
 		}
 		if err := loopExec(db, ctx, unwind); err != nil {
 			log.Error("Error", "err", err)
-			return err
+			return nil
 		}
 
 		return nil
@@ -260,7 +260,6 @@ func syncBySmallSteps(db ethdb.Database, miningConfig *params.MiningConfig, ctx
 			}
 			integrity.Trie(tx.(ethdb.HasTx).Tx(), integritySlow, quit)
 		}
-
 		//if err := tx.RollbackAndBegin(context.Background()); err != nil {
 		//	return err
 		//}
@@ -317,7 +316,7 @@ func syncBySmallSteps(db ethdb.Database, miningConfig *params.MiningConfig, ctx
 			if err := tx.RollbackAndBegin(context.Background()); err != nil {
 				return err
 			}
-			checkMinedBlock(nextBlock, minedBlock)
+			checkMinedBlock(nextBlock, minedBlock, chainConfig)
 		}
 
 		// Unwind all stages to `execStage - unwind` block
@@ -367,37 +366,21 @@ func miningTransactions(nextBlock *types.Block) (map[common.Address]types.Transa
 	return localTxs, nextBlock.Transactions()
 }
 
-func checkMinedBlock(b1, b2 *types.Block) {
+func checkMinedBlock(b1, b2 *types.Block, chainConfig *params.ChainConfig) {
 	h1 := b1.Header()
 	h2 := b2.Header()
 	if h1.Root != h2.Root ||
-		h1.ReceiptHash != h2.ReceiptHash ||
+		(chainConfig.IsByzantium(b1.Number()) && h1.ReceiptHash != h2.ReceiptHash) ||
 		h1.TxHash != h2.TxHash ||
 		h1.ParentHash != h2.ParentHash ||
 		h1.UncleHash != h2.UncleHash ||
 		h1.GasUsed != h2.GasUsed ||
 		!bytes.Equal(h1.Extra, h2.Extra) {
-		printBlocks(b1, b2)
+		debugprint.Headers(h1, h2)
 		panic("blocks are not same")
 	}
 }
 
-func printBlocks(b1, b2 *types.Block) {
-	h1 := b1.Header()
-	h2 := b2.Header()
-	fmt.Printf("==== Header ====\n")
-	fmt.Printf("root:        %x, %x\n", h1.Root, h2.Root)
-	fmt.Printf("nonce:       %d, %d\n", h1.Nonce.Uint64(), h2.Nonce.Uint64())
-	fmt.Printf("number:      %d, %d\n", h1.Number.Uint64(), h2.Number.Uint64())
-	fmt.Printf("gasLimit:    %d, %d\n", h1.GasLimit, h2.GasLimit)
-	fmt.Printf("gasUsed:     %d, %d\n", h1.GasUsed, h2.GasUsed)
-	fmt.Printf("Difficulty:  %d, %d\n", h1.Difficulty, h2.Difficulty)
-	fmt.Printf("ReceiptHash: %x, %x\n", h1.ReceiptHash, h2.ReceiptHash)
-	fmt.Printf("TxHash:      %x, %x\n", h1.TxHash, h2.TxHash)
-	fmt.Printf("UncleHash:   %x, %x\n", h1.UncleHash, h2.UncleHash)
-	fmt.Printf("ParentHash:  %x, %x\n", h1.ParentHash, h2.ParentHash)
-}
-
 func loopIh(db ethdb.Database, ctx context.Context, unwind uint64) error {
 	ch := ctx.Done()
 	var tx = ethdb.NewTxDbWithoutTransaction(db, ethdb.RW)

cmd/integration/main.go

@@ -2,9 +2,10 @@ package main
 
 import (
 	"fmt"
+	"os"
+
 	"github.com/ledgerwatch/turbo-geth/cmd/integration/commands"
 	"github.com/ledgerwatch/turbo-geth/cmd/utils"
-	"os"
 )
 
 func main() {

cmd/rpcdaemon/filters/filters.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/ledgerwatch/turbo-geth/core"
 	"github.com/ledgerwatch/turbo-geth/core/types"
-	"github.com/ledgerwatch/turbo-geth/ethdb/remote/remotedbserver"
 	"github.com/ledgerwatch/turbo-geth/gointerfaces/remote"
 	"github.com/ledgerwatch/turbo-geth/log"
 )
@@ -58,7 +57,7 @@ func (ff *Filters) OnNewEvent(event *remote.SubscribeReply) {
 	ff.mu.RLock()
 	defer ff.mu.RUnlock()
 
-	if remotedbserver.RpcEventType(event.Type) != remotedbserver.EventTypeHeader {
+	if event.Type != remote.Event_HEADER {
 		log.Warn("rpc filters: unsupported event type", "type", event.Type)
 		return
 	}

cmd/rpcdaemon/main.go

@@ -31,7 +31,11 @@ func main() {
 			log.Info("filters are not supported in chaindata mode")
 		}
 
-		return cli.StartRpcServer(cmd.Context(), *cfg, commands.APIList(ethdb.NewObjectDatabase(db), backend, ff, *cfg, nil))
+		if err := cli.StartRpcServer(cmd.Context(), *cfg, commands.APIList(ethdb.NewObjectDatabase(db), backend, ff, *cfg, nil)); err != nil {
+			log.Error(err.Error())
+			return nil
+		}
+		return nil
 	}
 
 	if err := cmd.ExecuteContext(utils.RootContext()); err != nil {

common/debugprint/receipts.go

@@ -0,0 +1,53 @@
+package debugprint
+
+import (
+	"fmt"
+
+	"github.com/ledgerwatch/turbo-geth/core/types"
+)
+
+//nolint
+func Receipts(rs1, rs2 types.Receipts) {
+	fmt.Printf("==== Receipts ====\n")
+	fmt.Printf("len(Receipts): %d, %d\n", len(rs1), len(rs2))
+	for len(rs2) < len(rs1) {
+		rs2 = append(rs2, &types.Receipt{})
+	}
+
+	for i := range rs1 {
+		r1, r2 := rs1[i], rs2[i]
+		fmt.Printf(" ==== Receipt ====\n")
+		fmt.Printf(" TxHash:            %x, %x\n", r1.TxHash, r2.TxHash)
+		fmt.Printf(" PostState:         %x, %x\n", r1.PostState, r2.PostState)
+		fmt.Printf(" Status:            %d, %d\n", r1.Status, r2.Status)
+		fmt.Printf(" CumulativeGasUsed: %d, %d\n", r1.CumulativeGasUsed, r2.CumulativeGasUsed)
+		fmt.Printf(" ContractAddress:   %x, %x\n", r1.ContractAddress, r2.ContractAddress)
+		fmt.Printf(" GasUsed:           %x, %x\n", r1.GasUsed, r2.GasUsed)
+		fmt.Printf(" len(Logs):         %d, %d\n", len(r1.Logs), len(r2.Logs))
+		for len(r2.Logs) < len(r1.Logs) {
+			r2.Logs = append(r2.Logs, &types.Log{})
+		}
+		for j := range r1.Logs {
+			l1, l2 := r1.Logs[j], r2.Logs[j]
+			fmt.Printf("  Logs[%d].Address: %x, %x\n", j, l1.Address, l2.Address)
+			fmt.Printf("  Logs[%d].Topic:   %x, %x\n", j, l1.Topics, l2.Topics)
+			fmt.Printf("  Logs[%d].Data:    %x, %x\n", j, l1.Data, l2.Data)
+		}
+		fmt.Printf(" Bloom: %x, %x\n", r1.Bloom, r1.Bloom)
+	}
+}
+
+//nolint
+func Headers(h1, h2 *types.Header) {
+	fmt.Printf("==== Header ====\n")
+	fmt.Printf("root:        %x, %x\n", h1.Root, h2.Root)
+	fmt.Printf("nonce:       %d, %d\n", h1.Nonce.Uint64(), h2.Nonce.Uint64())
+	fmt.Printf("number:      %d, %d\n", h1.Number.Uint64(), h2.Number.Uint64())
+	fmt.Printf("gasLimit:    %d, %d\n", h1.GasLimit, h2.GasLimit)
+	fmt.Printf("gasUsed:     %d, %d\n", h1.GasUsed, h2.GasUsed)
+	fmt.Printf("Difficulty:  %d, %d\n", h1.Difficulty, h2.Difficulty)
+	fmt.Printf("ReceiptHash: %x, %x\n", h1.ReceiptHash, h2.ReceiptHash)
+	fmt.Printf("TxHash:      %x, %x\n", h1.TxHash, h2.TxHash)
+	fmt.Printf("UncleHash:   %x, %x\n", h1.UncleHash, h2.UncleHash)
+	fmt.Printf("ParentHash:  %x, %x\n", h1.ParentHash, h2.ParentHash)
+}

core/rawdb/accessors_chain.go

@@ -508,7 +508,7 @@ func HasReceipts(db databaseReader, hash common.Hash, number uint64) bool {
 // ReadRawReceipts retrieves all the transaction receipts belonging to a block.
 // The receipt metadata fields are not guaranteed to be populated, so they
 // should not be used. Use ReadReceipts instead if the metadata is needed.
-func ReadRawReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Receipts {
+func ReadRawReceipts(db ethdb.Getter, hash common.Hash, number uint64) types.Receipts {
 	// Retrieve the flattened receipt slice
 	data, err := db.Get(dbutils.BlockReceiptsPrefix, dbutils.ReceiptsKey(number))
 	if err != nil && !errors.Is(err, ethdb.ErrKeyNotFound) {
@@ -546,7 +546,7 @@ func ReadRawReceipts(db ethdb.Database, hash common.Hash, number uint64) types.R
 // The current implementation populates these metadata fields by reading the receipts'
 // corresponding block body, so if the block body is not found it will return nil even
 // if the receipt itself is stored.
-func ReadReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Receipts {
+func ReadReceipts(db ethdb.Getter, hash common.Hash, number uint64) types.Receipts {
 	// We're deriving many fields from the block body, retrieve beside the receipt
 	receipts := ReadRawReceipts(db, hash, number)
 	if receipts == nil {
@@ -569,6 +569,11 @@ func ReadReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Rece
 	return receipts
 }
 
+func ReadReceiptsByNumber(db ethdb.Getter, number uint64) types.Receipts {
+	h, _ := ReadCanonicalHash(db, number)
+	return ReadReceipts(db, h, number)
+}
+
 // WriteReceipts stores all the transaction receipts belonging to a block.
 func WriteReceipts(tx DatabaseWriter, number uint64, receipts types.Receipts) error {
 	buf := bytes.NewBuffer(make([]byte, 0, 1024))

core/types/access_list_tx.go

@@ -21,7 +21,7 @@ import (
 	"github.com/ledgerwatch/turbo-geth/common"
 )
 
-//go:generate gencodec -type AccessTuple -out gen_access_tuple.go
+// go:generate gencodec -type AccessTuple -out gen_access_tuple.go
 
 // AccessList is an EIP-2930 access list.
 type AccessList []AccessTuple

eth/backend.go

@@ -452,8 +452,8 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		return nil, err
 	}
 	//if config.SyncMode != downloader.StagedSync {
-	eth.miner = miner.New(eth, &config.Miner, chainConfig, eth.EventMux(), eth.engine, eth.isLocalBlock)
-	_ = eth.miner.SetExtra(makeExtraData(config.Miner.ExtraData))
+	//eth.miner = miner.New(eth, &config.Miner, chainConfig, eth.EventMux(), eth.engine, eth.isLocalBlock)
+	//_ = eth.miner.SetExtra(makeExtraData(config.Miner.ExtraData))
 	//}
 	eth.snapDialCandidates, _ = setupDiscovery(eth.config.SnapDiscoveryURLs) //nolint:staticcheck
 	eth.handler.SetTmpDir(tmpdir)
@@ -724,9 +724,7 @@ func (s *Ethereum) StartMining(threads int) error {
 		// If mining is started, we can disable the transaction rejection mechanism
 		// introduced to speed sync times.
 		atomic.StoreUint32(&s.handler.acceptTxs, 1)
-		if s.config.SyncMode != downloader.StagedSync {
-			go s.miner.Start(eb)
-		}
+		//go s.miner.Start(eb)
 	}
 	return nil
 }
@@ -804,9 +802,7 @@ func (s *Ethereum) Stop() error {
 		}
 	}
 
-	if s.config.SyncMode != downloader.StagedSync {
-		s.miner.Stop()
-	}
+	//s.miner.Stop()
 	s.blockchain.Stop()
 	s.engine.Close()
 	s.eventMux.Stop()

eth/filters/filter_system.go

@@ -128,10 +128,11 @@ func NewEventSystem(backend Backend, lightMode bool) *EventSystem {
 	m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
 	m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
 	m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
-	m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh)
+	//m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh)
 
 	// Make sure none of the subscriptions are empty
-	if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.pendingLogsSub == nil {
+	//if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil || m.pendingLogsSub == nil {
+	if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil {
 		log.Crit("Subscribe for event system failed")
 	}
 
@@ -446,7 +447,7 @@ func (es *EventSystem) eventLoop() {
 		es.txsSub.Unsubscribe()
 		es.logsSub.Unsubscribe()
 		es.rmLogsSub.Unsubscribe()
-		es.pendingLogsSub.Unsubscribe()
+		//es.pendingLogsSub.Unsubscribe()
 		es.chainSub.Unsubscribe()
 	}()
 

eth/filters/filter_system_test.go

@@ -374,6 +374,7 @@ func TestInvalidGetLogsRequest(t *testing.T) {
 
 // TestLogFilter tests whether log filters match the correct logs that are posted to the event feed.
 func TestLogFilter(t *testing.T) {
+	t.Skip("TG doesn't have public API, move this test to RPCDaemon")
 	t.Parallel()
 
 	db := ethdb.NewMemDatabase()

eth/stagedsync/all_stages.go

@@ -286,7 +286,7 @@ func createStageBuilders(blocks []*types.Block, blockNum uint64, checkRoot bool)
 						logPrefix := s.state.LogPrefix()
 						log.Info(fmt.Sprintf("[%s] Update current block for the RPC API", logPrefix), "to", executionAt)
 
-						err = NotifyRpcDaemon(s.BlockNumber+1, executionAt, world.notifier, world.TX)
+						err = NotifyNewHeaders(s.BlockNumber+1, executionAt, world.notifier, world.TX)
 						if err != nil {
 							return err
 						}

eth/stagedsync/stage_finish.go

@@ -8,19 +8,15 @@ import (
 	"github.com/ledgerwatch/turbo-geth/log"
 )
 
-func NotifyRpcDaemon(from, to uint64, notifier ChainEventNotifier, db ethdb.Database) error {
+func NotifyNewHeaders(from, to uint64, notifier ChainEventNotifier, db ethdb.Database) error {
 	if notifier == nil {
 		log.Warn("rpc notifier is not set, rpc daemon won't be updated about headers")
 		return nil
 	}
 	for i := from; i <= to; i++ {
-		hash, err := rawdb.ReadCanonicalHash(db, i)
-		if err != nil {
-			return err
-		}
-		header := rawdb.ReadHeader(db, hash, i)
+		header := rawdb.ReadHeaderByNumber(db, i)
 		if header == nil {
-			return fmt.Errorf("could not find canonical header for hash: %x number: %d", hash, i)
+			return fmt.Errorf("could not find canonical header for number: %d", i)
 		}
 		notifier.OnNewHeader(header)
 	}

eth/stagedsync/stage_mining_create_block.go

@@ -35,7 +35,6 @@ type miningBlock struct {
 
 // SpawnMiningCreateBlockStage
 //TODO:
-// - interrupt - variable is not implemented, see miner/worker.go:798
 // - resubmitAdjustCh - variable is not implemented
 func SpawnMiningCreateBlockStage(s *StageState, tx ethdb.Database, current *miningBlock, chainConfig *params.ChainConfig, engine consensus.Engine, extra hexutil.Bytes, gasFloor, gasCeil uint64, coinbase common.Address, txPoolLocals []common.Address, pendingTxs map[common.Address]types.Transactions, quit <-chan struct{}) error {
 	const (