openvibers/openvibe.media Commits

Pack durable CMAF segments into ~60 s chunks and serve them as ranges

OP OpenVibers committed 4 Oct 2026 · 3573f61507 · 13 files, +517 −48

Asked: T4 F3.3 plus the F3.1 review follow-ups — OpenVibe.Media: pack the CMAF timeline into ~60 s chunk objects. ONE pull request from current origin/main (git fetch first; read with `git show origin/main:<path>`, the local checkout may be stale). F3.1 is merged: migrations/0002_media_timeline.sql (media…
Harness job media-t4-f3-3-plus-the-f3-1-162714 (plan T4).

Changes

ddiff --git a/README.md b/README.md
iindex 4f7427c..d762fd1 100644
---- a/README.md
++++ b/README.md
@@@ -108,6 +108,7 @@ server/
cancellation), routes (/api/v2/:app/jobs), types: thumbnail.regenerate,
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),
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)
@@@ -457,7 +458,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). Same check as `GET /o/:id`: private and sandbox objects only with the object's `?exp&sig` (`/api/v2/:app/objects/:id/download?format=json` → `hls_url`), carried onto every URI; deleted 410. The `object.cmaf` job makes the timeline. `noindex` |
++| `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`), carried onto every URI; deleted 410. The `object.cmaf` job makes the timeline. `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 813ab9b..58f0b3a 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -37,7 +37,8 @@
"placement demand rollups on Valkey (F2.4): reads per object × region × 5-minute bucket, hotness over the hour, media.object.hot once per object per hour",
"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",
-- "segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks"
++ "segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
++ "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"
],
"updated": "2026-10-03"
}
ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex 489ad73..897cbdf 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -91,7 +91,7 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
`0002_media_timeline.sql` adds `media_timeline` (object, rendition, segment number `seq` — 0 is the init segment —,
`start_ms`/`end_ms` (contiguous, half-open), `keyframe_ms`, the segment's `key` `<object>/<rendition>/<version>/<name>`
(`version` = the sha prefix of the bytes, so a re-cut writes beside the old segment, never over it), `local_path`,
-- `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length` (packing is F3.3; NULL until then),
++ `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length` (set by F3.3 packing; NULL until a segment is packed),
`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`**
@@@ -111,11 +111,34 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
`hls_url` when a timeline exists. A purge, or a vod/clip deleted for good, removes the segments and rows; a durable
copy that will not delete keeps its row so the next pass retries the key; a held object keeps them all. Storage orphan
reports count timeline files and keys as wanted.
++ The durable provider of the segments is the placement router's choice (`timeline.durableProvider`: the source's own
++ best-ranked remote copy, else the best configured healthy provider; purpose `durable` ranks B2 before R2), and the
++ segment route serves the router's ranked copy; a private or sandbox HLS answer carries no
++ `Access-Control-Allow-Origin` (as `GET /o/:id`), a public one `*`.
++- **F3.3, shipped: packing after finalize.** The heavy-lane job **`object.pack`** (`server/jobs/pack.js`, `params`
++ `rendition` (`source`), `target_seconds` 10–300, 60 by default) concatenates each run of contiguous, durable, unpacked
++ media segments of one rendition into chunk objects of about the target (a tail under half of it joins the chunk
++ before; runs of one segment and the init segment stay as they are; a segment not durable yet ends a run). Each
++ segment's bytes are checked against its row's sha256 (this node's copy, else the durable one through the router); the
++ chunk goes to the router's durable provider under `<object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s` and is
++ verified there (sha256 read back), this node keeps a copy where it kept the segments, and the rows move to it in one
++ transaction: `key`/`local_path`/`durable_provider` name the chunk, `packed_object_id` is its sha256 (content address),
++ `byte_offset`/`byte_length` the segment inside it; name, times and sha256 stay the segment's. Only after the commit are
++ the per-segment files and keys **deleted**: the chunk holds the same verified bytes, keeping both would double the
++ stored bytes and every per-segment object packing exists to remove, and a delete that fails leaves an orphan the
++ storage report names. A re-cut in between rolls the commit back and drops the chunk. The job never deletes a row, is
++ idempotent (a rerun packs nothing), refuses a held object, honours the abort signal and `derive.budgetMs` and removes
++ its work directory; `object.cmaf` keeps a packed row whose bytes it reproduces. The playlists do not change: the
++ segment route answers a packed segment as a byte range of its chunk (from this node's file, or a ranged GET of the
++ durable chunk, since a redirect cannot carry the range) under the same check and signature; the `EXT-X-BYTERANGE`
++ form over chunk URIs is not used, so URIs, signed playlists and caches stay as they were. Purge, VOD delete, holds
++ and the orphan report see a chunk through the rows that name it (each location deleted once, every row kept while its
++ chunk's delete fails). All of it behind `MEDIA_HLS_ENABLED`; no schema change (F3.1's columns carry it).
**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for
-- live segments and its upload-lag metric, packing into ~60 s chunks (F3.3), virtual and materialized clips over the
-- timeline, sprites and captions on it, the faststart MP4 fallback as a derivative, segment-bucket demand, a finalize
-- hook that queues `object.cmaf` by itself, signed playlists that outlive one signed-URL lifetime, and Live's player
-- moving to HLS. Renditions are F4.
++ live segments and its upload-lag metric, virtual and materialized clips over the timeline (chunk reference counts),
++ sprites and captions on it, the faststart MP4 fallback as a derivative, segment-bucket demand, a finalize hook that
++ queues `object.cmaf` and `object.pack` by itself, signed playlists that outlive one signed-URL lifetime, and Live's
++ player moving to HLS. Renditions are F4.
## 4. Clips reuse the source
@@@ -267,8 +290,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
and per-class monthly R2 storage ceilings on its moves (F2.6, shipped), the move cleanup job and its alert after
three failures (§6, shipped). Still open in F2: every class (not only native objects ↔ R2) through the one sweep.
- **F3 segment-native video:** the timeline index and the CMAF/HLS source representation of finished video (F3.1,
-- shipped; §3); still open: CMAF/HLS recording with the growing DVR playlist from OpenRe, packing after finalize,
-- virtual clips, sprites and captions on the timeline, MP4 fallback; Live's player moves to HLS.
++ shipped; §3), packing into ~60 s chunks (F3.3, shipped; §3); still open: live DVR from OpenRe (CMAF/HLS recording
++ with the growing playlist), virtual clips, sprites and captions on the timeline, MP4 fallback; 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.
- **F5 multi-CDN delivery:** Bunny path (B2 origin), R2 custom-domain path, OpenVibe edge first up to capacity, route
ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex 4a2dbab..533bfac 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -27,7 +27,7 @@ Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`,
| `media_holds` | retention holds (with a staff `note`), see [Holds](#retention-holds) |
| `media_namespaces` | every namespace: `namespace`, `app_id` (tenant), `parent`, `owner` (`service:<app>` or `project:<prj_…>`), `policy` (JSON), `quota_bytes`, `quota_objects`, and the usage snapshot `used_bytes`, `used_objects`, `reserved_bytes`, `reserved_objects`, `reconciled_at`. See [Namespaces](#namespaces-grants-and-quotas) |
| `media_quota_reservations` | quota held by an upload in progress: `object_id`, `app_id`, `namespace`, `bytes`, `expires_at` |
--| `media_timeline` | the timeline index of segment-native video (F3.1, [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)): one row per CMAF segment of one rendition (`object_id`, `rendition`, `seq`: 0 is `init.mp4`), `start_ms`/`end_ms`/`keyframe_ms`, `key` (`<object>/<rendition>/<version>/<name>`, `version` = the sha prefix so a re-cut writes beside the old segment), `local_path`, `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length`, `sha256`, `durability` (`local` `durable`). Written by `object.cmaf`; read by the HLS routes and `timeline.segmentAt` |
++| `media_timeline` | the timeline index of segment-native video (F3.1, [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)): one row per CMAF segment of one rendition (`object_id`, `rendition`, `seq`: 0 is `init.mp4`), `start_ms`/`end_ms`/`keyframe_ms`, `key` (`<object>/<rendition>/<version>/<name>`, `version` = the sha prefix so a re-cut writes beside the old segment), `local_path`, `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length`, `sha256`, `durability` (`local` `durable`). Written by `object.cmaf`; `object.pack` (F3.3) points packed rows at their ~60 s chunk (`key`/`local_path`/`durable_provider` the chunk's, `packed_object_id` its sha256, `byte_offset`/`byte_length` the segment in it); read by the HLS routes and `timeline.segmentAt` |
| `media_invariant_violations` | public playback objects over the size policy, see [Invariant](#public-object-size-invariant) |
Location `state`:
@@@ -759,7 +759,8 @@ job over only with the token it read and, past boot, only while the lease is sti
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `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 B2 configured, as `<object>/source/<version>/<name>` on B2 (`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)) |
++| `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 |
| `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 }` |
@@@ -986,7 +987,7 @@ deleted object's R2 copy stays (so its purge waits) until the gate is on again.
| `MEDIA_UPLOAD_MIN_FREE_MB` | 10240 | free disk kept in reserve by multipart uploads and split/remux jobs |
| `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` job and the HLS routes `/o/:id/master.m3u8`, `/o/:id/source/…` (off: the job is refused, the routes 404) |
++| `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_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 9c438c5..eced9c1 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -85,7 +85,7 @@ const config = {
},
// Job system (server/jobs/; docs/object-model.md#jobs).
-- // Segment-native video (docs/media-fabric.md §3, F3.1): the object.cmaf job and the HLS playlists read from
++ // Segment-native video (docs/media-fabric.md §3, F3.1/F3.3): the object.cmaf and object.pack jobs and the HLS playlists read from
// 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()),
ddiff --git a/server/jobs/cmaf.js b/server/jobs/cmaf.js
iindex 63f02af..a979fa5 100644
---- a/server/jobs/cmaf.js
++++ b/server/jobs/cmaf.js
@@@ -11,8 +11,11 @@
* result { source_id, rendition: 'source', segments, duration_ms, bytes, durable, local_only,
* rows: { inserted, updated, unchanged, removed }, uploaded, master_path }
*
-- * Bytes: each segment lands on this node's disk (OBJECTS_PATH/.timeline/…); with B2 configured it is uploaded under
-- * <object>/source/<version>/<name> (version = the sha's prefix) and its row says `durable` once B2 confirms the size.
++ * 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.
* 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
@@@ -32,7 +35,6 @@ const { JobError } = require('./queue');
const MB = 1024 * 1024;
const DEFAULT_SEGMENT_SECONDS = 4;
--const DURABLE = 'b2';
function disabled() {
return new JobError('media.hls.disabled', 'Segment-native video is off on this server (MEDIA_HLS_ENABLED)', { permanent: true, status: 409 });
@@@ -111,7 +113,7 @@ async function run(job, ctx) {
const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));
const vodStorage = require('../vod/vod-storage');
-- const durableOn = vodStorage.providerConfigured(DURABLE);
++ const target = await timeline.durableProvider(src);
const rows = [];
let uploaded = 0;
const failed = [];
@@@ -129,6 +131,11 @@ async function run(job, ctx) {
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);
@@@ -143,12 +150,12 @@ async function run(job, ctx) {
fs.renameSync(tmp, dest);
stagedLocal.push(dest);
}
-- if (!durable && durableOn) {
++ if (!durable && target) {
try {
-- if (!vodStorage.providerAvailable(DURABLE)) throw new Error(`${DURABLE} is unavailable`);
-- await vodStorage.uploadFile(DURABLE, key, localPath, f.seq === 0 ? 'video/mp4' : 'video/iso.segment', { signal: ctx.signal });
-- stagedRemote.push({ provider: DURABLE, key });
-- durable = DURABLE;
++ 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"
ddiff --git a/server/jobs/pack.js b/server/jobs/pack.js
nnew file mode 100644
iindex 0000000..3aa16cd
---- /dev/null
++++ b/server/jobs/pack.js
@@@ -0,0 +1,204 @@
++/**
++ * OpenVibe.Media — job type object.pack (heavy lane; docs/media-fabric.md §3, F3.3)
++ *
++ * Packing after finalize: the durable media segments of one rendition, contiguous and not packed yet, are concatenated
++ * into chunk objects of about `params.target_seconds` (60) each — ~30× fewer objects, PUTs, HEADs and keys — and every
++ * segment's media_timeline row then names its chunk: key/local_path/durable_provider the chunk's, packed_object_id its
++ * sha256 (the chunk's content address), byte_offset/byte_length the segment inside it. Its name, times and sha256 stay
++ * the segment's, so the playlists and their URIs do not change; the segment route serves a packed segment as a ranged
++ * read of its chunk. The init segment is not packed.
++ *
++ * params { rendition ('source'), target_seconds (10-300, 60) }
++ * result { source_id, rendition, packs, segments, bytes, provider, rows_packed, already_packed, waiting, removed_segments, left_behind }
++ *
++ * Per chunk, two-phase: every segment's bytes are checked against its row's sha256 (this node's copy, else the durable
++ * one through the placement router); the chunk is uploaded to the router's durable provider under
++ * <object>/<rendition>/<chunk sha prefix>/p<first seq>.m4s and verified there (size + sha256 read back); the rows move
++ * to it in one transaction, and only then are the per-segment files and keys deleted — the chunk holds the same,
++ * verified bytes, and keeping both would double the stored bytes and keep every per-segment object the packing exists
++ * to remove. A delete that fails leaves an orphan the storage report names. A chunk whose rows changed meanwhile (a
++ * re-cut) is dropped again before anything is deleted. Never deletes a timeline row.
++ * Idempotent: a rerun finds every segment packed and packs nothing. Runs of one segment are not packed. A segment
++ * that is not durable yet ends a run: it is packed by a later run once it is. Honours the abort signal (an upload in
++ * flight stops) and a budget of derive.budgetMs(the rendition's bytes); the work directory is always removed.
++ * Refused: MEDIA_HLS_ENABLED off, a non-media object or a missing timeline (permanent), an object under a hold (retried).
++ */
++'use strict';
++
++const fs = require('fs');
++const path = require('path');
++const crypto = require('crypto');
++const config = require('../config');
++const timeline = require('../objects/timeline');
++const derive = require('./derive');
++const { JobError } = require('./queue');
++
++const DEFAULT_TARGET_SECONDS = 60;
++
++function disabled() {
++ return new JobError('media.hls.disabled', 'Segment-native video is off on this server (MEDIA_HLS_ENABLED)', { permanent: true, status: 409 });
++}
++
++function validate({ obj, params }) {
++ if (!config.hls.enabled) throw disabled();
++ if (!obj) throw new JobError('media.job.invalid', 'object.pack needs object_id', { permanent: true });
++ if (!derive.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video/audio objects) have a timeline', { permanent: true });
++ const p = { ...(params || {}) };
++ if (p.rendition != null && !/^[a-z0-9_-]{1,32}$/.test(String(p.rendition))) throw new JobError('media.job.invalid', 'params.rendition is a rendition name', { permanent: true });
++ if (p.target_seconds != null) {
++ const s = Number(p.target_seconds);
++ if (!Number.isFinite(s) || s < 10 || s > 300) throw new JobError('media.job.invalid', 'params.target_seconds must be from 10 to 300', { permanent: true });
++ p.target_seconds = s;
++ }
++ return p;
++}
++
++const ms = (r) => Number(r.end_ms) - Number(r.start_ms);
++
++/**
++ * The chunks to make: maximal runs of contiguous, durable, unpacked media segments, cut once a chunk reaches the target;
++ * a short tail joins the chunk before it. Runs of one segment stay as they are. → [[row, …], …] plus the counts.
++ */
++function plan(rows, targetMs) {
++ const segs = rows.filter((r) => Number(r.seq) > 0);
++ const runs = [];
++ let run = [];
++ for (const r of segs) {
++ const fits = !r.packed_object_id && r.durability === 'durable' && r.durable_provider;
++ const next = run.length && Number(r.seq) === Number(run[run.length - 1].seq) + 1;
++ if (fits && (next || !run.length)) { run.push(r); continue; }
++ if (run.length) runs.push(run);
++ run = fits ? [r] : [];
++ }
++ if (run.length) runs.push(run);
++ const chunks = [];
++ for (const rn of runs) {
++ if (rn.length < 2) continue;
++ const mine = [];
++ let cur = [];
++ let dur = 0;
++ for (const r of rn) {
++ cur.push(r);
++ dur += ms(r);
++ if (dur >= targetMs) { mine.push(cur); cur = []; dur = 0; }
++ }
++ if (cur.length) {
++ if (mine.length && dur < targetMs / 2) mine[mine.length - 1].push(...cur);
++ else mine.push(cur);
++ }
++ chunks.push(...mine.filter((c) => c.length > 1));
++ }
++ return {
++ chunks,
++ alreadyPacked: segs.filter((r) => r.packed_object_id).length,
++ waiting: segs.filter((r) => !r.packed_object_id && r.durability !== 'durable').length,
++ };
++}
++
++const aborted = (ctx) => { if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted'); };
++
++/** The segment's bytes as a file in `dir`, checked against its row: this node's copy, else the durable one (router). */
++async function segmentFile(row, dir, ctx) {
++ if (row.local_path && derive.existing(row.local_path) && await derive.sha256File(row.local_path) === row.sha256) return row.local_path;
++ const router = require('../placement/router');
++ const vodStorage = require('../vod/vod-storage');
++ const decision = await router.route({ locations: timeline.locationsOf({ ...row, local_path: null }), purpose: 'derive', presign: false });
++ for (const c of decision.candidates || []) {
++ const url = await vodStorage.presignGet(c.provider, c.key, 900).catch(() => null);
++ if (!url) continue;
++ const res = await fetch(url, { signal: ctx.signal }).catch((err) => { aborted(ctx); return { ok: false, err }; });
++ if (!res.ok) continue;
++ const file = path.join(dir, `seg-${row.seq}`);
++ fs.writeFileSync(file, Buffer.from(await res.arrayBuffer()));
++ if (await derive.sha256File(file) === row.sha256) return file;
++ }
++ throw new JobError('segment_unreadable', `${row.name}: no copy with the indexed sha256`, { retryAfterS: 1800 });
++}
++
++async function run(job, ctx) {
++ if (!config.hls.enabled) throw disabled();
++ const started = Date.now();
++ const src = await derive.loadSource(job);
++ const rendition = String(job.params.rendition || timeline.SOURCE);
++ const rows = await timeline.list(src.id, rendition);
++ if (!rows.some((r) => Number(r.seq) > 0)) throw new JobError('media.timeline.missing', `No ${rendition} timeline to pack (run object.cmaf first)`, { permanent: true });
++ if (await require('../objects/model').isHeld(src.id)) throw new JobError('media.object.held', 'The object is under a hold: its segments stay as they are', { retryAfterS: 6 * 3600 });
++ const budget = derive.budgetMs(rows.reduce((a, r) => a + Number(r.byte_length), 0));
++ const { chunks, alreadyPacked, waiting } = plan(rows, (job.params.target_seconds || DEFAULT_TARGET_SECONDS) * 1000);
++ const out = { source_id: src.id, rendition, packs: 0, segments: 0, bytes: 0, provider: null, rows_packed: 0, already_packed: alreadyPacked, waiting, removed_segments: 0, left_behind: 0 };
++ if (!chunks.length) return out;
++ const provider = await timeline.durableProvider(src);
++ if (!provider) throw new JobError('media.provider.unavailable', 'No durable provider is configured and healthy for the chunks', { retryAfterS: 1800 });
++ out.provider = provider;
++ const vodStorage = require('../vod/vod-storage');
++ const dir = derive.workDir(job.id);
++ try {
++ for (const chunk of chunks) {
++ aborted(ctx);
++ if (Date.now() - started > budget) throw new JobError('budget_exceeded', `Packing ran past its ${Math.round(budget / 1000)} s budget; the rest is packed by a retry`, { retryAfterS: 60 });
++ // Concatenate, recording where each segment starts in the chunk.
++ const tmp = path.join(dir, 'chunk.m4s');
++ const fd = fs.openSync(tmp, 'w');
++ const hash = crypto.createHash('sha256');
++ const placed = [];
++ let offset = 0;
++ try {
++ for (const r of chunk) {
++ aborted(ctx);
++ const bytes = fs.readFileSync(await segmentFile(r, dir, ctx));
++ fs.writeSync(fd, bytes);
++ hash.update(bytes);
++ placed.push({ row: r, byte_offset: offset });
++ offset += bytes.length;
++ }
++ } finally { fs.closeSync(fd); }
++ const sha = hash.digest('hex');
++ const name = `p${timeline.segmentName(Number(chunk[0].seq))}`;
++ const key = timeline.keyFor(src.id, rendition, name, sha.slice(0, 12));
++ // This node keeps a copy of the chunk where it kept copies of the segments.
++ const keepLocal = chunk.some((r) => r.local_path && derive.existing(r.local_path));
++ const localPath = keepLocal ? timeline.localPathFor(src, rendition, name, sha.slice(0, 12)) : null;
++ let uploaded = false;
++ let committed = false;
++ try {
++ if (!vodStorage.providerAvailable(provider)) throw new JobError('media.provider.unavailable', `${provider} is unavailable`, { retryAfterS: 600 });
++ await vodStorage.uploadFile(provider, key, tmp, 'video/iso.segment', { signal: ctx.signal });
++ uploaded = true;
++ aborted(ctx);
++ // Durable means verified: the provider gives back exactly these bytes.
++ const back = await vodStorage.sha256Object(provider, key);
++ if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });
++ if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }
++ aborted(ctx);
++ committed = await timeline.markPacked(src.id, rendition, placed.map((p) => ({
++ row: p.row, key, local_path: localPath, durable_provider: provider, packed_object_id: sha, byte_offset: p.byte_offset,
++ })), { jobId: job.id });
++ if (!committed) throw new JobError('media.timeline.changed', 'The timeline changed while packing (a re-cut); a retry packs the new segments', { retryAfterS: 60 });
++ } finally {
++ if (!committed) {
++ if (uploaded) await vodStorage.deleteObject(provider, key).catch(() => { /* best effort: the orphan report names it */ });
++ if (localPath) { try { fs.unlinkSync(localPath); } catch { /* not moved */ } }
++ try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }
++ }
++ }
++ // Phase two, after the commit: the per-segment copies the chunk now replaces.
++ const { failed } = await timeline.deleteBytes(chunk);
++ if (failed.length) console.warn(`[Pack] ${job.id}: ${failed.length} packed segment key(s) left behind (the storage orphan report names them), first ${failed[0].provider}:${failed[0].key}`);
++ out.left_behind += failed.length;
++ out.removed_segments += chunk.length - failed.length;
++ out.packs++;
++ out.segments += chunk.length;
++ out.rows_packed += chunk.length;
++ out.bytes += offset;
++ for (const f of fs.readdirSync(dir)) { try { fs.unlinkSync(path.join(dir, f)); } catch { /* gone */ } }
++ }
++ return out;
++ } finally {
++ derive.cleanupWork(job.id);
++ }
++}
++
++module.exports = {
++ spec: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate, run },
++ plan,
++};
ddiff --git a/server/jobs/types.js b/server/jobs/types.js
iindex 354aed8..19482c6 100644
---- a/server/jobs/types.js
++++ b/server/jobs/types.js
@@@ -8,6 +8,7 @@
* object.waveform heavy the audio waveform (PNG) of a vod/clip as the source's `waveform` variant
* 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)
* 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
@@@ -26,6 +27,7 @@ queue.register('object.remux', require('./derive').remux);
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('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 f9334db..fccb53e 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 MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a
++ * heavy object.split, object.remux, object.cmaf, object.pack 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/routes.js b/server/objects/routes.js
iindex 09d3dea..d04517b 100644
---- a/server/objects/routes.js
++++ b/server/objects/routes.js
@@@ -737,6 +737,8 @@ const publicRouter = express.Router();
// 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
// (from /download), which a signed playlist carries onto each URI; deleted 410, not ready 404. Off, these paths 404.
++// 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,}\.m4s)$/;
@@@ -752,8 +754,9 @@ async function hlsObject(req, res, next) {
if (obj.lifecycle_status === 'deleted') { res.status(410).json({ error: 'Gone' }); return null; }
if (obj.lifecycle_status !== 'ready') { res.status(404).json({ error: 'Not found' }); return null; }
res.set({
-- 'X-Content-Type-Options': 'nosniff', 'X-Robots-Tag': 'noindex', 'Access-Control-Allow-Origin': '*',
++ 'X-Content-Type-Options': 'nosniff', 'X-Robots-Tag': 'noindex',
'Cache-Control': closed ? 'private, no-store' : 'public, max-age=300',
++ ...(!closed && { 'Access-Control-Allow-Origin': '*' }),
});
return { obj, closed, query: signed ? `exp=${encodeURIComponent(req.query.exp)}&sig=${encodeURIComponent(req.query.sig)}` : '' };
}
@@@ -782,24 +785,70 @@ publicRouter.get('/:id/source/index.m3u8', hlsRoute(async (req, res, { obj, quer
res.type(HLS_TYPE).send(timeline.mediaPlaylist(rows, { query }));
}));
++/** The client's Range over a `length`-byte body: { start, end } (inclusive), null for the whole, false if unsatisfiable. */
++function sliceRange(header, length) {
++ const m = /^bytes=(\d*)-(\d*)$/.exec(String(header || '').trim());
++ if (!m || (!m[1] && !m[2])) return null;
++ let start = m[1] ? Number(m[1]) : Math.max(0, length - Number(m[2]));
++ let end = m[1] && m[2] ? Math.min(Number(m[2]), length - 1) : length - 1;
++ if (start >= length || end < start) return false;
++ return { start, end };
++}
++
++/**
++ * A packed segment: bytes [offset, offset + length) of its chunk, from a local file or a ranged GET of a presigned URL,
++ * answered as if the segment were its own file (Range over the segment, 206/416, Content-Length).
++ */
++async function sendSlice(req, res, { file, url, offset, length, headers }) {
++ const r = sliceRange(req.headers.range, length);
++ if (r === false) { res.writeHead(416, { 'Content-Range': `bytes */${length}`, ...headers }); return res.end(); }
++ const { start, end } = r || { start: 0, end: length - 1 };
++ const head = { ...headers, 'Accept-Ranges': 'bytes', 'Content-Length': end - start + 1, ...(r && { 'Content-Range': `bytes ${start}-${end}/${length}` }) };
++ if (file) {
++ res.writeHead(r ? 206 : 200, head);
++ const stream = fs.createReadStream(file, { start: offset + start, end: offset + end });
++ stream.on('error', () => res.destroy());
++ return stream.pipe(res);
++ }
++ const up = await fetch(url, { headers: { range: `bytes=${offset + start}-${offset + end}` } }).catch(() => null);
++ const body = up && up.status === 206 ? Buffer.from(await up.arrayBuffer()) : null;
++ if (!body || body.length !== end - start + 1) return false;
++ res.writeHead(r ? 206 : 200, head);
++ return res.end(body);
++}
++
publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, { obj, closed }) => {
const timeline = require('./timeline');
const name = String(req.params.name || '');
const row = SEGMENT_NAME.test(name) ? await timeline.byName(obj.id, timeline.SOURCE, name) : null;
if (!row) return res.status(404).json({ error: 'Not found' });
const mime = Number(row.seq) === 0 ? 'video/mp4' : 'video/iso.segment';
-- // This node's copy first; otherwise a short presigned URL to the durable one.
-- if (row.local_path && fs.existsSync(row.local_path)) {
-- return require('../public/routes').streamFileWithRange(req, res, row.local_path, {
-- 'Content-Type': mime, 'Cache-Control': closed ? 'private, no-store' : 'public, max-age=3600',
-- });
++ const headers = { 'Content-Type': mime, 'Cache-Control': closed ? 'private, no-store' : 'public, max-age=3600' };
++ const packed = !!row.packed_object_id;
++ const slice = { offset: Number(row.byte_offset), length: Number(row.byte_length), headers };
++ // The router ranks this node's copy and the durable one (no viewer demand counted per segment: segment-bucket demand
++ // comes later). A local file is 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), purpose: 'playback', session: router.sessionFor(req, `${obj.id}/source`),
++ contentType: mime, expiresIn: 300, presign: !packed });
++ for (const loc of decision.candidates || []) {
++ if (loc.provider === 'local') {
++ if (!fs.existsSync(loc.key)) continue;
++ if (packed) return sendSlice(req, res, { file: loc.key, ...slice });
++ return require('../public/routes').streamFileWithRange(req, res, loc.key, headers);
++ }
++ const url = (!packed && loc.provider === decision.provider && decision.url)
++ || await require('../vod/vod-storage').presignGet(loc.provider, loc.key, 300, packed ? {} : { contentType: mime }).catch(() => null);
++ if (!url) continue;
++ if (packed) {
++ if (await sendSlice(req, res, { url, ...slice }) !== false) return;
++ continue;
++ }
++ res.set('Cache-Control', 'private, max-age=0');
++ return res.redirect(302, url);
}
-- const url = row.durable_provider
-- ? await require('../vod/vod-storage').presignGet(row.durable_provider, row.key, 300, { contentType: mime }).catch(() => null)
-- : null;
-- if (!url) return res.status(404).json({ error: 'Segment bytes unavailable' });
-- res.set('Cache-Control', 'private, max-age=0');
-- res.redirect(302, url);
++ res.status(404).json({ error: 'Segment bytes unavailable' });
}));
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.js b/server/objects/timeline.js
iindex c077765..4582e2e 100644
---- a/server/objects/timeline.js
++++ b/server/objects/timeline.js
@@@ -11,6 +11,11 @@
* on durable_provider. The name alone is the playlist URI, so a row carries the versioned location it serves from. The
* object.cmaf job (server/jobs/cmaf.js) writes the source rendition; the public routes
* /o/:id/master.m3u8 and /o/:id/source/… (objects/routes.js) serve it behind MEDIA_HLS_ENABLED.
++ *
++ * Packed (F3.3): object.pack (server/jobs/pack.js) concatenates ~60 s of durable segments into one chunk object. A packed
++ * row's key/local_path/durable_provider then name the chunk, packed_object_id its sha256 (content address) and
++ * byte_offset/byte_length the segment's bytes inside it; its name, times and sha256 stay the segment's. Several rows
++ * share one chunk, so the bytes are deleted once per location and kept (for a retry) while any delete of them fails.
*/
'use strict';
@@@ -96,12 +101,15 @@ async function deleteBytes(rows) {
const vodStorage = require('../vod/vod-storage');
const dirs = new Set();
const failed = [];
++ const done = new Set(); // a packed chunk is named by every row it holds: each location is deleted once
for (const r of rows) {
-- if (r.local_path && path.resolve(r.local_path).startsWith(root)) {
++ if (r.local_path && path.resolve(r.local_path).startsWith(root) && !done.has(`local\n${r.local_path}`)) {
++ done.add(`local\n${r.local_path}`);
try { fs.unlinkSync(r.local_path); } catch { /* already gone */ }
dirs.add(path.dirname(r.local_path));
}
-- if (r.durable_provider && vodStorage.providerConfigured(r.durable_provider)) {
++ if (r.durable_provider && vodStorage.providerConfigured(r.durable_provider) && !done.has(`${r.durable_provider}\n${r.key}`)) {
++ done.add(`${r.durable_provider}\n${r.key}`);
const ok = await vodStorage.deleteObject(r.durable_provider, r.key);
if (!ok) failed.push({ row: r, provider: r.durable_provider, key: r.key, error: 'delete failed' });
}
@@@ -123,17 +131,67 @@ async function removeObject(objectId) {
const rows = await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [objectId]);
if (!rows.length) return { removed: 0, pending: 0 };
const { failed } = await deleteBytes(rows);
-- const stuck = new Set(failed.map((f) => `${f.row.rendition}\n${f.row.seq}`));
++ // Every row naming a location whose delete failed is kept: for a packed chunk that is all the rows it holds.
++ const stuck = new Set(failed.map((f) => `${f.provider}\n${f.key}`));
let removed = 0;
let pending = 0;
for (const r of rows) {
-- if (stuck.has(`${r.rendition}\n${r.seq}`)) { pending++; continue; }
++ if (r.durable_provider && stuck.has(`${r.durable_provider}\n${r.key}`)) { pending++; continue; }
removed += (await db.run('DELETE FROM media_timeline WHERE object_id = ? AND rendition = ? AND seq = ?', [objectId, r.rendition, r.seq])).changes || 0;
}
if (pending) console.warn(`[Timeline] ${objectId}: ${pending} segment row(s) kept — their durable copy could not be deleted`);
return { removed, pending };
}
++/**
++ * Point rows at the chunk they were packed into, in one transaction. `updates` = [{ row, key, local_path,
++ * durable_provider, packed_object_id, byte_offset }]; each row changes only if it is still exactly the unpacked segment
++ * the packer read (same key and sha256, not packed): a re-cut in between makes the whole commit roll back and false is
++ * returned, so the caller drops its chunk. Never deletes a row.
++ */
++async function markPacked(objectId, rendition, updates, { jobId = null } = {}) {
++ const CONFLICT = new Error('timeline changed');
++ try {
++ await db.getDb().tx(async () => {
++ for (const u of updates) {
++ const r = await db.run(`UPDATE media_timeline SET key = ?, local_path = ?, durable_provider = ?, durability = 'durable',
++ packed_object_id = ?, byte_offset = ?, job_id = ?, updated_at = ov_now()
++ WHERE object_id = ? AND rendition = ? AND seq = ? AND key = ? AND sha256 = ? AND packed_object_id IS NULL`,
++ [u.key, u.local_path, u.durable_provider, u.packed_object_id, u.byte_offset, jobId, objectId, rendition, u.row.seq, u.row.key, u.row.sha256]);
++ if (!r.changes) throw CONFLICT;
++ }
++ });
++ return true;
++ } catch (err) {
++ if (err === CONFLICT) return false;
++ throw err;
++ }
++}
++
++/**
++ * The durable provider for bytes derived from `obj` (segments, packed chunks), chosen by the placement router like
++ * every other provider choice (derive.js): the provider of the object's own best-ranked remote copy, else the best
++ * configured, healthy one. Purpose `durable` ranks the canonical tier (B2) before the hot cache (R2). null = none.
++ */
++async function durableProvider(obj) {
++ const router = require('../placement/router');
++ const remote = (d) => (d.candidates || []).find((c) => c.provider !== 'local');
++ const own = remote(await router.route({ object: obj, purpose: 'durable', presign: false }));
++ if (own) return own.provider;
++ const vodStorage = require('../vod/vod-storage');
++ const configured = vodStorage.REMOTE_PROVIDERS.filter((p) => vodStorage.providerConfigured(p)).map((p) => ({ provider: p, key: p, state: 'present' }));
++ const any = remote(await router.route({ locations: configured, purpose: 'durable', presign: false }));
++ return any ? any.provider : null;
++}
++
++/** The copies of a row's bytes as router locations: this node's file, then the durable copy. */
++function locationsOf(row) {
++ const out = [];
++ if (row.local_path && fs.existsSync(row.local_path)) out.push({ provider: 'local', key: row.local_path, state: 'present' });
++ if (row.durable_provider && row.durability === 'durable') out.push({ provider: row.durable_provider, key: row.key, state: 'present' });
++ return out;
++}
++
// ── Playlists (from the rows, never from files) ──────────────
const secs = (ms) => (Number(ms) / 1000).toFixed(3);
@@@ -170,6 +228,7 @@ function masterPlaylist(renditions, { query = '' } = {}) {
}
module.exports = {
-- SOURCE, INIT_NAME, localRoot, localPathFor, keyFor, segmentName,
-- list, segments, byName, has, segmentAt, replace, deleteBytes, removeObject, mediaPlaylist, masterPlaylist,
++ SOURCE, INIT_NAME, FIELDS, localRoot, localPathFor, keyFor, segmentName,
++ list, segments, byName, has, segmentAt, replace, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,
++ mediaPlaylist, masterPlaylist,
};
ddiff --git a/server/placement/router.js b/server/placement/router.js
iindex 04c3c59..704ebd8 100644
---- a/server/placement/router.js
++++ b/server/placement/router.js
@@@ -16,7 +16,7 @@
* 4. Rank by preference tier, with the measured 5-minute EWMA latency breaking ties:
* playback : R2 with a verified present copy (popularity promotion put it there to move
* reads off the host; promoting a VOD also frees its local copy) > local > R2 > B2
-- * else : local > R2 > B2 (canonical); `download` puts B2 before R2 (free ops)
++ * else : local > R2 > B2 (canonical); `download` (free ops) and `durable` (a job's derived bytes) put B2 before R2
* A VOD or clip offers every verified cloud copy its object records (media_locations), so a
* VOD promoted to R2 keeps B2 as its fallback when R2's breaker opens.
* A lower tier displaces a higher one only when both have measurements and the lower
@@@ -140,6 +140,9 @@ function latencyOf(provider) {
* Preference tier (lower wins). Purpose decides only the order between the hot and canonical
* cloud copies; local is always preferred to a cloud copy that is not the hot cache.
*/
++// `download` (free B2 egress ops) and `durable` (where a job keeps bytes it derives: timeline segments and chunks)
++// rank the canonical copy before the hot cache.
++const CANONICAL_FIRST = new Set(['download', 'durable']);
function tierOf(candidate, purpose) {
const p = candidate.provider;
if (p === 'local') return 1;
@@@ -147,9 +150,9 @@ function tierOf(candidate, purpose) {
// A verified R2 copy is the hot cache: popularity promotion put it there precisely so reads leave the
// host (the design's spill, decided by popularity until F2 measures the port).
if (purpose === 'playback' && candidate.verified_at) return 0;
-- return purpose === 'download' ? 3 : 2;
++ return CANONICAL_FIRST.has(purpose) ? 3 : 2;
}
-- if (p === 'b2') return purpose === 'download' ? 2 : 3;
++ if (p === 'b2') return CANONICAL_FIRST.has(purpose) ? 2 : 3;
return 2;
}
ddiff --git a/test/media-timeline.test.js b/test/media-timeline.test.js
iindex 9b7d06a..173f09f 100644
---- a/test/media-timeline.test.js
++++ b/test/media-timeline.test.js
@@@ -4,6 +4,9 @@
// rows (ENDLIST, the rows' durations); segmentAt answers the segment at an instant, at the boundaries too; a rerun is a
// no-op; a private object's playlist refuses an anonymous reader and its signed one carries the signature on; with
// MEDIA_HLS_ENABLED off nothing answers; a purge removes the segments unless the object is held.
++// F3.3: object.pack concatenates the durable segments into a chunk, each row naming its byte range; a packed segment is
++// served as a ranged read of the chunk (from this node or the durable copy) with the original bytes; a rerun packs
++// nothing; the per-segment copies are gone; the durable provider comes from the placement router; private stays private.
const assert = require('assert');
const fs = require('fs');
const os = require('os');
@@@ -25,9 +28,19 @@ const { spawn, spawnSync } = require('child_process');
process.env.MEDIA_UPLOAD_MIN_FREE_MB = '0';
process.env.MEDIA_HLS_ENABLED = '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}`] = '';
++ // The durable providers' bytes ("<provider>:<key>"), served back with Range at /blob/… by the stub (presigned URLs).
++ const blobs = new Map();
const stub = http.createServer((req, res) => {
req.resume();
req.on('end', () => {
++ if (req.url.startsWith('/blob/')) {
++ const b = blobs.get(decodeURIComponent(req.url.slice(6)));
++ if (!b) { res.statusCode = 404; return res.end(); }
++ const m = /^bytes=(\d+)-(\d+)$/.exec(req.headers.range || '');
++ if (!m) return res.end(b);
++ res.statusCode = 206;
++ return res.end(b.subarray(Number(m[1]), Number(m[2]) + 1));
++ }
res.setHeader('Content-Type', 'application/json');
if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
res.statusCode = 404; res.end('{}');
@@@ -107,11 +120,11 @@ const { spawn, spawnSync } = require('child_process');
const priv = await native('private');
const before = JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]);
-- const runJob = async (objectId, params) => {
-- const r = await call('POST', '/api/v2/live/jobs', { body: { type: 'object.cmaf', object_id: objectId, ...(params && { params }) } });
++ const runJob = async (objectId, params, type = 'object.cmaf') => {
++ 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));
const fin = await waitFor(async () => ['succeeded', 'failed'].includes((await queue.get(r.body.job.id)).status));
-- assert.ok(fin, `object.cmaf finishes: ${JSON.stringify(queue.jobPublic(await queue.get(r.body.job.id)))}`);
++ assert.ok(fin, `${type} finishes: ${JSON.stringify(queue.jobPublic(await queue.get(r.body.job.id)))}`);
return queue.jobPublic(await queue.get(r.body.job.id));
};
@@@ -156,14 +169,16 @@ const { spawn, spawnSync } = require('child_process');
assert.strictEqual((await get(`/o/${pub}/source/..%2Findex.m3u8`)).status, 404);
// A real HLS reader agrees: the playlist plays back as the 7 s source.
// (spawned, not spawnSync: the server answering it runs in this process)
-- const probe = await new Promise((resolve) => {
++ const ffprobe = () => new Promise((resolve) => {
const p = spawn('ffprobe', ['-v', 'error', '-rw_timeout', '10000000', '-show_entries', 'format=duration', '-of', 'csv=p=0', `${api}/o/${pub}/master.m3u8`]);
let out = '';
p.stdout.on('data', (d) => { out += d; });
p.stderr.on('data', (d) => { out += d; });
p.on('close', () => resolve(out.trim()));
});
++ const probe = await ffprobe();
assert.ok(Math.abs(Number(probe) - 7) < 0.2, `ffprobe reads the playlist: ${probe}`);
++ assert.strictEqual(master.headers.get('access-control-allow-origin'), '*', 'a public playlist may be read cross-origin');
// ── segmentAt on the real timeline, at the boundaries ──
const seqAt = async (t) => { const r = await timeline.segmentAt(pub, 'source', t); return r ? Number(r.seq) : null; };
@@@ -179,6 +194,96 @@ const { spawn, spawnSync } = require('child_process');
assert.strictEqual(JSON.stringify(await timeline.list(pub)), JSON.stringify(rows), 'the rows are as they were (job_id, updated_at too)');
assert.deepStrictEqual(rows.map((r) => fs.statSync(r.local_path).mtimeMs), mtimes, 'no segment rewritten');
++ // ── F3.3: durable segments (stubbed B2/R2), the provider chosen by the placement router ──
++ const pack = require('../server/jobs/pack');
++ const vodStorage = require('../server/vod/vod-storage');
++ const router = require('../server/placement/router');
++ let configured = ['b2', 'r2'];
++ Object.assign(vodStorage, {
++ providerConfigured: (p) => configured.includes(p),
++ providerAvailable: (p) => configured.includes(p),
++ uploadFile: async (p, key, file) => { blobs.set(`${p}:${key}`, fs.readFileSync(file)); },
++ sha256Object: async (p, key) => (blobs.has(`${p}:${key}`) ? { sha256: sha(blobs.get(`${p}:${key}`)), size: blobs.get(`${p}:${key}`).length } : null),
++ deleteObject: async (p, key) => { blobs.delete(`${p}:${key}`); return true; },
++ presignGet: async (p, key) => `${base}/blob/${encodeURIComponent(`${p}:${key}`)}`,
++ });
++ const realRoute = router.route;
++ const purposes = [];
++ router.route = async (opts) => { purposes.push(opts.purpose); return realRoute(opts); };
++ const j3 = await runJob(pub, { segment_seconds: 2 });
++ assert.strictEqual(j3.status, 'succeeded', JSON.stringify(j3.error));
++ assert.deepStrictEqual([j3.result.uploaded, j3.result.durable, j3.result.rows.updated], [5, 5, 5], JSON.stringify(j3.result));
++ assert.ok(purposes.includes('durable'), 'object.cmaf asks the placement router for its provider');
++ const durableRows = await timeline.list(pub);
++ assert.ok(durableRows.every((r) => r.durability === 'durable' && r.durable_provider === 'b2' && blobs.has(`b2:${r.key}`)), 'the router ranks the canonical tier first');
++ configured = ['r2'];
++ assert.strictEqual(await timeline.durableProvider(await model.getObject(pub)), 'r2', 'no B2: the router picks R2, nothing is hard-coded');
++ configured = [];
++ assert.strictEqual(await timeline.durableProvider(await model.getObject(pub)), null);
++ configured = ['b2', 'r2'];
++
++ // ── Planning: contiguous durable unpacked runs, ~target each, a short tail joins the chunk before ──
++ const pr = (seq, start, end, extra = {}) => ({ ...row(seq, start, end), durability: 'durable', durable_provider: 'b2', ...extra });
++ const planned = pack.plan([pr(0, 0, 0), ...[1, 2, 3, 4, 5, 6, 7].map((n) => pr(n, (n - 1) * 25000, n * 25000)), pr(8, 175000, 180000, { durability: 'local', durable_provider: null }),
++ pr(9, 180000, 190000), pr(10, 190000, 200000), pr(11, 200000, 210000, { packed_object_id: 'x' })], 60000);
++ assert.deepStrictEqual(planned.chunks.map((c) => c.map((r) => r.seq)), [[1, 2, 3], [4, 5, 6, 7], [9, 10]], 'a 25 s tail joins the chunk before; a local segment ends a run');
++ assert.deepStrictEqual([planned.alreadyPacked, planned.waiting], [1, 1]);
++
++ // ── The pack job's abort signal: nothing packed, the work directory removed ──
++ const acp = new AbortController();
++ acp.abort(new Error('pack cancelled in test'));
++ await assert.rejects(pack.spec.run({ id: 'mjob_PACKABORT', app_id: 'live', object_id: pub, params: { target_seconds: 10 } }, { signal: acp.signal }), /pack cancelled in test/);
++ assert.strictEqual(JSON.stringify(await timeline.list(pub)), JSON.stringify(durableRows));
++ assert.strictEqual(fs.existsSync(path.join(process.env.OBJECTS_PATH, '.jobs', 'mjob_PACKABORT')), false);
++
++ // ── Packing: one chunk, each row its byte range, the per-segment copies deleted ──
++ const jp = await runJob(pub, { target_seconds: 10 }, 'object.pack');
++ assert.strictEqual(jp.status, 'succeeded', JSON.stringify(jp.error));
++ assert.deepStrictEqual([jp.result.packs, jp.result.segments, jp.result.removed_segments, jp.result.left_behind, jp.result.provider], [1, 4, 4, 0, 'b2'], JSON.stringify(jp.result));
++ const packed = await timeline.list(pub);
++ assert.strictEqual(JSON.stringify(packed[0]), JSON.stringify(durableRows[0]), 'the init segment is not packed');
++ const chunkId = packed[1].packed_object_id;
++ const chunkKey = `${pub}/source/${chunkId.slice(0, 12)}/p000001.m4s`;
++ const chunk = blobs.get(`b2:${chunkKey}`);
++ assert.ok(chunk && sha(chunk) === chunkId, 'packed_object_id is the chunk sha256, and the durable chunk reads back as it');
++ assert.deepStrictEqual(fs.readFileSync(packed[1].local_path), chunk, 'this node keeps the same chunk');
++ let at0 = 0;
++ for (const [i, r] of packed.slice(1).entries()) {
++ const was = durableRows[i + 1];
++ assert.deepStrictEqual([r.name, r.start_ms, r.end_ms, r.sha256, r.byte_length], [was.name, was.start_ms, was.end_ms, was.sha256, was.byte_length], `${r.name} keeps its identity`);
++ assert.deepStrictEqual([r.packed_object_id, r.key, r.durable_provider, r.durability, Number(r.byte_offset)], [chunkId, chunkKey, 'b2', 'durable', at0], `${r.name} names its range`);
++ assert.strictEqual(sha(chunk.subarray(at0, at0 + Number(r.byte_length))), r.sha256, `${r.name}'s range is its bytes`);
++ assert.ok(!blobs.has(`b2:${was.key}`) && !fs.existsSync(was.local_path), `${r.name}'s own key and file are deleted`);
++ at0 += Number(r.byte_length);
++ }
++ assert.strictEqual(at0, chunk.length);
++ assert.strictEqual((await get(`/o/${pub}/source/index.m3u8`)).buf.toString(), media, 'the playlist is unchanged by packing');
++ for (const r of packed) {
++ const s = await get(`/o/${pub}/source/${r.name}`);
++ assert.deepStrictEqual([s.status, sha(s.buf), Number(s.headers.get('content-length'))], [200, r.sha256, Number(r.byte_length)], `${r.name} served from the chunk`);
++ }
++ const part = await get(`/o/${pub}/source/000002.m4s`, { range: 'bytes=10-99' });
++ assert.deepStrictEqual([part.status, part.headers.get('content-range')], [206, `bytes 10-99/${packed[2].byte_length}`]);
++ assert.deepStrictEqual(part.buf, chunk.subarray(Number(packed[2].byte_offset) + 10, Number(packed[2].byte_offset) + 100), 'a range inside a packed segment');
++ assert.strictEqual((await get(`/o/${pub}/source/000002.m4s`, { range: `bytes=${packed[2].byte_length}-` })).status, 416);
++ assert.ok(Math.abs(Number(await ffprobe()) - 7) < 0.2, 'the packed timeline still plays as the 7 s source');
++ // This node's chunk gone: the router's durable copy, read as a byte range.
++ fs.renameSync(packed[1].local_path, `${packed[1].local_path}.away`);
++ for (const r of packed.slice(1)) {
++ const s = await get(`/o/${pub}/source/${r.name}`);
++ assert.deepStrictEqual([s.status, sha(s.buf)], [200, r.sha256], `${r.name} served from the durable chunk`);
++ }
++ fs.renameSync(`${packed[1].local_path}.away`, packed[1].local_path);
++
++ // ── A rerun packs nothing; a cmaf rerun keeps the packed rows ──
++ const jp2 = await runJob(pub, { target_seconds: 10 }, 'object.pack');
++ assert.deepStrictEqual([jp2.status, jp2.result.packs, jp2.result.already_packed], ['succeeded', 0, 4], JSON.stringify(jp2.result));
++ assert.strictEqual(JSON.stringify(await timeline.list(pub)), JSON.stringify(packed), 'a rerun changes no row');
++ const j4 = await runJob(pub, { segment_seconds: 2 });
++ assert.deepStrictEqual([j4.status, j4.result.uploaded, j4.result.rows], ['succeeded', 0, { inserted: 0, updated: 0, unchanged: 5, removed: 0 }], JSON.stringify(j4.result));
++ assert.ok(blobs.has(`b2:${chunkKey}`) && fs.existsSync(packed[1].local_path), 'and its chunk');
++ router.route = realRoute;
++
// ── The job's abort signal: nothing written, the work directory removed ──
const cmaf = require('../server/jobs/cmaf');
const ac = new AbortController();
@@@ -201,6 +306,15 @@ const { spawn, spawnSync } = require('child_process');
const smedia = (await get(`/o/${priv}/source/index.m3u8${signed.search}`)).buf.toString();
assert.ok(smedia.includes(`#EXT-X-MAP:URI="init.mp4${signed.search}"`) && smedia.includes(`000001.m4s${signed.search}`), smedia);
assert.strictEqual((await get(`/o/${priv}/source/000001.m4s${signed.search}`)).status, 200);
++ assert.strictEqual(sm.headers.get('access-control-allow-origin'), null, 'no CORS on a private playlist, as on GET /o/:id');
++ const jpp = await runJob(priv, { target_seconds: 10 }, 'object.pack');
++ assert.deepStrictEqual([jpp.status, jpp.result.packs], ['succeeded', 1], JSON.stringify(jpp));
++ const privRows = await timeline.list(priv);
++ assert.ok(privRows[1].packed_object_id);
++ assert.strictEqual((await get(`/o/${priv}/source/000002.m4s`)).status, 404, 'a packed private segment needs the signature');
++ assert.strictEqual((await get(`/o/${priv}/source/000002.m4s?exp=${signed.searchParams.get('exp')}&sig=${'0'.repeat(64)}`)).status, 404);
++ const sseg = await get(`/o/${priv}/source/000002.m4s${signed.search}`);
++ assert.deepStrictEqual([sseg.status, sha(sseg.buf), sseg.headers.get('access-control-allow-origin')], [200, privRows[2].sha256, null], 'signed: the packed bytes, no CORS');
assert.strictEqual((await get(`/o/${priv}/master.m3u8?exp=${signed.searchParams.get('exp')}&sig=${'0'.repeat(64)}`)).status, 404, 'a forged signature');
assert.strictEqual((await get(`/o/${pub}/master.m3u8${signed.search}`)).status, 200, 'a public object needs no signature');
assert.strictEqual((await call('GET', `/api/v2/live/objects/${pub}/download?format=json`)).body.hls_url, `${config.publicUrl}/o/${pub}/master.m3u8`);
@@@ -211,6 +325,8 @@ const { spawn, spawnSync } = require('child_process');
assert.strictEqual((await get(`/o/${pub}/source/000001.m4s`)).status, 404);
const off = await call('POST', '/api/v2/live/jobs', { body: { type: 'object.cmaf', object_id: pub } });
assert.deepStrictEqual([off.status, off.body.code], [409, 'media.hls.disabled']);
++ const offPack = await call('POST', '/api/v2/live/jobs', { body: { type: 'object.pack', object_id: pub } });
++ assert.deepStrictEqual([offPack.status, offPack.body.code], [409, 'media.hls.disabled']);
assert.deepStrictEqual(Object.keys((await call('GET', `/api/v2/live/objects/${pub}/download?format=json`)).body).sort(), ['expires_at', 'public', 'url']);
config.hls.enabled = true;
@@@ -223,6 +339,8 @@ const { spawn, spawnSync } = require('child_process');
await model.purgeExpired({ retentionDays: 0 });
assert.strictEqual(await timeline.has(pub), false, 'a purge removes the timeline');
assert.ok(pubPaths.every((p) => !fs.existsSync(p)), 'and its segment files');
++ assert.ok(!fs.existsSync(packed[1].local_path) && !blobs.has(`b2:${chunkKey}`) && !blobs.has(`b2:${packed[0].key}`), 'and its chunk, here and durable');
++ assert.ok(blobs.has(`b2:${privRows[1].key}`), 'a held object keeps its durable chunk');
assert.strictEqual(await timeline.has(priv), true, 'a held object keeps its timeline');
assert.ok((await timeline.list(priv)).every((r) => fs.existsSync(r.local_path)));
await model.releaseHold(hold.id);