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 // items the service will download, when its download manifest lists them
   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,13 @@
 // 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
+// How long the node waits for `du` on a service's manifest folder. The container is user-controlled
+// and the measurement is only progress reporting, so it must not hold the probe or the node's
+// shutdown that drains it. Compute disk-quota checks keep waiting for the answer.
+const MANIFEST_DU_TIMEOUT_MS = 15_000
+// A manifest folder is measured with `du` at most this often, while the manifest itself is re-read
+// on every sample to catch appended entries.
+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 +228,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 -> when a service the node cannot probe was last checked for a download manifest.
+  private manifestCheckedAt: Map<string, number> = new Map()
+  // serviceId -> the container a "still not ready" warning was logged for (once per container).
+  private readinessWarned: Map<string, string> = new Map()
+  // serviceId -> the last `du` of each manifest folder in its current container.
+  private manifestDirSizes: Map<
+    string,
+    { containerId: string; sizes: Map<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 +3088,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 +3101,49 @@
         return null
       }
 
+      // With a TTY the output arrives as plain text. Without one, Docker frames it with 8-byte
+      // headers whose length byte can read as an ASCII digit and get 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 })
 
       const chunks: Buffer[] = []
-      for await (const chunk of stream) {
-        chunks.push(chunk as Buffer)
+      const read = (async () => {
+        for await (const chunk of stream) {
+          chunks.push(chunk as Buffer)
+        }
+        return true
+      })()
+      // Destroying the stream on timeout rejects the read after the race has moved on.
+      read.catch(() => {})
+      let timer: NodeJS.Timeout
+      const timedOut = new Promise<false>((resolve) => {
+        if (timeoutMs !== undefined) {
+          timer = setTimeout(() => resolve(false), timeoutMs)
+        }
+      })
+      try {
+        if (!(await Promise.race([read, timedOut]))) {
+          stream.destroy()
+          CORE_LOGGER.warn(
+            `du ${path} in ${containerName} did not finish within ${timeoutMs / 1000}s`
+          )
+          return null
+        }
+      } finally {
+        clearTimeout(timer)
       }
 
       const output = Buffer.concat(chunks).toString()
 
-      const match = output.match(/(\d+)\s/)
+      // `du -sb` prints "<bytes>\t<path>", after any "du: cannot access" lines (e.g. for /proc),
+      // which the TTY merges into the same output.
+      const match = output.match(/^(\d+)\t/m)
       return match ? parseInt(match[1], 10) : null
     } catch (error) {
       CORE_LOGGER.error(
@@ -4027,7 +4077,7 @@
       let empty = false
       await pipeline(
         tarToZip(tar, (stats) => {
           skipped = stats.skipped
           empty = stats.files === 0 && stats.directories === 0
         }),
         createWriteStream(partialPath)
@@ -4639,9 +4689,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 +4773,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,6 +4879,24 @@
         ? undefined
         : await this.sampleModelDownload(job, engine)
 
+      // A service that never answers is usually a node that cannot reach its container: say so once,
+      // where an operator looks, rather than only through users stuck on "warming up".
+      if (
+        readiness.state === 'waiting' &&
+        Number.isFinite(startedAt) &&
+        now - startedAt >= READINESS_WARN_AFTER_SECONDS * 1000 &&
+        this.readinessWarned.get(job.serviceId) !== job.containerId
+      ) {
+        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 ?? '?'}%` : '')
+        )
+      }
+
       // Same cross-process guard as the metrics write: a lifecycle transition must win.
       if (
         this.serviceOpsInFlight.has(job.serviceId) ||
@@ -4845,7 +4923,93 @@
   }
 
   /**
-   * How much of its model the container has pulled down, read from the engine's own cache.
+   * Download progress for a service the node cannot probe: only what the launch script's download
+   * manifest says, with no readiness, so Running stays the signal that the service is usable.
+   *
+   * Throttled like the probe. A service with no manifest MANIFEST_DISCOVERY_MS after its container
+   * started is not checked again, and sampling stops once the listed download is complete or after
+   * MANIFEST_SAMPLING_MS, whichever comes first — with no
+   * readiness to end it, it would otherwise run for the whole session.
+   */
+  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) {
+      return
+    }
+    // Same cross-process guard as the readiness 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
+      },
+      undefined,
+      modelDownload
+    )
+  }
+
+  /** A manifest folder's size, re-measured with `du` at most every MANIFEST_DIR_MEASURE_MS. */
+  private async manifestDirectoryBytes(
+    job: ServiceJob,
+    path: string
+  ): Promise<number | null> {
+    let cache = this.manifestDirSizes.get(job.serviceId)
+    if (!cache || cache.containerId !== job.containerId) {
+      cache = { containerId: job.containerId, sizes: new Map() }
+      this.manifestDirSizes.set(job.serviceId, cache)
+    }
+    const now = Date.now()
+    const last = cache.sizes.get(path)
+    if (last && now - last.at < MANIFEST_DIR_MEASURE_MS) {
+      return last.bytes
+    }
+    const bytes = await this.getContainerDiskUsage(
+      job.containerId,
+      path,
+      MANIFEST_DU_TIMEOUT_MS
+    )
+    cache.sizes.set(path, { 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 +5023,24 @@
     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)
+      // The launch script's own manifest, when it writes one (see downloadManifest). Read before
+      // the completion check below: a script can append to it after its first downloads finish.
+      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