orphane jobs

src/components/c2d/compute_engine_docker.ts

@@ -1975,12 +1975,18 @@ export class C2DEngineDocker extends C2DEngine {
       }
 
       if (jobs.length > 0) {
-        const promises: any = []
-        for (const job of jobs) {
-          promises.push(this.processJob(job))
-        }
-        // wait for all promises, there is no return
-        await Promise.all(promises)
+        // allSettled, not all: one throwing job used to abort the rest of this tick —
+        // service health check, expiry sweep and GPU snapshot below all stopped running.
+        const results = await Promise.allSettled(jobs.map((job) => this.processJob(job)))
+        results.forEach((result, i) => {
+          if (result.status === 'rejected') {
+            CORE_LOGGER.error(
+              `processJob ${jobs[i].jobId} failed: ${
+                result.reason?.message ?? result.reason
+              }`
+            )
+          }
+        })
       }
 
       // Service-on-Demand health check: catch Running services whose container died on its
@@ -2169,6 +2175,22 @@ export class C2DEngineDocker extends C2DEngine {
       `Process job ${job.jobId} started: [STATUS: ${job.status}: ${job.statusText}]`
     )
 
+    // An env id is a hash of its fees, so editing fees renames the env and strands every
+    // job already pointing at the old id. Such a job can never run — fail it once here
+    // rather than let each stage below retry or throw on the undefined env.
+    const env = this.envs.find((e) => e.id === job.environment)
+    if (!env) {
+      job.status = C2DStatusNumber.ContainerCreationFailed
+      job.statusText = `Environment ${job.environment} no longer exists`
+      job.isRunning = false
+      job.dateFinished = String(Date.now() / 1000)
+      CORE_LOGGER.error(`Job ${job.jobId} failed: ${job.statusText}`)
+      this.recordJobFinished(job)
+      await this.db.updateJob(job)
+      await this.cleanupJob(job)
+      return
+    }
+
     // has to :
     //  - monitor running containers and stop them if over limits
     //  - monitor disc space and clean up
@@ -2238,6 +2260,7 @@ export class C2DEngineDocker extends C2DEngine {
       try {
         const chainId = job.payment && job.payment.chainId ? job.payment.chainId : null
         const allEnvs = await this.getComputeEnvironments(chainId)
+        // chain-filtered view of the same env (fees narrowed to job.payment.chainId)
         const env = allEnvs.find((e) => e.id === job.environment)
         if (!env) throw new Error(`Environment ${job.environment} not found`)
         await this.checkIfResourcesAreAvailable(job.resources, env, job.isFree, allEnvs)
@@ -2298,8 +2321,7 @@ export class C2DEngineDocker extends C2DEngine {
       // create the volume & create container
       // TO DO C2D:  Choose driver & size
       // get environment-specific resources for Docker device/hardware configuration
-      const env = this.envs.find((e) => e.id === job.environment)
-      const envResource = env?.resources || []
+      const envResource = env.resources || []
       const volume: VolumeCreateOptions = {
         Name: job.jobId + '-volume'
       }

src/test/unit/c2d/orphanedJobEnv.test.ts

@@ -0,0 +1,64 @@
+import { expect } from 'chai'
+import sinon from 'sinon'
+import { C2DEngineDocker } from '../../../components/c2d/compute_engine_docker.js'
+import { C2DStatusNumber } from '../../../@types/C2D/C2D.js'
+
+const HASH = 'cluster-1'
+const LIVE = `${HASH}-0xlive`
+
+// Keep the prototype, skip the Docker constructor (same as serviceRestartRace.test.ts).
+function makeEngine(): any {
+  const engine: any = Object.create(C2DEngineDocker.prototype)
+  engine.clusterConfig = { hash: HASH, connection: { resources: [] } }
+  engine.envs = [{ id: LIVE, resources: [], enableNetwork: true }]
+  engine.db = {
+    updateJob: sinon.stub().resolves(1),
+    getRunningJobs: sinon.stub().resolves([]),
+    getPendingServiceStarts: sinon.stub().resolves([]),
+    getExpiredServiceJobs: sinon.stub().resolves([])
+  }
+  engine.recordJobFinished = sinon.stub()
+  engine.cleanupJob = sinon.stub().resolves(undefined)
+  engine.activeBuildAborts = new Map()
+  engine.stopped = false
+  engine.isInternalLoopRunning = false
+  engine.cronTimer = null
+  engine.setNewTimer = sinon.stub()
+  engine.checkRunningServices = sinon.stub().resolves([])
+  engine.logMetricsSummary = sinon.stub()
+  engine.refreshHostGpuSnapshot = sinon.stub().resolves(undefined)
+  return engine
+}
+
+const makeJob = (environment: string): any => ({
+  jobId: 'job-1',
+  environment,
+  status: C2DStatusNumber.ConfiguringVolumes,
+  statusText: 'Configuring volumes',
+  isRunning: true,
+  dateCreated: String(Date.now() / 1000),
+  resources: []
+})
+
+describe('C2D job whose environment was renamed away', () => {
+  it('fails the job instead of throwing on the undefined env', async () => {
+    const engine = makeEngine()
+    const job = makeJob(`${HASH}-0xgone`)
+
+    await engine.processJob(job)
+
+    expect(job.isRunning).to.equal(false)
+    expect(job.statusText).to.contain('no longer exists')
+    expect(engine.cleanupJob.calledOnce).to.equal(true)
+  })
+
+  it('finishes the tick even when a job throws', async () => {
+    const engine = makeEngine()
+    engine.db.getRunningJobs.resolves([makeJob(LIVE)])
+    sinon.stub(engine, 'processJob').rejects(new Error('boom'))
+
+    await engine.InternalLoop()
+
+    expect(engine.db.getExpiredServiceJobs.calledOnce).to.equal(true)
+  })
+})