consensus/bor, eth, miner, internal/cli: add sequence store publisher

consensus/bor/bor.go

@@ -1129,12 +1129,14 @@ func (c *Bor) Prepare(chain consensus.ChainHeaderReader, header *types.Header, w
 	var succession int
 	// if signer is not empty
 	if currentSigner.signer != (common.Address{}) {
+		// A signer outside the active set (post-Rio: outside the span's
+		// producer set) does not build at all — its candidate could never
+		// seal, and its sequence must never reach the store. Nodes without
+		// a signer (RPC) skip this check and keep their pending snapshot
+		// fresh.
 		succession, err = snap.GetSignerSuccessionNumber(currentSigner.signer)
 		if err != nil {
-			// If the signer is not in the active validator set, use succession 0
-			// so that the pending block header is still valid for RPC queries.
-			// Seal() will independently reject the block if unauthorized.
-			succession = 0
+			return err
 		}
 	}
 

docs/cli/default_config.toml

@@ -267,3 +267,9 @@ devfakeauthor = false
   enable-preconfs = false
   enable-private-tx = false
   bp-rpc-endpoints = []
+
+[sequencer]
+  enabled = false
+  publisher-endpoint = ""
+  consumer-endpoint = ""
+  poll = "200ms"

docs/cli/server.md

@@ -104,6 +104,14 @@ The ```bor server``` command runs the Bor client.
 
 - ```rpc.returndatalimit```: Maximum size (in bytes) a result of an rpc request could have (use 0 for no limits) (default: 100000)
 
+- ```sequencer.consumer-endpoint```: Sequence store consumer service gRPC endpoint (tail reads during reconciliation)
+
+- ```sequencer.enabled```: Enable the sequence store integration (a mining node publishes the block lifecycle) (default: false)
+
+- ```sequencer.poll```: Producer txpool poll cadence while a block is open (continuous building); 0 keeps the one-shot fill (default: 200ms)
+
+- ```sequencer.publisher-endpoint```: Sequence store publisher service gRPC endpoint (publish stream)
+
 - ```snapshot```: Enables the snapshot-database mode (default: true)
 
 - ```state.scheme```: Scheme to use for storing ethereum state ('hash' or 'path') (default: path)

eth/backend.go

@@ -55,6 +55,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/eth/protocols/wit"
 	"github.com/ethereum/go-ethereum/eth/relay"
+	"github.com/ethereum/go-ethereum/eth/sequencer"
 	"github.com/ethereum/go-ethereum/eth/tracers"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
@@ -130,6 +131,11 @@ type Ethereum struct {
 	gasPrice  *big.Int
 	etherbase common.Address
 
+	// Sequence store integration (design docs/sequencer-bor.md): at most
+	// one is set, by role.
+	seqPublisher *sequencer.Publisher // mining node: publishes the block lifecycle
+	seqConsumer  *sequencer.Consumer  // RPC node: re-executes the stream for preconf receipts
+
 	networkID     uint64
 	netRPCService *ethapi.NetAPI
 
@@ -142,6 +148,45 @@ type Ethereum struct {
 	shutdownTracker *shutdowncheck.ShutdownTracker // Tracks if and when the node has shutdown ungracefully
 }
 
+// attachSequencer wires the sequence store integration by role (design
+// docs/sequencer-bor.md): a mining node publishes the block lifecycle, a
+// non-mining node follows the stream for preconf receipts.
+func (s *Ethereum) attachSequencer(config *ethconfig.Config) error {
+	switch config.SequencerRole {
+	case "producer":
+		if s.miner == nil {
+			return nil
+		}
+
+		publisher, err := sequencer.NewPublisher(config.SequencerPublisherEndpoint,
+			config.SequencerConsumerEndpoint, s.blockchain.Config().ChainID.Uint64(),
+			config.SequencerPoll, s.blockchain)
+		if err != nil {
+			return fmt.Errorf("sequencer publisher: %w", err)
+		}
+
+		s.seqPublisher = publisher
+		s.miner.SetSequencer(publisher)
+	case "consumer":
+		// A consumer that cannot start (not a bor chain) must not block the
+		// node: preconfs stay off, everything else runs.
+		consumer, err := sequencer.NewConsumer(config.SequencerConsumerEndpoint, s.blockchain)
+		if err != nil {
+			log.Error("Sequencer consumer disabled", "err", err)
+
+			return nil
+		}
+
+		s.seqConsumer = consumer
+		consumer.Start()
+	case "":
+	default:
+		return fmt.Errorf("unknown sequencer role %q", config.SequencerRole)
+	}
+
+	return nil
+}
+
 // New creates a new Ethereum object (including the initialisation of the common Ethereum object),
 // whose lifecycle will be managed by the provided node.
 func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
@@ -305,32 +350,30 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		trieJournalDirectory = stack.ResolvePath("triedb")
 	}
 
-	var (
-		options = &core.BlockChainConfig{
-			TrieCleanLimit:    config.TrieCleanCache,
-			NoPrefetch:        config.NoPrefetch,
-			TrieDirtyLimit:    config.TrieDirtyCache,
-			ArchiveMode:       config.NoPruning,
-			TrieTimeLimit:     config.TrieTimeout,
-			SnapshotLimit:     config.SnapshotCache,
-			Preimages:         config.Preimages,
-			StateHistory:      config.StateHistory,
-			StateScheme:       scheme,
-			TriesInMemory:     config.TriesInMemory,
-			ChainHistoryMode:  config.HistoryMode,
-			TxLookupLimit:     int64(min(config.TransactionHistory, math.MaxInt64)),
-			AddressCacheSizes: config.AddressCacheSizes,
-			PreloadRateLimit:  config.PreloadRateLimit,
-			VmConfig:          vmCfg,
-			Stateless:         config.SyncMode == downloader.StatelessSync,
-			// Enables file journaling for the trie database. The journal files will be stored
-			// within the data directory. The corresponding paths will be either:
-			// - DATADIR/triedb/merkle.journal
-			// - DATADIR/triedb/verkle.journal
-			TrieJournalDirectory: trieJournalDirectory,
-			StateSizeTracking:    config.EnableStateSizeTracking,
-		}
-	)
+	options := &core.BlockChainConfig{
+		TrieCleanLimit:    config.TrieCleanCache,
+		NoPrefetch:        config.NoPrefetch,
+		TrieDirtyLimit:    config.TrieDirtyCache,
+		ArchiveMode:       config.NoPruning,
+		TrieTimeLimit:     config.TrieTimeout,
+		SnapshotLimit:     config.SnapshotCache,
+		Preimages:         config.Preimages,
+		StateHistory:      config.StateHistory,
+		StateScheme:       scheme,
+		TriesInMemory:     config.TriesInMemory,
+		ChainHistoryMode:  config.HistoryMode,
+		TxLookupLimit:     int64(min(config.TransactionHistory, math.MaxInt64)),
+		AddressCacheSizes: config.AddressCacheSizes,
+		PreloadRateLimit:  config.PreloadRateLimit,
+		VmConfig:          vmCfg,
+		Stateless:         config.SyncMode == downloader.StatelessSync,
+		// Enables file journaling for the trie database. The journal files will be stored
+		// within the data directory. The corresponding paths will be either:
+		// - DATADIR/triedb/merkle.journal
+		// - DATADIR/triedb/verkle.journal
+		TrieJournalDirectory: trieJournalDirectory,
+		StateSizeTracking:    config.EnableStateSizeTracking,
+	}
 
 	checker := whitelist.NewService(chainDb, config.DisableBlindForkValidation, config.MaxBlindForkValidationLimit)
 
@@ -474,6 +517,10 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		eth.miner.SetPrioAddresses(config.TxPool.Locals)
 	}
 
+	if err := eth.attachSequencer(config); err != nil {
+		return nil, err
+	}
+
 	// 1.14.8: NewOracle function definition was changed to accept (startPrice *big.Int) param.
 	eth.APIBackend.gpo = gasprice.NewOracle(eth.APIBackend, config.GPO, config.Miner.GasPrice)
 	eth.APIBackend.gpo.ProcessCache()
@@ -696,13 +743,21 @@ func (s *Ethereum) StopMining() {
 func (s *Ethereum) IsMining() bool      { return s.miner.Mining() }
 func (s *Ethereum) Miner() *miner.Miner { return s.miner }
 
-func (s *Ethereum) AccountManager() *accounts.Manager  { return s.accountManager }
-func (s *Ethereum) BlockChain() *core.BlockChain       { return s.blockchain }
-func (s *Ethereum) TxPool() *txpool.TxPool             { return s.txPool }
-func (s *Ethereum) BlobTxPool() *blobpool.BlobPool     { return s.blobTxPool }
-func (s *Ethereum) Engine() consensus.Engine           { return s.engine }
-func (s *Ethereum) ChainDb() ethdb.Database            { return s.chainDb }
-func (s *Ethereum) IsListening() bool                  { return true } // Always listening
+func (s *Ethereum) AccountManager() *accounts.Manager { return s.accountManager }
+func (s *Ethereum) BlockChain() *core.BlockChain      { return s.blockchain }
+func (s *Ethereum) TxPool() *txpool.TxPool            { return s.txPool }
+func (s *Ethereum) BlobTxPool() *blobpool.BlobPool    { return s.blobTxPool }
+func (s *Ethereum) Engine() consensus.Engine          { return s.engine }
+func (s *Ethereum) ChainDb() ethdb.Database           { return s.chainDb }
+func (s *Ethereum) IsListening() bool                 { return true } // Always listening
+// WhitelistedMilestone implements miner.Backend: the newest Heimdall
+// milestone the downloader has whitelisted, or false when none has arrived.
+// The miner's finality gate compares it against the local chain before a
+// producer builds.
+func (s *Ethereum) WhitelistedMilestone() (bool, uint64, common.Hash) {
+	return s.Downloader().ChainValidator.GetWhitelistedMilestone()
+}
+
 func (s *Ethereum) Downloader() *downloader.Downloader { return s.handler.downloader }
 func (s *Ethereum) Synced() bool                       { return s.handler.synced.Load() }
 func (s *Ethereum) SetSynced()                         { s.handler.enableSyncedFeatures() }
@@ -1083,6 +1138,14 @@ func (s *Ethereum) Stop() error {
 	<-ch
 	s.filterMaps.Stop()
 	s.txPool.Close()
+	if s.seqPublisher != nil {
+		s.seqPublisher.Close()
+	}
+
+	if s.seqConsumer != nil {
+		s.seqConsumer.Close()
+	}
+
 	if s.miner != nil {
 		s.miner.Close()
 	}

eth/ethconfig/config.go

@@ -329,7 +329,20 @@ type Config struct {
 	TxSyncDefaultTimeout time.Duration `toml:",omitempty"`
 	TxSyncMaxTimeout     time.Duration `toml:",omitempty"`
 
+	// Sequence store integration (design docs/sequencer-bor.md). Role is
+	// derived, not configured: "producer" on a mining node (publishes the
+	// block lifecycle), "consumer" on a non-mining node (re-executes the
+	// stream for preconf receipts), empty when disabled. The publisher and
+	// consumer gRPC services have their own endpoints (the publisher reads
+	// the tail through the consumer service when it reconciles); Poll is
+	// the txpool poll cadence while a block is open.
+	SequencerRole              string
+	SequencerPublisherEndpoint string
+	SequencerConsumerEndpoint  string
+	SequencerPoll              time.Duration
+
 	// Preconf / Private transaction relay related settings
+
 	EnablePreconfs            bool
 	EnablePrivateTx           bool
 	BlockProducerRpcEndpoints []string