Add gateway connectivity and observe unprocessed jobs

config.example.json

@@ -2,7 +2,8 @@
   "rpcEndpoint": "https://nois-testnet-rpc.itrocket.net:443",
   "rpcEndpoint2": "",
   "rpcEndpoint3": "",
-  "contract": "nois14xef285hz5cx5q9hh32p9nztu3cct4g44sxjgx3dmftt2tj2rweqkjextk",
+  "drandAddress": "nois14xef285hz5cx5q9hh32p9nztu3cct4g44sxjgx3dmftt2tj2rweqkjextk",
+  "gatewayAddress": "",
   "mnemonic": "",
   "denom": "unois",
   "gasPrice": "0.05unois",

deps.ts

@@ -12,7 +12,7 @@ export {
 } from "npm:@cosmjs/stargate@^0.31.0";
 export { sha256 } from "npm:@cosmjs/crypto@^0.31.0";
 export { toUtf8 } from "npm:@cosmjs/encoding@^0.31.0";
-export { Decimal } from "npm:@cosmjs/math@^0.31.0";
+export { Decimal, Uint53 } from "npm:@cosmjs/math@^0.31.0";
 export { DirectSecp256k1HdWallet } from "npm:@cosmjs/proto-signing@^0.31.0";
 export { Tendermint34Client, Tendermint37Client } from "npm:@cosmjs/tendermint-rpc@^0.31.0";
 export { assert, isDefined, sleep } from "npm:@cosmjs/utils@^0.31.0";

drand.ts

@@ -1,6 +1,6 @@
 import { ChainOptions } from "./deps.ts";
 
-const chainHash = "dbd506d6ef76e5f386f41c651dcb808c5bcbd75471cc4eafa3f4df7ad4e4c493";
+export const chainHash = "dbd506d6ef76e5f386f41c651dcb808c5bcbd75471cc4eafa3f4df7ad4e4c493";
 const publicKey =
   "a0b862a7527fee3a731bcb59280ab6abd62d5c0b6ea03dc4ddf6612fdfc9d01f01c31542541771903475eb1ec6615f8d0df0b8b6dce385811d6dcf8cbefb8759e5e616a3dfd054c928940766d9a5b9db91e3b697e5d70a975181e007f87fca5e";
 

drand_contract.ts

@@ -37,13 +37,15 @@ export async function queryIsAllowListed(
 export async function queryIsIncentivized(
   client: CosmWasmClient,
   contractAddress: string,
-  rounds: number[],
+  round: number,
   botAddress: string,
-): Promise<boolean[]> {
+): Promise<boolean> {
   const { incentivized } = await client.queryContractSmart(contractAddress, {
-    is_incentivized: { rounds, sender: botAddress },
+    is_incentivized: { rounds: [round], sender: botAddress },
   });
   // console.log(`#${rounds[0]} incentivized query returned at ${publishedSince(rounds[0])}ms`)
   assert(Array.isArray(incentivized));
-  return incentivized;
+  const first = incentivized[0];
+  assert(typeof first === "boolean");
+  return first;
 }

jobs.ts

@@ -0,0 +1,57 @@
+import { assert, CosmWasmClient, Uint53 } from "./deps.ts";
+import { chainHash, timeOfRound } from "./drand.ts";
+
+interface Job {
+  /// A RNG specific randomness source identifier, e.g. `drand:<network id>:<round>`
+  source_id: string;
+  // The channel the job came from and we have to send the response to
+  channel: string;
+  origin: string;
+}
+
+interface JobsResponse {
+  jobs: Job[];
+}
+
+function parseRound(job: Job): number {
+  const [sourceType, networkId, round] = job.source_id.split(":");
+  assert(sourceType == "drand", "Source type must be 'drand'");
+  assert(networkId == chainHash, "Got wrong chain hash in job");
+  assert(round, "Round must be set");
+  return Uint53.fromString(round).toNumber();
+}
+
+function formatDuration(durationInMs: number): string {
+  const inSeconds = durationInMs / 1000;
+  return `${inSeconds.toFixed(1)}s`;
+}
+
+export class JobsObserver {
+  private readonly noisClient: CosmWasmClient;
+  private readonly gateway: string;
+
+  public constructor(
+    noisClient: CosmWasmClient,
+    gatewayAddress: string,
+  ) {
+    this.noisClient = noisClient;
+    this.gateway = gatewayAddress;
+  }
+
+  /**
+   * Checks gateway for pending jobs and returns the rounds of those jobs as a list
+   */
+  public async check(): Promise<number[]> {
+    const query = { jobs_desc: { offset: null, limit: 3 } };
+    const { jobs }: JobsResponse = await this.noisClient.queryContractSmart(this.gateway, query);
+    if (jobs.length === 0) return []; // Nothing to do for us
+
+    const rounds = jobs.map(parseRound);
+    const roundInfos = rounds.map((round) => {
+      const due = timeOfRound(round) - Date.now();
+      return `#${round} (due ${formatDuration(due)})`;
+    });
+    console.log(`Jobs pending for rounds: %c${roundInfos.join(", ")}`, "color: orange");
+    return rounds;
+  }
+}

main.ts

@@ -1,4 +1,4 @@
-import { drandOptions, drandUrls, publishedIn } from "./drand.ts";
+import { drandOptions, drandUrls, publishedIn, publishedSince } from "./drand.ts";
 import { group } from "./group.ts";
 import {
   assert,
@@ -14,8 +14,8 @@ import {
   sleep,
   watch,
 } from "./deps.ts";
-import { BeaconCache } from "./cache.ts";
-import { loop } from "./loop.ts";
+import { JobsObserver } from "./jobs.ts";
+import { Submitter } from "./submitter.ts";
 import { queryIsAllowListed, queryIsIncentivized } from "./drand_contract.ts";
 import { connectTendermint } from "./tendermint.ts";
 
@@ -53,7 +53,7 @@ if (import.meta.main) {
   const { default: config } = await import("./config.json", {
     assert: { type: "json" },
   });
-  assert(config.contract, `Config field "contract" must be set.`);
+  assert(config.drandAddress, `Config field "drandAddress" must be set.`);
   assert(config.rpcEndpoint, `Config field "rpcEndpoint" must be set.`);
 
   const mnemonic = await (async () => {
@@ -106,7 +106,7 @@ if (import.meta.main) {
     const fee = calculateFee(gasLimitRegister, config.gasPrice);
     await client.execute(
       botAddress,
-      config.contract,
+      config.drandAddress,
       { register_bot: { moniker: moniker } },
       fee,
     );
@@ -117,52 +117,92 @@ if (import.meta.main) {
   await Promise.all([
     sleep(500), // the min waiting time
     (async function () {
-      const listed = await queryIsAllowListed(client, config.contract, botAddress);
+      const listed = await queryIsAllowListed(client, config.drandAddress, botAddress);
       console.info(`Bot allow listed for rewards: ${listed}`);
     })(),
   ]);
 
+  let jobs: JobsObserver | undefined;
+  if (config.gatewayAddress) {
+    jobs = new JobsObserver(client, config.gatewayAddress);
+  }
+
   // Initialize local sign data
   await resetSignData();
 
   const incentivizedRounds = new Map<number, Promise<boolean>>();
 
   const fastestNodeClient = new FastestNodeClient(drandUrls, drandOptions);
+
+  const submitter = new Submitter({
+    client,
+    tmClient,
+    broadcaster2,
+    broadcaster3,
+    getNextSignData,
+    gasLimitAddBeacon,
+    gasPrice: config.gasPrice,
+    botAddress,
+    drandAddress: config.drandAddress,
+    userAgent,
+    incentivizedRounds,
+    drandClient: fastestNodeClient,
+  });
+
   fastestNodeClient.start();
-  const cache = new BeaconCache(fastestNodeClient, 200 /* 10 min of beacons */);
   const abortController = new AbortController();
   for await (const beacon of watch(fastestNodeClient, abortController)) {
     const n = beacon.round; // n is the round we just received and process now
     const m = n + 1; // m := n+1 refers to the next round in this current loop
 
-    cache.add(n, beacon.signature);
+    console.log(`➘ #${beacon.round} received after ${publishedSince(beacon.round)}ms`);
 
     setTimeout(() => {
       // This is called 100ms after publishing time (might be some ms later)
       // From here we have ~300ms until the beacon comes in which should be
       // enough for the query to finish. In case the query is not yet done,
       // we can wait for the promise to be resolved.
       // console.log(`Now         : ${new Date().toISOString()}\nPublish time: ${new Date(timeOfRound(round)).toISOString()}`);
-      const promise = queryIsIncentivized(client, config.contract, [m], botAddress).then(
-        (incentivized) => !!incentivized[0],
+      const promise = queryIsIncentivized(client, config.drandAddress, m, botAddress).catch(
         (_err) => false,
       );
       incentivizedRounds.set(m, promise);
     }, publishedIn(m) + 100);
 
-    const didSubmit = await loop({
-      client,
-      tmClient,
-      broadcaster2,
-      broadcaster3,
-      getNextSignData,
-      gasLimitAddBeacon,
-      gasPrice: config.gasPrice,
-      botAddress,
-      drandAddress: config.contract,
-      userAgent,
-      incentivizedRounds,
-    }, beacon);
+    const didSubmit = await submitter.handlePublishedBeacon(beacon);
+
+    // Check jobs every 1.5s, shifted 1200ms from the drand receiving
+    const shift = 1200;
+    setTimeout(() =>
+      jobs?.check().then(
+        (rounds) => {
+          if (!rounds.length) return;
+          const past = rounds.filter((r) => r <= n);
+          const future = rounds.filter((r) => r > n);
+          console.log(
+            `Past: %o, Future: %o`,
+            past,
+            future,
+          );
+          submitter.submitPastRounds(past);
+        },
+        (err) => console.error(err),
+      ), shift);
+    setTimeout(() =>
+      jobs?.check().then(
+        (rounds) => {
+          if (!rounds.length) return;
+          const past = rounds.filter((r) => r <= n);
+          const future = rounds.filter((r) => r > n);
+          console.log(
+            `Past: %o, Future: %o`,
+            past,
+            future,
+          );
+          submitter.submitPastRounds(past);
+        },
+        (err) => console.error(err),
+      ), shift + 1500);
 
     if (didSubmit) {
       // Some seconds after the submission when things are idle, check and log