From 8c470a35ab0dad1637d0eafac771f80d6e5afef7 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 14:57:24 +0300 Subject: [PATCH 1/8] PoC service status --- src/@types/C2D/ServiceOnDemand.ts | 77 +++++ src/components/c2d/compute_engine_docker.ts | 258 +++++++++++++++- src/components/c2d/modelDownload.ts | 252 +++++++++++++++ src/components/c2d/serviceEngines.ts | 159 ++++++++++ src/components/c2d/serviceReadiness.ts | 194 ++++++++++++ src/components/core/service/utils.ts | 17 +- src/components/database/C2DDatabase.ts | 25 +- src/components/database/sqliteCompute.ts | 76 +++-- src/test/integration/services.test.ts | 11 + .../unit/service/serviceReadiness.test.ts | 286 ++++++++++++++++++ 10 files changed, 1331 insertions(+), 24 deletions(-) create mode 100644 src/components/c2d/modelDownload.ts create mode 100644 src/components/c2d/serviceEngines.ts create mode 100644 src/components/c2d/serviceReadiness.ts create mode 100644 src/test/unit/service/serviceReadiness.test.ts diff --git a/src/@types/C2D/ServiceOnDemand.ts b/src/@types/C2D/ServiceOnDemand.ts index 644cedcfe..712ed5111 100644 --- a/src/@types/C2D/ServiceOnDemand.ts +++ b/src/@types/C2D/ServiceOnDemand.ts @@ -124,6 +124,76 @@ export interface ServiceOnDemandConfig { allowImageBuild?: boolean // default: false — gates Dockerfile-based services per daemon } +// ── Readiness + startup progress ────────────────────────────────────── + +/** + * Whether a service can actually serve requests, as opposed to merely having a running container. + * + * `Running` says the container process started. An inference engine then spends minutes downloading + * weights and warming up, during which its forwarded port either refuses connections or answers 503 + * — so a consumer handed the endpoint at `Running` gets nothing but errors. The node closes that gap + * by asking the workload itself, on a schedule, and reporting the answer here. + * + * Only reported for workloads the node recognizes (see components/c2d/serviceEngines). For anything + * else the field is ABSENT, which every client must read as "this node cannot tell me" and fall back + * to treating `Running` as usable — the behaviour that predates this feature. + */ +export type ServiceReadinessState = + | 'waiting' // not answering as expected yet — the normal warm-up window + | 'ready' // answered the engine's readiness request + | 'failing' // it WAS ready and stopped answering + +export interface ServiceReadiness { + state: ServiceReadinessState + engine: string // which profile decided this ('vllm'), so a client can say what was checked + readySince?: number // Unix ms of the first successful check of THIS container + lastCheckedAt?: number // Unix ms + consecutiveFailures?: number + httpStatus?: number // last response status (absent when the connection itself failed) + lastError?: string // owner-only: stripped from SERVICE_LIST + probedUrl?: string // owner-only: which candidate address answered (diagnostics) +} + +/** + * Live progress of the image pull, aggregated from the Docker daemon's own per-layer byte + * counts. Written only while the job sits in PullImage, and kept afterwards as the record of + * what was downloaded (a cached image never produces one — absence means "already on the node"). + * + * `totalBytes` is the sum of the layer totals Docker has ANNOUNCED so far, which grows as + * layers start, so `percent` is clamped monotonic rather than recomputed each tick. + */ +export interface ServiceImagePullProgress { + phase: 'downloading' | 'extracting' | 'complete' + downloadedBytes: number + totalBytes: number + percent: number + layersTotal: number + layersDone: number + updatedAt: number // Unix ms +} + +/** + * How much of its model a recognized engine has downloaded, read from the container's own cache. + * + * This is the wait the image pull does NOT cover: the image is pulled once per node and cached + * forever after, while the weights are fetched on every container start, by the engine, after it + * reports Running. + * + * `totalBytes`/`percent` are present only when the size could be established — the engine is + * serving a Hugging Face repo AND the Hub published a safetensors index for it. Pointed at a local + * path, an object-store URI or an unindexed repo, only `downloadedBytes` is reported and the client + * shows an indeterminate bar rather than a ratio against a guess. + */ +export interface ServiceModelDownload { + modelId?: string // the repo being fetched, when it is a Hub id + downloadedBytes: number + totalBytes?: number + percent?: number + filesComplete: number + filesInFlight: number // the hub fetches files in parallel, so only the aggregate is meaningful + updatedAt: number // Unix ms +} + // ── Runtime service job ─────────────────────────────────────────────── export interface ServiceEndpoint { @@ -219,4 +289,11 @@ export interface ServiceJob { // Best-effort Docker/NVML runtime metrics sampled while the service container runs. // DB-only: stripped from every public response by toPublicServiceJob / toListedServiceJob. runtimeMetrics?: ContainerMetricsSnapshot + // Whether the workload can serve requests yet, for engines the node recognizes. Absent otherwise, + // which clients read as "not reported" and fall back to treating Running as usable. + readiness?: ServiceReadiness + // Image pull byte progress, written while the job is in PullImage. + imagePull?: ServiceImagePullProgress + // Model-weight download progress, sampled from the container's cache while it warms up. + modelDownload?: ServiceModelDownload } diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index c8da48e51..0039c7069 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -77,7 +77,26 @@ import { ServiceStatusText, SERVICE_START_PENDING_STATUSES } from '../../@types/C2D/ServiceOnDemand.js' -import type { ServiceJob } from '../../@types/C2D/ServiceOnDemand.js' +import type { + ServiceJob, + ServiceImagePullProgress, + ServiceModelDownload, + ServiceReadiness +} from '../../@types/C2D/ServiceOnDemand.js' +import { + ImagePullTracker, + probeCandidates, + runReadinessProbe, + PROBE_INITIAL_DELAY_SECONDS, + PROBE_PERIOD_SECONDS, + READY_PROBE_PERIOD_SECONDS +} from './serviceReadiness.js' +import { resolveServiceEngine, type ServiceEngineProfile } from './serviceEngines.js' +import { + buildModelDownload, + fetchModelTotalBytes, + readModelDownloadBytes +} from './modelDownload.js' import type { DockerMountObject } from '../../@types/PersistentStorage.js' import { resolveServiceImage } from './serviceResourceMatching.js' import { @@ -181,6 +200,10 @@ export class C2DEngineDocker extends C2DEngine { // process; the service_locks DB lease (acquired alongside it) extends the guarantee // across processes sharing the same DB file + Docker daemon. private serviceOpsInFlight: Set = new Set() + // serviceId -> the address the readiness probe last reached the container on. Which address + // works depends on how this node is deployed (on the host, or itself in Docker), so the probe + // discovers it once and reuses it instead of walking the candidate list every few seconds. + private serviceProbeUrls: Map = new Map() private readonly serviceLockHolderId: string = makeServiceLockHolderId() private serviceLockHeartbeatTimer: NodeJS.Timeout | null = null // The in-flight service lifecycle promises — processServiceStart() launched @@ -3727,7 +3750,8 @@ export class C2DEngineDocker extends C2DEngine { private async pullImageRef( imageRef: string, encryptedDockerRegistryAuth?: string, - logFile?: string + logFile?: string, + onProgress?: (progress: ServiceImagePullProgress) => void ): Promise { const controller = new AbortController() const timer = setTimeout(() => controller.abort(), this.getImagePullTimeoutMs()) @@ -3766,6 +3790,10 @@ export class C2DEngineDocker extends C2DEngine { } const pullStream = await this.docker.pull(imageRef, pullOptions) + // Per-layer byte accounting, aggregated from the daemon's own progress events. Only built + // when someone is listening (service starts) — the compute path passes no callback and so + // keeps its previous, allocation-free behaviour. + const tracker = onProgress ? new ImagePullTracker(onProgress) : null await new Promise((resolve, reject) => { this.docker.modem.followProgress( pullStream, @@ -3774,10 +3802,12 @@ export class C2DEngineDocker extends C2DEngine { if (logFile) appendFileSync(logFile, String(err.message)) return reject(err) } + tracker?.finish() resolve() }, (progress: any) => { if (logFile) appendFileSync(logFile, (progress.status ?? '') + '\n') + tracker?.onEvent(progress) } ) }) @@ -3979,6 +4009,12 @@ export class C2DEngineDocker extends C2DEngine { resources: resources.map((r) => ({ id: r.id, amount: r.amount })), payment } + // Stated up front, so a client polling a service that has not reached Running yet can already + // tell "this node will report readiness for this workload" from "it never will". + const engine = resolveServiceEngine(job) + if (engine) { + job.readiness = { state: 'waiting', engine: engine.id } + } await this.db.newServiceJob(job) return job } @@ -4097,7 +4133,28 @@ export class C2DEngineDocker extends C2DEngine { job.status = ServiceStatusNumber.PullImage job.statusText = ServiceStatusText[ServiceStatusNumber.PullImage] await this.db.updateServiceJob(job) - await this.pullImageRef(job.containerImage) + // Persist the daemon's own byte counts as the pull runs, so a client polling status can + // show real progress on what is usually the longest visible wait (multi-GB engine + // images). Writes are chained rather than fired in parallel, and drained BEFORE the + // status moves on — an in-flight write still carrying status=PullImage must never land + // after the Claiming transition and resurrect the old status. + let pullWrites: Promise = Promise.resolve() + await this.pullImageRef( + job.containerImage, + undefined, + undefined, + (progress) => { + job.imagePull = progress + pullWrites = pullWrites + .then(() => this.db.updateServiceJob(job)) + .catch((e: any) => + CORE_LOGGER.debug( + `service ${serviceId}: pull progress write failed: ${e.message}` + ) + ) + } + ) + await pullWrites } } catch (e: any) { imageError = e @@ -4435,10 +4492,192 @@ export class C2DEngineDocker extends C2DEngine { await this.markServiceFailed(job, reason, details.State) return } - // Container healthy: sample live runtime metrics (throttled, best-effort, lease-free). + // Container up: ask the workload itself whether it can serve requests yet (throttled, + // best-effort, lease-free), then sample live runtime metrics on the same terms. + await this.probeServiceReadiness(job, details) await this.sampleAndPersistServiceMetrics(job) } + /** + * Asks a recognized workload whether it can serve requests yet, and persists the answer. + * + * "Running" only means the container process started. An inference engine then spends minutes + * downloading weights and warming up, during which its forwarded port either refuses connections + * or answers 503 — so a consumer handed the URL at Running gets nothing but errors. The engine is + * the only thing that knows when that ends, and this is the node asking it, from outside the + * container, on the consumer's behalf. + * + * Which request to make is engine-specific and comes from the node's own table + * (components/c2d/serviceEngines). An image the node does not recognize is left alone entirely: + * `readiness` stays absent, and clients keep treating Running as usable exactly as before. + * + * Same discipline as the metrics sampler: no lifecycle lease, skipped while a lifecycle op is in + * flight, persisted through a guarded readiness-ONLY write, and it never throws. + */ + private async probeServiceReadiness( + job: ServiceJob, + details: Dockerode.ContainerInspectInfo + ): Promise { + try { + const engine = resolveServiceEngine(job) + if (!engine) return + if (this.serviceOpsInFlight.has(job.serviceId)) return + + const now = Date.now() + const wasReady = job.readiness?.state === 'ready' + // Hold off until the container has had a moment to bind its port — probing a process that + // has not called listen() yet only produces noise in the node's own logs. + const startedAt = Date.parse(details.State?.StartedAt ?? '') + if ( + !wasReady && + Number.isFinite(startedAt) && + now - startedAt < PROBE_INITIAL_DELAY_SECONDS * 1000 + ) { + return + } + // The loop ticks every couple of seconds; the probe runs on its own, slower period. Once + // ready it slows right down — from then on it is a liveness check, not a wait. + const periodMs = + (wasReady ? READY_PROBE_PERIOD_SECONDS : PROBE_PERIOD_SECONDS) * 1000 + if (job.readiness?.lastCheckedAt && now - job.readiness.lastCheckedAt < periodMs) { + return + } + + const port = engine.probe.port ?? job.exposedPorts[0] + if (!port) return + const containerIps = Object.values(details.NetworkSettings?.Networks ?? {}) + .map((net: any) => net?.IPAddress) + .filter((ip: string) => !!ip) + const candidates = probeCandidates(job, containerIps, port, engine.probe.path) + // The address that answered last time, tried first — but never exclusively: a container + // restart can change its IP, so a cached candidate that stops connecting falls back to the + // full list on the same tick. + const cached = this.serviceProbeUrls.get(job.serviceId) + const ordered = cached + ? [cached, ...candidates.filter((url) => url !== cached)] + : candidates + + let result = null + for (const url of ordered) { + const attempt = await runReadinessProbe(url, engine.probe.expectStatus) + if (attempt.ok) { + result = attempt + this.serviceProbeUrls.set(job.serviceId, url) + break + } + // A real HTTP answer means we reached the workload — it simply is not ready. That is a + // conclusive result, so stop here rather than reporting a later candidate's connection + // error, which would read as "unreachable" instead of "still warming up". + if (attempt.httpStatus !== undefined) { + result = attempt + this.serviceProbeUrls.set(job.serviceId, url) + break + } + result = result ?? attempt + } + if (!result) return + + const readiness: ServiceReadiness = result.ok + ? { + state: 'ready', + engine: engine.id, + readySince: job.readiness?.readySince ?? now, + lastCheckedAt: now, + consecutiveFailures: 0, + httpStatus: result.httpStatus, + probedUrl: result.url + } + : { + // Only a service that HAD answered can be "failing"; one that never has is still + // warming up, which is the normal state for the first minutes of a model server. + state: wasReady ? 'failing' : 'waiting', + engine: engine.id, + readySince: job.readiness?.readySince, + lastCheckedAt: now, + consecutiveFailures: (job.readiness?.consecutiveFailures ?? 0) + 1, + httpStatus: result.httpStatus, + lastError: result.error, + probedUrl: result.url + } + + if (result.ok !== wasReady) { + CORE_LOGGER.info( + `[readiness] service ${job.serviceId} (${engine.id}): ${ + job.readiness?.state ?? 'unknown' + } -> ${readiness.state} via ${result.url}${ + result.error ? ` (${result.error})` : '' + }` + ) + } + + // While the engine is still warming up, sample how far its model download has got. Skipped + // once ready: the files are there, and the walk is pure overhead from then on. + const modelDownload = + readiness.state === 'ready' + ? undefined + : await this.sampleModelDownload(job, engine) + + // Same cross-process guard as the metrics write: a lifecycle transition must win. + if ( + this.serviceOpsInFlight.has(job.serviceId) || + (await this.db.isServiceLocked(job.serviceId, SERVICE_LOCK_STALE_MS)) + ) { + return + } + await this.db.updateServiceJobReadiness( + job.serviceId, + { + owner: job.owner, + clusterHash: job.clusterHash, + status: ServiceStatusNumber.Running, + containerId: job.containerId + }, + readiness, + modelDownload + ) + } catch (e: any) { + CORE_LOGGER.debug( + `[readiness] service ${job.serviceId}: probe failed: ${e?.message}` + ) + } + } + + /** + * How much of its model the container has pulled down, read from the engine's own cache. + * + * This is the wait the image pull does not cover: the image is fetched once per node and cached + * forever after, while the weights are fetched by the engine on EVERY container start, after it + * reports Running. Nothing serves that number over HTTP at the time (there is no server yet), so + * it is read from the cache files themselves — see components/c2d/modelDownload. + * + * Best-effort throughout: an unreadable cache, an engine with no known cache path, or a model + * whose size the Hub cannot tell us all degrade to less information, never to an error. + */ + private async sampleModelDownload( + job: ServiceJob, + engine: ServiceEngineProfile + ): Promise { + if (!engine.modelCachePath) return undefined + try { + const container = this.docker.getContainer(job.containerId) + const downloaded = await readModelDownloadBytes(container, engine.modelCachePath) + if (!downloaded) return undefined + // Only a Hugging Face repo has a size the node can look up; a local path or an object-store + // URI reports bytes with no total, which clients render as an indeterminate bar. + const modelId = engine.modelIdFromCommand?.(job.dockerCmd) ?? null + // A GGUF engine names the exact file it pulls; asking for that instead of the whole repo is + // the difference between a real denominator and one covering every quantization on offer. + const quant = engine.modelQuantFromCommand?.(job.dockerCmd) ?? undefined + const totalBytes = modelId ? await fetchModelTotalBytes(modelId, quant) : null + return buildModelDownload(downloaded, totalBytes, modelId) + } catch (e: any) { + CORE_LOGGER.debug( + `[model-download] service ${job.serviceId}: sample failed: ${e?.message}` + ) + return undefined + } + } + // Resolves a service's requested allocation as bytes/cores for the metrics snapshot. // Services have no /data volume, so disk usage comes from the container writable layer // (SizeRw) rather than a quota — diskBytes is left 0 (no quota to report a % against). @@ -5007,6 +5246,17 @@ export class C2DEngineDocker extends C2DEngine { // Reset the metrics accumulators: the new container is a fresh process, so peak memory // and CPU deltas must not carry over from the outgoing one. job.runtimeMetrics = undefined + // Same for readiness: the replacement container has to earn "ready" again (an Edit relaunch + // re-downloads the model), and reporting the outgoing container's ready state would hand the + // user a live endpoint minutes before there is one. The probe SPEC is kept — a restart reuses + // the stored container spec, so it reuses the check that matches it. + const restartEngine = resolveServiceEngine(job) + job.readiness = restartEngine + ? { state: 'waiting', engine: restartEngine.id } + : undefined + job.imagePull = undefined + job.modelDownload = undefined + this.serviceProbeUrls.delete(serviceId) await this.db.updateServiceJob(job) // Live Docker handles for the newly-created container/network, tracked so the diff --git a/src/components/c2d/modelDownload.ts b/src/components/c2d/modelDownload.ts new file mode 100644 index 000000000..6dd78a47d --- /dev/null +++ b/src/components/c2d/modelDownload.ts @@ -0,0 +1,252 @@ +import type Dockerode from 'dockerode' +import tarStream from 'tar-stream' +import type { Readable } from 'stream' +import type { ServiceModelDownload } from '../../@types/C2D/ServiceOnDemand.js' +import { CORE_LOGGER } from '../../utils/logging/common.js' + +/** + * How much of its model a service has downloaded, read from the container's Hugging Face cache. + * + * The weights are fetched by `huggingface_hub` INSIDE the container, after it reports Running and + * before the engine can serve anything — the single longest wait a consumer sees, and one nothing + * reports today. The engine exposes no HTTP endpoint while it happens (there is no server yet), so + * the only structured source is the cache the downloader writes into: + * + * /models----/blobs/ a finished file + * /blobs/.incomplete one being written, size == bytes received + * + * llama.cpp downloads through the same cache but names its partial files `.downloadInProgress`. + * Both are counted; only the complete/in-flight split depends on telling them apart. + * + * The `.incomplete` mechanism is what makes HF downloads resumable, so it is a far steadier contract + * than parsing progress bars out of the container log. Verified against a live download: the byte + * counts track the transfer exactly, and several `.incomplete` files coexist because the hub fetches + * files in parallel — so only the AGGREGATE is meaningful, never a per-file percentage. + * + * The cache cannot supply the denominator: `snapshots//*` are symlinks (tar reports size 0 for + * those) pointing at blobs that do not exist yet, and they appear one by one as files resolve. The + * total therefore comes from the Hub's own metadata — see fetchModelTotalBytes. + */ + +// Suffixes a partially-downloaded file carries while it is being written. `huggingface_hub` (vLLM) +// uses `.incomplete`; llama.cpp's own `-hf` downloader uses `.downloadInProgress`. Both verified +// against live downloads. +const IN_FLIGHT_SUFFIXES = ['.incomplete', '.downloadInProgress'] + +// Read at most this many tar entries. A model repo has tens of files; a bound stops a pathological +// cache (or a wrong path pointing at something huge) from walking forever. +const MAX_CACHE_ENTRIES = 5000 + +/** + * Sums the cache's byte counts by streaming the directory out of the container and reading ONLY the + * tar HEADERS (name, size, type) — the payload is discarded as it arrives, so nothing large is + * transferred and no shell, volume, or storage-driver assumption is involved. Returns null when the + * cache does not exist yet (the normal state for the first seconds) or cannot be read. + */ +export async function readModelDownloadBytes( + container: Dockerode.Container, + cachePath: string +): Promise<{ downloadedBytes: number; files: number; inFlight: number } | null> { + let archive: Readable + try { + archive = (await container.getArchive({ path: cachePath })) as Readable + } catch (error: any) { + // 404 until the engine creates the cache — not a failure, just nothing to report yet. + if (error?.statusCode !== 404) { + CORE_LOGGER.debug(`[model-download] archive ${cachePath} failed: ${error?.message}`) + } + return null + } + + return await new Promise((resolve) => { + const extract = tarStream.extract() + let downloadedBytes = 0 + let files = 0 + let inFlight = 0 + let entries = 0 + let settled = false + + const finish = ( + result: { downloadedBytes: number; files: number; inFlight: number } | null + ) => { + if (settled) return + settled = true + resolve(result) + } + + extract.on('entry', (header, stream, next) => { + entries++ + // Only regular files carry bytes. A symlink (every snapshots/ entry) reports size 0 and its + // target may not exist yet, so counting those would double-count or contribute nothing. + if (header.type === 'file' && header.name.includes('/blobs/')) { + downloadedBytes += header.size ?? 0 + if (IN_FLIGHT_SUFFIXES.some((suffix) => header.name.endsWith(suffix))) { + inFlight++ + } else { + files++ + } + } + stream.on('end', next) + stream.resume() // discard the payload; only the header matters + if (entries > MAX_CACHE_ENTRIES) { + extract.destroy() + finish({ downloadedBytes, files, inFlight }) + } + }) + extract.on('finish', () => finish({ downloadedBytes, files, inFlight })) + extract.on('error', (error: any) => { + CORE_LOGGER.debug(`[model-download] tar read failed: ${error?.message}`) + finish(null) + }) + archive.on('error', () => finish(null)) + archive.pipe(extract) + }) +} + +// Bytes on the wire per parameter, by the dtype key the Hub reports. Anything unrecognized counts +// as 2 — the overwhelmingly common half-precision case, and a wrong guess here only skews a bar. +const BYTES_PER_PARAM: Record = { + F64: 8, + I64: 8, + F32: 4, + I32: 4, + U32: 4, + BF16: 2, + F16: 2, + I16: 2, + U16: 2, + F8_E4M3: 1, + F8_E5M2: 1, + I8: 1, + U8: 1, + BOOL: 1, + I4: 0.5, + U4: 0.5 +} + +const HF_MODEL_API = 'https://huggingface.co/api/models' +const HF_TIMEOUT_MS = 8000 +// One lookup per model for the life of the process: the answer cannot change for a given repo, and +// this is read on the metrics cadence for every starting service. +const totalBytesCache = new Map() + +/** + * The size of the weights an engine will download for a model, from the Hub's safetensors index. + * + * Deliberately NOT the repo's `usedStorage`: that counts every artifact in the repo, including the + * .bin duplicates and GGUF quantizations of the same weights that nothing downloads — for a small + * model it reads roughly triple what is actually fetched, which would park a progress bar at a third + * of the truth for the entire wait. + * + * Returns null for a repo with no safetensors index (GGUF-only repos, gated repos, anything the Hub + * has not indexed) and whenever the Hub is unreachable. The caller then reports bytes downloaded + * with no percentage, rather than a ratio against a made-up denominator. + */ +export async function fetchModelTotalBytes( + modelId: string, + quant?: string +): Promise { + const key = quant ? `${modelId}:${quant}` : modelId + if (totalBytesCache.has(key)) return totalBytesCache.get(key) ?? null + + let total: number | null = null + try { + if (quant) { + const ggufTotal = await fetchGgufFileBytes(modelId, quant) + totalBytesCache.set(key, ggufTotal) + return ggufTotal + } + const path = modelId + .split('/') + .map((segment) => encodeURIComponent(segment)) + .join('/') + const response = await fetch(`${HF_MODEL_API}/${path}?expand[]=safetensors`, { + signal: AbortSignal.timeout(HF_TIMEOUT_MS) + }) + if (response.ok) { + const body: any = await response.json() + const parameters = body?.safetensors?.parameters + if (parameters && typeof parameters === 'object') { + const bytes = Object.entries(parameters).reduce( + (sum, [dtype, count]) => + sum + (Number(count) || 0) * (BYTES_PER_PARAM[dtype] ?? 2), + 0 + ) + total = bytes > 0 ? Math.round(bytes) : null + } else if (Number(body?.safetensors?.total) > 0) { + // Only a parameter count, no dtype breakdown: assume half precision, the default these + // repos are served in. + total = Math.round(Number(body.safetensors.total) * 2) + } + } + } catch (error: any) { + CORE_LOGGER.debug( + `[model-download] hub lookup for ${modelId} failed: ${error?.message}` + ) + } + totalBytesCache.set(key, total) + return total +} + +/** + * The size of ONE quantization file in a GGUF repo, for an engine that names it (llama.cpp's + * `-hf :`). + * + * A GGUF repo publishes no safetensors index, so the parameter-count route returns nothing for it — + * but the Hub does list every file with its size, and the engine downloads exactly one of them. + * Matching on the quant tag gives the exact denominator instead of the whole repo's contents, which + * for a repo carrying a dozen quantizations would be an order of magnitude out. + */ +async function fetchGgufFileBytes( + modelId: string, + quant: string +): Promise { + try { + const path = modelId + .split('/') + .map((segment) => encodeURIComponent(segment)) + .join('/') + const response = await fetch(`${HF_MODEL_API}/${path}?blobs=true`, { + signal: AbortSignal.timeout(HF_TIMEOUT_MS) + }) + if (!response.ok) return null + const body: any = await response.json() + const siblings: any[] = Array.isArray(body?.siblings) ? body.siblings : [] + const wanted = quant.toLowerCase() + const match = siblings.find((file) => { + const name = String(file?.rfilename ?? '').toLowerCase() + return name.endsWith('.gguf') && name.includes(wanted) + }) + const size = Number(match?.size ?? match?.lfs?.size) + return Number.isFinite(size) && size > 0 ? size : null + } catch (error: any) { + CORE_LOGGER.debug( + `[model-download] gguf lookup for ${modelId}:${quant} failed: ${error?.message}` + ) + return null + } +} + +/** Builds the record persisted on the job, capping the ratio (see the note on repo variants). */ +export function buildModelDownload( + downloaded: { downloadedBytes: number; files: number; inFlight: number }, + totalBytes: number | null, + modelId: string | null +): ServiceModelDownload { + // The Hub reports the REPO's safetensors size while the engine downloads only what it needs — + // identical in the common case, but a repo carrying several variants over-counts. Cap rather than + // report >100%, and treat "reached the total" as complete. + const percent = + totalBytes && totalBytes > 0 + ? Math.min(100, Math.round((downloaded.downloadedBytes / totalBytes) * 100)) + : undefined + return { + ...(modelId ? { modelId } : {}), + downloadedBytes: downloaded.downloadedBytes, + ...(totalBytes ? { totalBytes } : {}), + ...(percent !== undefined ? { percent } : {}), + filesComplete: downloaded.files, + filesInFlight: downloaded.inFlight, + updatedAt: Date.now() + } +} diff --git a/src/components/c2d/serviceEngines.ts b/src/components/c2d/serviceEngines.ts new file mode 100644 index 000000000..7234f26bf --- /dev/null +++ b/src/components/c2d/serviceEngines.ts @@ -0,0 +1,159 @@ +import type { ServiceJob } from '../../@types/C2D/ServiceOnDemand.js' + +/** + * What the node knows about the workloads it can report readiness and model-download progress for. + * + * Deliberately a closed table rather than something the client declares. A service's container + * reports Running the moment its process starts, but an inference engine then spends minutes + * downloading weights and warming up — during which its port either refuses connections or answers + * 503. Only the engine knows when that ends, and only the node can ask it (the endpoint is plain + * http, so a browser on an https page cannot). Knowing WHICH question to ask is engine-specific, so + * it lives here: one row per engine the node understands. + * + * An unrecognized image matches nothing, and the node then reports no readiness at all — which every + * client reads as "this node cannot tell me", falling back to the pre-existing behaviour where + * Running means the endpoint is handed over. New engines are added by adding a row. + */ +export interface ServiceEngineProfile { + /** Identifier carried on the job's readiness record, so a client can tell what was probed. */ + id: string + /** Matches the image reference (without tag/digest) this engine ships as. */ + matchesImage: (image: string) => boolean + /** The request that proves the engine can serve traffic. */ + probe: { + path: string + /** Container port to probe; defaults to the service's first exposed port when absent. */ + port?: number + expectStatus: number[] + } + /** + * Where this engine caches the model it downloads at startup, inside the container. Absent for an + * engine whose download the node cannot observe — readiness still works, progress simply isn't + * reported. + */ + modelCachePath?: string + /** + * Recovers the Hugging Face repo id from the container command, for the size lookup. Returns null + * when the engine was pointed at anything else (a local path, an object-store URI, another hub) — + * progress is then reported as bytes with no total. + */ + modelIdFromCommand?: (cmd: string[] | undefined) => string | null + /** + * The specific file within the repo this engine downloads, when it names one (llama.cpp's + * `-hf :`). Lets the size lookup ask for that file rather than the whole repo, which + * for a GGUF repo carrying a dozen quantizations differs by an order of magnitude. + */ + modelQuantFromCommand?: (cmd: string[] | undefined) => string | null +} + +/** + * The Hugging Face repo id llama.cpp is serving, or null. + * + * `-hf /[:]` fetches a GGUF from the Hub; the `:quant` suffix selects a file + * within the repo and is not part of the repo id. `-m ` (a local file) names no repo at all. + */ +function hfRepoIdFromLlamaCppCommand(cmd: string[] | undefined): string | null { + if (!cmd) return null + const index = cmd.indexOf('-hf') + if (index === -1 || index === cmd.length - 1) return null + const value = cmd[index + 1] + if (!value || value.startsWith('-')) return null + const repo = value.split(':')[0] + return isHubRepoId(repo, { requireOrg: true }) ? repo : null +} + +/** + * Whether a string is shaped like a Hugging Face repo id: `org/name`, or bare `name` when an org is + * not required. Checked segment by segment rather than with one pattern — a regex combining an + * optional group with repeated character classes backtracks ambiguously on caller-supplied input. + */ +function isHubRepoId(value: string, opts: { requireOrg?: boolean } = {}): boolean { + const segments = value.split('/') + if (segments.length > 2) return false + if (opts.requireOrg && segments.length !== 2) return false + return segments.every((segment) => segment.length > 0 && /^[\w.-]+$/.test(segment)) +} + +/** The quant tag in `-hf :`, e.g. `Q4_K_M`. Null when the command names no quant. */ +function quantFromLlamaCppCommand(cmd: string[] | undefined): string | null { + if (!cmd) return null + const index = cmd.indexOf('-hf') + if (index === -1 || index === cmd.length - 1) return null + const parts = String(cmd[index + 1] ?? '').split(':') + return parts.length > 1 && parts[1] ? parts[1] : null +} + +/** + * The Hugging Face repo id vLLM is serving, or null when it is not serving one. + * + * `--model` accepts far more than a Hub repo: an absolute path to weights already on disk, an + * object-store URI (`s3://`, `gs://`), or a ModelScope id when VLLM_USE_MODELSCOPE is set. Only a + * plain `org/name` is a Hub repo whose size the Hub can be asked for — everything else returns null, + * and the caller then reports bytes downloaded with no total (an indeterminate bar) rather than + * measuring against a repo that has nothing to do with what is being fetched. + */ +function hfRepoIdFromVllmCommand(cmd: string[] | undefined): string | null { + if (!cmd) return null + const index = cmd.indexOf('--model') + if (index === -1 || index === cmd.length - 1) return null + const value = cmd[index + 1] + if (!value || value.startsWith('-')) return null + // A local path or a remote URI is not a Hub repo. + if (value.startsWith('/') || value.startsWith('.') || value.includes('://')) return null + // A Hub repo is exactly `org/name` (no further slashes, no whitespace). `name` alone is a + // canonical-model shorthand the Hub also serves, so it is accepted too. + if (!isHubRepoId(value)) return null + // NOTE: a ModelScope id (VLLM_USE_MODELSCOPE=1) is shaped exactly like a Hub one and cannot be + // told apart here, since the env that selects it rides in encrypted userData. The Hub lookup then + // simply 404s and the caller falls back to bytes-with-no-total, which is the right outcome anyway. + return value +} + +export const SERVICE_ENGINE_PROFILES: ServiceEngineProfile[] = [ + { + id: 'vllm', + // The upstream image, and the common convention of mirroring it under another registry with the + // same repository path. A privately renamed build matches nothing, which is the documented + // "no readiness reported" case rather than a wrong answer. + matchesImage: (image) => + /(^|\/)vllm\/vllm-openai$/.test(image) || image === 'vllm-openai', + /** + * `/v1/models`, NOT `/health`: across vLLM versions `/health` has answered 200 from the moment + * the HTTP server binds, which on some builds is before the weights are loaded — it would + * declare readiness at exactly the wrong moment. `/v1/models` is the model-aware one. vLLM also + * often binds the port only after loading, so the probe simply gets a connection refusal until + * then; both shapes mean "not ready". + */ + probe: { path: '/v1/models', port: 8000, expectStatus: [200] }, + // huggingface_hub's cache, which vLLM inherits. HF_HOME defaults to ~/.cache/huggingface and the + // official image runs as root. + modelCachePath: '/root/.cache/huggingface/hub', + modelIdFromCommand: hfRepoIdFromVllmCommand + }, + { + id: 'llamacpp', + // Upstream publishes CPU (`:server`) and CUDA (`:server-cuda`) under the same repository, and + // NEXT_PUBLIC_LLAMACPP_IMAGE lets an operator point at their own build of it — matched on the + // repository path, whatever registry it is mirrored under. + matchesImage: (image) => /(^|\/)ggml-org\/llama\.cpp$/.test(image), + /** + * llama.cpp's server binds its port only once the model is in memory, and its `/health` is + * documented to answer 503 `{"status":"loading model"}` while loading and 200 `{"status":"ok"}` + * after — verified against the arm64 image: connection refused, then 200, with the port opening + * at "model loaded". Either shape reads as not-ready, so the check holds for both. + */ + probe: { path: '/health', port: 8080, expectStatus: [200] }, + // `-hf` downloads through the Hugging Face cache, exactly like vLLM's — NOT the + // `/root/.cache/llama.cpp` that older guides mention (confirmed absent on a live download). + modelCachePath: '/root/.cache/huggingface/hub', + modelIdFromCommand: hfRepoIdFromLlamaCppCommand, + modelQuantFromCommand: quantFromLlamaCppCommand + } +] + +/** The engine profile for a service, or null when the node does not recognize the image. */ +export function resolveServiceEngine(job: ServiceJob): ServiceEngineProfile | null { + const image = (job.image ?? '').trim() + if (!image) return null + return SERVICE_ENGINE_PROFILES.find((profile) => profile.matchesImage(image)) ?? null +} diff --git a/src/components/c2d/serviceReadiness.ts b/src/components/c2d/serviceReadiness.ts new file mode 100644 index 000000000..beae9b452 --- /dev/null +++ b/src/components/c2d/serviceReadiness.ts @@ -0,0 +1,194 @@ +import type { + ServiceImagePullProgress, + ServiceJob +} from '../../@types/C2D/ServiceOnDemand.js' + +// ── Image pull progress ─────────────────────────────────────────────── + +// How often the aggregated byte counts are pushed to the caller (which persists them). The daemon +// emits progress events per layer per ~100ms, which is far more often than a status poll reads. +const PULL_PROGRESS_EMIT_INTERVAL_MS = 1000 + +/** + * Aggregates the Docker daemon's per-layer pull events into one byte/percentage view. + * + * Docker reports each layer separately, and only announces a layer's `total` when that layer + * STARTS downloading — so the denominator grows during the pull and a naively recomputed + * percentage walks backwards. The emitted `percent` is therefore clamped monotonic: it is the + * honest lower bound at every moment, which is what a progress bar needs. + * + * Layers that are already on the host arrive as "Already exists" and contribute no bytes, so an + * image that is fully cached completes at 0 bytes — the caller reports that as "complete" rather + * than as a stalled 0%. + */ +export class ImagePullTracker { + private readonly layers = new Map< + string, + { current: number; total: number; done: boolean } + >() + + private lastEmitAt = 0 + private maxPercent = 0 + private extracting = false + + constructor(private readonly emit: (progress: ServiceImagePullProgress) => void) {} + + onEvent(event: any): void { + const id: string | undefined = typeof event?.id === 'string' ? event.id : undefined + const status: string = typeof event?.status === 'string' ? event.status : '' + // Not a layer: the pull's own narration. Most of it carries no id ("Digest: sha256:…"), but + // the opening "Pulling from library/python" line carries the TAG as its id — counted as a + // layer it inflates layersTotal by one, so the pull ends reading 4/5 instead of 4/4. + if (!id || status.startsWith('Pulling from') || status.startsWith('Digest:')) return + + const layer = this.layers.get(id) ?? { current: 0, total: 0, done: false } + const current = Number(event?.progressDetail?.current) + const total = Number(event?.progressDetail?.total) + + if (status.startsWith('Downloading')) { + if (Number.isFinite(current)) layer.current = current + if (Number.isFinite(total) && total > 0) layer.total = total + } else if ( + status.startsWith('Verifying Checksum') || + status === 'Download complete' + ) { + // The byte stream for this layer is over; the daemon stops reporting `current`. + if (layer.total > 0) layer.current = layer.total + } else if (status.startsWith('Extracting')) { + this.extracting = true + if (layer.total > 0) layer.current = layer.total + } else if (status === 'Pull complete' || status === 'Already exists') { + if (layer.total > 0) layer.current = layer.total + layer.done = true + } + + this.layers.set(id, layer) + this.maybeEmit() + } + + /** Pull finished — report 100% once, so a listener never sits on the last partial sample. */ + finish(): void { + const snapshot = this.snapshot() + this.emit({ ...snapshot, phase: 'complete', percent: 100, updatedAt: Date.now() }) + } + + private maybeEmit(): void { + const now = Date.now() + if (now - this.lastEmitAt < PULL_PROGRESS_EMIT_INTERVAL_MS) return + this.lastEmitAt = now + this.emit({ ...this.snapshot(), updatedAt: now }) + } + + private snapshot(): ServiceImagePullProgress { + let downloadedBytes = 0 + let totalBytes = 0 + let layersDone = 0 + for (const layer of this.layers.values()) { + downloadedBytes += layer.current + totalBytes += layer.total + if (layer.done) layersDone++ + } + const raw = totalBytes > 0 ? (downloadedBytes / totalBytes) * 100 : 0 + this.maxPercent = Math.min(100, Math.max(this.maxPercent, raw)) + return { + phase: this.extracting ? 'extracting' : 'downloading', + downloadedBytes, + totalBytes, + percent: Math.round(this.maxPercent), + layersTotal: this.layers.size, + layersDone, + updatedAt: Date.now() + } + } +} + +// ── Readiness probe ─────────────────────────────────────────────────── + +// Before the first check, giving the container a moment to bind its port. +export const PROBE_INITIAL_DELAY_SECONDS = 5 +// While warming up. The InternalLoop ticks faster than this, so the probe throttles itself. +export const PROBE_PERIOD_SECONDS = 5 +// Once ready the check keeps running — to catch an engine that dies without its container exiting — +// but far more slowly: it is a liveness check at that point, not a wait. +export const READY_PROBE_PERIOD_SECONDS = 30 +const PROBE_TIMEOUT_MS = 2000 + +export interface ReadinessProbeResult { + ok: boolean + httpStatus?: number + error?: string + url: string +} + +/** + * The addresses worth trying to reach a service container on, best first. + * + * Which one works depends entirely on how the node itself is deployed, and the node cannot know + * that up front: a node running on the host reaches the container's own bridge IP directly, a node + * that publishes ports reaches them on loopback, and a node running inside Docker reaches neither + * and has to go out through the host gateway. So we try them in order once and remember the winner. + */ +export function probeCandidates( + job: ServiceJob, + containerIps: string[], + containerPort: number, + path: string +): string[] { + const endpoint = + job.endpoints.find((ep) => ep.containerPort === containerPort) ?? job.endpoints[0] + const hostPort = endpoint?.hostPort + const urls: string[] = [] + for (const ip of containerIps) { + if (ip) urls.push(`http://${ip}:${containerPort}${path}`) + } + if (hostPort) { + urls.push(`http://127.0.0.1:${hostPort}${path}`) + // Docker Desktop always, and Linux compose when the node declares `host-gateway`. + urls.push(`http://host.docker.internal:${hostPort}${path}`) + // The public URL handed to consumers — last, because it depends on NAT hairpinning. + if (endpoint?.url) urls.push(`${endpoint.url}${path}`) + } + // De-duplicate while keeping order (nodeHost is often 'localhost', i.e. candidate 2 again). + return [...new Set(urls)] +} + +/** + * Performs ONE readiness request. Never throws — a failure IS the result. + * + * A refused connection is reported without an httpStatus, which the caller treats as "not ready" + * exactly like an unexpected status: an engine that binds its port only after loading its weights + * (vLLM, commonly) is unreachable rather than unhealthy for that whole window. + */ +export async function runReadinessProbe( + url: string, + expectStatus: number[] +): Promise { + const controller = new AbortController() + const timer = setTimeout(() => controller.abort(), PROBE_TIMEOUT_MS) + try { + const response = await fetch(url, { + method: 'GET', + signal: controller.signal, + headers: { accept: '*/*' } + }) + // Drain, so the socket is not left hanging on either path. + await response.body?.cancel().catch(() => {}) + if (!expectStatus.includes(response.status)) { + return { + ok: false, + httpStatus: response.status, + url, + error: `unexpected status ${response.status}` + } + } + return { ok: true, httpStatus: response.status, url } + } catch (e: any) { + return { + ok: false, + url, + error: e?.name === 'AbortError' ? 'timeout' : String(e?.message) + } + } finally { + clearTimeout(timer) + } +} diff --git a/src/components/core/service/utils.ts b/src/components/core/service/utils.ts index 77de3d63c..73463118c 100644 --- a/src/components/core/service/utils.ts +++ b/src/components/core/service/utils.ts @@ -95,9 +95,24 @@ export function toListedServiceJob( dockerEntrypoint, dockerfile, additionalDockerFiles, + readiness, ...pub } = job - return pub + // "Is this service actually serving?" is the listing's whole point for a node operator, so the + // readiness RESULT is kept — but reduced to what answers that, without the diagnostics (probed + // address, error text) meant for the owner. + return { + ...pub, + ...(readiness + ? { + readiness: { + state: readiness.state, + engine: readiness.engine, + readySince: readiness.readySince + } + } + : {}) + } } const SINCE_DURATION_RE = /^(\d+)(s|m|h|d)$/ diff --git a/src/components/database/C2DDatabase.ts b/src/components/database/C2DDatabase.ts index de6a658bd..fe83b3921 100755 --- a/src/components/database/C2DDatabase.ts +++ b/src/components/database/C2DDatabase.ts @@ -6,7 +6,11 @@ import { C2DStatusNumber, ContainerMetricsSnapshot } from '../../@types/C2D/C2D.js' -import { ServiceJob } from '../../@types/C2D/ServiceOnDemand.js' +import { + ServiceJob, + ServiceModelDownload, + ServiceReadiness +} from '../../@types/C2D/ServiceOnDemand.js' import { SQLiteCompute } from './sqliteCompute.js' import { DATABASE_LOGGER } from '../../utils/logging/common.js' import { OceanNodeDBConfig } from '../../@types/OceanNode.js' @@ -108,6 +112,25 @@ export class C2DDatabase extends AbstractDatabase { ) } + async updateServiceJobReadiness( + serviceId: string, + expected: { + owner: string + clusterHash: string + status: number + containerId: string + }, + readiness: ServiceReadiness, + modelDownload?: ServiceModelDownload + ): Promise { + return await this.provider.updateServiceJobReadiness( + serviceId, + expected, + readiness, + modelDownload + ) + } + async getRunningServiceJobs(clusterHash?: string): Promise { return await this.provider.getRunningServiceJobs(clusterHash) } diff --git a/src/components/database/sqliteCompute.ts b/src/components/database/sqliteCompute.ts index 00bff23be..e9e9affc5 100644 --- a/src/components/database/sqliteCompute.ts +++ b/src/components/database/sqliteCompute.ts @@ -8,7 +8,9 @@ import { import { ServiceStatusNumber, SERVICE_START_PENDING_STATUSES, - type ServiceJob + type ServiceJob, + type ServiceModelDownload, + type ServiceReadiness } from '../../@types/C2D/ServiceOnDemand.js' import { SqliteClient } from './sqliteClient.js' import { DATABASE_LOGGER } from '../../utils/logging/common.js' @@ -321,18 +323,17 @@ export class SQLiteCompute implements ComputeDatabaseProvider { } } - // Persists ONLY runtimeMetrics onto a service job — best-effort telemetry that must NOT clobber - // lifecycle fields another process may have changed. The read + validate + merge + write run - // inside a single `BEGIN IMMEDIATE` transaction so the sequence is atomic even across processes - // sharing the SQLite file: IMMEDIATE takes the write lock up front (waiting up to busy_timeout), - // so no other process can commit a lifecycle change (status, expiry, container) between our read - // and our write. Inside the transaction it re-reads the CURRENT row, re-validates - // owner/clusterHash/status/containerId, merges ONLY runtimeMetrics into that current body, and - // writes back ONLY the `body` column guarded on the unchanged status — leaving every lifecycle - // field (incl. expiresAt, in both column and body) exactly as last committed. Returns true when a - // row was written; any mismatch/error rolls back and returns false. - // eslint-disable-next-line require-await - async updateServiceJobMetrics( + // Persists ONE best-effort, non-lifecycle field onto a service job (runtime metrics, readiness) + // without clobbering lifecycle fields another process may have changed. The read + validate + + // merge + write run inside a single `BEGIN IMMEDIATE` transaction so the sequence is atomic even + // across processes sharing the SQLite file: IMMEDIATE takes the write lock up front (waiting up + // to busy_timeout), so no other process can commit a lifecycle change (status, expiry, container) + // between our read and our write. Inside the transaction it re-reads the CURRENT row, re-validates + // owner/clusterHash/status/containerId, applies `patch` to that current body, and writes back ONLY + // the `body` column guarded on the unchanged status — leaving every lifecycle field (incl. + // expiresAt, in both column and body) exactly as last committed. Returns true when a row was + // written; any mismatch/error rolls back and returns false. + private patchServiceJobBody( serviceId: string, expected: { owner: string @@ -340,12 +341,13 @@ export class SQLiteCompute implements ComputeDatabaseProvider { status: number containerId: string }, - runtimeMetrics: ContainerMetricsSnapshot - ): Promise { + what: string, + patch: (body: ServiceJob) => void + ): boolean { try { this.db.exec('BEGIN IMMEDIATE;') } catch (err) { - DATABASE_LOGGER.error(`metrics update: could not begin transaction: ${err.message}`) + DATABASE_LOGGER.error(`${what} update: could not begin transaction: ${err.message}`) return false } try { @@ -367,7 +369,7 @@ export class SQLiteCompute implements ComputeDatabaseProvider { this.db.exec('ROLLBACK;') return false } - body.runtimeMetrics = runtimeMetrics + patch(body) const { changes } = this.db.run( `UPDATE service_jobs SET body = ? WHERE serviceId = ? AND status = ?;`, [ @@ -379,7 +381,7 @@ export class SQLiteCompute implements ComputeDatabaseProvider { this.db.exec('COMMIT;') return changes > 0 } catch (err) { - DATABASE_LOGGER.error(`Error while updating service job metrics: ${err.message}`) + DATABASE_LOGGER.error(`Error while updating service job ${what}: ${err.message}`) try { this.db.exec('ROLLBACK;') } catch { @@ -389,6 +391,44 @@ export class SQLiteCompute implements ComputeDatabaseProvider { } } + // eslint-disable-next-line require-await + async updateServiceJobMetrics( + serviceId: string, + expected: { + owner: string + clusterHash: string + status: number + containerId: string + }, + runtimeMetrics: ContainerMetricsSnapshot + ): Promise { + return this.patchServiceJobBody(serviceId, expected, 'metrics', (body) => { + body.runtimeMetrics = runtimeMetrics + }) + } + + // Same guarantees as updateServiceJobMetrics, for the readiness probe result: it is sampled from + // the same lease-free background loop, so it must never overwrite a lifecycle transition either. + // eslint-disable-next-line require-await + async updateServiceJobReadiness( + serviceId: string, + expected: { + owner: string + clusterHash: string + status: number + containerId: string + }, + readiness: ServiceReadiness, + modelDownload?: ServiceModelDownload + ): Promise { + return this.patchServiceJobBody(serviceId, expected, 'readiness', (body) => { + body.readiness = readiness + // Only overwritten when a fresh sample was taken: once the engine is ready the walk stops, + // and the last figures stay as the record of what was downloaded. + if (modelDownload) body.modelDownload = modelDownload + }) + } + private mapServiceRows(rows: any[] | undefined): ServiceJob[] { if (!rows || rows.length === 0) return [] // BLOB comes back as Uint8Array from node:sqlite; decode through Buffer before parsing. diff --git a/src/test/integration/services.test.ts b/src/test/integration/services.test.ts index e04eec5a7..0f2228026 100644 --- a/src/test/integration/services.test.ts +++ b/src/test/integration/services.test.ts @@ -472,6 +472,17 @@ describe('********** Service on Demand', () => { assert(res.body.toLowerCase().includes('nginx'), 'body should be the nginx page') }) + it('(d2) reports no readiness for an image the node does not recognize', async function () { + this.timeout(DEFAULT_TEST_TIMEOUT * 4) + // nginx is not an engine the node knows, so it must be left exactly as it was before this + // feature existed: Running, and no readiness field for a client to gate on. + const job = await getServiceJob(serviceId) + assert(job, 'job not found') + expect(job.status).to.equal(ServiceStatusNumber.Running) + expect(job.readiness).to.equal(undefined) + expect(job.modelDownload).to.equal(undefined) + }) + it('(e) SERVICE_GET_STATUS returns the job with userData stripped', async () => { const job = await getServiceJob(serviceId) assert(job, 'job not found') diff --git a/src/test/unit/service/serviceReadiness.test.ts b/src/test/unit/service/serviceReadiness.test.ts new file mode 100644 index 000000000..d853f937e --- /dev/null +++ b/src/test/unit/service/serviceReadiness.test.ts @@ -0,0 +1,286 @@ +import { expect } from 'chai' +import { createServer, Server } from 'http' +import type { AddressInfo } from 'net' +import type { + ServiceImagePullProgress, + ServiceJob +} from '../../../@types/C2D/ServiceOnDemand.js' +import { + ImagePullTracker, + probeCandidates, + runReadinessProbe +} from '../../../components/c2d/serviceReadiness.js' +import { resolveServiceEngine } from '../../../components/c2d/serviceEngines.js' +import { buildModelDownload } from '../../../components/c2d/modelDownload.js' + +describe('ImagePullTracker', () => { + // The daemon's real event sequence for one layer, minus the narration lines. + function pull(tracker: ImagePullTracker, id: string, bytes: number) { + tracker.onEvent({ id, status: 'Pulling fs layer', progressDetail: {} }) + tracker.onEvent({ + id, + status: 'Downloading', + progressDetail: { current: bytes / 2, total: bytes } + }) + tracker.onEvent({ + id, + status: 'Downloading', + progressDetail: { current: bytes, total: bytes } + }) + tracker.onEvent({ id, status: 'Download complete', progressDetail: {} }) + tracker.onEvent({ id, status: 'Pull complete', progressDetail: {} }) + } + + it('aggregates layer bytes and reports 100% on finish', () => { + const emitted: ServiceImagePullProgress[] = [] + const tracker = new ImagePullTracker((p) => emitted.push(p)) + pull(tracker, 'layer-a', 1000) + pull(tracker, 'layer-b', 3000) + tracker.finish() + + const last = emitted[emitted.length - 1] + expect(last.phase).to.equal('complete') + expect(last.percent).to.equal(100) + expect(last.downloadedBytes).to.equal(4000) + expect(last.totalBytes).to.equal(4000) + expect(last.layersTotal).to.equal(2) + expect(last.layersDone).to.equal(2) + }) + + it('never walks the percentage backwards when a later layer announces its size', () => { + const emitted: ServiceImagePullProgress[] = [] + const tracker = new ImagePullTracker((p) => emitted.push(p)) + // First layer completes (100% of everything known so far)... + tracker.onEvent({ + id: 'a', + status: 'Downloading', + progressDetail: { current: 100, total: 100 } + }) + tracker.onEvent({ id: 'a', status: 'Pull complete', progressDetail: {} }) + // ...then a much bigger second layer appears, which would drop a naive ratio to 10%. + tracker.onEvent({ + id: 'b', + status: 'Downloading', + progressDetail: { current: 0, total: 900 } + }) + tracker.finish() + + const percents = emitted.map((e) => e.percent) + for (let i = 1; i < percents.length; i++) { + expect(percents[i]).to.be.at.least(percents[i - 1]) + } + }) + + it('reports a fully cached image as complete rather than stuck at 0%', () => { + const emitted: ServiceImagePullProgress[] = [] + const tracker = new ImagePullTracker((p) => emitted.push(p)) + tracker.onEvent({ id: 'a', status: 'Already exists', progressDetail: {} }) + tracker.onEvent({ id: 'b', status: 'Already exists', progressDetail: {} }) + tracker.finish() + + const last = emitted[emitted.length - 1] + expect(last.phase).to.equal('complete') + expect(last.percent).to.equal(100) + expect(last.downloadedBytes).to.equal(0) + expect(last.layersDone).to.equal(2) + }) + + it('does not count the "Pulling from " line as a layer', () => { + const emitted: ServiceImagePullProgress[] = [] + const tracker = new ImagePullTracker((p) => emitted.push(p)) + // The daemon sends this first, with the TAG as its id. + tracker.onEvent({ id: '3.12-slim', status: 'Pulling from library/python' }) + pull(tracker, 'layer-a', 1000) + tracker.finish() + + const last = emitted[emitted.length - 1] + expect(last.layersTotal).to.equal(1) + expect(last.layersDone).to.equal(1) + }) + + it("ignores events with no layer id (the pull's own narration)", () => { + const emitted: ServiceImagePullProgress[] = [] + const tracker = new ImagePullTracker((p) => emitted.push(p)) + tracker.onEvent({ status: 'Digest: sha256:abc' }) + tracker.onEvent({ status: 'Status: Downloaded newer image for x:latest' }) + tracker.finish() + expect(emitted[emitted.length - 1].layersTotal).to.equal(0) + }) +}) + +describe('resolveServiceEngine', () => { + const job = (image: string) => ({ image }) as ServiceJob + + it('recognizes the vLLM image, including a registry mirror of it', () => { + expect(resolveServiceEngine(job('vllm/vllm-openai'))?.id).to.equal('vllm') + expect(resolveServiceEngine(job('ghcr.io/vllm/vllm-openai'))?.id).to.equal('vllm') + }) + + it('returns null for an image it does not know, so nothing is gated', () => { + expect(resolveServiceEngine(job('nginxinc/nginx-unprivileged'))).to.equal(null) + expect(resolveServiceEngine(job('someone/vllm-openai-fork'))).to.equal(null) + expect(resolveServiceEngine(job(''))).to.equal(null) + }) + + it('reads the Hugging Face repo id out of the vLLM command', () => { + const vllm = resolveServiceEngine(job('vllm/vllm-openai'))! + expect( + vllm.modelIdFromCommand!(['--model', 'Qwen/Qwen2.5-7B-Instruct', '--port', '8000']) + ).to.equal('Qwen/Qwen2.5-7B-Instruct') + }) + + it('refuses to call a non-Hub source a repo id (no size can be claimed for it)', () => { + const vllm = resolveServiceEngine(job('vllm/vllm-openai'))! + // vLLM also serves weights already on disk, or from object storage — asking the Hub about + // either would measure progress against a repo that has nothing to do with the download. + expect(vllm.modelIdFromCommand!(['--model', '/models/local-weights'])).to.equal(null) + expect(vllm.modelIdFromCommand!(['--model', 's3://bucket/model'])).to.equal(null) + expect(vllm.modelIdFromCommand!(['--model'])).to.equal(null) + expect(vllm.modelIdFromCommand!(undefined)).to.equal(null) + }) +}) + +describe('resolveServiceEngine — llama.cpp', () => { + const job = (image: string, cmd?: string[]) => ({ image, dockerCmd: cmd }) as ServiceJob + + it('recognizes the llama.cpp image and probes /health', () => { + const engine = resolveServiceEngine(job('ghcr.io/ggml-org/llama.cpp')) + expect(engine?.id).to.equal('llamacpp') + // Verified against the arm64 image: the port opens only at "model loaded", and /health then + // answers 200 {"status":"ok"} — 503 "loading model" on builds that bind earlier. + expect(engine?.probe.path).to.equal('/health') + expect(engine?.probe.port).to.equal(8080) + }) + + it('splits `-hf repo:quant` into the repo and the file it pulls', () => { + const engine = resolveServiceEngine(job('ghcr.io/ggml-org/llama.cpp'))! + const cmd = ['-hf', 'Qwen/Qwen2.5-0.5B-Instruct-GGUF:Q4_K_M', '--port', '8080'] + expect(engine.modelIdFromCommand!(cmd)).to.equal('Qwen/Qwen2.5-0.5B-Instruct-GGUF') + expect(engine.modelQuantFromCommand!(cmd)).to.equal('Q4_K_M') + }) + + it('handles a repo with no quant, and a local -m path', () => { + const engine = resolveServiceEngine(job('ghcr.io/ggml-org/llama.cpp'))! + expect(engine.modelIdFromCommand!(['-hf', 'org/repo'])).to.equal('org/repo') + expect(engine.modelQuantFromCommand!(['-hf', 'org/repo'])).to.equal(null) + // `-m` points at a file already on disk — no repo to size. + expect(engine.modelIdFromCommand!(['-m', '/models/model.gguf'])).to.equal(null) + }) + + it('uses the same Hugging Face cache as vLLM', () => { + // NOT /root/.cache/llama.cpp, which older guides name — confirmed absent on a live download. + const llamacpp = resolveServiceEngine(job('ghcr.io/ggml-org/llama.cpp')) + const vllm = resolveServiceEngine(job('vllm/vllm-openai')) + expect(llamacpp?.modelCachePath).to.equal(vllm?.modelCachePath) + }) +}) + +describe('probeCandidates', () => { + const job = { + endpoints: [ + { containerPort: 8000, hostPort: 31000, url: 'http://node.example:31000' }, + { containerPort: 9000, hostPort: 31001, url: 'http://node.example:31001' } + ] + } as ServiceJob + + it('tries the container IP first, then the published port, then the public URL', () => { + const urls = probeCandidates(job, ['172.18.0.2'], 8000, '/v1/models') + expect(urls[0]).to.equal('http://172.18.0.2:8000/v1/models') + expect(urls).to.include('http://127.0.0.1:31000/v1/models') + expect(urls[urls.length - 1]).to.equal('http://node.example:31000/v1/models') + }) + + it('maps the probe port to ITS endpoint, not the first one', () => { + const urls = probeCandidates(job, [], 9000, '/health') + expect(urls).to.include('http://127.0.0.1:31001/health') + expect(urls).to.not.include('http://127.0.0.1:31000/health') + }) +}) + +describe('runReadinessProbe', () => { + let server: Server + let base: string + // Mimics the two engines: 503 while "loading", then 200 with a model list. + let loaded = false + + before(async () => { + server = createServer((req, res) => { + if (req.url === '/hang') { + return // never answers — exercises the timeout + } + if (!loaded) { + res.writeHead(503, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ status: 'loading model' })) + return + } + res.writeHead(200, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ object: 'list', data: [{ id: 'my-org/my-model' }] })) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + base = `http://127.0.0.1:${(server.address() as AddressInfo).port}` + }) + + after(async () => { + await new Promise((resolve) => server.close(() => resolve())) + }) + + it('reports not-ready while the engine answers 503', async () => { + loaded = false + const result = await runReadinessProbe(`${base}/v1/models`, [200]) + expect(result.ok).to.equal(false) + expect(result.httpStatus).to.equal(503) + }) + + it('reports ready once the engine answers 200 with the expected model', async () => { + loaded = true + const result = await runReadinessProbe(`${base}/v1/models`, [200]) + expect(result.ok).to.equal(true) + expect(result.httpStatus).to.equal(200) + }) + + it('treats a refused connection as not-ready, with no http status', async () => { + // Port 1 is reserved and never listening. + const result = await runReadinessProbe('http://127.0.0.1:1/v1/models', [200]) + expect(result.ok).to.equal(false) + expect(result.httpStatus).to.equal(undefined) + }) + + it('times out rather than hanging the loop', async () => { + const result = await runReadinessProbe(`${base}/hang`, [200]) + expect(result.ok).to.equal(false) + expect(result.error).to.equal('timeout') + }) +}) + +describe('buildModelDownload', () => { + const sample = { downloadedBytes: 1_500_000_000, files: 3, inFlight: 1 } + + it('reports a percentage when the model size is known', () => { + const result = buildModelDownload(sample, 3_000_000_000, 'Qwen/Qwen2.5-7B-Instruct') + expect(result.percent).to.equal(50) + expect(result.totalBytes).to.equal(3_000_000_000) + expect(result.modelId).to.equal('Qwen/Qwen2.5-7B-Instruct') + expect(result.filesInFlight).to.equal(1) + }) + + it('reports bytes with NO percentage when the size is unknown', () => { + // A local path, an object-store URI or a repo the Hub has not indexed: the client shows an + // indeterminate bar rather than a ratio against a guessed denominator. + const result = buildModelDownload(sample, null, null) + expect(result.downloadedBytes).to.equal(1_500_000_000) + expect(result.percent).to.equal(undefined) + expect(result.totalBytes).to.equal(undefined) + expect(result.modelId).to.equal(undefined) + }) + + it('caps at 100% rather than reporting more than the whole model', () => { + // The Hub reports the REPO's safetensors size while the engine downloads only what it needs, + // so a repo carrying several variants can be exceeded. + const result = buildModelDownload( + { ...sample, downloadedBytes: 4_000_000_000 }, + 3_000_000_000, + 'a/b' + ) + expect(result.percent).to.equal(100) + }) +}) From 97017194754e600dab67ed730f7eeb67729f8635 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 15:12:05 +0300 Subject: [PATCH 2/8] fix test --- src/components/c2d/compute_engine_docker.ts | 24 +++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index 0039c7069..293524dfe 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -5248,15 +5248,27 @@ export class C2DEngineDocker extends C2DEngine { job.runtimeMetrics = undefined // Same for readiness: the replacement container has to earn "ready" again (an Edit relaunch // re-downloads the model), and reporting the outgoing container's ready state would hand the - // user a live endpoint minutes before there is one. The probe SPEC is kept — a restart reuses - // the stored container spec, so it reuses the check that matches it. - const restartEngine = resolveServiceEngine(job) - job.readiness = restartEngine - ? { state: 'waiting', engine: restartEngine.id } + // user a live endpoint minutes before there is one. + // + // Resolved defensively: this runs OUTSIDE the try below that turns a failure into a persisted + // Error status, so anything thrown here would abandon the restart with the job stuck reading + // Restarting forever. Readiness is best-effort reporting — it must never be the reason a + // lifecycle operation fails. + let restartEngineId: string | undefined + try { + restartEngineId = resolveServiceEngine(job)?.id + } catch (e: any) { + CORE_LOGGER.debug(`restart ${serviceId}: engine detection failed: ${e?.message}`) + } + job.readiness = restartEngineId + ? { state: 'waiting', engine: restartEngineId } : undefined job.imagePull = undefined job.modelDownload = undefined - this.serviceProbeUrls.delete(serviceId) + // Optional-chained deliberately: this cache is best-effort bookkeeping for the next probe, and + // a restart must not fail because of it. (It is also absent on an engine built without running + // field initializers, which is how the unit tests construct one.) + this.serviceProbeUrls?.delete(serviceId) await this.db.updateServiceJob(job) // Live Docker handles for the newly-created container/network, tracked so the From f66801a9753153e4a497bc12b43d2a642b3ccb6a Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 15:53:14 +0300 Subject: [PATCH 3/8] fix automated code review --- src/components/c2d/compute_engine_docker.ts | 18 +- src/components/c2d/modelDownload.ts | 225 +++++++++++++++++--- src/components/c2d/serviceReadiness.ts | 6 +- src/components/core/service/utils.ts | 9 +- 4 files changed, 217 insertions(+), 41 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index 293524dfe..a6546a6d9 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -4525,6 +4525,11 @@ export class C2DEngineDocker extends C2DEngine { const now = Date.now() const wasReady = job.readiness?.state === 'ready' + // Has this container EVER answered? `wasReady` only covers the last sample, so a service + // that fails twice running would fall back to "waiting" on the second — reading as "still + // starting up" for something that already served and then stopped. readySince is set on the + // first success and kept through failures, so it is the durable answer. + const everReady = wasReady || job.readiness?.readySince !== undefined // Hold off until the container has had a moment to bind its port — probing a process that // has not called listen() yet only produces noise in the node's own logs. const startedAt = Date.parse(details.State?.StartedAt ?? '') @@ -4590,7 +4595,7 @@ export class C2DEngineDocker extends C2DEngine { : { // Only a service that HAD answered can be "failing"; one that never has is still // warming up, which is the normal state for the first minutes of a model server. - state: wasReady ? 'failing' : 'waiting', + state: everReady ? 'failing' : 'waiting', engine: engine.id, readySince: job.readiness?.readySince, lastCheckedAt: now, @@ -4611,11 +4616,12 @@ export class C2DEngineDocker extends C2DEngine { } // While the engine is still warming up, sample how far its model download has got. Skipped - // once ready: the files are there, and the walk is pure overhead from then on. - const modelDownload = - readiness.state === 'ready' - ? undefined - : await this.sampleModelDownload(job, engine) + // once the service has EVER been ready: the files are on disk from that point on, so the walk + // would only re-measure a finished download — including while a ready service is failing, + // when it says nothing about why. + const modelDownload = everReady + ? undefined + : await this.sampleModelDownload(job, engine) // Same cross-process guard as the metrics write: a lifecycle transition must win. if ( diff --git a/src/components/c2d/modelDownload.ts b/src/components/c2d/modelDownload.ts index 6dd78a47d..25d7649c7 100644 --- a/src/components/c2d/modelDownload.ts +++ b/src/components/c2d/modelDownload.ts @@ -33,69 +33,215 @@ import { CORE_LOGGER } from '../../utils/logging/common.js' // against live downloads. const IN_FLIGHT_SUFFIXES = ['.incomplete', '.downloadInProgress'] -// Read at most this many tar entries. A model repo has tens of files; a bound stops a pathological -// cache (or a wrong path pointing at something huge) from walking forever. -const MAX_CACHE_ENTRIES = 5000 +// A model repo has tens of files; a bound stops a pathological cache (or a mistyped path pointing +// at something huge) from issuing thousands of stat calls. +const MAX_BLOB_FILES = 512 + +// Docker returns a path's stat in this header, base64-encoded JSON, with NO response body. +const PATH_STAT_HEADER = 'x-docker-container-path-stat' + +/** One file's size, read from Docker's stat header. null when the path is gone or unreadable. */ +async function statContainerPath( + container: Dockerode.Container, + path: string +): Promise { + try { + const info: any = await container.infoArchive({ path }) + const header = info?.headers?.[PATH_STAT_HEADER] + if (!header) return null + const stat = JSON.parse(Buffer.from(String(header), 'base64').toString()) + const size = Number(stat?.size) + return Number.isFinite(size) ? size : null + } catch { + // Vanished between listing and stat (a rename on completion), or never existed. + return null + } +} /** - * Sums the cache's byte counts by streaming the directory out of the container and reading ONLY the - * tar HEADERS (name, size, type) — the payload is discarded as it arrives, so nothing large is - * transferred and no shell, volume, or storage-driver assumption is involved. Returns null when the - * cache does not exist yet (the normal state for the first seconds) or cannot be read. + * Sums the cache's byte counts WITHOUT transferring any of it. + * + * Docker's archive endpoint streams file CONTENTS, and it does so for any directory that contains + * them — measured at 367 MB of socket traffic to list a 350 MB `blobs/` folder, scaling with the + * model (a 15 GB one would move 15 GB, every few seconds). So `blobs/` is never listed. Instead: + * + * 1. `snapshots/` is listed, which holds only directories and symlinks — 2.5 KB and ~14 ms, + * whatever the model's size — and every symlink's target names a blob. + * 2. Each blob is then stat-ed through Docker's path-stat header, which carries the size and + * returns no body at all (~9 ms, zero content). + * + * In-flight files are found by stat-ing the in-progress name beside each target: a partially + * downloaded blob is written as `.incomplete` (huggingface_hub) or `.downloadInProgress` + * (llama.cpp) and only renamed to `` on completion, while its snapshot symlink already points + * at the final name. + * + * Returns null when the cache does not exist yet (the normal state for the first seconds) or cannot + * be read. Never throws: progress reporting must not be able to disturb a running service. */ export async function readModelDownloadBytes( container: Dockerode.Container, cachePath: string ): Promise<{ downloadedBytes: number; files: number; inFlight: number } | null> { + const repoDirs = await listDirectoryEntries(container, cachePath, 'directory') + if (!repoDirs) return null + + let downloadedBytes = 0 + let files = 0 + let inFlight = 0 + let seen = 0 + + for (const repoDir of repoDirs) { + if (!repoDir.startsWith('models--')) continue + const blobTargets = await listSnapshotBlobTargets( + container, + `${cachePath}/${repoDir}` + ) + for (const blobName of blobTargets) { + if (seen >= MAX_BLOB_FILES) break + seen++ + const blobPath = `${cachePath}/${repoDir}/blobs/${blobName}` + // The finished blob, if the download completed. + const size = await statContainerPath(container, blobPath) + if (size !== null) { + downloadedBytes += size + files++ + continue + } + // Otherwise it may still be arriving under its in-progress name. + for (const suffix of IN_FLIGHT_SUFFIXES) { + const partial = await statContainerPath(container, `${blobPath}${suffix}`) + if (partial !== null) { + downloadedBytes += partial + inFlight++ + break + } + } + } + } + return { downloadedBytes, files, inFlight } +} + +/** + * The blob names a repo's snapshots point at. + * + * `snapshots//` are symlinks into `blobs/`, so this directory carries no file data and + * its listing is cheap regardless of how large the model is. The symlink target's basename is the + * blob to stat. + */ +async function listSnapshotBlobTargets( + container: Dockerode.Container, + repoPath: string +): Promise { + const targets = await listSymlinkTargets(container, `${repoPath}/snapshots`) + return targets ?? [] +} + +/** + * Names of the immediate children of a container directory, of the given tar entry type. + * + * Only ever called on directories that hold no file data of their own (the cache root, which holds + * repo folders), so nothing large crosses the socket — see the note on readModelDownloadBytes. + */ +async function listDirectoryEntries( + container: Dockerode.Container, + path: string, + keep: 'file' | 'directory' +): Promise { + // `stopAtDepth` is what keeps this cheap: Docker tars a directory RECURSIVELY and there is no + // shallow-list option, so a cache root would stream every weight file before this could filter + // by depth. Entries arrive in tree order, so the stream is abandoned as soon as something below + // the level being listed appears — the payloads never start. + const entries = await readTarHeaders(container, path, 1) + if (!entries) return null + return entries + .filter((entry) => entry.depth === 1 && entry.type === keep) + .map((entry) => entry.name) +} + +/** + * Basenames of every symlink target under a directory tree (here: `snapshots//` points + * at `../../blobs/`). Symlinks carry no payload, so this stays cheap at any model size. + */ +async function listSymlinkTargets( + container: Dockerode.Container, + path: string +): Promise { + const entries = await readTarHeaders(container, path) + if (!entries) return null + const targets = new Set() + for (const entry of entries) { + if (entry.type !== 'symlink' || !entry.linkname) continue + const basename = entry.linkname.split('/').filter(Boolean).pop() + if (basename) targets.add(basename) + } + return [...targets] +} + +interface TarEntryHeader { + name: string + type: string + depth: number + linkname?: string +} + +/** Reads a container path's tar entry headers. Null when the path is absent or unreadable. */ +async function readTarHeaders( + container: Dockerode.Container, + path: string, + stopAtDepth?: number +): Promise { let archive: Readable try { - archive = (await container.getArchive({ path: cachePath })) as Readable + archive = (await container.getArchive({ path })) as Readable } catch (error: any) { // 404 until the engine creates the cache — not a failure, just nothing to report yet. if (error?.statusCode !== 404) { - CORE_LOGGER.debug(`[model-download] archive ${cachePath} failed: ${error?.message}`) + CORE_LOGGER.debug(`[model-download] listing ${path} failed: ${error?.message}`) } return null } return await new Promise((resolve) => { const extract = tarStream.extract() - let downloadedBytes = 0 - let files = 0 - let inFlight = 0 - let entries = 0 + const entries: TarEntryHeader[] = [] let settled = false - - const finish = ( - result: { downloadedBytes: number; files: number; inFlight: number } | null - ) => { + const finish = (result: TarEntryHeader[] | null) => { if (settled) return settled = true resolve(result) } extract.on('entry', (header, stream, next) => { - entries++ - // Only regular files carry bytes. A symlink (every snapshots/ entry) reports size 0 and its - // target may not exist yet, so counting those would double-count or contribute nothing. - if (header.type === 'file' && header.name.includes('/blobs/')) { - downloadedBytes += header.size ?? 0 - if (IN_FLIGHT_SUFFIXES.some((suffix) => header.name.endsWith(suffix))) { - inFlight++ - } else { - files++ - } + // Paths arrive prefixed with the requested directory's own name; drop it so `depth` counts + // from the directory that was asked for. + const relative = header.name.split('/').slice(1).filter(Boolean) + if (relative.length > 0) { + entries.push({ + name: relative[relative.length - 1], + type: String(header.type), + depth: relative.length, + linkname: (header as any).linkname + }) + } + // Deeper than asked for: everything wanted at this level has already been seen, and what + // follows is file data. Stop before any of it is transferred. + if (stopAtDepth !== undefined && relative.length > stopAtDepth) { + extract.destroy() + archive.destroy() + finish(entries) + return } stream.on('end', next) - stream.resume() // discard the payload; only the header matters - if (entries > MAX_CACHE_ENTRIES) { + stream.resume() + if (entries.length >= MAX_BLOB_FILES) { extract.destroy() - finish({ downloadedBytes, files, inFlight }) + archive.destroy() + finish(entries) } }) - extract.on('finish', () => finish({ downloadedBytes, files, inFlight })) + extract.on('finish', () => finish(entries)) extract.on('error', (error: any) => { - CORE_LOGGER.debug(`[model-download] tar read failed: ${error?.message}`) + CORE_LOGGER.debug(`[model-download] listing ${path} failed: ${error?.message}`) finish(null) }) archive.on('error', () => finish(null)) @@ -163,7 +309,15 @@ export async function fetchModelTotalBytes( const response = await fetch(`${HF_MODEL_API}/${path}?expand[]=safetensors`, { signal: AbortSignal.timeout(HF_TIMEOUT_MS) }) - if (response.ok) { + if (!response.ok) { + // A 404 is a definitive "no such repo" and worth remembering; anything else (429, 5xx, a + // proxy hiccup) is transient and must not poison the cache for the life of the process. + if (response.status === 404) { + totalBytesCache.set(key, null) + } + return null + } + { const body: any = await response.json() const parameters = body?.safetensors?.parameters if (parameters && typeof parameters === 'object') { @@ -180,10 +334,15 @@ export async function fetchModelTotalBytes( } } } catch (error: any) { + // Network error, timeout, unparseable body: no answer, but not evidence there is none. Return + // without caching so the next sample can ask again. CORE_LOGGER.debug( `[model-download] hub lookup for ${modelId} failed: ${error?.message}` ) + return null } + // A successful response that carried no index IS definitive (a GGUF-only repo has none), so it + // is cached like any other answer. totalBytesCache.set(key, total) return total } diff --git a/src/components/c2d/serviceReadiness.ts b/src/components/c2d/serviceReadiness.ts index beae9b452..8704bb5cb 100644 --- a/src/components/c2d/serviceReadiness.ts +++ b/src/components/c2d/serviceReadiness.ts @@ -169,7 +169,11 @@ export async function runReadinessProbe( const response = await fetch(url, { method: 'GET', signal: controller.signal, - headers: { accept: '*/*' } + headers: { accept: '*/*' }, + // Never chase a redirect: following one would send this request to an address the workload + // chose rather than the container we are probing, and a 3xx is not the engine saying it can + // serve. Returned as a response instead, it simply fails the expected-status check. + redirect: 'manual' }) // Drain, so the socket is not left hanging on either path. await response.body?.cancel().catch(() => {}) diff --git a/src/components/core/service/utils.ts b/src/components/core/service/utils.ts index 73463118c..9b5139e58 100644 --- a/src/components/core/service/utils.ts +++ b/src/components/core/service/utils.ts @@ -96,11 +96,17 @@ export function toListedServiceJob( dockerfile, additionalDockerFiles, readiness, + modelDownload, ...pub } = job // "Is this service actually serving?" is the listing's whole point for a node operator, so the // readiness RESULT is kept — but reduced to what answers that, without the diagnostics (probed // address, error text) meant for the owner. + // Progress is kept for the same reason, minus `modelId`: which model a consumer is running is + // their business, and this listing is readable by any caller, not just the owner. + const listedModelDownload = modelDownload + ? (({ modelId, ...rest }) => rest)(modelDownload) + : undefined return { ...pub, ...(readiness @@ -111,7 +117,8 @@ export function toListedServiceJob( readySince: readiness.readySince } } - : {}) + : {}), + ...(listedModelDownload ? { modelDownload: listedModelDownload } : {}) } } From d9d8ed882bc0fb4dbd8a83fb1eef4651b5f06494 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 18:47:28 +0300 Subject: [PATCH 4/8] fix gemini review --- src/components/c2d/serviceEngines.ts | 19 +++++++++++++++++-- .../unit/service/serviceReadiness.test.ts | 19 +++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/src/components/c2d/serviceEngines.ts b/src/components/c2d/serviceEngines.ts index 7234f26bf..98c07016f 100644 --- a/src/components/c2d/serviceEngines.ts +++ b/src/components/c2d/serviceEngines.ts @@ -151,9 +151,24 @@ export const SERVICE_ENGINE_PROFILES: ServiceEngineProfile[] = [ } ] +/** + * The repository path of an image reference, with any `@digest` and `:tag` removed. `image` is + * documented as the bare name (tag and digest ride in their own fields), but a caller that folds + * them in should still match. Only the last path segment carries a tag, so a registry port + * (`localhost:5000/...`) is left alone. + */ +function stripImageRef(image: string): string { + const withoutDigest = image.split('@')[0] + const lastSlash = withoutDigest.lastIndexOf('/') + const colon = withoutDigest.indexOf(':', lastSlash + 1) + return colon === -1 ? withoutDigest : withoutDigest.slice(0, colon) +} + /** The engine profile for a service, or null when the node does not recognize the image. */ export function resolveServiceEngine(job: ServiceJob): ServiceEngineProfile | null { - const image = (job.image ?? '').trim() - if (!image) return null + const image = stripImageRef((job.image || job.containerImage || '').trim()) + if (!image) { + return null + } return SERVICE_ENGINE_PROFILES.find((profile) => profile.matchesImage(image)) ?? null } diff --git a/src/test/unit/service/serviceReadiness.test.ts b/src/test/unit/service/serviceReadiness.test.ts index d853f937e..c7b3b16fd 100644 --- a/src/test/unit/service/serviceReadiness.test.ts +++ b/src/test/unit/service/serviceReadiness.test.ts @@ -116,6 +116,25 @@ describe('resolveServiceEngine', () => { expect(resolveServiceEngine(job('ghcr.io/vllm/vllm-openai'))?.id).to.equal('vllm') }) + it('matches even when a tag or digest is folded into the image', () => { + expect(resolveServiceEngine(job('vllm/vllm-openai:v0.28.0'))?.id).to.equal('vllm') + expect(resolveServiceEngine(job('vllm/vllm-openai@sha256:abc'))?.id).to.equal('vllm') + expect( + resolveServiceEngine(job('localhost:5000/vllm/vllm-openai:latest'))?.id + ).to.equal('vllm') + expect(resolveServiceEngine(job('localhost:5000/vllm/vllm-openai'))?.id).to.equal( + 'vllm' + ) + expect( + resolveServiceEngine(job('ghcr.io/ggml-org/llama.cpp:server-cuda'))?.id + ).to.equal('llamacpp') + }) + + it('falls back to containerImage when image is missing', () => { + const noImage = { containerImage: 'vllm/vllm-openai:v0.28.0' } as ServiceJob + expect(resolveServiceEngine(noImage)?.id).to.equal('vllm') + }) + it('returns null for an image it does not know, so nothing is gated', () => { expect(resolveServiceEngine(job('nginxinc/nginx-unprivileged'))).to.equal(null) expect(resolveServiceEngine(job('someone/vllm-openai-fork'))).to.equal(null) From 142d499de0832f0b5cd351291768a668a46d7583 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 20:25:17 +0300 Subject: [PATCH 5/8] use container changes instead of overlay2 to check download progress --- src/components/c2d/compute_engine_docker.ts | 42 +- src/components/c2d/modelDownload.ts | 455 +++++++++--------- src/components/c2d/serviceEngines.ts | 7 +- .../unit/service/serviceReadiness.test.ts | 151 +++++- 4 files changed, 418 insertions(+), 237 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index a6546a6d9..34d63694d 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -95,6 +95,7 @@ import { resolveServiceEngine, type ServiceEngineProfile } from './serviceEngine import { buildModelDownload, fetchModelTotalBytes, + isModelDownloadComplete, readModelDownloadBytes } from './modelDownload.js' import type { DockerMountObject } from '../../@types/PersistentStorage.js' @@ -204,6 +205,10 @@ export class C2DEngineDocker extends C2DEngine { // works depends on how this node is deployed (on the host, or itself in Docker), so the probe // discovers it once and reuses it instead of walking the candidate list every few seconds. private serviceProbeUrls: Map = new Map() + // serviceId -> its readiness probe (+ model-download sample) still running in the background. + // The probe is launched fire-and-forget so a slow engine, Docker daemon or Hub never holds up + // InternalLoop; this keeps one probe per service at a time and lets stop() drain them. + private serviceProbesInFlight: Map> = new Map() private readonly serviceLockHolderId: string = makeServiceLockHolderId() private serviceLockHeartbeatTimer: NodeJS.Timeout | null = null // The in-flight service lifecycle promises — processServiceStart() launched @@ -839,6 +844,11 @@ export class C2DEngineDocker extends C2DEngine { await Promise.allSettled([...this.serviceOpPromises]) this.serviceOpPromises.clear() } + // Background readiness probes write to the same DB; let them settle too. + if (this.serviceProbesInFlight.size > 0) { + await Promise.allSettled([...this.serviceProbesInFlight.values()]) + this.serviceProbesInFlight.clear() + } this.isInternalLoopRunning = false // Release any GPU metrics backends (e.g. nvmlShutdown). Best-effort, never throws; the // optional chain also covers engines built without the constructor (e.g. test doubles). @@ -4494,10 +4504,33 @@ export class C2DEngineDocker extends C2DEngine { } // Container up: ask the workload itself whether it can serve requests yet (throttled, // best-effort, lease-free), then sample live runtime metrics on the same terms. - await this.probeServiceReadiness(job, details) + this.launchServiceReadinessProbe(job, details) await this.sampleAndPersistServiceMetrics(job) } + /** + * Runs probeServiceReadiness without holding up the loop. The probe can walk several addresses + * with a timeout each, and its model-download sample waits on the Docker daemon and the Hub — + * seconds, on a bad day, that would otherwise delay every compute job and service start on the + * node. At most one probe per service runs at a time; a tick that finds one in flight skips it. + */ + private launchServiceReadinessProbe( + job: ServiceJob, + details: Dockerode.ContainerInspectInfo + ): void { + if (this.stopped || this.serviceProbesInFlight.has(job.serviceId)) { + return + } + // probeServiceReadiness never throws; the catch only keeps a surprise from surfacing as an + // unhandled rejection. + const probe = this.probeServiceReadiness(job, details) + .catch((e: any) => + CORE_LOGGER.debug(`[readiness] service ${job.serviceId}: ${e?.message}`) + ) + .finally(() => this.serviceProbesInFlight.delete(job.serviceId)) + this.serviceProbesInFlight.set(job.serviceId, probe) + } + /** * Asks a recognized workload whether it can serve requests yet, and persists the answer. * @@ -4664,9 +4697,14 @@ export class C2DEngineDocker extends C2DEngine { engine: ServiceEngineProfile ): Promise { if (!engine.modelCachePath) return undefined + // Downloaded: what remains is the engine loading weights, which the cache says nothing about. + // Returning nothing keeps the stored 100% record as it is. + if (isModelDownloadComplete(job.modelDownload)) { + return undefined + } try { const container = this.docker.getContainer(job.containerId) - const downloaded = await readModelDownloadBytes(container, engine.modelCachePath) + const downloaded = await readModelDownloadBytes(container) if (!downloaded) return undefined // Only a Hugging Face repo has a size the node can look up; a local path or an object-store // URI reports bytes with no total, which clients render as an indeterminate bar. diff --git a/src/components/c2d/modelDownload.ts b/src/components/c2d/modelDownload.ts index 25d7649c7..8a1282af5 100644 --- a/src/components/c2d/modelDownload.ts +++ b/src/components/c2d/modelDownload.ts @@ -1,6 +1,4 @@ import type Dockerode from 'dockerode' -import tarStream from 'tar-stream' -import type { Readable } from 'stream' import type { ServiceModelDownload } from '../../@types/C2D/ServiceOnDemand.js' import { CORE_LOGGER } from '../../utils/logging/common.js' @@ -12,25 +10,24 @@ import { CORE_LOGGER } from '../../utils/logging/common.js' * reports today. The engine exposes no HTTP endpoint while it happens (there is no server yet), so * the only structured source is the cache the downloader writes into: * - * /models----/blobs/ a finished file - * /blobs/.incomplete one being written, size == bytes received + * /models----/blobs/ a finished file + * /models----/blobs/[.].incomplete one being written * * llama.cpp downloads through the same cache but names its partial files `.downloadInProgress`. - * Both are counted; only the complete/in-flight split depends on telling them apart. + * Both are counted; only the complete/in-flight split depends on telling them apart. Recent + * `huggingface_hub` inserts a per-process id before `.incomplete`, so the partial name cannot be + * derived from the blob's — it has to be discovered. * - * The `.incomplete` mechanism is what makes HF downloads resumable, so it is a far steadier contract - * than parsing progress bars out of the container log. Verified against a live download: the byte - * counts track the transfer exactly, and several `.incomplete` files coexist because the hub fetches - * files in parallel — so only the AGGREGATE is meaningful, never a per-file percentage. + * The hub fetches files in parallel, so several partial files coexist — only the AGGREGATE is + * meaningful, never a per-file percentage. * - * The cache cannot supply the denominator: `snapshots//*` are symlinks (tar reports size 0 for - * those) pointing at blobs that do not exist yet, and they appear one by one as files resolve. The + * The cache cannot supply the denominator: the snapshot symlinks that name a repo's files are + * created only AFTER each file finishes, so they say nothing about what is still arriving. The * total therefore comes from the Hub's own metadata — see fetchModelTotalBytes. */ // Suffixes a partially-downloaded file carries while it is being written. `huggingface_hub` (vLLM) -// uses `.incomplete`; llama.cpp's own `-hf` downloader uses `.downloadInProgress`. Both verified -// against live downloads. +// uses `.incomplete`; llama.cpp's own `-hf` downloader uses `.downloadInProgress`. const IN_FLIGHT_SUFFIXES = ['.incomplete', '.downloadInProgress'] // A model repo has tens of files; a bound stops a pathological cache (or a mistyped path pointing @@ -40,18 +37,31 @@ const MAX_BLOB_FILES = 512 // Docker returns a path's stat in this header, base64-encoded JSON, with NO response body. const PATH_STAT_HEADER = 'x-docker-container-path-stat' -/** One file's size, read from Docker's stat header. null when the path is gone or unreadable. */ +// Kind values in Docker's container-changes listing. +const CHANGE_KIND_DELETED = 2 + +interface ContainerPathStat { + size: number + linkTarget?: string +} + +/** One path's stat, read from Docker's stat header. null when the path is gone or unreadable. */ async function statContainerPath( container: Dockerode.Container, path: string -): Promise { +): Promise { try { const info: any = await container.infoArchive({ path }) const header = info?.headers?.[PATH_STAT_HEADER] - if (!header) return null + if (!header) { + return null + } const stat = JSON.parse(Buffer.from(String(header), 'base64').toString()) const size = Number(stat?.size) - return Number.isFinite(size) ? size : null + if (!Number.isFinite(size)) { + return null + } + return { size, linkTarget: stat?.linkTarget || undefined } } catch { // Vanished between listing and stat (a rename on completion), or never existed. return null @@ -59,194 +69,142 @@ async function statContainerPath( } /** - * Sums the cache's byte counts WITHOUT transferring any of it. - * - * Docker's archive endpoint streams file CONTENTS, and it does so for any directory that contains - * them — measured at 367 MB of socket traffic to list a 350 MB `blobs/` folder, scaling with the - * model (a 15 GB one would move 15 GB, every few seconds). So `blobs/` is never listed. Instead: - * - * 1. `snapshots/` is listed, which holds only directories and symlinks — 2.5 KB and ~14 ms, - * whatever the model's size — and every symlink's target names a blob. - * 2. Each blob is then stat-ed through Docker's path-stat header, which carries the size and - * returns no body at all (~9 ms, zero content). - * - * In-flight files are found by stat-ing the in-progress name beside each target: a partially - * downloaded blob is written as `.incomplete` (huggingface_hub) or `.downloadInProgress` - * (llama.cpp) and only renamed to `` on completion, while its snapshot symlink already points - * at the final name. - * - * Returns null when the cache does not exist yet (the normal state for the first seconds) or cannot - * be read. Never throws: progress reporting must not be able to disturb a running service. + * A file's size, following one symlink. Docker's path stat is an lstat, and newer + * `huggingface_hub` can link a blob into a shared store — the link's own size would count a few + * bytes for a multi-gigabyte file. */ -export async function readModelDownloadBytes( +async function fileSize( container: Dockerode.Container, - cachePath: string -): Promise<{ downloadedBytes: number; files: number; inFlight: number } | null> { - const repoDirs = await listDirectoryEntries(container, cachePath, 'directory') - if (!repoDirs) return null - - let downloadedBytes = 0 - let files = 0 - let inFlight = 0 - let seen = 0 - - for (const repoDir of repoDirs) { - if (!repoDir.startsWith('models--')) continue - const blobTargets = await listSnapshotBlobTargets( - container, - `${cachePath}/${repoDir}` - ) - for (const blobName of blobTargets) { - if (seen >= MAX_BLOB_FILES) break - seen++ - const blobPath = `${cachePath}/${repoDir}/blobs/${blobName}` - // The finished blob, if the download completed. - const size = await statContainerPath(container, blobPath) - if (size !== null) { - downloadedBytes += size - files++ - continue - } - // Otherwise it may still be arriving under its in-progress name. - for (const suffix of IN_FLIGHT_SUFFIXES) { - const partial = await statContainerPath(container, `${blobPath}${suffix}`) - if (partial !== null) { - downloadedBytes += partial - inFlight++ - break - } - } - } + path: string +): Promise { + const stat = await statContainerPath(container, path) + if (!stat?.linkTarget) { + return stat?.size ?? null } - return { downloadedBytes, files, inFlight } + const target = stat.linkTarget.startsWith('/') + ? stat.linkTarget + : `${path.slice(0, path.lastIndexOf('/'))}/${stat.linkTarget}` + return (await statContainerPath(container, target))?.size ?? null } -/** - * The blob names a repo's snapshots point at. - * - * `snapshots//` are symlinks into `blobs/`, so this directory carries no file data and - * its listing is cheap regardless of how large the model is. The symlink target's basename is the - * blob to stat. - */ -async function listSnapshotBlobTargets( - container: Dockerode.Container, - repoPath: string -): Promise { - const targets = await listSymlinkTargets(container, `${repoPath}/snapshots`) - return targets ?? [] +/** The in-flight suffix a path ends with, or null for a finished file. */ +function inFlightSuffix(path: string): string | null { + return IN_FLIGHT_SUFFIXES.find((suffix) => path.endsWith(suffix)) ?? null } /** - * Names of the immediate children of a container directory, of the given tar entry type. - * - * Only ever called on directories that hold no file data of their own (the cache root, which holds - * repo folders), so nothing large crosses the socket — see the note on readModelDownloadBytes. + * The finished blob a partial file becomes: `.downloadInProgress`, `.incomplete` and + * `..incomplete` all rename to ``, and a blob name never contains a dot. */ -async function listDirectoryEntries( - container: Dockerode.Container, - path: string, - keep: 'file' | 'directory' -): Promise { - // `stopAtDepth` is what keeps this cheap: Docker tars a directory RECURSIVELY and there is no - // shallow-list option, so a cache root would stream every weight file before this could filter - // by depth. Entries arrive in tree order, so the stream is abandoned as soon as something below - // the level being listed appears — the payloads never start. - const entries = await readTarHeaders(container, path, 1) - if (!entries) return null - return entries - .filter((entry) => entry.depth === 1 && entry.type === keep) - .map((entry) => entry.name) +function finishedBlobPath(partialPath: string): string { + const slash = partialPath.lastIndexOf('/') + const name = partialPath.slice(slash + 1) + return `${partialPath.slice(0, slash + 1)}${name.split('.')[0]}` } /** - * Basenames of every symlink target under a directory tree (here: `snapshots//` points - * at `../../blobs/`). Symlinks carry no payload, so this stays cheap at any model size. + * The files in a Hugging Face cache's `blobs/` folders, from the container's changes listing. + * + * Matched on the cache's own layout (`models----/blobs/`) wherever it sits, not + * under one fixed root: HF_HOME / HF_HUB_CACHE can move it, and the env that would say so rides in + * encrypted userData the node cannot read. Snapshot symlinks and lock files live outside `blobs/` + * and carry no weight data. */ -async function listSymlinkTargets( - container: Dockerode.Container, - path: string -): Promise { - const entries = await readTarHeaders(container, path) - if (!entries) return null - const targets = new Set() - for (const entry of entries) { - if (entry.type !== 'symlink' || !entry.linkname) continue - const basename = entry.linkname.split('/').filter(Boolean).pop() - if (basename) targets.add(basename) +export function selectBlobPaths( + changes: Array<{ Path?: string; Kind?: number }> | null | undefined +): string[] { + const paths = new Set() + for (const change of changes ?? []) { + const path = change?.Path + if (!path || change.Kind === CHANGE_KIND_DELETED) { + continue + } + const segments = path.split('/') + const n = segments.length + if ( + n >= 3 && + segments[n - 3].startsWith('models--') && + segments[n - 2] === 'blobs' && + segments[n - 1] + ) { + paths.add(path) + } } - return [...targets] -} - -interface TarEntryHeader { - name: string - type: string - depth: number - linkname?: string + return [...paths].slice(0, MAX_BLOB_FILES) } -/** Reads a container path's tar entry headers. Null when the path is absent or unreadable. */ -async function readTarHeaders( - container: Dockerode.Container, - path: string, - stopAtDepth?: number -): Promise { - let archive: Readable +/** + * Sums the cache's byte counts WITHOUT transferring any of it. + * + * Docker's changes endpoint lists every path the container has added or modified on top of its + * image — names only, no contents, computed by the daemon from the container's writable layer. It + * sees partial files under whatever name the downloader picked, in every repo folder, with no + * dependence on when symlinks appear or what order a directory lists in. Each file is then stat-ed + * through Docker's path-stat header, which carries the size and returns no body. + * + * A partial file renamed between the listing and its stat is looked up again under its finished + * name, so a file completing mid-sample is counted once rather than dropped. + * + * Only sees the container's own filesystem: a cache mounted from a volume is invisible here. + * Engine-agnostic within that: anything downloading through the Hugging Face cache layout is + * measured, wherever the cache was put. + * + * Returns null when the cache does not exist yet (the normal state for the first seconds) or cannot + * be read. Never throws: progress reporting must not be able to disturb a running service. + */ +export async function readModelDownloadBytes( + container: Dockerode.Container +): Promise<{ downloadedBytes: number; files: number; inFlight: number } | null> { + let changes: Array<{ Path?: string; Kind?: number }> | null try { - archive = (await container.getArchive({ path })) as Readable + changes = await container.changes() } catch (error: any) { - // 404 until the engine creates the cache — not a failure, just nothing to report yet. - if (error?.statusCode !== 404) { - CORE_LOGGER.debug(`[model-download] listing ${path} failed: ${error?.message}`) - } + CORE_LOGGER.debug(`[model-download] listing changes failed: ${error?.message}`) + return null + } + const blobPaths = selectBlobPaths(changes) + if (blobPaths.length === 0) { return null } - return await new Promise((resolve) => { - const extract = tarStream.extract() - const entries: TarEntryHeader[] = [] - let settled = false - const finish = (result: TarEntryHeader[] | null) => { - if (settled) return - settled = true - resolve(result) - } + const listed = new Set(blobPaths) + const counted = new Set() + let downloadedBytes = 0 + let files = 0 + let inFlight = 0 - extract.on('entry', (header, stream, next) => { - // Paths arrive prefixed with the requested directory's own name; drop it so `depth` counts - // from the directory that was asked for. - const relative = header.name.split('/').slice(1).filter(Boolean) - if (relative.length > 0) { - entries.push({ - name: relative[relative.length - 1], - type: String(header.type), - depth: relative.length, - linkname: (header as any).linkname - }) - } - // Deeper than asked for: everything wanted at this level has already been seen, and what - // follows is file data. Stop before any of it is transferred. - if (stopAtDepth !== undefined && relative.length > stopAtDepth) { - extract.destroy() - archive.destroy() - finish(entries) - return - } - stream.on('end', next) - stream.resume() - if (entries.length >= MAX_BLOB_FILES) { - extract.destroy() - archive.destroy() - finish(entries) + for (const path of blobPaths) { + if (counted.has(path)) { + continue + } + const size = await fileSize(container, path) + if (size !== null) { + counted.add(path) + downloadedBytes += size + if (inFlightSuffix(path)) { + inFlight++ + } else { + files++ } - }) - extract.on('finish', () => finish(entries)) - extract.on('error', (error: any) => { - CORE_LOGGER.debug(`[model-download] listing ${path} failed: ${error?.message}`) - finish(null) - }) - archive.on('error', () => finish(null)) - archive.pipe(extract) - }) + continue + } + if (!inFlightSuffix(path)) { + continue + } + // Finished between the listing and the stat: count it under the name it now has, unless the + // listing already carries that name and it will be (or was) counted there. + const finished = finishedBlobPath(path) + if (listed.has(finished) || counted.has(finished)) { + continue + } + const finishedSize = await fileSize(container, finished) + if (finishedSize !== null) { + counted.add(finished) + downloadedBytes += finishedSize + files++ + } + } + return { downloadedBytes, files, inFlight } } // Bytes on the wire per parameter, by the dtype key the Hub reports. Anything unrecognized counts @@ -275,6 +233,11 @@ const HF_TIMEOUT_MS = 8000 // One lookup per model for the life of the process: the answer cannot change for a given repo, and // this is read on the metrics cadence for every starting service. const totalBytesCache = new Map() +// A lookup that failed transiently (timeout, 429, 5xx, network) is not asked again for this long. +// Without it every sample of every starting service would wait out the full timeout while the Hub +// is slow; with it the total simply appears a little later once the Hub recovers. +const HUB_RETRY_AFTER_MS = 60_000 +const hubFailedAt = new Map() /** * The size of the weights an engine will download for a model, from the Hub's safetensors index. @@ -294,57 +257,75 @@ export async function fetchModelTotalBytes( ): Promise { const key = quant ? `${modelId}:${quant}` : modelId if (totalBytesCache.has(key)) return totalBytesCache.get(key) ?? null + const failedAt = hubFailedAt.get(key) + if (failedAt !== undefined && Date.now() - failedAt < HUB_RETRY_AFTER_MS) { + return null + } + const total = await lookupModelTotalBytes(modelId, quant) + if (total === undefined) { + hubFailedAt.set(key, Date.now()) + return null + } + hubFailedAt.delete(key) + totalBytesCache.set(key, total) + return total +} - let total: number | null = null +/** + * One Hub lookup. The answer (a size, or null for a definitive "none") is cached by the caller; + * undefined means the Hub could not be asked, which is backed off rather than cached. + */ +async function lookupModelTotalBytes( + modelId: string, + quant?: string +): Promise { + if (quant) { + return await fetchGgufFileBytes(modelId, quant) + } try { - if (quant) { - const ggufTotal = await fetchGgufFileBytes(modelId, quant) - totalBytesCache.set(key, ggufTotal) - return ggufTotal - } - const path = modelId - .split('/') - .map((segment) => encodeURIComponent(segment)) - .join('/') - const response = await fetch(`${HF_MODEL_API}/${path}?expand[]=safetensors`, { - signal: AbortSignal.timeout(HF_TIMEOUT_MS) - }) - if (!response.ok) { - // A 404 is a definitive "no such repo" and worth remembering; anything else (429, 5xx, a - // proxy hiccup) is transient and must not poison the cache for the life of the process. - if (response.status === 404) { - totalBytesCache.set(key, null) + const response = await fetch( + `${HF_MODEL_API}/${hubPath(modelId)}?expand[]=safetensors`, + { + signal: AbortSignal.timeout(HF_TIMEOUT_MS) } - return null + ) + if (!response.ok) { + // A 404 is a definitive "no such repo"; anything else (429, 5xx, a proxy hiccup) is + // transient and must not poison the cache for the life of the process. + return response.status === 404 ? null : undefined } - { - const body: any = await response.json() - const parameters = body?.safetensors?.parameters - if (parameters && typeof parameters === 'object') { - const bytes = Object.entries(parameters).reduce( - (sum, [dtype, count]) => - sum + (Number(count) || 0) * (BYTES_PER_PARAM[dtype] ?? 2), - 0 - ) - total = bytes > 0 ? Math.round(bytes) : null - } else if (Number(body?.safetensors?.total) > 0) { - // Only a parameter count, no dtype breakdown: assume half precision, the default these - // repos are served in. - total = Math.round(Number(body.safetensors.total) * 2) - } + const body: any = await response.json() + const parameters = body?.safetensors?.parameters + if (parameters && typeof parameters === 'object') { + const bytes = Object.entries(parameters).reduce( + (sum, [dtype, count]) => + sum + (Number(count) || 0) * (BYTES_PER_PARAM[dtype] ?? 2), + 0 + ) + return bytes > 0 ? Math.round(bytes) : null } + if (Number(body?.safetensors?.total) > 0) { + // Only a parameter count, no dtype breakdown: assume half precision, the default these + // repos are served in. + return Math.round(Number(body.safetensors.total) * 2) + } + // A successful response that carried no index IS definitive (a GGUF-only repo has none). + return null } catch (error: any) { - // Network error, timeout, unparseable body: no answer, but not evidence there is none. Return - // without caching so the next sample can ask again. + // Network error, timeout, unparseable body: no answer, but not evidence there is none. CORE_LOGGER.debug( `[model-download] hub lookup for ${modelId} failed: ${error?.message}` ) - return null + return undefined } - // A successful response that carried no index IS definitive (a GGUF-only repo has none), so it - // is cached like any other answer. - totalBytesCache.set(key, total) - return total +} + +/** A repo id as a Hub API path, each segment encoded. */ +function hubPath(modelId: string): string { + return modelId + .split('/') + .map((segment) => encodeURIComponent(segment)) + .join('/') } /** @@ -355,20 +336,20 @@ export async function fetchModelTotalBytes( * but the Hub does list every file with its size, and the engine downloads exactly one of them. * Matching on the quant tag gives the exact denominator instead of the whole repo's contents, which * for a repo carrying a dozen quantizations would be an order of magnitude out. + * + * Same contract as lookupModelTotalBytes: null is a definitive "no size", undefined a failure to ask. */ async function fetchGgufFileBytes( modelId: string, quant: string -): Promise { +): Promise { try { - const path = modelId - .split('/') - .map((segment) => encodeURIComponent(segment)) - .join('/') - const response = await fetch(`${HF_MODEL_API}/${path}?blobs=true`, { + const response = await fetch(`${HF_MODEL_API}/${hubPath(modelId)}?blobs=true`, { signal: AbortSignal.timeout(HF_TIMEOUT_MS) }) - if (!response.ok) return null + if (!response.ok) { + return response.status === 404 ? null : undefined + } const body: any = await response.json() const siblings: any[] = Array.isArray(body?.siblings) ? body.siblings : [] const wanted = quant.toLowerCase() @@ -382,10 +363,22 @@ async function fetchGgufFileBytes( CORE_LOGGER.debug( `[model-download] gguf lookup for ${modelId}:${quant} failed: ${error?.message}` ) - return null + return undefined } } +/** + * Whether a recorded download has finished: the total reached AND nothing still arriving. The total + * is an estimate (dtype × parameters), so on its own it can read 100% while the last shard is still + * being written; no partial files left is what makes it final. A record with no total never + * completes here, and sampling simply carries on until the service is ready. + */ +export function isModelDownloadComplete( + download: ServiceModelDownload | undefined +): boolean { + return download?.percent === 100 && download.filesInFlight === 0 +} + /** Builds the record persisted on the job, capping the ratio (see the note on repo variants). */ export function buildModelDownload( downloaded: { downloadedBytes: number; files: number; inFlight: number }, diff --git a/src/components/c2d/serviceEngines.ts b/src/components/c2d/serviceEngines.ts index 98c07016f..621e6e000 100644 --- a/src/components/c2d/serviceEngines.ts +++ b/src/components/c2d/serviceEngines.ts @@ -27,9 +27,10 @@ export interface ServiceEngineProfile { expectStatus: number[] } /** - * Where this engine caches the model it downloads at startup, inside the container. Absent for an - * engine whose download the node cannot observe — readiness still works, progress simply isn't - * reported. + * Where this engine caches the model it downloads at startup, inside the container, by default. + * Its presence is what turns progress reporting on; the files themselves are found by the Hugging + * Face cache layout wherever it sits (see modelDownload). Absent for an engine whose download the + * node cannot observe — readiness still works, progress simply isn't reported. */ modelCachePath?: string /** diff --git a/src/test/unit/service/serviceReadiness.test.ts b/src/test/unit/service/serviceReadiness.test.ts index c7b3b16fd..614987199 100644 --- a/src/test/unit/service/serviceReadiness.test.ts +++ b/src/test/unit/service/serviceReadiness.test.ts @@ -11,7 +11,13 @@ import { runReadinessProbe } from '../../../components/c2d/serviceReadiness.js' import { resolveServiceEngine } from '../../../components/c2d/serviceEngines.js' -import { buildModelDownload } from '../../../components/c2d/modelDownload.js' +import { + buildModelDownload, + fetchModelTotalBytes, + isModelDownloadComplete, + readModelDownloadBytes, + selectBlobPaths +} from '../../../components/c2d/modelDownload.js' describe('ImagePullTracker', () => { // The daemon's real event sequence for one layer, minus the narration lines. @@ -303,3 +309,146 @@ describe('buildModelDownload', () => { expect(result.percent).to.equal(100) }) }) + +describe('model download bytes from container changes', () => { + const cache = '/root/.cache/huggingface/hub' + const repo = `${cache}/models--Qwen--Qwen2.5-7B` + + // A container double: `changes()` lists paths, `infoArchive()` answers Docker's stat header. + function fakeContainer( + changes: Array<{ Path: string; Kind: number }>, + files: Record + ) { + return { + changes: () => Promise.resolve(changes), + infoArchive: ({ path }: { path: string }) => { + const stat = files[path] + if (!stat) { + return Promise.reject( + Object.assign(new Error('not found'), { statusCode: 404 }) + ) + } + const header = Buffer.from(JSON.stringify(stat)).toString('base64') + return Promise.resolve({ headers: { 'x-docker-container-path-stat': header } }) + } + } as any + } + + it('keeps only files directly inside a repo blobs folder', () => { + const paths = selectBlobPaths([ + { Path: cache, Kind: 1 }, + { Path: `${cache}/.locks/models--Qwen--Qwen2.5-7B/abc.lock`, Kind: 1 }, + { Path: `${repo}/blobs`, Kind: 1 }, + { Path: `${repo}/blobs/abc`, Kind: 1 }, + { Path: `${repo}/blobs/def.1a2b3c4d.incomplete`, Kind: 1 }, + { Path: `${repo}/blobs/gone`, Kind: 2 }, + { Path: `${repo}/snapshots/sha/model.safetensors`, Kind: 1 }, + { Path: '/tmp/other', Kind: 1 }, + // HF_HOME moved the cache: still found by its layout. + { Path: '/data/hf/hub/models--org--name/blobs/xyz', Kind: 1 } + ]) + expect(paths).to.deep.equal([ + `${repo}/blobs/abc`, + `${repo}/blobs/def.1a2b3c4d.incomplete`, + '/data/hf/hub/models--org--name/blobs/xyz' + ]) + }) + + it('counts finished and in-flight files, whatever the partial file is named', async () => { + const container = fakeContainer( + [ + { Path: `${repo}/blobs/aaa`, Kind: 1 }, + { Path: `${repo}/blobs/bbb.1a2b3c4d.incomplete`, Kind: 1 }, + { Path: `${repo}/blobs/ccc.downloadInProgress`, Kind: 1 } + ], + { + [`${repo}/blobs/aaa`]: { size: 1000 }, + [`${repo}/blobs/bbb.1a2b3c4d.incomplete`]: { size: 300 }, + [`${repo}/blobs/ccc.downloadInProgress`]: { size: 200 } + } + ) + expect(await readModelDownloadBytes(container)).to.deep.equal({ + downloadedBytes: 1500, + files: 1, + inFlight: 2 + }) + }) + + it('counts a file that finished between the listing and the stat under its new name', async () => { + const container = fakeContainer( + [{ Path: `${repo}/blobs/bbb.1a2b3c4d.incomplete`, Kind: 1 }], + { [`${repo}/blobs/bbb`]: { size: 900 } } + ) + expect(await readModelDownloadBytes(container)).to.deep.equal({ + downloadedBytes: 900, + files: 1, + inFlight: 0 + }) + }) + + it('follows a blob symlinked into a shared store', async () => { + const container = fakeContainer([{ Path: `${repo}/blobs/aaa`, Kind: 1 }], { + [`${repo}/blobs/aaa`]: { size: 40, linkTarget: '../../shared/aaa' }, + [`${repo}/blobs/../../shared/aaa`]: { size: 5000 } + }) + expect((await readModelDownloadBytes(container))?.downloadedBytes).to.equal(5000) + }) + + it('reports nothing while the cache holds no blobs yet, or when Docker fails', async () => { + expect(await readModelDownloadBytes(fakeContainer([], {}))).to.equal(null) + const failing = { changes: () => Promise.reject(new Error('boom')) } as any + expect(await readModelDownloadBytes(failing)).to.equal(null) + }) +}) + +describe('fetchModelTotalBytes', () => { + const realFetch = globalThis.fetch + afterEach(() => { + globalThis.fetch = realFetch + }) + + it('backs off after a transient Hub failure instead of retrying every sample', async () => { + let calls = 0 + globalThis.fetch = (() => { + calls++ + return Promise.resolve(new Response('busy', { status: 503 })) + }) as typeof fetch + expect(await fetchModelTotalBytes('backoff-test/model-a')).to.equal(null) + expect(await fetchModelTotalBytes('backoff-test/model-a')).to.equal(null) + expect(calls).to.equal(1) + }) + + it('caches a successful answer for good', async () => { + let calls = 0 + globalThis.fetch = (() => { + calls++ + return Promise.resolve( + Response.json({ safetensors: { parameters: { BF16: 1000, F32: 10 } } }) + ) + }) as typeof fetch + expect(await fetchModelTotalBytes('backoff-test/model-b')).to.equal(2040) + expect(await fetchModelTotalBytes('backoff-test/model-b')).to.equal(2040) + expect(calls).to.equal(1) + }) +}) + +describe('isModelDownloadComplete', () => { + const record = (percent: number | undefined, filesInFlight: number) => + ({ + downloadedBytes: 1, + percent, + filesComplete: 1, + filesInFlight, + updatedAt: 0 + }) as any + + it('is complete only at 100% with nothing still arriving', () => { + expect(isModelDownloadComplete(record(100, 0))).to.equal(true) + // The total is an estimate: 100% with a shard still being written is not done. + expect(isModelDownloadComplete(record(100, 1))).to.equal(false) + expect(isModelDownloadComplete(record(99, 0))).to.equal(false) + // No total known: never declared complete, sampling carries on. + expect(isModelDownloadComplete(record(undefined, 0))).to.equal(false) + expect(isModelDownloadComplete(undefined)).to.equal(false) + }) +}) From 54a8953d65cfbbb8b30321a8fbcf42476ef2fd9a Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 20:41:44 +0300 Subject: [PATCH 6/8] hybrid model download discovery --- src/components/c2d/compute_engine_docker.ts | 15 +- src/components/c2d/modelDownload.ts | 153 ++++++++++++++---- .../unit/service/serviceReadiness.test.ts | 99 ++++++++++++ 3 files changed, 238 insertions(+), 29 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index 34d63694d..bd8018b4b 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -96,7 +96,7 @@ import { buildModelDownload, fetchModelTotalBytes, isModelDownloadComplete, - readModelDownloadBytes + ModelDownloadSampler } from './modelDownload.js' import type { DockerMountObject } from '../../@types/PersistentStorage.js' import { resolveServiceImage } from './serviceResourceMatching.js' @@ -205,6 +205,8 @@ export class C2DEngineDocker extends C2DEngine { // works depends on how this node is deployed (on the host, or itself in Docker), so the probe // discovers it once and reuses it instead of walking the candidate list every few seconds. private serviceProbeUrls: Map = new Map() + // Per-service model-download sampling state (which cache files to stat, when to re-list them). + private modelDownloadSampler = new ModelDownloadSampler() // serviceId -> its readiness probe (+ model-download sample) still running in the background. // The probe is launched fire-and-forget so a slow engine, Docker daemon or Hub never holds up // InternalLoop; this keeps one probe per service at a time and lets stop() drain them. @@ -4481,6 +4483,15 @@ export class C2DEngineDocker extends C2DEngine { !this.serviceOpsInFlight.has(svc.serviceId) ) await Promise.all(runningOnly.map((svc) => this.checkServiceContainerHealth(svc))) + // Forget per-service probe state for services that are no longer running (stopped, failed, + // expired), whichever path ended them. + const running = new Set(services.map((svc) => svc.serviceId)) + this.modelDownloadSampler.retain(running) + for (const serviceId of this.serviceProbeUrls.keys()) { + if (!running.has(serviceId)) { + this.serviceProbeUrls.delete(serviceId) + } + } return runningOnly } @@ -4704,7 +4715,7 @@ export class C2DEngineDocker extends C2DEngine { } try { const container = this.docker.getContainer(job.containerId) - const downloaded = await readModelDownloadBytes(container) + const downloaded = await this.modelDownloadSampler.sample(job.serviceId, container) if (!downloaded) return undefined // Only a Hugging Face repo has a size the node can look up; a local path or an object-store // URI reports bytes with no total, which clients render as an indeterminate bar. diff --git a/src/components/c2d/modelDownload.ts b/src/components/c2d/modelDownload.ts index 8a1282af5..073f1cc23 100644 --- a/src/components/c2d/modelDownload.ts +++ b/src/components/c2d/modelDownload.ts @@ -133,42 +133,48 @@ export function selectBlobPaths( return [...paths].slice(0, MAX_BLOB_FILES) } +export interface ModelDownloadBytes { + downloadedBytes: number + files: number + inFlight: number +} + /** - * Sums the cache's byte counts WITHOUT transferring any of it. - * - * Docker's changes endpoint lists every path the container has added or modified on top of its - * image — names only, no contents, computed by the daemon from the container's writable layer. It - * sees partial files under whatever name the downloader picked, in every repo folder, with no - * dependence on when symlinks appear or what order a directory lists in. Each file is then stat-ed - * through Docker's path-stat header, which carries the size and returns no body. - * - * A partial file renamed between the listing and its stat is looked up again under its finished - * name, so a file completing mid-sample is counted once rather than dropped. + * The cache's blob paths, from Docker's changes endpoint: every path the container has added or + * modified on top of its image — names only, no contents. It sees partial files under whatever + * name the downloader picked, in every repo folder, with no dependence on when symlinks appear or + * what order a directory lists in. * - * Only sees the container's own filesystem: a cache mounted from a volume is invisible here. - * Engine-agnostic within that: anything downloading through the Hugging Face cache layout is - * measured, wherever the cache was put. - * - * Returns null when the cache does not exist yet (the normal state for the first seconds) or cannot - * be read. Never throws: progress reporting must not be able to disturb a running service. + * The one expensive call here: on the containerd image store the daemon compares the container + * against its whole image (measured 0.1-1.6s, growing with image size), so ModelDownloadSampler + * calls it sparingly. Null when Docker could not answer. */ -export async function readModelDownloadBytes( +async function discoverBlobPaths( container: Dockerode.Container -): Promise<{ downloadedBytes: number; files: number; inFlight: number } | null> { - let changes: Array<{ Path?: string; Kind?: number }> | null +): Promise { try { - changes = await container.changes() + return selectBlobPaths(await container.changes()) } catch (error: any) { CORE_LOGGER.debug(`[model-download] listing changes failed: ${error?.message}`) return null } - const blobPaths = selectBlobPaths(changes) - if (blobPaths.length === 0) { - return null - } +} +/** + * Sizes the given blob paths through Docker's path-stat header, which carries the size and returns + * no body. A partial file renamed since it was listed is looked up again under its finished name, + * so a file completing mid-sample is counted once rather than dropped; `paths` comes back with + * those renames applied, and `renamed` says a download finished — usually the moment the + * downloader starts its next file. + */ +async function measureBlobPaths( + container: Dockerode.Container, + blobPaths: string[] +): Promise<{ bytes: ModelDownloadBytes; paths: string[]; renamed: boolean }> { const listed = new Set(blobPaths) const counted = new Set() + const paths: string[] = [] + let renamed = false let downloadedBytes = 0 let files = 0 let inFlight = 0 @@ -180,6 +186,7 @@ export async function readModelDownloadBytes( const size = await fileSize(container, path) if (size !== null) { counted.add(path) + paths.push(path) downloadedBytes += size if (inFlightSuffix(path)) { inFlight++ @@ -191,8 +198,9 @@ export async function readModelDownloadBytes( if (!inFlightSuffix(path)) { continue } - // Finished between the listing and the stat: count it under the name it now has, unless the - // listing already carries that name and it will be (or was) counted there. + // Finished since it was listed: count it under the name it now has, unless the list already + // carries that name and it will be (or was) counted there. + renamed = true const finished = finishedBlobPath(path) if (listed.has(finished) || counted.has(finished)) { continue @@ -200,11 +208,102 @@ export async function readModelDownloadBytes( const finishedSize = await fileSize(container, finished) if (finishedSize !== null) { counted.add(finished) + paths.push(finished) downloadedBytes += finishedSize files++ } } - return { downloadedBytes, files, inFlight } + return { bytes: { downloadedBytes, files, inFlight }, paths, renamed } +} + +/** + * Sums the cache's byte counts WITHOUT transferring any of it: one changes listing, then a stat + * per blob. Only sees the container's own filesystem — a cache mounted from a volume is invisible + * here. Engine-agnostic within that: anything downloading through the Hugging Face cache layout is + * measured, wherever the cache was put. + * + * Returns null when the cache does not exist yet (the normal state for the first seconds) or cannot + * be read. Never throws: progress reporting must not be able to disturb a running service. + */ +export async function readModelDownloadBytes( + container: Dockerode.Container +): Promise { + const blobPaths = await discoverBlobPaths(container) + if (!blobPaths || blobPaths.length === 0) { + return null + } + return (await measureBlobPaths(container, blobPaths)).bytes +} + +// Re-listing cadence. A known partial file finishing triggers a re-list on the next sample +// regardless, since that is when the downloader starts its next file; these bound how late a file +// that started on its own is noticed. +export const BLOB_REDISCOVER_MS = 30_000 +// Before the cache holds anything: the first bytes should show up quickly. +export const BLOB_EMPTY_REDISCOVER_MS = 10_000 + +interface BlobDiscovery { + containerId: string + paths: string[] + discoveredAt: number + /** Set when a known download finished: re-list on the next sample. */ + stale: boolean +} + +/** + * Samples model downloads per service, re-listing the container's changes only when needed and + * stat-ing the known paths in between — the stats are cheap, the listing is not. + * + * A file that starts without another finishing is counted up to BLOB_REDISCOVER_MS late, so the bar + * can pause and then catch up; it never goes backwards. Readiness does not depend on any of this. + */ +export class ModelDownloadSampler { + private discoveries = new Map() + + async sample( + serviceId: string, + container: Dockerode.Container, + now: number = Date.now() + ): Promise { + let discovery = this.discoveries.get(serviceId) + // A restart runs a new container, whose cache starts from nothing. + if (discovery && discovery.containerId !== container.id) { + discovery = undefined + } + if (!discovery || this.isDue(discovery, now)) { + const paths = await discoverBlobPaths(container) + if (!paths) { + return null + } + discovery = { containerId: container.id, paths, discoveredAt: now, stale: false } + this.discoveries.set(serviceId, discovery) + } + if (discovery.paths.length === 0) { + return null + } + const measured = await measureBlobPaths(container, discovery.paths) + discovery.paths = measured.paths + discovery.stale = discovery.stale || measured.renamed + return measured.bytes + } + + /** Drops every service not in `keep` — called with the services still running. */ + retain(keep: Set): void { + for (const serviceId of this.discoveries.keys()) { + if (!keep.has(serviceId)) { + this.discoveries.delete(serviceId) + } + } + } + + private isDue(discovery: BlobDiscovery, now: number): boolean { + if (discovery.stale) { + return true + } + const period = + discovery.paths.length === 0 ? BLOB_EMPTY_REDISCOVER_MS : BLOB_REDISCOVER_MS + return now - discovery.discoveredAt >= period + } } // Bytes on the wire per parameter, by the dtype key the Hub reports. Anything unrecognized counts diff --git a/src/test/unit/service/serviceReadiness.test.ts b/src/test/unit/service/serviceReadiness.test.ts index 614987199..e459e0086 100644 --- a/src/test/unit/service/serviceReadiness.test.ts +++ b/src/test/unit/service/serviceReadiness.test.ts @@ -14,7 +14,10 @@ import { resolveServiceEngine } from '../../../components/c2d/serviceEngines.js' import { buildModelDownload, fetchModelTotalBytes, + BLOB_EMPTY_REDISCOVER_MS, + BLOB_REDISCOVER_MS, isModelDownloadComplete, + ModelDownloadSampler, readModelDownloadBytes, selectBlobPaths } from '../../../components/c2d/modelDownload.js' @@ -452,3 +455,99 @@ describe('isModelDownloadComplete', () => { expect(isModelDownloadComplete(undefined)).to.equal(false) }) }) + +describe('ModelDownloadSampler', () => { + const repo = '/root/.cache/huggingface/hub/models--Qwen--Qwen2.5-7B' + + // A container whose cache the test mutates between samples; counts the expensive listing. + function liveContainer(id = 'c1') { + const files: Record = {} + const state = { changesCalls: 0 } + const container = { + id, + changes: () => { + state.changesCalls++ + return Promise.resolve(Object.keys(files).map((Path) => ({ Path, Kind: 1 }))) + }, + infoArchive: ({ path }: { path: string }) => { + if (!(path in files)) { + return Promise.reject(new Error('not found')) + } + const header = Buffer.from(JSON.stringify({ size: files[path] })).toString( + 'base64' + ) + return Promise.resolve({ headers: { 'x-docker-container-path-stat': header } }) + } + } as any + return { container, files, state } + } + + it('re-lists only on its own cadence, stat-ing known files in between', async () => { + const sampler = new ModelDownloadSampler() + const { container, files, state } = liveContainer() + files[`${repo}/blobs/a.1a2b3c4d.incomplete`] = 100 + + expect((await sampler.sample('svc', container, 0))?.downloadedBytes).to.equal(100) + files[`${repo}/blobs/a.1a2b3c4d.incomplete`] = 700 + // Growth of a known file is seen without listing again. + expect((await sampler.sample('svc', container, 5_000))?.downloadedBytes).to.equal(700) + expect(state.changesCalls).to.equal(1) + // A file that started on its own waits for the periodic re-list. + files[`${repo}/blobs/b.5e6f7a8b.incomplete`] = 50 + expect((await sampler.sample('svc', container, 10_000))?.downloadedBytes).to.equal( + 700 + ) + expect( + (await sampler.sample('svc', container, BLOB_REDISCOVER_MS))?.downloadedBytes + ).to.equal(750) + expect(state.changesCalls).to.equal(2) + }) + + it('re-lists right after a known download finishes', async () => { + const sampler = new ModelDownloadSampler() + const { container, files, state } = liveContainer() + files[`${repo}/blobs/a.1a2b3c4d.incomplete`] = 100 + await sampler.sample('svc', container, 0) + + // `a` finishes and the downloader starts `b`. + delete files[`${repo}/blobs/a.1a2b3c4d.incomplete`] + files[`${repo}/blobs/a`] = 1000 + files[`${repo}/blobs/b.5e6f7a8b.incomplete`] = 20 + const finished = await sampler.sample('svc', container, 5_000) + expect(finished).to.deep.equal({ downloadedBytes: 1000, files: 1, inFlight: 0 }) + expect(state.changesCalls).to.equal(1) + // The finish marked the list stale: the next sample picks `b` up well before 30s. + const next = await sampler.sample('svc', container, 10_000) + expect(next).to.deep.equal({ downloadedBytes: 1020, files: 1, inFlight: 1 }) + expect(state.changesCalls).to.equal(2) + }) + + it('polls an empty cache on the shorter cadence', async () => { + const sampler = new ModelDownloadSampler() + const { container, files, state } = liveContainer() + expect(await sampler.sample('svc', container, 0)).to.equal(null) + files[`${repo}/blobs/a.1a2b3c4d.incomplete`] = 10 + expect(await sampler.sample('svc', container, 5_000)).to.equal(null) + expect( + (await sampler.sample('svc', container, BLOB_EMPTY_REDISCOVER_MS))?.downloadedBytes + ).to.equal(10) + expect(state.changesCalls).to.equal(2) + }) + + it('starts over for a new container and forgets services that ended', async () => { + const sampler = new ModelDownloadSampler() + const first = liveContainer('c1') + first.files[`${repo}/blobs/a`] = 500 + await sampler.sample('svc', first.container, 0) + + // Restart: a new container with an empty cache must not be measured with the old paths. + const second = liveContainer('c2') + expect(await sampler.sample('svc', second.container, 1_000)).to.equal(null) + expect(second.state.changesCalls).to.equal(1) + + sampler.retain(new Set()) + first.files[`${repo}/blobs/b`] = 1 + await sampler.sample('svc', first.container, 2_000) + expect(first.state.changesCalls).to.equal(2) + }) +}) From b5779894a9b81bda2b1d1cdbf1195b6fd84c60f3 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 20:46:16 +0300 Subject: [PATCH 7/8] fix test --- src/components/c2d/compute_engine_docker.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index bd8018b4b..4f61eb3e1 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -846,8 +846,9 @@ export class C2DEngineDocker extends C2DEngine { await Promise.allSettled([...this.serviceOpPromises]) this.serviceOpPromises.clear() } - // Background readiness probes write to the same DB; let them settle too. - if (this.serviceProbesInFlight.size > 0) { + // Background readiness probes write to the same DB; let them settle too. The optional chain + // covers engines built without the constructor (e.g. test doubles). + if (this.serviceProbesInFlight?.size > 0) { await Promise.allSettled([...this.serviceProbesInFlight.values()]) this.serviceProbesInFlight.clear() } From f794919e90a40a90a285f702ca4fb983c127ba15 Mon Sep 17 00:00:00 2001 From: andreip136 <129227833+andreip136@users.noreply.github.com> Date: Wed, 23 Sep 2026 20:50:34 +0300 Subject: [PATCH 8/8] fix gemini review --- src/components/c2d/compute_engine_docker.ts | 13 ++++++++++++- src/components/c2d/serviceReadiness.ts | 6 +++++- src/test/unit/service/serviceReadiness.test.ts | 12 ++++++++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/components/c2d/compute_engine_docker.ts b/src/components/c2d/compute_engine_docker.ts index 4f61eb3e1..f28be4a97 100755 --- a/src/components/c2d/compute_engine_docker.ts +++ b/src/components/c2d/compute_engine_docker.ts @@ -4151,15 +4151,26 @@ export class C2DEngineDocker extends C2DEngine { // images). Writes are chained rather than fired in parallel, and drained BEFORE the // status moves on — an in-flight write still carrying status=PullImage must never land // after the Claiming transition and resurrect the old status. + // At most one write waits behind the running one: each write persists the shared `job`, + // which already carries the latest progress, so queuing more would only replay it. A + // slow DB then drops intermediate updates instead of growing the chain. let pullWrites: Promise = Promise.resolve() + let pullWriteQueued = false await this.pullImageRef( job.containerImage, undefined, undefined, (progress) => { job.imagePull = progress + if (pullWriteQueued) { + return + } + pullWriteQueued = true pullWrites = pullWrites - .then(() => this.db.updateServiceJob(job)) + .then(() => { + pullWriteQueued = false + return this.db.updateServiceJob(job) + }) .catch((e: any) => CORE_LOGGER.debug( `service ${serviceId}: pull progress write failed: ${e.message}` diff --git a/src/components/c2d/serviceReadiness.ts b/src/components/c2d/serviceReadiness.ts index 8704bb5cb..ac370e57b 100644 --- a/src/components/c2d/serviceReadiness.ts +++ b/src/components/c2d/serviceReadiness.ts @@ -139,7 +139,11 @@ export function probeCandidates( const hostPort = endpoint?.hostPort const urls: string[] = [] for (const ip of containerIps) { - if (ip) urls.push(`http://${ip}:${containerPort}${path}`) + if (ip) { + // Docker reports IPv4 here today; an IPv6 literal would need brackets to form a URL. + const host = ip.includes(':') ? `[${ip}]` : ip + urls.push(`http://${host}:${containerPort}${path}`) + } } if (hostPort) { urls.push(`http://127.0.0.1:${hostPort}${path}`) diff --git a/src/test/unit/service/serviceReadiness.test.ts b/src/test/unit/service/serviceReadiness.test.ts index e459e0086..cf55eabe6 100644 --- a/src/test/unit/service/serviceReadiness.test.ts +++ b/src/test/unit/service/serviceReadiness.test.ts @@ -313,6 +313,18 @@ describe('buildModelDownload', () => { }) }) +describe('probeCandidates', () => { + it('brackets an IPv6 container address so the URL parses', () => { + const job = { endpoints: [] } as any + const urls = probeCandidates(job, ['172.17.0.2', 'fd00::2'], 8080, '/health') + expect(urls).to.deep.equal([ + 'http://172.17.0.2:8080/health', + 'http://[fd00::2]:8080/health' + ]) + expect(() => new URL(urls[1])).to.not.throw() + }) +}) + describe('model download bytes from container changes', () => { const cache = '/root/.cache/huggingface/hub' const repo = `${cache}/models--Qwen--Qwen2.5-7B`