add download timeout

Author: dnsi0

docs/env.md

@@ -127,6 +127,8 @@ Environmental variables are also tracked in `ENVIRONMENT_VARIABLES` within `src/
 
 ## Compute
 
+- `C2D_DOWNLOAD_TIMEOUT`: Per-file timeout (in seconds) for downloading the algorithm and each input asset during a C2D job. If a download exceeds this timeout, the job fails instead of getting stuck. Defaults to `900` (15 minutes). Example: `900`
+
 The `DOCKER_COMPUTE_ENVIRONMENTS` environment variable is used to configure Docker-based compute environments in Ocean Node. This guide will walk you through the options available for defining `DOCKER_COMPUTE_ENVIRONMENTS` and how to set it up correctly. For configuring compute environments and setting prices for each resource (including pricing units and examples), see [Compute pricing](compute-pricing.md).
 
 Example Configuration

src/components/c2d/compute_engine_docker.ts

@@ -49,6 +49,7 @@ import {
 } from 'fs'
 import { pipeline } from 'node:stream/promises'
 import { CORE_LOGGER } from '../../utils/logging/common.js'
+import { ENVIRONMENT_VARIABLES } from '../../utils/constants.js'
 import { AssetUtils } from '../../utils/asset.js'
 import { FindDdoHandler } from '../core/handler/ddoHandler.js'
 import { OceanNode } from '../../OceanNode.js'
@@ -1694,6 +1695,31 @@ export class C2DEngineDocker extends C2DEngine {
     }
   }
 
+  private getDownloadTimeoutMs(): number {
+    const raw = ENVIRONMENT_VARIABLES.C2D_DOWNLOAD_TIMEOUT.value
+    const parsed = raw ? parseInt(raw, 10) : NaN
+    if (Number.isFinite(parsed) && parsed > 0) {
+      return parsed * 1000
+    }
+    return 15 * 60 * 1000
+  }
+
+  private async pipelineWithDownloadTimeout(
+    source: NodeJS.ReadableStream,
+    destination: NodeJS.WritableStream
+  ): Promise<void> {
+    const timeoutMs = this.getDownloadTimeoutMs()
+    const controller = new AbortController()
+    const timer = setTimeout(() => {
+      controller.abort(new Error(`Download timed out after ${timeoutMs / 1000}s`))
+    }, timeoutMs)
+    try {
+      await pipeline(source, destination, { signal: controller.signal })
+    } finally {
+      clearTimeout(timer)
+    }
+  }
+
   // eslint-disable-next-line require-await
   private async processJob(job: DBComputeJob) {
     CORE_LOGGER.info(
@@ -2883,7 +2909,7 @@ export class C2DEngineDocker extends C2DEngine {
         }
 
         if (storage) {
-          await pipeline(
+          await this.pipelineWithDownloadTimeout(
             (await storage.getReadableStream()).stream,
             createWriteStream(fullAlgoPath)
           )
@@ -2992,7 +3018,7 @@ export class C2DEngineDocker extends C2DEngine {
         const fullPath = jobFolderPath + '/data/inputs/' + fileInfo[0].name
         appendFileSync(configLogPath, `Downloading asset to ${fullPath}\n`)
         try {
-          await pipeline(
+          await this.pipelineWithDownloadTimeout(
             (await storage.getReadableStream()).stream,
             createWriteStream(fullPath)
           )

src/utils/constants.ts

@@ -531,6 +531,11 @@ export const ENVIRONMENT_VARIABLES: Record<any, EnvVariable> = {
     name: 'PERSISTENT_STORAGE',
     value: process.env.PERSISTENT_STORAGE,
     required: false
+  },
+  C2D_DOWNLOAD_TIMEOUT: {
+    name: 'C2D_DOWNLOAD_TIMEOUT',
+    value: process.env.C2D_DOWNLOAD_TIMEOUT,
+    required: false
   }
 }
 export const CONNECTION_HISTORY_DELETE_THRESHOLD = 300