op-node: fix driver step hot loop, improve events and add utils (#11040)

* op-node: fix driver step hot loop, improvement events and add utils

* op-node: fix verifier event processing liveness

* op-node: add events_rate_limited metric

* op-node: bump events rate limits

op-e2e/actions/l2_sequencer.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/driver"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
@@ -109,7 +110,7 @@ func (s *L2Sequencer) ActL2EndBlock(t Testing) {
 	// This will ensure the sync-status and such reflect the latest changes.
 	s.synchronousEvents.Emit(engine.TryUpdateEngineEvent{})
 	s.synchronousEvents.Emit(engine.ForkchoiceRequestEvent{})
-	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev rollup.Event) bool {
+	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev event.Event) bool {
 		x, ok := ev.(engine.ForkchoiceUpdateEvent)
 		return ok && x.UnsafeL2Head == s.engine.UnsafeL2Head()
 	}, false))

op-e2e/actions/l2_verifier.go

@@ -6,11 +6,13 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	gnode "github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum-optimism/optimism/op-node/node"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
@@ -19,6 +21,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/driver"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/finality"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/status"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
@@ -41,7 +44,7 @@ type L2Verifier struct {
 
 	syncStatus driver.SyncStatusTracker
 
-	synchronousEvents *rollup.SynchronousEvents
+	synchronousEvents event.EmitterDrainer
 
 	syncDeriver *driver.SyncDeriver
 
@@ -88,8 +91,13 @@ func NewL2Verifier(t Testing, log log.Logger, l1 derive.L1Fetcher, blobsSrc deri
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	rootDeriver := &rollup.SynchronousDerivers{}
-	synchronousEvents := rollup.NewSynchronousEvents(log, ctx, rootDeriver)
+	rootDeriver := &event.DeriverMux{}
+	var synchronousEvents event.EmitterDrainer
+	synchronousEvents = event.NewQueue(log, ctx, rootDeriver, event.NoopMetrics{})
+	synchronousEvents = event.NewLimiterDrainer(ctx, synchronousEvents, rate.Limit(1000), 20, func() {
+		log.Warn("Hitting events rate-limit. An events code-path may be hot-looping.")
+		t.Fatal("Tests must not hot-loop events")
+	})
 
 	metrics := &testutils.TestDerivationMetrics{}
 	ec := engine.NewEngineController(eng, log, metrics, cfg, syncCfg.SyncMode, synchronousEvents)
@@ -148,7 +156,7 @@ func NewL2Verifier(t Testing, log log.Logger, l1 derive.L1Fetcher, blobsSrc deri
 		synchronousEvents: synchronousEvents,
 	}
 
-	*rootDeriver = rollup.SynchronousDerivers{
+	*rootDeriver = event.DeriverMux{
 		syncStatusTracker,
 		syncDeriver,
 		engineResetDeriver,
@@ -275,7 +283,7 @@ func (s *L2Verifier) ActL1HeadSignal(t Testing) {
 	head, err := s.l1.L1BlockRefByLabel(t.Ctx(), eth.Unsafe)
 	require.NoError(t, err)
 	s.synchronousEvents.Emit(status.L1UnsafeEvent{L1Unsafe: head})
-	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev rollup.Event) bool {
+	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev event.Event) bool {
 		x, ok := ev.(status.L1UnsafeEvent)
 		return ok && x.L1Unsafe == head
 	}, false))
@@ -286,7 +294,7 @@ func (s *L2Verifier) ActL1SafeSignal(t Testing) {
 	safe, err := s.l1.L1BlockRefByLabel(t.Ctx(), eth.Safe)
 	require.NoError(t, err)
 	s.synchronousEvents.Emit(status.L1SafeEvent{L1Safe: safe})
-	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev rollup.Event) bool {
+	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev event.Event) bool {
 		x, ok := ev.(status.L1SafeEvent)
 		return ok && x.L1Safe == safe
 	}, false))
@@ -297,14 +305,14 @@ func (s *L2Verifier) ActL1FinalizedSignal(t Testing) {
 	finalized, err := s.l1.L1BlockRefByLabel(t.Ctx(), eth.Finalized)
 	require.NoError(t, err)
 	s.synchronousEvents.Emit(finality.FinalizeL1Event{FinalizedL1: finalized})
-	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev rollup.Event) bool {
+	require.NoError(t, s.synchronousEvents.DrainUntil(func(ev event.Event) bool {
 		x, ok := ev.(finality.FinalizeL1Event)
 		return ok && x.FinalizedL1 == finalized
 	}, false))
 	require.Equal(t, finalized, s.syncStatus.SyncStatus().FinalizedL1)
 }
 
-func (s *L2Verifier) OnEvent(ev rollup.Event) {
+func (s *L2Verifier) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case rollup.L1TemporaryErrorEvent:
 		s.log.Warn("L1 temporary error", "err", x.Err)
@@ -323,13 +331,13 @@ func (s *L2Verifier) OnEvent(ev rollup.Event) {
 }
 
 func (s *L2Verifier) ActL2EventsUntilPending(t Testing, num uint64) {
-	s.ActL2EventsUntil(t, func(ev rollup.Event) bool {
+	s.ActL2EventsUntil(t, func(ev event.Event) bool {
 		x, ok := ev.(engine.PendingSafeUpdateEvent)
 		return ok && x.PendingSafe.Number == num
 	}, 1000, false)
 }
 
-func (s *L2Verifier) ActL2EventsUntil(t Testing, fn func(ev rollup.Event) bool, max int, excl bool) {
+func (s *L2Verifier) ActL2EventsUntil(t Testing, fn func(ev event.Event) bool, max int, excl bool) {
 	t.Helper()
 	if s.l2Building {
 		t.InvalidAction("cannot derive new data while building L2 block")

op-e2e/actions/plasma_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
 	"github.com/ethereum-optimism/optimism/op-node/node/safedb"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	plasma "github.com/ethereum-optimism/optimism/op-plasma"
 	"github.com/ethereum-optimism/optimism/op-plasma/bindings"
@@ -500,7 +501,7 @@ func TestPlasma_SequencerStalledMultiChallenges(gt *testing.T) {
 
 	// advance the pipeline until it errors out as it is still stuck
 	// on deriving the first commitment
-	a.sequencer.ActL2EventsUntil(t, func(ev rollup.Event) bool {
+	a.sequencer.ActL2EventsUntil(t, func(ev event.Event) bool {
 		x, ok := ev.(rollup.EngineTemporaryErrorEvent)
 		if ok {
 			require.ErrorContains(t, x.Err, "failed to fetch input data")

op-e2e/actions/sync_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	engine2 "github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/sources"
@@ -446,10 +447,7 @@ func TestBackupUnsafeReorgForkChoiceInputError(gt *testing.T) {
 
 	// B3 is invalid block
 	// NextAttributes is called
-	sequencer.ActL2EventsUntil(t, func(ev rollup.Event) bool {
-		_, ok := ev.(engine2.ProcessAttributesEvent)
-		return ok
-	}, 100, true)
+	sequencer.ActL2EventsUntil(t, event.Is[engine2.ProcessAttributesEvent], 100, true)
 	// mock forkChoiceUpdate error while restoring previous unsafe chain using backupUnsafe.
 	seqEng.ActL2RPCFail(t, eth.InputError{Inner: errors.New("mock L2 RPC error"), Code: eth.InvalidForkchoiceState})
 
@@ -582,20 +580,14 @@ func TestBackupUnsafeReorgForkChoiceNotInputError(gt *testing.T) {
 
 	// B3 is invalid block
 	// wait till attributes processing (excl.) before mocking errors
-	sequencer.ActL2EventsUntil(t, func(ev rollup.Event) bool {
-		_, ok := ev.(engine2.ProcessAttributesEvent)
-		return ok
-	}, 100, true)
+	sequencer.ActL2EventsUntil(t, event.Is[engine2.ProcessAttributesEvent], 100, true)
 
 	serverErrCnt := 2
 	for i := 0; i < serverErrCnt; i++ {
 		// mock forkChoiceUpdate failure while restoring previous unsafe chain using backupUnsafe.
 		seqEng.ActL2RPCFail(t, engine.GenericServerError)
 		// TryBackupUnsafeReorg is called - forkChoiceUpdate returns GenericServerError so retry
-		sequencer.ActL2EventsUntil(t, func(ev rollup.Event) bool {
-			_, ok := ev.(rollup.EngineTemporaryErrorEvent)
-			return ok
-		}, 100, false)
+		sequencer.ActL2EventsUntil(t, event.Is[rollup.EngineTemporaryErrorEvent], 100, false)
 		// backupUnsafeHead not emptied yet
 		require.Equal(t, targetUnsafeHeadHash, sequencer.L2BackupUnsafe().Hash)
 	}
@@ -981,11 +973,8 @@ func TestSpanBatchAtomicity_Consolidation(gt *testing.T) {
 	verifier.l2PipelineIdle = false
 	for !verifier.l2PipelineIdle {
 		// wait for next pending block
-		verifier.ActL2EventsUntil(t, func(ev rollup.Event) bool {
-			_, pending := ev.(engine2.PendingSafeUpdateEvent)
-			_, idle := ev.(derive.DeriverIdleEvent)
-			return pending || idle
-		}, 1000, false)
+		verifier.ActL2EventsUntil(t, event.Any(
+			event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent]), 1000, false)
 		if verifier.L2PendingSafe().Number < targetHeadNumber {
 			// If the span batch is not fully processed, the safe head must not advance.
 			require.Equal(t, verifier.L2Safe().Number, uint64(0))
@@ -1033,11 +1022,8 @@ func TestSpanBatchAtomicity_ForceAdvance(gt *testing.T) {
 	verifier.l2PipelineIdle = false
 	for !verifier.l2PipelineIdle {
 		// wait for next pending block
-		verifier.ActL2EventsUntil(t, func(ev rollup.Event) bool {
-			_, pending := ev.(engine2.PendingSafeUpdateEvent)
-			_, idle := ev.(derive.DeriverIdleEvent)
-			return pending || idle
-		}, 1000, false)
+		verifier.ActL2EventsUntil(t, event.Any(
+			event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent]), 1000, false)
 		if verifier.L2PendingSafe().Number < targetHeadNumber {
 			// If the span batch is not fully processed, the safe head must not advance.
 			require.Equal(t, verifier.L2Safe().Number, uint64(0))

op-node/metrics/metrics.go

@@ -39,6 +39,9 @@ type Metricer interface {
 	RecordSequencingError()
 	RecordPublishingError()
 	RecordDerivationError()
+	RecordEmittedEvent(name string)
+	RecordProcessedEvent(name string)
+	RecordEventsRateLimited()
 	RecordReceivedUnsafePayload(payload *eth.ExecutionPayloadEnvelope)
 	RecordRef(layer string, name string, num uint64, timestamp uint64, h common.Hash)
 	RecordL1Ref(name string, ref eth.L1BlockRef)
@@ -92,6 +95,11 @@ type Metrics struct {
 	SequencingErrors *metrics.Event
 	PublishingErrors *metrics.Event
 
+	EmittedEvents   *prometheus.CounterVec
+	ProcessedEvents *prometheus.CounterVec
+
+	EventsRateLimited *metrics.Event
+
 	DerivedBatches metrics.EventVec
 
 	P2PReqDurationSeconds *prometheus.HistogramVec
@@ -195,6 +203,24 @@ func NewMetrics(procName string) *Metrics {
 		SequencingErrors: metrics.NewEvent(factory, ns, "", "sequencing_errors", "sequencing errors"),
 		PublishingErrors: metrics.NewEvent(factory, ns, "", "publishing_errors", "p2p publishing errors"),
 
+		EmittedEvents: factory.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: ns,
+				Subsystem: "events",
+				Name:      "emitted",
+				Help:      "number of emitted events",
+			}, []string{"event_type"}),
+
+		ProcessedEvents: factory.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: ns,
+				Subsystem: "events",
+				Name:      "processed",
+				Help:      "number of processed events",
+			}, []string{"event_type"}),
+
+		EventsRateLimited: metrics.NewEvent(factory, ns, "events", "rate_limited", "events rate limiter hits"),
+
 		DerivedBatches: metrics.NewEventVec(factory, ns, "", "derived_batches", "derived batches", []string{"type"}),
 
 		SequencerInconsistentL1Origin: metrics.NewEvent(factory, ns, "", "sequencer_inconsistent_l1_origin", "events when the sequencer selects an inconsistent L1 origin"),
@@ -441,6 +467,18 @@ func (m *Metrics) RecordPublishingError() {
 	m.PublishingErrors.Record()
 }
 
+func (m *Metrics) RecordEmittedEvent(name string) {
+	m.EmittedEvents.WithLabelValues(name).Inc()
+}
+
+func (m *Metrics) RecordProcessedEvent(name string) {
+	m.ProcessedEvents.WithLabelValues(name).Inc()
+}
+
+func (m *Metrics) RecordEventsRateLimited() {
+	m.EventsRateLimited.Record()
+}
+
 func (m *Metrics) RecordDerivationError() {
 	m.DerivationErrors.Record()
 }
@@ -642,6 +680,15 @@ func (n *noopMetricer) RecordPublishingError() {
 func (n *noopMetricer) RecordDerivationError() {
 }
 
+func (n *noopMetricer) RecordEmittedEvent(name string) {
+}
+
+func (n *noopMetricer) RecordProcessedEvent(name string) {
+}
+
+func (n *noopMetricer) RecordEventsRateLimited() {
+}
+
 func (n *noopMetricer) RecordReceivedUnsafePayload(payload *eth.ExecutionPayloadEnvelope) {
 }
 

op-node/rollup/attributes/attributes.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -31,12 +32,12 @@ type AttributesHandler struct {
 
 	mu sync.Mutex
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 
 	attributes *derive.AttributesWithParent
 }
 
-func NewAttributesHandler(log log.Logger, cfg *rollup.Config, ctx context.Context, l2 L2, emitter rollup.EventEmitter) *AttributesHandler {
+func NewAttributesHandler(log log.Logger, cfg *rollup.Config, ctx context.Context, l2 L2, emitter event.Emitter) *AttributesHandler {
 	return &AttributesHandler{
 		log:        log,
 		cfg:        cfg,
@@ -47,7 +48,7 @@ func NewAttributesHandler(log log.Logger, cfg *rollup.Config, ctx context.Contex
 	}
 }
 
-func (eq *AttributesHandler) OnEvent(ev rollup.Event) {
+func (eq *AttributesHandler) OnEvent(ev event.Event) {
 	// Events may be concurrent in the future. Prevent unsafe concurrent modifications to the attributes.
 	eq.mu.Lock()
 	defer eq.mu.Unlock()

op-node/rollup/clsync/clsync.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -25,14 +26,14 @@ type CLSync struct {
 	cfg     *rollup.Config
 	metrics Metrics
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 
 	mu sync.Mutex
 
 	unsafePayloads *PayloadsQueue // queue of unsafe payloads, ordered by ascending block number, may have gaps and duplicates
 }
 
-func NewCLSync(log log.Logger, cfg *rollup.Config, metrics Metrics, emitter rollup.EventEmitter) *CLSync {
+func NewCLSync(log log.Logger, cfg *rollup.Config, metrics Metrics, emitter event.Emitter) *CLSync {
 	return &CLSync{
 		log:            log,
 		cfg:            cfg,
@@ -63,7 +64,7 @@ func (ev ReceivedUnsafePayloadEvent) String() string {
 	return "received-unsafe-payload"
 }
 
-func (eq *CLSync) OnEvent(ev rollup.Event) {
+func (eq *CLSync) OnEvent(ev event.Event) {
 	// Events may be concurrent in the future. Prevent unsafe concurrent modifications to the payloads queue.
 	eq.mu.Lock()
 	defer eq.mu.Unlock()

op-node/rollup/derive/deriver.go

@@ -6,6 +6,7 @@ import (
 	"io"
 
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -67,12 +68,12 @@ type PipelineDeriver struct {
 
 	ctx context.Context
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 
 	needAttributesConfirmation bool
 }
 
-func NewPipelineDeriver(ctx context.Context, pipeline *DerivationPipeline, emitter rollup.EventEmitter) *PipelineDeriver {
+func NewPipelineDeriver(ctx context.Context, pipeline *DerivationPipeline, emitter event.Emitter) *PipelineDeriver {
 	return &PipelineDeriver{
 		pipeline: pipeline,
 		ctx:      ctx,
@@ -80,7 +81,7 @@ func NewPipelineDeriver(ctx context.Context, pipeline *DerivationPipeline, emitt
 	}
 }
 
-func (d *PipelineDeriver) OnEvent(ev rollup.Event) {
+func (d *PipelineDeriver) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case rollup.ResetEvent:
 		d.pipeline.Reset()

op-node/rollup/driver/driver.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"time"
 
+	"golang.org/x/time/rate"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 
@@ -14,6 +16,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/conductor"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/finality"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/status"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
@@ -46,6 +49,8 @@ type Metrics interface {
 	EngineMetrics
 	L1FetcherMetrics
 	SequencerMetrics
+	event.Metrics
+	RecordEventsRateLimited()
 }
 
 type L1Chain interface {
@@ -93,7 +98,7 @@ type AttributesHandler interface {
 
 type Finalizer interface {
 	FinalizedL1() eth.L1BlockRef
-	rollup.Deriver
+	event.Deriver
 }
 
 type PlasmaIface interface {
@@ -106,7 +111,7 @@ type PlasmaIface interface {
 }
 
 type SyncStatusTracker interface {
-	rollup.Deriver
+	event.Deriver
 	SyncStatus() *eth.SyncStatus
 	L1Head() eth.L1BlockRef
 }
@@ -149,6 +154,14 @@ type SequencerStateListener interface {
 	SequencerStopped() error
 }
 
+// 10,000 events per second is plenty.
+// If we are going through more events, the driver needs to breathe, and warn the user of a potential issue.
+const eventsLimit = rate.Limit(10_000)
+
+// 500 events of burst: the maximum amount of events to eat up
+// past the rate limit before the rate limit becomes applicable.
+const eventsBurst = 500
+
 // NewDriver composes an events handler that tracks L1 state, triggers L2 Derivation, and optionally sequences new L2 blocks.
 func NewDriver(
 	driverCfg *Config,
@@ -167,8 +180,13 @@ func NewDriver(
 	plasma PlasmaIface,
 ) *Driver {
 	driverCtx, driverCancel := context.WithCancel(context.Background())
-	rootDeriver := &rollup.SynchronousDerivers{}
-	synchronousEvents := rollup.NewSynchronousEvents(log, driverCtx, rootDeriver)
+	rootDeriver := &event.DeriverMux{}
+	var synchronousEvents event.EmitterDrainer
+	synchronousEvents = event.NewQueue(log, driverCtx, rootDeriver, metrics)
+	synchronousEvents = event.NewLimiterDrainer(context.Background(), synchronousEvents, eventsLimit, eventsBurst, func() {
+		metrics.RecordEventsRateLimited()
+		log.Warn("Driver is hitting events rate limit.")
+	})
 
 	statusTracker := status.NewStatusTracker(log, metrics)
 
@@ -240,7 +258,7 @@ func NewDriver(
 		sequencerConductor: sequencerConductor,
 	}
 
-	*rootDeriver = []rollup.Deriver{
+	*rootDeriver = []event.Deriver{
 		syncDeriver,
 		engineResetDeriver,
 		engDeriv,

op-node/rollup/driver/state.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/conductor"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/finality"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/status"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
@@ -41,7 +42,7 @@ type Driver struct {
 
 	sched *StepSchedulingDeriver
 
-	synchronousEvents *rollup.SynchronousEvents
+	synchronousEvents event.EmitterDrainer
 
 	// Requests to block the event loop for synchronous execution to avoid reading an inconsistent state
 	stateReq chan chan struct{}
@@ -365,7 +366,7 @@ func (s *Driver) eventLoop() {
 // OnEvent handles broadcasted events.
 // The Driver itself is a deriver to catch system-critical events.
 // Other event-handling should be encapsulated into standalone derivers.
-func (s *Driver) OnEvent(ev rollup.Event) {
+func (s *Driver) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case rollup.CriticalErrorEvent:
 		s.Log.Error("Derivation process critical error", "err", x.Err)
@@ -381,7 +382,7 @@ func (s *Driver) OnEvent(ev rollup.Event) {
 	}
 }
 
-func (s *Driver) Emit(ev rollup.Event) {
+func (s *Driver) Emit(ev event.Event) {
 	s.synchronousEvents.Emit(ev)
 }
 
@@ -408,7 +409,7 @@ type SyncDeriver struct {
 	L1 L1Chain
 	L2 L2Chain
 
-	Emitter rollup.EventEmitter
+	Emitter event.Emitter
 
 	Log log.Logger
 
@@ -417,7 +418,7 @@ type SyncDeriver struct {
 	Drain func() error
 }
 
-func (s *SyncDeriver) OnEvent(ev rollup.Event) {
+func (s *SyncDeriver) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case StepEvent:
 		s.onStepEvent()
@@ -508,7 +509,8 @@ func (s *SyncDeriver) onStepEvent() {
 		s.Log.Error("Derivation process error", "err", err)
 		s.Emitter.Emit(StepReqEvent{})
 	} else {
-		s.Emitter.Emit(StepReqEvent{ResetBackoff: true}) // continue with the next step if we can
+		// Revisit SyncStep in 1/2 of a L2 block.
+		s.Emitter.Emit(StepDelayedReqEvent{Delay: (time.Duration(s.Config.BlockTime) * time.Second) / 2})
 	}
 }
 

op-node/rollup/driver/steps.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 
-	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
 )
 
@@ -16,6 +16,14 @@ func (ev ResetStepBackoffEvent) String() string {
 	return "reset-step-backoff"
 }
 
+type StepDelayedReqEvent struct {
+	Delay time.Duration
+}
+
+func (ev StepDelayedReqEvent) String() string {
+	return "step-delayed-req"
+}
+
 type StepReqEvent struct {
 	ResetBackoff bool
 }
@@ -61,10 +69,10 @@ type StepSchedulingDeriver struct {
 
 	log log.Logger
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 }
 
-func NewStepSchedulingDeriver(log log.Logger, emitter rollup.EventEmitter) *StepSchedulingDeriver {
+func NewStepSchedulingDeriver(log log.Logger, emitter event.Emitter) *StepSchedulingDeriver {
 	return &StepSchedulingDeriver{
 		stepAttempts:   0,
 		bOffStrategy:   retry.Exponential(),
@@ -88,7 +96,7 @@ func (s *StepSchedulingDeriver) NextDelayedStep() <-chan time.Time {
 	return s.delayedStepReq
 }
 
-func (s *StepSchedulingDeriver) OnEvent(ev rollup.Event) {
+func (s *StepSchedulingDeriver) OnEvent(ev event.Event) {
 	step := func() {
 		s.delayedStepReq = nil
 		select {
@@ -99,6 +107,10 @@ func (s *StepSchedulingDeriver) OnEvent(ev rollup.Event) {
 	}
 
 	switch x := ev.(type) {
+	case StepDelayedReqEvent:
+		if s.delayedStepReq == nil {
+			s.delayedStepReq = time.After(x.Delay)
+		}
 	case StepReqEvent:
 		if x.ResetBackoff {
 			s.stepAttempts = 0

op-node/rollup/driver/steps_test.go

@@ -7,14 +7,14 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 
-	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 )
 
 func TestStepSchedulingDeriver(t *testing.T) {
 	logger := testlog.Logger(t, log.LevelError)
-	var queued []rollup.Event
-	emitter := rollup.EmitterFunc(func(ev rollup.Event) {
+	var queued []event.Event
+	emitter := event.EmitterFunc(func(ev event.Event) {
 		queued = append(queued, ev)
 	})
 	sched := NewStepSchedulingDeriver(logger, emitter)
@@ -26,7 +26,7 @@ func TestStepSchedulingDeriver(t *testing.T) {
 	require.Empty(t, queued, "only scheduled so far, no step attempts yet")
 	<-sched.NextStep()
 	sched.OnEvent(StepAttemptEvent{})
-	require.Equal(t, []rollup.Event{StepEvent{}}, queued, "got step event")
+	require.Equal(t, []event.Event{StepEvent{}}, queued, "got step event")
 	require.Nil(t, sched.NextDelayedStep(), "no delayed steps yet")
 	sched.OnEvent(StepReqEvent{})
 	require.NotNil(t, sched.NextDelayedStep(), "2nd attempt before backoff reset causes delayed step to be scheduled")

op-node/rollup/engine/engine_controller.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/async"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/conductor"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/clock"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
@@ -54,7 +55,7 @@ type EngineController struct {
 	elStart    time.Time
 	clock      clock.Clock
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 
 	// Block Head State
 	unsafeHead       eth.L2BlockRef
@@ -78,7 +79,7 @@ type EngineController struct {
 }
 
 func NewEngineController(engine ExecEngine, log log.Logger, metrics derive.Metrics,
-	rollupCfg *rollup.Config, syncMode sync.Mode, emitter rollup.EventEmitter) *EngineController {
+	rollupCfg *rollup.Config, syncMode sync.Mode, emitter event.Emitter) *EngineController {
 	syncStatus := syncStatusCL
 	if syncMode == sync.ELSync {
 		syncStatus = syncStatusWillStartEL

op-node/rollup/engine/engine_reset.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 )
 
@@ -27,11 +28,11 @@ type EngineResetDeriver struct {
 	l2      sync.L2Chain
 	syncCfg *sync.Config
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 }
 
 func NewEngineResetDeriver(ctx context.Context, log log.Logger, cfg *rollup.Config,
-	l1 sync.L1Chain, l2 sync.L2Chain, syncCfg *sync.Config, emitter rollup.EventEmitter) *EngineResetDeriver {
+	l1 sync.L1Chain, l2 sync.L2Chain, syncCfg *sync.Config, emitter event.Emitter) *EngineResetDeriver {
 	return &EngineResetDeriver{
 		ctx:     ctx,
 		log:     log,
@@ -43,7 +44,7 @@ func NewEngineResetDeriver(ctx context.Context, log log.Logger, cfg *rollup.Conf
 	}
 }
 
-func (d *EngineResetDeriver) OnEvent(ev rollup.Event) {
+func (d *EngineResetDeriver) OnEvent(ev event.Event) {
 	switch ev.(type) {
 	case ResetEngineRequestEvent:
 		result, err := sync.FindL2Heads(d.ctx, d.cfg, d.l1, d.l2, d.log, d.syncCfg)

op-node/rollup/engine/events.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/async"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/conductor"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -148,13 +149,13 @@ type EngDeriver struct {
 	cfg     *rollup.Config
 	ec      *EngineController
 	ctx     context.Context
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 }
 
-var _ rollup.Deriver = (*EngDeriver)(nil)
+var _ event.Deriver = (*EngDeriver)(nil)
 
 func NewEngDeriver(log log.Logger, ctx context.Context, cfg *rollup.Config,
-	ec *EngineController, emitter rollup.EventEmitter) *EngDeriver {
+	ec *EngineController, emitter event.Emitter) *EngDeriver {
 	return &EngDeriver{
 		log:     log,
 		cfg:     cfg,
@@ -164,7 +165,7 @@ func NewEngDeriver(log log.Logger, ctx context.Context, cfg *rollup.Config,
 	}
 }
 
-func (d *EngDeriver) OnEvent(ev rollup.Event) {
+func (d *EngDeriver) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case TryBackupUnsafeReorgEvent:
 		// If we don't need to call FCU to restore unsafeHead using backupUnsafe, keep going b/c

op-node/rollup/events.go → op-node/rollup/event.go

 package rollup
 
-import "github.com/ethereum/go-ethereum/log"
-
-type Event interface {
-	String() string
-}
-
-type Deriver interface {
-	OnEvent(ev Event)
-}
-
-type EventEmitter interface {
-	Emit(ev Event)
-}
-
-type EmitterFunc func(ev Event)
-
-func (fn EmitterFunc) Emit(ev Event) {
-	fn(ev)
-}
+import "github.com/ethereum-optimism/optimism/op-node/rollup/event"
 
 // L1TemporaryErrorEvent identifies a temporary issue with the L1 data.
 type L1TemporaryErrorEvent struct {
 	Err error
 }
 
-var _ Event = L1TemporaryErrorEvent{}
+var _ event.Event = L1TemporaryErrorEvent{}
 
 func (ev L1TemporaryErrorEvent) String() string {
 	return "l1-temporary-error"
@@ -39,7 +21,7 @@ type EngineTemporaryErrorEvent struct {
 	Err error
 }
 
-var _ Event = EngineTemporaryErrorEvent{}
+var _ event.Event = EngineTemporaryErrorEvent{}
 
 func (ev EngineTemporaryErrorEvent) String() string {
 	return "engine-temporary-error"
@@ -49,7 +31,7 @@ type ResetEvent struct {
 	Err error
 }
 
-var _ Event = ResetEvent{}
+var _ event.Event = ResetEvent{}
 
 func (ev ResetEvent) String() string {
 	return "reset-event"
@@ -59,43 +41,8 @@ type CriticalErrorEvent struct {
 	Err error
 }
 
-var _ Event = CriticalErrorEvent{}
+var _ event.Event = CriticalErrorEvent{}
 
 func (ev CriticalErrorEvent) String() string {
 	return "critical-error"
 }
-
-type SynchronousDerivers []Deriver
-
-func (s *SynchronousDerivers) OnEvent(ev Event) {
-	for _, d := range *s {
-		d.OnEvent(ev)
-	}
-}
-
-var _ Deriver = (*SynchronousDerivers)(nil)
-
-type DebugDeriver struct {
-	Log log.Logger
-}
-
-func (d DebugDeriver) OnEvent(ev Event) {
-	d.Log.Debug("on-event", "event", ev)
-}
-
-type NoopDeriver struct{}
-
-func (d NoopDeriver) OnEvent(ev Event) {}
-
-// DeriverFunc implements the Deriver interface as a function,
-// similar to how the std-lib http HandlerFunc implements a Handler.
-// This can be used for small in-place derivers, test helpers, etc.
-type DeriverFunc func(ev Event)
-
-func (fn DeriverFunc) OnEvent(ev Event) {
-	fn(ev)
-}
-
-type NoopEmitter struct{}
-
-func (e NoopEmitter) Emit(ev Event) {}

op-node/rollup/event/events.go

+package event
+
+import "github.com/ethereum/go-ethereum/log"
+
+type Event interface {
+	// String returns the name of the event.
+	// The name must be simple and identify the event type, not the event content.
+	// This name is used for metric-labeling.
+	String() string
+}
+
+type Deriver interface {
+	OnEvent(ev Event)
+}
+
+type Emitter interface {
+	Emit(ev Event)
+}
+
+type Drainer interface {
+	// Drain processes all events.
+	Drain() error
+	// DrainUntil processes all events until a condition is hit.
+	// If excl, the event that matches the condition is not processed yet.
+	// If not excl, the event that matches is processed.
+	DrainUntil(fn func(ev Event) bool, excl bool) error
+}
+
+type EmitterDrainer interface {
+	Emitter
+	Drainer
+}
+
+type EmitterFunc func(ev Event)
+
+func (fn EmitterFunc) Emit(ev Event) {
+	fn(ev)
+}
+
+// DeriverMux takes an event-signal as deriver, and synchronously fans it out to all contained Deriver ends.
+// Technically this is a DeMux: single input to multi output.
+type DeriverMux []Deriver
+
+func (s *DeriverMux) OnEvent(ev Event) {
+	for _, d := range *s {
+		d.OnEvent(ev)
+	}
+}
+
+var _ Deriver = (*DeriverMux)(nil)
+
+type DebugDeriver struct {
+	Log log.Logger
+}
+
+func (d DebugDeriver) OnEvent(ev Event) {
+	d.Log.Debug("on-event", "event", ev)
+}
+
+type NoopDeriver struct{}
+
+func (d NoopDeriver) OnEvent(ev Event) {}
+
+// DeriverFunc implements the Deriver interface as a function,
+// similar to how the std-lib http HandlerFunc implements a Handler.
+// This can be used for small in-place derivers, test helpers, etc.
+type DeriverFunc func(ev Event)
+
+func (fn DeriverFunc) OnEvent(ev Event) {
+	fn(ev)
+}
+
+type NoopEmitter struct{}
+
+func (e NoopEmitter) Emit(ev Event) {}

op-node/rollup/events_test.go → op-node/rollup/event/events_test.go

-package rollup
+package event
 
 import (
 	"fmt"
@@ -13,7 +13,7 @@ func (ev TestEvent) String() string {
 	return "X"
 }
 
-func TestSynchronousDerivers_OnEvent(t *testing.T) {
+func TestDeriverMux_OnEvent(t *testing.T) {
 	result := ""
 	a := DeriverFunc(func(ev Event) {
 		result += fmt.Sprintf("A:%s\n", ev)
@@ -25,26 +25,26 @@ func TestSynchronousDerivers_OnEvent(t *testing.T) {
 		result += fmt.Sprintf("C:%s\n", ev)
 	})
 
-	x := SynchronousDerivers{}
+	x := DeriverMux{}
 	x.OnEvent(TestEvent{})
 	require.Equal(t, "", result)
 
-	x = SynchronousDerivers{a}
+	x = DeriverMux{a}
 	x.OnEvent(TestEvent{})
 	require.Equal(t, "A:X\n", result)
 
 	result = ""
-	x = SynchronousDerivers{a, a}
+	x = DeriverMux{a, a}
 	x.OnEvent(TestEvent{})
 	require.Equal(t, "A:X\nA:X\n", result)
 
 	result = ""
-	x = SynchronousDerivers{a, b}
+	x = DeriverMux{a, b}
 	x.OnEvent(TestEvent{})
 	require.Equal(t, "A:X\nB:X\n", result)
 
 	result = ""
-	x = SynchronousDerivers{a, b, c}
+	x = DeriverMux{a, b, c}
 	x.OnEvent(TestEvent{})
 	require.Equal(t, "A:X\nB:X\nC:X\n", result)
 }

op-node/rollup/event/limiter.go

+package event
+
+import (
+	"context"
+
+	"golang.org/x/time/rate"
+)
+
+type Limiter[E Emitter] struct {
+	ctx       context.Context
+	emitter   E
+	rl        *rate.Limiter
+	onLimited func()
+}
+
+// NewLimiter returns an event rate-limiter.
+// This can be used to prevent event loops from (accidentally) running too hot.
+// The eventRate is the number of events per second.
+// The eventBurst is the margin of events to eat into until the rate-limit kicks in.
+// The onLimited function is optional, and will be called if an emitted event is getting rate-limited
+func NewLimiter[E Emitter](ctx context.Context, em E, eventRate rate.Limit, eventBurst int, onLimited func()) *Limiter[E] {
+	return &Limiter[E]{
+		ctx:       ctx,
+		emitter:   em,
+		rl:        rate.NewLimiter(eventRate, eventBurst),
+		onLimited: onLimited,
+	}
+}
+
+// Emit is thread-safe, multiple parallel derivers can safely emit events to it.
+func (l *Limiter[E]) Emit(ev Event) {
+	if l.onLimited != nil && l.rl.Tokens() < 1.0 {
+		l.onLimited()
+	}
+	if err := l.rl.Wait(l.ctx); err != nil {
+		return // ctx error, safe to ignore.
+	}
+	l.emitter.Emit(ev)
+}
+
+// LimiterDrainer is a variant of Limiter that supports event draining.
+type LimiterDrainer Limiter[EmitterDrainer]
+
+func NewLimiterDrainer(ctx context.Context, em EmitterDrainer, eventRate rate.Limit, eventBurst int, onLimited func()) *LimiterDrainer {
+	return (*LimiterDrainer)(NewLimiter(ctx, em, eventRate, eventBurst, onLimited))
+}
+
+func (l *LimiterDrainer) Emit(ev Event) {
+	l.emitter.Emit(ev)
+}
+
+func (l *LimiterDrainer) Drain() error {
+	return l.emitter.Drain()
+}
+
+func (l *LimiterDrainer) DrainUntil(fn func(ev Event) bool, excl bool) error {
+	return l.emitter.DrainUntil(fn, excl)
+}

op-node/rollup/event/limiter_test.go

+package event
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
+)
+
+func TestLimiter(t *testing.T) {
+	count := uint64(0)
+	em := EmitterFunc(func(ev Event) {
+		count += 1
+	})
+	hitRateLimitAt := uint64(0)
+	// Test that we are able to hit the specified rate limit, and no earlier
+	lim := NewLimiter(context.Background(), em, rate.Limit(10), 10, func() {
+		if hitRateLimitAt != 0 {
+			return
+		}
+		hitRateLimitAt = count
+	})
+	for i := 0; i < 30; i++ {
+		lim.Emit(TestEvent{})
+	}
+	require.LessOrEqual(t, uint64(10), hitRateLimitAt)
+}

op-node/rollup/event/metrics.go

+package event
+
+type Metrics interface {
+	RecordEmittedEvent(name string)
+	RecordProcessedEvent(name string)
+}
+
+type NoopMetrics struct {
+}
+
+func (n NoopMetrics) RecordEmittedEvent(name string) {}
+
+func (n NoopMetrics) RecordProcessedEvent(name string) {}
+
+var _ Metrics = NoopMetrics{}

op-node/rollup/synchronous.go → op-node/rollup/event/queue.go

-package rollup
+package event
 
 import (
 	"context"
@@ -12,11 +12,11 @@ import (
 // At some point it's better to drop events and warn something is exploding the number of events.
 const sanityEventLimit = 1000
 
-// SynchronousEvents is a rollup.EventEmitter that a rollup.Deriver can emit events to.
+// Queue is a event.Emitter that a event.Deriver can emit events to.
 // The events will be queued up, and can then be executed synchronously by calling the Drain function,
 // which will apply all events to the root Deriver.
 // New events may be queued up while events are being processed by the root rollup.Deriver.
-type SynchronousEvents struct {
+type Queue struct {
 	// The lock is no-op in FP execution, if running in synchronous FP-VM.
 	// This lock ensures that all emitted events are merged together correctly,
 	// if this util is used in a concurrent context.
@@ -29,20 +29,28 @@ type SynchronousEvents struct {
 	ctx context.Context
 
 	root Deriver
+
+	metrics Metrics
 }
 
-func NewSynchronousEvents(log log.Logger, ctx context.Context, root Deriver) *SynchronousEvents {
-	return &SynchronousEvents{
-		log:  log,
-		ctx:  ctx,
-		root: root,
+var _ EmitterDrainer = (*Queue)(nil)
+
+func NewQueue(log log.Logger, ctx context.Context, root Deriver, metrics Metrics) *Queue {
+	return &Queue{
+		log:     log,
+		ctx:     ctx,
+		root:    root,
+		metrics: metrics,
 	}
 }
 
-func (s *SynchronousEvents) Emit(event Event) {
+func (s *Queue) Emit(event Event) {
 	s.evLock.Lock()
 	defer s.evLock.Unlock()
 
+	s.log.Debug("Emitting event", "event", event)
+	s.metrics.RecordEmittedEvent(event.String())
+
 	if s.ctx.Err() != nil {
 		s.log.Warn("Ignoring emitted event during shutdown", "event", event)
 		return
@@ -56,7 +64,7 @@ func (s *SynchronousEvents) Emit(event Event) {
 	s.events = append(s.events, event)
 }
 
-func (s *SynchronousEvents) Drain() error {
+func (s *Queue) Drain() error {
 	for {
 		if s.ctx.Err() != nil {
 			return s.ctx.Err()
@@ -70,11 +78,13 @@ func (s *SynchronousEvents) Drain() error {
 		s.events = s.events[1:]
 		s.evLock.Unlock()
 
+		s.log.Debug("Processing event", "event", first)
 		s.root.OnEvent(first)
+		s.metrics.RecordProcessedEvent(first.String())
 	}
 }
 
-func (s *SynchronousEvents) DrainUntil(fn func(ev Event) bool, excl bool) error {
+func (s *Queue) DrainUntil(fn func(ev Event) bool, excl bool) error {
 	for {
 		if s.ctx.Err() != nil {
 			return s.ctx.Err()
@@ -93,11 +103,13 @@ func (s *SynchronousEvents) DrainUntil(fn func(ev Event) bool, excl bool) error
 		s.events = s.events[1:]
 		s.evLock.Unlock()
 
+		s.log.Debug("Processing event", "event", first)
 		s.root.OnEvent(first)
+		s.metrics.RecordProcessedEvent(first.String())
 		if stop {
 			return nil
 		}
 	}
 }
 
-var _ EventEmitter = (*SynchronousEvents)(nil)
+var _ Emitter = (*Queue)(nil)

op-node/rollup/synchronous_test.go → op-node/rollup/event/queue_test.go

-package rollup
+package event
 
 import (
 	"context"
@@ -11,14 +11,14 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 )
 
-func TestSynchronousEvents(t *testing.T) {
+func TestQueue(t *testing.T) {
 	logger := testlog.Logger(t, log.LevelError)
 	ctx, cancel := context.WithCancel(context.Background())
 	count := 0
 	deriver := DeriverFunc(func(ev Event) {
 		count += 1
 	})
-	syncEv := NewSynchronousEvents(logger, ctx, deriver)
+	syncEv := NewQueue(logger, ctx, deriver, NoopMetrics{})
 	require.NoError(t, syncEv.Drain(), "can drain, even if empty")
 
 	syncEv.Emit(TestEvent{})
@@ -38,13 +38,13 @@ func TestSynchronousEvents(t *testing.T) {
 	require.Equal(t, 3, count, "didn't process event after trigger close")
 }
 
-func TestSynchronousEventsSanityLimit(t *testing.T) {
-	logger := testlog.Logger(t, log.LevelError)
+func TestQueueSanityLimit(t *testing.T) {
+	logger := testlog.Logger(t, log.LevelCrit) // expecting error log of hitting sanity limit
 	count := 0
 	deriver := DeriverFunc(func(ev Event) {
 		count += 1
 	})
-	syncEv := NewSynchronousEvents(logger, context.Background(), deriver)
+	syncEv := NewQueue(logger, context.Background(), deriver, NoopMetrics{})
 	// emit 1 too many events
 	for i := 0; i < sanityEventLimit+1; i++ {
 		syncEv.Emit(TestEvent{})
@@ -67,7 +67,7 @@ func (ev CyclicEvent) String() string {
 
 func TestSynchronousCyclic(t *testing.T) {
 	logger := testlog.Logger(t, log.LevelError)
-	var emitter EventEmitter
+	var emitter Emitter
 	result := false
 	deriver := DeriverFunc(func(ev Event) {
 		logger.Info("received event", "event", ev)
@@ -80,7 +80,7 @@ func TestSynchronousCyclic(t *testing.T) {
 			}
 		}
 	})
-	syncEv := NewSynchronousEvents(logger, context.Background(), deriver)
+	syncEv := NewQueue(logger, context.Background(), deriver, NoopMetrics{})
 	emitter = syncEv
 	syncEv.Emit(CyclicEvent{Count: 0})
 	require.NoError(t, syncEv.Drain())

op-node/rollup/event/util.go

+package event
+
+// Is as helper function is syntax-sugar to do an Event type check as a boolean function
+func Is[T Event](ev Event) bool {
+	_, ok := ev.(T)
+	return ok
+}
+
+// Any as helper function combines different event conditions into a single function
+func Any(fns ...func(ev Event) bool) func(ev Event) bool {
+	return func(ev Event) bool {
+		for _, fn := range fns {
+			if fn(ev) {
+				return true
+			}
+		}
+		return false
+	}
+}

op-node/rollup/event/util_test.go

+package event
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type FooEvent struct{}
+
+func (ev FooEvent) String() string {
+	return "foo"
+}
+
+type BarEvent struct{}
+
+func (ev BarEvent) String() string {
+	return "bar"
+}
+
+func TestIs(t *testing.T) {
+	require.False(t, Is[TestEvent](FooEvent{}))
+	require.False(t, Is[TestEvent](BarEvent{}))
+	require.True(t, Is[FooEvent](FooEvent{}))
+	require.True(t, Is[BarEvent](BarEvent{}))
+}
+
+func TestAny(t *testing.T) {
+	require.False(t, Any()(FooEvent{}))
+	require.False(t, Any(Is[BarEvent])(FooEvent{}))
+	require.True(t, Any(Is[FooEvent])(FooEvent{}))
+	require.False(t, Any(Is[TestEvent], Is[BarEvent])(FooEvent{}))
+	require.True(t, Any(Is[TestEvent], Is[BarEvent], Is[FooEvent])(FooEvent{}))
+	require.True(t, Any(Is[FooEvent], Is[BarEvent], Is[TestEvent])(FooEvent{}))
+	require.True(t, Any(Is[FooEvent], Is[FooEvent], Is[FooEvent])(FooEvent{}))
+	require.False(t, Any(Is[FooEvent], Is[FooEvent], Is[FooEvent])(BarEvent{}))
+}

op-node/rollup/finality/finalizer.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
@@ -72,7 +73,7 @@ type Finalizer struct {
 
 	ctx context.Context
 
-	emitter rollup.EventEmitter
+	emitter event.Emitter
 
 	// finalizedL1 is the currently perceived finalized L1 block.
 	// This may be ahead of the current traversed origin when syncing.
@@ -93,7 +94,7 @@ type Finalizer struct {
 	l1Fetcher FinalizerL1Interface
 }
 
-func NewFinalizer(ctx context.Context, log log.Logger, cfg *rollup.Config, l1Fetcher FinalizerL1Interface, emitter rollup.EventEmitter) *Finalizer {
+func NewFinalizer(ctx context.Context, log log.Logger, cfg *rollup.Config, l1Fetcher FinalizerL1Interface, emitter event.Emitter) *Finalizer {
 	lookback := calcFinalityLookback(cfg)
 	return &Finalizer{
 		ctx:              ctx,
@@ -130,7 +131,7 @@ func (ev TryFinalizeEvent) String() string {
 	return "try-finalize"
 }
 
-func (fi *Finalizer) OnEvent(ev rollup.Event) {
+func (fi *Finalizer) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case FinalizeL1Event:
 		fi.onL1Finalized(x.FinalizedL1)

op-node/rollup/finality/plasma.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	plasma "github.com/ethereum-optimism/optimism/op-plasma"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
@@ -27,7 +28,7 @@ type PlasmaFinalizer struct {
 }
 
 func NewPlasmaFinalizer(ctx context.Context, log log.Logger, cfg *rollup.Config,
-	l1Fetcher FinalizerL1Interface, emitter rollup.EventEmitter,
+	l1Fetcher FinalizerL1Interface, emitter event.Emitter,
 	backend PlasmaBackend) *PlasmaFinalizer {
 
 	inner := NewFinalizer(ctx, log, cfg, l1Fetcher, emitter)
@@ -45,7 +46,7 @@ func NewPlasmaFinalizer(ctx context.Context, log log.Logger, cfg *rollup.Config,
 	}
 }
 
-func (fi *PlasmaFinalizer) OnEvent(ev rollup.Event) {
+func (fi *PlasmaFinalizer) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case FinalizeL1Event:
 		fi.backend.Finalize(x.FinalizedL1)

op-node/rollup/finality/plasma_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	plasma "github.com/ethereum-optimism/optimism/op-plasma"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
@@ -134,7 +135,7 @@ func TestPlasmaFinalityData(t *testing.T) {
 			emitter.AssertExpectations(t)
 		}
 		// might trigger finalization attempt, if expired finality delay
-		emitter.ExpectMaybeRun(func(ev rollup.Event) {
+		emitter.ExpectMaybeRun(func(ev event.Event) {
 			require.IsType(t, TryFinalizeEvent{}, ev)
 		})
 		fi.OnEvent(derive.DeriverIdleEvent{})
@@ -166,7 +167,7 @@ func TestPlasmaFinalityData(t *testing.T) {
 			l1F.ExpectL1BlockRefByNumber(commitmentInclusionFinalized.Number, commitmentInclusionFinalized, nil)
 			l1F.ExpectL1BlockRefByNumber(commitmentInclusionFinalized.Number, commitmentInclusionFinalized, nil)
 			var finalizedL2 eth.L2BlockRef
-			emitter.ExpectOnceRun(func(ev rollup.Event) {
+			emitter.ExpectOnceRun(func(ev event.Event) {
 				if x, ok := ev.(engine.PromoteFinalizedEvent); ok {
 					finalizedL2 = x.Ref
 				} else {

op-node/rollup/status/status.go

@@ -9,6 +9,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/finality"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
@@ -56,7 +57,7 @@ func NewStatusTracker(log log.Logger, metrics Metrics) *StatusTracker {
 	return st
 }
 
-func (st *StatusTracker) OnEvent(ev rollup.Event) {
+func (st *StatusTracker) OnEvent(ev event.Event) {
 	st.mu.Lock()
 	defer st.mu.Unlock()
 

op-program/client/driver/driver.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	plasma "github.com/ethereum-optimism/optimism/op-plasma"
 )
@@ -22,10 +23,10 @@ type EndCondition interface {
 type Driver struct {
 	logger log.Logger
 
-	events []rollup.Event
+	events []event.Event
 
 	end     EndCondition
-	deriver rollup.Deriver
+	deriver event.Deriver
 }
 
 func NewDriver(logger log.Logger, cfg *rollup.Config, l1Source derive.L1Fetcher,
@@ -51,7 +52,7 @@ func NewDriver(logger log.Logger, cfg *rollup.Config, l1Source derive.L1Fetcher,
 		targetBlockNum: targetBlockNum,
 	}
 
-	d.deriver = &rollup.SynchronousDerivers{
+	d.deriver = &event.DeriverMux{
 		prog,
 		engineDeriv,
 		pipelineDeriver,
@@ -62,7 +63,7 @@ func NewDriver(logger log.Logger, cfg *rollup.Config, l1Source derive.L1Fetcher,
 	return d
 }
 
-func (d *Driver) Emit(ev rollup.Event) {
+func (d *Driver) Emit(ev event.Event) {
 	if d.end.Closing() {
 		return
 	}

op-program/client/driver/driver_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 
-	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 )
 
@@ -26,21 +26,21 @@ func (d *fakeEnd) Result() error {
 }
 
 func TestDriver(t *testing.T) {
-	newTestDriver := func(t *testing.T, onEvent func(d *Driver, end *fakeEnd, ev rollup.Event)) *Driver {
+	newTestDriver := func(t *testing.T, onEvent func(d *Driver, end *fakeEnd, ev event.Event)) *Driver {
 		logger := testlog.Logger(t, log.LevelInfo)
 		end := &fakeEnd{}
 		d := &Driver{
 			logger: logger,
 			end:    end,
 		}
-		d.deriver = rollup.DeriverFunc(func(ev rollup.Event) {
+		d.deriver = event.DeriverFunc(func(ev event.Event) {
 			onEvent(d, end, ev)
 		})
 		return d
 	}
 
 	t.Run("insta complete", func(t *testing.T) {
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			end.closing = true
 		})
 		require.NoError(t, d.RunComplete())
@@ -48,7 +48,7 @@ func TestDriver(t *testing.T) {
 
 	t.Run("insta error", func(t *testing.T) {
 		mockErr := errors.New("mock error")
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			end.closing = true
 			end.result = mockErr
 		})
@@ -57,7 +57,7 @@ func TestDriver(t *testing.T) {
 
 	t.Run("success after a few events", func(t *testing.T) {
 		count := 0
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			if count > 3 {
 				end.closing = true
 				return
@@ -71,7 +71,7 @@ func TestDriver(t *testing.T) {
 	t.Run("error after a few events", func(t *testing.T) {
 		count := 0
 		mockErr := errors.New("mock error")
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			if count > 3 {
 				end.closing = true
 				end.result = mockErr
@@ -85,7 +85,7 @@ func TestDriver(t *testing.T) {
 
 	t.Run("exhaust events", func(t *testing.T) {
 		count := 0
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			if count < 3 { // stop generating events after a while, without changing end condition
 				d.Emit(TestEvent{})
 			}
@@ -96,7 +96,7 @@ func TestDriver(t *testing.T) {
 
 	t.Run("queued events", func(t *testing.T) {
 		count := 0
-		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev rollup.Event) {
+		d := newTestDriver(t, func(d *Driver, end *fakeEnd, ev event.Event) {
 			if count < 3 {
 				d.Emit(TestEvent{})
 				d.Emit(TestEvent{})

op-program/client/driver/program.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 )
 
 // ProgramDeriver expresses how engine and derivation events are
@@ -17,7 +18,7 @@ import (
 type ProgramDeriver struct {
 	logger log.Logger
 
-	Emitter rollup.EventEmitter
+	Emitter event.Emitter
 
 	closing        bool
 	result         error
@@ -32,7 +33,7 @@ func (d *ProgramDeriver) Result() error {
 	return d.result
 }
 
-func (d *ProgramDeriver) OnEvent(ev rollup.Event) {
+func (d *ProgramDeriver) OnEvent(ev event.Event) {
 	switch x := ev.(type) {
 	case engine.EngineResetConfirmedEvent:
 		d.Emitter.Emit(derive.ConfirmPipelineResetEvent{})

op-service/testutils/mock_emitter.go

 package testutils
 
 import (
+	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/stretchr/testify/mock"
-
-	"github.com/ethereum-optimism/optimism/op-node/rollup"
 )
 
 type MockEmitter struct {
 	mock.Mock
 }
 
-func (m *MockEmitter) Emit(ev rollup.Event) {
+func (m *MockEmitter) Emit(ev event.Event) {
 	m.Mock.MethodCalled("Emit", ev)
 }
 
-func (m *MockEmitter) ExpectOnce(expected rollup.Event) {
+func (m *MockEmitter) ExpectOnce(expected event.Event) {
 	m.Mock.On("Emit", expected).Once()
 }
 
-func (m *MockEmitter) ExpectMaybeRun(fn func(ev rollup.Event)) {
+func (m *MockEmitter) ExpectMaybeRun(fn func(ev event.Event)) {
 	m.Mock.On("Emit", mock.Anything).Maybe().Run(func(args mock.Arguments) {
-		fn(args.Get(0).(rollup.Event))
+		fn(args.Get(0).(event.Event))
 	})
 }
 
@@ -28,9 +27,9 @@ func (m *MockEmitter) ExpectOnceType(typ string) {
 	m.Mock.On("Emit", mock.AnythingOfType(typ)).Once()
 }
 
-func (m *MockEmitter) ExpectOnceRun(fn func(ev rollup.Event)) {
+func (m *MockEmitter) ExpectOnceRun(fn func(ev event.Event)) {
 	m.Mock.On("Emit", mock.Anything).Once().Run(func(args mock.Arguments) {
-		fn(args.Get(0).(rollup.Event))
+		fn(args.Get(0).(event.Event))
 	})
 }
 
@@ -38,4 +37,4 @@ func (m *MockEmitter) AssertExpectations(t mock.TestingT) {
 	m.Mock.AssertExpectations(t)
 }
 
-var _ rollup.EventEmitter = (*MockEmitter)(nil)
+var _ event.Emitter = (*MockEmitter)(nil)