Check Comfy model download status

docs/services.md

@@ -153,6 +153,36 @@ redeploy) cannot run conflicting operations on the same service. Leases are hear
 every 30 s while the operation runs; a lease not refreshed for 2 minutes belongs to a
 crashed process and is stolen automatically, so no manual cleanup is ever needed.
 
+## Readiness and startup progress
+
+`Running` only means the container started. For workloads it recognizes, the node also reports
+whether the service can serve requests and how far its startup downloads have got, on the job
+returned by `SERVICE_GET_STATUS`:
+
+- `readiness` — `waiting`, `ready` or `failing`, from a request the node makes to the workload
+  itself: vLLM `GET /v1/models` on 8000, llama.cpp `GET /health` on 8080, ComfyUI
+  `GET /system_stats` on 8188. A profile applies only when the service publishes that port, so an
+  app that runs an engine behind its own port reports no readiness and `Running` stays the signal.
+  A service still `waiting` 15 minutes after its container started is logged once as a node warning
+  with the last probe error — usually a node that cannot reach its containers.
+- `imagePull` — byte progress of the image pull.
+- `modelDownload` — bytes downloaded against the total, and `filesComplete` / `filesTotal` when the
+  download is listed up front.
+
+`modelDownload` comes from the launch script's download manifest when it writes one: a
+tab-separated `/tmp/ocean/downloads.tsv` in the container, one `<kind> <bytes> <path> [<source>]`
+line per item, written before the first download. A `file` entry is complete once `<path>` exists
+and in flight while `<path>.part` does; a `dir` entry is a folder filled under names of its own
+(huggingface_hub's snapshot cache), measured with `du -sb`. A `bytes` of `0` means unknown. Lines may
+be appended while the script runs; at most 64 entries are read. Without a manifest, the node falls
+back to the engine's own Hugging Face cache (vLLM, llama.cpp).
+
+A service with no engine profile (an app that runs its engine behind its own port) still reports
+`modelDownload` from its manifest, with no `readiness`. With nothing to say when such a service is
+ready, the node checks it only within 30 minutes of the container starting unless a manifest has
+appeared, and stops once everything listed is complete. All of it is best-effort and owner-only: a
+missing or wrong manifest only changes what that service's own progress report says.
+
 ## Service outputs
 
 A service writes its results to `/data/outputs`. With an `outputBucketId` that folder is the

src/@types/C2D/ServiceOnDemand.ts

@@ -193,6 +193,7 @@ export interface ServiceModelDownload {
   percent?: number
   filesComplete: number
   filesInFlight: number // the hub fetches files in parallel, so only the aggregate is meaningful
+  filesTotal?: number // from the download manifest
   updatedAt: number // Unix ms
 }
 

src/components/c2d/compute_engine_docker.ts

@@ -90,7 +90,8 @@
   runReadinessProbe,
   PROBE_INITIAL_DELAY_SECONDS,
   PROBE_PERIOD_SECONDS,
-  READY_PROBE_PERIOD_SECONDS
+  READY_PROBE_PERIOD_SECONDS,
+  READINESS_WARN_AFTER_SECONDS
 } from './serviceReadiness.js'
 import { resolveServiceEngine, type ServiceEngineProfile } from './serviceEngines.js'
 import {
@@ -99,6 +100,11 @@
   isModelDownloadComplete,
   ModelDownloadSampler
 } from './modelDownload.js'
+import {
+  MANIFEST_DISCOVERY_MS,
+  MANIFEST_SAMPLING_MS,
+  sampleDownloadManifest
+} from './downloadManifest.js'
 import type { DockerMountObject } from '../../@types/PersistentStorage.js'
 import { resolveServiceImage } from './serviceResourceMatching.js'
 import {
@@ -145,6 +151,9 @@
 // doesn't get a live operation's resources torn down under it.
 const SERVICE_LOCK_HEARTBEAT_MS = 30_000
 const SERVICE_LOCK_STALE_MS = 120_000
+// `du` on a manifest folder runs in a user-controlled container, so it must not hold the probe.
+const MANIFEST_DU_TIMEOUT_MS = 15_000
+const MANIFEST_DIR_MEASURE_MS = 15_000
 
 // Identifies one engine instance as a service-lock holder across processes. pid alone is
 // not enough: pids are reused, and one process can host several engine instances.
@@ -215,6 +224,15 @@
   private serviceProbeUrls: Map<string, string> = new Map()
   // Per-service model-download sampling state (which cache files to stat, when to re-list them).
   private modelDownloadSampler = new ModelDownloadSampler()
+  // serviceId -> last manifest check, for services the node cannot probe.
+  private manifestCheckedAt: Map<string, number> = new Map()
+  // serviceId -> containerId the "still not ready" warning was logged for.
+  private readinessWarned: Map<string, string> = new Map()
+  // serviceId -> manifest folder path -> last `du`.
+  private manifestDirSizes: Map<
+    string,
+    Map<string, { containerId: string; bytes: number | null; at: number }>
+  > = 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.
@@ -3066,7 +3084,8 @@
   // never report an unmeasurable container as "0 bytes used".
   private async getContainerDiskUsage(
     containerName: string,
-    path: string = '/data'
+    path: string = '/data',
+    timeoutMs?: number
   ): Promise<number | null> {
     try {
       const container = this.docker.getContainer(containerName)
@@ -3078,22 +3097,47 @@
         return null
       }
 
+      // Without a TTY, Docker's 8-byte stream headers can read as digits prepended to the size.
       const exec = await container.exec({
         Cmd: ['du', '-sb', path],
         AttachStdout: true,
-        AttachStderr: true
+        AttachStderr: true,
+        Tty: true
       })
 
-      const stream = await exec.start({ Detach: false, Tty: false })
+      const stream = await exec.start({ Detach: false, Tty: true })
 
+      let timedOut = false
+      const timer =
+        timeoutMs !== undefined
+          ? setTimeout(() => {
+              timedOut = true
+              stream.destroy()
+            }, timeoutMs)
+          : undefined
       const chunks: Buffer[] = []
-      for await (const chunk of stream) {
-        chunks.push(chunk as Buffer)
+      try {
+        for await (const chunk of stream) {
+          chunks.push(chunk as Buffer)
+        }
+      } catch (error) {
+        if (!timedOut) {
+          throw error
+        }
+      } finally {
+        clearTimeout(timer)
+      }
+      if (timedOut) {
+        CORE_LOGGER.warn(
+          `du ${path} in ${containerName} did not finish within ${timeoutMs / 1000}s`
+        )
+        return null
       }
 
       const output = Buffer.concat(chunks).toString()
 
-      const match = output.match(/(\d+)\s/)
+      // The size line follows any "du: cannot access" lines, which the TTY merges in.
+      const match = output.match(/^(\d+)\t/m)
       return match ? parseInt(match[1], 10) : null
     } catch (error) {
       CORE_LOGGER.error(
@@ -4027,7 +4071,7 @@
       let empty = false
       await pipeline(
         tarToZip(tar, (stats) => {
           skipped = stats.skipped
           empty = stats.files === 0 && stats.directories === 0
         }),
         createWriteStream(partialPath)
@@ -4639,9 +4683,16 @@
     // 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)
+    for (const state of [
+      this.serviceProbeUrls,
+      this.manifestCheckedAt,
+      this.readinessWarned,
+      this.manifestDirSizes
+    ]) {
+      for (const serviceId of state.keys()) {
+        if (!running.has(serviceId)) {
+          state.delete(serviceId)
+        }
       }
     }
     return runningOnly
@@ -4716,7 +4767,10 @@
   ): Promise<void> {
     try {
       const engine = resolveServiceEngine(job)
-      if (!engine) return
+      if (!engine) {
+        await this.sampleUnprobedServiceDownload(job, details)
+        return
+      }
       if (this.serviceOpsInFlight.has(job.serviceId)) return
 
       const now = Date.now()
@@ -4819,33 +4873,124 @@
         ? undefined
         : await this.sampleModelDownload(job, engine)
 
-      // Same cross-process guard as the metrics write: a lifecycle transition must win.
+      // Usually a node that cannot reach its container: tell the operator once per container.
       if (
-        this.serviceOpsInFlight.has(job.serviceId) ||
-        (await this.db.isServiceLocked(job.serviceId, SERVICE_LOCK_STALE_MS))
+        readiness.state === 'waiting' &&
+        Number.isFinite(startedAt) &&
+        now - startedAt >= READINESS_WARN_AFTER_SECONDS * 1000 &&
+        this.readinessWarned.get(job.serviceId) !== job.containerId
       ) {
-        return
+        this.readinessWarned.set(job.serviceId, job.containerId)
+        const download = modelDownload ?? job.modelDownload
+        CORE_LOGGER.warn(
+          `[readiness] service ${job.serviceId} (${engine.id}) not ready ` +
+            `${Math.round((now - startedAt) / 60_000)} min after its container started: ` +
+            `${readiness.lastError ?? `HTTP ${readiness.httpStatus}`} via ${readiness.probedUrl}` +
+            (download ? `; model download at ${download.percent ?? '?'}%` : '')
+        )
       }
-      await this.db.updateServiceJobReadiness(
-        job.serviceId,
-        {
-          owner: job.owner,
-          clusterHash: job.clusterHash,
-          status: ServiceStatusNumber.Running,
-          containerId: job.containerId
-        },
-        readiness,
-        modelDownload
-      )
+
+      await this.persistServiceReadiness(job, readiness, modelDownload)
     } catch (e: any) {
       CORE_LOGGER.debug(
         `[readiness] service ${job.serviceId}: probe failed: ${e?.message}`
       )
     }
   }
 
+  // Same cross-process guard as the metrics write: a lifecycle transition must win.
+  private async persistServiceReadiness(
+    job: ServiceJob,
+    readiness: ServiceReadiness | undefined,
+    modelDownload: ServiceModelDownload | undefined
+  ): Promise<void> {
+    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
+    )
+  }
+
   /**
-   * How much of its model the container has pulled down, read from the engine's own cache.
+   * Manifest download progress for a service the node cannot probe; no readiness, so Running stays
+   * the signal. With no readiness to end it, sampling is bounded by MANIFEST_DISCOVERY_MS /
+   * MANIFEST_SAMPLING_MS and stops once the listed download is complete.
+   */
+  private async sampleUnprobedServiceDownload(
+    job: ServiceJob,
+    details: Dockerode.ContainerInspectInfo
+  ): Promise<void> {
+    if (
+      this.serviceOpsInFlight.has(job.serviceId) ||
+      isModelDownloadComplete(job.modelDownload)
+    ) {
+      return
+    }
+    const now = Date.now()
+    const startedAt = Date.parse(details.State?.StartedAt ?? '')
+    if (
+      Number.isFinite(startedAt) &&
+      now - startedAt > (job.modelDownload ? MANIFEST_SAMPLING_MS : MANIFEST_DISCOVERY_MS)
+    ) {
+      return
+    }
+    if (
+      now - (this.manifestCheckedAt.get(job.serviceId) ?? 0) <
+      PROBE_PERIOD_SECONDS * 1000
+    ) {
+      return
+    }
+    this.manifestCheckedAt.set(job.serviceId, now)
+    const modelDownload = await sampleDownloadManifest(
+      this.docker.getContainer(job.containerId),
+      (path) => this.manifestDirectoryBytes(job, path)
+    )
+    if (modelDownload) {
+      await this.persistServiceReadiness(job, undefined, modelDownload)
+    }
+  }
+
+  private async manifestDirectoryBytes(
+    job: ServiceJob,
+    path: string
+  ): Promise<number | null> {
+    let sizes = this.manifestDirSizes.get(job.serviceId)
+    if (!sizes) {
+      sizes = new Map()
+      this.manifestDirSizes.set(job.serviceId, sizes)
+    }
+    const now = Date.now()
+    const last = sizes.get(path)
+    if (
+      last?.containerId === job.containerId &&
+      now - last.at < MANIFEST_DIR_MEASURE_MS
+    ) {
+      return last.bytes
+    }
+    const bytes = await this.getContainerDiskUsage(
+      job.containerId,
+      path,
+      MANIFEST_DU_TIMEOUT_MS
+    )
+    sizes.set(path, { containerId: job.containerId, bytes, at: now })
+    return bytes
+  }
+
+  /**
+   * How much of its model the container has pulled down: from the launch script's download
+   * manifest when it writes one, otherwise 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
@@ -4859,14 +5004,23 @@
     job: ServiceJob,
     engine: ServiceEngineProfile
   ): Promise<ServiceModelDownload | undefined> {
-    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)
+      // Before the completion check: a script can append to its manifest after earlier downloads.
+      const fromManifest = await sampleDownloadManifest(container, (path) =>
+        this.manifestDirectoryBytes(job, path)
+      )
+      if (fromManifest) {
+        return fromManifest
+      }
+      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
+      }
       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