openvibers/openvibe.media Commits
Media: a local timeline-cache budget (eviction of verified durable copies, backfill pacing) before HLS goes on (#94)
Changes
@@@ -143,6 +143,12 @@ MEDIA_CAPABILITY_GATE=report# /o/:id/master.m3u8, /o/:id/source/… and the hls_url field of GET …/download?format=json ; off (default):# the job is refused and the routes 404, nothing else changes.#MEDIA_HLS_ENABLED=0++# Maximum bytes of local timeline segments and packed chunks (default 10 GiB; 0 keeps only durable copies).++#MEDIA_TIMELINE_LOCAL_MAX_BYTES=10737418240++# Stop keeping new local timeline copies below this free space (default 8 GiB).++#MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES=8589934592++# At most this many old VODs are lazily queued per hour; recording finalize is exempt (default 20).++#MEDIA_HLS_BACKFILL_PER_HOUR=20# Lifetime in seconds of a signed HLS playlist (download?format=json → hls_url of a private or sandbox object; its own# signature purpose, accepted only by the HLS routes). Default 21600 (6 h), clamped to 60..43200.#MEDIA_HLS_PLAYLIST_TTL_S=21600nnew file mode 100644@@@ -0,0 +1,9 @@++# Changelog++++## Unreleased++++- Bound local CMAF and packed timeline copies with a configurable byte budget and free-space floor. The hourly sweep++ evicts least recently served copies only after durable size and SHA-256 verification, preserving all timeline rows++ and durable playback.++- Limit lazy HLS backfill to 20 VODs per hour by default while recording finalize remains immediate.++- Add timeline cache readiness and metrics, environment documentation and production enablement guidance.@@@ -446,6 +446,20 @@ that reaches port 4100 directly cannot choose its IP by sending `X-Forwarded-For`CF-Connecting-IP` itself. View counting and the live-frame rate limit read `req.ip`,never the headers (`test/trust-proxy.test.js`).++## Segment-native video settings++++| Environment variable | Default | Purpose |++|---|---:|---|++| `MEDIA_HLS_ENABLED` | `0` | Enable timeline generation and HLS routes. |++| `MEDIA_TIMELINE_LOCAL_MAX_BYTES` | `10737418240` | Maximum local timeline cache (10 GiB; `0` keeps no durable copies locally). |++| `MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES` | `8589934592` | Below this free disk level, new verified timeline bytes remain durable only. |++| `MEDIA_HLS_BACKFILL_PER_HOUR` | `20` | Maximum lazy `object.cmaf` queues for older VODs per rolling hour on a single Media host; finalize is exempt. |++| `MEDIA_JOBS_HEAVY_CONCURRENCY` | `1` | Keep at 1 while enabling HLS in production. |++++The hourly sweep evicts the least recently served verified local timeline files. A segment whose local copy was++evicted is served from its durable provider. Watch `timeline_cache` on `/api/ready` and job durations during backfill++(docs/media-fabric.md §3).++## Per-actor limitsnginx limits by address; the capability routes also limit by who calls (`server/actor-limits.js`,@@@ -506,8 +520,8 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.| path | behavior ||---|---|| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |--| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks` (user JWTs and service tokens; app keys work without it), `remote_b2` / `remote_r2` (HeadBucket, at most once a minute; present only when the provider is configured), `events_outbox` (backlog over 1000 events). Also `recordings_in_progress`. |--| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |++| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress`. |++| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_timeline_local_bytes`, `media_timeline_local_evictions`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). || `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. || `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |@@@ -41,11 +41,12 @@"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs","every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling","segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",++ "local timeline cache (F3): 10 GiB default budget and 8 GiB free-disk floor, least-recently-served eviction of verified durable copies in the hourly sweep, durable segment fallback, optional timeline_cache readiness and local bytes/evictions metrics; lazy CMAF backfill capped at 20 per hour while recording finalize is exempt","packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)","on-demand renditions (F4 slice 1, behind MEDIA_HLS_ENABLED + MEDIA_RENDITIONS): the rendition.create job transcodes one 720p H.264/AAC rung of a ready media object into media_timeline through the same content-addressed rows, placement-router durable provider and two-phase commit as object.cmaf, forcing the source timeline's segment boundaries as keyframes; a source already 720p or smaller is skipped and a held object is left alone. The master playlist lists a rendition only once its rows exist and queues the missing ones lazily from /o/:id/master.m3u8 and download?format=json (dedupeActive + an object/rendition idempotency key), and /o/:id/<rendition>/index.m3u8 plus /o/:id/<rendition>/<name> serve it","timeline sprites (F3.1, behind MEDIA_HLS_ENABLED): object.sprite cuts its sheet from the object's media_timeline rows — each sample time picks the row covering it and one frame is decoded from that row's bytes (this node's segment, or a packed row's byte range in its chunk), checked against the row's sha256 — with the same layout and metadata as the seek path, so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame","HLS discovery on the v1 shapes (F3.c, behind MEDIA_HLS_ENABLED): the VOD and clip answers (vodPublic/clipPublic — get, list, create, update) carry hls_url (and hls_expires_at when the object is private or sandbox), the same field and access rules as GET /o/:id/download?format=json, read through one shared discovery (server/objects/hls.js); an object without a timeline gets object.cmaf queued and the field appears once it ran, and a virtual clip plays its source's window without being cut a timeline of its own","materialized clips (F3.6, behind MEDIA_HLS_ENABLED + MEDIA_MATERIALIZED_CLIPS, off by default): clip.cut over a source timeline writes the clip's own media_timeline rows — the interior copies the source's segments (packed chunk ranges included), only the two window edges are re-encoded into fMP4 with their own inits (init-head.mp4/init-tail.mp4, negative seq, EXT-X-MAP + EXT-X-DISCONTINUITY in the playlist) — inserting them in one transaction that locks the named source rows FOR SHARE and aborts with a retryable error when one is gone or any location column moved (a pack), so a racing delete either waits and keeps the bytes or aborts the insert; the projection keeps virtual: true and gains materialized: true (readiness, playableSql and the verify job treat it as a virtual clip), sprites decode each edge with its own init, deleting the clip removes its rows and edge bytes while the source's bytes stay, and download?format=mp4 answers the playlist for a timeline-only clip instead of queueing a remux that cannot read its input"],-- "updated": "2026-10-11"++ "updated": "2026-10-12"}@@@ -98,7 +98,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}(`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,`params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it) into `init.mp4` +`000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured,-- under the key `<object>/source/<version>/<name>` on B2 (`durable` once B2 confirms the size), then writes all rows of++ under the key `<object>/source/<version>/<name>` on B2 (`durable` after a size and sha256 readback), then writes all rows ofthe rendition in one transaction and only then deletes the previous cut's bytes. It never touches the source object; arerun reuses rows whose sha256 matches and their location as they are and uploads only what is not durable yet; ithonours the abort signal (which stops an upload in flight) and, before the rows commit, removes the files and keys it@@@ -121,7 +121,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}before; runs of one segment and the init segment stay as they are; a segment not durable yet ends a run). Eachsegment's bytes are checked against its row's sha256 (this node's copy, else the durable one through the router); thechunk goes to the router's durable provider under `<object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s` and is-- verified there (sha256 read back), this node keeps a copy where it kept the segments, and the rows move to it in one++ verified there (size and sha256 read back), this node may keep a cache copy where it kept the segments, and the rows move to it in onetransaction: `key`/`local_path`/`durable_provider` name the chunk, `packed_object_id` is its sha256 (content address),`byte_offset`/`byte_length` the segment inside it; name, times and sha256 stay the segment's. Only after the commit arethe per-segment files and keys **deleted**: the chunk holds the same verified bytes, keeping both would double the@@@ -168,6 +168,26 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux(`dedupeActive`) and answers `202 { job_id }` while it is missing.- **F3.5, shipped: virtual clips over the timeline (§4).**++- **Local timeline cache, shipped:** `OBJECTS_PATH/.timeline` is bounded by++ `MEDIA_TIMELINE_LOCAL_MAX_BYTES` (10 GiB by default; 0 keeps no new local copies once durable) and++ `MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES` (8 GiB by default). CMAF and packing stage their work as before; once the++ durable upload passes size and sha256 readback, they keep no new local copy if the cache is full or free disk is++ below the floor. The hourly object sweep evicts least recently served local files until both limits are satisfied.++ Serving a local segment updates its file mtime, a portable recency signal that survives restarts without a database++ write per request. The sweep verifies the durable bytes again, including older size-only timeline uploads, and clears++ every row sharing that local path in its delete transaction. A file with an unverified or missing durable copy remains;++ no timeline row is deleted. The route then uses its existing durable fallback, including packed byte ranges. The++ storage orphan report still counts only row-named timeline files; an unlinked cache file and its cleared path are++ absent from both sides. `timeline_cache` is optional on `/api/ready` and degrades above budget or below the free++ floor; `/metrics` exposes `media_timeline_local_bytes` and `media_timeline_local_evictions`.++++ Lazy discovery of an older VOD without a timeline is capped at `MEDIA_HLS_BACKFILL_PER_HOUR` (20 per rolling hour,++ with job history surviving a restart on a single Media host); a capped read simply has no `hls_url` and its MP4/file playback remains available. A recording's++ own finalize bypasses the cap. List reads discover timelines in batches and do not queue missing ones.++++ **Turning HLS on in production:** set the cache budget and free-space floor for the host, set++ `MEDIA_HLS_ENABLED=1`, keep `MEDIA_JOBS_HEAVY_CONCURRENCY=1`, and watch `timeline_cache`, local timeline bytes,++ evictions, and `object.cmaf`/`object.pack` job durations during backfill.- **F3.6, shipped (behind `MEDIA_HLS_ENABLED` + `MEDIA_MATERIALIZED_CLIPS`, off by default): materialized clips overthe source's bytes (§4, docs/materialized-clips.md).** A clip whose source has a timeline gets its own persisted`media_timeline` rows: the interior segments are the source's rows (same location, packed chunk range and sha256,@@@ -93,6 +93,9 @@ const config = {// media_timeline (/o/:id/master.m3u8, /o/:id/source/…). Off by default: off, the job is refused and the routes 404.hls: {enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),++ localMaxBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MAX_BYTES', 10 * 1024 ** 3)),++ localMinFreeBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES', 8 * 1024 ** 3)),++ backfillPerHour: Math.max(0, intEnv('MEDIA_HLS_BACKFILL_PER_HOUR', 20)),// Lifetime of a signed playlist (purpose 'hls', accepted only by the HLS routes): 60 s to 12 h, 6 h by default.playlistTtlS: Math.min(43200, Math.max(60, intEnv('MEDIA_HLS_PLAYLIST_TTL_S', 21600))),// On-demand renditions (docs/media-fabric.md §5/F4): the rendition.create job and the extra master-playlist@@@ -266,6 +266,7 @@ const ready = (async () => {require('./vod/clip-jobs').start().catch((err) => console.warn('[Clips] re-cut job did not start:', err.message)); // failed clip re-cuts (auto-retry, hot-fetch)require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)++ require('./objects/timeline-cache').sweep().catch((err) => console.warn('[Objects] timeline cache boot sweep:', err.message));require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)// Usage readings → Billing (plan T5 step 14): off by default (MEDIA_BILLING_INTERVAL_MS=0 starts nothing).// With an interval it aggregates each closed UTC day into one platform.usage-sample@1 gb-month reading and@@@ -288,6 +289,7 @@ const ready = (async () => {try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }++ try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).@@@ -131,6 +131,8 @@ async function run(job, ctx) {if (!provider) throw new JobError('media.provider.unavailable', 'No durable provider is configured and healthy for the chunks', { retryAfterS: 1800 });out.provider = provider;const vodStorage = require('../vod/vod-storage');++ const cache = require('../objects/timeline-cache');++ let localRemaining = Math.max(0, config.hls.localMaxBytes - cache.status().bytes);const dir = derive.workDir(job.id);try {for (const chunk of chunks) {@@@ -156,7 +158,9 @@ async function run(job, ctx) {const name = `p${timeline.segmentName(Number(chunk[0].seq))}`;const key = timeline.keyFor(src.id, rendition, name, sha.slice(0, 12));// This node keeps a copy of the chunk where it kept copies of the segments.-- const keepLocal = chunk.some((r) => r.local_path && derive.existing(r.local_path));++ const keepLocal = chunk.some((r) => r.local_path && derive.existing(r.local_path))++ && localRemaining >= offset && cache.freeBytes() >= config.hls.localMinFreeBytes;++ if (keepLocal) localRemaining -= offset;const localPath = keepLocal ? timeline.localPathFor(src, rendition, name, sha.slice(0, 12)) : null;let uploaded = false;let committed = false;@@@ -167,7 +171,7 @@ async function run(job, ctx) {aborted(ctx);// Durable means verified: the provider gives back exactly these bytes.const back = await vodStorage.sha256Object(provider, key);-- if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });++ if (!back || back.sha256 !== sha || Number(back.size) !== offset) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }aborted(ctx);committed = await timeline.markPacked(src.id, placed.map((p) => ({@@@ -29,6 +29,7 @@const fs = require('fs');const path = require('path');const timeline = require('../objects/timeline');++const config = require('../config');const derive = require('./derive');const { JobError } = require('./queue');@@@ -96,6 +97,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = 'const target = await timeline.durableProvider(src);// B2/R2 configured but none healthy (breakers open) is "not durable yet", retried later, never a quiet success.const remoteConfigured = vodStorage.REMOTE_PROVIDERS.some((p) => vodStorage.providerConfigured(p));++ const cache = require('../objects/timeline-cache');++ // One tree scan per cut. Only this heavy job adds timeline files; charge every retained file against this++ // snapshot and recheck free space for each file, instead of walking the growing tree for every segment.++ let localRemaining = Math.max(0, config.hls.localMaxBytes - cache.status().bytes);const rows = [];let uploaded = 0;const failed = [];@@@ -103,6 +108,7 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '// again, so a cancelled run leaves neither untracked keys nor files.const stagedRemote = [];const stagedLocal = [];++ const dropLocal = [];let committed = false;try {for (const f of files) {@@@ -138,6 +144,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = 'if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);await vodStorage.uploadFile(target, key, localPath, f.seq === 0 ? 'video/mp4' : 'video/iso.segment', { signal: ctx.signal });stagedRemote.push({ provider: target, key });++ const back = await vodStorage.sha256Object(target, key);++ if (!back || back.sha256 !== sha || Number(back.size) !== size) throw new Error(`${target}:${key} failed size/sha256 readback`);durable = target;uploaded++;} catch (err) {@@@ -145,6 +153,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = 'failed.push(`${f.name}: ${err.message}`);}}++ if (durable && (localRemaining < size || cache.freeBytes() < config.hls.localMinFreeBytes)) {++ dropLocal.push(localPath);++ localPath = null;++ } else if (localPath) localRemaining = Math.max(0, localRemaining - size);rows.push({seq: f.seq, name: f.name, start_ms: f.start_ms, end_ms: f.end_ms, keyframe_ms: f.keyframe_ms, key, local_path: localPath,durable_provider: durable, packed_object_id: null, byte_offset: null, byte_length: size, sha256: sha,@@@ -163,6 +175,11 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = 'throw err;}committed = true;++ // A published durable row now resolves through the router. Remove the staged local copy only after commit;++ // another row may share this content-addressed path, in which case it remains its cache until a sweep.++ for (const p of dropLocal) {++ if (!await timeline.isNamed({ localPath: p })) { try { fs.unlinkSync(p); } catch { /* sweep reports orphan */ } }++ }// The master playlist reads these without touching bytes; a failure here leaves the timeline correct and only// the variant's RESOLUTION/CODECS absent, so it never fails the cut.if (meta) {@@@ -80,7 +80,7 @@ async function throttleOnce(map, objectId, queue) {/** Queue object.cmaf for an object at most once per QUEUE_TTL_MS in this process. Never throws. */async function queueCmafOnce(appId, objectId) {-- await throttleOnce(_queuedCmaf, objectId, () => require('./timeline-queue').queueCmaf(appId, objectId));++ await throttleOnce(_queuedCmaf, objectId, () => require('./timeline-queue').queueCmaf(appId, objectId, { lazy: true }));}/**@@@ -914,6 +914,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {for (const loc of decision.candidates || []) {if (loc.provider === 'local') {if (!fs.existsSync(loc.key)) continue;++ require('./timeline-cache').served(loc.key);if (packed) return sendSlice(req, res, { file: loc.key, ...slice });return require('../public/routes').streamFileWithRange(req, res, loc.key, headers);}@@@ -1066,4 +1067,5 @@ module.exports.contentHandlers = [contentAuth, tenantCors, limits('media.object.// minute is 50 MB/s) and its session already bounds them; this stops only a runaway loop.module.exports.partHandlers = [mpWrite, tenantCors, limits('media.object.part', { minute: 600, hour: 20000 }), putPart];module.exports.publicRouter = publicRouter;++module.exports.sendSegmentRow = sendSegmentRow; // socket-free verification of durable ranged fallbackmodule.exports.INLINE = INLINE;nnew file mode 100644@@@ -0,0 +1,104 @@++/** Local CMAF cache. Durable timeline rows remain authoritative after their local copy is evicted. */++'use strict';++++const fs = require('fs');++const path = require('path');++const config = require('../config');++const db = require('../db/database');++const timeline = require('./timeline');++++let evictions = 0;++let sweeping = false;++++function freeBytes() {++ try {++ const st = fs.statfsSync(config.objects.path);++ return Number(st.bavail) * Number(st.bsize);++ } catch { return 0; } // Unknown free space must not authorize another local copy.++}++++function files() {++ const root = timeline.localRoot();++ const out = [];++ function visit(dir) {++ let entries;++ try { entries = fs.readdirSync(dir, { withFileTypes: true }); } catch { return; }++ for (const entry of entries) {++ const file = path.join(dir, entry.name);++ if (entry.isDirectory()) visit(file);++ else if (entry.isFile()) {++ try { const st = fs.statSync(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }++ catch { /* disappeared while scanning */ }++ }++ }++ }++ visit(root);++ return out;++}++++// status() walks every local timeline file, so readiness and /metrics read a copy at most 30 s old; a write decision++// (shouldKeepLocal) and the sweep read it fresh, and a sweep that removed files drops the copy.++const STATUS_TTL_MS = 30_000;++let memo = null;++function status({ fresh = false } = {}) {++ const now = Date.now();++ // Only the measured numbers are kept; the budget and the floor are read from config on every call.++ if (fresh || !memo || now - memo.at >= STATUS_TTL_MS) memo = { at: now, bytes: files().reduce((n, f) => n + f.size, 0), free: freeBytes() };++ return { bytes: memo.bytes, max_bytes: config.hls.localMaxBytes, free_bytes: memo.free, min_free_bytes: config.hls.localMinFreeBytes };++}++++function shouldKeepLocal() {++ const s = status({ fresh: true });++ return s.max_bytes > 0 && s.bytes < s.max_bytes && s.free_bytes >= s.min_free_bytes;++}++++// mtime is explicitly maintained on a local read. atime is mount-option dependent; mtime is portable and survives++// process restarts. No database write is needed for each segment request. The sweep sorts one entry per physical file.++function served(file, now = Date.now()) {++ try { fs.utimesSync(file, now / 1000, now / 1000); } catch { /* an eviction may have just unlinked it */ }++}++++async function sweep() {++ if (sweeping) return { skipped: true };++ sweeping = true;++ try {++ const entries = files().sort((a, b) => a.servedAt - b.servedAt);++ let bytes = entries.reduce((n, f) => n + f.size, 0);++ let free = freeBytes();++ let removed = 0;++ for (const file of entries) {++ if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) break;++ // A path can be shared by source, clip and multiple packed rows. All must name a verified durable++ // provider before the path can go. The row's durable state is set only after size+sha readback.++ const rows = await db.all('SELECT durability, durable_provider, key, sha256, packed_object_id, byte_offset, byte_length FROM media_timeline WHERE local_path = ?', [file.path]);++ if (!rows.length || rows.some((r) => r.durability !== 'durable' || !r.durable_provider)) continue;++ // Existing timelines predate the cache and may have been marked durable after a size-only upload.++ // Read back every distinct immutable key before evicting, including the complete packed chunk.++ const copies = new Map();++ for (const r of rows) {++ const id = `${r.durable_provider}\n${r.key}`;++ const expected = r.packed_object_id || r.sha256;++ if (copies.has(id)) continue;++ const back = await require('../vod/vod-storage').sha256Object(r.durable_provider, r.key).catch(() => null);++ copies.set(id, !!back && back.sha256 === expected && Number(back.size) === file.size);++ }++ if ([...copies.values()].some((ok) => !ok)) continue;++ await db.getDb().tx(async () => {++ const current = await db.all('SELECT durability, durable_provider FROM media_timeline WHERE local_path = ? FOR UPDATE', [file.path]);++ if (!current.length || current.some((r) => r.durability !== 'durable' || !r.durable_provider)) return;++ // Keep the unlink and location change in the same transaction. A failed unlink rolls back the++ // UPDATE; readers already tolerate a missing local file and use the durable location.++ fs.unlinkSync(file.path);++ await db.run('UPDATE media_timeline SET local_path = NULL, updated_at = ov_now() WHERE local_path = ?', [file.path]);++ removed++;++ evictions++;++ bytes -= file.size;++ });++ free = freeBytes();++ }++ if (removed) memo = null;++ return { removed, ...status({ fresh: true }) };++ } finally { sweeping = false; }++}++++module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions };@@@ -26,6 +26,8 @@const crypto = require('crypto');const config = require('../config');++const lazyQueuedAt = [];++let pendingLazy = 0;/*** The identity of the source cut a rendition is made against: the first 16 hex of sha256 over the source rows' (seq,@@@ -37,19 +39,34 @@ function sourceVersion(rows) {return h.digest('hex').slice(0, 16);}--async function queueCmaf(appId, objectId, { segment_seconds = null } = {}) {++async function queueCmaf(appId, objectId, { segment_seconds = null, lazy = false, now = Date.now() } = {}) {if (!config.hls.enabled || !appId || !objectId) return null;const obj = await require('./model').getObject(objectId);if (!obj || obj.lifecycle_status !== 'ready' || !require('../jobs/derive').isMediaObject(obj)) return null;if (await require('./timeline').has(obj.id)) return null;++ // Count persisted jobs (including those from before a restart) and successful queues in this process. Reserve++ // synchronously after the count so concurrent reads cannot overshoot it while their enqueues are in flight.++ let recent = 0;++ if (lazy) {++ while (lazyQueuedAt.length && lazyQueuedAt[0] <= now - 3600000) lazyQueuedAt.shift();++ const row = await require('../db/database').get("SELECT COUNT(*) AS n FROM media_jobs WHERE created_by = 'system:timeline-backfill' AND created_at >= ov_now_iso('-1 hour')");++ recent = Number(row.n);++ }const queue = require('../jobs/queue');const params = require('../jobs/cmaf').spec.validate({ appId, obj, params: segment_seconds != null ? { segment_seconds } : {} });-- const r = await queue.enqueue({-- appId, type: 'object.cmaf', objectId: obj.id, params,-- dedupeActive: true, idempotencyKey: `object.cmaf:${obj.id}`, createdBy: 'system:timeline',-- });-- if (r.created) require('../jobs/worker').kick();-- return r.job ? r.job.id : null;++ if (lazy) {++ if (Math.max(recent, lazyQueuedAt.length) + pendingLazy >= config.hls.backfillPerHour) return null;++ pendingLazy++;++ }++ try {++ const r = await queue.enqueue({++ appId, type: 'object.cmaf', objectId: obj.id, params,++ dedupeActive: true, idempotencyKey: `object.cmaf:${obj.id}`, createdBy: lazy ? 'system:timeline-backfill' : 'system:timeline',++ });++ if (r.created && lazy) lazyQueuedAt.push(now);++ if (r.created) require('../jobs/worker').kick();++ return r.job ? r.job.id : null;++ } finally { if (lazy) pendingLazy--; }}async function queuePack(appId, objectId, { rendition = null } = {}) {@@@ -56,6 +56,9 @@ function storageDirs(config) {function domainMetrics(registry, { db, recorder, events }) {registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });++ const timelineCache = require('./objects/timeline-cache');++ registry.gauge({ name: 'media_timeline_local_bytes', help: 'Bytes used by local timeline files, including orphan files', collect: () => timelineCache.status().bytes });++ registry.gauge({ name: 'media_timeline_local_evictions', help: 'Local timeline files evicted since process start', collect: () => timelineCache.evictions() });// target vod: media_tier_decisions (VODs); object: media_object_tier_decisions (native v2 objects), whose// outcome dry_run is what the object sweep would have done while its activation gate is off.registry.gauge({@@@ -136,6 +139,12 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, remfor (const [name, dir] of Object.entries(storageDirs(config))) {checks.push({ name: `storage_${name}`, required: true, check: () => writable(dir) });}++ checks.push({ name: 'timeline_cache', required: false, check: () => {++ if (!config.hls.enabled) return { skipped: 'MEDIA_HLS_ENABLED off: no timeline is cut, nothing is cached' };++ const s = require('./objects/timeline-cache').status();++ const ok = s.bytes <= s.max_bytes && s.free_bytes >= s.min_free_bytes;++ return { ok, ...(ok ? {} : { error: 'Timeline cache over budget or disk below free-space floor' }), detail: s };++ } });}checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });for (const name of require('./vod/vod-storage').REMOTE_PROVIDERS) {@@@ -267,7 +267,8 @@ const http = require('http');assert.strictEqual(r.status, 202);assert.ok(await waitFor(() => false, 50) || true);await worker.start();-- assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded'), '?async=1 queues it for the worker');++ // ffprobe then ffmpeg (the audio-only check, Media#93) under a parallel suite: give the real job 15 s.++ assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded', 15000), '?async=1 queues it for the worker');r = await call('POST', '/api/v2/live/jobs', { body: { type: 'thumbnail.regenerate', object_id: `legacy:live:vod:1` } });assert.deepStrictEqual([r.status, r.body.job.params], [202, { kind: 'vod', id: 1 }], 'the v2 API takes the object and finds the row');assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded'));@@@ -63,13 +63,13 @@ const { spawn } = require('child_process');assert.strictEqual(body.service, 'media');assert.strictEqual(body.recordings_in_progress, 1);const names = Object.keys(body.checks).sort();-- assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'valkey']);++ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'timeline_cache', 'valkey']);// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optional// skip leaves the verified service ready.assert.deepStrictEqual([body.checks.remote_r2.status, body.checks.remote_r2.reason], ['skipped', 'not configured']);-- assert.deepStrictEqual(body.skipped.sort(), ['remote_r2', 'valkey'], 'Valkey too: no VALKEY_URL in this test');++ assert.deepStrictEqual(body.skipped.sort(), ['remote_r2', 'timeline_cache', 'valkey'], 'Valkey too (no VALKEY_URL in this test), and the timeline cache while HLS is off');for (const [n, c] of Object.entries(body.checks)) {-- if (n === 'remote_r2' || n === 'valkey') continue;++ if (n === 'remote_r2' || n === 'valkey' || n === 'timeline_cache') continue;assert.strictEqual(c.status, 'ok', n);assert.strictEqual(typeof c.latency_ms, 'number');assert.ok(Date.parse(c.checked_at));nnew file mode 100644@@@ -0,0 +1,230 @@++'use strict';++// No listening socket: real migrated PGlite rows, local files and a fake durable readback.++const assert = require('assert');++const crypto = require('crypto');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { EventEmitter } = require('events');++++(async () => {++ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-cache-'));++ process.env.OBJECTS_PATH = root;++ process.env.MEDIA_HLS_ENABLED = '1';++ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = '8';++ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';++ process.env.MEDIA_HLS_BACKFILL_PER_HOUR = '20';++ const config = require('../server/config');++ const db = require('../server/db/database');++ const timeline = require('../server/objects/timeline');++ const cache = require('../server/objects/timeline-cache');++ const storage = require('../server/vod/vod-storage');++ const blobs = new Map();++ const realRead = storage.sha256Object;++ storage.sha256Object = async (_provider, key) => {++ const bytes = blobs.get(key);++ return bytes ? { size: bytes.length, sha256: crypto.createHash('sha256').update(bytes).digest('hex') } : null;++ };++ try {++ const obj = { app_id: 'live', id: 'cache-test' };++ const make = async (seq, data, durable, at) => {++ const bytes = Buffer.from(data);++ const name = timeline.segmentName(seq);++ const sha = crypto.createHash('sha256').update(bytes).digest('hex');++ const key = `cache/${name}`;++ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));++ fs.mkdirSync(path.dirname(file), { recursive: true });++ fs.writeFileSync(file, bytes);++ cache.served(file, at);++ if (durable) blobs.set(key, bytes);++ await timeline.replace(`cache-${seq}`, 'source', [{ seq, name, start_ms: 0, end_ms: 1000, key, local_path: file,++ durable_provider: durable ? 'b2' : null, packed_object_id: null, byte_offset: null,++ byte_length: bytes.length, sha256: sha, durability: durable ? 'durable' : 'local' }]);++ return file;++ };++ const old = await make(1, 'aaaaaa', true, 1000);++ const recent = await make(2, 'bbbbbb', true, 3000);++ const only = await make(3, 'cccccc', false, 500);++ assert.strictEqual(cache.status().bytes, 18);++ assert.strictEqual(cache.shouldKeepLocal(), false);++ const first = await cache.sweep();++ assert.strictEqual(first.removed, 2);++ assert.strictEqual(fs.existsSync(old), false);++ assert.strictEqual(fs.existsSync(recent), false);++ assert.strictEqual(fs.existsSync(only), true, 'the only copy stays even above budget');++ assert.strictEqual((await timeline.byName('cache-1', 'source', timeline.segmentName(1))).local_path, null);++ assert.strictEqual(cache.evictions(), 2);++ assert.strictEqual(cache.status().bytes, 6);++ const orphanReport = await require('../server/vod/orphans-report').buildStorageReport({ providers: [] });++ assert.ok(!orphanReport.unreferenced_local.some((f) => f.path.includes('.timeline/')),++ 'evicted files are gone and the remaining local-only timeline file is still referenced');++ const observability = require('../server/observability');++ const gauges = new Map();++ const registry = {++ gauge: (spec) => { gauges.set(spec.name, spec.collect); return { set: () => {} }; },++ counter: () => ({ inc: () => {} }), histogram: () => ({ observe: () => {} }),++ };++ const events = { status: async () => ({ enabled: false }) };++ observability.domainMetrics(registry, { db, recorder: { activeCount: () => 0 }, events });++ assert.strictEqual(gauges.get('media_timeline_local_bytes')(), 6);++ assert.strictEqual(gauges.get('media_timeline_local_evictions')(), 2);++ for (const section of ['vod', 'pastes', 'files', 'thumbnails']) {++ if (section === 'vod') { config.vod.path = root; config.vod.clipsPath = root; }++ else config[section].path = root;++ }++ const ready = observability.createMediaReadiness({ release: null, db, config,++ auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 }, events,++ remote: { configured: () => false } });++ let report = await ready.run();++ assert.strictEqual(report.checks.timeline_cache.status, 'ok');++ assert.strictEqual(report.checks.timeline_cache.detail.bytes, 6);++ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;++ report = await ready.run();++ assert.strictEqual(report.checks.timeline_cache.status, 'fail');++ assert.ok(report.degraded.includes('timeline_cache'));++ config.hls.localMinFreeBytes = 0;++ const row = await timeline.byName('cache-1', 'source', timeline.segmentName(1));++ assert.deepStrictEqual(timeline.locationsOf(row).map((x) => x.provider), ['b2']);++ assert.deepStrictEqual(blobs.get(row.key), Buffer.from('aaaaaa'));++ // A packed row's byte range remains the same after its local chunk is evicted.++ const packed = Buffer.from('abcdef');++ const packedSha = crypto.createHash('sha256').update(packed).digest('hex');++ const packedFile = timeline.localPathFor(obj, 'source', 'p000001.m4s', packedSha.slice(0, 12));++ fs.mkdirSync(path.dirname(packedFile), { recursive: true });++ fs.writeFileSync(packedFile, packed);++ blobs.set('cache/packed', packed);++ await timeline.replace('cache-packed', 'source', [{ seq: 1, name: '000001.m4s', start_ms: 0, end_ms: 1000,++ key: 'cache/packed', local_path: packedFile, durable_provider: 'b2', packed_object_id: packedSha,++ byte_offset: 2, byte_length: 3, sha256: crypto.createHash('sha256').update('cde').digest('hex'), durability: 'durable' }]);++ config.hls.localMaxBytes = 0;++ await cache.sweep();++ const packedRow = await timeline.byName('cache-packed', 'source', '000001.m4s');++ assert.strictEqual(packedRow.local_path, null);++ assert.deepStrictEqual(blobs.get(packedRow.key).subarray(Number(packedRow.byte_offset), Number(packedRow.byte_offset) + Number(packedRow.byte_length)), Buffer.from('cde'));++ const router = require('../server/placement/router');++ const realRoute = router.route;++ const realSession = router.sessionFor;++ const realPresign = storage.presignGet;++ const realFetch = global.fetch;++ router.route = async () => ({ provider: 'b2', candidates: [{ provider: 'b2', key: packedRow.key }] });++ router.sessionFor = () => null;++ storage.presignGet = async () => 'https://durable.invalid/chunk';++ global.fetch = async (_url, opts) => {++ const m = /^bytes=(\d+)-(\d+)$/.exec(opts.headers.range);++ return new Response(packed.subarray(Number(m[1]), Number(m[2]) + 1), { status: 206 });++ };++ try {++ const req = new EventEmitter();++ req.headers = {};++ const res = { destroyed: false, writeHead(code, headers) { this.code = code; this.headers = headers; }, end(body) { this.body = body; } };++ await require('../server/objects/routes').sendSegmentRow(req, res, { obj: { id: 'cache-packed' }, closed: false }, packedRow, 'source');++ assert.strictEqual(res.code, 200);++ assert.deepStrictEqual(res.body, Buffer.from('cde'), 'the segment route reads the packed range from durable storage');++ } finally {++ router.route = realRoute; router.sessionFor = realSession; storage.presignGet = realPresign; global.fetch = realFetch;++ }++ assert.strictEqual(fs.existsSync(only), true);++ console.log('timeline cache: oldest verified copies and packed range evicted; sole copy retained');++++ // The CMAF publisher stages normally, verifies the upload, then publishes remote-only rows below the floor.++ const publishDir = path.join(root, 'publish');++ fs.mkdirSync(publishDir);++ fs.writeFileSync(path.join(publishDir, 'init.mp4'), 'init');++ fs.writeFileSync(path.join(publishDir, '000001.m4s'), 'media');++ const realProvider = timeline.durableProvider;++ const realAvailable = storage.providerAvailable;++ const realUpload = storage.uploadFile;++ const timelineQueue = require('../server/objects/timeline-queue');++ const realQueuePack = timelineQueue.queuePack;++ timeline.durableProvider = async () => 'b2';++ storage.providerAvailable = () => true;++ storage.uploadFile = async (_provider, key, file) => { blobs.set(key, fs.readFileSync(file)); };++ timelineQueue.queuePack = async () => null;++ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;++ try {++ const published = await require('../server/jobs/segments').publishRendition({ job: { id: 'cache-publish', app_id: 'live' },++ ctx: { signal: new AbortController().signal }, src: { id: 'cache-publish', app_id: 'live' },++ rendition: 'source', out: publishDir, list: [{ name: '000001.m4s', seconds: 1 }] });++ assert.strictEqual(published.durable, 2);++ assert.strictEqual(published.local_only, 0);++ const rows = await timeline.list('cache-publish');++ assert.ok(rows.every((r) => r.local_path === null && r.durable_provider === 'b2'));++ assert.ok(rows.every((r) => blobs.has(r.key)));++ } finally {++ timeline.durableProvider = realProvider; storage.providerAvailable = realAvailable;++ storage.uploadFile = realUpload; timelineQueue.queuePack = realQueuePack;++ config.hls.localMinFreeBytes = 0;++ }++ // Packing below the floor uploads and verifies its chunk, then points every row at durable bytes only.++ const packRows = [];++ for (let seq = 1; seq <= 2; seq++) {++ const bytes = Buffer.from(seq === 1 ? 'first' : 'second');++ const name = timeline.segmentName(seq);++ const sha = crypto.createHash('sha256').update(bytes).digest('hex');++ const key = `cache-pack/${name}`;++ const file = timeline.localPathFor({ app_id: 'live', id: 'cache-pack' }, 'source', name, sha.slice(0, 12));++ fs.mkdirSync(path.dirname(file), { recursive: true });++ fs.writeFileSync(file, bytes);++ blobs.set(key, bytes);++ packRows.push({ seq, name, start_ms: (seq - 1) * 6000, end_ms: seq * 6000, keyframe_ms: (seq - 1) * 6000,++ key, local_path: file, durable_provider: 'b2', packed_object_id: null, byte_offset: null,++ byte_length: bytes.length, sha256: sha, durability: 'durable' });++ }++ await timeline.replace('cache-pack', 'source', packRows);++ const derive = require('../server/jobs/derive');++ const realLoad = derive.loadSource;++ timeline.durableProvider = async () => 'b2';++ storage.providerAvailable = () => true;++ storage.uploadFile = async (_provider, key, file) => { blobs.set(key, fs.readFileSync(file)); };++ derive.loadSource = async () => ({ id: 'cache-pack', app_id: 'live' });++ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;++ try {++ const result = await require('../server/jobs/pack').spec.run({ id: 'cache-pack-job', params: { target_seconds: 10 } },++ { signal: new AbortController().signal });++ assert.strictEqual(result.packs, 1);++ const packedRows = await timeline.list('cache-pack');++ assert.ok(packedRows.every((r) => r.local_path === null && r.packed_object_id));++ assert.ok(packRows.every((r) => !fs.existsSync(r.local_path)));++ assert.deepStrictEqual(blobs.get(packedRows[0].key), Buffer.from('firstsecond'));++ } finally {++ derive.loadSource = realLoad; timeline.durableProvider = realProvider;++ storage.providerAvailable = realAvailable; storage.uploadFile = realUpload;++ config.hls.localMinFreeBytes = 0;++ }++++ const model = require('../server/objects/model');++ const queue = require('../server/jobs/queue');++ const worker = require('../server/jobs/worker');++ const realEnqueue = queue.enqueue;++ const realKick = worker.kick;++ let queued = 0;++ queue.enqueue = async () => ({ created: true, job: { id: `job-${++queued}` } });++ worker.kick = () => {};++ await db.upsertApp({ app_id: 'live', api_key: 'cache-key' });++ const queueCmaf = require('../server/objects/timeline-queue').queueCmaf;++ const base = Date.now();++ try {++ for (let n = 0; n < 21; n++) {++ const id = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });++ const result = await queueCmaf('live', id, { lazy: true, now: base });++ assert.strictEqual(!!result, n < 20);++ }++ assert.strictEqual(queued, 20);++ const id = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });++ assert.ok(await queueCmaf('live', id), 'finalize bypasses the lazy cap');++ assert.strictEqual(queued, 21);++ const realGet = db.get;++ db.get = async (sql, ...args) => sql.includes("created_by = 'system:timeline-backfill'") ? { n: 20 } : realGet(sql, ...args);++ try {++ const later = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });++ assert.strictEqual(await queueCmaf('live', later, { lazy: true, now: base + 3600001 }), null,++ 'persisted job count still caps lazy queues after the in-memory hour rolls over');++ } finally { db.get = realGet; }++ } finally { queue.enqueue = realEnqueue; worker.kick = realKick; }++ console.log('timeline cache: lazy queue capped at 20 per hour; finalize bypasses cap');++ } finally {++ storage.sha256Object = realRead;++ await db.close();++ fs.rmSync(root, { recursive: true, force: true });++ }++})().catch((err) => { console.error(err); process.exitCode = 1; });@@@ -185,7 +185,7 @@ const express = require('express');await call('GET', `/api/v2/live/objects/${old}/download?format=json`);cmaf = await jobs('object.cmaf', old);assert.strictEqual(cmaf.length, 1, 'queued once on demand');-- assert.strictEqual(cmaf[0].created_by, 'system:timeline');++ assert.strictEqual(cmaf[0].created_by, 'system:timeline-backfill');assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0);await call('GET', `/api/v2/live/objects/${pdf}/download?format=json`);assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0, 'a non-media download queues nothing');
@@ -143,6 +143,12 @@ MEDIA_CAPABILITY_GATE=report
@@ -143,6 +143,12 @@ MEDIA_CAPABILITY_GATE=report
# /o/:id/master.m3u8, /o/:id/source/… and the hls_url field of GET …/download?format=json ; off (default):
# /o/:id/master.m3u8, /o/:id/source/… and the hls_url field of GET …/download?format=json ; off (default):
# the job is refused and the routes 404, nothing else changes.
# the job is refused and the routes 404, nothing else changes.
#MEDIA_HLS_ENABLED=0
#MEDIA_HLS_ENABLED=0
+# Maximum bytes of local timeline segments and packed chunks (default 10 GiB; 0 keeps only durable copies).
+#MEDIA_TIMELINE_LOCAL_MAX_BYTES=10737418240
+# Stop keeping new local timeline copies below this free space (default 8 GiB).
+#MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES=8589934592
+# At most this many old VODs are lazily queued per hour; recording finalize is exempt (default 20).
+#MEDIA_HLS_BACKFILL_PER_HOUR=20
# Lifetime in seconds of a signed HLS playlist (download?format=json → hls_url of a private or sandbox object; its own
# Lifetime in seconds of a signed HLS playlist (download?format=json → hls_url of a private or sandbox object; its own
# signature purpose, accepted only by the HLS routes). Default 21600 (6 h), clamped to 60..43200.
# signature purpose, accepted only by the HLS routes). Default 21600 (6 h), clamped to 60..43200.
#MEDIA_HLS_PLAYLIST_TTL_S=21600
#MEDIA_HLS_PLAYLIST_TTL_S=21600
new file mode 100644
new file mode 100644
@@ -0,0 +1,9 @@
@@ -0,0 +1,9 @@
+# Changelog
+
+## Unreleased
+
+- Bound local CMAF and packed timeline copies with a configurable byte budget and free-space floor. The hourly sweep
+ evicts least recently served copies only after durable size and SHA-256 verification, preserving all timeline rows
+ and durable playback.
+- Limit lazy HLS backfill to 20 VODs per hour by default while recording finalize remains immediate.
+- Add timeline cache readiness and metrics, environment documentation and production enablement guidance.
@@ -446,6 +446,20 @@ that reaches port 4100 directly cannot choose its IP by sending `X-Forwarded-For
@@ -446,6 +446,20 @@ that reaches port 4100 directly cannot choose its IP by sending `X-Forwarded-For
`CF-Connecting-IP` itself. View counting and the live-frame rate limit read `req.ip`,
`CF-Connecting-IP` itself. View counting and the live-frame rate limit read `req.ip`,
never the headers (`test/trust-proxy.test.js`).
never the headers (`test/trust-proxy.test.js`).
+## Segment-native video settings
+
+| Environment variable | Default | Purpose |
+|---|---:|---|
+| `MEDIA_HLS_ENABLED` | `0` | Enable timeline generation and HLS routes. |
+| `MEDIA_TIMELINE_LOCAL_MAX_BYTES` | `10737418240` | Maximum local timeline cache (10 GiB; `0` keeps no durable copies locally). |
+| `MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES` | `8589934592` | Below this free disk level, new verified timeline bytes remain durable only. |
+| `MEDIA_HLS_BACKFILL_PER_HOUR` | `20` | Maximum lazy `object.cmaf` queues for older VODs per rolling hour on a single Media host; finalize is exempt. |
+| `MEDIA_JOBS_HEAVY_CONCURRENCY` | `1` | Keep at 1 while enabling HLS in production. |
+
+The hourly sweep evicts the least recently served verified local timeline files. A segment whose local copy was
+evicted is served from its durable provider. Watch `timeline_cache` on `/api/ready` and job durations during backfill
+(docs/media-fabric.md §3).
+
## Per-actor limits
## Per-actor limits
nginx limits by address; the capability routes also limit by who calls (`server/actor-limits.js`,
nginx limits by address; the capability routes also limit by who calls (`server/actor-limits.js`,
@@ -506,8 +520,8 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.
@@ -506,8 +520,8 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.
| path | behavior |
| path | behavior |
|---|---|
|---|---|
| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
-| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks` (user JWTs and service tokens; app keys work without it), `remote_b2` / `remote_r2` (HeadBucket, at most once a minute; present only when the provider is configured), `events_outbox` (backlog over 1000 events). Also `recordings_in_progress`. |
-| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |
+| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress`. |
+| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_timeline_local_bytes`, `media_timeline_local_evictions`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |
| `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. |
| `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. |
| `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |
| `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |
@@ -41,11 +41,12 @@
@@ -41,11 +41,12 @@
"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
"every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
"every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
"segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
"segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
+ "local timeline cache (F3): 10 GiB default budget and 8 GiB free-disk floor, least-recently-served eviction of verified durable copies in the hourly sweep, durable segment fallback, optional timeline_cache readiness and local bytes/evictions metrics; lazy CMAF backfill capped at 20 per hour while recording finalize is exempt",
"packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)",
"packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)",
"on-demand renditions (F4 slice 1, behind MEDIA_HLS_ENABLED + MEDIA_RENDITIONS): the rendition.create job transcodes one 720p H.264/AAC rung of a ready media object into media_timeline through the same content-addressed rows, placement-router durable provider and two-phase commit as object.cmaf, forcing the source timeline's segment boundaries as keyframes; a source already 720p or smaller is skipped and a held object is left alone. The master playlist lists a rendition only once its rows exist and queues the missing ones lazily from /o/:id/master.m3u8 and download?format=json (dedupeActive + an object/rendition idempotency key), and /o/:id/<rendition>/index.m3u8 plus /o/:id/<rendition>/<name> serve it",
"on-demand renditions (F4 slice 1, behind MEDIA_HLS_ENABLED + MEDIA_RENDITIONS): the rendition.create job transcodes one 720p H.264/AAC rung of a ready media object into media_timeline through the same content-addressed rows, placement-router durable provider and two-phase commit as object.cmaf, forcing the source timeline's segment boundaries as keyframes; a source already 720p or smaller is skipped and a held object is left alone. The master playlist lists a rendition only once its rows exist and queues the missing ones lazily from /o/:id/master.m3u8 and download?format=json (dedupeActive + an object/rendition idempotency key), and /o/:id/<rendition>/index.m3u8 plus /o/:id/<rendition>/<name> serve it",
"timeline sprites (F3.1, behind MEDIA_HLS_ENABLED): object.sprite cuts its sheet from the object's media_timeline rows — each sample time picks the row covering it and one frame is decoded from that row's bytes (this node's segment, or a packed row's byte range in its chunk), checked against the row's sha256 — with the same layout and metadata as the seek path, so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame",
"timeline sprites (F3.1, behind MEDIA_HLS_ENABLED): object.sprite cuts its sheet from the object's media_timeline rows — each sample time picks the row covering it and one frame is decoded from that row's bytes (this node's segment, or a packed row's byte range in its chunk), checked against the row's sha256 — with the same layout and metadata as the seek path, so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame",
"HLS discovery on the v1 shapes (F3.c, behind MEDIA_HLS_ENABLED): the VOD and clip answers (vodPublic/clipPublic — get, list, create, update) carry hls_url (and hls_expires_at when the object is private or sandbox), the same field and access rules as GET /o/:id/download?format=json, read through one shared discovery (server/objects/hls.js); an object without a timeline gets object.cmaf queued and the field appears once it ran, and a virtual clip plays its source's window without being cut a timeline of its own",
"HLS discovery on the v1 shapes (F3.c, behind MEDIA_HLS_ENABLED): the VOD and clip answers (vodPublic/clipPublic — get, list, create, update) carry hls_url (and hls_expires_at when the object is private or sandbox), the same field and access rules as GET /o/:id/download?format=json, read through one shared discovery (server/objects/hls.js); an object without a timeline gets object.cmaf queued and the field appears once it ran, and a virtual clip plays its source's window without being cut a timeline of its own",
"materialized clips (F3.6, behind MEDIA_HLS_ENABLED + MEDIA_MATERIALIZED_CLIPS, off by default): clip.cut over a source timeline writes the clip's own media_timeline rows — the interior copies the source's segments (packed chunk ranges included), only the two window edges are re-encoded into fMP4 with their own inits (init-head.mp4/init-tail.mp4, negative seq, EXT-X-MAP + EXT-X-DISCONTINUITY in the playlist) — inserting them in one transaction that locks the named source rows FOR SHARE and aborts with a retryable error when one is gone or any location column moved (a pack), so a racing delete either waits and keeps the bytes or aborts the insert; the projection keeps virtual: true and gains materialized: true (readiness, playableSql and the verify job treat it as a virtual clip), sprites decode each edge with its own init, deleting the clip removes its rows and edge bytes while the source's bytes stay, and download?format=mp4 answers the playlist for a timeline-only clip instead of queueing a remux that cannot read its input"
"materialized clips (F3.6, behind MEDIA_HLS_ENABLED + MEDIA_MATERIALIZED_CLIPS, off by default): clip.cut over a source timeline writes the clip's own media_timeline rows — the interior copies the source's segments (packed chunk ranges included), only the two window edges are re-encoded into fMP4 with their own inits (init-head.mp4/init-tail.mp4, negative seq, EXT-X-MAP + EXT-X-DISCONTINUITY in the playlist) — inserting them in one transaction that locks the named source rows FOR SHARE and aborts with a retryable error when one is gone or any location column moved (a pack), so a racing delete either waits and keeps the bytes or aborts the insert; the projection keeps virtual: true and gains materialized: true (readiness, playableSql and the verify job treat it as a virtual clip), sprites decode each edge with its own init, deleting the clip removes its rows and edge bytes while the source's bytes stay, and download?format=mp4 answers the playlist for a timeline-only clip instead of queueing a remux that cannot read its input"
],
],
- "updated": "2026-10-11"
+ "updated": "2026-10-12"
}
}
@@ -98,7 +98,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -98,7 +98,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
(`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,
(`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,
`params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it) into `init.mp4` +
`params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it) into `init.mp4` +
`000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured,
`000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured,
- under the key `<object>/source/<version>/<name>` on B2 (`durable` once B2 confirms the size), then writes all rows of
+ under the key `<object>/source/<version>/<name>` on B2 (`durable` after a size and sha256 readback), then writes all rows of
the rendition in one transaction and only then deletes the previous cut's bytes. It never touches the source object; a
the rendition in one transaction and only then deletes the previous cut's bytes. It never touches the source object; a
rerun reuses rows whose sha256 matches and their location as they are and uploads only what is not durable yet; it
rerun reuses rows whose sha256 matches and their location as they are and uploads only what is not durable yet; it
honours the abort signal (which stops an upload in flight) and, before the rows commit, removes the files and keys it
honours the abort signal (which stops an upload in flight) and, before the rows commit, removes the files and keys it
@@ -121,7 +121,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -121,7 +121,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
before; runs of one segment and the init segment stay as they are; a segment not durable yet ends a run). Each
before; runs of one segment and the init segment stay as they are; a segment not durable yet ends a run). Each
segment's bytes are checked against its row's sha256 (this node's copy, else the durable one through the router); the
segment's bytes are checked against its row's sha256 (this node's copy, else the durable one through the router); the
chunk goes to the router's durable provider under `<object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s` and is
chunk goes to the router's durable provider under `<object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s` and is
- verified there (sha256 read back), this node keeps a copy where it kept the segments, and the rows move to it in one
+ verified there (size and sha256 read back), this node may keep a cache copy where it kept the segments, and the rows move to it in one
transaction: `key`/`local_path`/`durable_provider` name the chunk, `packed_object_id` is its sha256 (content address),
transaction: `key`/`local_path`/`durable_provider` name the chunk, `packed_object_id` is its sha256 (content address),
`byte_offset`/`byte_length` the segment inside it; name, times and sha256 stay the segment's. Only after the commit are
`byte_offset`/`byte_length` the segment inside it; name, times and sha256 stay the segment's. Only after the commit are
the per-segment files and keys **deleted**: the chunk holds the same verified bytes, keeping both would double the
the per-segment files and keys **deleted**: the chunk holds the same verified bytes, keeping both would double the
@@ -168,6 +168,26 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -168,6 +168,26 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux
`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux
(`dedupeActive`) and answers `202 { job_id }` while it is missing.
(`dedupeActive`) and answers `202 { job_id }` while it is missing.
- **F3.5, shipped: virtual clips over the timeline (§4).**
- **F3.5, shipped: virtual clips over the timeline (§4).**
+- **Local timeline cache, shipped:** `OBJECTS_PATH/.timeline` is bounded by
+ `MEDIA_TIMELINE_LOCAL_MAX_BYTES` (10 GiB by default; 0 keeps no new local copies once durable) and
+ `MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES` (8 GiB by default). CMAF and packing stage their work as before; once the
+ durable upload passes size and sha256 readback, they keep no new local copy if the cache is full or free disk is
+ below the floor. The hourly object sweep evicts least recently served local files until both limits are satisfied.
+ Serving a local segment updates its file mtime, a portable recency signal that survives restarts without a database
+ write per request. The sweep verifies the durable bytes again, including older size-only timeline uploads, and clears
+ every row sharing that local path in its delete transaction. A file with an unverified or missing durable copy remains;
+ no timeline row is deleted. The route then uses its existing durable fallback, including packed byte ranges. The
+ storage orphan report still counts only row-named timeline files; an unlinked cache file and its cleared path are
+ absent from both sides. `timeline_cache` is optional on `/api/ready` and degrades above budget or below the free
+ floor; `/metrics` exposes `media_timeline_local_bytes` and `media_timeline_local_evictions`.
+
+ Lazy discovery of an older VOD without a timeline is capped at `MEDIA_HLS_BACKFILL_PER_HOUR` (20 per rolling hour,
+ with job history surviving a restart on a single Media host); a capped read simply has no `hls_url` and its MP4/file playback remains available. A recording's
+ own finalize bypasses the cap. List reads discover timelines in batches and do not queue missing ones.
+
+ **Turning HLS on in production:** set the cache budget and free-space floor for the host, set
+ `MEDIA_HLS_ENABLED=1`, keep `MEDIA_JOBS_HEAVY_CONCURRENCY=1`, and watch `timeline_cache`, local timeline bytes,
+ evictions, and `object.cmaf`/`object.pack` job durations during backfill.
- **F3.6, shipped (behind `MEDIA_HLS_ENABLED` + `MEDIA_MATERIALIZED_CLIPS`, off by default): materialized clips over
- **F3.6, shipped (behind `MEDIA_HLS_ENABLED` + `MEDIA_MATERIALIZED_CLIPS`, off by default): materialized clips over
the source's bytes (§4, docs/materialized-clips.md).** A clip whose source has a timeline gets its own persisted
the source's bytes (§4, docs/materialized-clips.md).** A clip whose source has a timeline gets its own persisted
`media_timeline` rows: the interior segments are the source's rows (same location, packed chunk range and sha256,
`media_timeline` rows: the interior segments are the source's rows (same location, packed chunk range and sha256,
@@ -93,6 +93,9 @@ const config = {
@@ -93,6 +93,9 @@ const config = {
// media_timeline (/o/:id/master.m3u8, /o/:id/source/…). Off by default: off, the job is refused and the routes 404.
// media_timeline (/o/:id/master.m3u8, /o/:id/source/…). Off by default: off, the job is refused and the routes 404.
hls: {
hls: {
enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
+ localMaxBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MAX_BYTES', 10 * 1024 ** 3)),
+ localMinFreeBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES', 8 * 1024 ** 3)),
+ backfillPerHour: Math.max(0, intEnv('MEDIA_HLS_BACKFILL_PER_HOUR', 20)),
// Lifetime of a signed playlist (purpose 'hls', accepted only by the HLS routes): 60 s to 12 h, 6 h by default.
// Lifetime of a signed playlist (purpose 'hls', accepted only by the HLS routes): 60 s to 12 h, 6 h by default.
playlistTtlS: Math.min(43200, Math.max(60, intEnv('MEDIA_HLS_PLAYLIST_TTL_S', 21600))),
playlistTtlS: Math.min(43200, Math.max(60, intEnv('MEDIA_HLS_PLAYLIST_TTL_S', 21600))),
// On-demand renditions (docs/media-fabric.md §5/F4): the rendition.create job and the extra master-playlist
// On-demand renditions (docs/media-fabric.md §5/F4): the rendition.create job and the extra master-playlist
@@ -266,6 +266,7 @@ const ready = (async () => {
@@ -266,6 +266,7 @@ const ready = (async () => {
require('./vod/clip-jobs').start().catch((err) => console.warn('[Clips] re-cut job did not start:', err.message)); // failed clip re-cuts (auto-retry, hot-fetch)
require('./vod/clip-jobs').start().catch((err) => console.warn('[Clips] re-cut job did not start:', err.message)); // failed clip re-cuts (auto-retry, hot-fetch)
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
require('./jobs/worker').start().catch((err) => console.warn('[Jobs] worker did not start:', err.message)); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
+ require('./objects/timeline-cache').sweep().catch((err) => console.warn('[Objects] timeline cache boot sweep:', err.message));
require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)
require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)
// Usage readings → Billing (plan T5 step 14): off by default (MEDIA_BILLING_INTERVAL_MS=0 starts nothing).
// Usage readings → Billing (plan T5 step 14): off by default (MEDIA_BILLING_INTERVAL_MS=0 starts nothing).
// With an interval it aggregates each closed UTC day into one platform.usage-sample@1 gb-month reading and
// With an interval it aggregates each closed UTC day into one platform.usage-sample@1 gb-month reading and
@@ -288,6 +289,7 @@ const ready = (async () => {
@@ -288,6 +289,7 @@ const ready = (async () => {
try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }
try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }
// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.
// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.
try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }
try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }
+ try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).
// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).
@@ -131,6 +131,8 @@ async function run(job, ctx) {
@@ -131,6 +131,8 @@ async function run(job, ctx) {
if (!provider) throw new JobError('media.provider.unavailable', 'No durable provider is configured and healthy for the chunks', { retryAfterS: 1800 });
if (!provider) throw new JobError('media.provider.unavailable', 'No durable provider is configured and healthy for the chunks', { retryAfterS: 1800 });
out.provider = provider;
out.provider = provider;
const vodStorage = require('../vod/vod-storage');
const vodStorage = require('../vod/vod-storage');
+ const cache = require('../objects/timeline-cache');
+ let localRemaining = Math.max(0, config.hls.localMaxBytes - cache.status().bytes);
const dir = derive.workDir(job.id);
const dir = derive.workDir(job.id);
try {
try {
for (const chunk of chunks) {
for (const chunk of chunks) {
@@ -156,7 +158,9 @@ async function run(job, ctx) {
@@ -156,7 +158,9 @@ async function run(job, ctx) {
const name = `p${timeline.segmentName(Number(chunk[0].seq))}`;
const name = `p${timeline.segmentName(Number(chunk[0].seq))}`;
const key = timeline.keyFor(src.id, rendition, name, sha.slice(0, 12));
const key = timeline.keyFor(src.id, rendition, name, sha.slice(0, 12));
// This node keeps a copy of the chunk where it kept copies of the segments.
// This node keeps a copy of the chunk where it kept copies of the segments.
- const keepLocal = chunk.some((r) => r.local_path && derive.existing(r.local_path));
+ const keepLocal = chunk.some((r) => r.local_path && derive.existing(r.local_path))
+ && localRemaining >= offset && cache.freeBytes() >= config.hls.localMinFreeBytes;
+ if (keepLocal) localRemaining -= offset;
const localPath = keepLocal ? timeline.localPathFor(src, rendition, name, sha.slice(0, 12)) : null;
const localPath = keepLocal ? timeline.localPathFor(src, rendition, name, sha.slice(0, 12)) : null;
let uploaded = false;
let uploaded = false;
let committed = false;
let committed = false;
@@ -167,7 +171,7 @@ async function run(job, ctx) {
@@ -167,7 +171,7 @@ async function run(job, ctx) {
aborted(ctx);
aborted(ctx);
// Durable means verified: the provider gives back exactly these bytes.
// Durable means verified: the provider gives back exactly these bytes.
const back = await vodStorage.sha256Object(provider, key);
const back = await vodStorage.sha256Object(provider, key);
- if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });
+ if (!back || back.sha256 !== sha || Number(back.size) !== offset) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });
if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }
if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }
aborted(ctx);
aborted(ctx);
committed = await timeline.markPacked(src.id, placed.map((p) => ({
committed = await timeline.markPacked(src.id, placed.map((p) => ({
@@ -29,6 +29,7 @@
@@ -29,6 +29,7 @@
const fs = require('fs');
const fs = require('fs');
const path = require('path');
const path = require('path');
const timeline = require('../objects/timeline');
const timeline = require('../objects/timeline');
+const config = require('../config');
const derive = require('./derive');
const derive = require('./derive');
const { JobError } = require('./queue');
const { JobError } = require('./queue');
@@ -96,6 +97,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -96,6 +97,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
const target = await timeline.durableProvider(src);
const target = await timeline.durableProvider(src);
// B2/R2 configured but none healthy (breakers open) is "not durable yet", retried later, never a quiet success.
// B2/R2 configured but none healthy (breakers open) is "not durable yet", retried later, never a quiet success.
const remoteConfigured = vodStorage.REMOTE_PROVIDERS.some((p) => vodStorage.providerConfigured(p));
const remoteConfigured = vodStorage.REMOTE_PROVIDERS.some((p) => vodStorage.providerConfigured(p));
+ const cache = require('../objects/timeline-cache');
+ // One tree scan per cut. Only this heavy job adds timeline files; charge every retained file against this
+ // snapshot and recheck free space for each file, instead of walking the growing tree for every segment.
+ let localRemaining = Math.max(0, config.hls.localMaxBytes - cache.status().bytes);
const rows = [];
const rows = [];
let uploaded = 0;
let uploaded = 0;
const failed = [];
const failed = [];
@@ -103,6 +108,7 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -103,6 +108,7 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
// again, so a cancelled run leaves neither untracked keys nor files.
// again, so a cancelled run leaves neither untracked keys nor files.
const stagedRemote = [];
const stagedRemote = [];
const stagedLocal = [];
const stagedLocal = [];
+ const dropLocal = [];
let committed = false;
let committed = false;
try {
try {
for (const f of files) {
for (const f of files) {
@@ -138,6 +144,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -138,6 +144,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);
if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);
await vodStorage.uploadFile(target, key, localPath, f.seq === 0 ? 'video/mp4' : 'video/iso.segment', { signal: ctx.signal });
await vodStorage.uploadFile(target, key, localPath, f.seq === 0 ? 'video/mp4' : 'video/iso.segment', { signal: ctx.signal });
stagedRemote.push({ provider: target, key });
stagedRemote.push({ provider: target, key });
+ const back = await vodStorage.sha256Object(target, key);
+ if (!back || back.sha256 !== sha || Number(back.size) !== size) throw new Error(`${target}:${key} failed size/sha256 readback`);
durable = target;
durable = target;
uploaded++;
uploaded++;
} catch (err) {
} catch (err) {
@@ -145,6 +153,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -145,6 +153,10 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
failed.push(`${f.name}: ${err.message}`);
failed.push(`${f.name}: ${err.message}`);
}
}
}
}
+ if (durable && (localRemaining < size || cache.freeBytes() < config.hls.localMinFreeBytes)) {
+ dropLocal.push(localPath);
+ localPath = null;
+ } else if (localPath) localRemaining = Math.max(0, localRemaining - size);
rows.push({
rows.push({
seq: f.seq, name: f.name, start_ms: f.start_ms, end_ms: f.end_ms, keyframe_ms: f.keyframe_ms, key, local_path: localPath,
seq: f.seq, name: f.name, start_ms: f.start_ms, end_ms: f.end_ms, keyframe_ms: f.keyframe_ms, key, local_path: localPath,
durable_provider: durable, packed_object_id: null, byte_offset: null, byte_length: size, sha256: sha,
durable_provider: durable, packed_object_id: null, byte_offset: null, byte_length: size, sha256: sha,
@@ -163,6 +175,11 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -163,6 +175,11 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
throw err;
throw err;
}
}
committed = true;
committed = true;
+ // A published durable row now resolves through the router. Remove the staged local copy only after commit;
+ // another row may share this content-addressed path, in which case it remains its cache until a sweep.
+ for (const p of dropLocal) {
+ if (!await timeline.isNamed({ localPath: p })) { try { fs.unlinkSync(p); } catch { /* sweep reports orphan */ } }
+ }
// The master playlist reads these without touching bytes; a failure here leaves the timeline correct and only
// The master playlist reads these without touching bytes; a failure here leaves the timeline correct and only
// the variant's RESOLUTION/CODECS absent, so it never fails the cut.
// the variant's RESOLUTION/CODECS absent, so it never fails the cut.
if (meta) {
if (meta) {
@@ -80,7 +80,7 @@ async function throttleOnce(map, objectId, queue) {
@@ -80,7 +80,7 @@ async function throttleOnce(map, objectId, queue) {
/** Queue object.cmaf for an object at most once per QUEUE_TTL_MS in this process. Never throws. */
/** Queue object.cmaf for an object at most once per QUEUE_TTL_MS in this process. Never throws. */
async function queueCmafOnce(appId, objectId) {
async function queueCmafOnce(appId, objectId) {
- await throttleOnce(_queuedCmaf, objectId, () => require('./timeline-queue').queueCmaf(appId, objectId));
+ await throttleOnce(_queuedCmaf, objectId, () => require('./timeline-queue').queueCmaf(appId, objectId, { lazy: true }));
}
}
/**
/**
@@ -914,6 +914,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
@@ -914,6 +914,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
for (const loc of decision.candidates || []) {
for (const loc of decision.candidates || []) {
if (loc.provider === 'local') {
if (loc.provider === 'local') {
if (!fs.existsSync(loc.key)) continue;
if (!fs.existsSync(loc.key)) continue;
+ require('./timeline-cache').served(loc.key);
if (packed) return sendSlice(req, res, { file: loc.key, ...slice });
if (packed) return sendSlice(req, res, { file: loc.key, ...slice });
return require('../public/routes').streamFileWithRange(req, res, loc.key, headers);
return require('../public/routes').streamFileWithRange(req, res, loc.key, headers);
}
}
@@ -1066,4 +1067,5 @@ module.exports.contentHandlers = [contentAuth, tenantCors, limits('media.object.
@@ -1066,4 +1067,5 @@ module.exports.contentHandlers = [contentAuth, tenantCors, limits('media.object.
// minute is 50 MB/s) and its session already bounds them; this stops only a runaway loop.
// minute is 50 MB/s) and its session already bounds them; this stops only a runaway loop.
module.exports.partHandlers = [mpWrite, tenantCors, limits('media.object.part', { minute: 600, hour: 20000 }), putPart];
module.exports.partHandlers = [mpWrite, tenantCors, limits('media.object.part', { minute: 600, hour: 20000 }), putPart];
module.exports.publicRouter = publicRouter;
module.exports.publicRouter = publicRouter;
+module.exports.sendSegmentRow = sendSegmentRow; // socket-free verification of durable ranged fallback
module.exports.INLINE = INLINE;
module.exports.INLINE = INLINE;
new file mode 100644
new file mode 100644
@@ -0,0 +1,104 @@
@@ -0,0 +1,104 @@
+/** Local CMAF cache. Durable timeline rows remain authoritative after their local copy is evicted. */
+'use strict';
+
+const fs = require('fs');
+const path = require('path');
+const config = require('../config');
+const db = require('../db/database');
+const timeline = require('./timeline');
+
+let evictions = 0;
+let sweeping = false;
+
+function freeBytes() {
+ try {
+ const st = fs.statfsSync(config.objects.path);
+ return Number(st.bavail) * Number(st.bsize);
+ } catch { return 0; } // Unknown free space must not authorize another local copy.
+}
+
+function files() {
+ const root = timeline.localRoot();
+ const out = [];
+ function visit(dir) {
+ let entries;
+ try { entries = fs.readdirSync(dir, { withFileTypes: true }); } catch { return; }
+ for (const entry of entries) {
+ const file = path.join(dir, entry.name);
+ if (entry.isDirectory()) visit(file);
+ else if (entry.isFile()) {
+ try { const st = fs.statSync(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }
+ catch { /* disappeared while scanning */ }
+ }
+ }
+ }
+ visit(root);
+ return out;
+}
+
+// status() walks every local timeline file, so readiness and /metrics read a copy at most 30 s old; a write decision
+// (shouldKeepLocal) and the sweep read it fresh, and a sweep that removed files drops the copy.
+const STATUS_TTL_MS = 30_000;
+let memo = null;
+function status({ fresh = false } = {}) {
+ const now = Date.now();
+ // Only the measured numbers are kept; the budget and the floor are read from config on every call.
+ if (fresh || !memo || now - memo.at >= STATUS_TTL_MS) memo = { at: now, bytes: files().reduce((n, f) => n + f.size, 0), free: freeBytes() };
+ return { bytes: memo.bytes, max_bytes: config.hls.localMaxBytes, free_bytes: memo.free, min_free_bytes: config.hls.localMinFreeBytes };
+}
+
+function shouldKeepLocal() {
+ const s = status({ fresh: true });
+ return s.max_bytes > 0 && s.bytes < s.max_bytes && s.free_bytes >= s.min_free_bytes;
+}
+
+// mtime is explicitly maintained on a local read. atime is mount-option dependent; mtime is portable and survives
+// process restarts. No database write is needed for each segment request. The sweep sorts one entry per physical file.
+function served(file, now = Date.now()) {
+ try { fs.utimesSync(file, now / 1000, now / 1000); } catch { /* an eviction may have just unlinked it */ }
+}
+
+async function sweep() {
+ if (sweeping) return { skipped: true };
+ sweeping = true;
+ try {
+ const entries = files().sort((a, b) => a.servedAt - b.servedAt);
+ let bytes = entries.reduce((n, f) => n + f.size, 0);
+ let free = freeBytes();
+ let removed = 0;
+ for (const file of entries) {
+ if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) break;
+ // A path can be shared by source, clip and multiple packed rows. All must name a verified durable
+ // provider before the path can go. The row's durable state is set only after size+sha readback.
+ const rows = await db.all('SELECT durability, durable_provider, key, sha256, packed_object_id, byte_offset, byte_length FROM media_timeline WHERE local_path = ?', [file.path]);
+ if (!rows.length || rows.some((r) => r.durability !== 'durable' || !r.durable_provider)) continue;
+ // Existing timelines predate the cache and may have been marked durable after a size-only upload.
+ // Read back every distinct immutable key before evicting, including the complete packed chunk.
+ const copies = new Map();
+ for (const r of rows) {
+ const id = `${r.durable_provider}\n${r.key}`;
+ const expected = r.packed_object_id || r.sha256;
+ if (copies.has(id)) continue;
+ const back = await require('../vod/vod-storage').sha256Object(r.durable_provider, r.key).catch(() => null);
+ copies.set(id, !!back && back.sha256 === expected && Number(back.size) === file.size);
+ }
+ if ([...copies.values()].some((ok) => !ok)) continue;
+ await db.getDb().tx(async () => {
+ const current = await db.all('SELECT durability, durable_provider FROM media_timeline WHERE local_path = ? FOR UPDATE', [file.path]);
+ if (!current.length || current.some((r) => r.durability !== 'durable' || !r.durable_provider)) return;
+ // Keep the unlink and location change in the same transaction. A failed unlink rolls back the
+ // UPDATE; readers already tolerate a missing local file and use the durable location.
+ fs.unlinkSync(file.path);
+ await db.run('UPDATE media_timeline SET local_path = NULL, updated_at = ov_now() WHERE local_path = ?', [file.path]);
+ removed++;
+ evictions++;
+ bytes -= file.size;
+ });
+ free = freeBytes();
+ }
+ if (removed) memo = null;
+ return { removed, ...status({ fresh: true }) };
+ } finally { sweeping = false; }
+}
+
+module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions };
@@ -26,6 +26,8 @@
@@ -26,6 +26,8 @@
const crypto = require('crypto');
const crypto = require('crypto');
const config = require('../config');
const config = require('../config');
+const lazyQueuedAt = [];
+let pendingLazy = 0;
/**
/**
* The identity of the source cut a rendition is made against: the first 16 hex of sha256 over the source rows' (seq,
* The identity of the source cut a rendition is made against: the first 16 hex of sha256 over the source rows' (seq,
@@ -37,19 +39,34 @@ function sourceVersion(rows) {
@@ -37,19 +39,34 @@ function sourceVersion(rows) {
return h.digest('hex').slice(0, 16);
return h.digest('hex').slice(0, 16);
}
}
-async function queueCmaf(appId, objectId, { segment_seconds = null } = {}) {
+async function queueCmaf(appId, objectId, { segment_seconds = null, lazy = false, now = Date.now() } = {}) {
if (!config.hls.enabled || !appId || !objectId) return null;
if (!config.hls.enabled || !appId || !objectId) return null;
const obj = await require('./model').getObject(objectId);
const obj = await require('./model').getObject(objectId);
if (!obj || obj.lifecycle_status !== 'ready' || !require('../jobs/derive').isMediaObject(obj)) return null;
if (!obj || obj.lifecycle_status !== 'ready' || !require('../jobs/derive').isMediaObject(obj)) return null;
if (await require('./timeline').has(obj.id)) return null;
if (await require('./timeline').has(obj.id)) return null;
+ // Count persisted jobs (including those from before a restart) and successful queues in this process. Reserve
+ // synchronously after the count so concurrent reads cannot overshoot it while their enqueues are in flight.
+ let recent = 0;
+ if (lazy) {
+ while (lazyQueuedAt.length && lazyQueuedAt[0] <= now - 3600000) lazyQueuedAt.shift();
+ const row = await require('../db/database').get("SELECT COUNT(*) AS n FROM media_jobs WHERE created_by = 'system:timeline-backfill' AND created_at >= ov_now_iso('-1 hour')");
+ recent = Number(row.n);
+ }
const queue = require('../jobs/queue');
const queue = require('../jobs/queue');
const params = require('../jobs/cmaf').spec.validate({ appId, obj, params: segment_seconds != null ? { segment_seconds } : {} });
const params = require('../jobs/cmaf').spec.validate({ appId, obj, params: segment_seconds != null ? { segment_seconds } : {} });
- const r = await queue.enqueue({
- appId, type: 'object.cmaf', objectId: obj.id, params,
- dedupeActive: true, idempotencyKey: `object.cmaf:${obj.id}`, createdBy: 'system:timeline',
- });
- if (r.created) require('../jobs/worker').kick();
- return r.job ? r.job.id : null;
+ if (lazy) {
+ if (Math.max(recent, lazyQueuedAt.length) + pendingLazy >= config.hls.backfillPerHour) return null;
+ pendingLazy++;
+ }
+ try {
+ const r = await queue.enqueue({
+ appId, type: 'object.cmaf', objectId: obj.id, params,
+ dedupeActive: true, idempotencyKey: `object.cmaf:${obj.id}`, createdBy: lazy ? 'system:timeline-backfill' : 'system:timeline',
+ });
+ if (r.created && lazy) lazyQueuedAt.push(now);
+ if (r.created) require('../jobs/worker').kick();
+ return r.job ? r.job.id : null;
+ } finally { if (lazy) pendingLazy--; }
}
}
async function queuePack(appId, objectId, { rendition = null } = {}) {
async function queuePack(appId, objectId, { rendition = null } = {}) {
@@ -56,6 +56,9 @@ function storageDirs(config) {
@@ -56,6 +56,9 @@ function storageDirs(config) {
function domainMetrics(registry, { db, recorder, events }) {
function domainMetrics(registry, { db, recorder, events }) {
registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });
registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });
+ const timelineCache = require('./objects/timeline-cache');
+ registry.gauge({ name: 'media_timeline_local_bytes', help: 'Bytes used by local timeline files, including orphan files', collect: () => timelineCache.status().bytes });
+ registry.gauge({ name: 'media_timeline_local_evictions', help: 'Local timeline files evicted since process start', collect: () => timelineCache.evictions() });
// target vod: media_tier_decisions (VODs); object: media_object_tier_decisions (native v2 objects), whose
// target vod: media_tier_decisions (VODs); object: media_object_tier_decisions (native v2 objects), whose
// outcome dry_run is what the object sweep would have done while its activation gate is off.
// outcome dry_run is what the object sweep would have done while its activation gate is off.
registry.gauge({
registry.gauge({
@@ -136,6 +139,12 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
@@ -136,6 +139,12 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
for (const [name, dir] of Object.entries(storageDirs(config))) {
for (const [name, dir] of Object.entries(storageDirs(config))) {
checks.push({ name: `storage_${name}`, required: true, check: () => writable(dir) });
checks.push({ name: `storage_${name}`, required: true, check: () => writable(dir) });
}
}
+ checks.push({ name: 'timeline_cache', required: false, check: () => {
+ if (!config.hls.enabled) return { skipped: 'MEDIA_HLS_ENABLED off: no timeline is cut, nothing is cached' };
+ const s = require('./objects/timeline-cache').status();
+ const ok = s.bytes <= s.max_bytes && s.free_bytes >= s.min_free_bytes;
+ return { ok, ...(ok ? {} : { error: 'Timeline cache over budget or disk below free-space floor' }), detail: s };
+ } });
}
}
checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });
checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });
for (const name of require('./vod/vod-storage').REMOTE_PROVIDERS) {
for (const name of require('./vod/vod-storage').REMOTE_PROVIDERS) {
@@ -267,7 +267,8 @@ const http = require('http');
@@ -267,7 +267,8 @@ const http = require('http');
assert.strictEqual(r.status, 202);
assert.strictEqual(r.status, 202);
assert.ok(await waitFor(() => false, 50) || true);
assert.ok(await waitFor(() => false, 50) || true);
await worker.start();
await worker.start();
- assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded'), '?async=1 queues it for the worker');
+ // ffprobe then ffmpeg (the audio-only check, Media#93) under a parallel suite: give the real job 15 s.
+ assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded', 15000), '?async=1 queues it for the worker');
r = await call('POST', '/api/v2/live/jobs', { body: { type: 'thumbnail.regenerate', object_id: `legacy:live:vod:1` } });
r = await call('POST', '/api/v2/live/jobs', { body: { type: 'thumbnail.regenerate', object_id: `legacy:live:vod:1` } });
assert.deepStrictEqual([r.status, r.body.job.params], [202, { kind: 'vod', id: 1 }], 'the v2 API takes the object and finds the row');
assert.deepStrictEqual([r.status, r.body.job.params], [202, { kind: 'vod', id: 1 }], 'the v2 API takes the object and finds the row');
assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded'));
assert.ok(await waitFor(async () => (await queue.get(r.body.job.id)).status === 'succeeded'));
@@ -63,13 +63,13 @@ const { spawn } = require('child_process');
@@ -63,13 +63,13 @@ const { spawn } = require('child_process');
assert.strictEqual(body.service, 'media');
assert.strictEqual(body.service, 'media');
assert.strictEqual(body.recordings_in_progress, 1);
assert.strictEqual(body.recordings_in_progress, 1);
const names = Object.keys(body.checks).sort();
const names = Object.keys(body.checks).sort();
- assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'valkey']);
+ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'timeline_cache', 'valkey']);
// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optional
// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optional
// skip leaves the verified service ready.
// skip leaves the verified service ready.
assert.deepStrictEqual([body.checks.remote_r2.status, body.checks.remote_r2.reason], ['skipped', 'not configured']);
assert.deepStrictEqual([body.checks.remote_r2.status, body.checks.remote_r2.reason], ['skipped', 'not configured']);
- assert.deepStrictEqual(body.skipped.sort(), ['remote_r2', 'valkey'], 'Valkey too: no VALKEY_URL in this test');
+ assert.deepStrictEqual(body.skipped.sort(), ['remote_r2', 'timeline_cache', 'valkey'], 'Valkey too (no VALKEY_URL in this test), and the timeline cache while HLS is off');
for (const [n, c] of Object.entries(body.checks)) {
for (const [n, c] of Object.entries(body.checks)) {
- if (n === 'remote_r2' || n === 'valkey') continue;
+ if (n === 'remote_r2' || n === 'valkey' || n === 'timeline_cache') continue;
assert.strictEqual(c.status, 'ok', n);
assert.strictEqual(c.status, 'ok', n);
assert.strictEqual(typeof c.latency_ms, 'number');
assert.strictEqual(typeof c.latency_ms, 'number');
assert.ok(Date.parse(c.checked_at));
assert.ok(Date.parse(c.checked_at));
new file mode 100644
new file mode 100644
@@ -0,0 +1,230 @@
@@ -0,0 +1,230 @@
+'use strict';
+// No listening socket: real migrated PGlite rows, local files and a fake durable readback.
+const assert = require('assert');
+const crypto = require('crypto');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { EventEmitter } = require('events');
+
+(async () => {
+ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-cache-'));
+ process.env.OBJECTS_PATH = root;
+ process.env.MEDIA_HLS_ENABLED = '1';
+ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = '8';
+ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';
+ process.env.MEDIA_HLS_BACKFILL_PER_HOUR = '20';
+ const config = require('../server/config');
+ const db = require('../server/db/database');
+ const timeline = require('../server/objects/timeline');
+ const cache = require('../server/objects/timeline-cache');
+ const storage = require('../server/vod/vod-storage');
+ const blobs = new Map();
+ const realRead = storage.sha256Object;
+ storage.sha256Object = async (_provider, key) => {
+ const bytes = blobs.get(key);
+ return bytes ? { size: bytes.length, sha256: crypto.createHash('sha256').update(bytes).digest('hex') } : null;
+ };
+ try {
+ const obj = { app_id: 'live', id: 'cache-test' };
+ const make = async (seq, data, durable, at) => {
+ const bytes = Buffer.from(data);
+ const name = timeline.segmentName(seq);
+ const sha = crypto.createHash('sha256').update(bytes).digest('hex');
+ const key = `cache/${name}`;
+ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));
+ fs.mkdirSync(path.dirname(file), { recursive: true });
+ fs.writeFileSync(file, bytes);
+ cache.served(file, at);
+ if (durable) blobs.set(key, bytes);
+ await timeline.replace(`cache-${seq}`, 'source', [{ seq, name, start_ms: 0, end_ms: 1000, key, local_path: file,
+ durable_provider: durable ? 'b2' : null, packed_object_id: null, byte_offset: null,
+ byte_length: bytes.length, sha256: sha, durability: durable ? 'durable' : 'local' }]);
+ return file;
+ };
+ const old = await make(1, 'aaaaaa', true, 1000);
+ const recent = await make(2, 'bbbbbb', true, 3000);
+ const only = await make(3, 'cccccc', false, 500);
+ assert.strictEqual(cache.status().bytes, 18);
+ assert.strictEqual(cache.shouldKeepLocal(), false);
+ const first = await cache.sweep();
+ assert.strictEqual(first.removed, 2);
+ assert.strictEqual(fs.existsSync(old), false);
+ assert.strictEqual(fs.existsSync(recent), false);
+ assert.strictEqual(fs.existsSync(only), true, 'the only copy stays even above budget');
+ assert.strictEqual((await timeline.byName('cache-1', 'source', timeline.segmentName(1))).local_path, null);
+ assert.strictEqual(cache.evictions(), 2);
+ assert.strictEqual(cache.status().bytes, 6);
+ const orphanReport = await require('../server/vod/orphans-report').buildStorageReport({ providers: [] });
+ assert.ok(!orphanReport.unreferenced_local.some((f) => f.path.includes('.timeline/')),
+ 'evicted files are gone and the remaining local-only timeline file is still referenced');
+ const observability = require('../server/observability');
+ const gauges = new Map();
+ const registry = {
+ gauge: (spec) => { gauges.set(spec.name, spec.collect); return { set: () => {} }; },
+ counter: () => ({ inc: () => {} }), histogram: () => ({ observe: () => {} }),
+ };
+ const events = { status: async () => ({ enabled: false }) };
+ observability.domainMetrics(registry, { db, recorder: { activeCount: () => 0 }, events });
+ assert.strictEqual(gauges.get('media_timeline_local_bytes')(), 6);
+ assert.strictEqual(gauges.get('media_timeline_local_evictions')(), 2);
+ for (const section of ['vod', 'pastes', 'files', 'thumbnails']) {
+ if (section === 'vod') { config.vod.path = root; config.vod.clipsPath = root; }
+ else config[section].path = root;
+ }
+ const ready = observability.createMediaReadiness({ release: null, db, config,
+ auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 }, events,
+ remote: { configured: () => false } });
+ let report = await ready.run();
+ assert.strictEqual(report.checks.timeline_cache.status, 'ok');
+ assert.strictEqual(report.checks.timeline_cache.detail.bytes, 6);
+ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;
+ report = await ready.run();
+ assert.strictEqual(report.checks.timeline_cache.status, 'fail');
+ assert.ok(report.degraded.includes('timeline_cache'));
+ config.hls.localMinFreeBytes = 0;
+ const row = await timeline.byName('cache-1', 'source', timeline.segmentName(1));
+ assert.deepStrictEqual(timeline.locationsOf(row).map((x) => x.provider), ['b2']);
+ assert.deepStrictEqual(blobs.get(row.key), Buffer.from('aaaaaa'));
+ // A packed row's byte range remains the same after its local chunk is evicted.
+ const packed = Buffer.from('abcdef');
+ const packedSha = crypto.createHash('sha256').update(packed).digest('hex');
+ const packedFile = timeline.localPathFor(obj, 'source', 'p000001.m4s', packedSha.slice(0, 12));
+ fs.mkdirSync(path.dirname(packedFile), { recursive: true });
+ fs.writeFileSync(packedFile, packed);
+ blobs.set('cache/packed', packed);
+ await timeline.replace('cache-packed', 'source', [{ seq: 1, name: '000001.m4s', start_ms: 0, end_ms: 1000,
+ key: 'cache/packed', local_path: packedFile, durable_provider: 'b2', packed_object_id: packedSha,
+ byte_offset: 2, byte_length: 3, sha256: crypto.createHash('sha256').update('cde').digest('hex'), durability: 'durable' }]);
+ config.hls.localMaxBytes = 0;
+ await cache.sweep();
+ const packedRow = await timeline.byName('cache-packed', 'source', '000001.m4s');
+ assert.strictEqual(packedRow.local_path, null);
+ assert.deepStrictEqual(blobs.get(packedRow.key).subarray(Number(packedRow.byte_offset), Number(packedRow.byte_offset) + Number(packedRow.byte_length)), Buffer.from('cde'));
+ const router = require('../server/placement/router');
+ const realRoute = router.route;
+ const realSession = router.sessionFor;
+ const realPresign = storage.presignGet;
+ const realFetch = global.fetch;
+ router.route = async () => ({ provider: 'b2', candidates: [{ provider: 'b2', key: packedRow.key }] });
+ router.sessionFor = () => null;
+ storage.presignGet = async () => 'https://durable.invalid/chunk';
+ global.fetch = async (_url, opts) => {
+ const m = /^bytes=(\d+)-(\d+)$/.exec(opts.headers.range);
+ return new Response(packed.subarray(Number(m[1]), Number(m[2]) + 1), { status: 206 });
+ };
+ try {
+ const req = new EventEmitter();
+ req.headers = {};
+ const res = { destroyed: false, writeHead(code, headers) { this.code = code; this.headers = headers; }, end(body) { this.body = body; } };
+ await require('../server/objects/routes').sendSegmentRow(req, res, { obj: { id: 'cache-packed' }, closed: false }, packedRow, 'source');
+ assert.strictEqual(res.code, 200);
+ assert.deepStrictEqual(res.body, Buffer.from('cde'), 'the segment route reads the packed range from durable storage');
+ } finally {
+ router.route = realRoute; router.sessionFor = realSession; storage.presignGet = realPresign; global.fetch = realFetch;
+ }
+ assert.strictEqual(fs.existsSync(only), true);
+ console.log('timeline cache: oldest verified copies and packed range evicted; sole copy retained');
+
+ // The CMAF publisher stages normally, verifies the upload, then publishes remote-only rows below the floor.
+ const publishDir = path.join(root, 'publish');
+ fs.mkdirSync(publishDir);
+ fs.writeFileSync(path.join(publishDir, 'init.mp4'), 'init');
+ fs.writeFileSync(path.join(publishDir, '000001.m4s'), 'media');
+ const realProvider = timeline.durableProvider;
+ const realAvailable = storage.providerAvailable;
+ const realUpload = storage.uploadFile;
+ const timelineQueue = require('../server/objects/timeline-queue');
+ const realQueuePack = timelineQueue.queuePack;
+ timeline.durableProvider = async () => 'b2';
+ storage.providerAvailable = () => true;
+ storage.uploadFile = async (_provider, key, file) => { blobs.set(key, fs.readFileSync(file)); };
+ timelineQueue.queuePack = async () => null;
+ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;
+ try {
+ const published = await require('../server/jobs/segments').publishRendition({ job: { id: 'cache-publish', app_id: 'live' },
+ ctx: { signal: new AbortController().signal }, src: { id: 'cache-publish', app_id: 'live' },
+ rendition: 'source', out: publishDir, list: [{ name: '000001.m4s', seconds: 1 }] });
+ assert.strictEqual(published.durable, 2);
+ assert.strictEqual(published.local_only, 0);
+ const rows = await timeline.list('cache-publish');
+ assert.ok(rows.every((r) => r.local_path === null && r.durable_provider === 'b2'));
+ assert.ok(rows.every((r) => blobs.has(r.key)));
+ } finally {
+ timeline.durableProvider = realProvider; storage.providerAvailable = realAvailable;
+ storage.uploadFile = realUpload; timelineQueue.queuePack = realQueuePack;
+ config.hls.localMinFreeBytes = 0;
+ }
+ // Packing below the floor uploads and verifies its chunk, then points every row at durable bytes only.
+ const packRows = [];
+ for (let seq = 1; seq <= 2; seq++) {
+ const bytes = Buffer.from(seq === 1 ? 'first' : 'second');
+ const name = timeline.segmentName(seq);
+ const sha = crypto.createHash('sha256').update(bytes).digest('hex');
+ const key = `cache-pack/${name}`;
+ const file = timeline.localPathFor({ app_id: 'live', id: 'cache-pack' }, 'source', name, sha.slice(0, 12));
+ fs.mkdirSync(path.dirname(file), { recursive: true });
+ fs.writeFileSync(file, bytes);
+ blobs.set(key, bytes);
+ packRows.push({ seq, name, start_ms: (seq - 1) * 6000, end_ms: seq * 6000, keyframe_ms: (seq - 1) * 6000,
+ key, local_path: file, durable_provider: 'b2', packed_object_id: null, byte_offset: null,
+ byte_length: bytes.length, sha256: sha, durability: 'durable' });
+ }
+ await timeline.replace('cache-pack', 'source', packRows);
+ const derive = require('../server/jobs/derive');
+ const realLoad = derive.loadSource;
+ timeline.durableProvider = async () => 'b2';
+ storage.providerAvailable = () => true;
+ storage.uploadFile = async (_provider, key, file) => { blobs.set(key, fs.readFileSync(file)); };
+ derive.loadSource = async () => ({ id: 'cache-pack', app_id: 'live' });
+ config.hls.localMinFreeBytes = Number.MAX_SAFE_INTEGER;
+ try {
+ const result = await require('../server/jobs/pack').spec.run({ id: 'cache-pack-job', params: { target_seconds: 10 } },
+ { signal: new AbortController().signal });
+ assert.strictEqual(result.packs, 1);
+ const packedRows = await timeline.list('cache-pack');
+ assert.ok(packedRows.every((r) => r.local_path === null && r.packed_object_id));
+ assert.ok(packRows.every((r) => !fs.existsSync(r.local_path)));
+ assert.deepStrictEqual(blobs.get(packedRows[0].key), Buffer.from('firstsecond'));
+ } finally {
+ derive.loadSource = realLoad; timeline.durableProvider = realProvider;
+ storage.providerAvailable = realAvailable; storage.uploadFile = realUpload;
+ config.hls.localMinFreeBytes = 0;
+ }
+
+ const model = require('../server/objects/model');
+ const queue = require('../server/jobs/queue');
+ const worker = require('../server/jobs/worker');
+ const realEnqueue = queue.enqueue;
+ const realKick = worker.kick;
+ let queued = 0;
+ queue.enqueue = async () => ({ created: true, job: { id: `job-${++queued}` } });
+ worker.kick = () => {};
+ await db.upsertApp({ app_id: 'live', api_key: 'cache-key' });
+ const queueCmaf = require('../server/objects/timeline-queue').queueCmaf;
+ const base = Date.now();
+ try {
+ for (let n = 0; n < 21; n++) {
+ const id = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });
+ const result = await queueCmaf('live', id, { lazy: true, now: base });
+ assert.strictEqual(!!result, n < 20);
+ }
+ assert.strictEqual(queued, 20);
+ const id = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });
+ assert.ok(await queueCmaf('live', id), 'finalize bypasses the lazy cap');
+ assert.strictEqual(queued, 21);
+ const realGet = db.get;
+ db.get = async (sql, ...args) => sql.includes("created_by = 'system:timeline-backfill'") ? { n: 20 } : realGet(sql, ...args);
+ try {
+ const later = await model.createObject({ app_id: 'live', kind: 'vod', mime_type: 'video/mp4', lifecycle_status: 'ready', size_bytes: 10 });
+ assert.strictEqual(await queueCmaf('live', later, { lazy: true, now: base + 3600001 }), null,
+ 'persisted job count still caps lazy queues after the in-memory hour rolls over');
+ } finally { db.get = realGet; }
+ } finally { queue.enqueue = realEnqueue; worker.kick = realKick; }
+ console.log('timeline cache: lazy queue capped at 20 per hour; finalize bypasses cap');
+ } finally {
+ storage.sha256Object = realRead;
+ await db.close();
+ fs.rmSync(root, { recursive: true, force: true });
+ }
+})().catch((err) => { console.error(err); process.exitCode = 1; });
@@ -185,7 +185,7 @@ const express = require('express');
@@ -185,7 +185,7 @@ const express = require('express');
await call('GET', `/api/v2/live/objects/${old}/download?format=json`);
await call('GET', `/api/v2/live/objects/${old}/download?format=json`);
cmaf = await jobs('object.cmaf', old);
cmaf = await jobs('object.cmaf', old);
assert.strictEqual(cmaf.length, 1, 'queued once on demand');
assert.strictEqual(cmaf.length, 1, 'queued once on demand');
- assert.strictEqual(cmaf[0].created_by, 'system:timeline');
+ assert.strictEqual(cmaf[0].created_by, 'system:timeline-backfill');
assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0);
assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0);
await call('GET', `/api/v2/live/objects/${pdf}/download?format=json`);
await call('GET', `/api/v2/live/objects/${pdf}/download?format=json`);
assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0, 'a non-media download queues nothing');
assert.strictEqual((await jobs('object.cmaf', pdf)).length, 0, 'a non-media download queues nothing');