openvibers/openvibe.media Commits

Media: H.264/AAC timelines for WebM and other incompatible recordings (MEDIA_CMAF_TRANSCODE), capped and niced (#95)

OP OpenVibers committed 12 Oct 2026 · 6e2a16c47a · 11 files, +239 −20

* Media: H.264/AAC timelines for WebM and other incompatible recordings (MEDIA_CMAF_TRANSCODE), capped and niced

HLS has been on since 15:52 UTC today. Browser recordings are WebM (VP8/VP9, Opus), so `object.cmaf`'s stream copy fails on them with `codec_unsupported` (VOD 5720), and they never get a timeline. A VP9 copy that does succeed still doesn't play in Safari. This change (plan T4, F4: "VP8/VP9 → H.264") makes those recordings playable.

- **Probe, then copy or encode** (`MEDIA_CMAF_TRANSCODE`, off by default). The job ffprobes the exact input ffmpeg will read (`derive.probeStreams`).
  - H.264 video is copied. AAC or MP3 audio is copied.
  - Other video is encoded as H.264: libx264 veryfast, CRF 23, yuv420p, high profile, source size and timing, keyframes forced on the segment grid.
  - Other audio is encoded as AAC (128k, stereo).
  - The result says `video: copy|h264` and `audio: copy|aac|none`, and the master playlist's `CODECS` follow what was written.
- **Bounds.**
  - A source that needs an encode and runs longer than `MEDIA_CMAF_TRANSCODE_MAX_SECONDS` (4 h) fails permanently with `too_long_to_transcode`.
  - An encode's time budget comes from the measured duration, not the byte size.
  - The job's own timeout is 13 h with transcoding on, 3 h as before when off.
- **CPU.** The host has 4 cores and serves live streams. The heavy lane waits for recordings only before it starts a job, so an encode can still be running when someone goes live. Encodes therefore use `MEDIA_CMAF_TRANSCODE_THREADS` libx264 threads (2) and run at nice 15.
- **One retry for the old failures.** With the knob on, a lazy timeline request for an object whose last `object.cmaf` failed with `codec_unsupported` queues one fresh job under the key `object.cmaf:<id>:transcode`. If that job fails, the same key replays it, so lazy reads can't loop. The backfill cap still applies. Queued jobs from before simply run with transcoding on.
- **Metrics:** `media_cmaf_transcodes_total{stream}`.
- **Docs:** `docs/media-fabric.md` (the cmaf paragraph and the HLS checklist), README, STATUS and `.env.example`.

Built by Codex and reviewed. During review I added the thread cap, the CPU priority (`derive.ffmpeg`'s `niceness`) and their docs. A production WebM (VOD 5720, VP8) carries a container duration, so the encode bound works for browser recordings.

Tests: the new `cmaf-transcode.test.js` covers:
- a VP8/Opus WebM fails with the knob off and is encoded to h264 + aac with it on (segments probed, master `CODECS` checked);
- H.264/AAC is still copied, and audio-only is still copied;
- the duration cap;
- one lazy retry, and no second one.

The full suite passes 103 of 105 files. The two exceptions:
- `media-timeline.test.js` passes on its own.
- `jobs.test.js`'s async thumbnail check fails under load on main too: two back-to-back runs failed on main and on this branch alike, and it passed when given more time. That is an existing timing flake in that test, not this change.

* Media: cmaf-transcode test skips without ffmpeg, like media-timeline

Changes

ddiff --git a/.env.example b/.env.example
iindex 9a4b84f..c3aaf8b 100644
---- a/.env.example
++++ b/.env.example
@@@ -143,6 +143,12 @@ MEDIA_CAPABILITY_GATE=report
  # /o/:id/master.m3u8, /o/:id/source/… and the hls_url field of GET …/download?format=json ; off (default):
  # the job is refused and the routes 404, nothing else changes.
  #MEDIA_HLS_ENABLED=0
++# Encode incompatible source streams as H.264/AAC (CPU intensive; off by default).
++#MEDIA_CMAF_TRANSCODE=0
++# Maximum duration in seconds of a source needing transcoding (default 4 h).
++#MEDIA_CMAF_TRANSCODE_MAX_SECONDS=14400
++# libx264 threads for one transcode (it also runs at nice 15: the heavy lane only waits for recordings before starting).
++#MEDIA_CMAF_TRANSCODE_THREADS=2
  # Maximum bytes of local timeline segments and packed chunks (default 10 GiB; 0 keeps only durable copies).
  #MEDIA_TIMELINE_LOCAL_MAX_BYTES=10737418240
  # Stop keeping new local timeline copies below this free space (default 8 GiB).
ddiff --git a/README.md b/README.md
iindex 50d6c55..c20db63 100644
---- a/README.md
++++ b/README.md
@@@ -451,6 +451,9 @@ never the headers (`test/trust-proxy.test.js`).
  | Environment variable | Default | Purpose |
  |---|---:|---|
  | `MEDIA_HLS_ENABLED` | `0` | Enable timeline generation and HLS routes. |
++| `MEDIA_CMAF_TRANSCODE` | `0` | Encode incompatible source streams to H.264/AAC for the source timeline; uses CPU. |
++| `MEDIA_CMAF_TRANSCODE_MAX_SECONDS` | `14400` | Maximum duration of a source needing CMAF transcoding (four hours). |
++| `MEDIA_CMAF_TRANSCODE_THREADS` | `2` | libx264 threads for one CMAF transcode, which also runs at nice 15 so live streams keep the CPU. |
  | `MEDIA_TIMELINE_LOCAL_MAX_BYTES` | `10737418240` | Maximum local timeline cache (10 GiB; `0` keeps no durable copies locally). |
  | `MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES` | `8589934592` | Below this free disk level, new verified timeline bytes remain durable only. |
  | `MEDIA_HLS_BACKFILL_PER_HOUR` | `20` | Maximum lazy `object.cmaf` queues for older VODs per rolling hour on a single Media host; finalize is exempt. |
ddiff --git a/STATUS.json b/STATUS.json
iindex e897479..3b8eb89 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -40,7 +40,7 @@
      "tiering on demand (F2.5): the object sweep promotes and demotes on the hour's Valkey reads against per-class hysteresis bands, one MGET per page, falling back to the daily PostgreSQL counts for a whole sweep when Valkey fails",
      "placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
      "every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
--    "segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
++    "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; optional MEDIA_CMAF_TRANSCODE encodes incompatible streams as H.264/AAC up to MEDIA_CMAF_TRANSCODE_MAX_SECONDS, default 14400) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
      "local timeline cache (F3): 10 GiB default budget and 8 GiB free-disk floor, least-recently-served eviction of verified durable copies in the hourly sweep, durable segment fallback, optional timeline_cache readiness and local bytes/evictions metrics; lazy CMAF backfill capped at 20 per hour while recording finalize is exempt",
      "packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)",
      "on-demand renditions (F4 slice 1, behind MEDIA_HLS_ENABLED + MEDIA_RENDITIONS): the rendition.create job transcodes one 720p H.264/AAC rung of a ready media object into media_timeline through the same content-addressed rows, placement-router durable provider and two-phase commit as object.cmaf, forcing the source timeline's segment boundaries as keyframes; a source already 720p or smaller is skipped and a held object is left alone. The master playlist lists a rendition only once its rows exist and queues the missing ones lazily from /o/:id/master.m3u8 and download?format=json (dedupeActive + an object/rendition idempotency key), and /o/:id/<rendition>/index.m3u8 plus /o/:id/<rendition>/<name> serve it",
ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex 7c01510..b4b1519 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -95,14 +95,21 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
    `sha256`, `durability` `local|durable`), keyed by (object, rendition, seq) for "segments in order" with an index on
    (object, rendition, start_ms) for "the segment at t". `server/objects/timeline.js` is the store (`segmentAt(objectId,
    rendition, tMs)`, `segments`, `replace`, `removeObject`, the playlist writers). The heavy-lane job **`object.cmaf`**
--  (`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,
--  `params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it) into `init.mp4` +
++  (`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object by default (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,
++  `params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it). With
++  `MEDIA_CMAF_TRANSCODE=1`, it probes the source and copies H.264 video and AAC/MP3 audio, encoding other video as
++  H.264 and other audio as AAC. Encoded video keeps its dimensions and frame timing and forces keyframes on the segment
++  grid; sources needing an encode above `MEDIA_CMAF_TRANSCODE_MAX_SECONDS` (14400 by default) fail permanently. An
++  encode uses `MEDIA_CMAF_TRANSCODE_THREADS` libx264 threads (2) at nice 15, because the heavy lane waits for recordings
++  only before it starts a job and an encode can outlast that. The
++  output is `init.mp4` +
    `000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured,
    under the key `<object>/source/<version>/<name>` on B2 (`durable` after a size and sha256 readback), then writes all rows of
    the rendition in one transaction and only then deletes the previous cut's bytes. It never touches the source object; a
    rerun reuses rows whose sha256 matches and their location as they are and uploads only what is not durable yet; it
    honours the abort signal (which stops an upload in flight) and, before the rows commit, removes the files and keys it
--  staged; the stream-copy budget is honoured and its work directory removed.
++  staged; the stream-copy or duration-based transcode budget is honoured and its work directory removed. The master
++  playlist reads CODECS from the written init segment, so an encoded source advertises its output codecs.
    Behind **`MEDIA_HLS_ENABLED`** (off by default: the job is refused with `409 media.hls.disabled` and the routes 404),
    `GET /o/:id/master.m3u8`, `/o/:id/source/index.m3u8` and `/o/:id/source/{init.mp4,NNNNNN.m4s}` serve the timeline
    under the same check as `GET /o/:id` (private and sandbox objects only with the object's `?exp&sig`, carried onto
@@@ -167,6 +174,9 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
    answers the object's `remux` variant (`object.remux`,
    `-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux
    (`dedupeActive`) and answers `202 { job_id }` while it is missing.
++  With transcoding enabled, new jobs use a `:transcode` idempotency suffix. Lazy discovery retries an earlier
++  copy-only `codec_unsupported` failure once under that key. A failed transcode job is not queued again; the hourly
++  lazy backfill cap still applies.
  - **F3.5, shipped: virtual clips over the timeline (§4).**
  - **Local timeline cache, shipped:** `OBJECTS_PATH/.timeline` is bounded by
    `MEDIA_TIMELINE_LOCAL_MAX_BYTES` (10 GiB by default; 0 keeps no new local copies once durable) and
@@@ -186,8 +196,10 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
    own finalize bypasses the cap. List reads discover timelines in batches and do not queue missing ones.
  
    **Turning HLS on in production:** set the cache budget and free-space floor for the host, set
--  `MEDIA_HLS_ENABLED=1`, keep `MEDIA_JOBS_HEAVY_CONCURRENCY=1`, and watch `timeline_cache`, local timeline bytes,
--  evictions, and `object.cmaf`/`object.pack` job durations during backfill.
++  `MEDIA_HLS_ENABLED=1`, and set `MEDIA_CMAF_TRANSCODE=1` to make WebM/other incompatible sources playable as H.264/AAC.
++  Encoding costs substantially more CPU than copying; keep `MEDIA_JOBS_HEAVY_CONCURRENCY=1`, choose a suitable
++  `MEDIA_CMAF_TRANSCODE_MAX_SECONDS` (default 14400, four hours), and watch `timeline_cache`, local timeline bytes,
++  evictions, `media_cmaf_transcodes_total`, and `object.cmaf`/`object.pack` job durations during backfill.
  - **F3.6, shipped (behind `MEDIA_HLS_ENABLED` + `MEDIA_MATERIALIZED_CLIPS`, off by default): materialized clips over
    the source's bytes (§4, docs/materialized-clips.md).** A clip whose source has a timeline gets its own persisted
    `media_timeline` rows: the interior segments are the source's rows (same location, packed chunk range and sha256,
ddiff --git a/server/config.js b/server/config.js
iindex 3145125..8375802 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -93,6 +93,10 @@ const config = {
      // media_timeline (/o/:id/master.m3u8, /o/:id/source/…). Off by default: off, the job is refused and the routes 404.
      hls: {
          enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
++        transcode: ['1', 'true', 'on'].includes(String(process.env.MEDIA_CMAF_TRANSCODE || '').toLowerCase()),
++        transcodeMaxSeconds: Math.max(0, intEnv('MEDIA_CMAF_TRANSCODE_MAX_SECONDS', 14400)),
++        // libx264 threads for one transcode (2 on a 4-core host leaves room for live streams; it also runs at nice 15).
++        transcodeThreads: Math.min(16, Math.max(1, intEnv('MEDIA_CMAF_TRANSCODE_THREADS', 2))),
          localMaxBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MAX_BYTES', 10 * 1024 ** 3)),
          localMinFreeBytes: Math.max(0, intEnv('MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES', 8 * 1024 ** 3)),
          backfillPerHour: Math.max(0, intEnv('MEDIA_HLS_BACKFILL_PER_HOUR', 20)),
ddiff --git a/server/jobs/cmaf.js b/server/jobs/cmaf.js
iindex f63a1fb..ce644ae 100644
---- a/server/jobs/cmaf.js
++++ b/server/jobs/cmaf.js
@@@ -1,8 +1,8 @@
  /**
   * OpenVibe.Media — job type object.cmaf (heavy lane; docs/media-fabric.md §3, F3.1)
   *
-- * The source representation of a finished video (or audio) object as CMAF under HLS: a stream copy (ffmpeg -c copy,
-- * no re-encode) into fragmented-MP4 segments cut on the source's own keyframes (variable duration), init.mp4 plus
++ * The source representation of a finished video (or audio) object as CMAF under HLS: a stream copy by default, or
++ * selective H.264/AAC transcoding with MEDIA_CMAF_TRANSCODE, into fragmented-MP4 segments, init.mp4 plus
   * 000001.m4s, …, stored under the object's `source/` prefix and indexed in media_timeline (objects/timeline.js), one
   * row per segment with its times, size, sha256 and durability. The source object is never touched: its bytes,
   * locations, visibility, metadata and lifecycle stay as they are. Once the rows are committed, `object.pack` is queued
@@@ -10,7 +10,7 @@
   *
   *   params { segment_seconds }   the target segment length, 1-10 (4): a segment ends at the first keyframe after it
   *   result { source_id, rendition: 'source', segments, duration_ms, bytes, durable, local_only,
-- *            rows: { inserted, updated, unchanged, removed }, uploaded, master_path }
++ *            rows: { inserted, updated, unchanged, removed }, uploaded, master_path, video, audio }
   *
   * 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
@@@ -34,6 +34,10 @@ const { JobError } = require('./queue');
  
  const DEFAULT_SEGMENT_SECONDS = 4;
  
++function transcodeBudgetMs(duration) {
++    return Math.min(13 * 3600 * 1000, Math.max(10 * 60000, Math.round(duration * 3000 + 10 * 60000)));
++}
++
  function disabled() {
      return new JobError('media.hls.disabled', 'Segment-native video is off on this server (MEDIA_HLS_ENABLED)', { permanent: true, status: 409 });
  }
@@@ -56,6 +60,36 @@ async function run(job, ctx) {
      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 });
++    let video = 'copy';
++    let audio = 'copy';
++    let codecArgs = ['-c', 'copy'];
++    let timeoutMs = derive.budgetMs(src.size_bytes);
++    const segmentSeconds = Number(job.params.segment_seconds || DEFAULT_SEGMENT_SECONDS);
++    let info;
++    try { info = await derive.probeStreams(source, { signal: ctx.signal }); }
++    catch (err) {
++        segments.aborted(ctx);
++        // Preserve the old copy path if an input cannot be probed while the new feature is off.
++        if (config.hls.transcode) throw new JobError('probe_failed', `ffprobe failed: ${err.message}`);
++    }
++    segments.aborted(ctx);
++    if (info && !info.audio) audio = 'none';
++    if (config.hls.transcode && info) {
++        video = info.video && info.video.codec_name !== 'h264' ? 'h264' : 'copy';
++        audio = !info.audio ? 'none' : ['aac', 'mp3'].includes(info.audio.codec_name) ? 'copy' : 'aac';
++        if (video === 'h264' || audio === 'aac') {
++            if (!(info.duration > 0)) throw new JobError('duration_unknown', 'Cannot bound a transcode without a measured duration', { permanent: true });
++            if (info.duration > config.hls.transcodeMaxSeconds) {
++                throw new JobError('too_long_to_transcode', `Source duration ${info.duration}s exceeds MEDIA_CMAF_TRANSCODE_MAX_SECONDS`, { permanent: true });
++            }
++            timeoutMs = transcodeBudgetMs(info.duration);
++            codecArgs = video === 'h264'
++                ? ['-c:v', 'libx264', '-preset', 'veryfast', '-crf', '23', '-pix_fmt', 'yuv420p', '-profile:v', 'high', '-threads', String(config.hls.transcodeThreads),
++                    '-force_key_frames', `expr:gte(t,n_forced*${segmentSeconds})`, '-sc_threshold', '0', '-fps_mode', 'passthrough']
++                : ['-c:v', 'copy'];
++            codecArgs.push(...(audio === 'aac' ? ['-c:a', 'aac', '-b:a', '128k', '-ac', '2'] : ['-c:a', 'copy']));
++        }
++    }
      segments.checkDisk(Number(src.size_bytes) || 0);
      const rendition = timeline.SOURCE;
      const dir = derive.workDir(job.id);
@@@ -63,11 +97,11 @@ async function run(job, ctx) {
          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?', '-c', 'copy', '-map_metadata', '-1', '-fflags', '+bitexact', '-flags', '+bitexact',
--            '-f', 'hls', '-hls_time', String(job.params.segment_seconds || DEFAULT_SEGMENT_SECONDS), '-hls_list_size', '0',
++            '-map', '0:v:0?', '-map', '0:a:0?', ...codecArgs, '-map_metadata', '-1', '-fflags', '+bitexact', '-flags', '+bitexact',
++            '-f', 'hls', '-hls_time', String(segmentSeconds), '-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: derive.budgetMs(src.size_bytes) });
++        { signal: ctx.signal, timeoutMs, niceness: video === 'h264' || audio === 'aac' ? 15 : 0 });
          segments.aborted(ctx);
          const tail = String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300);
          if (!r.ok) {
@@@ -78,10 +112,12 @@ async function run(job, ctx) {
          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: 'Cmaf' });
++        if (video === 'h264') require('../placement/metrics-binding').inc('media_cmaf_transcodes_total', { stream: 'video' });
++        if (audio === 'aac') require('../placement/metrics-binding').inc('media_cmaf_transcodes_total', { stream: 'audio' });
          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,
++            master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true, video, audio,
          };
      } finally {
          derive.cleanupWork(job.id);
@@@ -89,6 +125,6 @@ async function run(job, ctx) {
  }
  
  module.exports = {
--    spec: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate, run },
++    spec: { lane: 'heavy', maxAttempts: 3, timeoutMs: () => config.hls.transcode ? 13 * 3600 * 1000 : 3 * 3600 * 1000, needsObject: true, validate, run },
      parsePlaylist: segments.parsePlaylist,
  };
ddiff --git a/server/jobs/derive.js b/server/jobs/derive.js
iindex fae51bd..4edf94f 100644
---- a/server/jobs/derive.js
++++ b/server/jobs/derive.js
@@@ -109,10 +109,13 @@ function sha256File(p) {
  }
  
  /** ffmpeg with a deadline and the job's abort signal. Resolves { ok, code, stderr }. */
--function ffmpeg(args, { signal, timeoutMs }) {
++/** `niceness` (0-19) lowers the process's CPU priority, for a re-encode that may still be running when a stream starts:
++ *  the heavy lane only waits for recordings before it starts a job. */
++function ffmpeg(args, { signal, timeoutMs, niceness = 0 }) {
      return new Promise((resolve) => {
          let proc;
          try { proc = spawn('ffmpeg', args, { stdio: ['ignore', 'ignore', 'pipe'] }); } catch (err) { return resolve({ ok: false, code: -1, stderr: err.message }); }
++        if (niceness > 0 && proc.pid) { try { require('os').setPriority(proc.pid, niceness); } catch { /* best effort */ } }
          let stderr = '';
          let done = false;
          const finish = (r) => { if (done) return; done = true; clearTimeout(t); if (signal) signal.removeEventListener('abort', onAbort); resolve(r); };
@@@ -130,6 +133,41 @@ function inputArgs(src) {
      return src.remote ? ['-reconnect', '1', '-reconnect_streamed', '1', '-reconnect_delay_max', '10'] : [];
  }
  
++/** The first streams and measured duration of the exact input ffmpeg will read. */
++function probeStreams(source, { signal, timeoutMs = 60000 } = {}) {
++    return new Promise((resolve, reject) => {
++        const args = ['-v', 'error', ...inputArgs(source), '-show_entries', 'format=duration:stream=codec_type,codec_name,duration', '-of', 'json', source.input];
++        let proc;
++        try { proc = spawn('ffprobe', args, { stdio: ['ignore', 'pipe', 'pipe'] }); } catch (err) { return reject(err); }
++        let stdout = '';
++        let stderr = '';
++        let done = false;
++        const kill = () => { try { proc.kill('SIGKILL'); } catch { /* gone */ } };
++        const timer = setTimeout(kill, timeoutMs);
++        const finish = (err, value) => {
++            if (done) return;
++            done = true;
++            clearTimeout(timer);
++            if (signal) signal.removeEventListener('abort', kill);
++            if (err) reject(err); else resolve(value);
++        };
++        if (signal) { if (signal.aborted) kill(); else signal.addEventListener('abort', kill, { once: true }); }
++        proc.stdout.on('data', (d) => { stdout = (stdout + d).slice(-200000); });
++        proc.stderr.on('data', (d) => { stderr = (stderr + d).slice(-4000); });
++        proc.on('error', (err) => finish(err));
++        proc.on('close', (code) => {
++            if (code !== 0) return finish(new Error(`ffprobe exited ${code}: ${stderr.trim()}`));
++            try {
++                const info = JSON.parse(stdout);
++                const streams = info.streams || [];
++                const duration = Number(info.format && info.format.duration) || Math.max(0, ...streams.map((s) => Number(s.duration) || 0));
++                finish(null, { video: streams.find((s) => s.codec_type === 'video') || null,
++                    audio: streams.find((s) => s.codec_type === 'audio') || null, duration });
++            } catch (err) { finish(err); }
++        });
++    });
++}
++
  /** Budget for one stream copy of `bytes`: ~45 s per GB, at least 3 min, at most 2 h. */
  function budgetMs(bytes) {
      return Math.min(2 * 3600 * 1000, Math.max(180000, Math.round((Number(bytes) || 0) / 1e9 * 45000) + 60000));
@@@ -317,5 +355,5 @@ module.exports = {
      // for the preview jobs (previews.js)
      loadSource, adopt, ffmpeg, inputArgs, probeDuration, workDir, cleanupWork, existing, checkRoom,
      // and object.cmaf (cmaf.js)
--    budgetMs, sha256File,
++    budgetMs, probeStreams, sha256File,
  };
ddiff --git a/server/jobs/worker.js b/server/jobs/worker.js
iindex eeb97a9..c6c3770 100644
---- a/server/jobs/worker.js
++++ b/server/jobs/worker.js
@@@ -81,7 +81,7 @@ async function execute(row, laneName) {
          } catch { /* next beat */ }
      }, Math.max(1000, Math.floor(leaseS * 1000 / 3)));
      if (beat.unref) beat.unref();
--    const timeoutMs = spec ? spec.timeoutMs : 1000;
++    const timeoutMs = spec ? (typeof spec.timeoutMs === 'function' ? spec.timeoutMs(row) : spec.timeoutMs) : 1000;
      const limit = setTimeout(() => ac.abort(new Error(`timed out after ${Math.round(timeoutMs / 1000)} s`)), timeoutMs);
      if (limit.unref) limit.unref();
      if (cancelRequested) ac.abort(new Error('cancelled by its owner'));
ddiff --git a/server/objects/timeline-queue.js b/server/objects/timeline-queue.js
iindex 2fed806..a13d0dc 100644
---- a/server/objects/timeline-queue.js
++++ b/server/objects/timeline-queue.js
@@@ -5,7 +5,8 @@
   * finalize of a recording (server/vod/finalize.js), and lazily from GET …/download?format=json so a VOD older than F3.1
   * gets one on its first request. Nothing happens (null) with MEDIA_HLS_ENABLED off, for a missing, unready or non-media
   * 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.
++ * idempotency key `object.cmaf:<id>` (or `:transcode` when first queued with transcoding on) answers a later call with
++ * the same job. → 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)
@@@ -54,6 +55,16 @@ async function queueCmaf(appId, objectId, { segment_seconds = null, lazy = false
      }
      const queue = require('../jobs/queue');
      const params = require('../jobs/cmaf').spec.validate({ appId, obj, params: segment_seconds != null ? { segment_seconds } : {} });
++    // Mark jobs first queued with transcoding on. An older copy-only codec failure gets this one new key on a lazy
++    // request. A failed transcode job already has the key and is replayed, so lazy reads cannot start a retry loop.
++    let idempotencyKey = `object.cmaf:${obj.id}`;
++    if (config.hls.transcode) {
++        const last = await require('../db/database').get(`SELECT status, error_code, idempotency_key FROM media_jobs
++            WHERE app_id = ? AND object_id = ? AND job_type = 'object.cmaf'
++            ORDER BY created_at DESC, id DESC LIMIT 1`, [appId, obj.id]);
++        if (!last || last.idempotency_key === `${idempotencyKey}:transcode` ||
++            (lazy && last.status === 'failed' && last.error_code === 'codec_unsupported')) idempotencyKey += ':transcode';
++    }
      if (lazy) {
          if (Math.max(recent, lazyQueuedAt.length) + pendingLazy >= config.hls.backfillPerHour) return null;
          pendingLazy++;
@@@ -61,7 +72,7 @@ async function queueCmaf(appId, objectId, { segment_seconds = null, lazy = false
      try {
          const r = await queue.enqueue({
              appId, type: 'object.cmaf', objectId: obj.id, params,
--            dedupeActive: true, idempotencyKey: `object.cmaf:${obj.id}`, createdBy: lazy ? 'system:timeline-backfill' : 'system:timeline',
++            dedupeActive: true, idempotencyKey, createdBy: lazy ? 'system:timeline-backfill' : 'system:timeline',
          });
          if (r.created && lazy) lazyQueuedAt.push(now);
          if (r.created) require('../jobs/worker').kick();
ddiff --git a/server/placement/metrics-binding.js b/server/placement/metrics-binding.js
iindex 96231a4..27eaf58 100644
---- a/server/placement/metrics-binding.js
++++ b/server/placement/metrics-binding.js
@@@ -16,6 +16,7 @@
   *   media_presign_cache_total{result}                    counter
   *   media_storage_alerts_total{kind}                     counter
   *   media_sweep_demand_source_total{source}              counter (valkey | pg: the object sweep's demand signal)
++ *   media_cmaf_transcodes_total{stream}                  counter (video or audio streams encoded by object.cmaf)
   */
  'use strict';
  
@@@ -30,6 +31,7 @@ let decisions = null;   // counter
  let presign = null;     // counter
  let alerts = null;      // counter
  let demandSource = null; // counter
++let cmafTranscodes = null; // counter
  
  function bind(registry) {
      if (latency) return;   // idempotent
@@@ -77,6 +79,12 @@ function bind(registry) {
          labelNames: ['source'],
      });
      for (const source of ['valkey', 'pg']) demandSource.inc({ source }, 0);
++    cmafTranscodes = registry.counter({
++        name: 'media_cmaf_transcodes_total',
++        help: 'Streams transcoded by successful object.cmaf jobs',
++        labelNames: ['stream'],
++    });
++    for (const stream of ['video', 'audio']) cmafTranscodes.inc({ stream }, 0);
  
      // Bind every hook the placement modules use.
      router.bindMetric((name, labels) => inc(name, labels));
@@@ -90,6 +98,7 @@ function inc(name, labels) {
      else if (name === 'media_storage_alerts_total') alerts && alerts.inc(labels || {}, 1);
      else if (name === 'media_provider_errors_total') errors && errors.inc(labels || {}, 1);
      else if (name === 'media_sweep_demand_source_total') demandSource && demandSource.inc(labels || {}, 1);
++    else if (name === 'media_cmaf_transcodes_total') cmafTranscodes && cmafTranscodes.inc(labels || {}, 1);
  }
  
  /** Hook the signals module's recordSuccess/recordFailure to fill the histogram/counter. */
@@@ -97,4 +106,4 @@ function observeLatency(provider, op, ms) { if (latency && ms > 0) latency.obser
  function observeError(provider, op) { errors && errors.inc({ provider, op }, 1); }
  function alert(kind) { alerts && alerts.inc({ kind }, 1); }
  
--module.exports = { bind, inc, observeLatency, observeError, alert, _reset: () => { latency = errors = breakerGauge = decisions = presign = alerts = demandSource = null; } };
++module.exports = { bind, inc, observeLatency, observeError, alert, _reset: () => { latency = errors = breakerGauge = decisions = presign = alerts = demandSource = cmafTranscodes = null; } };
ddiff --git a/test/cmaf-transcode.test.js b/test/cmaf-transcode.test.js
nnew file mode 100644
iindex 0000000..8c54dd4
---- /dev/null
++++ b/test/cmaf-transcode.test.js
@@@ -0,0 +1,100 @@
++'use strict';
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { spawnSync } = require('child_process');
++
++(async () => {
++    // Like media-timeline.test.js: the encode needs ffmpeg/ffprobe (with libx264, libvpx, libopus), which CI may not have.
++    if (spawnSync('ffmpeg', ['-version']).status !== 0 || spawnSync('ffprobe', ['-version']).status !== 0) {
++        console.log('cmaf transcode: skipped (ffmpeg not found)');
++        return;
++    }
++    const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-cmaf-transcode-'));
++    const dir = (name) => { const p = path.join(tmp, name); fs.mkdirSync(p, { recursive: true }); return p; };
++    Object.assign(process.env, {
++        VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'), THUMBNAILS_PATH: dir('thumbnails'),
++        PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'), MEDIA_HLS_ENABLED: '1', MEDIA_CMAF_TRANSCODE: '0',
++        MEDIA_UPLOAD_MIN_FREE_MB: '0', MEDIA_INVARIANT_SCAN_HOURS: '0',
++    });
++    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 ffmpeg = (...args) => {
++        const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args], { encoding: 'utf8' });
++        assert.strictEqual(r.status, 0, r.stderr);
++    };
++    const probe = (file) => {
++        const r = spawnSync('ffprobe', ['-v', 'error', '-show_entries', 'stream=codec_name', '-of', 'json', file], { encoding: 'utf8' });
++        assert.strictEqual(r.status, 0, r.stderr);
++        return JSON.parse(r.stdout).streams.map((s) => s.codec_name);
++    };
++    try {
++        const config = require('../server/config');
++        const db = require('../server/db/database');
++        const model = require('../server/objects/model');
++        const timeline = require('../server/objects/timeline');
++        const cmaf = require('../server/jobs/cmaf');
++        const queue = require('../server/jobs/queue');
++        const { queueCmaf } = require('../server/objects/timeline-queue');
++        await db.upsertApp({ app_id: 'live', api_key: 'cmaf-test' });
++        const webm = path.join(tmp, 'browser.webm');
++        const mp4 = path.join(tmp, 'compatible.mp4');
++        const audioOnly = path.join(tmp, 'audio.m4a');
++        ffmpeg('-f', 'lavfi', '-i', 'testsrc=duration=7:size=160x90:rate=10', '-f', 'lavfi', '-i', 'sine=duration=7',
++            '-c:v', 'libvpx', '-deadline', 'realtime', '-cpu-used', '8', '-c:a', 'libopus', '-shortest', webm);
++        ffmpeg('-f', 'lavfi', '-i', 'testsrc=duration=7:size=160x90:rate=10', '-f', 'lavfi', '-i', 'sine=duration=7',
++            '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', mp4);
++        ffmpeg('-f', 'lavfi', '-i', 'sine=duration=7', '-c:a', 'aac', audioOnly);
++        const object = async (file, mime) => {
++            const id = await model.createObject({ app_id: 'live', owner_user_id: 1, visibility: 'public', lifecycle_status: 'ready',
++                kind: 'file', mime_type: mime, size_bytes: fs.statSync(file).size });
++            await model.upsertLocation(id, { provider: 'local', key: file, state: 'present', size_bytes: fs.statSync(file).size, verified: true });
++            return id;
++        };
++        const browser = await object(webm, 'video/webm');
++        const compatible = await object(mp4, 'video/mp4');
++        const onlyAudio = await object(audioOnly, 'audio/mp4');
++        const run = async (id, suffix) => await cmaf.spec.run({ id: `mjob_${suffix}`, app_id: 'live', object_id: id, params: { segment_seconds: 2 } },
++            { signal: new AbortController().signal });
++        await assert.rejects(run(browser, 'off'), (e) => e.code === 'codec_unsupported' && e.permanent);
++        config.hls.transcode = true;
++        config.hls.transcodeMaxSeconds = 1;
++        await assert.rejects(run(browser, 'long'), (e) => e.code === 'too_long_to_transcode' && e.permanent);
++        config.hls.transcodeMaxSeconds = 14400;
++        const made = await run(browser, 'on');
++        assert.deepStrictEqual([made.video, made.audio], ['h264', 'aac']);
++        const rows = await timeline.list(browser, 'source');
++        assert.ok(rows.length >= 3);
++        const joined = path.join(tmp, 'segment.mp4');
++        fs.writeFileSync(joined, Buffer.concat([fs.readFileSync(rows[0].local_path), fs.readFileSync(rows[1].local_path)]));
++        assert.deepStrictEqual(probe(joined).sort(), ['aac', 'h264']);
++        const meta = model.parseJson((await model.getObject(browser)).metadata, {}).renditions.source;
++        const master = timeline.masterPlaylist([{ name: 'source', rows, meta }]);
++        assert.match(master, /CODECS="avc1\.[0-9a-f]+,mp4a\.40\.2"/);
++        const copied = await run(compatible, 'copy');
++        assert.deepStrictEqual([copied.video, copied.audio], ['copy', 'copy']);
++        const audioResult = await run(onlyAudio, 'audio');
++        assert.deepStrictEqual([audioResult.video, audioResult.audio], ['copy', 'copy'], 'audio-only input keeps the copy path');
++
++        // An old failed copy gets one distinct lazy retry. A failed retry settles under its own key.
++        const old = await object(webm, 'video/webm');
++        config.hls.transcode = false;
++        const first = await queueCmaf('live', old);
++        await db.run("UPDATE media_jobs SET status = 'failed', error_code = 'codec_unsupported' WHERE id = ?", [first]);
++        config.hls.transcode = true;
++        const retry = await queueCmaf('live', old, { lazy: true });
++        assert.notStrictEqual(retry, first);
++        const retryRow = await queue.get(retry);
++        assert.strictEqual(retryRow.idempotency_key, `object.cmaf:${old}:transcode`);
++        await db.run("UPDATE media_jobs SET status = 'failed', error_code = 'codec_unsupported' WHERE id = ?", [retry]);
++        assert.strictEqual(await queueCmaf('live', old, { lazy: true }), retry);
++        assert.strictEqual((await db.all("SELECT id FROM media_jobs WHERE object_id = ? AND job_type = 'object.cmaf'", [old])).length, 2);
++        const fresh = await object(webm, 'video/webm');
++        const freshJob = await queueCmaf('live', fresh, { lazy: true });
++        await db.run("UPDATE media_jobs SET status = 'failed', error_code = 'codec_unsupported' WHERE id = ?", [freshJob]);
++        assert.strictEqual(await queueCmaf('live', fresh, { lazy: true }), freshJob, 'a new transcode-on failure does not retry');
++        config.hls.backfillPerHour = 0;
++        assert.strictEqual(await queueCmaf('live', old, { lazy: true }), null, 'the retry still respects the lazy cap');
++        console.log('✅ CMAF transcode, copy, duration cap, codecs and one lazy retry');
++    } finally { fs.rmSync(tmp, { recursive: true, force: true }); }
++})().catch((err) => { console.error(err); process.exitCode = 1; });