fix(dtl): kovan DTL halting issue (#2393)

* fix(dtl): kovan DTL halting issue

* fix(dtl): add consistency check for L2 sync

* dtl: add retry logic on startup

* dtl: fix startup
Co-authored-by: Mark Tyneway <mark.tyneway@gmail.com>

.changeset/brown-dots-whisper.md

+---
+'@eth-optimism/data-transport-layer': patch
+---
+
+Patch for Kovan DTL halting issue

packages/data-transport-layer/src/db/transport-db.ts

@@ -27,6 +27,7 @@ const TRANSPORT_DB_KEYS = {
   STARTING_L1_BLOCK: `l1:starting`,
   HIGHEST_L2_BLOCK: `l2:highest`,
   HIGHEST_SYNCED_BLOCK: `synced:highest`,
+  CONSISTENCY_CHECK: `consistency:checked`,
 }
 
 interface Indexed {
@@ -202,6 +203,20 @@ export class TransportDB {
     return this.db.get<number>(TRANSPORT_DB_KEYS.HIGHEST_L2_BLOCK, 0)
   }
 
+  public async getConsistencyCheckFlag(): Promise<boolean> {
+    return this.db.get<boolean>(TRANSPORT_DB_KEYS.CONSISTENCY_CHECK, 0)
+  }
+
+  public async putConsistencyCheckFlag(flag: boolean): Promise<void> {
+    return this.db.put<boolean>([
+      {
+        key: TRANSPORT_DB_KEYS.CONSISTENCY_CHECK,
+        index: 0,
+        value: flag,
+      },
+    ])
+  }
+
   public async putHighestL2BlockNumber(
     block: number | BigNumber
   ): Promise<void> {

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

+export type EventName = 'SequencerTransaction'
+
+export class MissingElementError extends Error {
+  constructor(public name: EventName) {
+    super(`missing event: ${name}`)
+  }
+}

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

@@ -11,6 +11,7 @@ import {
   TransactionEntry,
 } from '../../../types'
 import { parseSignatureVParam } from '../../../utils'
+import { MissingElementError } from './errors'
 
 export const handleSequencerBlock = {
   parseBlock: async (
@@ -22,9 +23,12 @@ export const handleSequencerBlock = {
   }> => {
     const transaction = block.transactions[0]
     const transactionIndex =
-      transaction.index === null || transaction.index === undefined
-        ? BigNumber.from(transaction.blockNumber).toNumber() - 1
-        : BigNumber.from(transaction.index).toNumber()
+      BigNumber.from(transaction.blockNumber).toNumber() - 1
+
+    // We make the assumption that you don't need to sync the genesis block
+    if (transactionIndex < 0) {
+      throw new Error('should not happen, attempted to sync genesis block')
+    }
 
     let transactionEntry: Partial<TransactionEntry> = {
       // Legacy support.
@@ -111,6 +115,17 @@ export const handleSequencerBlock = {
     },
     db: TransportDB
   ): Promise<void> => {
+    if (entry.transactionEntry.index > 0) {
+      const prevTransactionEntry = await db.getUnconfirmedTransactionByIndex(
+        entry.transactionEntry.index - 1
+      )
+
+      // We should *always* have a previous transaction here.
+      if (prevTransactionEntry === null) {
+        throw new MissingElementError('SequencerTransaction')
+      }
+    }
+
     // Having separate indices for confirmed/unconfirmed means we never have to worry about
     // accidentally overwriting a confirmed transaction with an unconfirmed one. Unconfirmed
     // transactions are purely extra information.

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

@@ -107,7 +107,58 @@ export class L2IngestionService extends BaseService<L2IngestionServiceOptions> {
         : this.options.l2RpcProvider
   }
 
+  protected async ensure(): Promise<void> {
+    let retries = 0
+    while (true) {
+      try {
+        await this.state.l2RpcProvider.getNetwork()
+        break
+      } catch (e) {
+        retries++
+        this.logger.info(`Cannot connect to L2, retrying ${retries}/20`)
+        if (retries >= 20) {
+          this.logger.info('Cannot connect to L2, shutting down')
+          await this.stop()
+          process.exit()
+        }
+        await sleep(1000 * retries)
+      }
+    }
+  }
+
+  protected async checkConsistency(): Promise<void> {
+    const network = await this.state.l2RpcProvider.getNetwork()
+    const shouldDoCheck = !(await this.state.db.getConsistencyCheckFlag())
+    if (shouldDoCheck && network.chainId === 69) {
+      this.logger.info('performing consistency check')
+      const highestBlock =
+        await this.state.db.getHighestSyncedUnconfirmedBlock()
+      for (let i = 0; i < highestBlock; i++) {
+        const block = await this.state.db.getUnconfirmedTransactionByIndex(i)
+        if (block === null) {
+          this.logger.info('resetting to null block', {
+            index: i,
+          })
+          await this.state.db.setHighestSyncedUnconfirmedBlock(i)
+          break
+        }
+
+        // Log some progress so people know what's goin on.
+        if (i % 10000 === 0) {
+          this.logger.info(`consistency check progress`, {
+            index: i,
+          })
+        }
+      }
+      this.logger.info('consistency check complete')
+      await this.state.db.putConsistencyCheckFlag(true)
+    }
+  }
+
   protected async _start(): Promise<void> {
+    await this.ensure()
+    await this.checkConsistency()
+
     while (this.running) {
       try {
         const highestSyncedL2BlockNumber =