Merge pull request #1356 from oceanprotocol/add-timeout

Add timeout

Author: dnsi0

docs/env.md

@@ -127,6 +127,8 @@ Environmental variables are also tracked in `ENVIRONMENT_VARIABLES` within `src/
 
 ## Compute
 
+- `C2D_DOWNLOAD_TIMEOUT`: Timeout (in seconds) for pulling the algorithm docker image during a C2D job. If the pull exceeds this timeout, the job fails with `PullImageFailed` 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'
@@ -1695,6 +1696,15 @@ export class C2DEngineDocker extends C2DEngine {
     }
   }
 
+  private getImagePullTimeoutMs(): 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
+  }
+
   // eslint-disable-next-line require-await
   private async processJob(job: DBComputeJob) {
     CORE_LOGGER.info(
@@ -2548,6 +2558,7 @@ export class C2DEngineDocker extends C2DEngine {
         )
       }
 
+      pullOptions.abortSignal = AbortSignal.timeout(this.getImagePullTimeoutMs())
       const pullStream = await this.docker.pull(job.containerImage, pullOptions)
       await new Promise((resolve, reject) => {
         let wroteStatusBanner = false

src/components/core/utils/statusHandler.ts

@@ -170,6 +170,9 @@ export async function status(
       CORE_LOGGER.log(LOG_LEVELS_STR.LEVEL_ERROR, `Error getting c2d clusters: ${error}`)
     }
     nodeStatus.supportedSchemas = typesenseSchemas.ddoSchemas
+  } else {
+    delete nodeStatus.c2dClusters
+    delete nodeStatus.supportedSchemas
   }
 
   if (config.persistentStorage) {

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