Execution service 4: execution->consensus interface

arbnode/inbox_reader.go

@@ -10,7 +10,6 @@ import (
 	"math"
 	"math/big"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -90,10 +89,6 @@ type InboxReader struct {
 
 	// Atomic
 	lastSeenBatchCount uint64
-
-	// Behind the mutex
-	lastReadMutex      sync.RWMutex
-	lastReadBlock      uint64
 	lastReadBatchCount uint64
 }
 
@@ -353,10 +348,7 @@ func (r *InboxReader) run(ctx context.Context, hadError bool) error {
 			// There's nothing to do
 			from = arbmath.BigAddByUint(currentHeight, 1)
 			blocksToFetch = config.DefaultBlocksToRead
-			r.lastReadMutex.Lock()
-			r.lastReadBlock = currentHeight.Uint64()
-			r.lastReadBatchCount = checkingBatchCount
-			r.lastReadMutex.Unlock()
+			atomic.StoreUint64(&r.lastReadBatchCount, checkingBatchCount)
 			storeSeenBatchCount()
 			if !r.caughtUp {
 				r.caughtUp = true
@@ -487,10 +479,7 @@ func (r *InboxReader) run(ctx context.Context, hadError bool) error {
 				}
 				if len(sequencerBatches) > 0 {
 					readAnyBatches = true
-					r.lastReadMutex.Lock()
-					r.lastReadBlock = to.Uint64()
-					r.lastReadBatchCount = sequencerBatches[len(sequencerBatches)-1].SequenceNumber + 1
-					r.lastReadMutex.Unlock()
+					atomic.StoreUint64(&r.lastReadBatchCount, sequencerBatches[len(sequencerBatches)-1].SequenceNumber+1)
 					storeSeenBatchCount()
 				}
 			}
@@ -517,10 +506,7 @@ func (r *InboxReader) run(ctx context.Context, hadError bool) error {
 		}
 
 		if !readAnyBatches {
-			r.lastReadMutex.Lock()
-			r.lastReadBlock = currentHeight.Uint64()
-			r.lastReadBatchCount = checkingBatchCount
-			r.lastReadMutex.Unlock()
+			atomic.StoreUint64(&r.lastReadBatchCount, checkingBatchCount)
 			storeSeenBatchCount()
 		}
 	}
@@ -590,10 +576,8 @@ func (r *InboxReader) GetSequencerMessageBytes(ctx context.Context, seqNum uint6
 	return nil, fmt.Errorf("sequencer batch %v not found in L1 block %v (found batches %v)", seqNum, metadata.ParentChainBlock, seenBatches)
 }
 
-func (r *InboxReader) GetLastReadBlockAndBatchCount() (uint64, uint64) {
-	r.lastReadMutex.RLock()
-	defer r.lastReadMutex.RUnlock()
-	return r.lastReadBlock, r.lastReadBatchCount
+func (r *InboxReader) GetLastReadBatchCount() uint64 {
+	return atomic.LoadUint64(&r.lastReadBatchCount)
 }
 
 // GetLastSeenBatchCount returns how many sequencer batches the inbox reader has read in from L1.

arbnode/inbox_tracker.go

@@ -194,6 +194,11 @@ func (t *InboxTracker) GetBatchMessageCount(seqNum uint64) (arbutil.MessageIndex
 	return metadata.MessageCount, err
 }
 
+func (t *InboxTracker) GetBatchParentChainBlock(seqNum uint64) (uint64, error) {
+	metadata, err := t.GetBatchMetadata(seqNum)
+	return metadata.ParentChainBlock, err
+}
+
 // GetBatchAcc is a convenience function wrapping GetBatchMetadata
 func (t *InboxTracker) GetBatchAcc(seqNum uint64) (common.Hash, error) {
 	metadata, err := t.GetBatchMetadata(seqNum)
@@ -213,6 +218,44 @@ func (t *InboxTracker) GetBatchCount() (uint64, error) {
 	return count, nil
 }
 
+func (t *InboxTracker) FindL1BatchForMessage(pos arbutil.MessageIndex) (uint64, error) {
+	batchCount, err := t.GetBatchCount()
+	if err != nil {
+		return 0, err
+	}
+	low := uint64(0)
+	high := batchCount - 1
+	// Iteration preconditions:
+	// - high >= low
+	// - msgCount(low - 1) <= pos implies low <= target
+	// - msgCount(high) > pos implies high >= target
+	// Therefore, if low == high, then low == high == target
+	for {
+		// Due to integer rounding, mid >= low && mid < high
+		mid := (low + high) / 2
+		count, err := t.GetBatchMessageCount(mid)
+		if err != nil {
+			return 0, err
+		}
+		if count < pos {
+			// Must narrow as mid >= low, therefore mid + 1 > low, therefore newLow > oldLow
+			// Keeps low precondition as msgCount(mid) < pos
+			low = mid + 1
+		} else if count == pos {
+			return mid + 1, err
+		} else if count == pos+1 || mid == low { // implied: count > pos
+			return mid, nil
+		} else { // implied: count > pos + 1
+			// Must narrow as mid < high, therefore newHigh < lowHigh
+			// Keeps high precondition as msgCount(mid) > pos
+			high = mid
+		}
+		if high == low {
+			return high, err
+		}
+	}
+}
+
 func (t *InboxTracker) PopulateFeedBacklog(broadcastServer *broadcaster.Broadcaster) error {
 	batchCount, err := t.GetBatchCount()
 	if err != nil {

arbnode/node.go

@@ -28,6 +28,7 @@ import (
 	"github.com/offchainlabs/nitro/arbnode/dataposter/storage"
 	"github.com/offchainlabs/nitro/arbnode/redislock"
 	"github.com/offchainlabs/nitro/arbnode/resourcemanager"
+	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcastclient"
 	"github.com/offchainlabs/nitro/broadcastclients"
@@ -391,6 +392,7 @@ func ConfigDefaultL1NonSequencerTest() *Config {
 	config.BatchPoster.Enable = false
 	config.SeqCoordinator.Enable = false
 	config.BlockValidator = staker.TestBlockValidatorConfig
+	config.SyncMonitor = TestSyncMonitorConfig
 	config.Staker = staker.TestL1ValidatorConfig
 	config.Staker.Enable = false
 	config.BlockValidator.ValidationServer.URL = ""
@@ -407,6 +409,7 @@ func ConfigDefaultL2Test() *Config {
 	config.SeqCoordinator.Signer.ECDSA.AcceptSequencer = false
 	config.SeqCoordinator.Signer.ECDSA.Dangerous.AcceptMissing = true
 	config.Staker = staker.TestL1ValidatorConfig
+	config.SyncMonitor = TestSyncMonitorConfig
 	config.Staker.Enable = false
 	config.BlockValidator.ValidationServer.URL = ""
 	config.TransactionStreamer = DefaultTransactionStreamerConfig
@@ -449,7 +452,6 @@ type Node struct {
 	SeqCoordinator          *SeqCoordinator
 	MaintenanceRunner       *MaintenanceRunner
 	DASLifecycleManager     *das.LifecycleManager
-	ClassicOutboxRetriever  *ClassicOutboxRetriever
 	SyncMonitor             *SyncMonitor
 	configFetcher           ConfigFetcher
 	ctx                     context.Context
@@ -564,17 +566,10 @@ func createNodeImpl(
 
 	l2ChainId := l2Config.ChainID.Uint64()
 
-	syncMonitor := NewSyncMonitor(&config.SyncMonitor)
-	var classicOutbox *ClassicOutboxRetriever
-	classicMsgDb, err := stack.OpenDatabase("classic-msg", 0, 0, "", true)
-	if err != nil {
-		if l2Config.ArbitrumChainParams.GenesisBlockNum > 0 {
-			log.Warn("Classic Msg Database not found", "err", err)
-		}
-		classicOutbox = nil
-	} else {
-		classicOutbox = NewClassicOutboxRetriever(classicMsgDb)
+	syncConfigFetcher := func() *SyncMonitorConfig {
+		return &configFetcher.Get().SyncMonitor
 	}
+	syncMonitor := NewSyncMonitor(syncConfigFetcher)
 
 	var l1Reader *headerreader.HeaderReader
 	if config.ParentChainReader.Enable {
@@ -670,7 +665,6 @@ func createNodeImpl(
 			SeqCoordinator:          coordinator,
 			MaintenanceRunner:       maintenanceRunner,
 			DASLifecycleManager:     nil,
-			ClassicOutboxRetriever:  classicOutbox,
 			SyncMonitor:             syncMonitor,
 			configFetcher:           configFetcher,
 			ctx:                     ctx,
@@ -873,7 +867,6 @@ func createNodeImpl(
 		SeqCoordinator:          coordinator,
 		MaintenanceRunner:       maintenanceRunner,
 		DASLifecycleManager:     dasLifecycleManager,
-		ClassicOutboxRetriever:  classicOutbox,
 		SyncMonitor:             syncMonitor,
 		configFetcher:           configFetcher,
 		ctx:                     ctx,
@@ -936,16 +929,19 @@ func (n *Node) Start(ctx context.Context) error {
 		execClient = nil
 	}
 	if execClient != nil {
-		err := execClient.Initialize(ctx, n, n.SyncMonitor)
+		err := execClient.Initialize(ctx)
 		if err != nil {
 			return fmt.Errorf("error initializing exec client: %w", err)
 		}
 	}
-	n.SyncMonitor.Initialize(n.InboxReader, n.TxStreamer, n.SeqCoordinator, n.Execution)
+	n.SyncMonitor.Initialize(n.InboxReader, n.TxStreamer, n.SeqCoordinator)
 	err := n.Stack.Start()
 	if err != nil {
 		return fmt.Errorf("error starting geth stack: %w", err)
 	}
+	if execClient != nil {
+		execClient.SetConsensusClient(n)
+	}
 	err = n.Execution.Start(ctx)
 	if err != nil {
 		return fmt.Errorf("error starting exec client: %w", err)
@@ -1049,6 +1045,7 @@ func (n *Node) Start(ctx context.Context) error {
 	if n.configFetcher != nil {
 		n.configFetcher.Start(ctx)
 	}
+	n.SyncMonitor.Start(ctx)
 	return nil
 }
 
@@ -1105,10 +1102,48 @@ func (n *Node) StopAndWait() {
 		// Just stops the redis client (most other stuff was stopped earlier)
 		n.SeqCoordinator.StopAndWait()
 	}
+	n.SyncMonitor.StopAndWait()
 	if n.DASLifecycleManager != nil {
 		n.DASLifecycleManager.StopAndWaitUntil(2 * time.Second)
 	}
 	if err := n.Stack.Close(); err != nil {
 		log.Error("error on stak close", "err", err)
 	}
 }
+
+func (n *Node) FetchBatch(ctx context.Context, batchNum uint64) ([]byte, error) {
+	return n.InboxReader.GetSequencerMessageBytes(ctx, batchNum)
+}
+
+func (n *Node) FindL1BatchForMessage(message arbutil.MessageIndex) (uint64, error) {
+	return n.InboxTracker.FindL1BatchForMessage(message)
+}
+
+func (n *Node) GetBatchParentChainBlock(seqNum uint64) (uint64, error) {
+	return n.InboxTracker.GetBatchParentChainBlock(seqNum)
+}
+
+func (n *Node) SyncProgressMap() map[string]interface{} {
+	return n.SyncMonitor.SyncProgressMap()
+}
+
+func (n *Node) SyncTargetMessageCount() arbutil.MessageIndex {
+	return n.SyncMonitor.SyncTargetMessageCount()
+}
+
+// TODO: switch from pulling to pushing safe/finalized
+func (n *Node) GetSafeMsgCount(ctx context.Context) (arbutil.MessageIndex, error) {
+	return n.InboxReader.GetSafeMsgCount(ctx)
+}
+
+func (n *Node) GetFinalizedMsgCount(ctx context.Context) (arbutil.MessageIndex, error) {
+	return n.InboxReader.GetFinalizedMsgCount(ctx)
+}
+
+func (n *Node) WriteMessageFromSequencer(pos arbutil.MessageIndex, msgWithMeta arbostypes.MessageWithMetadata) error {
+	return n.TxStreamer.WriteMessageFromSequencer(pos, msgWithMeta)
+}
+
+func (n *Node) ExpectChosenSequencer() error {
+	return n.TxStreamer.ExpectChosenSequencer()
+}