Merge branch 'master' into docker

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@kenshi.io/unchained",
-  "version": "0.8.2",
+  "version": "0.8.3",
   "description": "",
   "main": "dist/index.js",
   "scripts": {
@@ -17,13 +17,15 @@
     "@chainsafe/bls": "^7.1.2",
     "base-ex": "^0.8.1",
     "commander": "^11.1.0",
+    "console-table-printer": "^2.11.2",
     "enquirer": "^2.4.1",
     "ethers": "^6.8.1",
     "hyperswarm": "^4.7.9",
     "iso-websocket": "^0.1.6",
     "json-canon": "^1.0.1",
     "latest-version": "^7.0.0",
     "mongodb": "^6.3.0",
+    "node-cron": "^3.0.3",
     "ora": "^7.0.1",
     "semver": "^7.5.4",
     "unws": "^0.2.4",
@@ -36,6 +38,7 @@
   },
   "devDependencies": {
     "@types/node": "^20.10.4",
+    "@types/node-cron": "^3.0.11",
     "@types/semver": "^7.5.6",
     "@types/winston": "^2.4.4",
     "@types/ws": "^8.5.10",

src/lib/bls/keys.ts

@@ -1,12 +1,7 @@
 import bls from "@chainsafe/bls";
 import { Base58 } from "base-ex";
-
-import { SecretKey, PublicKey } from "@chainsafe/bls/types";
-
-interface KeyPair {
-  secretKey: SecretKey;
-  publicKey: PublicKey;
-}
+import { KeyPair } from "../types.js";
+import { SecretKey } from "@chainsafe/bls/types";
 
 interface EncodedKeyPair {
   secretKey: string;

src/lib/constants.ts

@@ -1,19 +1,19 @@
 import { sha } from "./utils/hash.js";
 import {
   State,
-  Keys,
+  KeyPair,
   Config,
   StringAnyObject,
   StringGossipMethodObject,
 } from "./types.js";
 
-export const version = "0.8.2";
+export const version = "0.8.3";
 export const protocolVersion = "0.8.0";
 
 export const topic = sha(`Kenshi.Unchained.Testnet.Topic.V${protocolVersion}`);
 
 export const sockets = new Map();
-export const keys: Keys = {};
+export const keys: KeyPair = Object({});
 
 export const config: Config = {
   name: "Change Me",

src/lib/daemon/index.ts

@@ -1,8 +1,8 @@
 import * as uniswap from "../plugins/uniswap/uniswap.js";
 import { gossip } from "../gossip/index.js";
-import { runAtNextInterval } from "../utils/time.js";
-import { runWithRetries, CALLERROR, TIMEOUT } from "../utils/retry.js";
-import { GossipRequest } from "../types.js";
+import { runWithRetries } from "../utils/retry.js";
+import { schedule } from "node-cron";
+import { printScores } from "../score/print.js";
 
 interface UniswapArgs {
   blockchain: string;
@@ -18,7 +18,7 @@ const uniswapArgs: [UniswapArgs, string, [number, number], boolean] = [
 ];
 
 export const runTasks = (): void => {
-  runAtNextInterval(async () => {
+  schedule("*/5 * * * * *", async () => {
     try {
       const result = await runWithRetries(uniswap.work, uniswapArgs);
       if (result && !(result instanceof Symbol)) {
@@ -27,5 +27,13 @@ export const runTasks = (): void => {
     } catch (error) {
       // Handle the error or log it
     }
-  }, 5);
+  }).start();
+
+  schedule("0 */5 * * * *", async () => {
+    try {
+      printScores();
+    } catch (error) {
+      // Handle the error or log it
+    }
+  }).start();
 };

src/lib/plugins/uniswap/uniswap.ts

@@ -8,6 +8,7 @@ import { encoder } from "../../bls/keys.js";
 import { WebSocketLike } from "ethers";
 import { WebSocket } from "unws";
 import { assetPrices } from "../../db/collections/AssetPrice.js";
+import { addOnePoint } from "../../score/index.js";
 
 import assert from "assert";
 
@@ -174,6 +175,10 @@ const setAttestations = async (
     ({ signer }) => !currentSignatures.includes(signer)
   );
 
+  for (const { signer } of newSignatureSet) {
+    addOnePoint(signer);
+  }
+
   const newSignatures = newSignatureSet.map((item) => item.signature);
   const currentAggregation = stored.aggregated || "";
   const signatureList = [currentAggregation, ...newSignatures].filter(Boolean);
@@ -194,6 +199,8 @@ const setAttestations = async (
     printAttestations(size, block, price, signersSet);
   }
 
+  earlyAttestations.set(block, []);
+
   for (const [key] of attestations.entries()) {
     // FIXME: Security problem where a validator can reset another
     // FIXME: peer's cache by sending a big block number

src/lib/score/index.ts

@@ -0,0 +1,11 @@
+const peerScoreMap = new Map<string, number>();
+
+export const addOnePoint = (peer: string) => {
+  const current = peerScoreMap.get(peer) || 0;
+  peerScoreMap.set(peer, current + 1);
+};
+
+export const resetScore = (peer: string) => peerScoreMap.set(peer, 0);
+export const resetAllScores = () => peerScoreMap.clear();
+export const getScoreOf = (peer: string): number => peerScoreMap.get(peer) || 0;
+export const getAllScores = () => peerScoreMap.entries();

src/lib/score/print.ts

@@ -0,0 +1,42 @@
+import { getAllScores, resetAllScores } from "./index.js";
+import { logger } from "../logger/index.js";
+import { Table } from "console-table-printer";
+import { sockets, keys } from "../constants.js";
+import { encodeKeys } from "../bls/keys.js";
+
+export const printScores = () => {
+  const table = new Table({
+    columns: [
+      { name: "peer", title: "Peer", alignment: "left", color: "blue" },
+      { name: "name", title: "Name", alignment: "left", color: "yellow" },
+      { name: "score", title: "Score", alignment: "center", color: "green" },
+    ],
+  });
+
+  const { publicKey: thisNode } = encodeKeys(keys);
+
+  const rows = [];
+  const publicKeyMap = new Map(
+    [...sockets.entries()].map(([_, { publicKey, name }]) => [publicKey, name])
+  );
+
+  for (const [peer, score] of getAllScores()) {
+    if (peer === thisNode) {
+      continue;
+    }
+    const name = publicKeyMap.get(peer) || "?";
+    rows.push({ peer, score, name });
+  }
+
+  if (!rows.length) {
+    return;
+  }
+
+  table.addRows(rows.sort((a, b) => b.score - a.score));
+
+  const sprint = Math.ceil(new Date().valueOf() / 300000);
+  logger.info(`Scores for sprint ${sprint} are:`);
+  table.printTable();
+
+  resetAllScores();
+};

src/lib/swarm/index.ts

@@ -70,7 +70,7 @@ swarm.on("connection", async (socket: Duplex, info: { publicKey: Buffer }) => {
     } else if (message.type === "call") {
       const result = await processRpc(message);
       try {
-        await socket.write(JSON.stringify(result));
+        socket.write(JSON.stringify(result));
       } catch (error) {
         const err = error as NodeSystemError;
         const info = err.code || err.errno || err.message;
@@ -100,7 +100,7 @@ swarm.on("connection", async (socket: Duplex, info: { publicKey: Buffer }) => {
       type: "call",
       request: { method: "introduce", args: {} },
     });
-    await socket.write(introducePayload);
+    socket.write(introducePayload);
   } catch (error) {}
 });
 

src/lib/types.ts

@@ -5,9 +5,9 @@ export interface State {
   connected: boolean;
 }
 
-export interface Keys {
-  publicKey?: PublicKey;
-  secretKey?: SecretKey;
+export interface KeyPair {
+  publicKey: PublicKey;
+  secretKey: SecretKey;
 }
 
 interface RPCList {