openvibers/openvibe.media Commits
Build sprite sheets from the timeline, not 100 source seeks (#34)
Changes
@@@ -39,7 +39,8 @@"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",-- "packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)"++ "packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)",++ "timeline sprites (F3.1, behind MEDIA_HLS_ENABLED): object.sprite cuts its sheet from the object's media_timeline rows — each sample time picks the row covering it and one frame is decoded from that row's bytes (this node's segment, or a packed row's byte range in its chunk), checked against the row's sha256 — with the same layout and metadata as the seek path, so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame"],"updated": "2026-10-04"}@@@ -149,9 +149,16 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux(`dedupeActive`) and answers `202 { job_id }` while it is missing.- **F3.5, shipped: virtual and materialized clips over the timeline (§4).**++- **F3.1 sprites, shipped: a sprite sheet from the timeline.** `object.sprite` (with `MEDIA_HLS_ENABLED`) cuts its++ frames from the object's `media_timeline` rows instead of seeking the source: each sample time picks the row covering++ it (`timeline.segmentAt`) and one frame is decoded from that row's bytes — this node's segment file, or a packed row's++ byte range in its chunk, with the init segment prepended — checked against the row's sha256. A packed long VOD is++ never decoded end to end. The layout (`sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`)++ and metadata are the seek path's, so the player contract is unchanged. A source without a timeline, or with++ `MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability forlive segments and its upload-lag metric, a materialized clip that references the source's chunks (chunk reference-- counts, boundary-only re-encode), sprites and captions on it, segment-bucket demand, and Live's player++ counts, boundary-only re-encode), captions on the timeline, segment-bucket demand, and Live's playermoving to HLS. Renditions are F4.## 4. Clips reuse the source@@@ -341,8 +348,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr- **F3 segment-native video:** the timeline index and the CMAF/HLS source representation of finished video (F3.1,shipped; §3), packing into ~60 s chunks (F3.3, shipped; §3), the timeline queued at finalize (and lazily ondownload), signed playlists outliving a download URL and the faststart MP4 fallback (F3.c, shipped; §3), virtual and-- materialized clips (F3.5, shipped; §4); still open: live DVR from OpenRe (CMAF/HLS recording with the growing-- playlist), materialized clips referencing the source's chunks, sprites and captions on the++ materialized clips (F3.5, shipped; §4) and sprite sheets cut from the timeline's rows (F3.1, shipped; §3); still++ open: live DVR from OpenRe (CMAF/HLS recording with the growing++ playlist), materialized clips referencing the source's chunks, captions on thetimeline, segment-bucket demand; Live's player moves to HLS.- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viralVODs, compute placement.@@@ -739,7 +739,7 @@ job over only with the token it read and, past boot, only while the lease is sti| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant || `clip.cut` | clips | Cut (or re-cut) a clip row from its VOD: `POST /api/v1/:app/clips` and `/:id/recut` answer 202 with `job_id`, which a UI polls and reattaches to (`GET /api/v2/:app/jobs/:id`). `clip.ready`/`clip.failed` and the webhook come from the cut as before. It retries after 2, 10 and 30 min (from attempt 2 a cloud VOD is fetched home first); a window with no footage fails at once. The `clips` lane (`MEDIA_JOBS_CLIPS_CONCURRENCY`, 2) runs while recording || `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent |--| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG, made from one fast seek per frame: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent |++| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). With `MEDIA_HLS_ENABLED` and a source timeline it is cut from the timeline's rows — each sample time's row decoded from this node's segment or the packed chunk's byte range, checked against the row's sha256 — so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame over the source. It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent || `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) |@@@ -12,7 +12,10 @@* result { source_id, object_id, size_bytes, width, height, method: 'waveform' }* object.sprite params { frames?, columns?, tile_width? } a JPEG sheet of evenly spaced frames for* seek previews: 100 frames by default (at most one per 2 s), 10 columns, 160 px wide-- * tiles. Each frame is its own fast seek, so a long remote VOD is never decoded end to end.++ * tiles. With MEDIA_HLS_ENABLED and a source timeline the sheet is cut from the timeline's++ * own segments: each sample time picks its row and one frame is decoded from that row's++ * bytes (this node's file, or the packed chunk's byte range), so a packed long VOD is never++ * decoded end to end. A source without a timeline falls back to one fast seek per frame.* The layout is in the object's metadata (sprite: { interval_seconds, count, columns,* rows, tile_width, tile_height }): frame i covers [i × interval, (i + 1) × interval) and* sits at column i % columns, row ⌊i / columns⌋.@@@ -25,6 +28,9 @@'use strict';const fs = require('fs');const path = require('path');++const crypto = require('crypto');++const config = require('../config');++const timeline = require('../objects/timeline');const { JobError } = require('./queue');const d = require('./derive');@@@ -103,10 +109,35 @@ function spriteLayout(duration, { frames, columns, tile_width }) {return { count, interval_seconds: Math.round(interval * 1000) / 1000, columns: cols, rows: Math.ceil(count / cols), tile_width, tile_height: Math.round(tile_width * 9 / 16 / 2) * 2 };}--async function runSprite(job, ctx) {-- const src = await d.loadSource(job);-- const p = validateSprite({ obj: src, params: job.params });-- if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }++const tileFilter = (layout) => `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;++++/** A frame that cannot be read (a gap in a recording) is black rather than failing the sheet. */++async function blackTile(file, layout, ctx) {++ await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', file], { signal: ctx.signal, timeoutMs: 60000 });++}++++/** Assemble the grid from the per-frame tiles, adopt it as the source's sprite variant, return the job result. */++async function finishSprite({ src, job, ctx, dir, layout, duration }) {++ const out = path.join(dir, 'sprite.jpg');++ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),++ '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });++ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {++ d.cleanupWork(job.id);++ throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);++ }++ const made = await d.adopt({++ src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',++ relation: { method: 'sprite', ...layout },++ metadata: { sprite: layout, source_duration_seconds: duration },++ saveCheckpoint: ctx.saveCheckpoint,++ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),++ });++ d.cleanupWork(job.id);++ return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };++}++++/** The old path: one fast seek per frame over the source itself. A source with no timeline uses it. */++async function runSeekSprite(job, ctx, src, p) {const source = await d.resolveSource(src);if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });await d.checkRoom(src, 16 * MB);@@@ -117,7 +148,7 @@ async function runSprite(job, ctx) {if (meta.video === false) throw new JobError('no_video', 'The source has no video track', { permanent: true });const layout = spriteLayout(duration, p);const dir = d.workDir(job.id);-- const tile = `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;++ const tile = tileFilter(layout);for (let i = 0; i < layout.count; i++) {if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);@@@ -125,31 +156,179 @@ async function runSprite(job, ctx) {const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-ss', String(Math.max(0, at).toFixed(3)), '-i', source.input,'-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });-- // A frame that cannot be read (a gap in a recording) is black rather than failing the sheet.-- if (!r.ok || !d.existing(frame)) {-- await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', frame], { signal: ctx.signal, timeoutMs: 60000 });++ if (!r.ok || !d.existing(frame)) await blackTile(frame, layout, ctx);++ }++ return finishSprite({ src, job, ctx, dir, layout, duration });++}++++/** `length` bytes at `offset` of a local file, or null (short read, gone). */++function readRange(file, offset, length) {++ let fd = null;++ try {++ fd = fs.openSync(file, 'r');++ const b = Buffer.allocUnsafe(length);++ let read = 0;++ while (read < length) {++ const n = fs.readSync(fd, b, read, length - read, offset + read);++ if (!n) break;++ read += n;}++ return read === length ? b : null;++ } catch { return null; } finally { if (fd != null) { try { fs.closeSync(fd); } catch { /* gone */ } } }++}++++/**++ * A row's bytes, checked against the row's sha256: this node's file (a segment, or the packed chunk's byte range++ * [byte_offset, byte_offset + byte_length)), else a ranged read of the durable copy through the placement router.++ * null = no verified copy (the frame falls back to black).++ */++async function rowBuffer(row, ctx) {++ const length = Number(row.byte_length);++ if (row.local_path && d.existing(row.local_path)) {++ const b = row.packed_object_id ? readRange(row.local_path, Number(row.byte_offset), length) : fs.readFileSync(row.local_path);++ if (b && b.length === length && crypto.createHash('sha256').update(b).digest('hex') === row.sha256) return b;}-- const out = path.join(dir, 'sprite.jpg');-- const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),-- '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });-- if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {-- d.cleanupWork(job.id);-- throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);++ if (row.durable_provider && row.durability === 'durable' && row.key) {++ const vodStorage = require('../vod/vod-storage');++ const url = await vodStorage.presignGet(row.durable_provider, row.key, 900).catch(() => null);++ if (!url) return null;++ const packed = !!row.packed_object_id;++ const headers = packed ? { range: `bytes=${Number(row.byte_offset)}-${Number(row.byte_offset) + length - 1}` } : {};++ const res = await fetch(url, { headers, signal: ctx.signal }).catch(() => null);++ if (!res) return null;++ // A packed row is a byte range of the chunk: a server that ignores Range answers 200 with the whole++ // (many MB) chunk. Refuse that without reading a byte, and never read past the length we asked for.++ const refuse = () => {++ try { const p = res.body && res.body.cancel ? res.body.cancel() : null; if (p && typeof p.catch === 'function') p.catch(() => { /* gone */ }); } catch { /* gone */ }++ return null;++ };++ if (packed) {++ if (res.status !== 206) return refuse();++ if (Number(res.headers.get('content-length')) !== length) return refuse();++ } else if (!res.ok) return refuse();++ const b = Buffer.from(await res.arrayBuffer());++ if (b.length === length && crypto.createHash('sha256').update(b).digest('hex') === row.sha256) return b;}-- const made = await d.adopt({-- src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',-- relation: { method: 'sprite', ...layout },-- metadata: { sprite: layout, source_duration_seconds: duration },-- saveCheckpoint: ctx.saveCheckpoint,-- checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),-- });-- d.cleanupWork(job.id);-- return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };++ return null;++}++++/**++ * The layout duration for a timeline sprite: the object's own duration (the seek path's metadata first, then its++ * probe), never the timeline end unless neither is known. A truncated or in-progress cut must not shrink the sheet++ * or its seek metadata; sample times past the timeline end already fall back to black tiles.++ */++async function spriteDuration(src, segs) {++ const md = require('../objects/model').parseJson(src.metadata, {});++ const known = Number(md.duration_seconds) || 0;++ if (known > 0) return known;++ const source = await d.resolveSource(src).catch(() => null);++ if (source) {++ const meta = await streamsOf(source.input).catch(() => ({ duration: 0 }));++ if (Number(meta.duration) > 0) return Number(meta.duration);++ }++ return Number(segs[segs.length - 1].end_ms) / 1000;++}++++/** Remove the sampled pieces of a failed run; the frame checkpoints (f*.jpg) stay, so a retry resumes. */++function sweepPieces(dir) {++ let names = [];++ try { names = fs.readdirSync(dir); } catch { return; }++ for (const n of names) if (/^seg-\d+\.mp4$/.test(n)) { try { fs.unlinkSync(path.join(dir, n)); } catch { /* gone */ } }++}++++/**++ * The sheet from the timeline (F3.1): each sample time picks the row covering it (segmentAt) and one frame is++ * decoded from that row's bytes — init.mp4 + the segment as one small file, never the source, never a full-file++ * decode. Layout, metadata and the player contract are the seek path's.++ */++async function runTimelineSprite(job, ctx, src, p, rows, segs) {++ const initRow = rows.find((r) => Number(r.seq) === 0) || null;++ // Pieces are deleted as the samples move past them, so only one or two are on disk at once.++ const initLen = initRow ? Number(initRow.byte_length) || 0 : 0;++ const maxSeg = segs.reduce((m, r) => Math.max(m, Number(r.byte_length) || 0), 0);++ await d.checkRoom(src, 2 * (maxSeg + initLen) + 4 * MB);++ const duration = await spriteDuration(src, segs);++ if (!(duration > 0)) throw new JobError('no_duration', 'The timeline has no duration to spread frames over', { permanent: true });++ const layout = spriteLayout(duration, p);++ const dir = d.workDir(job.id);++ // The init segment is read once and prepended to every piece; a media segment alone is not playable.++ const initBody = initRow ? await rowBuffer(initRow, ctx) : null;++ const pieceFile = (seq) => path.join(dir, `seg-${String(seq).padStart(6, '0')}.mp4`);++ const dropPiece = (seq) => { if (seq != null) { try { fs.unlinkSync(pieceFile(seq)); } catch { /* already gone */ } } };++ sweepPieces(dir); // pieces a previous attempt left behind are rebuilt from the rows as needed++ const cache = new Map();++ const piece = async (row) => {++ const seq = Number(row.seq);++ if (cache.has(seq)) return cache.get(seq);++ const body = await rowBuffer(row, ctx);++ let file = null;++ if (body) {++ file = pieceFile(seq);++ fs.writeFileSync(file, initBody && seq !== 0 ? Buffer.concat([initBody, body]) : body);++ }++ cache.set(seq, file);++ return file;++ };++ const tile = tileFilter(layout);++ let decoded = 0;++ let lastSeq = null;++ try {++ // An audio-only source has a timeline but no frame to cut: permanent, like the seek path's `no_video`.++ const first = await piece(segs[0]);++ if (first) {++ const info = await require('../vod/media-tools').probeVodInfo(first);++ if (info.streams.length && !info.streams.some((s) => s.codec_type === 'video')) {++ d.cleanupWork(job.id);++ throw new JobError('no_video', 'The source has no video track', { permanent: true });++ }++ }++ for (let i = 0; i < layout.count; i++) {++ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');++ const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);++ if (d.existing(frame)) { decoded++; continue; }++ const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);++ const row = await timeline.segmentAt(src.id, timeline.SOURCE, Math.max(0, at) * 1000);++ let ok = false;++ if (row) {++ const seq = Number(row.seq);++ // Samples are in time order: once the row advances, the previous piece has taken its last sample.++ if (lastSeq != null && seq > lastSeq) dropPiece(lastSeq);++ lastSeq = seq;++ const file = await piece(row);++ if (file) {++ // Seek is relative to the segment's own start (ffmpeg shifts the fMP4 segment's tfdt to 0 with the init).++ const rel = Math.max(0, at - Number(row.start_ms) / 1000);++ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...(rel > 0 ? ['-ss', rel.toFixed(3)] : []), '-i', file,++ '-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });++ ok = r.ok && d.existing(frame);++ }++ }++ if (ok) decoded++;++ else await blackTile(frame, layout, ctx);++ }++ if (!decoded) throw new JobError('segment_unreadable', 'No frame could be read from the timeline segments; a retry reads the durable copies', { retryAfterS: 1800 });++ return await finishSprite({ src, job, ctx, dir, layout, duration });++ } finally {++ // However the run ends (abort, ffmpeg, no_video), the pieces go: hundreds of MB must not be left behind.++ // The frame checkpoints stay for a retry.++ dropPiece(lastSeq);++ sweepPieces(dir);++ }++}++++async function runSprite(job, ctx) {++ const src = await d.loadSource(job);++ const p = validateSprite({ obj: src, params: job.params });++ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }++ // Sprites ride the timeline when there is one (MEDIA_HLS_ENABLED); a source without one keeps the seek path.++ const rows = config.hls.enabled ? await timeline.list(src.id) : [];++ const segs = rows.filter((r) => Number(r.seq) > 0);++ return segs.length ? runTimelineSprite(job, ctx, src, p, rows, segs) : runSeekSprite(job, ctx, src, p);}module.exports = {waveform: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateWaveform, run: runWaveform },sprite: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateSprite, run: runSprite },spriteLayout,++ rowBuffer, // tested directly: the durable ranged-GET guards};@@@ -51,6 +51,7 @@ const { spawn, spawnSync } = require('child_process');const sleep = (ms) => new Promise((r) => setTimeout(r, ms));const waitFor = async (fn, ms = 60000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(25); } return false; };const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');++ const size = (file) => { const r = spawnSync('ffprobe', ['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height', '-of', 'csv=p=0', file], { encoding: 'utf8' }); return r.stdout.trim(); };const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;(async () => {@@@ -153,6 +154,18 @@ const { spawn, spawnSync } = require('child_process');assert.strictEqual(JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]), before, 'the source is untouched');assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');++ // ── A truncated timeline (an in-progress cut) must not shrink the sheet: the object's own 7 s duration++ // lays it out, exactly as the seek path would, and the samples past the rows become black tiles ──++ const shorty = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility: 'private', lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: 1, metadata: { duration_seconds: 7 } });++ await timeline.replace(shorty, 'source', rows.slice(0, 2)); // init + one 3 s segment: the timeline ends at 3 s++ const st = await runJob(shorty, {}, 'object.sprite');++ assert.strictEqual(st.status, 'succeeded', JSON.stringify(st.error));++ assert.deepStrictEqual(st.result.sprite, require('../server/jobs/previews').spriteLayout(7, { frames: 100, columns: 10, tile_width: 160 }), 'the object duration, not the 3 s timeline end, lays the sheet out');++ assert.deepStrictEqual([st.result.sprite.count, st.result.sprite.interval_seconds], [3, 2.333], 'same interval/count as the seek path over the 7 s source');++ const sto = await model.getObject(st.result.object_id);++ assert.strictEqual(model.parseJson(sto.metadata, {}).source_duration_seconds, 7, 'the seek metadata carries the real duration');++ assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');++// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;const autoPack = await packJobs(pub);@@@ -303,6 +316,104 @@ const { spawn, spawnSync } = require('child_process');}fs.renameSync(`${packed[1].local_path}.away`, packed[1].local_path);++ // ── Sprites come from the timeline (F3.1): each frame from the row covering its instant, decoded from the++ // packed chunk's byte range, never the source. The seek layout and the player contract are unchanged ──++ const { spriteLayout } = require('../server/jobs/previews');++ const brightness = (file) => {++ const r = spawnSync('ffmpeg', ['-v', 'error', '-i', file, '-frames:v', '1', '-vf', 'scale=1:1,format=gray', '-f', 'rawvideo', '-'], { encoding: null, maxBuffer: 1 << 20 });++ return r.stdout && r.stdout.length ? r.stdout[r.stdout.length - 1] : -1;++ };++ const sr = await runJob(pub, {}, 'object.sprite');++ assert.strictEqual(sr.status, 'succeeded', JSON.stringify(sr.error));++ assert.deepStrictEqual(sr.result.sprite, spriteLayout(7, { frames: 100, columns: 10, tile_width: 160 }), 'the timeline sheet keeps the seek layout');++ assert.deepStrictEqual([sr.result.sprite.count, sr.result.sprite.interval_seconds, sr.result.sprite.columns], [3, 2.333, 3]);++ const so = await model.getObject(sr.result.object_id);++ assert.deepStrictEqual([so.kind, so.mime_type, so.visibility], ['asset', 'image/jpeg', 'private']);++ assert.strictEqual(size((await model.listLocations(so.id))[0].key), '480,90', 'the image is the layout: 3 × 160 by 1 × 90');++ assert.deepStrictEqual(model.parseJson(so.metadata, {}).sprite, sr.result.sprite, 'the layout is in the metadata');++ assert.strictEqual((await model.getVariant(pub, 'sprite')).derived_object_id, so.id);++ assert.ok(brightness((await model.listLocations(so.id))[0].key) > 30, 'real frames from the rows, not black tiles');++++ // The source file gone: the seek path could not run at all, yet the rows' bytes cut the same sheet.++ const pubFile = path.join(process.env.VOD_PATH, 'public.mp4');++ fs.renameSync(pubFile, `${pubFile}.away`);++ const sr2 = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');++ assert.strictEqual(sr2.status, 'succeeded', JSON.stringify(sr2.error));++ assert.deepStrictEqual([sr2.result.sprite.count, sr2.result.sprite.rows, sr2.result.sprite.columns], [3, 2, 2], 'the same layout from rows, with no source to seek');++ assert.strictEqual(size((await model.listLocations(sr2.result.object_id))[0].key), '192,108');++ fs.renameSync(`${pubFile}.away`, pubFile);++++ // ── The durable fallback and its guards. A packed row is a byte range of the chunk: with this node's++ // chunk gone the frame must come from a 206 ranged GET of the durable copy. A 200 (Range ignored) or a++ // wrong content-length is refused before the body is read; a sha mismatch is a black tile, not a failure ──++ const previews = require('../server/jobs/previews');++ assert.strictEqual(typeof previews.rowBuffer, 'function');++ const durableRow = { durability: 'durable', durable_provider: 'b2', key: 'k', packed_object_id: 'chunk', byte_offset: 10, byte_length: 4, sha256: sha(Buffer.from('abcd')) };++ const realFetch = global.fetch;++ let cancelled = 0;++ let read = 0;++ global.fetch = async () => ({ ok: true, status: 200, headers: { get: () => null }, body: { cancel: async () => { cancelled++; } }, arrayBuffer: async () => { read++; return Buffer.from('abcd'); } });++ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a 200 answer to a packed ranged GET is refused');++ assert.deepStrictEqual([cancelled, read], [1, 0], 'refused without buffering the whole chunk');++ cancelled = 0;++ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '999' }, body: { cancel: async () => { cancelled++; } }, arrayBuffer: async () => { read++; return Buffer.alloc(999); } });++ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a 206 of the wrong length is refused');++ assert.deepStrictEqual([cancelled, read], [1, 0], 'refused before this one was read either');++ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '4' }, body: null, arrayBuffer: async () => Buffer.from('0000') });++ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a sha mismatch is a black tile, not a failure');++ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '4' }, body: null, arrayBuffer: async () => Buffer.from('abcd') });++ assert.deepStrictEqual(await previews.rowBuffer(durableRow, {}), Buffer.from('abcd'), 'the verified range');++ global.fetch = realFetch;++++ const tileBrightness = (file, i, layout) => {++ const x = (i % layout.columns) * layout.tile_width;++ const y = Math.floor(i / layout.columns) * layout.tile_height;++ const r = spawnSync('ffmpeg', ['-v', 'error', '-i', file, '-frames:v', '1', '-vf', `crop=${layout.tile_width}:${layout.tile_height}:${x}:${y},scale=1:1,format=gray`, '-f', 'rawvideo', '-'], { encoding: null, maxBuffer: 1 << 20 });++ return r.stdout && r.stdout.length ? r.stdout[r.stdout.length - 1] : -1;++ };++ fs.renameSync(packed[1].local_path, `${packed[1].local_path}.sprite-away`);++ const sd = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');++ assert.strictEqual(sd.status, 'succeeded', JSON.stringify(sd.error));++ assert.deepStrictEqual([sd.result.sprite.count, sd.result.sprite.columns, sd.result.sprite.rows], [3, 2, 2], 'the durable ranges cut the same sheet');++ const sdFile = (await model.listLocations(sd.result.object_id))[0].key;++ assert.ok([0, 1, 2].every((i) => tileBrightness(sdFile, i, sd.result.sprite) > 30), 'real frames from the durable chunk');++ // One durable range corrupted: its tile is black, the others still decode.++ const cleanChunk = blobs.get(`b2:${chunkKey}`);++ const corruptChunk = Buffer.from(cleanChunk);++ corruptChunk.fill(0, Number(packed[1].byte_offset), Number(packed[1].byte_offset) + Number(packed[1].byte_length));++ blobs.set(`b2:${chunkKey}`, corruptChunk);++ const sc = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');++ assert.strictEqual(sc.status, 'succeeded', JSON.stringify(sc.error));++ const scFile = (await model.listLocations(sc.result.object_id))[0].key;++ assert.ok(tileBrightness(scFile, 0, sc.result.sprite) < 20, 'the sha-mismatched range is a black tile');++ assert.ok(tileBrightness(scFile, 1, sc.result.sprite) > 30, 'the intact ranges still cut real frames');++ blobs.set(`b2:${chunkKey}`, cleanChunk);++ fs.renameSync(`${packed[1].local_path}.sprite-away`, packed[1].local_path);++++ // ── A failed run leaves no seg-*.mp4 behind, only the resumable frame checkpoints ──++ {++ const derive = require('../server/jobs/derive');++ const acs = new AbortController();++ acs.abort(new Error('sprite cancelled in test'));++ const wdir = derive.workDir('mjob_SPRITEABORT');++ fs.writeFileSync(path.join(wdir, 'f0000.jpg'), 'checkpoint');++ await assert.rejects(previews.sprite.run({ id: 'mjob_SPRITEABORT', app_id: 'live', object_id: pub, params: {} }, { signal: acs.signal }), /sprite cancelled in test/);++ assert.deepStrictEqual(fs.readdirSync(wdir).filter((n) => /^seg-.*\.mp4$/.test(n)), [], 'no sampled segment is left behind');++ assert.ok(fs.existsSync(path.join(wdir, 'f0000.jpg')), 'the frame checkpoint survives for a retry');++ derive.cleanupWork('mjob_SPRITEABORT');++ }++++ // A source with no timeline uses the old seek path (and writes no rows).++ const plainFile = path.join(process.env.VOD_PATH, 'plain.mp4');++ fs.copyFileSync(src, plainFile);++ const plain = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility: 'private', lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: fs.statSync(plainFile).size });++ await model.upsertLocation(plain, { provider: 'local', key: plainFile, state: 'present', size_bytes: fs.statSync(plainFile).size, verified: true });++ assert.strictEqual(await timeline.has(plain), false);++ const sp = await runJob(plain, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');++ assert.strictEqual(sp.status, 'succeeded', JSON.stringify(sp.error));++ assert.deepStrictEqual([sp.result.sprite.count, sp.result.sprite.columns, sp.result.sprite.rows], [3, 2, 2], 'the seek path lays out the sheet the same way');++ assert.strictEqual(await timeline.has(plain), false, 'and writes no timeline rows');++// ── 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));@@@ -36,6 +36,7 @@ const https = require('https');'server/public/dev-data.js': 'the app\'s internal URL (APP_INTERNAL_URLS), ids and encoded names in the path','server/jobs/pack.js': 'a presigned GET on B2/R2 (operator-configured buckets) for a segment\'s durable copy','server/objects/routes.js': 'a ranged GET on a presigned B2/R2 URL (operator-configured buckets) for a packed segment\'s chunk',++ 'server/jobs/previews.js': 'a (ranged) GET on a presigned B2/R2 URL (operator-configured buckets) for a timeline segment, to cut a sprite frame','server/drill.js': 'restore drill: only this server itself','server/me/client.js': 'browser script, same-origin',};
@@ -39,7 +39,8 @@
@@ -39,7 +39,8 @@
"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
"every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
"every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
"segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
"segment-native video (F3.1, behind MEDIA_HLS_ENABLED): the media_timeline index, the object.cmaf job (CMAF/fMP4 segments of a finished video on its own keyframes, sha256 and durability per row, idempotent) and HLS playlists written from the timeline at /o/:id/master.m3u8 under the object's access checks",
- "packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)"
+ "packing after finalize (F3.3, behind MEDIA_HLS_ENABLED): the object.pack job concatenates durable segments into ~60 s chunk objects verified by sha256, each media_timeline row naming its byte range; the segment route serves a packed segment as a ranged read of its chunk; the per-segment keys are deleted once the chunk is durable; segment providers come from the placement router; a finished object.cmaf queues object.pack by itself (dedupeActive: a rerun joins an active pack, a later cut queues a new one)",
+ "timeline sprites (F3.1, behind MEDIA_HLS_ENABLED): object.sprite cuts its sheet from the object's media_timeline rows — each sample time picks the row covering it and one frame is decoded from that row's bytes (this node's segment, or a packed row's byte range in its chunk), checked against the row's sha256 — with the same layout and metadata as the seek path, so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame"
],
],
"updated": "2026-10-04"
"updated": "2026-10-04"
}
}
@@ -149,9 +149,16 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -149,9 +149,16 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux
`-movflags +faststart` for an MP4 source; the container stays the source's) as a signed URL, or queues the remux
(`dedupeActive`) and answers `202 { job_id }` while it is missing.
(`dedupeActive`) and answers `202 { job_id }` while it is missing.
- **F3.5, shipped: virtual and materialized clips over the timeline (§4).**
- **F3.5, shipped: virtual and materialized clips over the timeline (§4).**
+- **F3.1 sprites, shipped: a sprite sheet from the timeline.** `object.sprite` (with `MEDIA_HLS_ENABLED`) cuts its
+ frames from the object's `media_timeline` rows instead of seeking the source: each sample time picks the row covering
+ it (`timeline.segmentAt`) and one frame is decoded from that row's bytes — this node's segment file, or a packed row's
+ byte range in its chunk, with the init segment prepended — checked against the row's sha256. A packed long VOD is
+ never decoded end to end. The layout (`sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`)
+ and metadata are the seek path's, so the player contract is unchanged. A source without a timeline, or with
+ `MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.
**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for
**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, a materialized clip that references the source's chunks (chunk reference
live segments and its upload-lag metric, a materialized clip that references the source's chunks (chunk reference
- counts, boundary-only re-encode), sprites and captions on it, segment-bucket demand, and Live's player
+ counts, boundary-only re-encode), captions on the timeline, segment-bucket demand, and Live's player
moving to HLS. Renditions are F4.
moving to HLS. Renditions are F4.
## 4. Clips reuse the source
## 4. Clips reuse the source
@@ -341,8 +348,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
@@ -341,8 +348,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
- **F3 segment-native video:** the timeline index and the CMAF/HLS source representation of finished video (F3.1,
- **F3 segment-native video:** the timeline index and the CMAF/HLS source representation of finished video (F3.1,
shipped; §3), packing into ~60 s chunks (F3.3, shipped; §3), the timeline queued at finalize (and lazily on
shipped; §3), packing into ~60 s chunks (F3.3, shipped; §3), the timeline queued at finalize (and lazily on
download), signed playlists outliving a download URL and the faststart MP4 fallback (F3.c, shipped; §3), virtual and
download), signed playlists outliving a download URL and the faststart MP4 fallback (F3.c, shipped; §3), virtual and
- materialized clips (F3.5, shipped; §4); still open: live DVR from OpenRe (CMAF/HLS recording with the growing
- playlist), materialized clips referencing the source's chunks, sprites and captions on the
+ materialized clips (F3.5, shipped; §4) and sprite sheets cut from the timeline's rows (F3.1, shipped; §3); still
+ open: live DVR from OpenRe (CMAF/HLS recording with the growing
+ playlist), materialized clips referencing the source's chunks, captions on the
timeline, segment-bucket demand; Live's player moves to HLS.
timeline, segment-bucket demand; Live's player moves to HLS.
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
VODs, compute placement.
VODs, compute placement.
@@ -739,7 +739,7 @@ job over only with the token it read and, past boot, only while the lease is sti
@@ -739,7 +739,7 @@ job over only with the token it read and, past boot, only while the lease is sti
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
| `clip.cut` | clips | Cut (or re-cut) a clip row from its VOD: `POST /api/v1/:app/clips` and `/:id/recut` answer 202 with `job_id`, which a UI polls and reattaches to (`GET /api/v2/:app/jobs/:id`). `clip.ready`/`clip.failed` and the webhook come from the cut as before. It retries after 2, 10 and 30 min (from attempt 2 a cloud VOD is fetched home first); a window with no footage fails at once. The `clips` lane (`MEDIA_JOBS_CLIPS_CONCURRENCY`, 2) runs while recording |
| `clip.cut` | clips | Cut (or re-cut) a clip row from its VOD: `POST /api/v1/:app/clips` and `/:id/recut` answer 202 with `job_id`, which a UI polls and reattaches to (`GET /api/v2/:app/jobs/:id`). `clip.ready`/`clip.failed` and the webhook come from the cut as before. It retries after 2, 10 and 30 min (from attempt 2 a cloud VOD is fetched home first); a window with no footage fails at once. The `clips` lane (`MEDIA_JOBS_CLIPS_CONCURRENCY`, 2) runs while recording |
| `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent |
| `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent |
-| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG, made from one fast seek per frame: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent |
+| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). With `MEDIA_HLS_ENABLED` and a source timeline it is cut from the timeline's rows — each sample time's row decoded from this node's segment or the packed chunk's byte range, checked against the row's sha256 — so a packed long VOD is never decoded end to end; a source without a timeline keeps one fast seek per frame over the source. It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent |
| `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) |
| `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) |
| `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) |
| `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) |
@@ -12,7 +12,10 @@
@@ -12,7 +12,10 @@
* result { source_id, object_id, size_bytes, width, height, method: 'waveform' }
* result { source_id, object_id, size_bytes, width, height, method: 'waveform' }
* object.sprite params { frames?, columns?, tile_width? } a JPEG sheet of evenly spaced frames for
* object.sprite params { frames?, columns?, tile_width? } a JPEG sheet of evenly spaced frames for
* seek previews: 100 frames by default (at most one per 2 s), 10 columns, 160 px wide
* seek previews: 100 frames by default (at most one per 2 s), 10 columns, 160 px wide
- * tiles. Each frame is its own fast seek, so a long remote VOD is never decoded end to end.
+ * tiles. With MEDIA_HLS_ENABLED and a source timeline the sheet is cut from the timeline's
+ * own segments: each sample time picks its row and one frame is decoded from that row's
+ * bytes (this node's file, or the packed chunk's byte range), so a packed long VOD is never
+ * decoded end to end. A source without a timeline falls back to one fast seek per frame.
* The layout is in the object's metadata (sprite: { interval_seconds, count, columns,
* The layout is in the object's metadata (sprite: { interval_seconds, count, columns,
* rows, tile_width, tile_height }): frame i covers [i × interval, (i + 1) × interval) and
* rows, tile_width, tile_height }): frame i covers [i × interval, (i + 1) × interval) and
* sits at column i % columns, row ⌊i / columns⌋.
* sits at column i % columns, row ⌊i / columns⌋.
@@ -25,6 +28,9 @@
@@ -25,6 +28,9 @@
'use strict';
'use strict';
const fs = require('fs');
const fs = require('fs');
const path = require('path');
const path = require('path');
+const crypto = require('crypto');
+const config = require('../config');
+const timeline = require('../objects/timeline');
const { JobError } = require('./queue');
const { JobError } = require('./queue');
const d = require('./derive');
const d = require('./derive');
@@ -103,10 +109,35 @@ function spriteLayout(duration, { frames, columns, tile_width }) {
@@ -103,10 +109,35 @@ function spriteLayout(duration, { frames, columns, tile_width }) {
return { count, interval_seconds: Math.round(interval * 1000) / 1000, columns: cols, rows: Math.ceil(count / cols), tile_width, tile_height: Math.round(tile_width * 9 / 16 / 2) * 2 };
return { count, interval_seconds: Math.round(interval * 1000) / 1000, columns: cols, rows: Math.ceil(count / cols), tile_width, tile_height: Math.round(tile_width * 9 / 16 / 2) * 2 };
}
}
-async function runSprite(job, ctx) {
- const src = await d.loadSource(job);
- const p = validateSprite({ obj: src, params: job.params });
- if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }
+const tileFilter = (layout) => `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;
+
+/** A frame that cannot be read (a gap in a recording) is black rather than failing the sheet. */
+async function blackTile(file, layout, ctx) {
+ await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', file], { signal: ctx.signal, timeoutMs: 60000 });
+}
+
+/** Assemble the grid from the per-frame tiles, adopt it as the source's sprite variant, return the job result. */
+async function finishSprite({ src, job, ctx, dir, layout, duration }) {
+ const out = path.join(dir, 'sprite.jpg');
+ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),
+ '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });
+ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {
+ d.cleanupWork(job.id);
+ throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);
+ }
+ const made = await d.adopt({
+ src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',
+ relation: { method: 'sprite', ...layout },
+ metadata: { sprite: layout, source_duration_seconds: duration },
+ saveCheckpoint: ctx.saveCheckpoint,
+ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),
+ });
+ d.cleanupWork(job.id);
+ return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };
+}
+
+/** The old path: one fast seek per frame over the source itself. A source with no timeline uses it. */
+async function runSeekSprite(job, ctx, src, p) {
const source = await d.resolveSource(src);
const source = await d.resolveSource(src);
if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
await d.checkRoom(src, 16 * MB);
await d.checkRoom(src, 16 * MB);
@@ -117,7 +148,7 @@ async function runSprite(job, ctx) {
@@ -117,7 +148,7 @@ async function runSprite(job, ctx) {
if (meta.video === false) throw new JobError('no_video', 'The source has no video track', { permanent: true });
if (meta.video === false) throw new JobError('no_video', 'The source has no video track', { permanent: true });
const layout = spriteLayout(duration, p);
const layout = spriteLayout(duration, p);
const dir = d.workDir(job.id);
const dir = d.workDir(job.id);
- const tile = `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;
+ const tile = tileFilter(layout);
for (let i = 0; i < layout.count; i++) {
for (let i = 0; i < layout.count; i++) {
if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');
if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');
const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);
const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);
@@ -125,31 +156,179 @@ async function runSprite(job, ctx) {
@@ -125,31 +156,179 @@ async function runSprite(job, ctx) {
const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);
const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);
const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-ss', String(Math.max(0, at).toFixed(3)), '-i', source.input,
const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-ss', String(Math.max(0, at).toFixed(3)), '-i', source.input,
'-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });
'-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });
- // A frame that cannot be read (a gap in a recording) is black rather than failing the sheet.
- if (!r.ok || !d.existing(frame)) {
- await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', frame], { signal: ctx.signal, timeoutMs: 60000 });
+ if (!r.ok || !d.existing(frame)) await blackTile(frame, layout, ctx);
+ }
+ return finishSprite({ src, job, ctx, dir, layout, duration });
+}
+
+/** `length` bytes at `offset` of a local file, or null (short read, gone). */
+function readRange(file, offset, length) {
+ let fd = null;
+ try {
+ fd = fs.openSync(file, 'r');
+ const b = Buffer.allocUnsafe(length);
+ let read = 0;
+ while (read < length) {
+ const n = fs.readSync(fd, b, read, length - read, offset + read);
+ if (!n) break;
+ read += n;
}
}
+ return read === length ? b : null;
+ } catch { return null; } finally { if (fd != null) { try { fs.closeSync(fd); } catch { /* gone */ } } }
+}
+
+/**
+ * A row's bytes, checked against the row's sha256: this node's file (a segment, or the packed chunk's byte range
+ * [byte_offset, byte_offset + byte_length)), else a ranged read of the durable copy through the placement router.
+ * null = no verified copy (the frame falls back to black).
+ */
+async function rowBuffer(row, ctx) {
+ const length = Number(row.byte_length);
+ if (row.local_path && d.existing(row.local_path)) {
+ const b = row.packed_object_id ? readRange(row.local_path, Number(row.byte_offset), length) : fs.readFileSync(row.local_path);
+ if (b && b.length === length && crypto.createHash('sha256').update(b).digest('hex') === row.sha256) return b;
}
}
- const out = path.join(dir, 'sprite.jpg');
- const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),
- '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });
- if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {
- d.cleanupWork(job.id);
- throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);
+ if (row.durable_provider && row.durability === 'durable' && row.key) {
+ const vodStorage = require('../vod/vod-storage');
+ const url = await vodStorage.presignGet(row.durable_provider, row.key, 900).catch(() => null);
+ if (!url) return null;
+ const packed = !!row.packed_object_id;
+ const headers = packed ? { range: `bytes=${Number(row.byte_offset)}-${Number(row.byte_offset) + length - 1}` } : {};
+ const res = await fetch(url, { headers, signal: ctx.signal }).catch(() => null);
+ if (!res) return null;
+ // A packed row is a byte range of the chunk: a server that ignores Range answers 200 with the whole
+ // (many MB) chunk. Refuse that without reading a byte, and never read past the length we asked for.
+ const refuse = () => {
+ try { const p = res.body && res.body.cancel ? res.body.cancel() : null; if (p && typeof p.catch === 'function') p.catch(() => { /* gone */ }); } catch { /* gone */ }
+ return null;
+ };
+ if (packed) {
+ if (res.status !== 206) return refuse();
+ if (Number(res.headers.get('content-length')) !== length) return refuse();
+ } else if (!res.ok) return refuse();
+ const b = Buffer.from(await res.arrayBuffer());
+ if (b.length === length && crypto.createHash('sha256').update(b).digest('hex') === row.sha256) return b;
}
}
- const made = await d.adopt({
- src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',
- relation: { method: 'sprite', ...layout },
- metadata: { sprite: layout, source_duration_seconds: duration },
- saveCheckpoint: ctx.saveCheckpoint,
- checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),
- });
- d.cleanupWork(job.id);
- return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };
+ return null;
+}
+
+/**
+ * The layout duration for a timeline sprite: the object's own duration (the seek path's metadata first, then its
+ * probe), never the timeline end unless neither is known. A truncated or in-progress cut must not shrink the sheet
+ * or its seek metadata; sample times past the timeline end already fall back to black tiles.
+ */
+async function spriteDuration(src, segs) {
+ const md = require('../objects/model').parseJson(src.metadata, {});
+ const known = Number(md.duration_seconds) || 0;
+ if (known > 0) return known;
+ const source = await d.resolveSource(src).catch(() => null);
+ if (source) {
+ const meta = await streamsOf(source.input).catch(() => ({ duration: 0 }));
+ if (Number(meta.duration) > 0) return Number(meta.duration);
+ }
+ return Number(segs[segs.length - 1].end_ms) / 1000;
+}
+
+/** Remove the sampled pieces of a failed run; the frame checkpoints (f*.jpg) stay, so a retry resumes. */
+function sweepPieces(dir) {
+ let names = [];
+ try { names = fs.readdirSync(dir); } catch { return; }
+ for (const n of names) if (/^seg-\d+\.mp4$/.test(n)) { try { fs.unlinkSync(path.join(dir, n)); } catch { /* gone */ } }
+}
+
+/**
+ * The sheet from the timeline (F3.1): each sample time picks the row covering it (segmentAt) and one frame is
+ * decoded from that row's bytes — init.mp4 + the segment as one small file, never the source, never a full-file
+ * decode. Layout, metadata and the player contract are the seek path's.
+ */
+async function runTimelineSprite(job, ctx, src, p, rows, segs) {
+ const initRow = rows.find((r) => Number(r.seq) === 0) || null;
+ // Pieces are deleted as the samples move past them, so only one or two are on disk at once.
+ const initLen = initRow ? Number(initRow.byte_length) || 0 : 0;
+ const maxSeg = segs.reduce((m, r) => Math.max(m, Number(r.byte_length) || 0), 0);
+ await d.checkRoom(src, 2 * (maxSeg + initLen) + 4 * MB);
+ const duration = await spriteDuration(src, segs);
+ if (!(duration > 0)) throw new JobError('no_duration', 'The timeline has no duration to spread frames over', { permanent: true });
+ const layout = spriteLayout(duration, p);
+ const dir = d.workDir(job.id);
+ // The init segment is read once and prepended to every piece; a media segment alone is not playable.
+ const initBody = initRow ? await rowBuffer(initRow, ctx) : null;
+ const pieceFile = (seq) => path.join(dir, `seg-${String(seq).padStart(6, '0')}.mp4`);
+ const dropPiece = (seq) => { if (seq != null) { try { fs.unlinkSync(pieceFile(seq)); } catch { /* already gone */ } } };
+ sweepPieces(dir); // pieces a previous attempt left behind are rebuilt from the rows as needed
+ const cache = new Map();
+ const piece = async (row) => {
+ const seq = Number(row.seq);
+ if (cache.has(seq)) return cache.get(seq);
+ const body = await rowBuffer(row, ctx);
+ let file = null;
+ if (body) {
+ file = pieceFile(seq);
+ fs.writeFileSync(file, initBody && seq !== 0 ? Buffer.concat([initBody, body]) : body);
+ }
+ cache.set(seq, file);
+ return file;
+ };
+ const tile = tileFilter(layout);
+ let decoded = 0;
+ let lastSeq = null;
+ try {
+ // An audio-only source has a timeline but no frame to cut: permanent, like the seek path's `no_video`.
+ const first = await piece(segs[0]);
+ if (first) {
+ const info = await require('../vod/media-tools').probeVodInfo(first);
+ if (info.streams.length && !info.streams.some((s) => s.codec_type === 'video')) {
+ d.cleanupWork(job.id);
+ throw new JobError('no_video', 'The source has no video track', { permanent: true });
+ }
+ }
+ for (let i = 0; i < layout.count; i++) {
+ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');
+ const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);
+ if (d.existing(frame)) { decoded++; continue; }
+ const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);
+ const row = await timeline.segmentAt(src.id, timeline.SOURCE, Math.max(0, at) * 1000);
+ let ok = false;
+ if (row) {
+ const seq = Number(row.seq);
+ // Samples are in time order: once the row advances, the previous piece has taken its last sample.
+ if (lastSeq != null && seq > lastSeq) dropPiece(lastSeq);
+ lastSeq = seq;
+ const file = await piece(row);
+ if (file) {
+ // Seek is relative to the segment's own start (ffmpeg shifts the fMP4 segment's tfdt to 0 with the init).
+ const rel = Math.max(0, at - Number(row.start_ms) / 1000);
+ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...(rel > 0 ? ['-ss', rel.toFixed(3)] : []), '-i', file,
+ '-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });
+ ok = r.ok && d.existing(frame);
+ }
+ }
+ if (ok) decoded++;
+ else await blackTile(frame, layout, ctx);
+ }
+ if (!decoded) throw new JobError('segment_unreadable', 'No frame could be read from the timeline segments; a retry reads the durable copies', { retryAfterS: 1800 });
+ return await finishSprite({ src, job, ctx, dir, layout, duration });
+ } finally {
+ // However the run ends (abort, ffmpeg, no_video), the pieces go: hundreds of MB must not be left behind.
+ // The frame checkpoints stay for a retry.
+ dropPiece(lastSeq);
+ sweepPieces(dir);
+ }
+}
+
+async function runSprite(job, ctx) {
+ const src = await d.loadSource(job);
+ const p = validateSprite({ obj: src, params: job.params });
+ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }
+ // Sprites ride the timeline when there is one (MEDIA_HLS_ENABLED); a source without one keeps the seek path.
+ const rows = config.hls.enabled ? await timeline.list(src.id) : [];
+ const segs = rows.filter((r) => Number(r.seq) > 0);
+ return segs.length ? runTimelineSprite(job, ctx, src, p, rows, segs) : runSeekSprite(job, ctx, src, p);
}
}
module.exports = {
module.exports = {
waveform: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateWaveform, run: runWaveform },
waveform: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateWaveform, run: runWaveform },
sprite: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateSprite, run: runSprite },
sprite: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateSprite, run: runSprite },
spriteLayout,
spriteLayout,
+ rowBuffer, // tested directly: the durable ranged-GET guards
};
};
@@ -51,6 +51,7 @@ const { spawn, spawnSync } = require('child_process');
@@ -51,6 +51,7 @@ const { spawn, spawnSync } = require('child_process');
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const waitFor = async (fn, ms = 60000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(25); } return false; };
const waitFor = async (fn, ms = 60000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(25); } return false; };
const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');
const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');
+ const size = (file) => { const r = spawnSync('ffprobe', ['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height', '-of', 'csv=p=0', file], { encoding: 'utf8' }); return r.stdout.trim(); };
const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
(async () => {
(async () => {
@@ -153,6 +154,18 @@ const { spawn, spawnSync } = require('child_process');
@@ -153,6 +154,18 @@ const { spawn, spawnSync } = require('child_process');
assert.strictEqual(JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]), before, 'the source is untouched');
assert.strictEqual(JSON.stringify([await model.getObject(pub), await model.listLocations(pub)]), before, 'the source is untouched');
assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
+ // ── A truncated timeline (an in-progress cut) must not shrink the sheet: the object's own 7 s duration
+ // lays it out, exactly as the seek path would, and the samples past the rows become black tiles ──
+ const shorty = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility: 'private', lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: 1, metadata: { duration_seconds: 7 } });
+ await timeline.replace(shorty, 'source', rows.slice(0, 2)); // init + one 3 s segment: the timeline ends at 3 s
+ const st = await runJob(shorty, {}, 'object.sprite');
+ assert.strictEqual(st.status, 'succeeded', JSON.stringify(st.error));
+ assert.deepStrictEqual(st.result.sprite, require('../server/jobs/previews').spriteLayout(7, { frames: 100, columns: 10, tile_width: 160 }), 'the object duration, not the 3 s timeline end, lays the sheet out');
+ assert.deepStrictEqual([st.result.sprite.count, st.result.sprite.interval_seconds], [3, 2.333], 'same interval/count as the seek path over the 7 s source');
+ const sto = await model.getObject(st.result.object_id);
+ assert.strictEqual(model.parseJson(sto.metadata, {}).source_duration_seconds, 7, 'the seek metadata carries the real duration');
+ assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
+
// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──
// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──
const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;
const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;
const autoPack = await packJobs(pub);
const autoPack = await packJobs(pub);
@@ -303,6 +316,104 @@ const { spawn, spawnSync } = require('child_process');
@@ -303,6 +316,104 @@ const { spawn, spawnSync } = require('child_process');
}
}
fs.renameSync(`${packed[1].local_path}.away`, packed[1].local_path);
fs.renameSync(`${packed[1].local_path}.away`, packed[1].local_path);
+ // ── Sprites come from the timeline (F3.1): each frame from the row covering its instant, decoded from the
+ // packed chunk's byte range, never the source. The seek layout and the player contract are unchanged ──
+ const { spriteLayout } = require('../server/jobs/previews');
+ const brightness = (file) => {
+ const r = spawnSync('ffmpeg', ['-v', 'error', '-i', file, '-frames:v', '1', '-vf', 'scale=1:1,format=gray', '-f', 'rawvideo', '-'], { encoding: null, maxBuffer: 1 << 20 });
+ return r.stdout && r.stdout.length ? r.stdout[r.stdout.length - 1] : -1;
+ };
+ const sr = await runJob(pub, {}, 'object.sprite');
+ assert.strictEqual(sr.status, 'succeeded', JSON.stringify(sr.error));
+ assert.deepStrictEqual(sr.result.sprite, spriteLayout(7, { frames: 100, columns: 10, tile_width: 160 }), 'the timeline sheet keeps the seek layout');
+ assert.deepStrictEqual([sr.result.sprite.count, sr.result.sprite.interval_seconds, sr.result.sprite.columns], [3, 2.333, 3]);
+ const so = await model.getObject(sr.result.object_id);
+ assert.deepStrictEqual([so.kind, so.mime_type, so.visibility], ['asset', 'image/jpeg', 'private']);
+ assert.strictEqual(size((await model.listLocations(so.id))[0].key), '480,90', 'the image is the layout: 3 × 160 by 1 × 90');
+ assert.deepStrictEqual(model.parseJson(so.metadata, {}).sprite, sr.result.sprite, 'the layout is in the metadata');
+ assert.strictEqual((await model.getVariant(pub, 'sprite')).derived_object_id, so.id);
+ assert.ok(brightness((await model.listLocations(so.id))[0].key) > 30, 'real frames from the rows, not black tiles');
+
+ // The source file gone: the seek path could not run at all, yet the rows' bytes cut the same sheet.
+ const pubFile = path.join(process.env.VOD_PATH, 'public.mp4');
+ fs.renameSync(pubFile, `${pubFile}.away`);
+ const sr2 = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');
+ assert.strictEqual(sr2.status, 'succeeded', JSON.stringify(sr2.error));
+ assert.deepStrictEqual([sr2.result.sprite.count, sr2.result.sprite.rows, sr2.result.sprite.columns], [3, 2, 2], 'the same layout from rows, with no source to seek');
+ assert.strictEqual(size((await model.listLocations(sr2.result.object_id))[0].key), '192,108');
+ fs.renameSync(`${pubFile}.away`, pubFile);
+
+ // ── The durable fallback and its guards. A packed row is a byte range of the chunk: with this node's
+ // chunk gone the frame must come from a 206 ranged GET of the durable copy. A 200 (Range ignored) or a
+ // wrong content-length is refused before the body is read; a sha mismatch is a black tile, not a failure ──
+ const previews = require('../server/jobs/previews');
+ assert.strictEqual(typeof previews.rowBuffer, 'function');
+ const durableRow = { durability: 'durable', durable_provider: 'b2', key: 'k', packed_object_id: 'chunk', byte_offset: 10, byte_length: 4, sha256: sha(Buffer.from('abcd')) };
+ const realFetch = global.fetch;
+ let cancelled = 0;
+ let read = 0;
+ global.fetch = async () => ({ ok: true, status: 200, headers: { get: () => null }, body: { cancel: async () => { cancelled++; } }, arrayBuffer: async () => { read++; return Buffer.from('abcd'); } });
+ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a 200 answer to a packed ranged GET is refused');
+ assert.deepStrictEqual([cancelled, read], [1, 0], 'refused without buffering the whole chunk');
+ cancelled = 0;
+ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '999' }, body: { cancel: async () => { cancelled++; } }, arrayBuffer: async () => { read++; return Buffer.alloc(999); } });
+ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a 206 of the wrong length is refused');
+ assert.deepStrictEqual([cancelled, read], [1, 0], 'refused before this one was read either');
+ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '4' }, body: null, arrayBuffer: async () => Buffer.from('0000') });
+ assert.strictEqual(await previews.rowBuffer(durableRow, {}), null, 'a sha mismatch is a black tile, not a failure');
+ global.fetch = async () => ({ ok: true, status: 206, headers: { get: () => '4' }, body: null, arrayBuffer: async () => Buffer.from('abcd') });
+ assert.deepStrictEqual(await previews.rowBuffer(durableRow, {}), Buffer.from('abcd'), 'the verified range');
+ global.fetch = realFetch;
+
+ const tileBrightness = (file, i, layout) => {
+ const x = (i % layout.columns) * layout.tile_width;
+ const y = Math.floor(i / layout.columns) * layout.tile_height;
+ const r = spawnSync('ffmpeg', ['-v', 'error', '-i', file, '-frames:v', '1', '-vf', `crop=${layout.tile_width}:${layout.tile_height}:${x}:${y},scale=1:1,format=gray`, '-f', 'rawvideo', '-'], { encoding: null, maxBuffer: 1 << 20 });
+ return r.stdout && r.stdout.length ? r.stdout[r.stdout.length - 1] : -1;
+ };
+ fs.renameSync(packed[1].local_path, `${packed[1].local_path}.sprite-away`);
+ const sd = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');
+ assert.strictEqual(sd.status, 'succeeded', JSON.stringify(sd.error));
+ assert.deepStrictEqual([sd.result.sprite.count, sd.result.sprite.columns, sd.result.sprite.rows], [3, 2, 2], 'the durable ranges cut the same sheet');
+ const sdFile = (await model.listLocations(sd.result.object_id))[0].key;
+ assert.ok([0, 1, 2].every((i) => tileBrightness(sdFile, i, sd.result.sprite) > 30), 'real frames from the durable chunk');
+ // One durable range corrupted: its tile is black, the others still decode.
+ const cleanChunk = blobs.get(`b2:${chunkKey}`);
+ const corruptChunk = Buffer.from(cleanChunk);
+ corruptChunk.fill(0, Number(packed[1].byte_offset), Number(packed[1].byte_offset) + Number(packed[1].byte_length));
+ blobs.set(`b2:${chunkKey}`, corruptChunk);
+ const sc = await runJob(pub, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');
+ assert.strictEqual(sc.status, 'succeeded', JSON.stringify(sc.error));
+ const scFile = (await model.listLocations(sc.result.object_id))[0].key;
+ assert.ok(tileBrightness(scFile, 0, sc.result.sprite) < 20, 'the sha-mismatched range is a black tile');
+ assert.ok(tileBrightness(scFile, 1, sc.result.sprite) > 30, 'the intact ranges still cut real frames');
+ blobs.set(`b2:${chunkKey}`, cleanChunk);
+ fs.renameSync(`${packed[1].local_path}.sprite-away`, packed[1].local_path);
+
+ // ── A failed run leaves no seg-*.mp4 behind, only the resumable frame checkpoints ──
+ {
+ const derive = require('../server/jobs/derive');
+ const acs = new AbortController();
+ acs.abort(new Error('sprite cancelled in test'));
+ const wdir = derive.workDir('mjob_SPRITEABORT');
+ fs.writeFileSync(path.join(wdir, 'f0000.jpg'), 'checkpoint');
+ await assert.rejects(previews.sprite.run({ id: 'mjob_SPRITEABORT', app_id: 'live', object_id: pub, params: {} }, { signal: acs.signal }), /sprite cancelled in test/);
+ assert.deepStrictEqual(fs.readdirSync(wdir).filter((n) => /^seg-.*\.mp4$/.test(n)), [], 'no sampled segment is left behind');
+ assert.ok(fs.existsSync(path.join(wdir, 'f0000.jpg')), 'the frame checkpoint survives for a retry');
+ derive.cleanupWork('mjob_SPRITEABORT');
+ }
+
+ // A source with no timeline uses the old seek path (and writes no rows).
+ const plainFile = path.join(process.env.VOD_PATH, 'plain.mp4');
+ fs.copyFileSync(src, plainFile);
+ const plain = await model.createObject({ app_id: 'live', owner_user_id: 5, visibility: 'private', lifecycle_status: 'ready', kind: 'file', mime_type: 'video/mp4', size_bytes: fs.statSync(plainFile).size });
+ await model.upsertLocation(plain, { provider: 'local', key: plainFile, state: 'present', size_bytes: fs.statSync(plainFile).size, verified: true });
+ assert.strictEqual(await timeline.has(plain), false);
+ const sp = await runJob(plain, { frames: 4, columns: 2, tile_width: 96 }, 'object.sprite');
+ assert.strictEqual(sp.status, 'succeeded', JSON.stringify(sp.error));
+ assert.deepStrictEqual([sp.result.sprite.count, sp.result.sprite.columns, sp.result.sprite.rows], [3, 2, 2], 'the seek path lays out the sheet the same way');
+ assert.strictEqual(await timeline.has(plain), false, 'and writes no timeline rows');
+
// ── A rerun packs nothing; a cmaf rerun keeps the packed rows ──
// ── A rerun packs nothing; a cmaf rerun keeps the packed rows ──
const jp2 = await runJob(pub, { target_seconds: 10 }, 'object.pack');
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.deepStrictEqual([jp2.status, jp2.result.packs, jp2.result.already_packed], ['succeeded', 0, 4], JSON.stringify(jp2.result));
@@ -36,6 +36,7 @@ const https = require('https');
@@ -36,6 +36,7 @@ const https = require('https');
'server/public/dev-data.js': 'the app\'s internal URL (APP_INTERNAL_URLS), ids and encoded names in the path',
'server/public/dev-data.js': 'the app\'s internal URL (APP_INTERNAL_URLS), ids and encoded names in the path',
'server/jobs/pack.js': 'a presigned GET on B2/R2 (operator-configured buckets) for a segment\'s durable copy',
'server/jobs/pack.js': 'a presigned GET on B2/R2 (operator-configured buckets) for a segment\'s durable copy',
'server/objects/routes.js': 'a ranged GET on a presigned B2/R2 URL (operator-configured buckets) for a packed segment\'s chunk',
'server/objects/routes.js': 'a ranged GET on a presigned B2/R2 URL (operator-configured buckets) for a packed segment\'s chunk',
+ 'server/jobs/previews.js': 'a (ranged) GET on a presigned B2/R2 URL (operator-configured buckets) for a timeline segment, to cut a sprite frame',
'server/drill.js': 'restore drill: only this server itself',
'server/drill.js': 'restore drill: only this server itself',
'server/me/client.js': 'browser script, same-origin',
'server/me/client.js': 'browser script, same-origin',
};
};