fix: peer disconnection (#296)

Author: BlobMaster41

package.json

@@ -1,7 +1,7 @@
 {
     "type": "module",
     "name": "@btc-vision/opnet-node",
-    "version": "1.0.4",
+    "version": "1.0.5",
     "description": "Bitcoin Smart Contracts Node",
     "main": "build/index.js",
     "scripts": {
@@ -40,7 +40,8 @@
         "typescript-eslint": {
             "typescript": "$typescript"
         },
-        "jsonpath-plus": "^10.4.0"
+        "jsonpath-plus": "^10.4.0",
+        "interface-datastore": "^10.0.1"
     },
     "dependencies": {
         "@asyncapi/generator": "^3.2.1",
@@ -64,25 +65,25 @@
         "@chainsafe/libp2p-yamux": "^8.0.1",
         "@datastructures-js/priority-queue": "^6.3.5",
         "@eslint/js": "^10.0.1",
-        "@libp2p/autonat-v2": "^2.0.18",
-        "@libp2p/bootstrap": "^12.0.20",
-        "@libp2p/crypto": "^5.1.17",
-        "@libp2p/identify": "^4.1.3",
+        "@libp2p/autonat-v2": "^2.0.19",
+        "@libp2p/bootstrap": "^12.0.21",
+        "@libp2p/crypto": "^5.1.18",
+        "@libp2p/identify": "^4.1.4",
         "@libp2p/interface": "^3.2.2",
         "@libp2p/interface-transport": "^4.0.3",
-        "@libp2p/kad-dht": "^16.2.4",
-        "@libp2p/logger": "^6.2.6",
-        "@libp2p/mdns": "^12.0.20",
-        "@libp2p/peer-id": "^6.0.8",
-        "@libp2p/peer-store": "^12.0.18",
-        "@libp2p/ping": "^3.1.3",
-        "@libp2p/tcp": "^11.0.18",
-        "@libp2p/upnp-nat": "^4.0.18",
-        "@libp2p/websockets": "^10.1.11",
+        "@libp2p/kad-dht": "^16.2.5",
+        "@libp2p/logger": "^6.2.7",
+        "@libp2p/mdns": "^12.0.21",
+        "@libp2p/peer-id": "^6.0.9",
+        "@libp2p/peer-store": "^12.0.19",
+        "@libp2p/ping": "^3.1.4",
+        "@libp2p/tcp": "^11.0.19",
+        "@libp2p/upnp-nat": "^4.0.19",
+        "@libp2p/websockets": "^10.1.12",
         "@multiformats/multiaddr": "^13.0.1",
         "@noble/curves": "^2.2.0",
         "@noble/secp256k1": "^3.1.0",
-        "@typescript/native-preview": "^7.0.0-dev.20260506.1",
+        "@typescript/native-preview": "^7.0.0-dev.20260511.1",
         "asyncapi-validator": "^5.1.1",
         "bignumber.js": "^11.1.1",
         "bip174": "^3.0.0",
@@ -91,43 +92,43 @@
         "bytenode": "^1.5.7",
         "chalk": "^5.6.2",
         "cors": "^2.8.6",
-        "datastore-level": "^12.0.4",
+        "datastore-level": "^13.0.1",
         "eslint-plugin-openapi": "^0.0.4",
         "express": "^5.2.1",
         "figlet": "^1.11.0",
-        "interface-datastore": "^9.0.3",
-        "it-length-prefixed-stream": "^2.0.6",
-        "libp2p": "^3.2.4",
+        "interface-datastore": "^10.0.1",
+        "it-length-prefixed-stream": "^3.0.0",
+        "libp2p": "^3.3.0",
         "long": "^5.3.2",
         "lru-cache": "^11.3.6",
         "mongodb": "^7.2.0",
-        "multiformats": "^13.4.2",
+        "multiformats": "^14.0.0",
         "openapi-comment-parser": "^1.0.0",
-        "protobufjs": "^8.0.3",
+        "protobufjs": "^8.2.0",
         "sodium-native": "^5.1.0",
         "ssh2": "^1.17.0",
         "swagger-ui-express": "^5.0.1",
         "tiny-secp256k1": "^2.2.4",
         "toml": "^4.1.1",
         "typescript": "^6.0.3",
-        "typescript-eslint": "^8.59.2",
-        "uint8arraylist": "^2.4.9"
+        "typescript-eslint": "^8.59.3",
+        "uint8arraylist": "^3.0.2"
     },
     "devDependencies": {
         "@types/cors": "^2.8.19",
-        "@types/node": "^25.6.0",
+        "@types/node": "^25.7.0",
         "@types/semver": "^7.7.1",
         "@types/sodium-native": "^2.3.9",
         "@types/ssh2": "^1.15.5",
         "@types/swagger-ui-express": "^4.1.8",
-        "@vitest/coverage-v8": "^4.1.5",
+        "@vitest/coverage-v8": "^4.1.6",
         "eslint": "^10.3.0",
         "gulp": "^5.0.1",
         "gulp-cached": "^1.1.1",
         "gulp-clean": "^0.4.0",
         "gulp-eslint-new": "^2.6.2",
         "patch-package": "^8.0.1",
         "prettier": "^3.8.3",
-        "vitest": "^4.1.5"
+        "vitest": "^4.1.6"
     }
 }

patches/@libp2p+utils+7.1.0.patch

@@ -44,8 +44,25 @@ export class RPCSubWorkerManager extends Logger {
     }
 
     private requestToWorker(data: string): void {
-        const worker = this.getWorker();
-        worker.send(data);
+        const total = this.workers.length;
+        for (let attempt = 0; attempt < total; attempt++) {
+            const worker = this.getWorker();
+            if (!worker.connected || worker.killed || worker.exitCode !== null) continue;
+
+            try {
+                worker.send(data);
+                return;
+            } catch (e) {
+                const code = (e as NodeJS.ErrnoException).code;
+                if (code === 'ERR_IPC_CHANNEL_CLOSED' || code === 'ERR_IPC_DISCONNECTED') {
+                    this.error(`RPC sub-worker pid=${worker.pid} send failed (${code}).`);
+                    continue;
+                }
+                throw e;
+            }
+        }
+
+        this.error('All RPC sub-workers are unavailable; dropping request.');
     }
 
     private createTaskId(): string {
@@ -79,9 +96,22 @@ export class RPCSubWorkerManager extends Logger {
         });
 
         worker.on('exit', (code: number) => {
-            this.error(`Worker exited with code ${code}`);
+            this.error(`Worker pid=${worker.pid} exited with code ${code}; respawning.`);
+            this.replaceWorker(worker);
         });
 
         return worker;
     }
+
+    private replaceWorker(dead: ChildProcess): void {
+        const idx = this.workers.indexOf(dead);
+        if (idx === -1) return;
+
+        try {
+            const fresh = this.createWorker();
+            this.workers[idx] = fresh;
+        } catch (e) {
+            this.error(`Failed to respawn RPC sub-worker: ${(e as Error).stack}`);
+        }
+    }
 }

src/src/blockchain-indexer/rpc/thread/RPCSubWorkerManager.ts

@@ -52,7 +52,7 @@ export class PoC extends Logger {
     ): Promise<ThreadData> {
         switch (m.type) {
             case MessageType.BLOCK_PROCESSED: {
-                return await this.onBlockProcessed(m as BlockProcessedMessage);
+                return this.onBlockProcessed(m as BlockProcessedMessage);
             }
             case MessageType.RPC_METHOD: {
                 return await this.handleRPCMessage(m as RPCMessage<BitcoinRPCThreadMessageType>);
@@ -107,11 +107,8 @@ export class PoC extends Logger {
     }
 
     private async onBlockProcessed(m: BlockProcessedMessage): Promise<ThreadData> {
-        // Wait for previous block to finish so height + proof are always in order.
-        // Use catch so a failed broadcast doesn't permanently jam the lock.
         await this.blockProcessedLock.catch(() => {});
 
-        // Broadcast height to ALL witness instances
         this.blockProcessedLock = this.sendMessageToAllThreads(ThreadTypes.WITNESS, {
             type: MessageType.WITNESS_HEIGHT_UPDATE,
             data: { blockNumber: m.data.blockNumber },
@@ -123,15 +120,13 @@ export class PoC extends Logger {
             this.error(`Failed to broadcast height update: ${(e as Error).stack}`);
         }
 
-        // Round-robin proof generation to ONE witness instance
         void this.sendMessageToThread(ThreadTypes.WITNESS, {
             type: MessageType.WITNESS_BLOCK_PROCESSED,
             data: m.data,
         }).catch((e: unknown) => {
             this.error(`Failed to dispatch WITNESS_BLOCK_PROCESSED: ${(e as Error).stack}`);
         });
 
-        // Update consensus height on this thread
         this.p2p.updateConsensusHeight(m.data.blockNumber);
 
         return {};

src/src/poc/PoC.ts

@@ -23,7 +23,6 @@ import { tcp } from '@libp2p/tcp';
 import { uPnPNAT } from '@libp2p/upnp-nat';
 import { multiaddr, Multiaddr } from '@multiformats/multiaddr';
 import figlet, { FontName } from 'figlet';
-import type { Datastore } from 'interface-datastore';
 import { createLibp2p, ServiceFactoryMap } from 'libp2p';
 import { BtcIndexerConfig } from '../../config/BtcIndexerConfig.js';
 import { DBManagerInstance } from '../../db/DBManager.js';
@@ -87,6 +86,7 @@ import {
     isPrivateOrLoopbackAddress,
 } from './AddressExtractor.js';
 import { Components } from 'libp2p/src/components.js';
+import { Datastore } from 'interface-datastore';
 
 if (Config.P2P.ENABLE_P2P_LOGGING) {
     enable('libp2p:*');
@@ -539,29 +539,99 @@ export class P2PManager extends Logger {
     }
 
     private async broadcastMempoolTransaction(transaction: ITransactionPacket): Promise<number> {
+        const PEER_BROADCAST_TIMEOUT_MS = 8_000;
+
         const broadcastPromises: Promise<void>[] = [];
-        for (const peer of this.peers.values()) {
+        for (const [peerIdStr, peer] of this.peers.entries()) {
             if (!peer.isAuthenticated) continue;
 
-            broadcastPromises.push(peer.broadcastMempoolTransaction(transaction));
+            broadcastPromises.push(
+                this.runPeerOp(
+                    peerIdStr,
+                    PEER_BROADCAST_TIMEOUT_MS,
+                    'Mempool broadcast',
+                    peer.broadcastMempoolTransaction(transaction),
+                ),
+            );
         }
 
         await Promise.safeAll(broadcastPromises);
 
         return broadcastPromises.length;
     }
 
+    /**
+     * Wrap a per-peer fan-out op with a hard timeout and full error isolation.
+     * Every fan-out path through libp2p (mempool gossip, witness broadcast,
+     * witness sync request) shares the same hazard: a single slow / dead /
+     * misbehaving peer would otherwise pin `Promise.safeAll` until the upstream
+     * 240s thread timeout fires and starts cascading. Always resolves; never
+     * rejects, so callers can treat the batch as best-effort.
+     */
+    private runPeerOp(
+        peerIdStr: string,
+        timeoutMs: number,
+        opName: string,
+        op: Promise<unknown>,
+    ): Promise<void> {
+        return new Promise<void>((resolve) => {
+            let settled = false;
+            const timer = setTimeout(() => {
+                if (settled) return;
+                settled = true;
+
+                if (Config.DEBUG_LEVEL >= DebugLevel.DEBUG) {
+                    this.warn(
+                        `${opName} to peer ${peerIdStr} timed out after ${timeoutMs}ms; skipping.`,
+                    );
+                }
+
+                resolve();
+            }, timeoutMs);
+
+            op.then(
+                () => {
+                    if (settled) return;
+                    settled = true;
+                    clearTimeout(timer);
+                    resolve();
+                },
+                (err: unknown) => {
+                    if (settled) return;
+                    settled = true;
+                    clearTimeout(timer);
+
+                    if (Config.DEV_MODE) {
+                        const details = err instanceof Error ? err.message : String(err);
+                        this.warn(`${opName} to peer ${peerIdStr} failed: ${details}`);
+                    }
+
+                    resolve();
+                },
+            );
+        });
+    }
+
     private async requestBlockWitnessesFromPeer(blockNumber: bigint): Promise<void> {
+        const PEER_WITNESS_REQUEST_TIMEOUT_MS = 8_000;
+
         const promises: Promise<void>[] = [];
-        for (const [_peerId, peer] of this.peers) {
+        for (const [peerIdStr, peer] of this.peers.entries()) {
             if (!peer.isAuthenticated) continue;
 
             // We skip asking proofs to light nodes, this is in TODO.
             // TODO: Handle correct proof validations for light nodes.
             const peerMode = peer.peerMode();
             if (peerMode === undefined || peerMode === OPNetIndexerMode.LIGHT) continue;
 
-            promises.push(peer.requestBlockWitnessesFromPeer(blockNumber));
+            promises.push(
+                this.runPeerOp(
+                    peerIdStr,
+                    PEER_WITNESS_REQUEST_TIMEOUT_MS,
+                    'Witness sync request',
+                    peer.requestBlockWitnessesFromPeer(blockNumber),
+                ),
+            );
         }
 
         await Promise.safeAll(promises);
@@ -599,12 +669,19 @@ export class P2PManager extends Logger {
             return;
         }
 
-        // send to all peers
+        const PEER_WITNESS_TIMEOUT_MS = 8_000;
         const promises: Promise<void>[] = [];
-        for (const [_peerId, peer] of this.peers) {
+        for (const [peerIdStr, peer] of this.peers.entries()) {
             if (!peer.isAuthenticated) continue;
 
-            promises.push(peer.sendFromServer(generatedWitness));
+            promises.push(
+                this.runPeerOp(
+                    peerIdStr,
+                    PEER_WITNESS_TIMEOUT_MS,
+                    'Witness broadcast',
+                    peer.sendFromServer(generatedWitness),
+                ),
+            );
         }
 
         await Promise.safeAll(promises);