LibP2P V3

package.json

@@ -48,25 +48,25 @@
     "changelog": "auto-changelog -p"
   },
   "dependencies": {
-    "@chainsafe/libp2p-noise": "^15.1.0",
-    "@chainsafe/libp2p-yamux": "^6.0.2",
+    "@chainsafe/libp2p-noise": "^17.0.0",
+    "@chainsafe/libp2p-yamux": "^8.0.1",
     "@elastic/elasticsearch": "^8.14.0",
-    "@libp2p/autonat": "^2.0.0",
-    "@libp2p/bootstrap": "^10.1.1",
-    "@libp2p/circuit-relay-v2": "^1.1.1",
-    "@libp2p/crypto": "^4.1.5",
-    "@libp2p/dcutr": "^1.1.1",
-    "@libp2p/identify": "^2.1.1",
-    "@libp2p/kad-dht": "^12.1.1",
-    "@libp2p/mdns": "^10.1.1",
-    "@libp2p/peer-id": "^4.1.4",
-    "@libp2p/peer-id-factory": "^4.1.4",
-    "@libp2p/ping": "^1.1.1",
-    "@libp2p/pubsub": "^9.0.22",
-    "@libp2p/pubsub-peer-discovery": "^10.0.2",
-    "@libp2p/tcp": "^9.1.1",
-    "@libp2p/upnp-nat": "^1.2.1",
-    "@libp2p/websockets": "^8.1.1",
+    "@libp2p/autonat": "^3.0.9",
+    "@libp2p/bootstrap": "^12.0.10",
+    "@libp2p/circuit-relay-v2": "^4.1.2",
+    "@libp2p/crypto": "^5.1.13",
+    "@libp2p/dcutr": "^3.0.9",
+    "@libp2p/identify": "^4.0.9",
+    "@libp2p/kad-dht": "^16.1.2",
+    "@libp2p/mdns": "^12.0.10",
+    "@libp2p/peer-id": "^6.0.4",
+    "@libp2p/peer-id-factory": "^4.2.4",
+    "@libp2p/ping": "^3.0.9",
+    "@libp2p/pubsub": "^10.1.18",
+    "@libp2p/pubsub-peer-discovery": "^12.0.0",
+    "@libp2p/tcp": "^11.0.9",
+    "@libp2p/upnp-nat": "^4.0.9",
+    "@libp2p/websockets": "^10.1.2",
     "@multiformats/multiaddr": "^12.2.3",
     "@oceanprotocol/contracts": "^2.5.0",
     "@oceanprotocol/ddo-js": "^0.1.4",
@@ -85,7 +85,7 @@
     "ipaddr.js": "^2.3.0",
     "it-pipe": "^3.0.1",
     "jsonwebtoken": "^9.0.2",
-    "libp2p": "^1.8.0",
+    "libp2p": "^3.1.2",
     "lodash": "^4.17.21",
     "lzma-purejs-requirejs": "^1.0.0",
     "node-cron": "^3.0.3",

src/OceanNode.ts

@@ -6,9 +6,6 @@ import { Database } from './components/database/index.js'
 import { Escrow } from './components/core/utils/escrow.js'
 import { CoreHandlersRegistry } from './components/core/handler/coreHandlersRegistry.js'
 import { OCEAN_NODE_LOGGER } from './utils/logging/common.js'
-import { ReadableString } from './components/P2P/handleProtocolCommands.js'
-import StreamConcat from 'stream-concat'
-import { pipe } from 'it-pipe'
 import { GENERIC_EMOJIS, LOG_LEVELS_STR } from './utils/logging/Logger.js'
 import { BaseHandler } from './components/core/handler/handler.js'
 import { C2DEngines } from './components/c2d/compute_engines.js'
@@ -147,57 +144,30 @@ export class OceanNode {
   }
 
   /**
-   * Use this method to direct calls to the node as node cannot dial into itself
-   * @param message command message
-   * @param sink transform function
+   * v3: Direct protocol command handler - no P2P, just call handler directly
+   * Returns {status, stream} without buffering
+   * @param message - JSON command string
    */
-  async handleDirectProtocolCommand(
-    message: string,
-    sink: any
-  ): Promise<P2PCommandResponse> {
+  async handleDirectProtocolCommand(message: string): Promise<P2PCommandResponse> {
     OCEAN_NODE_LOGGER.logMessage('Incoming direct command for ocean peer', true)
-    let status = null
-    // let statusStream
-    let sendStream = null
-    let response: P2PCommandResponse = null
-
     OCEAN_NODE_LOGGER.logMessage('Performing task: ' + message, true)
 
     try {
       const task = JSON.parse(message)
       const handler: BaseHandler = this.coreHandlers.getHandler(task.command)
-      if (handler === null) {
-        status = {
-          httpStatus: 501,
-          error: 'Unknown command or unexisting handler for command: ' + task.command
-        }
-      } else {
-        response = await handler.handle(task)
-      }
-
-      if (response) {
-        // eslint-disable-next-line prefer-destructuring
-        status = response.status
-        sendStream = response.stream
-      }
 
-      const statusStream = new ReadableString(JSON.stringify(status))
-      if (sendStream == null) {
-        pipe(statusStream, sink)
-      } else {
-        const combinedStream = new StreamConcat([statusStream, sendStream], {
-          highWaterMark: JSON.stringify(status).length
-          // the size of the buffer is important!
-        })
-        pipe(combinedStream, sink)
+      if (!handler) {
+        return {
+          stream: null,
+          status: {
+            httpStatus: 501,
+            error: 'Unknown command or missing handler for: ' + task.command
+          }
+        }
       }
 
-      return (
-        response || {
-          status,
-          stream: null
-        }
-      )
+      // Return response directly without buffering
+      return await handler.handle(task)
     } catch (err) {
       OCEAN_NODE_LOGGER.logMessageWithEmoji(
         'handleDirectProtocolCommands Error: ' + err.message,
@@ -207,8 +177,8 @@ export class OceanNode {
       )
 
       return {
-        status: { httpStatus: 500, error: err.message },
-        stream: null
+        stream: null,
+        status: { httpStatus: 500, error: err.message }
       }
     }
   }

src/components/Indexer/processors/BaseProcessor.ts

@@ -22,7 +22,7 @@ import { getDatabase } from '../../../utils/database.js'
 import { INDEXER_LOGGER } from '../../../utils/logging/common.js'
 import { LOG_LEVELS_STR } from '../../../utils/logging/Logger.js'
 import { URLUtils } from '../../../utils/url.js'
-import { streamToString } from '../../../utils/util.js'
+import { streamToString, streamToUint8Array } from '../../../utils/util.js'
 import ERC721Template from '@oceanprotocol/contracts/artifacts/contracts/templates/ERC721Template.sol/ERC721Template.json' with { type: 'json' }
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import ERC20Template from '@oceanprotocol/contracts/artifacts/contracts/templates/ERC20TemplateEnterprise.sol/ERC20TemplateEnterprise.json' with { type: 'json' }
@@ -359,42 +359,7 @@ export abstract class BaseEventProcessor {
         } else {
           try {
             const p2pNode = await node.getP2PNode()
-            let isBinaryContent = false
-            const sink = async function (source: any) {
-              let first = true
-              for await (const chunk of source) {
-                if (first) {
-                  first = false
-                  try {
-                    const str = uint8ArrayToString(chunk.subarray()) // Obs: we need to specify the length of the subarrays
-                    const decoded = JSON.parse(str)
-                    if ('headers' in decoded) {
-                      if (str?.toLowerCase().includes('application/octet-stream')) {
-                        isBinaryContent = true
-                      }
-                    }
-                    if (decoded.httpStatus !== 200) {
-                      INDEXER_LOGGER.logMessage(
-                        `Error in sink method  : ${decoded.httpStatus} errro: ${decoded.error}`
-                      )
-                      throw new Error('Error in sink method', decoded.error)
-                    }
-                  } catch (e) {
-                    INDEXER_LOGGER.logMessage(
-                      `Error in sink method  } error: ${e.message}`
-                    )
-                    throw new Error(`Error in sink method ${e.message}`)
-                  }
-                } else {
-                  if (isBinaryContent) {
-                    return chunk.subarray()
-                  } else {
-                    const str = uint8ArrayToString(chunk.subarray())
-                    return str
-                  }
-                }
-              }
-            }
+
             const message = {
               command: PROTOCOL_COMMANDS.DECRYPT_DDO,
               transactionId: txId,
@@ -406,12 +371,20 @@ export abstract class BaseEventProcessor {
               signature,
               nonce
             }
-            const response = await p2pNode.sendTo(
-              decryptorURL,
-              JSON.stringify(message),
-              sink
-            )
-            ddo = JSON.parse(await streamToString(response.stream as Readable))
+
+            const response = await p2pNode.sendTo(decryptorURL, JSON.stringify(message))
+
+            if (response.status.httpStatus !== 200) {
+              throw new Error(`Decrypt failed: ${response.status.error}`)
+            }
+
+            if (!response.stream) {
+              throw new Error('No data received from decrypt')
+            }
+
+            // Convert stream to Uint8Array
+            const data = await streamToUint8Array(response.stream as Readable)
+            ddo = JSON.parse(uint8ArrayToString(data))
           } catch (error) {
             const message = `Node exception on decrypt DDO. Status: ${error.message}`
             INDEXER_LOGGER.log(LOG_LEVELS_STR.LEVEL_ERROR, message)