fix[dtl]: defend against RPC provider missing events

.changeset/brown-boxes-compete.md

@@ -0,0 +1,5 @@
+---
+'@eth-optimism/data-transport-layer': patch
+---
+
+Adds additional code into the DTL to defend against situations where an RPC provider might be missing an event.

packages/data-transport-layer/src/services/l1-ingestion/handlers/errors.ts

@@ -0,0 +1,5 @@
+export class MissingElementError extends Error {
+  constructor(event: string) {
+    super(`missing event: ${event}`)
+  }
+}

packages/data-transport-layer/src/services/l1-ingestion/handlers/sequencer-batch-appended.ts

@@ -22,6 +22,7 @@ import {
   SEQUENCER_GAS_LIMIT,
   parseSignatureVParam,
 } from '../../../utils'
+import { MissingElementError } from './errors'
 
 export const handleEventsSequencerBatchAppended: EventHandlerSet<
   EventArgsSequencerBatchAppended,
@@ -181,6 +182,18 @@ export const handleEventsSequencerBatchAppended: EventHandlerSet<
     }
   },
   storeEvent: async (entry, db) => {
+    // Defend against situations where we missed an event because the RPC provider
+    // (infura/alchemy/whatever) is missing an event.
+    if (entry.transactionBatchEntry.index > 0) {
+      if (
+        (await db.getTransactionBatchByIndex(
+          entry.transactionBatchEntry.index - 1
+        )) === null
+      ) {
+        throw new MissingElementError('SequencerBatchAppended')
+      }
+    }
+
     await db.putTransactionBatchEntries([entry.transactionBatchEntry])
     await db.putTransactionEntries(entry.transactionEntries)
 

packages/data-transport-layer/src/services/l1-ingestion/handlers/state-batch-appended.ts

@@ -11,6 +11,7 @@ import {
   StateRootEntry,
   EventHandlerSet,
 } from '../../../types'
+import { MissingElementError } from './errors'
 
 export const handleEventsStateBatchAppended: EventHandlerSet<
   EventArgsStateBatchAppended,
@@ -67,6 +68,18 @@ export const handleEventsStateBatchAppended: EventHandlerSet<
     }
   },
   storeEvent: async (entry, db) => {
+    // Defend against situations where we missed an event because the RPC provider
+    // (infura/alchemy/whatever) is missing an event.
+    if (entry.stateRootBatchEntry.index > 0) {
+      if (
+        (await db.getStateRootBatchByIndex(
+          entry.stateRootBatchEntry.index - 1
+        )) === null
+      ) {
+        throw new MissingElementError('StateBatchAppended')
+      }
+    }
+
     await db.putStateRootBatchEntries([entry.stateRootBatchEntry])
     await db.putStateRootEntries(entry.stateRootEntries)
   },

packages/data-transport-layer/src/services/l1-ingestion/handlers/transaction-enqueued.ts

@@ -3,6 +3,7 @@ import { EventArgsTransactionEnqueued } from '@eth-optimism/core-utils'
 /* Imports: Internal */
 import { BigNumber } from 'ethers'
 import { EnqueueEntry, EventHandlerSet } from '../../../types'
+import { MissingElementError } from './errors'
 
 export const handleEventsTransactionEnqueued: EventHandlerSet<
   EventArgsTransactionEnqueued,
@@ -25,6 +26,14 @@ export const handleEventsTransactionEnqueued: EventHandlerSet<
     }
   },
   storeEvent: async (entry, db) => {
+    // Defend against situations where we missed an event because the RPC provider
+    // (infura/alchemy/whatever) is missing an event.
+    if (entry.index > 0) {
+      if ((await db.getEnqueueByIndex(entry.index - 1)) === null) {
+        throw new MissingElementError('TransactionEnqueued')
+      }
+    }
+
     await db.putEnqueueEntries([entry])
   },
 }

packages/data-transport-layer/src/services/l1-ingestion/service.ts

@@ -19,6 +19,7 @@ import { handleEventsTransactionEnqueued } from './handlers/transaction-enqueued
 import { handleEventsSequencerBatchAppended } from './handlers/sequencer-batch-appended'
 import { handleEventsStateBatchAppended } from './handlers/state-batch-appended'
 import { L1DataTransportServiceOptions } from '../main/service'
+import { MissingElementError } from './handlers/errors'
 
 export interface L1IngestionServiceOptions
   extends L1DataTransportServiceOptions {
@@ -205,7 +206,29 @@ export class L1IngestionService extends BaseService<L1IngestionServiceOptions> {
           await sleep(this.options.pollingInterval)
         }
       } catch (err) {
-        if (!this.running || this.options.dangerouslyCatchAllErrors) {
+        if (err instanceof MissingElementError) {
+          // Different functions for getting the last good element depending on the event type.
+          const handlers = {
+            SequencerBatchAppended: this.state.db.getLatestTransactionBatch,
+            StateBatchAppended: this.state.db.getLatestStateRootBatch,
+            TransactionEnqueued: this.state.db.getLatestEnqueue,
+          }
+
+          // Find the last good element and reset the highest synced L1 block to go back to the
+          // last good element. Will resync other event types too but we have no issues with
+          // syncing the same events more than once.
+          const eventName = err.message.split('missing event: ')[1]
+          const lastGoodElement = await handlers[eventName]()
+          await this.state.db.setHighestSyncedL1Block(
+            lastGoodElement.blockNumber
+          )
+
+          // Something we should be keeping track of.
+          this.logger.warn('recovering from a missing event', {
+            eventName,
+            lastGoodBlockNumber: lastGoodElement.blockNumber,
+          })
+        } else if (!this.running || this.options.dangerouslyCatchAllErrors) {
           this.logger.error('Caught an unhandled error', {
             message: err.toString(),
             stack: err.stack,