diff --git a/packages/adapters/src/media-ingest-analyze.test.ts b/packages/adapters/src/media-ingest-analyze.test.ts new file mode 100644 index 0000000..d551dde --- /dev/null +++ b/packages/adapters/src/media-ingest-analyze.test.ts @@ -0,0 +1,154 @@ +import type { MediaCapabilities } from "@rakazo/contracts"; +import { defaultTeaserWeights } from "@rakazo/media"; +import { describe, expect, it } from "vitest"; +import type { AnalysisBundle } from "./media-ingest-analyze.js"; +import { + analysisModelVersion, + buildSegments, + isTransientInferenceError, + keyframeTimes, +} from "./media-ingest-analyze.js"; +import { MediaSidecarError } from "./media-sidecar.js"; + +const empty: AnalysisBundle = { + scenes: null, + transcript: null, + faceSpans: null, + embeddings: null, + audio: null, + saliency: null, +}; +const weights = defaultTeaserWeights("teaser"); + +describe("keyframeTimes", () => { + it("samples once per second, a still once, and spreads long clips over 60 frames", () => { + expect(keyframeTimes(null, true)).toEqual([0]); + expect(keyframeTimes(800, false)).toEqual([0]); + expect(keyframeTimes(4000, false)).toEqual([0, 1000, 2000, 3000]); + const long = keyframeTimes(600_000, false); + expect(long).toHaveLength(60); + expect(long[1]).toBe(10_000); + }); +}); + +describe("buildSegments", () => { + it("uses detected shots and scores each from the analyses that exist", () => { + const unit = (x: number) => Float32Array.from([x, Math.sqrt(1 - x * x)]); + const segments = buildSegments({ + durationMs: 4000, + still: false, + width: 1920, + height: 1080, + weights, + bundle: { + ...empty, + scenes: { + shots: [ + { startMs: 0, endMs: 2000, score: 1 }, + { startMs: 2000, endMs: 4000, score: 1 }, + ], + scenes: [{ startMs: 0, endMs: 4000 }], + }, + transcript: { + language: "en", + words: [ + { startMs: 2100, endMs: 2400, text: "Wait", confidence: 0.9 }, + { startMs: 2500, endMs: 2800, text: "what?", confidence: 0.9 }, + ], + }, + faceSpans: [{ startMs: 0, endMs: 1000 }], + embeddings: { + timesMs: [0, 1000, 2000, 3000], + vectors: [unit(1), unit(1), unit(0), unit(0.1)], + }, + audio: { + beatsMs: [0, 2100], + energy: [ + { tMs: 500, value: 0.2 }, + { tMs: 2500, value: 0.8 }, + ], + }, + saliency: null, + }, + }); + expect(segments.map((segment) => [segment.kind, segment.startMs, segment.endMs])).toEqual([ + ["shot", 0, 2000], + ["shot", 2000, 4000], + ]); + const [first, second] = segments; + expect(first?.scores?.faces).toBe(0.5); + expect(second?.scores?.faces).toBe(0); + expect(first?.scores?.sharpness).toBe(1); + expect(second?.scores?.audioPeak).toBe(1); + expect(first?.scores?.audioPeak).toBe(0.25); + expect(first?.scores?.beatProximity).toBe(1); + expect(second?.transcriptExcerpt).toBe("Wait what?"); + expect(first?.transcriptExcerpt).toBeNull(); + expect(first?.scores?.novelty).toBe(0.5); + expect(second?.scores?.novelty).toBeGreaterThan(0.5); + expect(first?.scores?.aesthetics).toBeUndefined(); + for (const segment of segments) { + expect(segment.teaserScore).toBeGreaterThanOrEqual(0); + expect(segment.teaserScore).toBeLessThanOrEqual(1); + } + }); + + it("falls back to 3 s windows without shots and to one span for a still", () => { + const windows = buildSegments({ + durationMs: 7000, + still: false, + width: null, + height: null, + weights, + bundle: empty, + }); + expect(windows.map((segment) => [segment.kind, segment.startMs, segment.endMs])).toEqual([ + ["scene", 0, 3000], + ["scene", 3000, 6000], + ["scene", 6000, 7000], + ]); + expect(windows[0]?.scores).toEqual({}); + const still = buildSegments({ + durationMs: null, + still: true, + width: 640, + height: 480, + weights, + bundle: { ...empty, faceSpans: [{ startMs: 0, endMs: 1 }] }, + }); + expect(still).toHaveLength(1); + expect(still[0]?.scores?.faces).toBe(1); + }); +}); + +describe("analysis versions", () => { + it("derives the version from the provider id and the operation's model", () => { + const capabilities = { + configured: true, + accelerator: "cpu", + hardwareEncoders: [], + operations: { + embedFrames: { + available: true, + experimental: false, + model: "siglip-b16", + license: "Apache-2.0", + commercialUse: true, + }, + }, + } as unknown as MediaCapabilities; + expect(analysisModelVersion("sidecar", capabilities, "embeddings")).toBe("sidecar:siglip-b16"); + expect(analysisModelVersion("sidecar", capabilities, "scenes")).toBe("sidecar:default"); + expect(analysisModelVersion("emulator", null, "faces")).toBe("emulator:default"); + }); + + it("treats an unreachable or saturated sidecar as transient", () => { + const error = (code: "unreachable" | "busy" | "timeout" | "internal") => + new MediaSidecarError(code, "x", null, "detectScenes"); + expect(isTransientInferenceError(error("unreachable"))).toBe(true); + expect(isTransientInferenceError(error("busy"))).toBe(true); + expect(isTransientInferenceError(error("timeout"))).toBe(true); + expect(isTransientInferenceError(error("internal"))).toBe(false); + expect(isTransientInferenceError(new Error("unreachable"))).toBe(false); + }); +}); diff --git a/packages/adapters/src/media-ingest-analyze.ts b/packages/adapters/src/media-ingest-analyze.ts index a2fb404..f54902b 100644 --- a/packages/adapters/src/media-ingest-analyze.ts +++ b/packages/adapters/src/media-ingest-analyze.ts @@ -34,6 +34,7 @@ import { assertInside, resolveInside } from "@rakazo/media/node"; import { z } from "zod"; import type { MediaServiceDeps } from "./media-context.js"; import { assetDir, writeFileAtomic } from "./media-ingest-files.js"; +import { MediaSidecarError } from "./media-sidecar.js"; /* * Sidecar analyses for one asset and the scored segments built from them. @@ -54,7 +55,7 @@ const KEYFRAME_LIMIT = 60; const WINDOW_MS = 3000; const STILL_SEGMENT_MS = 3000; const FACE_SAMPLE_FPS = 2; -const CLASSIFIER_SAMPLE_FPS = 1; +const CLASSIFIER_SAMPLE_FPS = 2; const SAME_FACE_SIMILARITY = 0.6; const EXCERPT_LIMIT = 500; @@ -155,10 +156,6 @@ function inSpan(tMs: number, span: { startMs: number; endMs: number }): boolean return tMs >= span.startMs && tMs < span.endMs; } -function overlapMs(a: { startMs: number; endMs: number }, b: { startMs: number; endMs: number }) { - return Math.max(0, Math.min(a.endMs, b.endMs) - Math.max(a.startMs, b.startMs)); -} - /** Union length of `spans` clipped to `window`. */ function coveredMs( spans: Array<{ startMs: number; endMs: number }>, @@ -269,7 +266,11 @@ export function buildSegments(input: { const words = bundle.transcript.words.filter((word) => inSpan((word.startMs + word.endMs) / 2, span), ); - excerpt = words.map((word) => word.text.trim()).filter(Boolean).join(" ") || null; + excerpt = + words + .map((word) => word.text.trim()) + .filter(Boolean) + .join(" ") || null; scores.speechHook = excerpt ? round3(speechHookScore(excerpt, language)) : 0; } const mean = frames.length > 0 ? meanVector(frames) : null; @@ -281,11 +282,15 @@ export function buildSegments(input: { previousMean = mean; } if (bundle.audio && bundle.audio.beatsMs.length > 0 && !input.still) { - const distance = Math.min(...bundle.audio.beatsMs.map((beat) => Math.abs(beat - span.startMs))); + const distance = Math.min( + ...bundle.audio.beatsMs.map((beat) => Math.abs(beat - span.startMs)), + ); scores.beatProximity = round3(1 - distance / 500); } if (bundle.saliency && bundle.saliency.points.length > 0) { - const inside = bundle.saliency.points.filter((point) => input.still || inSpan(point.tMs, span)); + const inside = bundle.saliency.points.filter( + (point) => input.still || inSpan(point.tMs, span), + ); const points = inside.length > 0 ? inside : bundle.saliency.points; scores.aesthetics = round3( points.reduce((sum, point) => sum + point.strength, 0) / points.length, @@ -309,9 +314,17 @@ export function buildSegments(input: { type AssetRow = Awaited>; -function modelOf(capabilities: MediaCapabilities | null, operation: MediaInferenceOperation) { - const entry = capabilities?.operations[operation]; - return { available: entry?.available !== false, model: entry?.model ?? null }; +/** Sidecar errors that say nothing about the asset (down, saturated, timed out). */ +export function isTransientInferenceError(error: unknown): boolean { + return ( + error instanceof MediaSidecarError && + (error.code === "unreachable" || error.code === "busy" || error.code === "timeout") + ); +} + +/** JSON-safe copy of a provider result for a Prisma Json column. */ +function json(value: unknown): Prisma.InputJsonValue { + return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue; } /** Run `fn`, or return null when the provider does not support the operation. */ @@ -328,6 +341,20 @@ function safeName(value: string): string { return value.replace(/[^A-Za-z0-9_-]/g, "_").slice(0, 64) || "track"; } +/** + * Model version of an analysis kind: ":" from the + * provider's capabilities, so a model change writes new rows instead of + * overwriting, and the skip check sees the same key the write uses. + */ +export function analysisModelVersion( + providerId: string, + capabilities: MediaCapabilities | null, + kind: MediaAnalysisKind, +): string { + const model = capabilities?.operations[ANALYSIS_OPERATION[kind]]?.model ?? "default"; + return `${providerId}:${model}`; +} + /** * Re-analysis keeps the person a track was assigned to when the new track of the * same asset carries the same identity; tracks of other assets are never matched here. @@ -385,14 +412,18 @@ async function storedBundle( export interface AnalyzeOutcome { ran: MediaAnalysisKind[]; + /** Requested but unsupported by the provider or not applicable to a still. */ skipped: MediaAnalysisKind[]; + /** Already stored at the current model version, so not run again. */ + reused: MediaAnalysisKind[]; segments: number; qualityScore: number | null; } /** - * Run the requested analyses (all when omitted) on `inputPath`, store them, and - * rebuild the asset's segments. The caller owns ingest state transitions. + * Run analyses on `inputPath`, store them, and rebuild the asset's segments. + * Explicit `kinds` always run; without them only kinds that have no result at + * the provider's current model version run. The caller owns ingest state. */ export async function analyzeAsset(input: { deps: MediaServiceDeps; @@ -401,77 +432,101 @@ export async function analyzeAsset(input: { asset: AssetRow; inputPath: string; kinds?: MediaAnalysisKind[]; + signal?: AbortSignal; }): Promise { - const { deps, inference, spaceId, asset, inputPath } = input; + const { deps, inference, spaceId, asset, inputPath, signal } = input; const still = asset.kind === "photo" || !asset.durationMs; - const requested = new Set(input.kinds ?? MediaAnalysisKindSchema.options); - const capabilities = await inference.capabilities().catch(() => null); + const capabilities = await inference.capabilities(signal).catch((error: unknown) => { + if (signal?.aborted || isTransientInferenceError(error)) throw error; + return null; + }); const times = keyframeTimes(asset.durationMs, still); const bundle: AnalysisBundle = { ...EMPTY_BUNDLE }; const ran: MediaAnalysisKind[] = []; const skipped: MediaAnalysisKind[] = []; + const reused: MediaAnalysisKind[] = []; + const request = { path: inputPath, ...(signal ? { signal } : {}) }; + + const versionOf = (kind: MediaAnalysisKind) => + analysisModelVersion(inference.id, capabilities, kind); + + let requested: Set; + if (input.kinds) { + requested = new Set(input.kinds); + } else { + const existing = await deps.prisma.mediaAnalysis.findMany({ + where: { spaceId, assetId: asset.id }, + select: { kind: true, modelVersion: true }, + }); + const done = new Set(existing.map((row) => `${row.kind}\n${row.modelVersion}`)); + requested = new Set(); + for (const kind of MediaAnalysisKindSchema.options) { + if (done.has(`${kind}\n${versionOf(kind)}`)) reused.push(kind); + else requested.add(kind); + } + } - const version = (kind: MediaAnalysisKind, resultModel?: string) => { - const model = - modelOf(capabilities, ANALYSIS_OPERATION[kind]).model ?? resultModel ?? "default"; - return { modelId: model, modelVersion: `${inference.id}:${model}` }; - }; const store = async ( kind: MediaAnalysisKind, data: Prisma.InputJsonValue, extra: { vectors?: Uint8Array; resultModel?: string } = {}, ) => { + const modelVersion = versionOf(kind); await deps.repos.upsertAnalysis(spaceId, asset.id, { kind, - ...version(kind, extra.resultModel), + modelId: extra.resultModel ?? modelVersion.slice(inference.id.length + 1), + modelVersion, data, vectors: extra.vectors ?? null, }); ran.push(kind); }; const runnable = (kind: MediaAnalysisKind) => { - if (!requested.has(kind) || (still && VIDEO_ONLY.has(kind))) return false; - if (!modelOf(capabilities, ANALYSIS_OPERATION[kind]).available) { + if (!requested.has(kind)) return false; + const unavailable = capabilities?.operations[ANALYSIS_OPERATION[kind]]?.available === false; + if ((still && VIDEO_ONLY.has(kind)) || unavailable) { skipped.push(kind); return false; } return true; }; const attempt = async (kind: MediaAnalysisKind, fn: () => Promise) => { + signal?.throwIfAborted(); const result = await unlessUnsupported(fn); if (result === null) skipped.push(kind); return result; }; if (runnable("scenes")) { - const scenes = await attempt("scenes", () => inference.detectScenes({ path: inputPath })); + const scenes = await attempt("scenes", () => inference.detectScenes(request)); if (scenes) { bundle.scenes = scenes; - await store("scenes", { shots: scenes.shots, scenes: scenes.scenes }); + await store("scenes", json({ shots: scenes.shots, scenes: scenes.scenes })); } } if (runnable("transcript")) { - const transcript = await attempt("transcript", () => inference.transcribe({ path: inputPath })); + const transcript = await attempt("transcript", () => inference.transcribe(request)); if (transcript) { bundle.transcript = transcript; const words: TranscriptWord[] = transcript.words; - await store("transcript", { - language: transcript.language, - words: words.map((word) => ({ ...word })), - text: words - .map((word) => word.text.trim()) - .filter(Boolean) - .join(" "), - }); + await store( + "transcript", + json({ + language: transcript.language, + words, + text: words + .map((word) => word.text.trim()) + .filter(Boolean) + .join(" "), + }), + ); } } if (runnable("faces")) { const sampleFps = still ? 1 : FACE_SAMPLE_FPS; - const tracked = await attempt("faces", () => - inference.trackFaces({ path: inputPath, sampleFps }), - ); + const tracked = await attempt("faces", () => inference.trackFaces({ ...request, sampleFps })); if (tracked) { const facesDir = `${assetDir(spaceId, asset.id)}/faces`; await rm(await resolveInside(deps.mediaRoot, facesDir), { recursive: true, force: true }); @@ -515,17 +570,19 @@ export async function analyzeAsset(input: { }); spans.push({ startMs, endMs, keyframeCount: keyframes.length }); } - const { modelVersion } = version("faces"); - await deps.repos.replaceFaceTracks(spaceId, asset.id, modelVersion, tracks); + await deps.repos.replaceFaceTracks(spaceId, asset.id, versionOf("faces"), tracks); bundle.faceSpans = spans; // Spans and counts only: identity embeddings never enter an analysis row. - await store("faces", { sampleFps: tracked.sampleFps, trackCount: spans.length, tracks: spans }); + await store( + "faces", + json({ sampleFps: tracked.sampleFps, trackCount: spans.length, tracks: spans }), + ); } } if (runnable("embeddings")) { const embedded = await attempt("embeddings", () => - inference.embedFrames({ path: inputPath, timesMs: times }), + inference.embedFrames({ ...request, timesMs: times }), ); if (embedded && embedded.vectors.length > 0) { const timesMs = times.slice(0, embedded.vectors.length); @@ -533,47 +590,49 @@ export async function analyzeAsset(input: { timesMs, vectors: embedded.vectors.map((vector) => Float32Array.from(vector)), }; - await store( - "embeddings", - { model: embedded.model, dim: embedded.dim, timesMs }, - { vectors: encodeVectors(embedded.vectors), resultModel: embedded.model }, - ); + await store("embeddings", json({ model: embedded.model, dim: embedded.dim, timesMs }), { + vectors: encodeVectors(embedded.vectors), + resultModel: embedded.model, + }); } } if (runnable("classifier")) { - const boundaries = (bundle.scenes?.shots ?? []).map((shot) => shot.startMs).filter((t) => t > 0); + // Every shot boundary on top of the regular samples. + const boundaries = [ + ...new Set( + (bundle.scenes?.shots ?? []).flatMap((shot) => [shot.startMs, Math.max(0, shot.endMs - 1)]), + ), + ]; const classified = await attempt("classifier", () => inference.classifyContent({ - path: inputPath, + ...request, sampleFps: CLASSIFIER_SAMPLE_FPS, ...(boundaries.length > 0 ? { extraTimesMs: boundaries } : {}), }), ); if (classified) { - await store( - "classifier", - { model: classified.model, frames: classified.frames }, - { resultModel: classified.model }, - ); + await store("classifier", json({ model: classified.model, frames: classified.frames }), { + resultModel: classified.model, + }); } } if (runnable("audio")) { - const audio = await attempt("audio", () => inference.audioFeatures({ path: inputPath })); + const audio = await attempt("audio", () => inference.audioFeatures(request)); if (audio) { bundle.audio = audio; - await store("audio", { ...audio }); + await store("audio", json(audio)); } } if (runnable("saliency")) { const saliency = await attempt("saliency", () => - inference.saliency({ path: inputPath, timesMs: times }), + inference.saliency({ ...request, timesMs: times }), ); if (saliency) { bundle.saliency = saliency; - await store("saliency", { points: saliency.points }); + await store("saliency", json({ points: saliency.points })); } } @@ -581,7 +640,7 @@ export async function analyzeAsset(input: { const missing = MediaAnalysisKindSchema.options.filter( (kind) => !ran.includes(kind) && kind !== "classifier", ); - if (input.kinds && missing.length > 0) { + if (missing.length > 0) { const stored = await storedBundle(deps, spaceId, asset.id, missing); bundle.scenes ??= stored.scenes; bundle.transcript ??= stored.transcript; @@ -609,7 +668,7 @@ export async function analyzeAsset(input: { typeof segment.teaserScore === "number" ? [segment.teaserScore] : [], ); const qualityScore = teaserScores.length > 0 ? Math.max(...teaserScores) : null; - return { ran, skipped, segments: segments.length, qualityScore }; + return { ran, skipped, reused, segments: segments.length, qualityScore }; } /** diff --git a/packages/adapters/src/media-ingest-cluster.test.ts b/packages/adapters/src/media-ingest-cluster.test.ts new file mode 100644 index 0000000..4a30113 --- /dev/null +++ b/packages/adapters/src/media-ingest-cluster.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "vitest"; +import { reconcileUserEdits } from "./media-ingest-cluster.js"; + +const computed = [ + { key: "session:a1", assetIds: ["a1", "a2", "a3"] }, + { key: "session:b1", assetIds: ["b1", "b2"] }, + { key: "session:c1", assetIds: ["c1", "c2", "c3"] }, +]; + +const keys = (clusters: Array<{ key: string; assetIds: string[] }>) => + clusters.map((cluster) => [cluster.key, cluster.assetIds]); + +describe("reconcileUserEdits", () => { + it("passes computed clusters through when the user edited nothing", () => { + expect(reconcileUserEdits(computed, [], "session")).toEqual(computed); + }); + + it("never re-adds the assets of a merged collection to computed clusters", () => { + // The user merged a1.. and b1.. into the a1 collection. + const merged = { clusterKey: "session:a1", assetIds: ["a1", "a2", "a3", "b1", "b2"] }; + expect(keys(reconcileUserEdits(computed, [merged], "session"))).toEqual([ + ["session:c1", ["c1", "c2", "c3"]], + ]); + }); + + it("keeps both halves of a split and places new assets outside them", () => { + // Split c1 | c2 c3; a later sync adds c4 to the computed session. + const grown = [ + ...computed.slice(0, 2), + { key: "session:c1", assetIds: ["c1", "c2", "c3", "c4"] }, + ]; + const kept = { clusterKey: "session:c1", assetIds: ["c1"] }; + const splitOff = { clusterKey: null, assetIds: ["c2", "c3"] }; + expect(keys(reconcileUserEdits(grown, [kept, splitOff], "session"))).toEqual([ + ["session:a1", ["a1", "a2", "a3"]], + ["session:b1", ["b1", "b2"]], + ["session:c1", ["c1", "c4"]], + ]); + // When the kept half is not a subset any more it freezes; c4 gets its own key. + const moved = [...computed.slice(0, 2), { key: "session:c0", assetIds: ["c0", "c4"] }]; + expect( + keys(reconcileUserEdits(moved, [{ ...kept, assetIds: ["c1", "c9"] }], "session")), + ).toEqual([ + ["session:a1", ["a1", "a2", "a3"]], + ["session:b1", ["b1", "b2"]], + ["session:c0", ["c0", "c4"]], + ]); + }); + + it("lets a renamed collection follow its cluster, even when the cluster key changes", () => { + const renamed = { clusterKey: "session:a1", assetIds: ["a1", "a2"] }; + const earlier = [ + { key: "session:a0", assetIds: ["a0", "a1", "a2", "a3"] }, + ...computed.slice(1), + ]; + expect(keys(reconcileUserEdits(earlier, [renamed], "session"))).toEqual([ + ["session:a1", ["a0", "a1", "a2", "a3"]], + ["session:b1", ["b1", "b2"]], + ["session:c1", ["c1", "c2", "c3"]], + ]); + }); + + it("re-keys a cluster that lost its first asset to a split-off collection", () => { + const splitOff = { clusterKey: null, assetIds: ["a1"] }; + expect(keys(reconcileUserEdits(computed, [splitOff], "theme"))).toEqual([ + ["theme:a2", ["a2", "a3"]], + ["session:b1", ["b1", "b2"]], + ["session:c1", ["c1", "c2", "c3"]], + ]); + }); + + it("ignores empty user collections", () => { + expect( + reconcileUserEdits(computed, [{ clusterKey: "session:a1", assetIds: [] }], "session"), + ).toEqual(computed); + }); +}); diff --git a/packages/adapters/src/media-ingest-cluster.ts b/packages/adapters/src/media-ingest-cluster.ts index 3bd141b..83b0811 100644 --- a/packages/adapters/src/media-ingest-cluster.ts +++ b/packages/adapters/src/media-ingest-cluster.ts @@ -1,7 +1,8 @@ import { readFile } from "node:fs/promises"; import type { AgentInputImage } from "@rakazo/adapter-kit"; import type { MediaClusterInput } from "@rakazo/db"; -import type { ClusterAsset, VectorItem } from "@rakazo/media"; +import { getLogger } from "@rakazo/logging"; +import type { ClusterAsset, SessionOptions, VectorItem } from "@rakazo/media"; import { buildSessions, clusterFaces, @@ -13,7 +14,6 @@ import { ruleLabel, } from "@rakazo/media"; import { resolveInside } from "@rakazo/media/node"; -import { getLogger } from "@rakazo/logging"; import { z } from "zod"; import type { MediaServiceDeps } from "./media-context.js"; import { errorMessage, publishSafe } from "./media-ingest-files.js"; @@ -32,11 +32,8 @@ import { errorMessage, publishSafe } from "./media-ingest-files.js"; * thumbnails and nothing else. Face embeddings and person data never leave * this process. * - * Collections the user edited (labelSource "user") keep their membership when - * it no longer matches the computed cluster (merged or split): their assets are - * withheld from every other computed cluster so re-clustering never re-adds - * them elsewhere. A collection whose stored assets are still a subset of the - * computed cluster (only renamed) keeps updating. + * Collections the user merged or split keep their membership; re-clustering + * never re-adds their assets elsewhere (see reconcileUserEdits). */ export const THEME_BATCH_LIMIT = 2000; @@ -61,9 +58,17 @@ interface UserCollection { } /** - * Apply the user-edit rules above to computed clusters of one kind. Clusters - * that lose assets are re-keyed by their first remaining asset (the same rule - * the clustering uses), and dropped when empty or colliding with a kept key. + * Apply the user's edits to computed clusters of one kind: + * - A user collection without a cluster key (split off) keeps its assets; they + * are withheld from every computed cluster. + * - A user collection whose assets all still fall into one computed cluster + * (renamed only) follows that cluster: the cluster takes the collection's key + * so the collection keeps updating, also when an earlier asset changes the + * cluster's own key. + * - Any other user collection (merged, or split and kept) is frozen: its key + * and assets are withheld, so re-clustering never re-adds them elsewhere. + * Clusters that lose their first asset are re-keyed by the first remaining one + * (the clustering's own rule); empty clusters and key collisions are dropped. */ export function reconcileUserEdits( clusters: T[], @@ -73,31 +78,35 @@ export function reconcileUserEdits (collection.clusterKey ? [] : collection.assetIds)), ); - const computed = new Map( - clusters.map((cluster) => [ - cluster.key, - new Set(cluster.assetIds.filter((id) => !splitOff.has(id))), - ]), - ); + const hostOf = new Map(); + for (const cluster of clusters) { + for (const id of cluster.assetIds) if (!splitOff.has(id)) hostOf.set(id, cluster); + } + const followedBy = new Map(); const frozenKeys = new Set(); const frozenAssets = new Set(); for (const collection of userCollections) { - if (!collection.clusterKey) continue; - const members = computed.get(collection.clusterKey); - if (!members || collection.assetIds.some((id) => !members.has(id))) { + const [first] = collection.assetIds; + if (!collection.clusterKey || first === undefined) continue; + const host = hostOf.get(first); + const contained = host && collection.assetIds.every((id) => hostOf.get(id) === host); + if (host && contained && !followedBy.has(host.key)) { + followedBy.set(host.key, collection.clusterKey); + } else { frozenKeys.add(collection.clusterKey); for (const id of collection.assetIds) frozenAssets.add(id); } } - const seen = new Set(); + const seen = new Set(frozenKeys); const result: T[] = []; for (const cluster of clusters) { const assetIds = cluster.assetIds.filter((id) => !splitOff.has(id) && !frozenAssets.has(id)); const first = assetIds[0]; - if (!first) continue; - const keep = !frozenKeys.has(cluster.key) && first === cluster.assetIds[0]; - const key = keep ? cluster.key : `${prefix}:${first}`; - if (frozenKeys.has(key) || seen.has(key)) continue; + if (first === undefined) continue; + const key = + followedBy.get(cluster.key) ?? + (first === cluster.assetIds[0] ? cluster.key : `${prefix}:${first}`); + if (seen.has(key)) continue; seen.add(key); result.push({ ...cluster, key, assetIds }); } @@ -135,7 +144,8 @@ async function meanEmbeddings( /** The vectors of the most common dimension (a model change mixes dimensions). */ function sameDimension(items: T[]): T[] { const counts = new Map(); - for (const item of items) counts.set(item.vector.length, (counts.get(item.vector.length) ?? 0) + 1); + for (const item of items) + counts.set(item.vector.length, (counts.get(item.vector.length) ?? 0) + 1); const dim = [...counts.entries()].sort((a, b) => b[1] - a[1] || b[0] - a[0])[0]?.[0]; return items.filter((item) => item.vector.length === dim); } @@ -147,7 +157,10 @@ async function userCollections( ): Promise { const rows = await deps.prisma.mediaCollection.findMany({ where: { spaceId, kind, labelSource: "user" }, - select: { clusterKey: true, items: { select: { assetId: true }, orderBy: { position: "asc" } } }, + select: { + clusterKey: true, + items: { select: { assetId: true }, orderBy: { position: "asc" } }, + }, }); return rows.map((row) => ({ clusterKey: row.clusterKey, @@ -170,6 +183,16 @@ async function libraryOwner(deps: MediaServiceDeps, spaceId: string): Promise } = {}, ): Promise { + const userId = await libraryOwner(deps, spaceId); + const profile = userId + ? await deps.prisma.userProfile.findUnique({ + where: { userId }, + select: { timeZone: true, locale: true }, + }) + : null; + const timeZone = validTimeZone(profile?.timeZone) ?? "UTC"; + const locale = profile?.locale ?? undefined; + const label = (range: { startAt: Date | null; endAt: Date | null }, assetIds: string[]) => + ruleLabel({ ...range, albumNames: albums(assetIds) }, locale, timeZone); const rows = await deps.repos.listAssetsForClustering(spaceId); const extras = await deps.prisma.mediaAsset.findMany({ where: { spaceId }, @@ -228,20 +262,19 @@ export async function clusterSpace( albumNames: row.albumNames, embedding: embeddings.get(row.id) ?? null, })); - const computedSessions = buildSessions(clusterAssets); - const centroids = new Map(computedSessions.map((session) => [session.key, session.centroid])); + const computedSessions = buildSessions(clusterAssets, { timeZone, ...options.sessionOptions }); const sessions: MediaClusterInput[] = reconcileUserEdits( - computedSessions.map((session) => ({ key: session.key, assetIds: session.assetIds })), + computedSessions, await userCollections(deps, spaceId, "session"), "session", ).map((session) => { const range = span(session.assetIds); return { key: session.key, - title: ruleLabel({ ...range, albumNames: albums(session.assetIds) }), + title: label(range, session.assetIds), labelSource: "rule", ...range, - centroid: centroids.get(session.key) ?? null, + centroid: session.centroid, coverAssetId: cover(session.assetIds), assetIds: session.assetIds, }; @@ -275,7 +308,6 @@ export async function clusterSpace( row.labelSource === "model" && row.clusterKey ? [[row.clusterKey, row.title] as const] : [], ), ); - const userId = await libraryOwner(deps, spaceId); let modelCalls = 0; let modelLabels = 0; const themes: MediaClusterInput[] = []; @@ -291,7 +323,7 @@ export async function clusterSpace( if (title) modelLabels += 1; themes.push({ key: theme.key, - title: title ?? ruleLabel({ ...range, albumNames: albums(theme.assetIds) }), + title: title ?? label(range, theme.assetIds), labelSource: title ? "model" : "rule", ...range, centroid: null, @@ -330,7 +362,11 @@ async function themeTitle( for (const [index, row] of covers.entries()) { try { const file = await resolveInside(deps.mediaRoot, row.thumbPath as string); - images.push({ name: `cover-${index + 1}.jpg`, mimeType: "image/jpeg", data: await readFile(file) }); + images.push({ + name: `cover-${index + 1}.jpg`, + mimeType: "image/jpeg", + data: await readFile(file), + }); } catch { // A missing thumbnail only shrinks the set. } diff --git a/packages/adapters/src/media-ingest-files.ts b/packages/adapters/src/media-ingest-files.ts index de6938e..7ffa92f 100644 --- a/packages/adapters/src/media-ingest-files.ts +++ b/packages/adapters/src/media-ingest-files.ts @@ -1,11 +1,11 @@ import { createHash, randomUUID } from "node:crypto"; import { createReadStream } from "node:fs"; -import { copyFile, mkdir, rename, rm, unlink, writeFile } from "node:fs/promises"; +import { copyFile, lstat, mkdir, rename, rm, unlink, writeFile } from "node:fs/promises"; import path from "node:path"; import { pipeline } from "node:stream/promises"; -import type { MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js"; import { getLogger } from "@rakazo/logging"; import { resolveInside } from "@rakazo/media/node"; +import type { MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js"; import { FOLDER_MEDIA_EXTENSIONS } from "./media-folder-source.js"; /* @@ -100,8 +100,11 @@ export function errorMessage(error: unknown): string { return (error instanceof Error ? error.message : String(error)).slice(0, 500); } -/** A full disk is retried later instead of failing the asset for good. */ -export function isDiskFull(error: unknown): boolean { - if ((error as NodeJS.ErrnoException | null)?.code === "ENOSPC") return true; - return error instanceof Error && /No space left on device/i.test(error.message); +/** Whether a media-root-relative path is an existing regular file. */ +export async function fileExists(mediaRoot: string, relative: string): Promise { + try { + return (await lstat(await resolveInside(mediaRoot, relative))).isFile(); + } catch { + return false; + } } diff --git a/packages/adapters/src/media-ingest-proxy.test.ts b/packages/adapters/src/media-ingest-proxy.test.ts new file mode 100644 index 0000000..3d78fa4 --- /dev/null +++ b/packages/adapters/src/media-ingest-proxy.test.ts @@ -0,0 +1,185 @@ +import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { ffmpegAvailable } from "./media-emulator-clips.js"; +import { runMediaTool } from "./media-folder-source-probe.js"; +import { buildProxy, probeOriginal } from "./media-ingest-proxy.js"; + +/* + * Real ffmpeg runs on synthetic lavfi fixtures (test patterns and tones only). + * Skipped where ffmpeg is not installed. + */ + +const describeFfmpeg = ffmpegAvailable() ? describe : describe.skip; + +interface Probe { + streams: Array<{ + codec_type: string; + codec_name: string; + width?: number; + height?: number; + r_frame_rate?: string; + avg_frame_rate?: string; + color_transfer?: string; + side_data_list?: Array<{ rotation?: number }>; + }>; + format: { tags?: Record }; +} + +async function ffprobe(file: string): Promise { + const { stdout } = await runMediaTool("ffprobe", [ + ...["-v", "error", "-print_format", "json", "-show_format", "-show_streams", file], + ]); + return JSON.parse(stdout.toString("utf8")) as Probe; +} + +async function keyframeTimes(file: string): Promise { + const { stdout } = await runMediaTool("ffprobe", [ + ...["-v", "error", "-select_streams", "v:0", "-skip_frame", "nokey"], + ...["-show_entries", "frame=pts_time", "-of", "csv=p=0", file], + ]); + return stdout + .toString("utf8") + .trim() + .split("\n") + .map((line) => Number(line.split(",")[0])); +} + +describeFfmpeg("buildProxy (ffmpeg)", () => { + let root: string; + let originals: string; + + async function ffmpeg(args: string[]): Promise { + await runMediaTool("ffmpeg", ["-v", "error", "-y", ...args]); + } + + beforeAll(async () => { + root = await mkdtemp(path.join(tmpdir(), "media-proxy-")); + originals = path.join(root, "originals"); + await mkdir(originals, { recursive: true }); + }); + + afterAll(async () => { + await rm(root, { recursive: true, force: true }); + }); + + it("builds a 720p CFR proxy with 1 s keyframes, AAC, poster, and sprite without metadata", async () => { + const source = path.join(originals, "landscape.mp4"); + // Variable frame rate: every third frame is held a little longer. + await ffmpeg([ + ...["-f", "lavfi", "-i", "testsrc2=s=1920x1080:r=30:d=3"], + ...["-f", "lavfi", "-i", "sine=f=440:sample_rate=44100:d=3"], + ...["-vf", "setpts='PTS+if(eq(mod(N,3),0),0.02/TB,0)'", "-fps_mode", "vfr"], + ...["-c:v", "libx264", "-preset", "ultrafast", "-g", "90", "-pix_fmt", "yuv420p"], + ...["-c:a", "aac", "-shortest"], + ...["-metadata", "location=+52.5200+013.4050/", "-metadata", "make=TestCam", source], + ]); + const outputs = await buildProxy({ + mediaRoot: root, + spaceId: "space", + assetId: "landscape", + original: source, + still: false, + }); + expect(outputs.proxyPath).toBe("space/assets/landscape/proxy.mp4"); + expect(outputs.spritePath).toBe("space/assets/landscape/sprite.jpg"); + expect(outputs.hdr).toBe(false); + + const proxy = await ffprobe(path.join(root, outputs.proxyPath)); + const video = proxy.streams.find((stream) => stream.codec_type === "video"); + const audio = proxy.streams.find((stream) => stream.codec_type === "audio"); + expect([video?.width, video?.height]).toEqual([1280, 720]); + expect(video?.codec_name).toBe("h264"); + expect(video?.r_frame_rate).toBe("30/1"); + expect(audio?.codec_name).toBe("aac"); + expect(proxy.format.tags?.location).toBeUndefined(); + expect(proxy.format.tags?.make).toBeUndefined(); + + const keyframes = await keyframeTimes(path.join(root, outputs.proxyPath)); + expect(keyframes.length).toBeGreaterThanOrEqual(3); + for (let i = 1; i < keyframes.length; i++) { + expect((keyframes[i] ?? 0) - (keyframes[i - 1] ?? 0)).toBeLessThanOrEqual(1.05); + } + + const sprite = await ffprobe(path.join(root, outputs.spritePath as string)); + expect(sprite.streams[0]?.width).toBe(1600); + const thumb = await ffprobe(path.join(root, outputs.thumbPath)); + expect(thumb.streams[0]?.width).toBe(480); + expect(thumb.format.tags?.location).toBeUndefined(); + }); + + it("applies display rotation without upscaling", async () => { + const plain = path.join(originals, "plain.mp4"); + const rotated = path.join(originals, "rotated.mp4"); + await ffmpeg([ + ...["-f", "lavfi", "-i", "testsrc2=s=640x360:r=25:d=2"], + ...["-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", plain], + ]); + await ffmpeg(["-display_rotation", "90", "-i", plain, "-c", "copy", rotated]); + const { probe } = await probeOriginal(rotated, {}); + expect(Math.abs(probe.rotation)).toBe(90); + + const outputs = await buildProxy({ + mediaRoot: root, + spaceId: "space", + assetId: "rotated", + original: rotated, + still: false, + }); + const proxy = await ffprobe(path.join(root, outputs.proxyPath)); + const video = proxy.streams.find((stream) => stream.codec_type === "video"); + expect([video?.width, video?.height]).toEqual([360, 640]); + expect(video?.side_data_list?.some((data) => data.rotation)).toBeFalsy(); + expect(proxy.streams.some((stream) => stream.codec_type === "audio")).toBe(false); + }); + + it("tone-maps HDR sources to SDR", async () => { + const hdr = path.join(originals, "hdr.mp4"); + await ffmpeg([ + ...["-f", "lavfi", "-i", "testsrc2=s=1280x720:r=25:d=1"], + ...["-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p10le"], + ...["-color_primaries", "bt2020", "-color_trc", "smpte2084", "-colorspace", "bt2020nc"], + hdr, + ]); + const outputs = await buildProxy({ + mediaRoot: root, + spaceId: "space", + assetId: "hdr", + original: hdr, + still: false, + }); + expect(outputs.hdr).toBe(true); + expect(outputs.tonemapped).toBe(true); + const proxy = await ffprobe(path.join(root, outputs.proxyPath)); + expect(proxy.streams[0]?.color_transfer).not.toBe("smpte2084"); + }); + + it("gives a photo a resized JPEG proxy and poster but no sprite", async () => { + const photo = path.join(originals, "photo.jpg"); + await ffmpeg(["-f", "lavfi", "-i", "smptehdbars=s=1600x1200", "-frames:v", "1", photo]); + const outputs = await buildProxy({ + mediaRoot: root, + spaceId: "space", + assetId: "photo", + original: photo, + still: true, + }); + expect(outputs.proxyPath).toBe("space/assets/photo/proxy.jpg"); + expect(outputs.spritePath).toBeNull(); + const proxy = await ffprobe(path.join(root, outputs.proxyPath)); + expect([proxy.streams[0]?.width, proxy.streams[0]?.height]).toEqual([960, 720]); + }); + + it("refuses to write outside the media root", async () => { + await expect( + buildProxy({ + mediaRoot: root, + spaceId: "../escape", + assetId: "x", + original: path.join(originals, "plain.mp4"), + still: false, + }), + ).rejects.toThrow(/outside/); + }); +}); diff --git a/packages/adapters/src/media-ingest-proxy.ts b/packages/adapters/src/media-ingest-proxy.ts index 6726ce7..dc503d9 100644 --- a/packages/adapters/src/media-ingest-proxy.ts +++ b/packages/adapters/src/media-ingest-proxy.ts @@ -190,7 +190,14 @@ export async function buildProxy(input: { tools, signal, ); - return { proxyPath: proxyRel, thumbPath: thumbRel, spritePath: spriteRel, probe, hdr, tonemapped }; + return { + proxyPath: proxyRel, + thumbPath: thumbRel, + spritePath: spriteRel, + probe, + hdr, + tonemapped, + }; } async function writeThumb( @@ -204,7 +211,9 @@ async function writeThumb( target, ".jpg", (out) => [ - ...["-v", "error", "-y", "-ss", (atMs / 1000).toFixed(3), "-i", source], + // Seeking a single image, even to 0, yields no frame. + ...["-v", "error", "-y", ...(atMs > 0 ? ["-ss", (atMs / 1000).toFixed(3)] : [])], + ...["-i", source], ...["-map", "0:v:0", "-map_metadata", "-1", "-frames:v", "1", "-update", "1"], ...["-vf", `scale='min(${THUMB_WIDTH},iw)':-2`, "-q:v", "3", out], ], diff --git a/packages/adapters/src/media-ingest.postgres.test.ts b/packages/adapters/src/media-ingest.postgres.test.ts new file mode 100644 index 0000000..acf915c --- /dev/null +++ b/packages/adapters/src/media-ingest.postgres.test.ts @@ -0,0 +1,697 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, readdir, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { + BackgroundJob, + BackgroundJobHandlers, + JobPublisher, + MediaInferenceProvider, +} from "@rakazo/adapter-kit"; +import { dispatchBackgroundJob } from "@rakazo/adapter-kit"; +import type { MediaCapabilities } from "@rakazo/contracts"; +import type { MediaRepos, PrismaClient } from "@rakazo/db"; +import { bootstrapUserSpace, createDb, createMediaRepos } from "@rakazo/db"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import type { MediaModelClient, MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js"; +import { EmulatorMediaInferenceProvider } from "./media-emulator.js"; +import { ffmpegAvailable } from "./media-emulator-clips.js"; +import { EmulatorMediaSourceProvider } from "./media-emulator-source.js"; +import { FolderMediaSourceProvider } from "./media-folder-source.js"; +import type { MediaIngestHandlerOptions } from "./media-ingest.js"; +import { + createMediaIngestHandlers, + createMediaIngestService, + MEDIA_BUSY_RETRY_MS, + MEDIA_PAUSE_RETRY_MS, + MEDIA_STALE_CLAIM_MS, + resumeMediaIngest, +} from "./media-ingest.js"; +import { MediaSidecarError } from "./media-sidecar.js"; + +/* + * The ingest job chain against PostgreSQL, the deterministic media emulator, + * and real ffmpeg on synthetic clips. Jobs are recorded and drained by hand. + */ + +const databaseUrl = process.env.DATABASE_URL; +const describePostgres = + process.env.VERIFY_DATABASE && databaseUrl && ffmpegAvailable() + ? describe.sequential + : describe.skip; + +const signupEnv = { signupsEnabled: "true", signupAllowlist: "" }; +const LIMITS = { gpu: 1, cpu: 2 }; + +/** Records jobs keyed like the real queue: a replace key replaces the pending job. */ +class RecordingJobs implements JobPublisher { + readonly pending = new Map(); + readonly log: BackgroundJob[] = []; + + async enqueue(job: BackgroundJob): Promise { + this.log.push(job); + this.pending.set(job.replaceKey ?? `${job.name}:${randomUUID()}`, job); + } + + async cancel(key: string): Promise { + this.pending.delete(key); + } + + async close(): Promise {} + + reset(): void { + this.pending.clear(); + this.log.length = 0; + } + + named(name: BackgroundJob["name"]): BackgroundJob[] { + return this.log.filter((job) => job.name === name); + } +} + +/** Run pending jobs (delays ignored) until the queue is empty. */ +async function drain( + handlers: ReturnType, + jobs: RecordingJobs, + limit = 1000, +): Promise { + for (let ran = 0; ran < limit; ran++) { + const next = jobs.pending.entries().next(); + if (next.done) return ran; + const [key, job] = next.value; + jobs.pending.delete(key); + await dispatchBackgroundJob(handlers as BackgroundJobHandlers, job.name, job.payload); + } + throw new Error("media job queue did not drain"); +} + +describePostgres("media ingest jobs (PostgreSQL)", () => { + let prisma: PrismaClient; + let repos: MediaRepos; + let closeDb: () => Promise; + let mediaRoot: string; + let scratch: string; + const userIds: string[] = []; + + beforeAll(async () => { + const db = createDb(databaseUrl as string, { poolMax: 8 }); + prisma = db.prisma; + closeDb = async () => { + await db.prisma.$disconnect(); + await db.pool.end(); + }; + repos = createMediaRepos(prisma); + scratch = await mkdtemp(path.join(tmpdir(), "media-ingest-")); + mediaRoot = path.join(scratch, "media"); + await mkdir(mediaRoot, { recursive: true }); + }); + + afterAll(async () => { + if (!prisma) return; + const memberships = await prisma.member.findMany({ + where: { userId: { in: userIds } }, + select: { organizationId: true }, + }); + await prisma.organization.deleteMany({ + where: { id: { in: memberships.map((member) => member.organizationId) } }, + }); + await prisma.user.deleteMany({ where: { id: { in: userIds } } }); + await closeDb(); + await rm(scratch, { recursive: true, force: true }); + }); + + async function newSpace() { + const id = `mi${randomUUID().replaceAll("-", "")}`; + userIds.push(id); + await prisma.user.create({ data: { id, name: "Test", email: `${id}@rakazo.test` } }); + const { spaceId } = await bootstrapUserSpace(prisma, { id }, signupEnv, { + claimDeploymentOwner: false, + }); + return { userId: id, spaceId }; + } + + function setup( + overrides: { + inference?: MediaInferenceProvider | null; + model?: MediaModelClient; + options?: MediaIngestHandlerOptions; + } = {}, + ) { + const jobs = new RecordingJobs(); + const events: Array<{ spaceId: string; event: MediaRealtimeEvent }> = []; + const inference = + overrides.inference === undefined + ? new EmulatorMediaInferenceProvider({ root: mediaRoot }) + : overrides.inference; + const deps: MediaServiceDeps = { + prisma, + repos, + jobs, + inference, + renderer: {} as MediaServiceDeps["renderer"], + motion: {} as MediaServiceDeps["motion"], + publisher: {} as MediaServiceDeps["publisher"], + shortener: null, + sources: { + folder: new FolderMediaSourceProvider(), + emulator: new EmulatorMediaSourceProvider({ + fixturesDir: path.join(mediaRoot, "emulator-library"), + }), + }, + mediaRoot, + profiles: async () => { + throw new Error("not used"); + }, + artifacts: {} as MediaServiceDeps["artifacts"], + model: overrides.model ?? { generate: async () => null }, + publish: async (spaceId, event) => { + events.push({ spaceId, event }); + }, + now: () => new Date(), + }; + const handlers = createMediaIngestHandlers(deps, { + limits: LIMITS, + minFreeBytes: 0, + ...overrides.options, + }); + return { deps, jobs, events, handlers, service: createMediaIngestService(deps) }; + } + + async function emulatorLibrary(count: number, seed: number) { + const space = await newSpace(); + const source = await repos.createSource(space.spaceId, { + kind: "emulator", + label: "Emulator", + ownerUserId: space.userId, + config: { count, seed }, + }); + return { ...space, sourceId: source.id }; + } + + async function states(spaceId: string) { + const rows = await prisma.mediaAsset.findMany({ + where: { spaceId }, + select: { ingestState: true }, + }); + return rows.reduce>((acc, row) => { + acc[row.ingestState] = (acc[row.ingestState] ?? 0) + 1; + return acc; + }, {}); + } + + async function sessionMembers(spaceId: string) { + const rows = await prisma.mediaCollection.findMany({ + where: { spaceId, kind: "session" }, + select: { + id: true, + clusterKey: true, + labelSource: true, + items: { select: { assetId: true }, orderBy: { position: "asc" } }, + }, + orderBy: [{ startAt: "asc" }, { id: "asc" }], + }); + return rows.map((row) => ({ ...row, assetIds: row.items.map((item) => item.assetId) })); + } + + it("indexes a 1,000-file folder incrementally; an unchanged second sync enqueues nothing", async () => { + const space = await newSpace(); + const folder = path.join(scratch, "folder-1000"); + await mkdir(path.join(folder, "trip"), { recursive: true }); + for (let index = 0; index < 1000; index++) { + const dir = index < 500 ? folder : path.join(folder, "trip"); + await writeFile(path.join(dir, `clip-${String(index).padStart(4, "0")}.mp4`), ""); + } + const source = await repos.createSource(space.spaceId, { + kind: "folder", + label: "Folder", + ownerUserId: space.userId, + config: { path: folder }, + }); + const { jobs, handlers, events } = setup(); + + await handlers["media.sync"]({ sourceId: source.id }); + expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1000); + // Only a window of proxy work is queued; finished jobs refill it. + const proxyKeys = new Set(jobs.named("media.proxy").map((job) => job.replaceKey)); + expect(proxyKeys.size).toBe(LIMITS.cpu * 2); + expect(jobs.named("media.cluster")).toHaveLength(1); + expect(events.map(({ event }) => event)).toContainEqual({ + type: "media.source", + sourceId: source.id, + status: "idle", + error: null, + }); + + jobs.reset(); + await handlers["media.sync"]({ sourceId: source.id }); + expect(jobs.log).toEqual([]); + expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1000); + + // One file changes and one disappears. + await writeFile(path.join(folder, "clip-0001.mp4"), "changed"); + await rm(path.join(folder, "trip", "clip-0999.mp4")); + await handlers["media.sync"]({ sourceId: source.id }); + expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(999); + const changed = await prisma.mediaAsset.findFirstOrThrow({ + where: { spaceId: space.spaceId, externalId: "clip-0001.mp4" }, + }); + expect(changed.sizeBytes).toBe(7n); + expect(jobs.named("media.cluster")).toHaveLength(1); + const source2 = await repos.getSource(space.spaceId, source.id); + expect(source2.status).toBe("idle"); + }); + + it("does not delete a library when its source suddenly lists nothing", async () => { + const space = await newSpace(); + const folder = path.join(scratch, "unmounted"); + await mkdir(folder, { recursive: true }); + await writeFile(path.join(folder, "a.mp4"), ""); + const source = await repos.createSource(space.spaceId, { + kind: "folder", + label: "Folder", + ownerUserId: space.userId, + config: { path: folder }, + }); + const { handlers } = setup(); + await handlers["media.sync"]({ sourceId: source.id }); + await rm(path.join(folder, "a.mp4")); + await handlers["media.sync"]({ sourceId: source.id }); + expect(await prisma.mediaAsset.count({ where: { spaceId: space.spaceId } })).toBe(1); + const record = await repos.getSource(space.spaceId, source.id); + expect(record.status).toBe("error"); + expect(record.error).toMatch(/listed no files/); + }); + + describe("emulator library end to end", () => { + let library: Awaited>; + let env: ReturnType; + + beforeAll(async () => { + library = await emulatorLibrary(15, 3); + env = setup(); + await env.handlers["media.sync"]({ sourceId: library.sourceId }); + await drain(env.handlers, env.jobs); + }, 240_000); + + it("proxies and analyses every asset and writes scored segments", async () => { + expect(await states(library.spaceId)).toEqual({ analyzed: 15 }); + const assets = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } }); + for (const asset of assets) { + expect(asset.proxyPath).toMatch(/^.+\/assets\/.+\/proxy\.(mp4|jpg)$/); + expect(asset.contentHash).toMatch(/^[0-9a-f]{64}$/); + expect(asset.analysisVersion).toBe(1); + expect((await stat(path.join(mediaRoot, asset.proxyPath as string))).size).toBeGreaterThan( + 0, + ); + } + const segments = await prisma.mediaSegment.count({ where: { spaceId: library.spaceId } }); + expect(segments).toBeGreaterThanOrEqual(15); + const kinds = await prisma.mediaAnalysis.groupBy({ + by: ["kind"], + where: { spaceId: library.spaceId }, + _count: true, + }); + expect(kinds.map((row) => row.kind).sort()).toEqual( + ["audio", "classifier", "embeddings", "faces", "saliency", "scenes", "transcript"].sort(), + ); + const progress = env.events + .filter(({ event }) => event.type === "media.asset") + .map(({ event }) => (event as { ingestState: string }).ingestState); + expect(new Set(progress)).toEqual(new Set(["proxying", "proxied", "analyzing", "analyzed"])); + }); + + it("keeps face embeddings in face tracks only and suggests one person per identity", async () => { + const tracks = await prisma.mediaFaceTrack.findMany({ where: { spaceId: library.spaceId } }); + expect(tracks).toHaveLength(4); + expect(tracks.every((track) => track.embedding && track.personId)).toBe(true); + const people = await prisma.mediaPerson.findMany({ where: { spaceId: library.spaceId } }); + expect(people).toHaveLength(2); + expect(new Set(tracks.map((track) => track.personId)).size).toBe(2); + const faces = await prisma.mediaAnalysis.findMany({ + where: { spaceId: library.spaceId, kind: "faces" }, + }); + for (const row of faces) { + expect(row.vectors).toBeNull(); + expect(JSON.stringify(row.data)).not.toMatch(/embedding/i); + } + }); + + it("builds sessions by time and place and themes with rule labels without consent", async () => { + const sessions = await sessionMembers(library.spaceId); + expect(sessions).toHaveLength(3); + expect(sessions.flatMap((session) => session.assetIds)).toHaveLength(15); + const themes = await prisma.mediaCollection.findMany({ + where: { spaceId: library.spaceId, kind: "theme" }, + include: { items: true }, + }); + expect(themes).toHaveLength(2); + expect(themes.every((theme) => theme.labelSource === "rule")).toBe(true); + expect(env.events.map(({ event }) => event)).toContainEqual({ + type: "media.collections", + kind: "theme", + }); + }); + + it("labels themes through the model only from cover thumbnails, never calling out with face data", async () => { + const calls: Array[0]> = []; + const model: MediaModelClient = { + generate: async (input: Parameters[0]) => { + calls.push(input); + return { title: "Test pattern" } as T; + }, + }; + const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { + throw new Error("no outbound calls during clustering"); + }); + try { + const withModel = setup({ model }); + await withModel.handlers["media.cluster"]({ spaceId: library.spaceId }); + } finally { + expect(fetchSpy).not.toHaveBeenCalled(); + fetchSpy.mockRestore(); + } + expect(calls).toHaveLength(2); + for (const call of calls) { + expect(call.userId).toBe(library.userId); + expect(call.images?.length).toBeGreaterThan(0); + expect(call.images?.length).toBeLessThanOrEqual(9); + expect(call.images?.every((image) => image.mimeType === "image/jpeg")).toBe(true); + expect(`${call.prompt}${call.instructions}`).not.toMatch(/person|face|embedding/i); + } + const themes = await prisma.mediaCollection.findMany({ + where: { spaceId: library.spaceId, kind: "theme" }, + }); + expect(themes.map((theme) => [theme.title, theme.labelSource])).toEqual([ + ["Test pattern", "model"], + ["Test pattern", "model"], + ]); + }); + + it("re-clustering keeps the user's merges and splits", async () => { + const [first, second, third] = await sessionMembers(library.spaceId); + if (!first || !second || !third) throw new Error("expected three sessions"); + await repos.mergeCollections(library.spaceId, [first.id, second.id]); + const splitAt = third.assetIds[2] as string; + await repos.splitCollection(library.spaceId, third.id, splitAt); + const before = await sessionMembers(library.spaceId); + + await env.handlers["media.cluster"]({ spaceId: library.spaceId }); + const after = await sessionMembers(library.spaceId); + expect(after.map((session) => session.assetIds)).toEqual( + before.map((session) => session.assetIds), + ); + const all = after.flatMap((session) => session.assetIds); + expect(new Set(all).size).toBe(all.length); + expect(all).toHaveLength(15); + }); + + it("re-runs nothing for unchanged content and adds rows for a new model version", async () => { + const asset = await prisma.mediaAsset.findFirstOrThrow({ + where: { spaceId: library.spaceId, kind: "video" }, + orderBy: { externalId: "asc" }, + }); + const analyses = await prisma.mediaAnalysis.count({ where: { assetId: asset.id } }); + const proxyFile = path.join(mediaRoot, asset.proxyPath as string); + const proxyMtime = (await stat(proxyFile)).mtimeMs; + const inference = new EmulatorMediaInferenceProvider({ root: mediaRoot }); + const scenes = vi.spyOn(inference, "detectScenes"); + const run = setup({ inference }); + + // A repeated job for an analysed asset is a no-op. + await run.handlers["media.analyze"]({ assetId: asset.id }); + expect(scenes).not.toHaveBeenCalled(); + + // Re-proxy with unchanged bytes keeps the proxy and reuses every analysis. + await prisma.mediaAsset.update({ where: { id: asset.id }, data: { ingestState: "indexed" } }); + await run.handlers["media.proxy"]({ assetId: asset.id }); + expect((await stat(proxyFile)).mtimeMs).toBe(proxyMtime); + await run.handlers["media.analyze"]({ assetId: asset.id }); + expect(scenes).not.toHaveBeenCalled(); + expect(await prisma.mediaAnalysis.count({ where: { assetId: asset.id } })).toBe(analyses); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState, + ).toBe("analyzed"); + + // A new embedding model writes new rows and keeps the old ones. + const upgraded = new EmulatorMediaInferenceProvider({ root: mediaRoot }); + const base = upgraded.capabilities.bind(upgraded); + upgraded.capabilities = async (signal?: AbortSignal): Promise => { + const caps = await base(signal); + return { + ...caps, + operations: { + ...caps.operations, + embedFrames: { ...caps.operations.embedFrames, model: "emulator-v2" }, + }, + }; + }; + await prisma.mediaAsset.update({ where: { id: asset.id }, data: { ingestState: "proxied" } }); + await setup({ inference: upgraded }).handlers["media.analyze"]({ assetId: asset.id }); + const versions = await prisma.mediaAnalysis.findMany({ + where: { assetId: asset.id, kind: "embeddings" }, + select: { modelVersion: true }, + orderBy: { modelVersion: "asc" }, + }); + expect(versions.map((row) => row.modelVersion)).toEqual([ + "emulator:emulator", + "emulator:emulator-v2", + ]); + expect(await prisma.mediaAnalysis.count({ where: { assetId: asset.id } })).toBe(analyses + 1); + }); + + it("removes a source with its assets and derived files", async () => { + const assets = await prisma.mediaAsset.findMany({ + where: { spaceId: library.spaceId }, + select: { id: true }, + }); + await env.service.removeSource( + { spaceId: library.spaceId, userId: library.userId }, + library.sourceId, + ); + expect(await prisma.mediaAsset.count({ where: { spaceId: library.spaceId } })).toBe(0); + const dir = path.join(mediaRoot, library.spaceId, "assets"); + const left = await readdir(dir).catch(() => []); + expect(left.filter((name) => assets.some((asset) => asset.id === name))).toEqual([]); + }); + }); + + it("pauses on a full disk with a clear message and resumes afterwards", async () => { + const library = await emulatorLibrary(2, 11); + const full = setup({ options: { minFreeBytes: Number.MAX_SAFE_INTEGER } }); + await full.handlers["media.sync"]({ sourceId: library.sourceId }); + const [asset] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } }); + if (!asset) throw new Error("expected an asset"); + full.jobs.reset(); + const before = Date.now(); + + await full.handlers["media.proxy"]({ assetId: asset.id }); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState, + ).toBe("indexed"); + const source = await repos.getSource(library.spaceId, library.sourceId); + expect(source.error).toBe("Media processing is paused because the disk is full"); + const [retry] = full.jobs.named("media.proxy"); + expect(retry?.replaceKey).toBe(`media.proxy:${asset.id}`); + expect(retry?.availableAt?.getTime()).toBeGreaterThanOrEqual(before + MEDIA_PAUSE_RETRY_MS); + + const roomy = setup(); + await roomy.handlers["media.proxy"]({ assetId: asset.id }); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } })).ingestState, + ).toBe("proxied"); + expect((await repos.getSource(library.spaceId, library.sourceId)).error).toBeNull(); + }); + + it("defers jobs beyond the concurrency limit instead of holding a worker slot", async () => { + const library = await emulatorLibrary(2, 12); + const run = setup({ options: { limits: { gpu: 1, cpu: 1 } } }); + await run.handlers["media.sync"]({ sourceId: library.sourceId }); + const assets = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } }); + run.jobs.reset(); + const before = Date.now(); + await Promise.all(assets.map((asset) => run.handlers["media.proxy"]({ assetId: asset.id }))); + const deferred = run.jobs + .named("media.proxy") + .filter((job) => (job.availableAt?.getTime() ?? 0) >= before + MEDIA_BUSY_RETRY_MS); + expect(deferred).toHaveLength(1); + expect(await states(library.spaceId)).toEqual({ indexed: 1, proxied: 1 }); + }); + + it("defers analysis while the sidecar is unreachable", async () => { + const library = await emulatorLibrary(1, 13); + const inference = new EmulatorMediaInferenceProvider({ root: mediaRoot }); + vi.spyOn(inference, "detectScenes").mockRejectedValue( + new MediaSidecarError("unreachable", "media sidecar is unreachable", null, "detectScenes"), + ); + const run = setup({ inference }); + await run.handlers["media.sync"]({ sourceId: library.sourceId }); + const [asset] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } }); + if (!asset) throw new Error("expected an asset"); + await run.handlers["media.proxy"]({ assetId: asset.id }); + run.jobs.reset(); + await run.handlers["media.analyze"]({ assetId: asset.id }); + expect(await states(library.spaceId)).toEqual({ proxied: 1 }); + expect(run.jobs.named("media.analyze")[0]?.availableAt).toBeInstanceOf(Date); + }); + + it("returns an aborted proxy to the queue state and takes over stale claims", async () => { + const library = await emulatorLibrary(2, 14); + const controller = new AbortController(); + controller.abort(new Error("worker stopping")); + const stopping = setup({ options: { signal: controller.signal } }); + await setup().handlers["media.sync"]({ sourceId: library.sourceId }); + const [first, second] = await prisma.mediaAsset.findMany({ + where: { spaceId: library.spaceId }, + orderBy: { externalId: "asc" }, + }); + if (!first || !second) throw new Error("expected two assets"); + await expect(stopping.handlers["media.proxy"]({ assetId: first.id })).rejects.toThrow( + /abort|stopping/, + ); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: first.id } })).ingestState, + ).toBe("indexed"); + + // A fresh claim by another worker is left alone; a stale one is taken over. + const run = setup(); + await prisma.mediaAsset.update({ where: { id: second.id }, data: { ingestState: "proxying" } }); + await run.handlers["media.proxy"]({ assetId: second.id }); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: second.id } })).ingestState, + ).toBe("proxying"); + const stale = new Date(Date.now() - MEDIA_STALE_CLAIM_MS - 60_000); + await prisma.$executeRaw`UPDATE media_assets SET "updatedAt" = ${stale} WHERE id = ${second.id}`; + await run.handlers["media.proxy"]({ assetId: second.id }); + expect( + (await prisma.mediaAsset.findUniqueOrThrow({ where: { id: second.id } })).ingestState, + ).toBe("proxied"); + }); + + it("resumes stranded work after a restart", async () => { + const library = await emulatorLibrary(3, 15); + const run = setup(); + await run.handlers["media.sync"]({ sourceId: library.sourceId }); + const [stranded] = await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } }); + if (!stranded) throw new Error("expected an asset"); + // The queue was lost and one asset was claimed by a worker that died. + run.jobs.reset(); + const stale = new Date(Date.now() - MEDIA_STALE_CLAIM_MS - 60_000); + await prisma.$executeRaw`UPDATE media_assets SET "ingestState" = 'proxying', "updatedAt" = ${stale} WHERE id = ${stranded.id}`; + + expect(await resumeMediaIngest(run.deps, LIMITS)).toBeGreaterThanOrEqual(1); + expect(run.jobs.named("media.proxy").map((job) => job.replaceKey)).toContain( + `media.proxy:${stranded.id}`, + ); + // Other tests' spaces share this database; only this library's work runs here. + const ours = new Set( + (await prisma.mediaAsset.findMany({ where: { spaceId: library.spaceId } })).map( + (asset) => `media.proxy:${asset.id}`, + ), + ); + for (const key of [...run.jobs.pending.keys()]) { + if (!ours.has(key)) run.jobs.pending.delete(key); + } + await drain(run.handlers, run.jobs); + expect(await states(library.spaceId)).toEqual({ analyzed: 3 }); + expect(run.jobs.named("media.cluster").length).toBeGreaterThan(0); + }); + + it("ignores jobs for assets and sources that no longer exist", async () => { + const { handlers } = setup(); + await expect(handlers["media.proxy"]({ assetId: "missing" })).resolves.toBeUndefined(); + await expect(handlers["media.analyze"]({ assetId: "missing" })).resolves.toBeUndefined(); + await expect(handlers["media.sync"]({ sourceId: "missing" })).resolves.toBeUndefined(); + }); + + it("moves a completed upload into place and starts its proxy", async () => { + const space = await newSpace(); + const actor = { spaceId: space.spaceId, userId: space.userId }; + const device = await repos.createSource(space.spaceId, { + kind: "ios_library", + label: "Phone", + ownerUserId: space.userId, + config: { deviceId: "device-1" }, + }); + const { service, jobs } = setup(); + const pushed = await service.pushItems(actor, { + sourceId: device.id, + items: [ + { + externalId: "IMG_0001", + kind: "video", + capturedAt: new Date(Date.UTC(2026, 0, 2)).toISOString(), + durationMs: 4000, + width: 320, + height: 180, + fps: 25, + gpsLat: null, + gpsLon: null, + deviceModel: null, + albumNames: [], + isFavorite: false, + fileName: "IMG_0001.mp4", + }, + ], + deletedExternalIds: [], + syncToken: "token-1", + complete: true, + }); + expect(pushed).toEqual({ created: 1, updated: 0, unchanged: 0, deleted: 0 }); + const asset = await prisma.mediaAsset.findFirstOrThrow({ where: { spaceId: space.spaceId } }); + expect(asset.ingestState).toBe("awaiting_upload"); + + const uploads = path.join(mediaRoot, space.spaceId, "uploads"); + await mkdir(uploads, { recursive: true }); + const temp = path.join(uploads, "part-1"); + const bytes = Buffer.from("synthetic upload bytes"); + await writeFile(temp, bytes); + const sha256 = createHash("sha256").update(bytes).digest("hex"); + + await expect( + service.uploadCompleted(actor, { + assetId: asset.id, + tempPath: temp, + sha256: "0".repeat(64), + fileName: "IMG_0001.mp4", + }), + ).rejects.toThrow(/hash mismatch/); + const outside = path.join(mediaRoot, "elsewhere"); + await writeFile(outside, bytes); + await expect( + service.uploadCompleted(actor, { + assetId: asset.id, + tempPath: outside, + sha256, + fileName: "IMG_0001.mp4", + }), + ).rejects.toThrow(/upload area/); + await expect( + service.uploadCompleted(actor, { + assetId: asset.id, + tempPath: temp, + sha256, + fileName: "notes.txt", + }), + ).rejects.toThrow(/Unsupported/); + + jobs.reset(); + await service.uploadCompleted(actor, { + assetId: asset.id, + tempPath: temp, + sha256, + fileName: "IMG_0001.mp4", + }); + const stored = await prisma.mediaAsset.findUniqueOrThrow({ where: { id: asset.id } }); + expect(stored.storage).toBe("uploaded"); + expect(stored.ingestState).toBe("indexed"); + expect(stored.contentHash).toBe(sha256); + expect(stored.storagePath).toBe(`${space.spaceId}/assets/${asset.id}/original.mp4`); + expect((await stat(path.join(mediaRoot, stored.storagePath as string))).size).toBe( + bytes.length, + ); + expect(jobs.named("media.proxy").map((job) => job.replaceKey)).toEqual([ + `media.proxy:${asset.id}`, + ]); + await expect(service.requestSync(actor, device.id)).rejects.toThrow(/device/); + }); +}); diff --git a/packages/adapters/src/media-ingest.ts b/packages/adapters/src/media-ingest.ts index 69311c6..be0e439 100644 --- a/packages/adapters/src/media-ingest.ts +++ b/packages/adapters/src/media-ingest.ts @@ -1,53 +1,82 @@ -import { rename } from "node:fs/promises"; +import { lstat, mkdir, rename } from "node:fs/promises"; import path from "node:path"; -import type { MediaJobHandlers, MediaSourceProvider } from "@rakazo/adapter-kit"; -import { - mediaAnalyzeJob, - mediaClusterJob, - mediaProxyJob, - mediaSyncJob, -} from "@rakazo/adapter-kit"; -import type { MediaSourceKind } from "@rakazo/contracts"; +import type { BackgroundJob, MediaJobHandlers, MediaSourceProvider } from "@rakazo/adapter-kit"; +import { mediaAnalyzeJob, mediaClusterJob, mediaProxyJob, mediaSyncJob } from "@rakazo/adapter-kit"; +import type { MediaIngestState, MediaSourceKind } from "@rakazo/contracts"; import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts"; import type { MediaSourceRecord } from "@rakazo/db"; -import { clientMediaItemToSourceItem, InvalidMediaInputError } from "@rakazo/db"; +import { clientMediaItemToSourceItem, InvalidMediaInputError, IsolationError } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; +import type { SessionOptions } from "@rakazo/media"; import { assertInside, resolveInside } from "@rakazo/media/node"; import type { MediaIngestService, MediaServiceDeps } from "./media-context.js"; import type { MediaToolPaths } from "./media-folder-source-probe.js"; -import { analysisInputPath, analyzeAsset, MEDIA_ANALYSIS_VERSION } from "./media-ingest-analyze.js"; +import { + analysisInputPath, + analyzeAsset, + isTransientInferenceError, + MEDIA_ANALYSIS_VERSION, +} from "./media-ingest-analyze.js"; import { clusterSpace } from "./media-ingest-cluster.js"; import { assetDir, deleteMediaPaths, errorMessage, - isDiskFull, + fileExists, moveFile, originalExtension, publishSafe, sha256File, } from "./media-ingest-files.js"; import { buildProxy } from "./media-ingest-proxy.js"; +import type { MediaJobLimits } from "./media-jobs.js"; +import { + assertDiskSpace, + defaultMediaJobLimits, + isDiskFull, + MEDIA_MIN_FREE_BYTES, + MediaDiskFullError, + MediaJobLimiter, +} from "./media-jobs.js"; /* * Media ingest: client pushes, uploads, and the sync → proxy → analyze → * cluster job chain. Jobs carry only record ids; every step re-reads its * record, scopes by the owning space, and is safe to repeat. * - * A full sync walks every page from the start on each run: positional cursors - * (folder paths) cannot express "changed since", and per-item revisions make - * repeated pages cheap. Unchanged items enqueue nothing. Memory stays bounded - * by the page size plus the set of external ids seen. + * Sync walks every page of a server-side source from the start: positional + * cursors (folder paths) cannot express "changed since", and per-item + * revisions make unchanged pages cheap. Unchanged items enqueue nothing. + * + * Proxy and analysis work is pumped, not fanned out: the queue holds at most a + * small window of proxy and analyze jobs per space (the oldest assets waiting + * in `indexed` or `proxied`), and every finished job refills the window. A + * 50k-asset library therefore never floods the shared job queue that chat runs + * also use. Job keys are per asset, so refills replace instead of duplicating. + * + * Idempotency: a proxy is rebuilt only when the original's content hash + * differs from the one it was built from; analysis runs only the kinds that + * have no result at the provider's current model version (explicit kinds + * always re-run). A crashed claim (proxying/analyzing) is taken over after + * MEDIA_STALE_CLAIM_MS. */ export const MEDIA_SYNC_PAGE = 500; /** Clustering waits for a quiet period so a burst of ingest work clusters once. */ export const MEDIA_CLUSTER_DEBOUNCE_MS = 30_000; +/** A job that found no free permit runs again after this delay. */ +export const MEDIA_BUSY_RETRY_MS = 15_000; +/** A job paused by a full disk or an unreachable sidecar retries after this delay. */ +export const MEDIA_PAUSE_RETRY_MS = 5 * 60_000; +/** An in-flight claim older than this belongs to a crashed worker. */ +export const MEDIA_STALE_CLAIM_MS = 30 * 60_000; +const DISK_PAUSED = "Media processing is paused because the disk is full"; const SERVER_SOURCES = ["folder", "emulator"] as const; const REMOVE_BATCH = 500; type ServerSourceKind = (typeof SERVER_SOURCES)[number]; +type AssetRecord = Awaited>; function isServerSource(kind: MediaSourceKind): kind is ServerSourceKind { return (SERVER_SOURCES as readonly string[]).includes(kind); @@ -95,6 +124,101 @@ async function removeItems( return deleted; } +/** + * Content changed under an asset: analyses, segments, and face tracks describe + * the old bytes and are dropped so the next analysis starts clean. + */ +async function dropStaleAnalysis( + deps: MediaServiceDeps, + spaceId: string, + assetId: string, +): Promise { + await deps.repos.replaceFaceTracks(spaceId, assetId, "stale", []); + await deps.repos.replaceSegments(spaceId, assetId, 0, []); + await deps.prisma.mediaAnalysis.deleteMany({ where: { spaceId, assetId } }); + await deleteMediaPaths(deps.mediaRoot, [`${assetDir(spaceId, assetId)}/faces`]); +} + +/** Assets waiting in `waiting`, or stuck in `inFlight` by a worker that died. */ +function waitingWhere( + deps: MediaServiceDeps, + spaceId: string, + waiting: MediaIngestState, + inFlight: MediaIngestState, +) { + const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS); + const storage = { in: ["referenced", "uploaded"] }; + return { + spaceId, + storage, + OR: [{ ingestState: waiting }, { ingestState: inFlight, updatedAt: { lt: staleBefore } }], + }; +} + +/** Enqueue proxy jobs for the oldest assets waiting for one. */ +async function refillProxy(deps: MediaServiceDeps, spaceId: string, window: number) { + const rows = await deps.prisma.mediaAsset.findMany({ + where: waitingWhere(deps, spaceId, "indexed", "proxying"), + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + take: window, + select: { id: true }, + }); + for (const row of rows) await deps.jobs.enqueue(mediaProxyJob(row.id)); +} + +/** Enqueue analyze jobs for the oldest assets waiting for one. */ +async function refillAnalyze(deps: MediaServiceDeps, spaceId: string, window: number) { + if (!deps.inference) return; + const rows = await deps.prisma.mediaAsset.findMany({ + where: waitingWhere(deps, spaceId, "proxied", "analyzing"), + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: window, + select: { id: true }, + }); + for (const row of rows) await deps.jobs.enqueue(mediaAnalyzeJob(row.id)); +} + +/** + * Restart the proxy and analysis pumps of every space with waiting or stranded + * work. The worker calls this on start (after a crash or a sidecar was added) + * and periodically; it only enqueues jobs, so repeating it is harmless. + */ +export async function resumeMediaIngest( + deps: MediaServiceDeps, + limits: MediaJobLimits = defaultMediaJobLimits(), +): Promise { + const states: MediaIngestState[] = deps.inference + ? ["indexed", "proxying", "proxied", "analyzing"] + : ["indexed", "proxying"]; + const spaces = await deps.prisma.mediaAsset.findMany({ + where: { ingestState: { in: states }, storage: { in: ["referenced", "uploaded"] } }, + distinct: ["spaceId"], + select: { spaceId: true }, + }); + for (const { spaceId } of spaces) { + await refillProxy(deps, spaceId, limits.cpu * 2); + await refillAnalyze(deps, spaceId, limits.gpu * 2); + } + return spaces.length; +} + +/** Schedule clustering once the space has no proxy or analysis work left. */ +async function clusterWhenDrained(deps: MediaServiceDeps, spaceId: string): Promise { + const pending: MediaIngestState[] = deps.inference + ? ["proxying", "proxied", "analyzing"] + : ["proxying"]; + const busy = await deps.prisma.mediaAsset.count({ + where: { + spaceId, + OR: [ + { ingestState: { in: pending } }, + { ingestState: "indexed", storage: { in: ["referenced", "uploaded"] } }, + ], + }, + }); + if (busy === 0) await scheduleCluster(deps, spaceId); +} + export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestService { return { async pushItems(actor, input) { @@ -162,6 +286,7 @@ export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestSer if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) { throw new InvalidMediaInputError("Upload is outside this space's upload area"); } + if (!(await lstat(temp)).isFile()) throw new InvalidMediaInputError("Upload is not a file"); const actual = await sha256File(temp); if (actual !== input.sha256) throw new InvalidMediaInputError("Content hash mismatch"); @@ -197,6 +322,9 @@ export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestSer if (asset.storagePath && asset.storagePath !== storagePath) { await deleteMediaPaths(deps.mediaRoot, [asset.storagePath]); } + if (asset.contentHash && asset.contentHash !== actual) { + await dropStaleAnalysis(deps, actor.spaceId, asset.id); + } await publishSafe(deps, actor.spaceId, { type: "media.asset", assetId: asset.id, @@ -226,12 +354,19 @@ export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestSer export interface MediaIngestHandlerOptions extends MediaToolPaths { /** Aborts long-running ingest work when the worker shuts down. */ signal?: AbortSignal; + /** Per-process job limits; defaults to defaultMediaJobLimits(). */ + limits?: MediaJobLimits; + /** Free bytes the media root keeps in reserve; jobs pause below it. 0 disables the check. */ + minFreeBytes?: number; + /** Session rules (gap minutes and so on); the time zone defaults to the library owner's. */ + sessionOptions?: Partial; } async function syncSource( deps: MediaServiceDeps, source: MediaSourceRecord, provider: MediaSourceProvider, + onChanged: () => Promise, signal?: AbortSignal, ): Promise<{ changed: number; deleted: number }> { const seen = new Set(); @@ -247,8 +382,20 @@ async function syncSource( }); for (const item of page.items) seen.add(item.externalId); const result = await deps.repos.upsertSourceItems(source.spaceId, source.id, page.items); - for (const assetId of result.changedAssetIds) await deps.jobs.enqueue(mediaProxyJob(assetId)); - changed += result.changedAssetIds.length; + if (result.changedAssetIds.length > 0) { + // A changed revision means new bytes: the asset goes back through the pipeline. + await deps.prisma.mediaAsset.updateMany({ + where: { + spaceId: source.spaceId, + id: { in: result.changedAssetIds }, + storage: "referenced", + ingestState: { not: "indexed" }, + }, + data: { ingestState: "indexed", ingestError: null }, + }); + changed += result.changedAssetIds.length; + await onChanged(); + } cursor = page.nextCursor; } while (cursor); @@ -271,40 +418,174 @@ export function createMediaIngestHandlers( options: MediaIngestHandlerOptions = {}, ): Pick { const { signal } = options; + const limiter = new MediaJobLimiter(options.limits); + const proxyWindow = limiter.limits.cpu * 2; + const analyzeWindow = limiter.limits.gpu * 2; + const minFreeBytes = options.minFreeBytes ?? MEDIA_MIN_FREE_BYTES; const tools: MediaToolPaths = { ...(options.ffmpegPath ? { ffmpegPath: options.ffmpegPath } : {}), ...(options.ffprobePath ? { ffprobePath: options.ffprobePath } : {}), }; - const setAssetState = async ( + const later = async (job: BackgroundJob, delayMs: number) => { + await deps.jobs.enqueue({ ...job, availableAt: new Date(deps.now().getTime() + delayMs) }); + }; + + const checkDisk = async () => { + await mkdir(deps.mediaRoot, { recursive: true }); + await assertDiskSpace(deps.mediaRoot, minFreeBytes); + }; + + /** Pause on a full disk: tell the source's owner why and try again later. */ + const pauseForDisk = async (spaceId: string, sourceId: string, job: BackgroundJob) => { + getLogger().warn("media job paused: disk is full", { job: job.name }); + const source = await deps.repos + .updateSourceSync(spaceId, sourceId, { error: DISK_PAUSED }) + .catch(() => null); + if (source) { + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: source.status, + error: DISK_PAUSED, + }); + } + await later(job, MEDIA_PAUSE_RETRY_MS); + }; + + const resumeAfterDisk = async (spaceId: string, sourceId: string) => { + const result = await deps.prisma.mediaSource.updateMany({ + where: { id: sourceId, spaceId, error: DISK_PAUSED }, + data: { error: null }, + }); + if (result.count === 0) return; + const source = await deps.repos.getSource(spaceId, sourceId).catch(() => null); + if (source) { + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: source.status, + error: null, + }); + } + }; + + /** + * Claim an asset for a step: from one of `from`, or from `inFlight` when that + * claim is stale (its worker died). Only one worker wins. + */ + const claim = async ( spaceId: string, assetId: string, - from: Parameters[2]["from"], - to: Parameters[2]["to"], + from: MediaIngestState[], + inFlight: MediaIngestState, + ): Promise => { + const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS); + const result = await deps.prisma.mediaAsset.updateMany({ + where: { + id: assetId, + spaceId, + OR: [ + { ingestState: { in: from } }, + { ingestState: inFlight, updatedAt: { lt: staleBefore } }, + ], + }, + data: { ingestState: inFlight, ingestError: null }, + }); + if (result.count !== 1) return false; + await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: inFlight }); + return true; + }; + + const setState = async ( + spaceId: string, + assetId: string, + from: MediaIngestState, + to: MediaIngestState, error?: string, ) => { - const applied = await deps.repos.transitionIngestState(spaceId, assetId, { - from, - to, - ...(error !== undefined ? { error } : {}), - }); - if (applied) await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: to }); + const applied = await deps.repos + .transitionIngestState(spaceId, assetId, { + from: [from], + to, + ...(error !== undefined ? { error } : {}), + }) + .catch((cause) => { + if (cause instanceof IsolationError) return false; + throw cause; + }); + if (applied) + await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: to }); return applied; }; /** Absolute path of the original, or null for a device asset that has not been uploaded. */ - const originalOf = async ( - spaceId: string, - asset: Awaited>, - ): Promise<{ path: string; referenced: boolean } | null> => { + const originalOf = async (spaceId: string, asset: AssetRecord): Promise => { if (asset.storage === "uploaded" && asset.storagePath) { - return { path: await resolveInside(deps.mediaRoot, asset.storagePath), referenced: false }; + return resolveInside(deps.mediaRoot, asset.storagePath); } if (asset.storage !== "referenced") return null; const source = await deps.repos.getSource(spaceId, asset.sourceId); const provider = isServerSource(source.kind) ? deps.sources[source.kind] : undefined; if (!provider) throw new Error(`No ${source.kind} source provider is configured`); - return { path: await provider.resolve(source.config, asset.externalId), referenced: true }; + return provider.resolve(source.config, asset.externalId); + }; + + /** Load the asset a job names; null when it (or its space) is gone. */ + const loadAsset = async (assetId: string) => { + const spaceId = await deps.repos.mediaSpaceOf("asset", assetId); + if (!spaceId) return null; + try { + return { spaceId, asset: await deps.repos.getAssetRecord(spaceId, assetId) }; + } catch (error) { + if (error instanceof IsolationError) return null; + throw error; + } + }; + + const proxyAsset = async (spaceId: string, asset: AssetRecord): Promise => { + const original = await originalOf(spaceId, asset); + if (!original) throw new Error("The original has not been uploaded"); + const contentHash = await sha256File(original, signal); + const unchanged = + contentHash === asset.contentHash && + asset.proxyPath !== null && + asset.thumbPath !== null && + (await fileExists(deps.mediaRoot, asset.proxyPath)) && + (await fileExists(deps.mediaRoot, asset.thumbPath)); + if (unchanged) { + await deps.repos.recordProxy(spaceId, asset.id, { + contentHash, + proxyPath: asset.proxyPath, + thumbPath: asset.thumbPath, + }); + return; + } + if (asset.contentHash && asset.contentHash !== contentHash) { + await dropStaleAnalysis(deps, spaceId, asset.id); + } + const outputs = await buildProxy({ + mediaRoot: deps.mediaRoot, + spaceId, + assetId: asset.id, + original, + still: asset.kind === "photo", + tools, + ...(signal ? { signal } : {}), + }); + const still = outputs.spritePath === null; + await deps.repos.recordProxy(spaceId, asset.id, { + contentHash, + proxyPath: outputs.proxyPath, + thumbPath: outputs.thumbPath, + spritePath: outputs.spritePath, + durationMs: still ? asset.durationMs : outputs.probe.durationMs, + width: outputs.probe.width, + height: outputs.probe.height, + fps: still ? asset.fps : outputs.probe.fps, + codec: outputs.probe.codec, + orientation: outputs.probe.rotation, + }); }; return { @@ -325,21 +606,46 @@ export function createMediaIngestHandlers( return; } await deps.repos.updateSourceSync(spaceId, sourceId, { status: "syncing", error: null }); - await publishSafe(deps, spaceId, { type: "media.source", sourceId, status: "syncing" }); + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: "syncing", + error: null, + }); try { - const { changed, deleted } = await syncSource(deps, source, provider, signal); + const { changed, deleted } = await syncSource( + deps, + source, + provider, + () => refillProxy(deps, spaceId, proxyWindow), + signal, + ); await deps.repos.updateSourceSync(spaceId, sourceId, { status: "idle", lastSyncedAt: deps.now(), error: null, }); - await publishSafe(deps, spaceId, { type: "media.source", sourceId, status: "idle" }); + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: "idle", + error: null, + }); if (changed + deleted > 0) await scheduleCluster(deps, spaceId); } catch (error) { - if (signal?.aborted) throw error; + if (signal?.aborted) { + await deps.repos + .updateSourceSync(spaceId, sourceId, { status: "idle" }) + .catch(() => undefined); + throw error; + } + // The source was removed while it synced. + if (error instanceof IsolationError) return; const message = errorMessage(error); getLogger().error("media sync failed", error, { sourceId }); - await deps.repos.updateSourceSync(spaceId, sourceId, { status: "error", error: message }); + await deps.repos + .updateSourceSync(spaceId, sourceId, { status: "error", error: message }) + .catch(() => undefined); await publishSafe(deps, spaceId, { type: "media.source", sourceId, @@ -350,103 +656,155 @@ export function createMediaIngestHandlers( }, async "media.proxy"({ assetId }) { - const spaceId = await deps.repos.mediaSpaceOf("asset", assetId); - if (!spaceId) return; - const asset = await deps.repos.getAssetRecord(spaceId, assetId); - if (asset.storage === "remote") return; - const started = await setAssetState( - spaceId, - assetId, - ["indexed", "proxying", "proxied", "analyzing", "analyzed", "failed"], - "proxying", - ); - if (!started) return; - try { - const original = await originalOf(spaceId, asset); - if (!original) throw new Error("The original has not been uploaded"); - const contentHash = await sha256File(original.path, signal); - const outputs = await buildProxy({ - mediaRoot: deps.mediaRoot, - spaceId, - assetId, - original: original.path, - still: asset.kind === "photo", - tools, - ...(signal ? { signal } : {}), - }); - const still = outputs.spritePath === null; - await deps.repos.recordProxy(spaceId, assetId, { - contentHash, - proxyPath: outputs.proxyPath, - thumbPath: outputs.thumbPath, - spritePath: outputs.spritePath, - durationMs: still ? asset.durationMs : outputs.probe.durationMs, - width: outputs.probe.width, - height: outputs.probe.height, - fps: still ? asset.fps : outputs.probe.fps, - codec: outputs.probe.codec, - orientation: outputs.probe.rotation, - }); - await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: "proxied" }); - await deps.jobs.enqueue(mediaAnalyzeJob(assetId)); - } catch (error) { - if (signal?.aborted) throw error; - getLogger().error("media proxy failed", error, { assetId }); - if (isDiskFull(error)) { - await setAssetState(spaceId, assetId, ["proxying"], "failed", "Disk is full; retrying"); - throw error; - } - await setAssetState(spaceId, assetId, ["proxying"], "failed", errorMessage(error)); + const loaded = await loadAsset(assetId); + if (!loaded || loaded.asset.storage === "remote") return; + const { spaceId, asset } = loaded; + const job = mediaProxyJob(assetId); + const release = limiter.tryAcquire("cpu"); + if (!release) { + await later(job, MEDIA_BUSY_RETRY_MS); + return; } + try { + try { + await checkDisk(); + } catch (error) { + if (!(error instanceof MediaDiskFullError)) throw error; + await pauseForDisk(spaceId, asset.sourceId, job); + return; + } + if (!(await claim(spaceId, assetId, ["indexed", "failed"], "proxying"))) return; + try { + await proxyAsset(spaceId, asset); + await publishSafe(deps, spaceId, { + type: "media.asset", + assetId, + ingestState: "proxied", + }); + await resumeAfterDisk(spaceId, asset.sourceId); + } catch (error) { + if (signal?.aborted) { + await setState(spaceId, assetId, "proxying", "indexed"); + throw error; + } + if (error instanceof IsolationError) { + // Deleted while it was processed: drop whatever was written for it. + await deleteMediaPaths(deps.mediaRoot, [assetDir(spaceId, assetId)]); + return; + } + if (isDiskFull(error)) { + await setState(spaceId, assetId, "proxying", "indexed"); + await pauseForDisk(spaceId, asset.sourceId, job); + return; + } + getLogger().error("media proxy failed", error, { assetId }); + await setState(spaceId, assetId, "proxying", "failed", errorMessage(error)); + } + } finally { + release(); + } + await refillProxy(deps, spaceId, proxyWindow); + await refillAnalyze(deps, spaceId, analyzeWindow); + await clusterWhenDrained(deps, spaceId); }, async "media.analyze"({ assetId, kinds }) { const inference = deps.inference; if (!inference) return; - const spaceId = await deps.repos.mediaSpaceOf("asset", assetId); - if (!spaceId) return; - const asset = await deps.repos.getAssetRecord(spaceId, assetId); - if (!asset.proxyPath) return; - const started = await setAssetState( - spaceId, - assetId, - ["proxied", "analyzing", "analyzed", "failed"], - "analyzing", - ); - if (!started) return; - try { - const original = asset.storage === "referenced" ? await originalOf(spaceId, asset) : null; - const inputPath = await analysisInputPath({ - mediaRoot: deps.mediaRoot, - proxyPath: asset.proxyPath, - referencedOriginal: original?.path ?? null, - }); - const outcome = await analyzeAsset({ - deps, - inference, - spaceId, - asset, - inputPath, - ...(kinds ? { kinds } : {}), - }); - await deps.repos.recordAnalysisVersion( - spaceId, - assetId, - MEDIA_ANALYSIS_VERSION, - outcome.qualityScore, - ); - await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: "analyzed" }); - await scheduleCluster(deps, spaceId); - } catch (error) { - if (signal?.aborted) throw error; - getLogger().error("media analysis failed", error, { assetId }); - await setAssetState(spaceId, assetId, ["analyzing"], "failed", errorMessage(error)); - if (isDiskFull(error)) throw error; + const loaded = await loadAsset(assetId); + if (!loaded?.asset.proxyPath) return; + const { spaceId, asset } = loaded; + // A repeated default request for an asset already at the current pipeline version. + if (!kinds && asset.ingestState === "analyzed") { + if (asset.analysisVersion >= MEDIA_ANALYSIS_VERSION) return; } + const job = mediaAnalyzeJob(assetId, kinds); + const release = limiter.tryAcquire("gpu"); + if (!release) { + await later(job, MEDIA_BUSY_RETRY_MS); + return; + } + try { + try { + await checkDisk(); + } catch (error) { + if (!(error instanceof MediaDiskFullError)) throw error; + await pauseForDisk(spaceId, asset.sourceId, job); + return; + } + if (!(await claim(spaceId, assetId, ["proxied", "analyzed", "failed"], "analyzing"))) { + return; + } + try { + const original = asset.storage === "referenced" ? await originalOf(spaceId, asset) : null; + const inputPath = await analysisInputPath({ + mediaRoot: deps.mediaRoot, + proxyPath: asset.proxyPath as string, + referencedOriginal: original, + }); + const outcome = await analyzeAsset({ + deps, + inference, + spaceId, + asset, + inputPath, + ...(kinds ? { kinds } : {}), + ...(signal ? { signal } : {}), + }); + await deps.repos.recordAnalysisVersion( + spaceId, + assetId, + MEDIA_ANALYSIS_VERSION, + outcome.qualityScore, + ); + await publishSafe(deps, spaceId, { + type: "media.asset", + assetId, + ingestState: "analyzed", + }); + await resumeAfterDisk(spaceId, asset.sourceId); + } catch (error) { + if (signal?.aborted) { + await setState(spaceId, assetId, "analyzing", "proxied"); + throw error; + } + if (error instanceof IsolationError) return; + if (isDiskFull(error)) { + await setState(spaceId, assetId, "analyzing", "proxied"); + await pauseForDisk(spaceId, asset.sourceId, job); + return; + } + if (isTransientInferenceError(error)) { + // Sidecar down or saturated: not the asset's fault, try again later. + getLogger().warn("media analysis deferred", { assetId, reason: errorMessage(error) }); + await setState(spaceId, assetId, "analyzing", "proxied"); + await later(job, MEDIA_PAUSE_RETRY_MS); + return; + } + getLogger().error("media analysis failed", error, { assetId }); + await setState(spaceId, assetId, "analyzing", "failed", errorMessage(error)); + } + } finally { + release(); + } + await refillAnalyze(deps, spaceId, analyzeWindow); + await clusterWhenDrained(deps, spaceId); }, async "media.cluster"({ spaceId }) { - await clusterSpace(deps, spaceId, signal ? { signal } : {}); + const release = limiter.tryAcquire("cpu"); + if (!release) { + await later(mediaClusterJob(spaceId), MEDIA_BUSY_RETRY_MS); + return; + } + try { + await clusterSpace(deps, spaceId, { + ...(signal ? { signal } : {}), + ...(options.sessionOptions ? { sessionOptions: options.sessionOptions } : {}), + }); + } finally { + release(); + } }, }; } diff --git a/packages/adapters/src/media-jobs.test.ts b/packages/adapters/src/media-jobs.test.ts new file mode 100644 index 0000000..448e8d2 --- /dev/null +++ b/packages/adapters/src/media-jobs.test.ts @@ -0,0 +1,58 @@ +import { tmpdir } from "node:os"; +import { describe, expect, it } from "vitest"; +import { + assertDiskSpace, + defaultMediaJobLimits, + freeBytes, + isDiskFull, + MediaDiskFullError, + MediaJobLimiter, +} from "./media-jobs.js"; + +describe("media job limits", () => { + it("defaults to one GPU job and a bounded number of CPU jobs", () => { + expect(defaultMediaJobLimits(1)).toEqual({ gpu: 1, cpu: 1 }); + expect(defaultMediaJobLimits(4)).toEqual({ gpu: 1, cpu: 2 }); + expect(defaultMediaJobLimits(64)).toEqual({ gpu: 1, cpu: 2 }); + }); + + it("hands out permits up to the limit per class and never waits", () => { + const limiter = new MediaJobLimiter({ gpu: 1, cpu: 2 }); + const gpu = limiter.tryAcquire("gpu"); + expect(gpu).not.toBeNull(); + expect(limiter.tryAcquire("gpu")).toBeNull(); + const cpu = [limiter.tryAcquire("cpu"), limiter.tryAcquire("cpu")]; + expect(cpu.every(Boolean)).toBe(true); + expect(limiter.tryAcquire("cpu")).toBeNull(); + + gpu?.(); + gpu?.(); + expect(limiter.running("gpu")).toBe(0); + expect(limiter.tryAcquire("gpu")).not.toBeNull(); + cpu[0]?.(); + expect(limiter.running("cpu")).toBe(1); + }); + + it("rejects limits below one", () => { + expect(() => new MediaJobLimiter({ gpu: 0, cpu: 1 })).toThrow(); + expect(() => new MediaJobLimiter({ gpu: 1, cpu: 1.5 })).toThrow(); + }); +}); + +describe("disk backpressure", () => { + it("reports free space and pauses below the reserve", async () => { + const dir = tmpdir(); + expect(await freeBytes(dir)).toBeGreaterThan(0); + await expect(assertDiskSpace(dir, 0)).resolves.toBeUndefined(); + await expect(assertDiskSpace(dir, Number.MAX_SAFE_INTEGER)).rejects.toBeInstanceOf( + MediaDiskFullError, + ); + }); + + it("recognizes a full disk from errno or ffmpeg output", () => { + expect(isDiskFull(Object.assign(new Error("write"), { code: "ENOSPC" }))).toBe(true); + expect(isDiskFull(new Error("ffmpeg failed: No space left on device"))).toBe(true); + expect(isDiskFull(new Error("Invalid data found"))).toBe(false); + expect(isDiskFull(null)).toBe(false); + }); +}); diff --git a/packages/adapters/src/media-jobs.ts b/packages/adapters/src/media-jobs.ts new file mode 100644 index 0000000..2ebcbfb --- /dev/null +++ b/packages/adapters/src/media-jobs.ts @@ -0,0 +1,85 @@ +import { statfs } from "node:fs/promises"; +import { availableParallelism } from "node:os"; + +/* + * Resource limits for media jobs. + * + * Every job slot of the shared job worker also serves chat runs, so a media job + * never waits for capacity inside a slot: it takes a permit or steps aside + * (the caller re-enqueues it with a delay). Analysis is the GPU class + * (sidecar models), proxies and clustering are the CPU class. The limits are + * per worker process. + * + * Disk backpressure: before writing derived files a job checks the free space + * of the media root and pauses (re-enqueues itself later) when it falls under + * the reserve, instead of failing assets or crashing the worker. + */ + +export type MediaJobClass = "gpu" | "cpu"; + +export interface MediaJobLimits { + /** Concurrent analysis jobs. Default 1. */ + gpu: number; + /** Concurrent proxy and cluster jobs. Default: half the cores, at most 2. */ + cpu: number; +} + +export function defaultMediaJobLimits(cores = availableParallelism()): MediaJobLimits { + return { gpu: 1, cpu: Math.max(1, Math.min(2, Math.floor(cores / 2))) }; +} + +/** Non-blocking counting semaphore. */ +export class MediaJobLimiter { + private readonly active: Record = { gpu: 0, cpu: 0 }; + + constructor(readonly limits: MediaJobLimits = defaultMediaJobLimits()) { + for (const value of Object.values(limits)) { + if (!Number.isInteger(value) || value < 1) throw new Error("media job limits must be ≥ 1"); + } + } + + /** A release function when a permit was free, otherwise null. Release is idempotent. */ + tryAcquire(kind: MediaJobClass): (() => void) | null { + if (this.active[kind] >= this.limits[kind]) return null; + this.active[kind] += 1; + let released = false; + return () => { + if (released) return; + released = true; + this.active[kind] -= 1; + }; + } + + running(kind: MediaJobClass): number { + return this.active[kind]; + } +} + +/** Free space under the media root reserve: 2 GiB. */ +export const MEDIA_MIN_FREE_BYTES = 2 * 1024 ** 3; + +/** Bytes available to this process on the file system holding `dir`. */ +export async function freeBytes(dir: string): Promise { + const stats = await statfs(dir); + return Number(stats.bavail) * Number(stats.bsize); +} + +/** A full disk is retried later instead of failing the asset for good. */ +export function isDiskFull(error: unknown): boolean { + if ((error as NodeJS.ErrnoException | null)?.code === "ENOSPC") return true; + return error instanceof Error && /No space left on device/i.test(error.message); +} + +export class MediaDiskFullError extends Error { + constructor(readonly availableBytes: number | null) { + super("Media processing is paused because the disk is full"); + this.name = "MediaDiskFullError"; + } +} + +/** Throws MediaDiskFullError when less than `reserve` bytes are free under `dir`. */ +export async function assertDiskSpace(dir: string, reserve: number): Promise { + if (reserve <= 0) return; + const available = await freeBytes(dir); + if (available < reserve) throw new MediaDiskFullError(available); +}