Fix validator waiting on start

packages/lodestar/src/validator/rpc/abstract.ts

@@ -4,10 +4,12 @@ import {computeEpochOfSlot} from "../../chain/stateTransition/util";
 import {IBeaconApi} from "../../api/rpc/api/beacon";
 import {IBeaconConfig} from "@chainsafe/eth2.0-config";
 import {getCurrentSlot} from "../../chain/stateTransition/util/genesis";
-import {INewEpochCallback, INewSlotCallback, IRpcClient} from "./interface";
+import {INewEpochCallback, INewSlotCallback, IRpcClient, RpcClientEventEmitter} from "./interface";
+import {EventEmitter} from "ws";
 
 
-export abstract class AbstractRpcClient implements IRpcClient {
+export abstract class AbstractRpcClient extends (EventEmitter as { new(): RpcClientEventEmitter }) 
+  implements IRpcClient {
 
   protected config: IBeaconConfig;
 
@@ -16,6 +18,7 @@ export abstract class AbstractRpcClient implements IRpcClient {
   private newSlotCallbacks: INewSlotCallback[] = [];
   private newEpochCallbacks: INewEpochCallback[] = [];
   private running = false;
+  private beaconNodeInterval: NodeJS.Timeout;
 
   public onNewEpoch(cb: INewEpochCallback): void {
     if (cb) {
@@ -31,10 +34,25 @@ export abstract class AbstractRpcClient implements IRpcClient {
 
   public async connect(): Promise<void> {
     await this.startSlotCounting();
+    this.beaconNodeInterval = setInterval(this.pollBeaconNode.bind(this), 1000);
   }
 
   public async disconnect(): Promise<void> {
     this.running = false;
+    if(this.beaconNodeInterval) {
+      clearInterval(this.beaconNodeInterval);
+    }
+  }
+
+  private async pollBeaconNode(): Promise<void> {
+    if (!this.running) {
+      return;
+    }
+    const genesisTime =  await this.beacon.getGenesisTime();
+    if (genesisTime && (Date.now() / 1000) > genesisTime) {
+      this.emit("beaconChainStarted");
+      clearInterval(this.beaconNodeInterval);
+    }
   }
 
   private async startSlotCounting(): Promise<void> {

packages/lodestar/src/validator/rpc/interface.ts

@@ -1,5 +1,7 @@
 import {IBeaconApi, IValidatorApi} from "../../api/rpc/api";
 import {Epoch, Slot} from "@chainsafe/eth2.0-types";
+import StrictEventEmitter from "strict-event-emitter-types";
+import {EventEmitter} from "events";
 
 export interface INewSlotCallback {
   (slot: Slot): void;
@@ -9,7 +11,13 @@ export interface INewEpochCallback {
   (slot: Epoch): void;
 }
 
-export interface IRpcClient {
+export interface IRpcClientEvents {
+  beaconChainStarted: () => void;
+}
+
+export type RpcClientEventEmitter = StrictEventEmitter<EventEmitter, IRpcClientEvents>;
+
+export interface IRpcClient extends RpcClientEventEmitter {
 
   beacon: IBeaconApi;
 

packages/lodestar/src/validator/validator.ts

@@ -16,7 +16,6 @@
 import BlockProposingService from "./services/block";
 import {Slot} from "@chainsafe/eth2.0-types";
 import {IBeaconConfig} from "@chainsafe/eth2.0-config";
-import {GenesisInfo} from "./types";
 import {IRpcClient, RpcClientOverWs} from "./rpc";
 import {AttestationService} from "./services/attestation";
 import {IValidatorDB, LevelDbController, ValidatorDB} from "../db";
@@ -40,18 +39,14 @@ class Validator {
   private blockService: BlockProposingService;
   // @ts-ignore
   private attestationService: AttestationService;
-  // @ts-ignore
-  private genesisInfo: GenesisInfo;
   private db: IValidatorDB;
   private logger: ILogger;
-  private isActive: boolean;
   private isRunning: boolean;
 
   public constructor(opts: Partial<IValidatorOptions>, modules: {config: IBeaconConfig; logger: ILogger}) {
     this.opts = deepmerge(defaultValidatorOptions, opts, {isMergeableObject: isPlainObject});
     this.config = modules.config;
     this.logger = modules.logger.child(this.opts.logger);
-    this.isActive = false;
     this.isRunning = false;
     this.db = new ValidatorDB({
       config: this.config,
@@ -70,7 +65,14 @@ class Validator {
   public async start(): Promise<void> {
     this.isRunning = true;
     await this.setup();
-    this.run();
+    this.logger.info("Checking if chain has started...");
+    this.apiClient.once("beaconChainStarted", this.run.bind(this));
+  }
+
+  public run(): void {
+    this.logger.info("Chain start has occured!");
+    this.apiClient.onNewSlot(this.checkDuties);
+    // this.apiClient.onNewEpoch(this.lookAhead);
   }
 
   /**
@@ -79,6 +81,7 @@ class Validator {
   public async stop(): Promise<void> {
     this.isRunning = false;
     await this.apiClient.disconnect();
+
   }
 
   private initApiClient(): void {
@@ -103,9 +106,6 @@ class Validator {
 
     await this.setupRPC();
 
-    // Wait for the ChainStart log and grab validator index
-    this.isActive = await this.isChainLive();
-
     this.blockService = new BlockProposingService(
       this.config,
       this.opts.keypair,
@@ -132,29 +132,6 @@ class Validator {
     this.logger.info(`RPC connection successfully established: ${this.apiClient.url}!`);
   }
 
-  /**
-   * Recursively checks for the chain start log event from the ETH1.x deposit contract
-   */
-  private isChainLive = async (): Promise<boolean> => {
-    this.logger.info("Checking if chain has started...");
-    const genesisTime =  await this.apiClient.beacon.getGenesisTime();
-    if (genesisTime && (Date.now() / 1000) > genesisTime) {
-      this.genesisInfo = {
-        startTime: genesisTime,
-      };
-      this.logger.info("Chain start has occured!");
-      return true;
-    }
-    if(this.isRunning) {
-      setTimeout(this.isChainLive, 1000);
-    }
-    return false;
-  };
-
-  private run(): void {
-    this.apiClient.onNewSlot(this.checkDuties);
-    // this.apiClient.onNewEpoch(this.lookAhead);
-  }
 
   private checkDuties = async (slot: Slot): Promise<void> => {
     const validatorDuty =

packages/lodestar/test/unit/validator/validator.test.ts

@@ -1,6 +1,7 @@
 import {expect} from "chai";
 import {Keypair} from "@chainsafe/bls/lib/keypair";
 import {config} from "@chainsafe/eth2.0-config/lib/presets/mainnet";
+import sinon from "sinon";
 import Validator from "../../../src/validator";
 import {RpcClientOverInstance} from "../../../src/validator/rpc";
 import {MockBeaconApi} from "../../utils/mocks/rpc/beacon";
@@ -34,8 +35,10 @@ describe("Validator", () => {
     };
 
     const validator = new Validator(validatorCtx, {config, logger});
+    const runSpy = sinon.spy(validator, "run");
     await expect(validator.start()).to.not.throw;
-    await validator.stop();
+    setTimeout(async () => validator.stop(), 1100);
+    setTimeout(() => expect(runSpy.calledOnce).to.be.true, 1100);
   });
 
 });