op-node: handle crit error events, lift event-system out of driver

op-node/node/node.go

@@ -8,13 +8,11 @@ import (
  "sync/atomic"
  "time"
 
- "github.com/ethereum-optimism/optimism/op-node/rollup/sequencing"
-
  "github.com/hashicorp/go-multierror"
  "github.com/libp2p/go-libp2p/core/peer"
 
  "github.com/ethereum/go-ethereum"
- "github.com/ethereum/go-ethereum/event"
+ gethevent "github.com/ethereum/go-ethereum/event"
  "github.com/ethereum/go-ethereum/log"
 
  altda "github.com/ethereum-optimism/optimism/op-alt-da"
@@ -24,6 +22,8 @@ import (
  "github.com/ethereum-optimism/optimism/op-node/rollup"
  "github.com/ethereum-optimism/optimism/op-node/rollup/conductor"
  "github.com/ethereum-optimism/optimism/op-node/rollup/driver"
+ "github.com/ethereum-optimism/optimism/op-node/rollup/event"
+ "github.com/ethereum-optimism/optimism/op-node/rollup/sequencing"
  "github.com/ethereum-optimism/optimism/op-node/rollup/sync"
  "github.com/ethereum-optimism/optimism/op-service/client"
  "github.com/ethereum-optimism/optimism/op-service/eth"
@@ -52,6 +52,9 @@ type OpNode struct {
  l1SafeSub ethereum.Subscription // Subscription to get L1 safe blocks, a.k.a. justified data (polling)
  l1FinalizedSub ethereum.Subscription // Subscription to get L1 safe blocks, a.k.a. justified data (polling)
 
+ eventSys event.System
+ eventDrain event.Drainer
+
  l1Source *sources.L1Client // L1 Client to fetch data from
  l2Driver *driver.Driver // L2 Engine to Sync
  l2Source *sources.EngineClient // L2 Execution Engine RPC bindings
@@ -126,6 +129,7 @@ func (n *OpNode) init(ctx context.Context, cfg *Config) error {
  if err := n.initTracer(ctx, cfg); err != nil {
  return fmt.Errorf("failed to init the trace: %w", err)
  }
+ n.initEventSystem()
  if err := n.initL1(ctx, cfg); err != nil {
  return fmt.Errorf("failed to init L1: %w", err)
  }
@@ -159,6 +163,16 @@ func (n *OpNode) init(ctx context.Context, cfg *Config) error {
  return nil
 }
 
+func (n *OpNode) initEventSystem() {
+ // This executor will be configurable in the future, for parallel event processing
+ executor := event.NewGlobalSynchronous(n.resourcesCtx)
+ sys := event.NewSystem(n.log, executor)
+ sys.AddTracer(event.NewMetricsTracer(n.metrics))
+ sys.Register("node", event.DeriverFunc(n.onEvent), event.DefaultRegisterOpts())
+ n.eventSys = sys
+ n.eventDrain = executor
+}
+
 func (n *OpNode) initTracer(ctx context.Context, cfg *Config) error {
  if cfg.Tracer != nil {
  n.tracer = cfg.Tracer
@@ -185,7 +199,7 @@ func (n *OpNode) initL1(ctx context.Context, cfg *Config) error {
  }
 
  // Keep subscribed to the L1 heads, which keeps the L1 maintainer pointing to the best headers to sync
- n.l1HeadsSub = event.ResubscribeErr(time.Second*10, func(ctx context.Context, err error) (event.Subscription, error) {
+ n.l1HeadsSub = gethevent.ResubscribeErr(time.Second*10, func(ctx context.Context, err error) (gethevent.Subscription, error) {
  if err != nil {
  n.log.Warn("resubscribing after failed L1 subscription", "err", err)
  }
@@ -410,7 +424,7 @@ func (n *OpNode) initL2(ctx context.Context, cfg *Config) error {
  } else {
  n.safeDB = safedb.Disabled
  }
- n.l2Driver = driver.NewDriver(&cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source,
+ n.l2Driver = driver.NewDriver(n.eventSys, n.eventDrain, &cfg.Driver, &cfg.Rollup, n.l2Source, n.l1Source,
  n.supervisor, n.beacon, n, n, n.log, n.metrics, cfg.ConfigPersistence, n.safeDB, &cfg.Sync, sequencerConductor, altDA)
  return nil
 }
@@ -509,6 +523,20 @@ func (n *OpNode) Start(ctx context.Context) error {
  return nil
 }
 
+// onEvent handles broadcast events.
+// The OpNode itself is a deriver to catch system-critical events.
+// Other event-handling should be encapsulated into standalone derivers.
+func (n *OpNode) onEvent(ev event.Event) bool {
+ switch x := ev.(type) {
+ case rollup.CriticalErrorEvent:
+ n.log.Error("Critical error", "err", x.Err)
+ n.cancel(fmt.Errorf("critical error: %w", x.Err))
+ return true
+ default:
+ return false
+ }
+}
+
 func (n *OpNode) OnNewL1Head(ctx context.Context, sig eth.L1BlockRef) {
  n.tracer.OnNewL1Head(ctx, sig)
 
@@ -679,6 +707,10 @@ func (n *OpNode) Stop(ctx context.Context) error {
  }
  }
 
+ if n.eventSys != nil {
+ n.eventSys.Stop()
+ }
+
  if n.safeDB != nil {
  if err := n.safeDB.Close(); err != nil {
  result = multierror.Append(result, fmt.Errorf("failed to close safe head db: %w", err))

op-node/rollup/driver/driver.go

@@ -150,8 +150,14 @@ type SequencerStateListener interface {
  SequencerStopped() error
 }
 
+type Drain interface {
+ Drain() error
+}
+
 // NewDriver composes an events handler that tracks L1 state, triggers L2 Derivation, and optionally sequences new L2 blocks.
 func NewDriver(
+ sys event.Registry,
+ drain Drain,
  driverCfg *Config,
  cfg *rollup.Config,
  l2 L2Chain,
@@ -170,17 +176,6 @@ func NewDriver(
 ) *Driver {
  driverCtx, driverCancel := context.WithCancel(context.Background())
 
- var executor event.Executor
- var drain func() error
- // This instantiation will be one of more options: soon there will be a parallel events executor
- {
- s := event.NewGlobalSynchronous(driverCtx)
- executor = s
- drain = s.Drain
- }
- sys := event.NewSystem(log, executor)
- sys.AddTracer(event.NewMetricsTracer(metrics))
-
  opts := event.DefaultRegisterOpts()
 
  // If interop is scheduled we start the driver.
@@ -236,7 +231,7 @@ func NewDriver(
  L2: l2,
  Log: log,
  Ctx: driverCtx,
- Drain: drain,
+ Drain: drain.Drain,
  }
  sys.Register("sync", syncDeriver, opts)
 
@@ -260,12 +255,11 @@ func NewDriver(
 
  driverEmitter := sys.Register("driver", nil, opts)
  driver := &Driver{
- eventSys: sys,
  statusTracker: statusTracker,
  SyncDeriver: syncDeriver,
  sched: schedDeriv,
  emitter: driverEmitter,
- drain: drain,
+ drain: drain.Drain,
  stateReq: make(chan chan struct{}),
  forceReset: make(chan chan struct{}, 10),
  driverConfig: driverCfg,

op-node/rollup/driver/state.go

@@ -26,8 +26,6 @@ import (
 type SyncStatus = eth.SyncStatus
 
 type Driver struct {
- eventSys event.System
-
  statusTracker SyncStatusTracker
 
  *SyncDeriver
@@ -100,7 +98,6 @@ func (s *Driver) Start() error {
 func (s *Driver) Close() error {
  s.driverCancel()
  s.wg.Wait()
- s.eventSys.Stop()
  s.sequencer.Close()
  return nil
 }
@@ -282,27 +279,6 @@ 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 event.Event) bool {
- switch x := ev.(type) {
- case rollup.CriticalErrorEvent:
- s.Log.Error("Derivation process critical error", "err", x.Err)
- // we need to unblock event-processing to be able to close
- go func() {
- logger := s.Log
- err := s.Close()
- if err != nil {
- logger.Error("Failed to shutdown driver on critical error", "err", err)
- }
- }()
- return true
- default:
- return false
- }
-}
-
 type SyncDeriver struct {
  // The derivation pipeline is reset whenever we reorg.
  // The derivation pipeline determines the new l2Safe.

op-node/rollup/event.go

@@ -37,12 +37,5 @@ func (ev ResetEvent) String() string {
  return "reset-event"
 }
 
-type CriticalErrorEvent struct {
- Err error
-}
-
-var _ event.Event = CriticalErrorEvent{}
-
-func (ev CriticalErrorEvent) String() string {
- return "critical-error"
-}
+// CriticalErrorEvent is an alias for event.CriticalErrorEvent
+type CriticalErrorEvent = event.CriticalErrorEvent

op-node/rollup/event/events.go

@@ -75,3 +75,13 @@ func (fn DeriverFunc) OnEvent(ev Event) bool {
 type NoopEmitter struct{}
 
 func (e NoopEmitter) Emit(ev Event) {}
+
+type CriticalErrorEvent struct {
+ Err error
+}
+
+var _ Event = CriticalErrorEvent{}
+
+func (ev CriticalErrorEvent) String() string {
+ return "critical-error"
+}

op-node/rollup/event/system.go

@@ -11,7 +11,7 @@ import (
  "github.com/ethereum/go-ethereum/log"
 )
 
-type System interface {
+type Registry interface {
  // Register registers a named event-emitter, optionally processing events itself:
  // deriver may be nil, not all registrants have to process events.
  // A non-nil deriver may implement AttachEmitter to automatically attach the Emitter to it,
@@ -20,6 +20,10 @@ type System interface {
  // Unregister removes a named emitter,
  // also removing it from the set of events-receiving derivers (if registered with non-nil deriver).
  Unregister(name string) (old Emitter)
+}
+
+type System interface {
+ Registry
  // AddTracer registers a tracer to capture all event deriver/emitter work. It runs until RemoveTracer is called.
  // Duplicate tracers are allowed.
  AddTracer(t Tracer)
@@ -73,6 +77,10 @@ func (r *systemActor) RunEvent(ev AnnotatedEvent) {
  if r.ctx.Err() != nil {
  return
  }
+ if r.sys.abort.Load() && !Is[CriticalErrorEvent](ev.Event) {
+ // if aborting, and not the CriticalErrorEvent itself, then do not process the event
+ return
+ }
 
  prev := r.currentEvent
  start := time.Now()
@@ -99,6 +107,9 @@ type Sys struct {
 
  tracers []Tracer
  tracersLock sync.RWMutex
+
+ // if true, no events may be processed, except CriticalError itself
+ abort atomic.Bool
 }
 
 func NewSystem(log log.Logger, ex Executor) *Sys {
@@ -240,6 +251,12 @@ func (s *Sys) emit(name string, derivContext uint64, ev Event) {
  emitContext := s.emitContext.Add(1)
  annotated := AnnotatedEvent{Event: ev, EmitContext: emitContext}
 
+ // As soon as anything emits a critical event,
+ // make the system aware, before the executor event schedules it for processing.
+ if Is[CriticalErrorEvent](ev) {
+ s.abort.Store(true)
+ }
+
  emitTime := time.Now()
  s.recordEmit(name, annotated, derivContext, emitTime)
 

op-node/rollup/event/system_test.go

@@ -2,6 +2,7 @@ package event
 
 import (
  "context"
+ "errors"
  "testing"
 
  "github.com/stretchr/testify/require"
@@ -104,3 +105,46 @@ func TestSystemBroadcast(t *testing.T) {
  require.Equal(t, 3, fooCount)
  require.Equal(t, 3, barCount)
 }
+
+func TestCriticalError(t *testing.T) {
+ logger := testlog.Logger(t, log.LevelError)
+ count := 0
+ seenCrit := 0
+ deriverFn := DeriverFunc(func(ev Event) bool {
+ switch ev.(type) {
+ case CriticalErrorEvent:
+ seenCrit += 1
+ default:
+ count += 1
+ }
+ return true
+ })
+ exec := NewGlobalSynchronous(context.Background())
+ sys := NewSystem(logger, exec)
+ emitterA := sys.Register("a", deriverFn, DefaultRegisterOpts())
+ emitterB := sys.Register("b", deriverFn, DefaultRegisterOpts())
+
+ require.NoError(t, exec.Drain(), "can drain, even if empty")
+ emitterA.Emit(TestEvent{})
+ require.Equal(t, 0, count, "no processing yet, queued event")
+ require.NoError(t, exec.Drain())
+ require.Equal(t, 2, count, "both A and B processed the event")
+
+ emitterA.Emit(TestEvent{})
+ emitterB.Emit(TestEvent{})
+ testErr := errors.New("test crit error")
+ emitterB.Emit(CriticalErrorEvent{Err: testErr})
+ require.Equal(t, 2, count, "no processing yet, queued events")
+ require.Equal(t, 0, seenCrit, "critical error events are still scheduled like normal")
+ require.True(t, sys.abort.Load(), "we are aware of the crit")
+ require.NoError(t, exec.Drain())
+ require.Equal(t, 2, count, "still no processing, since we hit a crit error, the events are ignored")
+ require.Equal(t, 2, seenCrit, "but everyone has seen the crit now")
+
+ // We are able to stop the processing now
+ sys.Stop()
+
+ emitterA.Emit(TestEvent{})
+ require.NoError(t, exec.Drain(), "system is closed, no further event processing")
+ require.Equal(t, 2, count)
+}