improve(SpokePoolClient): Support restart of failed listener

In the event that a child process terminates improperly, support
restarting it with its initial lookback.

Author: pxrl

src/clients/SpokePoolClient.ts

@@ -79,6 +79,7 @@ export class IndexedSpokePoolClient extends clients.SpokePoolClient {
    * @returns void
    */
   protected startWorker(): void {
+    // fromBlock is retained for the life of the SpokePoolClient and is reused in case of listener crash.
     const {
       finality,
       eventSearchConfig: { fromBlock, maxBlockLookBack: blockRange },
@@ -92,6 +93,7 @@ export class IndexedSpokePoolClient extends clients.SpokePoolClient {
       stdio: ["ignore", "inherit", "inherit", "ipc"],
     });
 
+    this.worker.on("exit", (code, signal) => this.childExit(code, signal));
     this.worker.on("message", (message) => this.indexerUpdate(message));
     this.logger.debug({
       at: "SpokePoolClient#startWorker",
@@ -100,6 +102,50 @@ export class IndexedSpokePoolClient extends clients.SpokePoolClient {
     });
   }
 
+  /**
+   * The worker process has exited. Optionally restart it based on the exit code.
+   * See also: https://nodejs.org/api/child_process.html#event-exit
+   * @param code Optional exit code.
+   * @param signal Optional signal resulting in termination.
+   * @returns void
+   */
+  protected childExit(code?: number, signal?: string): void {
+    if (code === 0) {
+      return;
+    }
+
+    this.logger.warn({
+      at: "SpokePoolClient#childExit",
+      message: `${this.chain} SpokePool listener exited.`,
+      code,
+      signal
+    });
+
+    // Flush all ingested deposits to protect against re-org.
+    // xxx this probably belongs upstream in a `protected SpokePoolClient.init()` method
+    // that is called in the constructor, such that it can be re-called by this method to
+    // re-initialise the same defaults.
+    this.currentTime = 0;
+    this.oldestTime = 0;
+    this.depositHashes = {};
+    this.depositHashesToFills = {};
+    this.speedUps = {};
+    this.slowFillRequests = {};
+    this.fills = {};
+    this.depositRoutes = {};
+
+    this.tokensBridged = [];
+    this.rootBundleRelays = [];
+    this.relayerRefundExecutions = [];
+    this.earliestDepositIdQueried = Number.MAX_SAFE_INTEGER;
+    this.latestDepositIdQueried = 0;
+    this.firstDepositIdForSpokePool = Number.MAX_SAFE_INTEGER;
+    this.lastDepositIdForSpokePool = Number.MAX_SAFE_INTEGER;
+
+    // Restart the listener process from the initial `fromBlock`.
+    this.startWorker();
+  }
+
   /**
    * Receive an update from the external indexer process.
    * @param rawMessage Message to be parsed.