Fix consensus routine synchronization on 4+ nodes

src/index.ts

@@ -242,7 +242,7 @@ async function preMainLoop() {
     getSharedState.connectionString = ourselves
     log.info("Our connection string is: " + ourselves)
     // And saves the public key file
-    fs.writeFileSync(
+    await fs.promises.writeFile(
         "publickey_" + getSharedState.signingAlgorithm + "_" + publicKeyHex,
         publicKeyHex + "\n",
     )

src/libs/blockchain/gcr/gcr.ts

@@ -87,7 +87,7 @@ export class OperationsRegistry {
     }
 
     // INFO Adding an operation to the registry
-    add(operation: Operation) {
+    async add(operation: Operation) {
         this.operations.push({
             operation: operation,
             status: "pending",
@@ -97,7 +97,7 @@ export class OperationsRegistry {
             },
             timestamp: Date.now(),
         })
-        fs.writeFileSync(this.path, JSON.stringify(this.operations))
+        await fs.promises.writeFile(this.path, JSON.stringify(this.operations))
     }
 
     // INFO Getting the full list of operations currently in the registry

src/libs/blockchain/mempool_v2.ts

@@ -1,4 +1,10 @@
-import { FindManyOptions, In, QueryFailedError, Repository } from "typeorm"
+import {
+    FindManyOptions,
+    In,
+    LessThanOrEqual,
+    QueryFailedError,
+    Repository,
+} from "typeorm"
 import Datasource from "@/model/datasource"
 
 import TxUtils from "./transaction"
@@ -7,6 +13,7 @@ import { MempoolTx } from "@/model/entities/Mempool"
 import { Transaction } from "@kynesyslabs/demosdk/types"
 import SecretaryManager from "../consensus/v2/types/secretaryManager"
 import Chain from "./chain"
+import { getSharedState } from "@/utilities/sharedState"
 
 export default class Mempool {
     public static repo: Repository<MempoolTx> = null
@@ -17,6 +24,7 @@ export default class Mempool {
 
     /**
      * Returns the mempool. If `blockNumber` is not provided, returns all transactions.
+     * When `blockNumber` is transaction past from a previous block number are included.
      *
      * @param blockNumber - The block number to filter by
      */
@@ -31,7 +39,7 @@ export default class Mempool {
 
         if (blockNumber) {
             options.where = {
-                blockNumber: blockNumber,
+                blockNumber: LessThanOrEqual(blockNumber),
             }
         }
 
@@ -83,11 +91,10 @@ export default class Mempool {
         }
 
         let blockNumber: number
-        const manager = SecretaryManager.getInstance()
 
         // INFO: If we're in consensus, move tx to next block
-        if (manager.shard?.blockRef) {
-            blockNumber = manager.shard.blockRef + 1
+        if (getSharedState.inConsensusLoop) {
+            blockNumber = SecretaryManager.lastBlockRef + 1
         }
 
         if (!blockNumber) {
@@ -152,15 +159,13 @@ export default class Mempool {
             }
         }
 
-        const manager = SecretaryManager.getInstance()
-        const blockNumber = manager.shard?.blockRef
-
-        if (!blockNumber) {
+        if (!getSharedState.inConsensusLoop) {
             return {
                 success: false,
                 mempool: [],
             }
         }
+        const blockNumber = SecretaryManager.lastBlockRef
 
         const existingHashes = await this.getMempoolHashMap(blockNumber)
         const incomingSet = {}

src/libs/blockchain/routines/Sync.ts

@@ -250,6 +250,7 @@ async function verifyLastBlockIntegrity(
 }
 
 async function downloadBlock(peer: Peer, blockToAsk: number) {
+    log.debug("Downloading block: " + blockToAsk)
     const blockRequest: RPCRequest = {
         method: "nodeCall",
         params: [
@@ -264,6 +265,7 @@ async function downloadBlock(peer: Peer, blockToAsk: number) {
     const blockResponse = await peer.longCall(blockRequest, false, 250, 3, [
         404,
     ])
+    log.debug("Block response: " + blockResponse.result)
 
     // INFO: Handle max retries reached
     if (blockResponse.result === 400) {
@@ -290,6 +292,8 @@ async function downloadBlock(peer: Peer, blockToAsk: number) {
 
         log.info("[downloadBlock] Block received: " + block.hash)
         await Chain.insertBlock(block, [], null, false)
+        log.debug("Block inserted successfully")
+        log.debug("Last block number: " + getSharedState.lastBlockNumber + " Last block hash: " + getSharedState.lastBlockHash)
         log.info(
             "[fastSync] Block inserted successfully at the head of the chain!",
         )

src/libs/consensus/routines/consensusTime.ts

@@ -12,7 +12,9 @@ export async function checkConsensusTime(
 ): Promise<boolean> {
     // Safeguard to prevent the consensus time from being checked before the last block is forged
     if (getSharedState.inConsensusLoop) {
-        log.warning("[CONSENSUS TIME] Cannot check consensus time while in consensus loop, skipping")
+        log.warning(
+            "[CONSENSUS TIME] Cannot check consensus time while in consensus loop, skipping",
+        )
         return false
     }
     let isConsensusTime = false
@@ -27,16 +29,21 @@ export async function checkConsensusTime(
     // REVIEW Using the UTC timestamp as per mainLoop.ts settings
     const currentTimestamp = getNetworkTimestamp() // Date.now()
     const delta = currentTimestamp - lastTimestamp
-    const consensusIntervalTime =
-        getSharedState.getConsensusTime() || 10 // 10 seconds, use 10000 for 10 seconds in ms
+    const consensusIntervalTime = getSharedState.getConsensusTime()
     log.debug("[CONSENSUS TIME] lastTimestamp: " + lastTimestamp, true)
     log.debug("[CONSENSUS TIME] currentTimestamp: " + currentTimestamp, true)
-    log.debug("[CONSENSUS TIME] delta: " + delta, true)
-    log.debug("[CONSENSUS TIME] consensusIntervalTime: " + consensusIntervalTime, true)
+    log.debug("[CONSENSUS TIME] delta: " + delta)
+    log.debug(
+        "[CONSENSUS TIME] consensusIntervalTime: " + consensusIntervalTime,
+        true,
+    )
     //process.exit(0)
 
     // If the delta is greater than the consensus interval time, then the consensus time has passed
-    log.info("[CONSENSUS TIME] consensusIntervalTime: " + consensusIntervalTime, false)
+    log.info(
+        "[CONSENSUS TIME] consensusIntervalTime: " + consensusIntervalTime,
+        false,
+    )
     if (delta >= consensusIntervalTime) {
         isConsensusTime = true
         log.info("[CONSENSUS TIME] Consensus time reached", true)
@@ -48,7 +55,12 @@ export async function checkConsensusTime(
             const minDelta = consensusIntervalTime - flextime
             if (delta > minDelta && delta < maxDelta) {
                 isConsensusTime = true
-                log.info("[CONSENSUS TIME] Consensus time reached (with flexible time and delta: " + delta + ")", true)
+                log.info(
+                    "[CONSENSUS TIME] Consensus time reached (with flexible time and delta: " +
+                        delta +
+                        ")",
+                    true,
+                )
             }
         }
     }
@@ -58,5 +70,3 @@ export async function checkConsensusTime(
     // We can return the result
     return isConsensusTime
 }
-
-