openvibers/openvibe.media Commits

Generate a 720p rendition on demand and list it in the master playlist (F4 slice 1, off by default) (#42)

OP OpenVibers committed 5 Oct 2026 · 166fe229b8 · 17 files, +1074 −195

## What changed

**New**
- `server/jobs/rendition.js` — the `rendition.create` heavy job: transcodes one 720p H.264/AAC rung of a ready media object (libx264, source timeline boundaries forced as keyframes), skips a source already ≤720p, refuses a held object / a source with no video, publishes rows via `timeline.replace(objectId, rendition, rows, {jobId})`, durable copy via the placement router, then queues `object.pack` for the rendition.
- `server/jobs/segments.js` — shared "publish a produced rendition" module (parse the HLS output, content-addressed local + router-chosen durable placement, one-transaction commit, two-phase stale-byte deletion, pack enqueue), extracted verbatim from `object.cmaf` so both jobs share it.
- `test/rendition-job.test.js` — job registration, lazy enqueue/dedupe, 720p rows, master playlist + variant playback, idempotent rerun, ≤720p skip, held object, both flags off.

**Modified**
- `server/jobs/cmaf.js` — run() now calls the shared `segments.publishRendition`; behaviour unchanged (its `parsePlaylist` export kept for tests).
- `server/jobs/types.js` / `server/jobs/worker.js` — register `rendition.create` (heavy) and list it in the lane docs.
- `server/config.js` — new `hls.renditions` from `MEDIA_RENDITIONS` (default off).
- `server/objects/timeline-queue.js` — `queueRendition()` (ladder loop, `dedupeActive` + `rendition.create:<id>:<name>` key, skips virtual clips), and `queuePack(..., { rendition })`.
- `server/objects/hls.js` — `renditionsFor()` (source + every rendition whose rows exist) and throttled `queueRenditionsOnce()`; `discoveryForObject` queues missing renditions.
- `server/objects/routes.js` — `master.m3u8` lists all existing renditions and queues missing ones; segment serving factored into `sendSegmentRow`; new `/:id/:rendition/index.m3u8` and `/:id/:rendition/:name` routes so a listed variant actually plays.
- `test/media-timeline.test.js` — store-level two-rendition master playlist, and a route block proving a rendition is listed/served only once its rows exist.
- `README.md`, `docs/object-model.md` (job table + env table), `docs/media-fabric.md` (§5 F4 slice 1, §10 F4), `STATUS.json` — documentation.

## Tests

`ov test test/rendition-job.test.js test/media-timeline.test.js test/timeline-queue.test.js test/clip-timeline.test.js test/jobs.test.js test/objects-api.test.js` → **6 passed** (23 s). Then the full reachable set, `ov test` (no args) → **69 passed, 6 skipped, 0 failed** (178 s); a final `ov test` after the last edits → 2 passed / 73 skipped (unchanged-since-green). `STATUS.json` parses.

## Notes / not done

- The brief's decision section says the rung is **`720p`** (the test paragraph said "480p"); I implemented 720p.
- The master playlist keeps its existing `BANDWIDTH`/`AVERAGE-BANDWIDTH` writer; I did **not** add `RESOLUTION`/`CODECS`. The rows carry no resolution/codec fields, and emitting them "from the produced segments" would mean parsing the init segment — out of scope for this slice, and the tests assert per-variant `BANDWIDTH`.
- Rendition segment/playlist serving was added (the brief only named `master.m3u8`); without it the listed variant would 404.
- I refactored `cmaf.js` onto the new shared `segments.js` rather than duplicating ~110 lines; `media-timeline.test.js` exercises it and passes.
- Nothing was committed, pushed, deployed, or installed.
- The `ov test` selector initially warned that no test statically reached `server/jobs/rendition.js` (it's only required la

Changes

ddiff --git a/README.md b/README.md
iindex bfcfaaa..eca48ee 100644
---- a/README.md
++++ b/README.md
@@@ -110,6 +110,8 @@ server/
invariant.scan (proposes split/remux), object.split, object.remux, object.waveform, object.sprite,
object.cmaf (CMAF/HLS segments of a finished video into media_timeline; MEDIA_HLS_ENABLED),
object.pack (durable segments packed into ~60 s chunk objects; MEDIA_HLS_ENABLED),
++ rendition.create (an on-demand 720p H.264/AAC rendition of a media object into media_timeline;
++ MEDIA_HLS_ENABLED + MEDIA_RENDITIONS),
vod.finalize (orphans and failed finalizes, with backoff),
vod.duration.reconcile (stored vs measured durations, local and B2/R2),
storage.orphans.scan (monthly storage orphan report, service-wide, report only),
@@@ -463,7 +465,7 @@ Counters are per process (a restart forgets them). `test/actor-limits.test.js`.
| `GET /live/:sel/transcript.json` | **transcript + AI timeline API** — full audio-transcription log and AI overview timeline. `:sel` = slot id / slot slug → slot-scoped sessions; **`@username`** → user-scoped (all their slots, works while offline). Returns `{ live, current, sessions[] (each: ai_overview, transcript, duration…), streamer (overview + stream memories), user }`. `?limit=1..50` sessions (default 10), `?app=`. Cached **30s** (that's the rate limit), CORS-open. |
| `GET /live/:sel/chat-insight.json` | **chat insight API** — a user's chat-related AI insight/timeline (`:sel` = `@username` or numeric user id): today-vs-alltime chat overviews, condensed memory, event timeline, plus their streamer overview + stream memories when they stream. Proxied from the app's public chat-AI API over loopback; cached **30s**, CORS-open. |
| `GET /v/:id/transcript.json` | **VOD transcript API** — transcript + AI overview for one existing VOD id (`{ vod_id, title, duration_seconds, ai_overview, transcript, ai_analyzed_at }`). Private VODs → 404. Cached 30s, CORS-open. |
--| `GET /o/:id/master.m3u8`, `/o/:id/source/index.m3u8`, `/o/:id/source/{init.mp4,NNNNNN.m4s}` | segment-native video (F3.1), only with `MEDIA_HLS_ENABLED=1` (off by default; off, these 404): HLS playlists written from the object's `media_timeline` rows (VOD, `#EXT-X-ENDLIST`) and its CMAF segments (Range; a segment packed by `object.pack` (F3.3) is a byte range of its chunk). Same check as `GET /o/:id` (and, like it, no `Access-Control-Allow-Origin` on a private or sandbox answer): private and sandbox objects only with the object's `?exp&sig` (`/api/v2/:app/objects/:id/download?format=json` → `hls_url`, a playlist token valid `MEDIA_HLS_PLAYLIST_TTL_S` (6 h; 60 s to 12 h) that no other route accepts; a download signature works too), carried onto every URI; deleted 410. The `object.cmaf` job makes the timeline; a recording's finalize queues it, and `download?format=json` queues it for an object that has none yet; the v1 VOD and clip answers carry the same `hls_url` (and `hls_expires_at` when signed). `noindex` |
++| `GET /o/:id/master.m3u8`, `/o/:id/source/index.m3u8`, `/o/:id/source/{init.mp4,NNNNNN.m4s}`, `/o/:id/<rendition>/…` | segment-native video (F3.1), only with `MEDIA_HLS_ENABLED=1` (off by default; off, these 404): HLS playlists written from the object's `media_timeline` rows (VOD, `#EXT-X-ENDLIST`) and its CMAF segments (Range; a segment packed by `object.pack` (F3.3) is a byte range of its chunk). The master playlist lists one variant per rendition whose rows exist — the `source` cut plus, with `MEDIA_RENDITIONS=1`, an on-demand 720p rung cut by `rendition.create` and queued lazily by this route and by `download?format=json` (F4 slice 1, docs/media-fabric.md §5). Same check as `GET /o/:id` (and, like it, no `Access-Control-Allow-Origin` on a private or sandbox answer): private and sandbox objects only with the object's `?exp&sig` (`/api/v2/:app/objects/:id/download?format=json` → `hls_url`, a playlist token valid `MEDIA_HLS_PLAYLIST_TTL_S` (6 h; 60 s to 12 h) that no other route accepts; a download signature works too), carried onto every URI; deleted 410. The `object.cmaf` job makes the timeline; a recording's finalize queues it, and `download?format=json` queues it for an object that has none yet; the v1 VOD and clip answers carry the same `hls_url` (and `hls_expires_at` when signed). `noindex` |
| `GET /robots.txt` | crawler policy (`openvibe-shared/seo.robotsTxt`: AI and search crawlers named, `/api/`, `/auth/`, `/internal/`, `/o/`, `/metrics`, `/live/`, `/me` disallowed) and the sitemap |
| `GET /sitemap.xml` | the media index and the watch pages Media is the canonical home of: public, playable VODs and clips of apps other than Live (Live lists its own `/vod` and `/clip` pages), never AI clips, private, unlisted, sandbox, recording or failed items; thumbnails as image entries. Every listed page renders `index, follow` |
| `GET /llms.txt` | a map of the site for language-model crawlers |
ddiff --git a/STATUS.json b/STATUS.json
iindex 0d1cb79..9c72064 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -41,6 +41,7 @@
"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",
"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"
],
ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex 46a5e3c..765ae30 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -233,6 +233,17 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
- **codec economics:** for a viral VOD, an AV1 rendition is produced when the egress/CDN bytes it saves exceed its encode
cost; for a cold VOD, never;
- compute is placed by the Compute fabric (cheapest available CPU/GPU, `openvibe-sdk/placement`).
++- **F4 slice 1, shipped: an on-demand rendition.** When a client asks a ready media object's master playlist or its
++ JSON download and a ladder rung is missing, `rendition.create` (heavy) is queued lazily — one job per object and
++ rendition, an idempotency key answering a later request with the same job. This slice ships one extra rung, `720p`
++ (H.264/AAC, libx264), transcoded from the source (or its timeline) with the source timeline's segment boundaries
++ forced as keyframes so a switch lands on one; a source already 720p or smaller is skipped, and a source with no video
++ stream is refused. The rows are published exactly as the source cut's (content-addressed placement, the placement
++ router's durable provider, one transaction, then the previous cut's unnamed bytes deleted; object.pack is queued for
++ the rendition) and the source object is untouched. Behind `MEDIA_RENDITIONS`, which also needs `MEDIA_HLS_ENABLED`:
++ off by default, no new rendition is cut or queued (one already produced keeps serving). The master playlist lists a
++ rendition only once its rows exist, and the variant plays at `/o/:id/<rendition>/index.m3u8`. Keep-vs-regenerate
++ economics, AV1 and compute placement stay open.
## 6. Placement: one engine, every class, down to the segment bucket
@@@ -389,7 +400,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
playlist), materialized clips referencing the source's chunks, captions on the
timeline (the demand rollups already record reads per timeline segment, F2.4); Live's player moves to HLS.
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
-- VODs, compute placement.
++ VODs, compute placement. (Slice 1 shipped: one on-demand 720p rung, queued by the master playlist / JSON download
++ and cut into the timeline; the economics, AV1 and compute placement are still open.)
- **F5 multi-CDN delivery:** Bunny path (B2 origin), R2 custom-domain path, OpenVibe edge first up to capacity, route
epochs, canary shifts with auto-revert, player QoE beacons, seek SLOs.
- **F6 cost truth:** bill imports, forecasts, budget guards, the owner dashboard.
ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex 9c06882..826187f 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -747,6 +747,7 @@ job over only with the token it read and, past boot, only while the lease is sti
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
| `object.cmaf` | heavy | Segment-native video (F3.1; only with `MEDIA_HLS_ENABLED`, else `409 media.hls.disabled`): a stream copy of a ready vod/clip/video object into CMAF segments on its own keyframes (`init.mp4` + `NNNNNN.m4s`; `segment_seconds` 1–10, default 4: a segment ends at the first keyframe after it), stored under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with a durable provider, as `<object>/source/<version>/<name>` on the provider the placement router picks (`timeline.durableProvider`: the source's own remote copy first, B2 before R2; `version` = the sha prefix: a re-cut writes beside the old bytes); one `media_timeline` row per segment (times, size, sha256, `durability`), all written in one transaction before the old bytes are deleted. The source is untouched. A rerun reuses rows whose sha256 matches and uploads only what is not durable; a failed upload leaves its row `local` and retries the job; an abort removes staged files and keys. Result `{ source_id, rendition, segments, duration_ms, bytes, durable, local_only, rows: { inserted, updated, unchanged, removed }, uploaded, master_path }`. Served at `/o/:id/master.m3u8` (see [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)). A packed row whose bytes it reproduces is kept as it is |
| `object.pack` | heavy | Packing after finalize (F3.3; only with `MEDIA_HLS_ENABLED`): runs of contiguous, durable, unpacked segments of one rendition (`rendition`, default `source`) concatenated into chunks of about `target_seconds` (10–300, 60), each verified on the router's durable provider (sha256 read back) as `<object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s` (and on this node where the segments were); the rows then name the chunk and their byte range in one transaction, and only then are the per-segment files and keys deleted. Never deletes a row; a rerun packs nothing; a held object is retried later (`media.object.held`); a re-cut meanwhile drops the chunk (`media.timeline.changed`). Result `{ source_id, rendition, packs, segments, bytes, provider, rows_packed, already_packed, waiting, removed_segments, left_behind }`. The segment URIs do not change: a packed segment is served as a byte range of its chunk |
++| `rendition.create` | heavy | An on-demand rendition (F4 slice 1; only with `MEDIA_HLS_ENABLED` and `MEDIA_RENDITIONS`, else `409 media.hls.disabled`/`media.renditions.disabled`): one extra ladder rung, `params.rendition` (`720p`), a libx264 + AAC re-encode of the source (or its timeline), with the source timeline's segment boundaries forced as keyframes so a switch lands on one. The rows are published exactly as `object.cmaf`'s (content-addressed placement, the router's durable provider, one transaction, then the previous cut's unnamed bytes deleted) and `object.pack` is queued for the rendition; the source is untouched. A source already the rung's height or smaller is skipped (`skipped: true`), as is an object whose rows already exist (`already: true`); a source with no video stream is `no_video` (permanent) and a held object is retried later. Queued lazily by `GET /o/:id/master.m3u8` and `GET …/download?format=json` (objects/hls.js), one job per object and rendition (idempotency key `rendition.create:<id>:<name>`). The master playlist lists a rendition only once its rows exist, and `/o/:id/<rendition>/index.m3u8` + `/o/:id/<rendition>/<name>` serve it. Result `{ source_id, rendition, target_height, segments, duration_ms, bytes, durable, local_only, rows, uploaded, master_path, skipped, already }` |
| `storage.move.cleanup` | light | Placement moves whose delete keeps failing: an object with three `failed` rows in `media_object_tier_decisions` in the last 7 days (a later `done` resets them) raises one `storage.alert` of kind `move_cleanup_failed`. Service-wide under the system app `_media`; report only. `params {}`. Result `{ scanned, alerted, objects }`. Scheduled every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never); tenants cannot enqueue it (403) |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
@@@ -985,6 +986,7 @@ deleted object's R2 copy stays (so its purge waits) until the gate is on again.
| `MEDIA_UPLOAD_RESERVATION_HOURS` | 72 | an upload's quota reservation lasts this long after its last step; then the hourly sweep fails it and frees its bytes |
| `MEDIA_NAMESPACE_MAX_CHILDREN` | 100 | namespaces a tenant may have below its root |
| `MEDIA_HLS_ENABLED` | off | `1` turns segment-native video on: the `object.cmaf` and `object.pack` jobs and the HLS routes `/o/:id/master.m3u8`, `/o/:id/source/…` (off: the job is refused, the routes 404) |
++| `MEDIA_RENDITIONS` | off | `1` (with `MEDIA_HLS_ENABLED`) turns on-demand renditions on: the `rendition.create` job and the extra master-playlist variants `/o/:id/<rendition>/…` (off: no new rendition is cut or queued; one already produced keeps serving) |
| `MEDIA_JOBS_ENABLED` | on | `0` stops the job worker |
| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |
| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane |
ddiff --git a/server/config.js b/server/config.js
iindex 6dfbee6..f516194 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -91,6 +91,9 @@ const config = {
enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
// 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
++ // variants. Also needs `enabled`; off by default, so no new rendition is cut or queued (an existing one serves).
++ renditions: ['1', 'true', 'on'].includes(String(process.env.MEDIA_RENDITIONS || '').toLowerCase()),
},
jobs: {
ddiff --git a/server/jobs/cmaf.js b/server/jobs/cmaf.js
iindex f81b570..f63a1fb 100644
---- a/server/jobs/cmaf.js
++++ b/server/jobs/cmaf.js
@@@ -15,13 +15,10 @@
* Bytes: each segment lands on this node's disk (OBJECTS_PATH/.timeline/…); with a durable provider it is uploaded under
* <object>/source/<version>/<name> (version = the sha's prefix) and its row says `durable` once the provider confirms the
* size. The provider is the placement router's choice (timeline.durableProvider: the source's own canonical copy first),
-- * never a hard-coded one. A segment already packed into a chunk (object.pack, F3.3) whose bytes are unchanged keeps its
-- * packed row as it is.
++ * never a hard-coded one. The row materialization and the two-phase commit live in server/jobs/segments.js, shared with
++ * rendition.create (F4).
* Idempotent: a rerun makes the same bytes (bitexact), reuses a row whose sha256 matches and its location as it is, and
-- * uploads only what is not durable yet. Two-phase: a re-cut writes beside the published segments and the rows commit in
-- * one transaction before the old bytes go, so playback never reads bytes that disagree with the playlist; a run that
-- * aborts or fails before the commit removes its staged files and keys. An upload that fails leaves its row `local` and
-- * fails the attempt, so the retry finishes it.
++ * uploads only what is not durable yet. A run that aborts or fails before the commit removes its staged files and keys.
* Refused: MEDIA_HLS_ENABLED off, a source that is not ready or not media (permanent), less free disk than the source
* size plus MEDIA_UPLOAD_MIN_FREE_MB (retried later), a codec fMP4 cannot carry (permanent).
*/
@@@ -32,9 +29,9 @@ const path = require('path');
const config = require('../config');
const timeline = require('../objects/timeline');
const derive = require('./derive');
++const segments = require('./segments');
const { JobError } = require('./queue');
--const MB = 1024 * 1024;
const DEFAULT_SEGMENT_SECONDS = 4;
function disabled() {
@@@ -54,35 +51,12 @@ function validate({ obj, params }) {
return p;
}
--/** The segments ffmpeg wrote, in order: [{ name, seconds }] from its playlist (#EXTINF before each URI). */
--function parsePlaylist(text) {
-- const out = [];
-- let dur = null;
-- for (const line of String(text).split(/\r?\n/)) {
-- const inf = /^#EXTINF:([\d.]+)/.exec(line);
-- if (inf) { dur = Number(inf[1]); continue; }
-- if (line && !line.startsWith('#') && dur != null) { out.push({ name: line.trim(), seconds: dur }); dur = null; }
-- }
-- return out;
--}
--
--function checkDisk(needBytes) {
-- const reserve = config.objects.uploadMinFreeMb * MB;
-- let free = Infinity;
-- try { const st = fs.statfsSync(config.objects.path); free = Number(st.bavail) * Number(st.bsize); } catch { /* unknown */ }
-- if (free < needBytes + reserve) {
-- throw new JobError('insufficient_disk', `Not enough free disk: need ${Math.ceil((needBytes + reserve) / MB)} MB, ${Math.floor(free / MB)} MB free`, { retryAfterS: 1800 });
-- }
--}
--
--const aborted = (ctx) => { if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted'); };
--
async function run(job, ctx) {
if (!config.hls.enabled) throw disabled();
const src = await derive.loadSource(job);
const source = await derive.resolveSource(src);
if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
-- checkDisk(Number(src.size_bytes) || 0);
++ segments.checkDisk(Number(src.size_bytes) || 0);
const rendition = timeline.SOURCE;
const dir = derive.workDir(job.id);
try {
@@@ -94,133 +68,21 @@ async function run(job, ctx) {
'-hls_playlist_type', 'vod', '-hls_segment_type', 'fmp4', '-hls_fmp4_init_filename', timeline.INIT_NAME,
'-start_number', '1', '-hls_segment_filename', path.join(out, '%06d.m4s'), path.join(out, 'index.m3u8')],
{ signal: ctx.signal, timeoutMs: derive.budgetMs(src.size_bytes) });
-- aborted(ctx);
++ segments.aborted(ctx);
const tail = String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300);
if (!r.ok) {
const unsupported = /not currently supported in container|could not find tag for codec|codec not supported/i.test(r.stderr);
throw new JobError(unsupported ? 'codec_unsupported' : 'ffmpeg_failed', `ffmpeg exited ${r.code}: ${tail}`, { permanent: unsupported });
}
-- const list = derive.existing(path.join(out, 'index.m3u8')) ? parsePlaylist(fs.readFileSync(path.join(out, 'index.m3u8'), 'utf8')) : [];
++ const list = derive.existing(path.join(out, 'index.m3u8')) ? segments.parsePlaylist(fs.readFileSync(path.join(out, 'index.m3u8'), 'utf8')) : [];
if (!list.length || !derive.existing(path.join(out, timeline.INIT_NAME))) throw new JobError('ffmpeg_failed', `ffmpeg wrote no segments: ${tail}`);
-- // Times on the object's own axis: cumulative EXTINF in ms, so each segment ends where the next starts.
-- const files = [{ seq: 0, name: timeline.INIT_NAME, start_ms: 0, end_ms: 0, keyframe_ms: null }];
-- let cum = 0;
-- list.forEach((s, i) => {
-- const start = Math.round(cum * 1000);
-- cum += s.seconds;
-- files.push({ seq: i + 1, name: timeline.segmentName(i + 1), file: s.name, start_ms: start, end_ms: Math.round(cum * 1000), keyframe_ms: start });
-- });
--
-- const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));
-- const vodStorage = require('../vod/vod-storage');
-- 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 rows = [];
-- let uploaded = 0;
-- const failed = [];
-- // Bytes written before the new rows commit are staged: an abort or a failure before the commit removes them
-- // again, so a cancelled run leaves neither untracked keys nor files.
-- const stagedRemote = [];
-- const stagedLocal = [];
-- let committed = false;
-- try {
-- for (const f of files) {
-- aborted(ctx);
-- const tmp = path.join(out, f.file || f.name);
-- if (!derive.existing(tmp)) throw new JobError('ffmpeg_failed', `ffmpeg listed ${f.file || f.name} but did not write it`);
-- const size = fs.statSync(tmp).size;
-- const sha = await derive.sha256File(tmp);
-- const old = prev.get(f.seq);
-- const match = !!old && old.sha256 === sha;
-- if (match && old.packed_object_id) {
-- // Already packed into a chunk (object.pack): the same bytes, so the row and its chunk stay as they are.
-- rows.push(Object.fromEntries(['seq', ...timeline.FIELDS].map((k) => [k, k === 'seq' ? f.seq : old[k]])));
-- continue;
-- }
-- // Content-addressed: the same bytes always land at the same versioned location, so a rerun rewrites
-- // nothing and a re-cut writes beside the old segment until the new rows are published.
-- const version = sha.slice(0, 12);
-- const dest = timeline.localPathFor(src, rendition, f.name, version);
-- const key = timeline.keyFor(src.id, rendition, f.name, version);
-- let localPath = dest;
-- let durable = match && old.durability === 'durable' ? old.durable_provider : null;
-- if (match && old.local_path && derive.existing(old.local_path)) {
-- localPath = old.local_path; // the previous copy is exactly these bytes: keep it and its location
-- } else {
-- fs.mkdirSync(path.dirname(dest), { recursive: true });
-- fs.renameSync(tmp, dest);
-- stagedLocal.push(dest);
-- }
-- if (!durable && !target && remoteConfigured) failed.push(`${f.name}: no healthy durable provider`);
-- if (!durable && target) {
-- try {
-- 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 });
-- durable = target;
-- uploaded++;
-- } catch (err) {
-- if (ctx.signal.aborted) throw ctx.signal.reason || err; // a cancelled upload is not "not durable yet"
-- failed.push(`${f.name}: ${err.message}`);
-- }
-- }
-- 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,
-- durability: durable ? 'durable' : 'local',
-- });
-- }
-- aborted(ctx);
-- let written;
-- try {
-- written = await timeline.replace(src.id, rendition, rows, { jobId: job.id, expect: prev });
-- } catch (err) {
-- // An object.pack committed after `prev` was read: these rows would name the segment keys it deleted.
-- if (err instanceof timeline.TimelineChanged) throw new JobError('media.timeline.changed', 'The timeline changed while cutting (a pack); a retry cuts against the new rows', { retryAfterS: 60 });
-- throw err;
-- }
-- committed = true;
-- // Phase two, only once the new rows are published: the bytes a previous cut named that this one does not.
-- // A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.
-- const named = new Set(rows.map((r) => `${r.key}\n${r.local_path || ''}`));
-- const stale = [...prev.values()].filter((old) => !named.has(`${old.key}\n${old.local_path || ''}`));
-- // A location any row still names — another object's, or the source's own (another rendition/seq, or a
-- // different local_path with the same key) — is kept: the old cut's rows go, the shared bytes stay, and the
-- // last naming object's removal deletes them. `isNamed` counts every object, no exception.
-- const { failed: stuck, kept } = await timeline.deleteBytes(stale);
-- if (stuck.length) console.warn(`[Cmaf] ${job.id}: ${stuck.length} old segment(s) left behind (the storage orphan report names them), first ${stuck[0].provider}:${stuck[0].key}`);
-- if (kept.length) console.warn(`[Cmaf] ${job.id}: ${kept.length} old segment location(s) kept — another object's rows name them`);
-- if (failed.length) {
-- throw new JobError('upload_failed', `${failed.length} segment(s) not durable yet (${failed[0].slice(0, 200)}); a retry uploads only those`, { retryAfterS: 600 });
-- }
-- const segs = rows.filter((x) => x.seq > 0);
-- // Packing follows the cut by itself (F3): once the rows and their bytes are committed, object.pack is
-- // queued for the object (dedupeActive joins an active pack; a later cut queues a new one,
-- // objects/timeline-queue.js). A queue error never fails the cut (the timeline is already published); a run
-- // whose upload failed is left to its retry.
-- try { await require('../objects/timeline-queue').queuePack(job.app_id, src.id); }
-- catch (err) { console.warn(`[Cmaf] ${job.id}: pack queue failed: ${err.message}`); }
-- return {
-- source_id: src.id, rendition, segments: segs.length, duration_ms: segs[segs.length - 1].end_ms,
-- bytes: rows.reduce((a, x) => a + x.byte_length, 0),
-- durable: rows.filter((x) => x.durability === 'durable').length, local_only: rows.filter((x) => x.durability === 'local').length,
-- rows: written, uploaded, master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true,
-- };
-- } catch (err) {
-- if (!committed) {
-- // Content-addressed: another run may have committed these very keys and files, so only unnamed ones go.
-- for (const k of stagedRemote) {
-- if (!await timeline.isNamed({ provider: k.provider, key: k.key }).catch(() => true)) await vodStorage.deleteObject(k.provider, k.key).catch(() => { /* best effort */ });
-- }
-- for (const p of stagedLocal) {
-- if (await timeline.isNamed({ localPath: p }).catch(() => true)) continue;
-- try { fs.unlinkSync(p); } catch { /* already gone */ }
-- }
-- }
-- throw err;
-- }
++ const published = await segments.publishRendition({ job, ctx, src, rendition, out, list, label: 'Cmaf' });
++ return {
++ source_id: src.id, rendition, segments: published.segments, duration_ms: published.duration_ms, bytes: published.bytes,
++ durable: published.durable, local_only: published.local_only, rows: published.rows, uploaded: published.uploaded,
++ master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true,
++ };
} finally {
derive.cleanupWork(job.id);
}
@@@ -228,5 +90,5 @@ async function run(job, ctx) {
module.exports = {
spec: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate, run },
-- parsePlaylist,
++ parsePlaylist: segments.parsePlaylist,
};
ddiff --git a/server/jobs/rendition.js b/server/jobs/rendition.js
nnew file mode 100644
iindex 0000000..ee01944
---- /dev/null
++++ b/server/jobs/rendition.js
@@@ -0,0 +1,148 @@
++/**
++ * OpenVibe.Media — job type rendition.create (heavy lane; docs/media-fabric.md §5, F4 slice 1)
++ *
++ * One extra rendition of a ready media object, transcoded on demand: a 720p H.264/AAC ladder rung (`params.rendition`,
++ * '720p'). It is a re-encode (libx264 + AAC), not a stream copy, of the source — or, equivalently, of the source
++ * timeline, whose segment boundaries are forced keyframes here so ABR switches line up. The rows go through
++ * timeline.replace(objectId, rendition, rows, { jobId }) (server/jobs/segments.js: the same content-addressed
++ * placement, durable upload through the placement router, two-phase commit and stale-byte deletion as object.cmaf), and
++ * object.pack is queued for the rendition once they are committed. The source object is never touched.
++ *
++ * params { rendition ('720p') }
++ * result { source_id, rendition, target_height, segments, duration_ms, bytes, durable, local_only,
++ * rows: { inserted, updated, unchanged, removed }, uploaded, master_path }
++ * result { …, skipped: true, reason } when the source is already this rendition's height or smaller (a still is not
++ * upscaled), or the rows already exist (skipped: false, already: true)
++ *
++ * Refused: MEDIA_HLS_ENABLED or MEDIA_RENDITIONS off, a source that is not ready or not media (permanent), an object
++ * under a hold (retried), a source with no video stream (permanent), less free disk than the source size plus
++ * MEDIA_UPLOAD_MIN_FREE_MB (retried later). Idempotent: a re-run whose bytes match reuses the rows and locations, and
++ * only uploads what is not durable yet; an abort or a failure before the commit removes what it staged.
++ *
++ * The master playlist (GET /o/:id/master.m3u8) lists a rendition only once its rows exist, and the route / /download
++ * enqueue this job lazily when one is missing (objects/hls.js, objects/timeline-queue.js): an idempotency key per
++ * object and rendition answers a later request with the same job.
++ */
++'use strict';
++
++const fs = require('fs');
++const path = require('path');
++const config = require('../config');
++const timeline = require('../objects/timeline');
++const derive = require('./derive');
++const segments = require('./segments');
++const { JobError } = require('./queue');
++
++const DEFAULT_SEGMENT_SECONDS = 4;
++const DEFAULT_RENDITION = '720p';
++// The ladder this slice ships: one rung. The name is the rendition key in media_timeline and the master playlist's URI
++// prefix (`720p/index.m3u8`); the value is the target height (the source's aspect ratio is kept, width even).
++const RENDITIONS = { '720p': 720 };
++
++function hlsDisabled() {
++ return new JobError('media.hls.disabled', 'Segment-native video is off on this server (MEDIA_HLS_ENABLED)', { permanent: true, status: 409 });
++}
++
++function disabled() {
++ return new JobError('media.renditions.disabled', 'On-demand renditions are off on this server (MEDIA_RENDITIONS)', { permanent: true, status: 409 });
++}
++
++function validate({ obj, params }) {
++ if (!config.hls.enabled) throw hlsDisabled();
++ if (!config.hls.renditions) throw disabled();
++ if (!obj) throw new JobError('media.job.invalid', 'rendition.create needs object_id', { permanent: true });
++ if (!derive.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video/audio objects) have renditions', { permanent: true });
++ const p = { ...(params || {}) };
++ const name = String(p.rendition || DEFAULT_RENDITION);
++ if (!Object.prototype.hasOwnProperty.call(RENDITIONS, name)) throw new JobError('media.job.invalid', `params.rendition must be one of ${Object.keys(RENDITIONS).join(', ')}`, { permanent: true });
++ p.rendition = name;
++ return p;
++}
++
++/** A re-encode is far slower than the stream copy derive.budgetMs budgets: 8× it, under the job's own 3 h cap. */
++function encodeBudgetMs(bytes) { return Math.min(3 * 3600 * 1000 - 60000, derive.budgetMs(bytes) * 8); }
++
++/** The target segment length: the caller's, else the source timeline's own average (1-10 s), else 4. */
++function segmentSeconds(job, srcRows) {
++ if (job.params.segment_seconds) return Number(job.params.segment_seconds);
++ if (srcRows.length) {
++ const avg = srcRows.reduce((a, r) => a + (Number(r.end_ms) - Number(r.start_ms)), 0) / srcRows.length;
++ const s = Math.round(avg / 1000);
++ if (s >= 1 && s <= 10) return s;
++ }
++ return DEFAULT_SEGMENT_SECONDS;
++}
++
++async function run(job, ctx) {
++ if (!config.hls.enabled) throw hlsDisabled();
++ if (!config.hls.renditions) throw disabled();
++ const src = await derive.loadSource(job);
++ const rendition = String(job.params.rendition || DEFAULT_RENDITION);
++ const targetHeight = RENDITIONS[rendition];
++ if (!targetHeight) throw new JobError('media.job.invalid', `${rendition} is not a rendition of this server`, { permanent: true });
++ if (await require('../objects/model').isHeld(src.id)) throw new JobError('media.object.held', 'The object is under a hold: no rendition is made', { retryAfterS: 6 * 3600 });
++ // Rows already committed (another run, or a previous request): nothing to do. A pack may have moved them since.
++ if (await timeline.has(src.id, rendition)) {
++ const rows = await timeline.segments(src.id, rendition);
++ return {
++ source_id: src.id, rendition, target_height: targetHeight, already: true, skipped: false,
++ segments: rows.length, duration_ms: rows.length ? Number(rows[rows.length - 1].end_ms) : 0,
++ bytes: rows.reduce((a, r) => a + Number(r.byte_length), 0), master_path: `/o/${src.id}/master.m3u8`,
++ };
++ }
++ const source = await derive.resolveSource(src);
++ if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
++ const info = await require('../vod/media-tools').probeVodInfo(source.input);
++ const video = (info.streams || []).find((s) => s.codec_type === 'video');
++ if (!video) throw new JobError('no_video', 'The source has no video stream to rendition', { permanent: true });
++ const height = Number(video.height) || 0;
++ // A source already this rendition's height or smaller is not upscaled (a re-encode would only cost bytes).
++ if (height > 0 && height <= targetHeight) {
++ return { source_id: src.id, rendition, target_height: targetHeight, skipped: true, reason: `the source is ${height}p, not above ${rendition}`, segments: 0, duration_ms: 0, bytes: 0 };
++ }
++ segments.checkDisk(Number(src.size_bytes) || 0);
++ const srcRows = await timeline.segments(src.id, timeline.SOURCE);
++ const segSec = segmentSeconds(job, srcRows);
++ // Force keyframes at the source timeline's segment boundaries (docs/media-fabric.md §3), so a switch to this
++ // rendition lands on a boundary. A list longer than 2000 boundaries is skipped: the argument would be unwieldy,
++ // and a source that long is packed long before it is re-encoded.
++ const boundaries = srcRows.length && srcRows.length <= 2000 ? srcRows.map((r) => (Number(r.start_ms) / 1000).toFixed(3)) : [];
++ const dir = derive.workDir(job.id);
++ try {
++ const out = path.join(dir, 'out');
++ fs.mkdirSync(out, { recursive: true });
++ const r = await derive.ffmpeg(['-y', '-nostdin', '-v', 'error', ...derive.inputArgs(source), '-i', source.input,
++ '-map', '0:v:0', '-map', '0:a:0?',
++ '-vf', `scale=-2:${targetHeight}`, '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '21', '-profile:v', 'main', '-pix_fmt', 'yuv420p',
++ '-c:a', 'aac', '-b:a', '128k', '-ac', '2',
++ ...(boundaries.length ? ['-force_key_frames', boundaries.join(',')] : []),
++ '-map_metadata', '-1',
++ '-f', 'hls', '-hls_time', String(segSec), '-hls_list_size', '0', '-hls_playlist_type', 'vod',
++ '-hls_segment_type', 'fmp4', '-hls_fmp4_init_filename', timeline.INIT_NAME,
++ '-start_number', '1', '-hls_segment_filename', path.join(out, '%06d.m4s'), path.join(out, 'index.m3u8')],
++ { signal: ctx.signal, timeoutMs: encodeBudgetMs(src.size_bytes) });
++ segments.aborted(ctx);
++ const tail = String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300);
++ if (!r.ok) {
++ const unsupported = /Unknown encoder|not currently supported in container|could not find tag for codec|codec not supported/i.test(r.stderr);
++ throw new JobError(unsupported ? 'codec_unsupported' : 'ffmpeg_failed', `ffmpeg exited ${r.code}: ${tail}`, { permanent: unsupported });
++ }
++ const list = derive.existing(path.join(out, 'index.m3u8')) ? segments.parsePlaylist(fs.readFileSync(path.join(out, 'index.m3u8'), 'utf8')) : [];
++ if (!list.length || !derive.existing(path.join(out, timeline.INIT_NAME))) throw new JobError('ffmpeg_failed', `ffmpeg wrote no segments: ${tail}`);
++
++ const published = await segments.publishRendition({ job, ctx, src, rendition, out, list, label: 'Rendition' });
++ return {
++ source_id: src.id, rendition, target_height: targetHeight, skipped: false, already: false,
++ segments: published.segments, duration_ms: published.duration_ms, bytes: published.bytes,
++ durable: published.durable, local_only: published.local_only, rows: published.rows, uploaded: published.uploaded,
++ master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true,
++ };
++ } finally {
++ derive.cleanupWork(job.id);
++ }
++}
++
++module.exports = {
++ spec: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate, run },
++ RENDITIONS, DEFAULT_RENDITION, renditionNames: () => Object.keys(RENDITIONS),
++};
ddiff --git a/server/jobs/segments.js b/server/jobs/segments.js
nnew file mode 100644
iindex 0000000..0990e5a
---- /dev/null
++++ b/server/jobs/segments.js
@@@ -0,0 +1,216 @@
++/**
++ * OpenVibe.Media — publishing a produced rendition's segments into the timeline (docs/media-fabric.md §3, F3.1/F4).
++ *
++ * Shared by the jobs that cut one rendition of a media object — object.cmaf's stream copy (server/jobs/cmaf.js) and
++ * rendition.create's transcode (server/jobs/rendition.js). Both run ffmpeg into an HLS fMP4 directory; this module
++ * turns that output into media_timeline rows and publishes them:
++ *
++ * parsePlaylist ffmpeg's index.m3u8 → [{ name, seconds }], in order.
++ * producedFiles those plus the init segment → [{ seq, name, file, start_ms, end_ms, keyframe_ms }], times on the
++ * object's own axis (cumulative EXTINF, so each segment ends where the next starts).
++ * publishRendition hash each file, place it content-addressed on this node (timeline.localPathFor/keyFor, versioned
++ * by the sha prefix) and — through the placement router's durable provider (timeline.durableProvider,
++ * purpose `durable`) — on B2/R2, then commit every row of the rendition in ONE transaction
++ * (timeline.replace). Only then are the bytes a previous cut named and this one does not deleted, and
++ * a location any row of any object still names is kept (F3.4; timeline.deleteBytes). A source re-cut
++ * also drops every other rendition's rows in that same transaction (their bytes are deleted after it),
++ * so a stale rendition never plays against the new source. The init segment's width/height/codecs are
++ * parsed (objects/mp4-codecs.js) and stored in the object's metadata under renditions.<name> for the
++ * master playlist. Finally object.pack is queued for the rendition so packing follows the cut by itself.
++ *
++ * Idempotent: a rerun makes the same bytes (bitexact), reuses a row whose sha256 matches and its location as it is, keeps
++ * a packed row whose chunk holds the same bytes, and uploads only what is not durable yet. Two-phase: the rows commit
++ * before the old bytes go, so playback never reads bytes that disagree with the playlist; a run that aborts or fails
++ * before the commit removes the files and keys it staged (content-addressed: never one another run already committed). A
++ * durable delete that fails leaves an orphan the storage report names, never bytes a live row still points at.
++ */
++'use strict';
++
++const fs = require('fs');
++const path = require('path');
++const timeline = require('../objects/timeline');
++const derive = require('./derive');
++const { JobError } = require('./queue');
++
++/** The segments ffmpeg wrote, in order: [{ name, seconds }] from its playlist (#EXTINF before each URI). */
++function parsePlaylist(text) {
++ const out = [];
++ let dur = null;
++ for (const line of String(text).split(/\r?\n/)) {
++ const inf = /^#EXTINF:([\d.]+)/.exec(line);
++ if (inf) { dur = Number(inf[1]); continue; }
++ if (line && !line.startsWith('#') && dur != null) { out.push({ name: line.trim(), seconds: dur }); dur = null; }
++ }
++ return out;
++}
++
++/** The produced files as timeline rows in the making: the init segment (seq 0) then each media segment, in order. */
++function producedFiles(list) {
++ const files = [{ seq: 0, name: timeline.INIT_NAME, start_ms: 0, end_ms: 0, keyframe_ms: null }];
++ let cum = 0;
++ list.forEach((s, i) => {
++ const start = Math.round(cum * 1000);
++ cum += s.seconds;
++ files.push({ seq: i + 1, name: timeline.segmentName(i + 1), file: s.name, start_ms: start, end_ms: Math.round(cum * 1000), keyframe_ms: start });
++ });
++ return files;
++}
++
++function checkDisk(needBytes) {
++ const reserve = require('../config').objects.uploadMinFreeMb * 1024 * 1024;
++ let free = Infinity;
++ try { const st = fs.statfsSync(require('../config').objects.path); free = Number(st.bavail) * Number(st.bsize); } catch { /* unknown */ }
++ if (free < needBytes + reserve) {
++ throw new JobError('insufficient_disk', `Not enough free disk: need ${Math.ceil((needBytes + reserve) / 1048576)} MB, ${Math.floor(free / 1048576)} MB free`, { retryAfterS: 1800 });
++ }
++}
++
++const aborted = (ctx) => { if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted'); };
++
++/**
++ * Store what the cut's init segment says (width/height/codecs) under the object's metadata.renditions.<name>, in one
++ * UPDATE: the source cut and a rendition job may finish together, and a read-modify-write would drop one of them.
++ */
++async function storeRenditionMeta(objectId, rendition, meta) {
++ const value = JSON.stringify({ width: meta.width, height: meta.height, codecs: meta.codecs });
++ // The expression reads the row's own column, so a concurrent writer's committed metadata is what it builds on.
++ const doc = `COALESCE(NULLIF(metadata, ''), '{}')::jsonb`;
++ await require('../db/database').run(`UPDATE media_objects SET metadata = jsonb_set(${doc}, '{renditions}',
++ COALESCE(${doc}->'renditions', '{}'::jsonb) || jsonb_build_object(?::text, ?::jsonb))::text, updated_at = ov_now()
++ WHERE id = ?`, [rendition, value, objectId]);
++}
++
++/**
++ * Publish `files` (producedFiles) of `rendition` for `src` from the work directory `out`. Returns the row counts and
++ * the totals the job result is built from. Throws JobError: the timeline changed under a pack, or a segment that is not
++ * durable yet. `label` names the job in warnings (e.g. 'Cmaf').
++ */
++async function publishRendition({ job, ctx, src, rendition, out, list, label = 'Segments' }) {
++ const files = producedFiles(list);
++ // Read the init segment while it is still in the work directory (the loop below moves it to its content-addressed
++ // location): its codecs/resolution go into the object's metadata for the master playlist.
++ let meta = null;
++ try { meta = require('../objects/mp4-codecs').parseInit(fs.readFileSync(path.join(out, timeline.INIT_NAME))); } catch { /* no attributes */ }
++ const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));
++ const vodStorage = require('../vod/vod-storage');
++ 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 rows = [];
++ let uploaded = 0;
++ const failed = [];
++ // Bytes written before the new rows commit are staged: an abort or a failure before the commit removes them
++ // again, so a cancelled run leaves neither untracked keys nor files.
++ const stagedRemote = [];
++ const stagedLocal = [];
++ let committed = false;
++ try {
++ for (const f of files) {
++ aborted(ctx);
++ const tmp = path.join(out, f.file || f.name);
++ if (!derive.existing(tmp)) throw new JobError('ffmpeg_failed', `ffmpeg listed ${f.file || f.name} but did not write it`);
++ const size = fs.statSync(tmp).size;
++ const sha = await derive.sha256File(tmp);
++ const old = prev.get(f.seq);
++ const match = !!old && old.sha256 === sha;
++ if (match && old.packed_object_id) {
++ // Already packed into a chunk (object.pack): the same bytes, so the row and its chunk stay as they are.
++ rows.push(Object.fromEntries(['seq', ...timeline.FIELDS].map((k) => [k, k === 'seq' ? f.seq : old[k]])));
++ continue;
++ }
++ // Content-addressed: the same bytes always land at the same versioned location, so a rerun rewrites
++ // nothing and a re-cut writes beside the old segment until the new rows are published.
++ const version = sha.slice(0, 12);
++ const dest = timeline.localPathFor(src, rendition, f.name, version);
++ const key = timeline.keyFor(src.id, rendition, f.name, version);
++ let localPath = dest;
++ let durable = match && old.durability === 'durable' ? old.durable_provider : null;
++ if (match && old.local_path && derive.existing(old.local_path)) {
++ localPath = old.local_path; // the previous copy is exactly these bytes: keep it and its location
++ } else {
++ fs.mkdirSync(path.dirname(dest), { recursive: true });
++ fs.renameSync(tmp, dest);
++ stagedLocal.push(dest);
++ }
++ if (!durable && !target && remoteConfigured) failed.push(`${f.name}: no healthy durable provider`);
++ if (!durable && target) {
++ try {
++ 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 });
++ durable = target;
++ uploaded++;
++ } catch (err) {
++ if (ctx.signal.aborted) throw ctx.signal.reason || err; // a cancelled upload is not "not durable yet"
++ failed.push(`${f.name}: ${err.message}`);
++ }
++ }
++ 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,
++ durability: durable ? 'durable' : 'local',
++ });
++ }
++ aborted(ctx);
++ let written;
++ try {
++ // A source re-cut (object.cmaf) drops every other rendition's rows here, in this same transaction: the
++ // rendition was cut against the old source and would otherwise play beside the new one.
++ written = await timeline.replace(src.id, rendition, rows, { jobId: job.id, expect: prev, dropOtherRenditions: rendition === timeline.SOURCE });
++ } catch (err) {
++ // An object.pack committed after `prev` was read: these rows would name the segment keys it deleted.
++ if (err instanceof timeline.TimelineChanged) throw new JobError('media.timeline.changed', 'The timeline changed while cutting (a pack); a retry cuts against the new rows', { retryAfterS: 60 });
++ throw err;
++ }
++ committed = true;
++ // 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) {
++ try { await storeRenditionMeta(src.id, rendition, meta); }
++ catch (err) { console.warn(`[${label}] ${job.id}: rendition metadata store failed: ${err.message}`); }
++ }
++ // Phase two, only once the new rows are published: the bytes a previous cut named that this one does not, and
++ // — after a source re-cut — the bytes every dropped rendition row named.
++ // A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.
++ const named = new Set(rows.map((r) => `${r.key}\n${r.local_path || ''}`));
++ const stale = [...prev.values()].filter((old) => !named.has(`${old.key}\n${old.local_path || ''}`)).concat(written.dropped || []);
++ // A location any row still names — another object's, or the source's own (another rendition/seq, or a
++ // different local_path with the same key) — is kept: the old cut's rows go, the shared bytes stay, and the
++ // last naming object's removal deletes them. `isNamed` counts every object, no exception.
++ const { failed: stuck, kept } = await timeline.deleteBytes(stale);
++ if (stuck.length) console.warn(`[${label}] ${job.id}: ${stuck.length} old segment(s) left behind (the storage orphan report names them), first ${stuck[0].provider}:${stuck[0].key}`);
++ if (kept.length) console.warn(`[${label}] ${job.id}: ${kept.length} old segment location(s) kept — another object's rows name them`);
++ if ((written.dropped || []).length) console.log(`[${label}] ${job.id}: dropped ${written.dropped.length} row(s) of an earlier rendition (the source was re-cut)`);
++ if (failed.length) {
++ throw new JobError('upload_failed', `${failed.length} segment(s) not durable yet (${failed[0].slice(0, 200)}); a retry uploads only those`, { retryAfterS: 600 });
++ }
++ const segs = rows.filter((x) => x.seq > 0);
++ // Packing follows the cut by itself (F3): once the rows and their bytes are committed, object.pack is queued
++ // for this rendition (dedupeActive joins an active pack; a later cut queues a new one, objects/timeline-queue.js).
++ // A queue error never fails the cut (the timeline is already published); a run whose upload failed is left to
++ // its retry.
++ try { await require('../objects/timeline-queue').queuePack(job.app_id, src.id, { rendition }); }
++ catch (err) { console.warn(`[${label}] ${job.id}: pack queue failed: ${err.message}`); }
++ return {
++ rows: written, uploaded,
++ durable: rows.filter((x) => x.durability === 'durable').length,
++ local_only: rows.filter((x) => x.durability === 'local').length,
++ segments: segs.length, bytes: rows.reduce((a, x) => a + x.byte_length, 0),
++ duration_ms: segs.length ? Number(segs[segs.length - 1].end_ms) : 0,
++ };
++ } catch (err) {
++ if (!committed) {
++ // Content-addressed: another run may have committed these very keys and files, so only unnamed ones go.
++ for (const k of stagedRemote) {
++ if (!await timeline.isNamed({ provider: k.provider, key: k.key }).catch(() => true)) await vodStorage.deleteObject(k.provider, k.key).catch(() => { /* best effort */ });
++ }
++ for (const p of stagedLocal) {
++ if (await timeline.isNamed({ localPath: p }).catch(() => true)) continue;
++ try { fs.unlinkSync(p); } catch { /* already gone */ }
++ }
++ }
++ throw err;
++ }
++}
++
++module.exports = { parsePlaylist, producedFiles, checkDisk, aborted, publishRendition };
ddiff --git a/server/jobs/types.js b/server/jobs/types.js
iindex 19482c6..faae5b7 100644
---- a/server/jobs/types.js
++++ b/server/jobs/types.js
@@@ -9,6 +9,7 @@
* object.sprite heavy a seek-preview sprite sheet (JPEG, layout in its metadata) as the `sprite` variant
* object.cmaf heavy the source as CMAF/HLS segments on its own keyframes, indexed in media_timeline (MEDIA_HLS_ENABLED)
* object.pack heavy a timeline's durable segments packed into ~60 s chunk objects (byte ranges; MEDIA_HLS_ENABLED)
++ * rendition.create heavy an on-demand 720p H.264/AAC rendition of a media object, indexed in media_timeline (MEDIA_HLS_ENABLED + MEDIA_RENDITIONS)
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
* clip.cut clips cut (or re-cut) a clip row from its VOD; retries with backoff (server/vod/clip-jobs.js)
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
@@@ -28,6 +29,7 @@ queue.register('object.waveform', require('./previews').waveform);
queue.register('object.sprite', require('./previews').sprite);
queue.register('object.cmaf', require('./cmaf').spec);
queue.register('object.pack', require('./pack').spec);
++queue.register('rendition.create', require('./rendition').spec);
queue.register('vod.finalize', require('./vod-finalize').spec);
queue.register('clip.cut', require('../vod/clip-jobs').spec);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
ddiff --git a/server/jobs/worker.js b/server/jobs/worker.js
iindex fccb53e..eeb97a9 100644
---- a/server/jobs/worker.js
++++ b/server/jobs/worker.js
@@@ -3,7 +3,7 @@
*
* Runs media_jobs in-process, in two lanes so a long split never holds up a thumbnail:
* light thumbnail.regenerate, invariant.scan MEDIA_JOBS_LIGHT_CONCURRENCY (2)
-- * heavy object.split, object.remux, object.cmaf, object.pack MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a
++ * heavy object.split, object.remux, object.cmaf, object.pack, rendition.create MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a
* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1
* finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording
* (it is the work the recorder does when a stream ends)
ddiff --git a/server/objects/hls.js b/server/objects/hls.js
iindex 3be80d2..a21b186 100644
---- a/server/objects/hls.js
++++ b/server/objects/hls.js
@@@ -8,8 +8,9 @@
* A playlist exists when MEDIA_HLS_ENABLED is on and the object's source timeline has a media row (seq > 0) — its own
* rows, or for a virtual clip (F3.5) its source's window rows. With none, object.cmaf is queued (idempotent) and no URL
* is answered, unless the object is a virtual clip (its source's window plays it; it is never cut a timeline of its own).
-- * A single-item read takes that path (deduped in-process); a list read uses discoveryMany, which answers a whole page in
-- * a constant number of queries and never queues anything.
++ * With a source timeline and MEDIA_RENDITIONS on, the missing renditions (F4: rendition.create) are queued the same lazy
++ * way, so the master playlist lists them once they exist. A single-item read takes that path (deduped in-process); a
++ * list read uses discoveryMany, which answers a whole page in a constant number of queries and never queues anything.
*
* The URL is built exactly as the v2 route does: an open object (not private, not in a sandbox tenant) gets the plain
* /o/<id>/master.m3u8; a closed one gets a signed playlist (purpose 'hls', signing.signedPlaylistUrl) and its expiry.
@@@ -62,18 +63,51 @@ async function isOpen(obj) {
const QUEUE_TTL_MS = 10 * 60 * 1000;
const QUEUE_MAX = 1000;
const _queuedCmaf = new Map();
++const _queuedRenditions = new Map();
--/** Queue object.cmaf for an object at most once per QUEUE_TTL_MS in this process. Never throws. */
--async function queueCmafOnce(appId, objectId) {
++/** Run `queue()` for an object at most once per QUEUE_TTL_MS in this process. Never throws. */
++async function throttleOnce(map, objectId, queue) {
const now = Date.now();
-- const last = _queuedCmaf.get(objectId);
++ const last = map.get(objectId);
if (last != null && now - last < QUEUE_TTL_MS) return;
-- if (_queuedCmaf.size >= QUEUE_MAX) {
-- for (const [id, at] of _queuedCmaf) if (now - at >= QUEUE_TTL_MS) _queuedCmaf.delete(id);
-- while (_queuedCmaf.size >= QUEUE_MAX) _queuedCmaf.delete(_queuedCmaf.keys().next().value);
++ if (map.size >= QUEUE_MAX) {
++ for (const [id, at] of map) if (now - at >= QUEUE_TTL_MS) map.delete(id);
++ while (map.size >= QUEUE_MAX) map.delete(map.keys().next().value);
}
-- _queuedCmaf.set(objectId, now);
-- try { await require('./timeline-queue').queueCmaf(appId, objectId); } catch (err) { console.warn(`[HLS] Timeline queue failed for ${objectId}:`, err.message); }
++ map.set(objectId, now);
++ try { await queue(); } catch (err) { console.warn(`[HLS] Queue failed for ${objectId}:`, err.message); }
++}
++
++/** 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));
++}
++
++/**
++ * Queue the missing renditions (rendition.create, F4) for an object at most once per QUEUE_TTL_MS in this process: the
++ * lazy path from the master playlist route and GET …/download?format=json. Never throws. Nothing is queued when
++ * MEDIA_RENDITIONS is off (timeline-queue.queueRendition checks) or for a virtual clip.
++ */
++async function queueRenditionsOnce(obj) {
++ if (!config.hls.enabled || !config.hls.renditions) return;
++ await throttleOnce(_queuedRenditions, obj.id, () => require('./timeline-queue').queueRendition(obj.app_id, obj.id));
++}
++
++/**
++ * The renditions an object's master playlist offers: `source` first, then every other rendition whose rows exist (the
++ * route lists a rendition only once it was produced). `sourceRows` is the object's own source rows when the caller
++ * already has them (a virtual clip's window over its source), so it is not read twice. `meta` is the resolution/codecs
++ * the cut stored in the object's metadata (renditions.<name>), so the playlist carries RESOLUTION/CODECS without
++ * reading bytes; it is null for a rendition cut before this was stored.
++ */
++async function renditionsFor(obj, sourceRows = null) {
++ const timeline = require('./timeline');
++ const stored = model.parseJson(obj.metadata, {}).renditions || {};
++ const source = sourceRows || (await timelineOf(obj)).rows;
++ const out = [{ name: timeline.SOURCE, rows: source, meta: stored[timeline.SOURCE] || null }];
++ const extra = await db.all('SELECT DISTINCT rendition FROM media_timeline WHERE object_id = ? AND rendition <> ? AND seq > 0 ORDER BY rendition', [obj.id, timeline.SOURCE]);
++ for (const r of extra) out.push({ name: r.rendition, rows: await timeline.list(obj.id, r.rendition), meta: stored[r.rendition] || null });
++ return out;
}
/** What an object with a serviceable timeline is discovered by: { hls_url, hls_expires_at? } (open stays unsigned). */
@@@ -96,6 +130,8 @@ async function discoveryForObject(obj) {
if (!tl.window) await queueCmafOnce(obj.app_id, obj.id);
return null;
}
++ // The source timeline exists: queue any missing rendition (F4) so the master playlist can list it next time.
++ if (!tl.window) await queueRenditionsOnce(obj);
return playlistFor(obj, await isOpen(obj));
} catch (err) {
console.warn(`[HLS] Discovery failed for ${obj.id}:`, err.message);
@@@ -187,4 +223,4 @@ async function discoveryMany(rows) {
return out;
}
--module.exports = { timelineOf, discoveryForObject, discovery, discoveryMany };
++module.exports = { timelineOf, renditionsFor, queueRenditionsOnce, discoveryForObject, discovery, discoveryMany };
ddiff --git a/server/objects/mp4-codecs.js b/server/objects/mp4-codecs.js
nnew file mode 100644
iindex 0000000..bb6b9d6
---- /dev/null
++++ b/server/objects/mp4-codecs.js
@@@ -0,0 +1,147 @@
++/**
++ * OpenVibe.Media — codecs and resolution from an init segment (docs/media-fabric.md §5, F4).
++ *
++ * The master playlist names each variant's RESOLUTION and CODECS, but serving it must not read segment bytes: at
++ * publish time (server/jobs/segments.js) this pure parser reads the init.mp4 a cut just wrote and the answer is stored
++ * in the object's metadata under renditions.<name> = { width, height, codecs }. parseInit never throws — an unreadable
++ * box answers nulls and the variant is listed without the attributes, as every pre-F4 object already is.
++ *
++ * Recognised tracks: video avc1 → `avc1.<profile><compat><level>` from its avcC box (width/height from the sample entry,
++ * else tkhd); audio mp4a → `mp4a.40.<audioObjectType>` from the esds DecoderSpecificInfo (AAC-LC is `mp4a.40.2`); an
++ * `Opus` sample entry → `opus`. One track whose codec is anything else makes `codecs` null: a partial CODECS is worse
++ * than none, and a client falls back to probing. width/height stay known even then.
++ */
++'use strict';
++
++/** The child boxes of [start, end), header included: { type, start, content, end }. A malformed run ends the walk. */
++function boxes(buf, start, end) {
++ const out = [];
++ let p = start;
++ while (p + 8 <= end) {
++ let size = buf.readUInt32BE(p);
++ const type = buf.toString('latin1', p + 4, p + 8);
++ let header = 8;
++ if (size === 1) { if (p + 16 > end) break; size = Number(buf.readBigUInt64BE(p + 8)); header = 16; }
++ else if (size === 0) size = end - p;
++ if (size < header || p + size > end) break;
++ out.push({ type, start: p, content: p + header, end: p + size });
++ p += size;
++ }
++ return out;
++}
++
++/** The first child of `parent` (a box, or { content, end } for a raw range) named one of `types`. */
++function child(buf, parent, ...types) {
++ for (const b of boxes(buf, parent.content, parent.end)) if (types.includes(b.type)) return b;
++ return null;
++}
++
++/** avc1 width/height from the VisualSampleEntry body, or null. */
++function entrySize(buf, entry) {
++ if (entry.content + 28 > entry.end) return null;
++ const w = buf.readUInt16BE(entry.content + 24);
++ const h = buf.readUInt16BE(entry.content + 26);
++ return w > 0 && h > 0 ? { width: w, height: h } : null;
++}
++
++/** The 16.16 fixed width/height of a trak's tkhd, or null (the fallback when the sample entry gave none). */
++function tkhdSize(buf, trak) {
++ const tkhd = child(buf, trak, 'tkhd');
++ if (!tkhd || tkhd.content >= tkhd.end) return null;
++ const base = (buf[tkhd.content] === 1 ? tkhd.content + 88 : tkhd.content + 76);
++ if (base + 8 > tkhd.end) return null;
++ const w = buf.readUInt32BE(base) / 65536;
++ const h = buf.readUInt32BE(base + 4) / 65536;
++ return w > 0 && h > 0 ? { width: Math.round(w), height: Math.round(h) } : null;
++}
++
++/** `avc1.<profile><compat><level>` from an avc1 sample entry's avcC box, or null. */
++function avcCodec(buf, entry) {
++ const avcC = child(buf, { content: entry.content + 78, end: entry.end }, 'avcC');
++ if (!avcC || avcC.content + 4 > avcC.end) return null;
++ const hex = (n) => n.toString(16).padStart(2, '0');
++ return `avc1.${hex(buf[avcC.content + 1])}${hex(buf[avcC.content + 2])}${hex(buf[avcC.content + 3])}`;
++}
++
++/** The descriptors in [start, end): { tag, payload, end }. A truncated length ends the walk. */
++function descriptors(buf, start, end) {
++ const out = [];
++ let p = start;
++ while (p + 2 <= end) {
++ const tag = buf[p];
++ let size = 0;
++ let q = p + 1;
++ let n = 0;
++ while (q < end && n < 4) { const b = buf[q++]; size = (size << 7) | (b & 0x7f); n++; if (!(b & 0x80)) break; }
++ if (size <= 0 || q + size > end) break;
++ out.push({ tag, payload: q, end: q + size });
++ p = q + size;
++ }
++ return out;
++}
++
++/** The first descriptor with `tag`, descending into the ES (0x03) and DecoderConfig (0x04) descriptors that nest. */
++function findDescriptor(buf, start, end, tag) {
++ for (const d of descriptors(buf, start, end)) {
++ if (d.tag === tag) return d;
++ if (d.tag === 0x03 || d.tag === 0x04) {
++ // ES_Descriptor payload: ES_ID(2)+flags(1); DecoderConfigDescriptor payload: 13 bytes before nested ones.
++ const inner = d.tag === 0x03 ? d.payload + 3 : d.payload + 13;
++ if (inner < d.end) { const found = findDescriptor(buf, inner, d.end, tag); if (found) return found; }
++ }
++ }
++ return null;
++}
++
++/** `mp4a.40.<objectType>` from an mp4a sample entry's esds DecoderSpecificInfo, or null (not AAC, or unreadable). */
++function aacCodec(buf, entry) {
++ const esds = child(buf, { content: entry.content + 28, end: entry.end }, 'esds');
++ if (!esds || esds.content + 4 > esds.end) return null;
++ const dc = findDescriptor(buf, esds.content + 4, esds.end, 0x04);
++ if (!dc || dc.payload >= dc.end || buf[dc.payload] !== 0x40) return null; // objectTypeIndication 0x40 = AAC
++ const dsi = findDescriptor(buf, dc.payload + 13, dc.end, 0x05);
++ if (!dsi || dsi.payload >= dsi.end) return null;
++ let aot = buf[dsi.payload] >> 3; // AudioSpecificConfig: 5 bits of audioObjectType (AAC-LC = 2)
++ if (aot === 31) {
++ if (dsi.payload + 2 > dsi.end) return null;
++ aot = 32 + (((buf[dsi.payload] & 0x07) << 3) | (buf[dsi.payload + 1] >> 5));
++ }
++ return aot > 0 ? `mp4a.40.${aot}` : null;
++}
++
++/**
++ * { width, height, codecs } for an init segment (`codecs` a comma-joined RFC 6381 list, or null). Never throws.
++ */
++function parseInit(input) {
++ const out = { width: null, height: null, codecs: null };
++ try {
++ const buf = Buffer.isBuffer(input) ? input : Buffer.from(input || []);
++ const moov = child(buf, { content: 0, end: buf.length }, 'moov');
++ if (!moov) return out;
++ const codecs = [];
++ let unknown = false;
++ const traks = boxes(buf, moov.content, moov.end).filter((b) => b.type === 'trak');
++ for (const trak of traks) {
++ const stsd = child(buf, child(buf, child(buf, child(buf, trak, 'mdia'), 'minf'), 'stbl'), 'stsd');
++ const entry = stsd ? boxes(buf, stsd.content + 8, stsd.end)[0] : null;
++ if (!entry) { unknown = true; continue; }
++ if (entry.type === 'avc1') {
++ const codec = avcCodec(buf, entry);
++ if (!codec) { unknown = true; continue; }
++ codecs.push(codec);
++ if (out.width == null) { const size = entrySize(buf, entry); if (size) { out.width = size.width; out.height = size.height; } }
++ } else if (entry.type === 'mp4a') {
++ const codec = aacCodec(buf, entry);
++ if (!codec) { unknown = true; continue; }
++ codecs.push(codec);
++ } else if (entry.type === 'Opus') {
++ codecs.push('opus');
++ } else { unknown = true; }
++ }
++ if (out.width == null && traks.length) { const size = tkhdSize(buf, traks[0]); if (size) { out.width = size.width; out.height = size.height; } }
++ if (!unknown && codecs.length) out.codecs = codecs.join(',');
++ } catch { /* a malformed init: no attributes */ }
++ return out;
++}
++
++module.exports = { parseInit };
ddiff --git a/server/objects/routes.js b/server/objects/routes.js
iindex 90bff35..71ab312 100644
---- a/server/objects/routes.js
++++ b/server/objects/routes.js
@@@ -770,15 +770,18 @@ const INLINE = /^(image\/(?!svg)|video\/|audio\/|application\/pdf$|text\/plain$)
const publicRouter = express.Router();
// ── Segment-native video (docs/media-fabric.md §3, F3.1; MEDIA_HLS_ENABLED) ──
--// GET /o/:id/master.m3u8, /o/:id/source/index.m3u8 and /o/:id/source/{init.mp4,NNNNNN.m4s}: the playlists are written
--// from the object's media_timeline rows (objects/timeline.js), never read from files. Every one passes the same check
--// as GET /o/:id: public and unlisted objects openly, private and sandbox ones only with the object's valid ?exp&sig
++// GET /o/:id/master.m3u8, /o/:id/source/index.m3u8, /o/:id/source/{init.mp4,NNNNNN.m4s} and — for a produced
++// rendition (F4) — /o/:id/<rendition>/index.m3u8 and /o/:id/<rendition>/<name>: the playlists are written from the
++// object's media_timeline rows (objects/timeline.js), never read from files. Every one passes the same check as
++// GET /o/:id: public and unlisted objects openly, private and sandbox ones only with the object's valid ?exp&sig
// (a playlist token from /download's hls_url, or a download signature), which a signed playlist carries onto each URI;
--// deleted 410, not ready 404. Off, these paths 404.
++// deleted 410, not ready 404. Off, these paths 404. The master playlist lists a rendition only once its rows exist
++// and queues the missing ones on demand (objects/hls.js, MEDIA_RENDITIONS).
// Like GET /o/:id, a private or sandbox answer carries no Access-Control-Allow-Origin; a public one may be read cross-origin.
// A segment's copy is the placement router's choice; a packed one (F3.3) is a ranged read of its ~60 s chunk.
const HLS_TYPE = 'application/vnd.apple.mpegurl';
const SEGMENT_NAME = /^(init\.mp4|\d{6,9}\.m4s)$/;
++const RENDITION_NAME = /^[a-z0-9_-]{1,32}$/;
const SLICE_FETCH_MS = 15000; // a ranged read of one packed segment from B2/R2
// clipWindow/timelineOf (a clip's window over its source, and the rows its playlists are written from) live in
// objects/hls.js, shared with the v1 VOD/clip discovery.
@@@ -817,7 +820,10 @@ publicRouter.get('/:id/master.m3u8', hlsRoute(async (req, res, { obj, query }) =
const timeline = require('./timeline');
const { rows } = await hls.timelineOf(obj);
if (!rows.some((r) => Number(r.seq) > 0)) return res.status(404).json({ error: 'No timeline' });
-- res.type(HLS_TYPE).send(timeline.masterPlaylist([{ name: timeline.SOURCE, rows }], { query }));
++ // A missing rendition (F4) is queued lazily; only the ones whose rows exist are listed, so a client never follows
++ // a variant that is not there yet.
++ await hls.queueRenditionsOnce(obj);
++ res.type(HLS_TYPE).send(timeline.masterPlaylist(await hls.renditionsFor(obj, rows), { query }));
}));
publicRouter.get('/:id/source/index.m3u8', hlsRoute(async (req, res, { obj, query }) => {
@@@ -871,14 +877,13 @@ async function sendSlice(req, res, { file, url, offset, length, headers }) {
return res.end(body);
}
--publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, { obj, closed }) => {
++/**
++ * Send one timeline row's bytes: this node's file streamed, a remote unpacked segment 302'd to a short presigned URL,
++ * a remote packed one read as a byte range of its chunk here (a redirect cannot carry the range). `rendition` keys the
++ * router's sticky session and names the prefix the row's bytes live under.
++ */
++async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
const timeline = require('./timeline');
-- const name = String(req.params.name || '');
-- let row = SEGMENT_NAME.test(name) ? await timeline.byName(obj.id, timeline.SOURCE, name) : null;
-- // A virtual clip serves only the source segments inside its window (timelineOf): any other name is a 404, whatever
-- // the signature, so a clip's token never reaches the rest of its source.
-- if (!row && SEGMENT_NAME.test(name) && obj.kind === 'clip') row = (await hls.timelineOf(obj)).rows.find((r) => r.name === name) || null;
-- if (!row) return res.status(404).json({ error: 'Not found' });
const mime = Number(row.seq) === 0 ? 'video/mp4' : 'video/iso.segment';
const headers = { 'Content-Type': mime, 'Cache-Control': closed ? 'private, no-store' : 'public, max-age=3600' };
const packed = !!row.packed_object_id;
@@@ -888,8 +893,8 @@ publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, { obj, closed })
// streamed; a remote unpacked segment is a redirect to a short presigned URL; a remote packed one is read as a byte
// range of its chunk here, since a redirect cannot carry the segment's range.
const router = require('../placement/router');
-- const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: name,
-- purpose: 'playback', session: router.sessionFor(req, `${obj.id}/source`),
++ const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: row.name,
++ purpose: 'playback', session: router.sessionFor(req, `${obj.id}/${rendition}`),
contentType: mime, expiresIn: 300, presign: !packed });
for (const loc of decision.candidates || []) {
if (loc.provider === 'local') {
@@@ -908,6 +913,38 @@ publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, { obj, closed })
return res.redirect(302, url);
}
res.status(404).json({ error: 'Segment bytes unavailable' });
++}
++
++publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, h) => {
++ const timeline = require('./timeline');
++ const name = String(req.params.name || '');
++ let row = SEGMENT_NAME.test(name) ? await timeline.byName(h.obj.id, timeline.SOURCE, name) : null;
++ // A virtual clip serves only the source segments inside its window (timelineOf): any other name is a 404, whatever
++ // the signature, so a clip's token never reaches the rest of its source.
++ if (!row && SEGMENT_NAME.test(name) && h.obj.kind === 'clip') row = (await hls.timelineOf(h.obj)).rows.find((r) => r.name === name) || null;
++ if (!row) return res.status(404).json({ error: 'Not found' });
++ return sendSegmentRow(req, res, h, row, timeline.SOURCE);
++}));
++
++// A produced rendition (F4, rendition.create): its own media playlist and segments, under /o/:id/<rendition>/…. The
++// master playlist's variant URI is `<rendition>/index.m3u8`; a rendition with no rows answers 404 (nothing to play).
++publicRouter.get('/:id/:rendition/index.m3u8', hlsRoute(async (req, res, { obj, query }) => {
++ const timeline = require('./timeline');
++ const rendition = String(req.params.rendition || '');
++ if (!RENDITION_NAME.test(rendition) || rendition === timeline.SOURCE) return res.status(404).json({ error: 'Not found' });
++ const rows = await timeline.list(obj.id, rendition);
++ if (!rows.some((r) => Number(r.seq) > 0)) return res.status(404).json({ error: 'No timeline' });
++ res.type(HLS_TYPE).send(timeline.mediaPlaylist(rows, { query }));
++}));
++
++publicRouter.get('/:id/:rendition/:name', hlsRoute(async (req, res, h) => {
++ const timeline = require('./timeline');
++ const rendition = String(req.params.rendition || '');
++ const name = String(req.params.name || '');
++ if (!RENDITION_NAME.test(rendition) || rendition === timeline.SOURCE) return res.status(404).json({ error: 'Not found' });
++ const row = SEGMENT_NAME.test(name) ? await timeline.byName(h.obj.id, rendition, name) : null;
++ if (!row) return res.status(404).json({ error: 'Not found' });
++ return sendSegmentRow(req, res, h, row, rendition);
}));
publicRouter.get('/:id', async (req, res) => {
// A restore drill (MEDIA_DRILL) serves no stored bytes, local or by a B2/R2 redirect.
ddiff --git a/server/objects/timeline-queue.js b/server/objects/timeline-queue.js
iindex f64b2fe..36d9c91 100644
---- a/server/objects/timeline-queue.js
++++ b/server/objects/timeline-queue.js
@@@ -7,15 +7,36 @@
* object, or once the timeline exists. One job per object: dedupeActive joins a queued or running cut, and the
* idempotency key `object.cmaf:<id>` answers a later call with the same job. → the job id, or null.
*
-- * queuePack(appId, objectId) enqueues `object.pack` once a cut's rows are committed (server/jobs/cmaf.js), so packing
-- * follows the cut by itself instead of an operator's hand. Nothing happens (null) with MEDIA_HLS_ENABLED off or for a
-- * missing or non-media object. One job at a time per object: dedupeActive joins a queued or running pack, so a rerun
-- * while it is active joins it and a later cut queues a new one. → the job id, or null.
++ * queuePack(appId, objectId, { rendition }) enqueues `object.pack` once a cut's rows are committed (server/jobs/cmaf.js,
++ * server/jobs/rendition.js), so packing follows the cut by itself instead of an operator's hand. Nothing happens (null)
++ * with MEDIA_HLS_ENABLED off or for a missing or non-media object. One job at a time per object and rendition:
++ * dedupeActive joins a queued or running pack, so a rerun while it is active joins it and a later cut queues a new one.
++ * → the job id, or null.
++ *
++ * queueRendition(appId, objectId, { rendition }) enqueues `rendition.create` for a ready media object that does not have
++ * that rendition yet (docs/media-fabric.md §5, F4): lazily, from the master playlist route and from
++ * GET …/download?format=json (objects/hls.js). Nothing happens (null) with MEDIA_HLS_ENABLED or MEDIA_RENDITIONS off,
++ * for a missing, unready, held or non-media object, for a virtual clip (it plays its source's window), or once the rows
++ * exist. One job per object, rendition AND source cut: dedupeActive joins a queued or running one, and the idempotency
++ * key `rendition.create:<id>:<name>:<srcver>` (srcver = a short hash of the source rows) answers a later request with
++ * the same job — so a source below the rung's height settles after the one job that decided to skip it, while a re-cut
++ * source gets a new key and is queued again. → the job id, or null.
*/
'use strict';
++const crypto = require('crypto');
const config = require('../config');
++/**
++ * The identity of the source cut a rendition is made against: the first 16 hex of sha256 over the source rows' (seq,
++ * sha256) in seq order. A re-cut changes the rows and so the key; the same cut keeps deduping a skipped or failed job.
++ */
++function sourceVersion(rows) {
++ const h = crypto.createHash('sha256');
++ for (const r of rows) h.update(`${Number(r.seq)}:${r.sha256 || ''}\n`);
++ return h.digest('hex').slice(0, 16);
++}
++
async function queueCmaf(appId, objectId, { segment_seconds = null } = {}) {
if (!config.hls.enabled || !appId || !objectId) return null;
const obj = await require('./model').getObject(objectId);
@@@ -31,12 +52,12 @@ async function queueCmaf(appId, objectId, { segment_seconds = null } = {}) {
return r.job ? r.job.id : null;
}
--async function queuePack(appId, objectId) {
++async function queuePack(appId, objectId, { rendition = null } = {}) {
if (!config.hls.enabled || !appId || !objectId) return null;
const obj = await require('./model').getObject(objectId);
if (!obj || !require('../jobs/derive').isMediaObject(obj)) return null;
const queue = require('../jobs/queue');
-- const params = require('../jobs/pack').spec.validate({ appId, obj, params: {} });
++ const params = require('../jobs/pack').spec.validate({ appId, obj, params: rendition ? { rendition } : {} });
const r = await queue.enqueue({
appId, type: 'object.pack', objectId: obj.id, params,
dedupeActive: true, createdBy: 'system:timeline',
@@@ -45,4 +66,40 @@ async function queuePack(appId, objectId) {
return r.job ? r.job.id : null;
}
--module.exports = { queueCmaf, queuePack };
++/**
++ * The renditions this server ships for the master playlist: the ladder names of rendition.create. Kept here so the
++ * routes and this module agree without objects/hls.js requiring the job.
++ */
++function renditionNames() { return require('../jobs/rendition').renditionNames(); }
++
++/**
++ * Queue `rendition.create` for each rung the object does not have yet (one, '720p'), joined per object and rendition.
++ * `rendition` names a single rung instead of the whole ladder. → the first job id, or null when nothing was queued.
++ */
++async function queueRendition(appId, objectId, { rendition = null } = {}) {
++ if (!config.hls.enabled || !config.hls.renditions || !appId || !objectId) return null;
++ const obj = await require('./model').getObject(objectId);
++ if (!obj || obj.lifecycle_status !== 'ready' || !require('../jobs/derive').isMediaObject(obj)) return null;
++ // A held object is left alone (as the rendition job itself refuses it): queue nothing, so a hold does not burn jobs.
++ if (await require('./model').isHeld(obj.id)) return null;
++ const timeline = require('./timeline');
++ // A virtual clip (no timeline of its own) plays its source's window: it gets no rendition of its own.
++ if (obj.kind === 'clip' && !await timeline.has(obj.id)) return null;
++ const queue = require('../jobs/queue');
++ const names = rendition ? [String(rendition)] : renditionNames();
++ const srcVer = sourceVersion(await timeline.list(obj.id, timeline.SOURCE));
++ let first = null;
++ for (const name of names) {
++ if (await timeline.has(obj.id, name)) continue;
++ const params = require('../jobs/rendition').spec.validate({ appId, obj, params: { rendition: name } });
++ const r = await queue.enqueue({
++ appId, type: 'rendition.create', objectId: obj.id, params,
++ dedupeActive: true, idempotencyKey: `rendition.create:${obj.id}:${name}:${srcVer}`, createdBy: 'system:rendition',
++ });
++ if (r.created) require('../jobs/worker').kick();
++ if (first == null && r.job) first = r.job.id;
++ }
++ return first;
++}
++
++module.exports = { queueCmaf, queuePack, queueRendition };
ddiff --git a/server/objects/timeline.js b/server/objects/timeline.js
iindex 31237db..4943ecb 100644
---- a/server/objects/timeline.js
++++ b/server/objects/timeline.js
@@@ -89,9 +89,13 @@ class TimelineChanged extends Error {}
* says the same is not written, a changed one is updated, rows past the new end are dropped. → { inserted, updated, unchanged, removed }.
* `expect` (the rows the caller read, by seq) makes it compare-and-set: if any row's location or packing changed meanwhile
* (an object.pack committed), nothing is written and TimelineChanged is thrown.
++ * `dropOtherRenditions` (a source re-cut) also removes every other rendition's rows of the object IN THE SAME
++ * TRANSACTION: a rendition cut against the old source can never play against the new one. The removed rows are returned
++ * as `dropped`, for the caller to delete their bytes after the commit (deleteBytes, guarded by isNamed).
*/
--async function replace(objectId, rendition, rows, { jobId = null, expect = null } = {}) {
++async function replace(objectId, rendition, rows, { jobId = null, expect = null, dropOtherRenditions = false } = {}) {
const out = { inserted: 0, updated: 0, unchanged: 0, removed: 0 };
++ let dropped = null;
await db.getDb().tx(async () => {
await lockObject(objectId);
const before = new Map((await list(objectId, rendition)).map((r) => [Number(r.seq), r]));
@@@ -112,8 +116,12 @@ async function replace(objectId, rendition, rows, { jobId = null, expect = null
}
const last = rows.reduce((m, r) => Math.max(m, r.seq), -1);
out.removed = (await db.run('DELETE FROM media_timeline WHERE object_id = ? AND rendition = ? AND seq > ?', [objectId, rendition, last])).changes || 0;
++ if (dropOtherRenditions) {
++ dropped = await db.all('SELECT * FROM media_timeline WHERE object_id = ? AND rendition <> ?', [objectId, rendition]);
++ if (dropped.length) await db.run('DELETE FROM media_timeline WHERE object_id = ? AND rendition <> ?', [objectId, rendition]);
++ }
});
-- return out;
++ return dropped && dropped.length ? { ...out, dropped } : out;
}
/**
@@@ -302,16 +310,25 @@ function mediaPlaylist(rows, { query = '' } = {}) {
return lines.join('\n');
}
--/** The master playlist: one variant per rendition (only `source` in F3.1), BANDWIDTH = its peak segment bitrate. */
++/**
++ * The master playlist: one variant per rendition (only `source` in F3.1), BANDWIDTH = its peak segment bitrate.
++ * `meta` (the object's stored renditions.<name> = { width, height, codecs }, F4) adds RESOLUTION/CODECS; a rendition
++ * cut before metadata was stored — or one whose init could not be parsed — lists without them, exactly as before.
++ */
function masterPlaylist(renditions, { query = '' } = {}) {
const lines = ['#EXTM3U', '#EXT-X-VERSION:7', '#EXT-X-INDEPENDENT-SEGMENTS'];
-- for (const { name, rows } of renditions) {
++ for (const { name, rows, meta } of renditions) {
const segs = rows.filter((r) => Number(r.seq) > 0);
const rate = (r) => Math.ceil(Number(r.byte_length) * 8 * 1000 / Math.max(1, Number(r.end_ms) - Number(r.start_ms)));
const peak = Math.max(1, ...segs.map(rate));
const total = segs.reduce((a, r) => a + Number(r.end_ms) - Number(r.start_ms), 0);
const avg = Math.max(1, Math.ceil(segs.reduce((a, r) => a + Number(r.byte_length), 0) * 8 * 1000 / Math.max(1, total)));
-- lines.push(`#EXT-X-STREAM-INF:BANDWIDTH=${peak},AVERAGE-BANDWIDTH=${avg}`, withQuery(`${name}/index.m3u8`, query));
++ const attrs = [`BANDWIDTH=${peak}`, `AVERAGE-BANDWIDTH=${avg}`];
++ if (meta) {
++ if (Number(meta.width) > 0 && Number(meta.height) > 0) attrs.push(`RESOLUTION=${Number(meta.width)}x${Number(meta.height)}`);
++ if (meta.codecs) attrs.push(`CODECS="${meta.codecs}"`);
++ }
++ lines.push(`#EXT-X-STREAM-INF:${attrs.join(',')}`, withQuery(`${name}/index.m3u8`, query));
}
lines.push('');
return lines.join('\n');
ddiff --git a/test/media-timeline.test.js b/test/media-timeline.test.js
iindex 2da8d4d..b46172f 100644
---- a/test/media-timeline.test.js
++++ b/test/media-timeline.test.js
@@@ -75,6 +75,10 @@ const { spawn, spawnSync } = require('child_process');
const mp = timeline.mediaPlaylist(await timeline.list('med_HAND'));
assert.ok(mp.includes('#EXT-X-TARGETDURATION:3\n') && mp.includes('#EXT-X-MAP:URI="init.mp4"') && mp.includes('#EXTINF:1.932,\n000003.m4s') && mp.trim().endsWith('#EXT-X-ENDLIST'), mp);
assert.ok(timeline.masterPlaylist([{ name: 'source', rows: hand }]).includes('#EXT-X-STREAM-INF:BANDWIDTH=80000,AVERAGE-BANDWIDTH='), 'peak = 30000 B over 3 s');
++ // F4 slice 1: one variant per rendition, source first — the route passes every rendition whose rows exist.
++ const twoUp = timeline.masterPlaylist([{ name: 'source', rows: hand }, { name: '720p', rows: hand }]);
++ assert.deepStrictEqual([...twoUp.matchAll(/#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+[^\n]*\n(\S+)\n/g)].map((m) => m[1]),
++ ['source/index.m3u8', '720p/index.m3u8'], twoUp);
assert.deepStrictEqual(await timeline.replace('med_HAND', 'source', hand.slice(0, 2)), { inserted: 0, updated: 0, unchanged: 2, removed: 2 }, 'rows past a shorter end are dropped');
assert.deepStrictEqual(await timeline.removeObject('med_HAND'), { removed: 2, pending: 0 });
if (!hasFfmpeg) { console.log('media timeline: skipped (ffmpeg not found)'); process.exit(0); }
@@@ -121,7 +125,9 @@ const { spawn, spawnSync } = require('child_process');
};
const pub = await native('public');
const priv = await native('private');
-- const before = JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]);
++ // The cut adds only derived metadata (renditions.<name> = width/height/codecs): everything the object IS stays.
++ const untouched = async () => { const o = await model.getObject(pub); const md = model.parseJson(o.metadata, {}); delete md.renditions; return [{ ...o, metadata: JSON.stringify(md) }, await model.listLocations(pub)]; };
++ const before = JSON.stringify(await untouched());
const runJob = async (objectId, params, type = 'object.cmaf') => {
const r = await call('POST', '/api/v2/live/jobs', { body: { type, object_id: objectId, ...(params && { params }) } });
@@@ -151,7 +157,7 @@ const { spawn, spawnSync } = require('child_process');
assert.strictEqual(r.durability, 'local');
assert.ok(r.local_path.startsWith(path.join(path.resolve(process.env.OBJECTS_PATH), '.timeline', 'live', pub, 'source')));
}
-- assert.strictEqual(JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]), before, 'the source is untouched');
++ assert.strictEqual(JSON.stringify(await untouched()), before, 'the source bytes, locations and row are untouched (only derived rendition metadata is added)');
assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
// ── A truncated timeline (an in-progress cut) must not shrink the sheet: the object's own 7 s duration
@@@ -181,7 +187,7 @@ const { spawn, spawnSync } = require('child_process');
const master = await get(`/o/${pub}/master.m3u8`);
assert.strictEqual(master.status, 200);
assert.ok(/^application\/vnd\.apple\.mpegurl/.test(master.headers.get('content-type')));
-- assert.ok(/#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+\nsource\/index\.m3u8\n/.test(master.buf.toString()), master.buf.toString());
++ assert.ok(/#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+[^\n]*\nsource\/index\.m3u8\n/.test(master.buf.toString()), master.buf.toString());
const media = (await get(`/o/${pub}/source/index.m3u8`)).buf.toString();
assert.ok(media.trim().endsWith('#EXT-X-ENDLIST') && media.includes('#EXT-X-PLAYLIST-TYPE:VOD') && media.includes('#EXT-X-TARGETDURATION:3\n'), media);
assert.deepStrictEqual(parsePlaylist(media), [['000001.m4s', 3], ['000002.m4s', 1.5], ['000003.m4s', 1.5], ['000004.m4s', 1]].map(([name, seconds]) => ({ name, seconds })), 'the durations are the rows');
@@@ -497,6 +503,26 @@ const { spawn, spawnSync } = require('child_process');
assert.ok(many.get(String(sb)).hls_expires_at, 'the batched sandbox answer carries its expiry');
}
++ // ── F4 slice 1: the master playlist lists a rendition only once its rows exist, and serves it then ──
++ {
++ const two = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility: 'public', lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: 1, metadata: { duration_seconds: 3 } });
++ const rrow = (rendition, seq, start, end, len) => ({ ...row(seq, start, end, len), key: `${two}/${rendition}/${timeline.segmentName(seq)}` });
++ await timeline.replace(two, 'source', [row(0, 0, 0, 500), row(1, 0, 3000, 30000)]);
++ config.hls.renditions = false; // off: no lazy enqueue while the negative case is asserted
++ assert.ok(!(await get(`/o/${two}/master.m3u8`)).buf.toString().includes('720p/index.m3u8'), 'no 720p rows: source only');
++ assert.strictEqual((await get(`/o/${two}/720p/index.m3u8`)).status, 404, 'and the rendition playlist is not there');
++ config.hls.renditions = true;
++ await timeline.replace(two, '720p', [rrow('720p', 0, 0, 0, 500), rrow('720p', 1, 0, 3000, 15000)]);
++ const m2 = (await get(`/o/${two}/master.m3u8`)).buf.toString();
++ assert.ok(/source\/index\.m3u8\n/.test(m2) && /720p\/index\.m3u8\n/.test(m2), 'source and 720p are listed');
++ assert.ok(m2.includes('BANDWIDTH=80000') && m2.includes('BANDWIDTH=40000'), 'each variant carries its own BANDWIDTH');
++ const r720 = await get(`/o/${two}/720p/index.m3u8`);
++ assert.strictEqual(r720.status, 200);
++ assert.ok(r720.buf.toString().includes('#EXT-X-MAP:URI="init.mp4"') && r720.buf.toString().includes('000001.m4s'), 'the 720p media playlist comes from its rows');
++ assert.strictEqual((await get(`/o/${two}/720p/000009.m4s`)).status, 404, 'a name with no row is 404');
++ config.hls.renditions = false;
++ }
++
// ── MEDIA_HLS_ENABLED off: the routes 404, the job is refused, the download answer is as before ──
config.hls.enabled = false;
assert.strictEqual((await get(`/o/${pub}/master.m3u8`)).status, 404);
ddiff --git a/test/rendition-job.test.js b/test/rendition-job.test.js
nnew file mode 100644
iindex 0000000..0ac7b2c
---- /dev/null
++++ b/test/rendition-job.test.js
@@@ -0,0 +1,311 @@
++'use strict';
++// On-demand renditions, F4 slice 1 (docs/media-fabric.md §5): a ready media object's master playlist lists the
++// renditions that exist and queues the missing ones lazily — one rendition.create job per object and rendition,
++// deduped while it is active and answered by its idempotency key afterwards. The job transcodes one 720p H.264/AAC
++// rung (libx264) into media_timeline through the same content-addressed rows and two-phase commit as object.cmaf, with
++// the source timeline's segment boundaries forced as keyframes; the master playlist then lists source + 720p with each
++// variant's own BANDWIDTH and /o/:id/720p/… serves it. A source already 720p or smaller is skipped; an object under a
++// hold is left alone; with MEDIA_RENDITIONS or MEDIA_HLS_ENABLED off nothing is cut or listed.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const { spawn, spawnSync } = require('child_process');
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-rendition-'));
++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++ process.env.VOD_PATH = dir('vods');
++ process.env.CLIPS_PATH = dir('clips');
++ process.env.FILES_PATH = dir('files');
++ process.env.THUMBNAILS_PATH = dir('thumbnails');
++ process.env.PASTES_PATH = dir('pastes');
++ process.env.OBJECTS_PATH = dir('objects');
++ process.env.MEDIA_PUBLIC_URL = 'https://media.test';
++ process.env.MEDIA_JOBS_POLL_MS = '50';
++ process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';
++ process.env.MEDIA_UPLOAD_MIN_FREE_MB = '0';
++ process.env.MEDIA_HLS_ENABLED = '1';
++ process.env.MEDIA_RENDITIONS = '1';
++ for (const p of ['B2', 'R2']) for (const k of ['ENDPOINT', 'BUCKET', 'KEY_ID', 'APP_KEY', 'ACCESS_KEY_ID', 'SECRET_ACCESS_KEY']) process.env[`MEDIA_${p}_${k}`] = '';
++
++ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
++ const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');
++ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
++ const hasX264 = hasFfmpeg && /libx264/.test(String(spawnSync('ffmpeg', ['-hide_banner', '-encoders']).stdout));
++
++ (async () => {
++ const config = require('../server/config');
++ assert.strictEqual(config.hls.enabled, true);
++ assert.strictEqual(config.hls.renditions, true);
++ const types = require('../server/jobs/types');
++ assert.ok(types.names().includes('rendition.create'), 'rendition.create is registered');
++ assert.strictEqual(require('../server/jobs/queue').typeSpec('rendition.create').lane, 'heavy');
++ assert.deepStrictEqual(require('../server/jobs/rendition').renditionNames(), ['720p'], 'this slice ships one rung');
++ console.log('✅ rendition.create is registered on the heavy lane; MEDIA_RENDITIONS reads true');
++
++ if (!hasFfmpeg || !hasX264) { console.log('rendition job: skipped (ffmpeg/libx264 not found)'); process.exit(0); }
++
++ const db = require('../server/db/database');
++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-rendition' });
++ const model = require('../server/objects/model');
++ const timeline = require('../server/objects/timeline');
++ const queue = require('../server/jobs/queue');
++ const worker = require('../server/jobs/worker');
++ const express = require('express');
++
++ // ── The init parser (objects/mp4-codecs.js) on hand-built minimal boxes ──
++ {
++ const { parseInit } = require('../server/objects/mp4-codecs');
++ const bx = (type, ...parts) => {
++ const body = Buffer.concat(parts.map((x) => (Buffer.isBuffer(x) ? x : Buffer.from(x))));
++ const out = Buffer.alloc(8 + body.length);
++ out.writeUInt32BE(out.length, 0);
++ out.write(type, 4, 'latin1');
++ body.copy(out, 8);
++ return out;
++ };
++ const u16 = (n) => { const b = Buffer.alloc(2); b.writeUInt16BE(n); return b; };
++ const u32 = (n) => { const b = Buffer.alloc(4); b.writeUInt32BE(n); return b; };
++ const avc1 = (w, h, profile, compat, level) => bx('avc1', Buffer.alloc(6), u16(1), Buffer.alloc(16), u16(w), u16(h), Buffer.alloc(50),
++ bx('avcC', Buffer.from([1, profile, compat, level])));
++ const desc = (tag, payload) => Buffer.concat([Buffer.from([tag, payload.length]), payload]); // MPEG-4 descriptors, not boxes
++ const mp4a = (aot) => {
++ const dsi = desc(0x05, Buffer.from([aot << 3, 0x10]));
++ const dcd = desc(0x04, Buffer.concat([Buffer.from([0x40, 0x15, 0, 0, 0]), u32(0), u32(0), dsi]));
++ const es = desc(0x03, Buffer.concat([u16(1), Buffer.from([0]), dcd]));
++ const esds = bx('esds', Buffer.alloc(4), es);
++ return bx('mp4a', Buffer.alloc(6), u16(1), Buffer.alloc(8), u16(2), u16(16), u16(0), u16(0), u32(44100 * 65536), esds);
++ };
++ const trakOf = (e) => bx('trak', bx('mdia', bx('minf', bx('stbl', bx('stsd', Buffer.alloc(4), u32(1), e)))));
++ const moovOf = (...es) => bx('moov', ...es.map(trakOf));
++ assert.deepStrictEqual(parseInit(moovOf(avc1(1280, 800, 0x4d, 0x40, 0x1e))),
++ { width: 1280, height: 800, codecs: 'avc1.4d401e' }, 'avc1 profile/compat/level and sample-entry size');
++ assert.deepStrictEqual(parseInit(moovOf(mp4a(2))), { width: null, height: null, codecs: 'mp4a.40.2' }, 'AAC-LC from the esds DecoderSpecificInfo');
++ assert.deepStrictEqual(parseInit(moovOf(bx('Opus', Buffer.alloc(28)))), { width: null, height: null, codecs: 'opus' }, 'an Opus sample entry');
++ assert.deepStrictEqual(parseInit(moovOf(avc1(640, 360, 0x42, 0x00, 0x1e), mp4a(2))),
++ { width: 640, height: 360, codecs: 'avc1.42001e,mp4a.40.2' }, 'video + audio, in track order');
++ // A track whose codec is not recognised: CODECS is omitted entirely, RESOLUTION stays.
++ assert.deepStrictEqual(parseInit(moovOf(avc1(1280, 800, 0x4d, 0x40, 0x1e), bx('zzzz', Buffer.alloc(28)))),
++ { width: 1280, height: 800, codecs: null }, 'one unknown track omits CODECS, never a partial string');
++ assert.deepStrictEqual(parseInit(Buffer.alloc(0)), { width: null, height: null, codecs: null }, 'empty bytes never throw');
++ console.log('✅ mp4-codecs parses avc1/mp4a/Opus, omits a partial CODECS, and never throws');
++ }
++ const app = express();
++ app.use(express.json());
++ app.use('/api/v2/:app/jobs', require('../server/jobs/routes'));
++ app.use('/api/v2/:app/objects', require('../server/objects/routes'));
++ app.use('/o', require('../server/objects/routes').publicRouter);
++ const server = http.createServer(app);
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const api = `http://127.0.0.1:${server.address().port}`;
++ const call = async (method, p, { body } = {}) => {
++ const res = await fetch(api + p, { method, headers: { authorization: 'Bearer live-key-rendition', 'content-type': 'application/json' }, body: body === undefined ? undefined : JSON.stringify(body) });
++ return { status: res.status, body: await res.json().catch(() => null) };
++ };
++ const get = async (url, headers = {}) => {
++ const res = await fetch(url.startsWith('http') ? url : api + url, { headers, redirect: 'manual' });
++ return { status: res.status, headers: res.headers, buf: Buffer.from(await res.arrayBuffer()) };
++ };
++ // The worker is never started: lazily queued jobs stay queued until this runs one directly.
++ const runJob = async (type, objectId, params) => {
++ const r = await call('POST', '/api/v2/live/jobs', { body: { type, object_id: objectId, ...(params && { params }) } });
++ assert.strictEqual(r.status, 202, JSON.stringify(r.body));
++ assert.ok(await worker.runNow(r.body.job.id), `${type} runs`);
++ return queue.jobPublic(await queue.get(r.body.job.id));
++ };
++ const rendJobs = async (objectId) => await db.all('SELECT * FROM media_jobs WHERE job_type = ? AND object_id = ?', ['rendition.create', objectId]);
++
++ // A 6 s, 1280x800, 10 fps source with keyframes every 1.5 s: cuts at 3.0, 4.5, 6.0 for a 2 s target.
++ const srcFile = path.join(process.env.VOD_PATH, 'src.mp4');
++ const mk = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=6:size=1280x800:rate=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=6',
++ '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', '-g', '15', '-c:a', 'aac', '-shortest', srcFile]);
++ assert.strictEqual(mk.status, 0, String(mk.stderr));
++ const media = async (file, visibility = 'public', metadata = {}) => {
++ const id = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility, lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: fs.statSync(file).size, metadata: { duration_seconds: 6, ...metadata } });
++ await model.upsertLocation(id, { provider: 'local', key: file, state: 'present', size_bytes: fs.statSync(file).size, verified: true });
++ return id;
++ };
++ const pub = await media(srcFile);
++
++ // The source timeline, cut by a real object.cmaf (stream copy): 0-3000, 3000-4500, 4500-6000 ms.
++ const cmaf = await runJob('object.cmaf', pub, { segment_seconds: 2 });
++ assert.strictEqual(cmaf.status, 'succeeded', JSON.stringify(cmaf.error));
++ const srcRows = await timeline.segments(pub, 'source');
++ assert.deepStrictEqual(srcRows.map((r) => [Number(r.start_ms), Number(r.end_ms)]), [[0, 3000], [3000, 4500], [4500, 6000]], 'the source cut on its keyframes');
++ const sourceMaster = (await get(`/o/${pub}/master.m3u8`)).buf.toString();
++ assert.ok(!sourceMaster.includes('720p/index.m3u8'), 'no 720p rows yet: the master playlist lists source only');
++ assert.strictEqual((await get(`/o/${pub}/720p/index.m3u8`)).status, 404, 'and the rendition has nothing to serve');
++ console.log('✅ a source timeline exists; the master playlist lists source only until a rendition does');
++
++ // ── Lazy: asking for the master playlist queues exactly one rendition.create; a second request joins it ──
++ const m1 = await get(`/o/${pub}/master.m3u8`);
++ assert.strictEqual(m1.status, 200);
++ let jobs = await rendJobs(pub);
++ assert.strictEqual(jobs.length, 1, 'the first master playlist request queues one rendition.create');
++ const srcVer = crypto.createHash('sha256').update((await timeline.list(pub, 'source')).map((r) => `${Number(r.seq)}:${r.sha256 || ''}\n`).join('')).digest('hex').slice(0, 16);
++ assert.deepStrictEqual([jobs[0].status, jobs[0].created_by, jobs[0].idempotency_key, JSON.parse(jobs[0].params).rendition],
++ ['queued', 'system:rendition', `rendition.create:${pub}:720p:${srcVer}`, '720p']);
++ await get(`/o/${pub}/master.m3u8`);
++ jobs = await rendJobs(pub);
++ assert.strictEqual(jobs.length, 1, 'a second request joins the active job');
++ // The route throttles per process; the queue itself joins an active job and replays on its key.
++ const { queueRendition } = require('../server/objects/timeline-queue');
++ assert.strictEqual(await queueRendition('live', pub), jobs[0].id, 'queueRendition answers the one active job');
++ assert.strictEqual((await rendJobs(pub)).length, 1, 'and does not duplicate it');
++ assert.ok(!(await get(`/o/${pub}/master.m3u8`)).buf.toString().includes('720p/index.m3u8'), 'still not listed while the job is queued');
++ console.log('✅ the master playlist queues one rendition.create lazily; a second request joins it');
++
++ // ── The job: 720p rows land, contiguous, content-addressed, bytes on this node ──
++ const done = queue.jobPublic(await worker.runNow(jobs[0].id));
++ assert.strictEqual(done.status, 'succeeded', JSON.stringify(done.error));
++ assert.deepStrictEqual([done.result.rendition, done.result.target_height, done.result.segments, done.result.skipped], ['720p', 720, 3, false], JSON.stringify(done.result));
++ const rows = await timeline.list(pub, '720p');
++ assert.deepStrictEqual(rows.map((r) => Number(r.seq)), [0, 1, 2, 3]);
++ assert.deepStrictEqual(rows.map((r) => r.name), ['init.mp4', '000001.m4s', '000002.m4s', '000003.m4s']);
++ const segs = rows.slice(1);
++ assert.deepStrictEqual(segs.map((r) => [Number(r.start_ms), Number(r.end_ms)]), [[0, 3000], [3000, 4500], [4500, 6000]], 'the rung is cut on the source boundaries it forced');
++ for (const r of rows) {
++ assert.strictEqual(sha(fs.readFileSync(r.local_path)), r.sha256, `${r.name} sha256`);
++ assert.strictEqual(r.key, `${pub}/720p/${r.sha256.slice(0, 12)}/${r.name}`, 'the key is that rendition\'s, versioned by the sha');
++ assert.strictEqual(r.durability, 'local', 'no durable provider configured here');
++ assert.ok(r.local_path.startsWith(path.join(path.resolve(process.env.OBJECTS_PATH), '.timeline', 'live', pub, '720p')), r.local_path);
++ }
++ assert.ok(segs.every((r) => Number(r.byte_length) > 0));
++ assert.strictEqual(JSON.stringify(await timeline.segments(pub, 'source')), JSON.stringify(srcRows), 'the source timeline is untouched');
++ assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
++ console.log('✅ the job lands 720p rows with the produced bytes, times and sha256; the source is untouched');
++
++ // ── The master playlist now lists source + 720p, each with its own BANDWIDTH, and the variant plays ──
++ const master = (await get(`/o/${pub}/master.m3u8`)).buf.toString();
++ assert.ok(/#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+[^\n]*\nsource\/index\.m3u8\n/.test(master), master);
++ assert.ok(/#EXT-X-STREAM-INF:BANDWIDTH=\d+,AVERAGE-BANDWIDTH=\d+[^\n]*\n720p\/index\.m3u8\n/.test(master), master);
++ const bw = [...master.matchAll(/#EXT-X-STREAM-INF:BANDWIDTH=(\d+)/g)].map((m) => Number(m[1]));
++ assert.strictEqual(bw.length, 2, 'one variant per rendition whose rows exist');
++ const media720 = (await get(`/o/${pub}/720p/index.m3u8`)).buf.toString();
++ assert.ok(media720.includes('#EXT-X-PLAYLIST-TYPE:VOD') && media720.trim().endsWith('#EXT-X-ENDLIST') && media720.includes('000001.m4s'), media720);
++ const bw720 = Number(/#EXT-X-STREAM-INF:BANDWIDTH=(\d+),AVERAGE-BANDWIDTH=\d+[^\n]*\n720p/.exec(master)[1]);
++ assert.strictEqual(bw720, Math.max(...segs.map((r) => Math.ceil(Number(r.byte_length) * 8 * 1000 / Math.max(1, Number(r.end_ms) - Number(r.start_ms))))), 'the 720p BANDWIDTH is its own segments\' peak');
++
++ // ── Item 2: the stored init metadata feeds RESOLUTION/CODECS, without a bytes read per request ──
++ const sourceLine = /#EXT-X-STREAM-INF:([^\n]*)\nsource\/index\.m3u8\n/.exec(master)[1];
++ const v720Line = /#EXT-X-STREAM-INF:([^\n]*)\n720p\/index\.m3u8\n/.exec(master)[1];
++ assert.ok(sourceLine.includes('RESOLUTION=1280x800'), sourceLine);
++ assert.ok(v720Line.includes('RESOLUTION=1152x720'), v720Line);
++ assert.ok(/CODECS="avc1\.[0-9a-f]{6},mp4a\.40\.2"/.test(sourceLine), sourceLine);
++ assert.ok(/CODECS="avc1\.[0-9a-f]{6},mp4a\.40\.2"/.test(v720Line), v720Line);
++ assert.ok(/CODECS="avc1\.4d4[0-9a-f]/.test(v720Line), `the 720p rung is libx264 main: ${v720Line}`);
++ const mdPub = model.parseJson((await model.getObject(pub)).metadata, {});
++ assert.deepStrictEqual([mdPub.renditions['720p'].width, mdPub.renditions['720p'].height], [1152, 720], 'the cut stored the rendition\'s size');
++ assert.ok(/^avc1\.[0-9a-f]{6},mp4a\.40\.2$/.test(mdPub.renditions['720p'].codecs), mdPub.renditions['720p'].codecs);
++ const { parseInit } = require('../server/objects/mp4-codecs');
++ assert.deepStrictEqual(parseInit(fs.readFileSync(rows[0].local_path)), mdPub.renditions['720p'], 'the parser on the real init.mp4 agrees with what was stored');
++ console.log('✅ the master playlist carries RESOLUTION and CODECS from the stored init metadata');
++ for (const r of rows) {
++ const s = await get(`/o/${pub}/720p/${r.name}`);
++ assert.deepStrictEqual([s.status, sha(s.buf)], [200, r.sha256], `${r.name} served`);
++ }
++ assert.strictEqual((await get(`/o/${pub}/720p/000009.m4s`)).status, 404);
++ assert.strictEqual((await get(`/o/${pub}/720p/..%2Findex.m3u8`)).status, 404);
++ // A real HLS reader sees 720p, with the source's duration (the playlist is the master, so ffprobe takes a variant).
++ const probed = await new Promise((resolve) => {
++ const p = spawn('ffprobe', ['-v', 'error', '-rw_timeout', '10000000', '-select_streams', 'v:0', '-show_entries', 'stream=height', '-of', 'csv=p=0', `${api}/o/${pub}/720p/index.m3u8`]);
++ let out = '';
++ let err = '';
++ p.stdout.on('data', (d) => { out += d; });
++ p.stderr.on('data', (d) => { err += d; });
++ p.on('close', () => resolve(`${out.trim()}${err ? ` (${err.trim()})` : ''}`));
++ });
++ const heights = probed.split(/\s+/).filter(Boolean).map(Number);
++ assert.ok(heights.length && heights.every((h) => h === 720), `the produced rendition is 720p: ${JSON.stringify(probed)}`);
++ assert.ok(bw.every((x) => x > 0), 'both variants carry a bitrate');
++ console.log('✅ the master playlist lists source + 720p with per-variant BANDWIDTH; /o/:id/720p/… plays');
++
++ // ── Idempotent: a rerun finds the rows and re-encodes nothing ──
++ const mtimes = rows.map((r) => fs.statSync(r.local_path).mtimeMs);
++ const again = await runJob('rendition.create', pub, { rendition: '720p' });
++ assert.deepStrictEqual([again.status, again.result.already, again.result.segments], ['succeeded', true, 3], JSON.stringify(again.result));
++ assert.deepStrictEqual(rows.map((r) => fs.statSync(r.local_path).mtimeMs), mtimes, 'no segment re-encoded');
++
++ // ── A source already 720p or smaller is not upscaled ──
++ const smallFile = path.join(process.env.VOD_PATH, 'small.mp4');
++ const mkSmall = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=2:size=640x360:rate=10', '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', '-g', '10', smallFile]);
++ assert.strictEqual(mkSmall.status, 0, String(mkSmall.stderr));
++ const small = await media(smallFile, 'public', { duration_seconds: 2 });
++ const skip = await runJob('rendition.create', small, {});
++ assert.deepStrictEqual([skip.status, skip.result.skipped], ['succeeded', true], JSON.stringify(skip.result));
++ assert.strictEqual(await timeline.has(small, '720p'), false, 'nothing is written for a source at or below the rung');
++ console.log('✅ a source at or below 720p is skipped, with no rows');
++
++ // ── A held object is left alone (retried, not cut) ──
++ const held = await media(srcFile, 'public', { duration_seconds: 6 });
++ const hold = await model.placeHold({ object_id: held, kind: 'admin', reason: 'rendition test' });
++ const hj = await runJob('rendition.create', held, {});
++ assert.deepStrictEqual([hj.status, hj.error_code], ['queued', 'media.object.held'], JSON.stringify(hj));
++ assert.strictEqual(await timeline.has(held, '720p'), false, 'no rows on a held object');
++ await model.releaseHold(hold.id);
++
++ // ── The rendition key is scoped to the source cut: a failure replays against the same source, a re-cut requeues ──
++ const scope = await media(srcFile, 'public', { duration_seconds: 6 });
++ await runJob('object.cmaf', scope, { segment_seconds: 2 });
++ const first = await queueRendition('live', scope);
++ assert.ok(first, 'a missing rendition is queued');
++ await db.run("UPDATE media_jobs SET status = 'failed', attempts = 3, error_code = 'ffmpeg_failed' WHERE id = ?", [first]);
++ assert.strictEqual(await queueRendition('live', scope), first, 'the same source cut answers the failed job instead of requeueing it');
++ assert.strictEqual((await rendJobs(scope)).length, 1, 'and creates nothing new');
++ // A re-cut changes the source rows (new segmenting, new shas): a new key, a new job.
++ await runJob('object.cmaf', scope, { segment_seconds: 1 });
++ const second = await queueRendition('live', scope);
++ assert.ok(second && second !== first, 'a re-cut source queues a new job');
++ const keys = (await rendJobs(scope)).map((j) => j.idempotency_key).sort();
++ assert.strictEqual(new Set(keys).size, 2, `two different source-versioned keys: ${keys}`);
++ console.log('✅ rendition idempotency is scoped to the source cut: same source replays, a re-cut requeues');
++
++ // ── A source re-cut drops the stale rendition's rows and bytes; the master lists only source until it is re-cut ──
++ const staleFiles = (await timeline.list(pub, '720p')).map((r) => r.local_path).filter(Boolean);
++ assert.ok(staleFiles.length && staleFiles.every((f) => fs.existsSync(f)), 'the 720p bytes are there before the re-cut');
++ await runJob('object.cmaf', pub, { segment_seconds: 1 });
++ assert.strictEqual(await timeline.has(pub, '720p'), false, 'the re-cut removed the stale rendition rows');
++ assert.ok(staleFiles.every((f) => !fs.existsSync(f)), 'and their bytes');
++ const afterCut = (await get(`/o/${pub}/master.m3u8`)).buf.toString();
++ assert.ok(!afterCut.includes('720p/index.m3u8'), 'the master lists only source until a new rendition lands');
++ assert.ok(/CODECS="avc1\.[0-9a-f]{6},mp4a\.40\.2"/.test(afterCut), 'the source variant keeps its stored CODECS');
++ console.log('✅ a source re-cut drops the stale rendition rows and bytes; the master lists source only');
++
++ // ── A held object is queued no rendition at all (the lazy path checks the hold first) ──
++ const heldQ = await media(srcFile, 'public', { duration_seconds: 6 });
++ await runJob('object.cmaf', heldQ, { segment_seconds: 2 });
++ const qh = await model.placeHold({ object_id: heldQ, kind: 'admin', reason: 'queue test' });
++ assert.strictEqual(await queueRendition('live', heldQ), null, 'nothing is queued for a held object');
++ assert.strictEqual((await rendJobs(heldQ)).length, 0, 'no rendition.create row exists');
++ await model.releaseHold(qh.id);
++ console.log('✅ a held object queues no rendition');
++
++ // ── MEDIA_RENDITIONS off: nothing is queued or cut; the job is refused ──
++ const off = await media(srcFile, 'public', { duration_seconds: 6 });
++ await runJob('object.cmaf', off, { segment_seconds: 2 });
++ config.hls.renditions = false;
++ assert.strictEqual((await get(`/o/${off}/master.m3u8`)).status, 200, 'the master playlist still serves the source timeline');
++ assert.ok(!(await get(`/o/${off}/master.m3u8`)).buf.toString().includes('720p/index.m3u8'), 'no rendition is listed');
++ const offJob = await call('POST', '/api/v2/live/jobs', { body: { type: 'rendition.create', object_id: off, params: { rendition: '720p' } } });
++ assert.deepStrictEqual([offJob.status, offJob.body.code], [409, 'media.renditions.disabled'], JSON.stringify(offJob.body));
++ assert.strictEqual((await rendJobs(off)).length, 0, 'nothing queued with the flag off');
++ assert.strictEqual((await get(`/o/${off}/720p/index.m3u8`)).status, 404, 'and the rendition route has nothing');
++ config.hls.renditions = true;
++ console.log('✅ MEDIA_RENDITIONS off: no rendition queued, cut or listed; the job is refused 409');
++
++ // ── MEDIA_HLS_ENABLED off: the routes 404 and the job is refused as before ──
++ config.hls.enabled = false;
++ assert.strictEqual((await get(`/o/${pub}/master.m3u8`)).status, 404);
++ assert.strictEqual((await get(`/o/${pub}/720p/index.m3u8`)).status, 404);
++ const hlsOff = await call('POST', '/api/v2/live/jobs', { body: { type: 'rendition.create', object_id: pub, params: { rendition: '720p' } } });
++ assert.deepStrictEqual([hlsOff.status, hlsOff.body.code], [409, 'media.hls.disabled']);
++ config.hls.enabled = true;
++
++ server.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('rendition job: all checks passed');
++ process.exit(0);
++ })().catch((err) => { console.error(err); process.exit(1); });
++})().catch((err) => { console.error(err); process.exit(1); });