diff --git a/src/@types/C2D/C2D.ts b/src/@types/C2D/C2D.ts index 74bd3e267..467b654da 100644 --- a/src/@types/C2D/C2D.ts +++ b/src/@types/C2D/C2D.ts @@ -181,9 +181,25 @@ export interface ComputeEnvironment extends ComputeEnvironmentBaseConfig { consumerAddress: string // v1 queuedJobs: number queuedFreeJobs: number + /** + * Seconds of remaining queue wait, **summed across every queued job** in this environment + * (not a per-job maximum, despite the name). Each job contributes + * `max(0, queueMaxWaitTime - elapsed since it was created)`. + */ queMaxWaitTime: number + /** As `queMaxWaitTime`, summed over queued *free* jobs only. */ queMaxWaitTimeFree: number + /** + * Seconds of remaining runtime, **summed across every running job** in this environment + * (not a per-job maximum, despite the name). Each job contributes + * `max(0, maxJobDuration - elapsed since it started)`; a job that has been allocated but + * has not started executing yet contributes its full `maxJobDuration`. + * + * Because it is a sum, a client cannot derive "when does capacity free up" from this field: + * 15 jobs with 1 second left each report 15, the same as one job with 15 seconds left. + */ runMaxWaitTime: number + /** As `runMaxWaitTime`, summed over running *free* jobs only. */ runMaxWaitTimeFree: number } diff --git a/src/components/c2d/compute_engine_base.ts b/src/components/c2d/compute_engine_base.ts index 977ceb126..bb10f585f 100644 --- a/src/components/c2d/compute_engine_base.ts +++ b/src/components/c2d/compute_engine_base.ts @@ -17,7 +17,7 @@ import type { ComputeEnvFees } from '../../@types/C2D/C2D.js' import type { ServiceJob } from '../../@types/C2D/ServiceOnDemand.js' -import { C2DClusterType } from '../../@types/C2D/C2D.js' +import { C2DClusterType, C2DStatusNumber } from '../../@types/C2D/C2D.js' import { C2DDatabase } from '../database/C2DDatabase.js' import { Escrow } from '../core/utils/escrow.js' import { KeyManager } from '../KeyManager/index.js' @@ -30,6 +30,24 @@ import { ValidateParams } from '../httpRoutes/validateCommands.js' import { EncryptMethod } from '../../@types/fileObject.js' import { CORE_LOGGER } from '../../utils/logging/common.js' import { DockerRegistryAuthSchema } from '../../utils/config/schemas.js' + +/** + * Parse a job timestamp stored as decimal seconds in a string column. + * + * Job timestamps are initialized to the *string* `'0'` (not null/empty) and only get a real + * value once the corresponding phase starts. `'0'` is truthy, so a truthiness check happily + * accepts it and `Number.parseFloat('0')` then places the job's start at the Unix epoch, + * producing ~56 years of "elapsed" time. Always go through this helper: it returns 0 for the + * `'0'` sentinel, for null/undefined/empty, and for anything non-finite or negative, so + * callers only have to test `> 0` to know whether a real timestamp is present. + */ +export function parseJobTimestamp(raw?: string): number { + if (!raw) return 0 + const parsed = Number.parseFloat(raw) + if (!Number.isFinite(parsed) || parsed <= 0) return 0 + return parsed +} + export abstract class C2DEngine { private clusterConfig: C2DClusterInfo public db: C2DDatabase @@ -570,6 +588,36 @@ export abstract class C2DEngine { } } + /** + * Seconds of runtime budget a job still has, clamped to [0, maxJobDuration]. + * + * Build time counts against `maxJobDuration`, consistent with the runtime-expiry check in + * the docker engine, so `buildStartTimestamp` wins when both are set. A job that has been + * allocated but has not started ticking yet (no valid timestamp on either field — e.g. it + * is still in PullImage/BuildImage/ConfiguringVolumes) still has its whole budget ahead of + * it, so report the full `maxJobDuration` rather than measuring from the epoch. + */ + protected getJobRemainingRuntimeSeconds(job: DBComputeJob, nowSec: number): number { + const budget = Number.isFinite(job?.maxJobDuration) ? job.maxJobDuration : 0 + const buildStart = parseJobTimestamp(job?.buildStartTimestamp) + const start = buildStart > 0 ? buildStart : parseJobTimestamp(job?.algoStartTimestamp) + if (start === 0) return budget + return Math.max(0, budget - (nowSec - start)) + } + + /** + * Seconds a queued job may still wait before its queue request expires, clamped to + * [0, queueMaxWaitTime]. Mirrors the queue-expiry check in the docker engine. As with the + * runtime helper, a missing/sentinel `dateCreated` means "not measurable yet", so report the + * full requested wait instead of measuring from the epoch. + */ + protected getJobRemainingQueueWaitSeconds(job: DBComputeJob, nowSec: number): number { + const requested = Number.isFinite(job?.queueMaxWaitTime) ? job.queueMaxWaitTime : 0 + const created = parseJobTimestamp(job?.dateCreated) + if (created === 0) return requested + return Math.max(0, requested - (nowSec - created)) + } + public async getUsedResources(env: ComputeEnvironment): Promise { const usedResources: { [x: string]: any } = {} const usedFreeResources: { [x: string]: any } = {} @@ -591,32 +639,38 @@ export abstract class C2DEngine { let maxRunningTime = 0 let maxRunningTimeFree = 0 + // one instant for the whole response, so every job in it is measured against the same clock + const nowSec = Date.now() / 1000 + for (const job of jobs) { const isThisEnv = job.environment === env.id - const isRunning = job.queueMaxWaitTime === 0 + // A job holds resources from the moment it leaves the queue. JobQueued is the only + // pre-allocation state; every later state means volumes/containers exist or are being + // created. queueMaxWaitTime is the caller's *requested* wait, not live state, and is + // never reset on release — do not use it as a liveness flag. + const isQueued = job.status === C2DStatusNumber.JobQueued if (isThisEnv) { - if (isRunning) { - const timeElapsed = job.buildStartTimestamp - ? new Date().getTime() / 1000 - Number.parseFloat(job?.buildStartTimestamp) - : new Date().getTime() / 1000 - Number.parseFloat(job?.algoStartTimestamp) - totalJobs++ - maxRunningTime += job.maxJobDuration - timeElapsed + if (isQueued) { + const waitLeft = this.getJobRemainingQueueWaitSeconds(job, nowSec) + queuedJobs++ + maxWaitTime += waitLeft if (job.isFree) { - totalFreeJobs++ - maxRunningTimeFree += job.maxJobDuration - timeElapsed + queuedFreeJobs++ + maxWaitTimeFree += waitLeft } } else { - queuedJobs++ - maxWaitTime += job.maxJobDuration + const runtimeLeft = this.getJobRemainingRuntimeSeconds(job, nowSec) + totalJobs++ + maxRunningTime += runtimeLeft if (job.isFree) { - queuedFreeJobs++ - maxWaitTimeFree += job.maxJobDuration + totalFreeJobs++ + maxRunningTimeFree += runtimeLeft } } } - if (isRunning) { + if (!isQueued) { for (const resource of job.resources) { const envRes = envResourceMap.get(resource.id) if (envRes) { @@ -731,7 +785,15 @@ export abstract class C2DEngine { // Gate 1 (per-env ceiling) — fungible resources only. // envResource.total = env aggregate ceiling (from EnvironmentResourceRef.total). - if (isFungible && envResource.total - (envResource.inUse ?? 0) < request.amount) + // Both operands are defaulted: a resource object missing either field would make the + // subtraction NaN, and `NaN < amount` is false, i.e. the gate would silently admit the + // request. Missing must deny, not admit. (`total` is required by the type, but these + // objects also arrive from JSON config and older persisted shapes, which the compiler + // cannot vouch for — free-resource entries with no `total` have been seen in the field.) + if ( + isFungible && + (envResource.total ?? 0) - (envResource.inUse ?? 0) < request.amount + ) throw new Error(`Not enough available ${request.id} in this environment`) // Gate 2 (engine-wide pool ceiling) — fungible + exclusive discrete. @@ -744,7 +806,11 @@ export abstract class C2DEngine { if (!env.free) throw new Error(`No free resources`) envResource = this.getResource(env.free?.resources, request.id) if (!envResource) throw new Error(`No such free resource ${request.id}`) - if (envResource.total - envResource.inUse < request.amount) + // Same NaN-admits hazard as gate 1, and more reachable here: `inUse` is optional on + // ComputeResource, and free-resource entries carrying `max`/`inUse` but no `total` + // have been observed on live nodes (builds predating free-resource pool resolution). + // Unguarded, that yields NaN and allows unlimited free allocation. + if ((envResource.total ?? 0) - (envResource.inUse ?? 0) < request.amount) throw new Error(`Not enough available ${request.id} for free`) } } diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index 0c250addb..b48ee05ab 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -27,7 +27,7 @@ import type { EnvironmentResourceRef } from '../../@types/C2D/C2D.js' import { BASE_CHAIN_ID, USDC_TOKEN_ADDRESS_BASE } from '../../utils/config.js' -import { C2DEngine } from './compute_engine_base.js' +import { C2DEngine, parseJobTimestamp } from './compute_engine_base.js' import { C2DDatabase } from '../database/C2DDatabase.js' import { Escrow } from '../core/utils/escrow.js' import { create256Hash } from '../../utils/crypt.js' @@ -2407,10 +2407,15 @@ export class C2DEngineDocker extends C2DEngine { let expiry const buildDuration = this.getValidBuildDurationSeconds(job) + // Never measure from the '0' sentinel: parseFloat('0') would place the start at the + // Unix epoch and make an otherwise healthy container look instantly expired. If the + // algo start is not recorded yet, treat the clock as starting now — the next sweep + // will see the real timestamp. + const algoStart = parseJobTimestamp(job.algoStartTimestamp) || timeNow if (buildDuration > 0) { // if job has build time, reduce the remaining algorithm runtime budget - expiry = parseFloat(job.algoStartTimestamp) + job.maxJobDuration - buildDuration - } else expiry = parseFloat(job.algoStartTimestamp) + job.maxJobDuration + expiry = algoStart + job.maxJobDuration - buildDuration + } else expiry = algoStart + job.maxJobDuration CORE_LOGGER.debug( 'container running since timeNow: ' + timeNow + ' , Expiry: ' + expiry ) @@ -4741,13 +4746,11 @@ export class C2DEngineDocker extends C2DEngine { } private getValidBuildDurationSeconds(job: DBComputeJob): number { - const startRaw = job.buildStartTimestamp - const stopRaw = job.buildStopTimestamp - if (!startRaw || !stopRaw) return 0 - const start = Number.parseFloat(startRaw) - const stop = Number.parseFloat(stopRaw) - if (!Number.isFinite(start) || !Number.isFinite(stop)) return 0 - if (start <= 0) return 0 + // parseJobTimestamp returns 0 for the '0' sentinel, empty/missing values and anything + // non-finite, so a job that never built an image reports no build duration. + const start = parseJobTimestamp(job.buildStartTimestamp) + const stop = parseJobTimestamp(job.buildStopTimestamp) + if (start === 0 || stop === 0) return 0 if (stop < start) return 0 return stop - start } diff --git a/src/components/core/handler/getJobs.ts b/src/components/core/handler/getJobs.ts index df0bf05d5..c8be29541 100644 --- a/src/components/core/handler/getJobs.ts +++ b/src/components/core/handler/getJobs.ts @@ -4,13 +4,24 @@ import { CORE_LOGGER } from '../../../utils/logging/common.js' import { buildInvalidRequestMessage } from '../../httpRoutes/validateCommands.js' import { CommandHandler } from './handler.js' import { P2PCommandResponse } from '../../../@types/OceanNode.js' +import { parseFromTimestampSeconds } from '../utils/timestamps.js' export class GetJobsHandler extends CommandHandler { validate(command: GetJobsCommand) { - if (command.fromTimestamp && typeof command.fromTimestamp !== 'string') { - return buildInvalidRequestMessage( - 'Parameter : "fromTimestamp" is not a valid string' - ) + // absent / empty means "no filter", as before + if (command.fromTimestamp) { + if (typeof command.fromTimestamp !== 'string') { + return buildInvalidRequestMessage( + 'Parameter : "fromTimestamp" is not a valid string' + ) + } + // Reject unparseable values instead of passing them to SQL, where they used to match + // nothing and return 200 + [] — indistinguishable from "no jobs in that window". + if (!Number.isFinite(parseFromTimestampSeconds(command.fromTimestamp))) { + return buildInvalidRequestMessage( + `Parameter "fromTimestamp" is not a valid date: "${command.fromTimestamp}" — use an ISO date or a Unix timestamp` + ) + } } return { valid: true } } @@ -27,9 +38,14 @@ export class GetJobsHandler extends CommandHandler { throw new Error('C2D database not initialized') } + // The DB columns store decimal seconds; validate() has already rejected anything + // unparseable, so this is either a finite seconds value or undefined (no filter). + const fromTimestamp = task.fromTimestamp + ? parseFromTimestampSeconds(task.fromTimestamp) + : undefined const jobs = await c2d.getJobs( task.environments, - task.fromTimestamp, + fromTimestamp ?? undefined, task.consumerAddrs, undefined, task.runningJobs diff --git a/src/components/core/service/getServices.ts b/src/components/core/service/getServices.ts index c5fdb3e4c..e1e98cad9 100644 --- a/src/components/core/service/getServices.ts +++ b/src/components/core/service/getServices.ts @@ -12,20 +12,11 @@ import { type ServiceJob } from '../../../@types/C2D/ServiceOnDemand.js' import { toListedServiceJob } from './utils.js' +import { parseFromTimestamp } from '../utils/timestamps.js' -// Parses the `fromTimestamp` filter into Unix milliseconds. Accepts an ISO date string -// or a Unix timestamp (seconds or milliseconds) given as a string / number-like string. -// Returns undefined for "no filter" and null for an unparseable value (caller → 400). -export function parseFromTimestamp(value?: string): number | undefined | null { - if (value === undefined || value === null || value === '') return undefined - if (/^\d+$/.test(String(value))) { - const n = Number(value) - // 1e12 ms ≈ Sep 2001; any plausible seconds value is far below it - return n > 1e12 ? n : n * 1000 - } - const t = Date.parse(String(value)) - return Number.isNaN(t) ? null : t -} +// Re-exported for backwards compatibility: the parser now lives in core/utils/timestamps.ts +// so the compute-jobs listing can share it. +export { parseFromTimestamp } // SERVICE_LIST: the node-wide service listing, shaped like GetJobsHandler. Default (no // filters) returns exactly what the engines count against the shared resource pools diff --git a/src/components/core/utils/timestamps.ts b/src/components/core/utils/timestamps.ts new file mode 100644 index 000000000..da9f62835 --- /dev/null +++ b/src/components/core/utils/timestamps.ts @@ -0,0 +1,29 @@ +// Normalization for caller-supplied `fromTimestamp`-style query filters. +// +// This is for *request parameters*, not for stored job timestamps: use +// `parseJobTimestamp` (compute_engine_base.ts) to read a persisted timestamp column, which +// treats the `'0'` sentinel as "not set". Here `0`/garbage must be distinguishable from +// "no filter" so the handler can answer 400 instead of silently returning an empty list. + +// Parses the `fromTimestamp` filter into Unix milliseconds. Accepts an ISO date string +// or a Unix timestamp (seconds or milliseconds) given as a string / number-like string. +// Returns undefined for "no filter" and null for an unparseable value (caller → 400). +export function parseFromTimestamp(value?: string): number | undefined | null { + if (value === undefined || value === null || value === '') return undefined + if (/^\d+$/.test(String(value))) { + const n = Number(value) + // 1e12 ms ≈ Sep 2001; any plausible seconds value is far below it + return n > 1e12 ? n : n * 1000 + } + const t = Date.parse(String(value)) + return Number.isNaN(t) ? null : t +} + +// Same parse, expressed in Unix *seconds* — the unit the compute_jobs table stores its +// dateCreated/dateFinished columns in. Kept as an explicit wrapper so no call site has to +// guess whether it is holding seconds or milliseconds. +export function parseFromTimestampSeconds(value?: string): number | undefined | null { + const ms = parseFromTimestamp(value) + if (ms === undefined || ms === null) return ms + return ms / 1000 +} diff --git a/src/components/database/C2DDatabase.ts b/src/components/database/C2DDatabase.ts index b19d30dea..0261d990a 100755 --- a/src/components/database/C2DDatabase.ts +++ b/src/components/database/C2DDatabase.ts @@ -130,9 +130,10 @@ export class C2DDatabase extends AbstractDatabase { return await this.provider.getFinishedJobs(environments) } + /** @param fromTimestamp lower bound in Unix seconds — see SQLiteCompute.getJobs. */ async getJobs( environments?: string[], - fromTimestamp?: string, + fromTimestamp?: number, consumerAddrs?: string[], status?: C2DStatusNumber, runningJobs?: boolean diff --git a/src/components/database/sqliteCompute.ts b/src/components/database/sqliteCompute.ts index 7511350b5..33b217643 100644 --- a/src/components/database/sqliteCompute.ts +++ b/src/components/database/sqliteCompute.ts @@ -22,8 +22,10 @@ interface ComputeDatabaseProvider { getFinishedJobs(environments?: string[]): Promise getJobs( environments?: string[], - fromTimestamp?: string, - consumerAddrs?: string[] + fromTimestamp?: number, + consumerAddrs?: string[], + status?: C2DStatusNumber, + runningJobs?: boolean ): Promise updateImage(image: string): Promise getOldImages(retentionDays: number): Promise @@ -713,16 +715,24 @@ export class SQLiteCompute implements ComputeDatabaseProvider { return [] } + /** + * @param fromTimestamp lower bound in Unix **seconds** (the unit this table's + * dateCreated/dateFinished TEXT columns store). Callers must normalize before getting + * here — see parseFromTimestampSeconds in core/utils/timestamps.ts. Taking a number + * rather than a string is deliberate: a raw string bound to these columns is compared + * with memcmp, which happens to work for 10-digit seconds and silently matches nothing + * for milliseconds, ISO dates or garbage. + */ async getJobs( environments?: string[], - fromTimestamp?: string, + fromTimestamp?: number, consumerAddrs?: string[], status?: C2DStatusNumber, runningJobs?: boolean ): Promise { let selectSQL = `SELECT * FROM ${this.schema.name}` - const params: string[] = [] + const params: Array = [] const conditions: string[] = [] if (environments && environments.length > 0) { @@ -731,19 +741,23 @@ export class SQLiteCompute implements ComputeDatabaseProvider { params.push(...environments) } + // dateCreated/dateFinished are TEXT holding decimal seconds, so they must be CAST for + // the comparison to be numeric instead of lexicographic. if (runningJobs) { conditions.push(`status = ?`) params.push(C2DStatusNumber.RunningAlgorithm.toString()) - if (fromTimestamp) { - conditions.push(`dateCreated >= ?`) + if (fromTimestamp !== undefined && fromTimestamp !== null) { + conditions.push(`CAST(dateCreated AS REAL) >= ?`) params.push(fromTimestamp) } } else { - if (fromTimestamp) { - conditions.push(`dateFinished >= ?`) + if (fromTimestamp !== undefined && fromTimestamp !== null) { + conditions.push(`CAST(dateFinished AS REAL) >= ?`) params.push(fromTimestamp) } - if (status) { + // C2DStatusNumber.JobStarted is 0, so a truthiness check would silently drop the + // filter and return every status instead. + if (status !== undefined && status !== null) { conditions.push(`status = ?`) params.push(status.toString()) } @@ -758,7 +772,10 @@ export class SQLiteCompute implements ComputeDatabaseProvider { if (conditions.length > 0) { selectSQL += ` WHERE ${conditions.join(' AND ')}` } - selectSQL += ` ORDER BY dateCreated DESC` + // Numeric ordering for the same reason as the comparison above. NOTE: do NOT copy this + // CAST to the service_jobs table — its dateCreated is an ISO string, which sorts + // correctly as text and would CAST to its year. + selectSQL += ` ORDER BY CAST(dateCreated AS REAL) DESC` return await this.doQuery(selectSQL, params, environments) } diff --git a/src/test/unit/c2d/usedResources.test.ts b/src/test/unit/c2d/usedResources.test.ts new file mode 100644 index 000000000..0c1606d76 --- /dev/null +++ b/src/test/unit/c2d/usedResources.test.ts @@ -0,0 +1,389 @@ +import { assert, expect } from 'chai' +import sinon from 'sinon' +import { Readable } from 'stream' +import { + C2DClusterType, + C2DStatusNumber, + ComputeEnvironment, + ComputeJob, + DBComputeJob +} from '../../../@types/C2D/C2D.js' +import { + C2DEngine, + parseJobTimestamp +} from '../../../components/c2d/compute_engine_base.js' +import { C2DDatabase } from '../../../components/database/C2DDatabase.js' +import { ValidateParams } from '../../../components/httpRoutes/validateCommands.js' +import { ServiceJob } from '../../../@types/C2D/ServiceOnDemand.js' + +const CLUSTER_HASH = '0xcluster' +const ENV_ID = 'test-env' +// Frozen clock, so every remaining-time assertion can be exact instead of tolerance-based. +const NOW_SEC = 1785760660 +const NOW_MS = NOW_SEC * 1000 + +/* eslint-disable require-await */ +class TestEngine extends C2DEngine { + constructor(jobs: DBComputeJob[], serviceJobs: ServiceJob[] = []) { + super( + { type: C2DClusterType.DOCKER, hash: CLUSTER_HASH }, + { + getRunningJobs: () => Promise.resolve(jobs), + getRunningServiceJobs: () => Promise.resolve(serviceJobs) + } as unknown as C2DDatabase, + null, + null, + null + ) + } + + async getComputeEnvironments(): Promise { + return [] + } + + async checkDockerImage(): Promise { + return { valid: true, reason: null as string, status: 200 } + } + + async startComputeJob(): Promise { + return [] + } + + async stopComputeJob(): Promise { + return [] + } + + async getComputeJobStatus(): Promise { + return [] + } + + async getComputeJobResult(): Promise<{ stream: Readable; headers: any }> { + return null + } + + async cleanupExpiredStorage(): Promise { + return true + } +} +/* eslint-enable require-await */ + +function makeEnv(): ComputeEnvironment { + return { + id: ENV_ID, + resources: [ + { id: 'cpu', kind: 'fungible', type: 'cpu', total: 64, max: 64, min: 1, inUse: 0 }, + { id: 'ram', kind: 'fungible', type: 'ram', total: 62, max: 62, min: 1, inUse: 0 } + ], + runningJobs: 0, + runningfreeJobs: 0, + queuedJobs: 0, + queuedFreeJobs: 0, + queMaxWaitTime: 0, + queMaxWaitTimeFree: 0, + runMaxWaitTime: 0, + runMaxWaitTimeFree: 0, + consumerAddress: '0x0', + fees: {}, + access: { addresses: [], accessLists: null }, + platform: { architecture: 'x86_64', os: 'linux' }, + minJobDuration: 60, + maxJobDuration: 3600, + maxJobs: 20 + } as unknown as ComputeEnvironment +} + +function makeJob(overrides: Partial = {}): DBComputeJob { + return { + clusterHash: CLUSTER_HASH, + jobId: 'job-1', + owner: '0xowner', + environment: ENV_ID, + status: C2DStatusNumber.RunningAlgorithm, + maxJobDuration: 3600, + queueMaxWaitTime: 0, + isFree: false, + isRunning: true, + dateCreated: String(NOW_SEC), + // both timestamps default to the '0' sentinel exactly as newJob() writes them + buildStartTimestamp: '0', + algoStartTimestamp: '0', + resources: [ + { id: 'cpu', amount: 2 }, + { id: 'ram', amount: 4 } + ], + ...overrides + } as unknown as DBComputeJob +} + +async function used(jobs: DBComputeJob[], serviceJobs: ServiceJob[] = []) { + return await new TestEngine(jobs, serviceJobs).getUsedResources(makeEnv()) +} + +describe('parseJobTimestamp', () => { + it("returns 0 for the '0' sentinel and for missing / malformed values", () => { + // '0' is truthy: this is the whole reason the helper exists + expect(parseJobTimestamp('0')).to.equal(0) + expect(parseJobTimestamp('0.0')).to.equal(0) + expect(parseJobTimestamp('')).to.equal(0) + expect(parseJobTimestamp(undefined)).to.equal(0) + expect(parseJobTimestamp(null as unknown as string)).to.equal(0) + expect(parseJobTimestamp('abc')).to.equal(0) + expect(parseJobTimestamp('-5')).to.equal(0) + }) + + it('returns the parsed value for a real timestamp', () => { + expect(parseJobTimestamp('1785760660.961')).to.equal(1785760660.961) + }) +}) + +describe('C2DEngine.getUsedResources', () => { + let clock: sinon.SinonStub + + beforeEach(() => { + clock = sinon.stub(Date, 'now').returns(NOW_MS) + }) + + afterEach(() => { + clock.restore() + }) + + it('reports the full budget for a job with both timestamps still at the sentinel', async () => { + const res = await used([makeJob()]) + assert(res.maxRunningTime === 3600, `expected 3600, got ${res.maxRunningTime}`) + assert(res.totalJobs === 1, `expected 1 running job, got ${res.totalJobs}`) + }) + + it('does not report an epoch-sized negative for 15 sentinel jobs (reported bug)', async () => { + const jobs = Array.from({ length: 15 }, (_, i) => makeJob({ jobId: `job-${i}` })) + const res = await used(jobs) + // pre-fix this was -26786355914.46 (15 × (3600 - now)) + assert(res.maxRunningTime > 0, `must never be negative, got ${res.maxRunningTime}`) + // sum semantics: 15 jobs × full 3600s budget + assert(res.maxRunningTime === 54000, `expected 54000, got ${res.maxRunningTime}`) + assert(res.totalJobs === 15, `expected 15 running jobs, got ${res.totalJobs}`) + }) + + it('uses algoStartTimestamp when buildStartTimestamp is the sentinel (pull path)', async () => { + // The sharpest case: pre-fix the truthy '0' shadowed this valid timestamp and the + // remaining time came out around -1.79e9. + const res = await used([ + makeJob({ buildStartTimestamp: '0', algoStartTimestamp: String(NOW_SEC - 100) }) + ]) + assert(res.maxRunningTime === 3500, `expected 3500, got ${res.maxRunningTime}`) + }) + + it('prefers buildStartTimestamp when both are set (build time counts to the budget)', async () => { + const res = await used([ + makeJob({ + buildStartTimestamp: String(NOW_SEC - 100), + algoStartTimestamp: String(NOW_SEC - 10) + }) + ]) + assert(res.maxRunningTime === 3500, `expected 3500, got ${res.maxRunningTime}`) + }) + + it('clamps an overdue job to 0 instead of publishing a negative', async () => { + const res = await used([ + makeJob({ algoStartTimestamp: String(NOW_SEC - 7200), maxJobDuration: 3600 }) + ]) + assert(res.maxRunningTime === 0, `expected 0, got ${res.maxRunningTime}`) + }) + + it('treats malformed timestamps as not-yet-started, never NaN', async () => { + for (const bad of ['abc', '', undefined, '-100']) { + const res = await used([ + makeJob({ buildStartTimestamp: bad, algoStartTimestamp: bad }) + ]) + assert( + !Number.isNaN(res.maxRunningTime), + `maxRunningTime must not be NaN for ${JSON.stringify(bad)}` + ) + assert( + res.maxRunningTime === 3600, + `expected 3600 for ${JSON.stringify(bad)}, got ${res.maxRunningTime}` + ) + } + }) + + it('counts a JobQueued job as queued, with no resources held', async () => { + const res = await used([ + makeJob({ + status: C2DStatusNumber.JobQueued, + queueMaxWaitTime: 600, + dateCreated: String(NOW_SEC - 100) + }) + ]) + assert(res.queuedJobs === 1, `expected 1 queued job, got ${res.queuedJobs}`) + assert(res.totalJobs === 0, `expected 0 running jobs, got ${res.totalJobs}`) + // remaining queue wait, not the runtime budget: 600 - 100 + assert(res.maxWaitTime === 500, `expected 500, got ${res.maxWaitTime}`) + expect(res.usedResources).to.deep.equal({}) + }) + + it('reports the full queue wait for a queued job with no usable dateCreated', async () => { + const res = await used([ + makeJob({ + status: C2DStatusNumber.JobQueued, + queueMaxWaitTime: 600, + dateCreated: '0' + }) + ]) + assert(res.maxWaitTime === 600, `expected 600, got ${res.maxWaitTime}`) + }) + + it('counts a released job (queueMaxWaitTime > 0, status running) as running and holding resources', async () => { + // Regression test for the queueMaxWaitTime-as-liveness-flag bug: queueMaxWaitTime is + // never reset on release, so this job used to stay in queuedJobs forever with its + // cpu/ram invisible to the availability gate. + const res = await used([ + makeJob({ + status: C2DStatusNumber.RunningAlgorithm, + queueMaxWaitTime: 600, + algoStartTimestamp: String(NOW_SEC - 100) + }) + ]) + assert(res.totalJobs === 1, `expected 1 running job, got ${res.totalJobs}`) + assert(res.queuedJobs === 0, `expected 0 queued jobs, got ${res.queuedJobs}`) + assert(res.maxRunningTime === 3500, `expected 3500, got ${res.maxRunningTime}`) + expect(res.usedResources).to.deep.equal({ cpu: 2, ram: 4 }) + }) + + it('counts mid-pipeline states (PullImage/BuildImage/ConfiguringVolumes) as holding resources', async () => { + for (const status of [ + C2DStatusNumber.PullImage, + C2DStatusNumber.BuildImage, + C2DStatusNumber.ConfiguringVolumes + ]) { + const res = await used([makeJob({ status })]) + expect(res.usedResources, `status ${status}`).to.deep.equal({ cpu: 2, ram: 4 }) + assert(res.totalJobs === 1, `status ${status}: expected 1 running job`) + } + }) + + it('separates free from paid usage', async () => { + const res = await used([ + makeJob({ + jobId: 'paid', + algoStartTimestamp: String(NOW_SEC - 100), + isFree: false + }), + makeJob({ + jobId: 'free', + algoStartTimestamp: String(NOW_SEC - 600), + isFree: true + }) + ]) + expect(res.usedResources).to.deep.equal({ cpu: 4, ram: 8 }) + expect(res.usedFreeResources).to.deep.equal({ cpu: 2, ram: 4 }) + assert(res.totalJobs === 2, `expected 2 running jobs, got ${res.totalJobs}`) + assert(res.totalFreeJobs === 1, `expected 1 free job, got ${res.totalFreeJobs}`) + // paid+free: 3500 + 3000 ; free only: 3000 + assert(res.maxRunningTime === 6500, `expected 6500, got ${res.maxRunningTime}`) + assert( + res.maxRunningTimeFree === 3000, + `expected 3000, got ${res.maxRunningTimeFree}` + ) + }) + + it('accumulates remaining runtime as a sum across running jobs', async () => { + // Pins the documented (surprising) sum semantics so a refactor cannot silently switch + // this field to a per-job maximum. + const res = await used([ + makeJob({ + jobId: 'a', + maxJobDuration: 3600, + algoStartTimestamp: String(NOW_SEC - 3500) + }), + makeJob({ + jobId: 'b', + maxJobDuration: 3600, + algoStartTimestamp: String(NOW_SEC - 1600) + }), + makeJob({ + jobId: 'c', + maxJobDuration: 3600, + algoStartTimestamp: String(NOW_SEC - 600) + }) + ]) + // 100 + 2000 + 3000 + assert(res.maxRunningTime === 5100, `expected 5100, got ${res.maxRunningTime}`) + }) +}) + +describe('C2DEngine.checkIfResourcesAreAvailable availability gates', () => { + function envWithFree(freeResources: any[]): ComputeEnvironment { + const env = makeEnv() + env.resources[0].inUse = 0 + ;(env as any).free = { + resources: freeResources, + maxJobs: 100, + access: { addresses: [], accessLists: null } + } + return env + } + + async function rejectionOf(promise: Promise): Promise { + try { + await promise + return null + } catch (e) { + return e.message + } + } + + it('rejects an oversized paid request when the env resource has no total/inUse', async () => { + // Gate 1 has the same NaN-admits hazard as the free gate below: unguarded, + // `undefined - 0 < amount` is false and the request is silently admitted. + const engine = new TestEngine([]) + for (const cpu of [ + { id: 'cpu', kind: 'fungible' }, + { id: 'cpu', kind: 'fungible', inUse: 4 } + ]) { + const env = makeEnv() + ;(env as any).resources = [cpu] + const message = await rejectionOf( + // no allEnvironments → only gate 1 applies + engine.checkIfResourcesAreAvailable([{ id: 'cpu', amount: 8 }], env, false) + ) + assert( + message === 'Not enough available cpu in this environment', + `expected gate 1 to reject ${JSON.stringify(cpu)}, got ${message}` + ) + } + }) + + it('rejects an oversized free request when the free resource has no total/inUse', async () => { + // A sparse free-resource entry made `undefined - undefined < amount` evaluate to NaN < + // amount === false, silently passing the gate and allowing unlimited free allocation. + const engine = new TestEngine([]) + for (const freeRes of [ + [{ id: 'cpu' }], + [{ id: 'cpu', inUse: 4 }], + [{ id: 'cpu', total: undefined as number, inUse: undefined as number }] + ]) { + const message = await rejectionOf( + engine.checkIfResourcesAreAvailable( + [{ id: 'cpu', amount: 8 }], + envWithFree(freeRes), + true + ) + ) + assert( + message === 'Not enough available cpu for free', + `expected the free gate to reject ${JSON.stringify(freeRes)}, got ${message}` + ) + } + }) + + it('still admits a request that fits the declared free capacity', async () => { + const engine = new TestEngine([]) + const message = await rejectionOf( + engine.checkIfResourcesAreAvailable( + [{ id: 'cpu', amount: 8 }], + envWithFree([{ id: 'cpu', total: 16, inUse: 4 }]), + true + ) + ) + assert(message === null, `expected no rejection, got ${message}`) + }) +}) diff --git a/src/test/unit/database/getJobsFilters.test.ts b/src/test/unit/database/getJobsFilters.test.ts new file mode 100644 index 000000000..6d2ac8639 --- /dev/null +++ b/src/test/unit/database/getJobsFilters.test.ts @@ -0,0 +1,304 @@ +import { assert, expect } from 'chai' +import { C2DDatabase } from '../../../components/database/C2DDatabase.js' +import { typesenseSchemas } from '../../../components/database/TypesenseSchemas.js' +import { getConfiguration } from '../../../utils/config.js' +import { + C2DStatusNumber, + C2DStatusText, + ComputeAlgorithm, + ComputeAsset, + DBComputeJob +} from '../../../@types/C2D/C2D.js' +import { + buildEnvOverrideConfig, + OverrideEnvConfig, + setupEnvironment, + tearDownEnvironment, + TEST_ENV_CONFIG_FILE +} from '../../utils/utils.js' +import { ENVIRONMENT_VARIABLES, PROTOCOL_COMMANDS } from '../../../utils/constants.js' +import { OceanNodeConfig } from '../../../@types/OceanNode.js' +import { GetJobsHandler } from '../../../components/core/handler/getJobs.js' +import { GetJobsCommand } from '../../../@types/commands.js' +import { + parseFromTimestamp, + parseFromTimestampSeconds +} from '../../../components/core/utils/timestamps.js' + +// dateCreated/dateFinished are TEXT holding decimal seconds — the format the compute_jobs +// table actually stores. The fixture values are chosen so a lexicographic (memcmp) +// comparison gives a *different* answer than a numeric one for every input format below: +// +// bound 1785760660 (s) / "1785760660000" (ms) / "2026-…" (ISO) +// NEW_FINISHED "1785760660.961" — numerically above the bound, but as text below the ms +// bound ('.' 0x2E < '0' 0x30 at index 10) and below ISO +// NINE_DIGIT "999999999.5" — numerically far below the bound (year 2001), but as +// text above every one of the three bounds ('9' > '1','2') +const BOUND_SEC = 1785760660 +const OLD_SEC = BOUND_SEC - 7200 +const NEW_FINISHED = '1785760660.961' +const NINE_DIGIT = '999999999.5' + +// The C2D SQLite file survives between test runs, so every environment/jobId below is +// namespaced per run: each test then sees only the rows it seeded. +const RUN = String(Date.now()) +const envName = (base: string) => `${base}-${RUN}` + +const algorithm: ComputeAlgorithm = { documentId: 'did:op:1', serviceId: '0xabc' } +const dataset: ComputeAsset = { documentId: 'did:op:2', serviceId: '0xdef' } + +function baseJob(overrides: Partial): DBComputeJob { + return { + owner: '0xe2DD09d719Da89e5a3D0F2549c7E24566e947260', + jobId: null, + jobIdHash: null, + dateCreated: null, + dateFinished: null, + status: C2DStatusNumber.JobFinished, + statusText: C2DStatusText.JobFinished, + results: null, + inputDID: [], + maxJobDuration: 3600, + clusterHash: 'clusterHash', + configlogURL: null, + publishlogURL: null, + algologURL: null, + outputsURL: null, + stopRequested: false, + algorithm, + assets: [dataset], + isRunning: false, + isStarted: false, + containerImage: 'image', + resources: [], + environment: 'unset', + agreementId: '0xagreement', + isFree: false, + algoStartTimestamp: '0', + algoStopTimestamp: '0', + algoDuration: 0, + queueMaxWaitTime: 0, + ...overrides + } as unknown as DBComputeJob +} + +describe('getJobs filters', () => { + let envOverrides: OverrideEnvConfig[] + let config: OceanNodeConfig + let db: C2DDatabase = null + + before(async () => { + envOverrides = buildEnvOverrideConfig( + [ENVIRONMENT_VARIABLES.DOCKER_COMPUTE_ENVIRONMENTS], + [ + '[{"socketPath":"/var/run/docker.sock","environments":[{"storageExpiry":604800,"maxJobDuration":3600,"minJobDuration":60,"resources":[{"id":"cpu","total":4,"max":4,"min":1,"type":"cpu"}],"fees":{"1":[{"feeToken":"0x123","prices":[{"id":"cpu","price":1}]}]}}]}]' + ] + ) + envOverrides = await setupEnvironment(TEST_ENV_CONFIG_FILE, envOverrides) + config = await getConfiguration(true) + db = await new C2DDatabase(config.dbConfig, typesenseSchemas.c2dSchemas) + }) + + after(async () => { + await tearDownEnvironment(envOverrides) + }) + + // Seeds finished jobs into a run-private environment, with dateCreated == dateFinished so + // one fixture serves both the fromTimestamp filter (dateFinished) and the ordering + // assertion (dateCreated). + async function seedFinished(envId: string, finishedSecs: string[]): Promise { + const ids: string[] = [] + for (const [i, finished] of finishedSecs.entries()) { + const jobId = await db.newJob( + baseJob({ + jobId: `${envId}-${i}`, + environment: envId, + dateCreated: finished + }) + ) + // dateFinished is only written by updateJob + await db.updateJob( + baseJob({ + jobId, + environment: envId, + dateCreated: finished, + dateFinished: finished + }) + ) + ids.push(jobId) + } + return ids + } + + describe('fromTimestamp is compared numerically', () => { + const envId = envName('env-fromTimestamp') + + before(async () => { + await seedFinished(envId, [String(OLD_SEC), NINE_DIGIT, NEW_FINISHED]) + }) + + // Only NEW_FINISHED is numerically at/after the bound, whichever format the caller used. + async function expectOnlyTheNewRow(fromTimestamp: number) { + const jobs = await db.getJobs([envId], fromTimestamp) + assert( + jobs.length === 1, + `expected 1 job, got ${jobs.length}: ${jobs.map((j) => j.dateFinished).join()}` + ) + assert( + jobs[0].dateFinished === NEW_FINISHED, + `expected ${NEW_FINISHED}, got ${jobs[0].dateFinished}` + ) + } + + it('accepts Unix seconds — pre-fix this also matched the 9-digit row', async () => { + await expectOnlyTheNewRow(BOUND_SEC) + }) + + it('accepts Unix milliseconds — pre-fix this missed the matching row', async () => { + const fromTimestamp = parseFromTimestampSeconds(String(BOUND_SEC * 1000)) + await expectOnlyTheNewRow(fromTimestamp as number) + }) + + it('accepts an ISO date string — pre-fix this missed the matching row', async () => { + const iso = new Date(BOUND_SEC * 1000).toISOString() + const fromTimestamp = parseFromTimestampSeconds(iso) + await expectOnlyTheNewRow(fromTimestamp as number) + }) + + it('returns everything when no fromTimestamp is given', async () => { + const jobs = await db.getJobs([envId]) + assert(jobs.length === 3, `expected 3 jobs, got ${jobs.length}`) + }) + }) + + it('includes a row sitting exactly on the bound', async () => { + const envId = envName('env-boundary') + // the filter is >=, so an exact match must be returned + await seedFinished(envId, [String(BOUND_SEC)]) + const jobs = await db.getJobs([envId], BOUND_SEC) + assert(jobs.length === 1, `expected 1 job, got ${jobs.length}`) + }) + + it('orders by numeric dateCreated, not lexicographically', async () => { + const envId = envName('env-ordering') + // 9-digit vs 10-digit: as text "999999999.5" sorts above "1785760660.961", so pre-fix + // the OLDER job came back first under ORDER BY dateCreated DESC. + const tenDigits = NEW_FINISHED + await seedFinished(envId, [NINE_DIGIT, tenDigits]) + const jobs = await db.getJobs([envId]) + assert(jobs.length === 2, `expected 2 jobs, got ${jobs.length}`) + assert( + jobs[0].dateCreated === tenDigits, + `newest first: expected ${tenDigits}, got ${jobs[0].dateCreated}` + ) + }) + + describe('status filter', () => { + const envId = envName('env-status') + + before(async () => { + // JobStarted is 0 — the value a truthiness guard drops + await db.newJob( + baseJob({ + jobId: `${envId}-started`, + environment: envId, + status: C2DStatusNumber.JobStarted, + statusText: C2DStatusText.JobStarted, + dateCreated: String(BOUND_SEC) + }) + ) + await db.newJob( + baseJob({ + jobId: `${envId}-finished`, + environment: envId, + status: C2DStatusNumber.JobFinished, + statusText: C2DStatusText.JobFinished, + dateCreated: String(BOUND_SEC + 1) + }) + ) + }) + + it('filters on status 0 (JobStarted) — pre-fix this returned every status', async () => { + const jobs = await db.getJobs( + [envId], + undefined, + undefined, + C2DStatusNumber.JobStarted + ) + assert(jobs.length === 1, `expected 1 job, got ${jobs.length}`) + assert(jobs[0].status === C2DStatusNumber.JobStarted) + }) + + it('filters on a non-zero status', async () => { + const jobs = await db.getJobs( + [envId], + undefined, + undefined, + C2DStatusNumber.JobFinished + ) + assert(jobs.length === 1, `expected 1 job, got ${jobs.length}`) + assert(jobs[0].status === C2DStatusNumber.JobFinished) + }) + + it('does not treat an omitted status as a filter', async () => { + const jobs = await db.getJobs([envId]) + assert(jobs.length === 2, `expected 2 jobs, got ${jobs.length}`) + }) + }) +}) + +describe('GetJobsHandler.validate fromTimestamp', () => { + // validate() does not touch the node, so a null node is enough here + const handler = new GetJobsHandler(null) + + function command(fromTimestamp?: any): GetJobsCommand { + return { + command: PROTOCOL_COMMANDS.JOBS, + fromTimestamp + } as GetJobsCommand + } + + it('rejects an unparseable fromTimestamp instead of returning an empty list', () => { + const result = handler.validate(command('abc')) + assert(result.valid === false, 'expected validation to fail') + assert(result.status === 400, `expected 400, got ${result.status}`) + expect(result.reason).to.contain('not a valid date') + }) + + it('accepts seconds, milliseconds and ISO dates', () => { + for (const value of [ + String(BOUND_SEC), + String(BOUND_SEC * 1000), + new Date(BOUND_SEC * 1000).toISOString() + ]) { + const result = handler.validate(command(value)) + assert(result.valid === true, `expected ${value} to be accepted`) + } + }) + + it('treats an absent or empty fromTimestamp as no filter', () => { + assert(handler.validate(command(undefined)).valid === true) + assert(handler.validate(command('')).valid === true) + }) + + it('rejects a non-string fromTimestamp', () => { + const result = handler.validate(command(12345)) + assert(result.valid === false, 'expected validation to fail') + }) +}) + +describe('timestamp query-parameter parsing', () => { + it('distinguishes "no filter" from "unparseable"', () => { + expect(parseFromTimestamp(undefined)).to.equal(undefined) + expect(parseFromTimestamp('')).to.equal(undefined) + expect(parseFromTimestamp('abc')).to.equal(null) + }) + + it('converts to seconds for the compute jobs table', () => { + expect(parseFromTimestampSeconds(String(BOUND_SEC))).to.equal(BOUND_SEC) + expect(parseFromTimestampSeconds(String(BOUND_SEC * 1000))).to.equal(BOUND_SEC) + expect(parseFromTimestampSeconds(new Date(BOUND_SEC * 1000).toISOString())).to.equal( + BOUND_SEC + ) + }) +})