openvibers/openvibe.media Commits
Recordings open from R2, later segments from B2 (288 part 2) (#405)
Changes
nnew file mode 100644@@@ -0,0 +1,13 @@++# Recording opening tier (ask 288)++++With HLS enabled, `MEDIA_HLS_STARTUP_SECONDS` chooses how many opening seconds use R2 (180 by default, 0 disables, maximum 300). Ask 288 explicitly requests this placement enabled. Later segments use B2. Small initialization segments remain on R2; a packed chunk crossing the boundary stays on R2 as a whole. The original complete file is untouched.++++The existing heavy worker runs `object.startup-tier` after cuts and lazily when a master playlist is read. A public read does not wait for a copy. Lazy requests are capped by `MEDIA_HLS_BACKFILL_PER_HOUR` (20 by default), the queue's tenant limit, a ten-minute in-process throttle and a source/policy idempotency key. First-time access to an old recording can therefore still use B2 until the job completes. Each job copies at most 512 MiB and takes at most 30 minutes; unavailable providers, holds and verification failures retain the current routing.++++A joined recording uses its complete timeline: a later part does not get another three hot minutes. Clips do not retier their sources. Every move reads back the target's size and SHA-256, then commits under the same advisory locks as recuts/packs with a source snapshot check. Cancelled, changed or unverified copies are never published. Both R2 and B2 must be healthy. The read router still filters unhealthy providers; the segment route then uses the selected durable tier ahead of a retained local cache. `X-OV-Media-Provider` reports the actual provider on the Media response (including redirects and packed slices).++++A durable, internal-only `storage.startup.cleanup` job keeps old copies for 360 seconds: the presigned URL lifetime is 300 seconds and the joined timeline cache is 20 seconds. It refuses held objects and removes a copy only if no timeline row still names it. Failed deletes retry in the light worker. Packing keeps provider boundaries and preserves the current provider for each chunk.++++Validation: `ov test test/startup-tier.test.js` and `ov test test/media-timeline.test.js`, then the full suite. Production baseline and after-deploy measurements are in the PR and `~/openvibe/research-2026-11-10/player-starts.md`. The baseline recording (`/o/med_01M729VRTW6E3NPTTM008H0GAA`) used B2 at both `source/p1-000001.m4s` (0 s) and `source/p2-000019.m4s` (217.297 s). Once tiered, the first should report `X-OV-Media-Provider: r2` and the latter `b2`.++++Next for the design: none; this is storage, jobs and diagnostic headers only.@@@ -103,6 +103,8 @@ const config = {// 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: {++ // Ask 288 explicitly enables the opening tier. Zero is the owner's off switch.++ startupSeconds: Math.min(300, Math.max(0, intEnv('MEDIA_HLS_STARTUP_SECONDS', 180))),enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),transcode: ['1', 'true', 'on'].includes(String(process.env.MEDIA_CMAF_TRANSCODE || '').toLowerCase()),transcodeMaxSeconds: Math.max(0, intEnv('MEDIA_CMAF_TRANSCODE_MAX_SECONDS', 14400)),@@@ -65,7 +65,8 @@ function plan(rows, targetMs) {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;++ const next = run.length && Number(r.seq) === Number(run[run.length - 1].seq) + 1++ && r.durable_provider === run[run.length - 1].durable_provider;if (fits && (next || !run.length)) { run.push(r); continue; }if (run.length) runs.push(run);run = fits ? [r] : [];@@@ -136,6 +137,8 @@ async function run(job, ctx) {const dir = derive.workDir(job.id);try {for (const chunk of chunks) {++ // Keep an R2 opening hot when packing; plan never joins two provider tiers.++ const provider = chunk[0].durable_provider;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.@@@ -194,6 +197,8 @@ async function run(job, ctx) {if (kept.length) console.warn(`[Pack] ${job.id}: ${kept.length} packed segment location(s) kept — another object's rows name them`);out.left_behind += failed.length;out.removed_segments += chunk.length - failed.length;++ out.providers = [...new Set([...(out.providers || []), provider])];++ out.provider = out.providers.length === 1 ? provider : 'mixed';out.packs++;out.segments += chunk.length;out.rows_packed += chunk.length;@@@ -222,6 +222,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '// its retry.try { await require('../objects/timeline-queue').queuePack(job.app_id, src.id, { rendition }); }catch (err) { console.warn(`[${label}] ${job.id}: pack queue failed: ${err.message}`); }++ try { await require('./startup-tier').queueOnce(src); }++ catch (err) { console.warn(`[${label}] ${job.id}: startup queue failed: ${err.message}`); }return {rows: written, uploaded,durable: rows.filter((x) => x.durability === 'durable').length,nnew file mode 100644@@@ -0,0 +1,153 @@++/** Ask 288: the first three minutes of a recording on R2, later segments on B2. */++'use strict';++const crypto = require('crypto');++const config = require('../config');++const db = require('../db/database');++const timeline = require('../objects/timeline');++const model = require('../objects/model');++const { JobError } = require('./queue');++const MAX_BYTES = 512 * 1024 * 1024;++const recent = new Map();++const queuedAt = [];++let pending = 0;++const identity = (r) => `${r.object_id}\n${r.rendition}\n${r.key}`;++++// Packed chunks that straddle the boundary stay hot as a whole. Inits are small and always hot, including later++// discontinuities. Joined recording rows carry recording time and retain their original object_id/key.++function plan(rows, seconds = config.hls.startupSeconds) {++ const groups = new Map();++ for (const row of rows) {++ const id = identity(row);++ const hot = Number(row.seq) <= 0 || Number(row.start_ms) < seconds * 1000;++ const previous = groups.get(id);++ groups.set(id, { row, provider: hot || (previous && previous.provider === 'r2') ? 'r2' : 'b2' });++ }++ return groups;++}++// The shared read router applies this preference only after its health and capability checks.++function playbackPreference(row) {++ return config.hls.startupSeconds && row.durability === 'durable' ? row.durable_provider : null;++}++++async function subject(obj) {++ if (!obj || obj.kind === 'clip') return null; // A clip must not retier its source around a different zero point.++ const parent = await db.get('SELECT p.recording_object_id FROM media_recording_parts p JOIN vods v ON v.id = p.vod_id WHERE v.object_id = ? LIMIT 1', [obj.id]);++ return parent ? await model.getObject(parent.recording_object_id) : obj;++}++async function rowsOf(obj) {++ if (require('../vod/recordings').isRecording(obj)) return (await require('../objects/hls').timelineOf(obj)).rows;++ return db.all('SELECT * FROM media_timeline WHERE object_id = ? ORDER BY rendition, seq', [obj.id]);++}++function providersReady() {++ const storage = require('../vod/vod-storage');++ const providers = require('../placement/providers');++ return storage.providerAvailable('r2') && storage.providerAvailable('b2')++ && providers.placementGate('r2','online-hot').eligible++ && providers.placementGate('b2','online-canonical').eligible;++}++async function queueOnce(obj) {++ if (!config.hls.enabled || !config.hls.startupSeconds) return null;++ if (!providersReady()) return null;++ obj = await subject(obj);++ if (!obj || obj.lifecycle_status !== 'ready') return null;++ const now = Date.now();++ if (recent.has(obj.id) && now - recent.get(obj.id) < 600000) return null;++ const rows = await rowsOf(obj);++ if (!rows.some(r => Number(r.seq) > 0)) return null;++ const groups = plan(rows);++ if (![...groups.values()].some(({row,provider}) => row.durable_provider !== provider || row.durability !== 'durable')) return null;++ const hash = crypto.createHash('sha256').update(String(config.hls.startupSeconds));++ for (const [key, {row,provider}] of groups) hash.update(`${key}:${row.sha256}:${provider}\n`);++ const recentJobs = await db.get("SELECT COUNT(*) AS n FROM media_jobs WHERE job_type = 'object.startup-tier' AND created_at >= ov_now_iso('-1 hour')");++ while (queuedAt.length && queuedAt[0] <= now - 3600000) queuedAt.shift();++ if (Math.max(Number(recentJobs.n),queuedAt.length) + pending >= config.hls.backfillPerHour) return null;++ pending++;++ let result;++ try { result = await require('./queue').enqueue({appId:obj.app_id,type:'object.startup-tier',objectId:obj.id,params:{},++ dedupeActive:true,idempotencyKey:`startup-tier:${obj.id}:${hash.digest('hex').slice(0,20)}`,createdBy:'system:startup-tier'});++ if (result.created) queuedAt.push(now);++ } finally { pending--; }++ if (recent.size >= 1000) recent.delete(recent.keys().next().value);++ recent.set(obj.id,now);++ return result;++}++function validate({obj,params}) {++ if (!obj || obj.kind === 'clip') throw new JobError('media.job.invalid','Startup placement requires a recording or media object',{permanent:true});++ if (Object.keys(params || {}).length) throw new JobError('media.job.invalid','Startup placement takes no parameters',{permanent:true});++ return {};++}++async function run(job, ctx) {++ if (!config.hls.enabled || !config.hls.startupSeconds) return {skipped:'startup placement disabled'};++ const obj = await model.getObject(job.object_id);++ if (!obj || obj.app_id !== job.app_id || obj.lifecycle_status !== 'ready') throw new JobError('not_ready','Recording is not ready',{permanent:true});++ const rows = await rowsOf(obj), groups = plan(rows);++ const vs = require('../vod/vod-storage');++ if (!providersReady()) throw new JobError('provider_unavailable','R2 and B2 must both be healthy and eligible',{retryAfterS:300});++ const ids = [...new Set([obj.id,...rows.map(r=>r.object_id)])].sort();++ for (const id of ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Held recording stays in place',{retryAfterS:21600});++ const originals = [];++ for (const id of ids) originals.push(...await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [id]));++ const moving = originals.filter(r=>groups.has(identity(r)) && (r.durable_provider !== groups.get(identity(r)).provider || r.durability !== 'durable'));++ if (!moving.length) return {moved:0,bytes:0};++ const copies = new Map(), staged = [], updated = [];++ let bytes = 0, committed = false;++ try {++ for (const row of moving) {++ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('Cancelled');++ const provider = groups.get(identity(row)).provider;++ const copyId = `${provider}\n${row.key}`;++ if (!copies.has(copyId)) {++ const local = row.local_path && require('fs').existsSync(row.local_path);++ const size = local ? require('fs').statSync(row.local_path).size++ : Number((await vs.headObject(row.durable_provider,row.key) || {}).size);++ if (!(size > 0) || bytes + size > MAX_BYTES) throw new JobError('startup_budget','Startup move exceeds 512 MiB or has no readable source',{permanent:true});++ bytes += size;++ staged.push({...row,local_path:null,durable_provider:provider});++ if (local) await vs.uploadFile(provider,row.key,row.local_path,Number(row.seq)<=0?'video/mp4':'video/iso.segment',{signal:ctx.signal});++ else if (row.durability === 'durable') await vs.copyBetweenProviders(row.durable_provider,provider,row.key);++ else throw new JobError('segment_unreadable','No durable source',{retryAfterS:300});++ const back = await vs.sha256Object(provider,row.key);++ if (!back || back.sha256 !== (row.packed_object_id || row.sha256) || Number(back.size) !== size)++ throw new JobError('copy_unverified','Startup copy failed size/SHA256 verification',{retryAfterS:300});++ copies.set(copyId,true);++ }++ updated.push({...row,durable_provider:provider,durability:'durable'});++ }++ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('Cancelled');++ await db.getDb().tx(async()=>{++ // Same locks as timeline.replace and markPacked; the complete snapshot is checked before any row moves.++ for (const id of ids) await db.get('SELECT pg_advisory_xact_lock(hashtext(?)) AS locked',[`media.timeline:${id}`]);++ for (const id of ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Hold added during copy',{retryAfterS:21600});++ for (const row of moving) {++ const current = await timeline.byName(row.object_id,row.rendition,row.name);++ if (!current || ['key','sha256','durable_provider','packed_object_id','local_path'].some(k=>current[k]!==row[k]))++ throw new JobError('media.timeline.changed','Timeline changed during startup copy',{retryAfterS:60});++ }++ const queue = require('./queue');++ await queue.enqueue({appId:queue.SYSTEM_APP,type:'storage.startup.cleanup',params:{rows:moving.map(r=>({object_id:r.object_id,durable_provider:r.durable_provider,key:r.key})),object_ids:ids},runAfterS:360,idempotencyKey:`startup-cleanup:${job.id}`,createdBy:'system:startup-tier'});++ for (const row of updated) await db.run("UPDATE media_timeline SET durable_provider = ?, durability = 'durable', updated_at = ov_now() WHERE object_id = ? AND rendition = ? AND seq = ?",[row.durable_provider,row.object_id,row.rendition,row.seq]);++ });++ committed = true;++ // Old presigned URLs (300 s) and joined-row caches (20 s) stay valid through a durable 360 s cleanup job.++ return {moved:updated.length,bytes,hot_seconds:config.hls.startupSeconds,cleanup_after_s:360};++ } finally {++ if (!committed) await timeline.deleteBytes(staged).catch(()=>{});++ }++}++const cleanup = {++ lane:'light',maxAttempts:6,timeoutMs:300000,needsObject:false,++ validate({appId,params}) {++ if (appId !== require('./queue').SYSTEM_APP) throw new JobError('media.job.forbidden','Internal cleanup only',{permanent:true});++ if (!params || !Array.isArray(params.rows) || params.rows.length > 10000 || !Array.isArray(params.object_ids))++ throw new JobError('media.job.invalid','Cleanup rows required',{permanent:true});++ return params;++ },++ async run(job) {++ cleanup.validate({appId:job.app_id,params:job.params});++ for (const id of job.params.object_ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Held copies stay in place',{retryAfterS:21600});++ const result=await timeline.deleteBytes(job.params.rows.map(r=>({durable_provider:r.durable_provider,key:r.key,local_path:null})));++ if (result.failed.length) throw new JobError('cleanup_failed','Old segment copy deletion failed',{retryAfterS:600});++ return {kept:result.kept.length,removed:job.params.rows.length-result.kept.length};++ },++};++module.exports = {plan,queueOnce,playbackPreference,cleanup, spec:{lane:'heavy',maxAttempts:3,timeoutMs:30*60*1000,needsObject:true,validate,run}};@@@ -29,6 +29,8 @@ 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('object.startup-tier', require('./startup-tier').spec);++queue.register('storage.startup.cleanup', require('./startup-tier').cleanup);queue.register('rendition.create', require('./rendition').spec);queue.register('vod.finalize', require('./vod-finalize').spec);queue.register('clip.cut', require('../vod/clip-jobs').spec);@@@ -1014,6 +1014,8 @@ publicRouter.get('/:id/master.m3u8', hlsRoute(async (req, res, { obj, query }) =// A missing rendition (F4) is queued lazily; only the ones whose rows exist are listed, so a client never follows// a variant that is not there yet.await hls.queueRenditionsOnce(obj);++ // Existing recordings are warmed in the bounded worker queue; playlist delivery never waits for a copy.++ require('../jobs/startup-tier').queueOnce(obj).catch(err => console.warn('[HLS] startup queue:', err.message));res.type(HLS_TYPE).send(timeline.masterPlaylist(await hls.renditionsFor(obj, rows), { query }));}));@@@ -1086,8 +1088,10 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {const router = require('../placement/router');const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: row.name,purpose: 'playback', session: router.sessionFor(req, `${obj.id}/${rendition}`),-- contentType: mime, expiresIn: 300, presign: !packed });++ contentType: mime, expiresIn: 300, presign: !packed,++ preferredProvider: require('../jobs/startup-tier').playbackPreference(row) });for (const loc of decision.candidates || []) {++ headers['X-OV-Media-Provider'] = loc.provider;if (loc.provider === 'local') {if (!fs.existsSync(loc.key)) continue;require('./timeline-cache').served(loc.key);@@@ -1102,6 +1106,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {continue;}res.set('Cache-Control', 'private, max-age=0');++ res.set('X-OV-Media-Provider', loc.provider);return res.redirect(302, url);}res.status(404).json({ error: 'Segment bytes unavailable' });@@@ -167,8 +167,10 @@ function tierOf(candidate, purpose) {}/** Rank the candidates: by tier, with a measured latency advantage (> TIE_BAND) allowed to promote. */--function rankCandidates(candidates, { purpose = 'playback' } = {}) {++function rankCandidates(candidates, { purpose = 'playback', preferredProvider = null } = {}) {return [...candidates].sort((a, b) => {++ const ap = a.provider === preferredProvider, bp = b.provider === preferredProvider;++ if (ap !== bp) return ap ? -1 : 1;const ta = tierOf(a, purpose), tb = tierOf(b, purpose);if (ta === tb) {const ma = measuredMs(a.provider), mb = measuredMs(b.provider);@@@ -217,6 +219,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);async function route(opts = {}) {const purpose = opts.purpose || 'playback';const session = opts.session || null;++ // A verified timeline names its opening/tail tier. This never admits a missing or unhealthy copy.++ const preferredProvider = purpose === 'playback' && (opts.preferredProvider === 'r2' || opts.preferredProvider === 'b2') ? opts.preferredProvider : null;const locs = (await locationsFor(opts)).filter((l) => l && l.state === 'present' && l.key);// One candidate per provider: the first present location it has. A copy on an unhealthy (or avoided) provider@@@ -234,7 +238,7 @@ async function route(opts = {}) {}const lastResort = candidates.length === 0 && degraded.length > 0;if (lastResort) candidates = degraded;-- const ranked = rankCandidates(candidates, { purpose });++ const ranked = rankCandidates(candidates, { purpose, preferredProvider });// Sticky: same session -> same provider for STICKY_TTL_MS unless its copy/breaker is gone, or a better tier// is available again. Stickiness stops latency flapping between copies of one tier; it never holds a viewer@@@ -244,7 +248,7 @@ async function route(opts = {}) {let stickyHit = false;if (sticky) {const at = ranked.findIndex((c) => c.provider === sticky.provider);-- if (at >= 0 && tierOf(ranked[at], purpose) <= tierOf(ranked[0], purpose)) { ordered = [ranked[at], ...ranked.slice(0, at), ...ranked.slice(at + 1)]; stickyHit = true; }++ if (at >= 0 && (!preferredProvider || ranked[0].provider !== preferredProvider || ranked[at].provider === preferredProvider) && tierOf(ranked[at], purpose) <= tierOf(ranked[0], purpose)) { ordered = [ranked[at], ...ranked.slice(0, at), ...ranked.slice(at + 1)]; stickyHit = true; }else stickyByKey.delete(session);}@@@ -261,7 +265,7 @@ async function route(opts = {}) {const fallbacks = ordered.filter((c) => c !== primary).map((c) => c.provider);const reason = primary-- ? lastResort ? 'last-resort' : (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy')++ ? lastResort ? 'last-resort' : primary.provider === preferredProvider ? 'timeline-tier' : (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy'): 'no-healthy-copy';if (primary) stickyPut(session, primary.provider);@@@ -1,4 +1,6 @@'use strict';++// This fixture verifies local/remote packing explicitly; startup placement has its own move and routing tests.++process.env.MEDIA_HLS_STARTUP_SECONDS = '0';// Segment-native video, first slice (F3.1; docs/media-fabric.md §3): object.cmaf stream-copies a finished video into// CMAF segments on its own keyframes and writes one media_timeline row per segment; the HLS playlists come from the// rows (ENDLIST, the rows' durations); segmentAt answers the segment at an instant, at the boundaries too; a rerun is a@@@ -179,6 +179,20 @@ const assert = require('assert');assert.strictEqual(decision.provider, null, 'never an unconfigured provider, even as a last resort');signals.reset(); router.reset();++ // A timeline's selected cloud tier beats its retained local cache and the old sticky choice.++ providers.reset(); signals.reset(); router.reset();++ providers.isConfigured = () => true;++ const startupCopies = [{provider:'local',state:'present',key:'/tmp/cache'},{provider:'b2',state:'present',key:'tail'}];++ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false});++ assert.strictEqual(decision.provider,'local');++ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false,preferredProvider:'b2'});++ assert.strictEqual(decision.provider,'b2'); assert.strictEqual(decision.reason,'timeline-tier');++ assert.strictEqual(decision.candidates[0].provider,'b2','metrics, sticky choice and served order agree');++ for (let i=0;i<10;i++) signals.recordFailure('b2','presign',50);++ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false,preferredProvider:'b2'});++ assert.strictEqual(decision.provider,'local','an unhealthy preferred provider is still filtered out');++ signals.reset(); router.reset();++providers.isConfigured = realIsConfigured;console.log('placement-router: all checks passed');process.exit(0);nnew file mode 100644@@@ -0,0 +1,88 @@++'use strict';++const assert = require('assert');++const crypto = require('crypto');++process.env.MEDIA_HLS_ENABLED='1';++process.env.MEDIA_HLS_STARTUP_SECONDS='180';++(async()=>{++ const db=require('../server/db/database'), timeline=require('../server/objects/timeline');++ const model=require('../server/objects/model'), storage=require('../server/vod/vod-storage');++ const startup=require('../server/jobs/startup-tier'), config=require('../server/config');++ const obj={id:'startup-test',app_id:'live',kind:'vod',lifecycle_status:'ready'};++ const queue=require('../server/jobs/queue');++ const original={get:model.getObject,held:model.isHeld,enqueue:queue.enqueue};++ model.getObject=async()=>obj; model.isHeld=async()=>false;++ const blobs=new Map(), sha=b=>crypto.createHash('sha256').update(b).digest('hex');++ const source=(seq,start)=>{++ const bytes=Buffer.from(`segment-${seq}`), key=`startup/${seq}.m4s`; blobs.set(`b2:${key}`,bytes);++ return {seq,name:seq?`${seq}.m4s`:'init.mp4',start_ms:start,end_ms:seq?start+4000:0,keyframe_ms:seq?start:null,++ key,local_path:null,durable_provider:'b2',packed_object_id:null,byte_offset:null,byte_length:bytes.length,++ sha256:sha(bytes),durability:'durable',init_name:null};++ };++ const rows=[source(0,0),source(1,176000),source(2,180000),source(3,220000)];++ await timeline.replace(obj.id,'source',rows);++ const native={};++ for(const k of ['providerAvailable','providerConfigured','headObject','copyBetweenProviders','sha256Object','deleteObject'])native[k]=storage[k];++ storage.providerAvailable=storage.providerConfigured=()=>true;++ storage.headObject=async(p,k)=>blobs.has(`${p}:${k}`)?{size:blobs.get(`${p}:${k}`).length}:null;++ storage.copyBetweenProviders=async(a,b,k)=>{blobs.set(`${b}:${k}`,blobs.get(`${a}:${k}`));return true;};++ storage.sha256Object=async(p,k)=>{const b=blobs.get(`${p}:${k}`);return b?{size:b.length,sha256:sha(b)}:null;};++ storage.deleteObject=async(p,k)=>{blobs.delete(`${p}:${k}`);return true;};++ require('../server/jobs/queue').register('storage.startup.cleanup',startup.cleanup);++ const ctx={signal:new AbortController().signal};++ try {++ let queuedByRead=null; queue.enqueue=async request=>{queuedByRead=request;return request;};++ await startup.queueOnce(obj);++ assert.strictEqual(queuedByRead.objectId,obj.id,'master queues its existing source through the real recording-part lookup');++ queue.enqueue=original.enqueue;++ const read=storage.sha256Object;++ storage.sha256Object=async()=>({size:9,sha256:'wrong'});++ await assert.rejects(startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx),/verification/);++ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'failed copy leaves every original route');++ assert(blobs.has('b2:startup/0.m4s')); assert(!blobs.has('r2:startup/0.m4s'),'uncommitted copy cleaned');++ storage.sha256Object=read;++ const copy=storage.copyBetweenProviders;++ const cancelled=new AbortController();++ storage.copyBetweenProviders=async(a,b,k)=>{await copy(a,b,k);cancelled.abort(new Error('Test cancelled'));};++ await assert.rejects(startup.spec.run({id:'cancelled-job',app_id:'live',object_id:obj.id},{signal:cancelled.signal}),/Test cancelled/);++ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'cancellation keeps original routes');++ assert(!blobs.has('r2:startup/0.m4s'),'cancelled copies cleaned');++ let raced=false;++ storage.copyBetweenProviders=async(a,b,k)=>{++ await copy(a,b,k);++ if(!raced){raced=true;await db.run('UPDATE media_timeline SET packed_object_id = ? WHERE object_id = ? AND seq = 1',['concurrent-pack',obj.id]);}++ };++ await assert.rejects(startup.spec.run({id:'raced-job',app_id:'live',object_id:obj.id},ctx),/Timeline changed/);++ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'a concurrent pack is never overwritten');++ await db.run('UPDATE media_timeline SET packed_object_id = NULL WHERE object_id = ?',[obj.id]);++ storage.copyBetweenProviders=copy;++ await timeline.replace('startup-alias','source',[rows[1]]);++ const result=await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx);++ assert.strictEqual(result.moved,2);++ assert(blobs.has('b2:startup/0.m4s'),'old signed URLs stay readable until cleanup');++ const queued=await db.get("SELECT * FROM media_jobs WHERE job_type = 'storage.startup.cleanup'");++ assert(queued.run_after,'cleanup is durably delayed');++ assert.throws(()=>startup.cleanup.validate({appId:'live',params:{rows:[],object_ids:[]}}),/Internal/);++ const cleaned=await startup.cleanup.run({...queued,params:JSON.parse(queued.params)});++ assert.strictEqual(cleaned.kept,1,'another timeline retains its old copy');++ assert(blobs.has('b2:startup/1.m4s'));++ assert(!blobs.has('b2:startup/0.m4s'));++ assert(blobs.has('r2:startup/0.m4s'));++ assert.deepStrictEqual((await timeline.list(obj.id)).map(r=>r.durable_provider),['r2','r2','b2','b2']);++ assert.strictEqual((await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx)).moved,0,'idempotent');++ const joined=[{...rows[1],object_id:'part1',rendition:'source',start_ms:179000,key:'chunk'},++ {...rows[2],object_id:'part1',rendition:'source',start_ms:181000,key:'chunk'},++ {...rows[1],object_id:'part2',rendition:'source',start_ms:220000,key:'later'}];++ assert.deepStrictEqual([...startup.plan(joined).values()].map(x=>x.provider),['r2','b2'],'whole crossing chunk stays hot; part2 uses recording time');++ assert.strictEqual(startup.playbackPreference({durability:'durable',durable_provider:'b2'}),'b2');++ assert.strictEqual(startup.playbackPreference({durability:'local',durable_provider:null}),null);++ config.hls.startupSeconds=0;++ assert.strictEqual((await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx)).skipped,'startup placement disabled');++ config.hls.startupSeconds=180;++ model.isHeld=async()=>true;++ await assert.rejects(startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx),/Held/);++ console.log('startup tier: verified R2 prefix/B2 tail, failed-copy rollback, idempotency, packed boundary, recording time, hold and off switch passed');++ } finally {++ Object.assign(storage,native);model.getObject=original.get;model.isHeld=original.held;queue.enqueue=original.enqueue;++ await db.run('DELETE FROM media_timeline WHERE object_id IN (?, ?)',[obj.id,'startup-alias']); await db.close();++ }++})().catch(e=>{console.error(e);process.exitCode=1});@@@ -119,6 +119,7 @@ const { EventEmitter } = require('events');const res = { destroyed: false, writeHead(code, headers) { this.code = code; this.headers = headers; }, end(body) { this.body = body; } };await require('../server/objects/routes').sendSegmentRow(req, res, { obj: { id: 'cache-packed' }, closed: false }, packedRow, 'source');assert.strictEqual(res.code, 200);++ assert.strictEqual(res.headers['X-OV-Media-Provider'], 'b2', 'packed responses identify their backing tier');assert.deepStrictEqual(res.body, Buffer.from('cde'), 'the segment route reads the packed range from durable storage');} finally {router.route = realRoute; router.sessionFor = realSession; storage.presignGet = realPresign; global.fetch = realFetch;
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+# Recording opening tier (ask 288)
+
+With HLS enabled, `MEDIA_HLS_STARTUP_SECONDS` chooses how many opening seconds use R2 (180 by default, 0 disables, maximum 300). Ask 288 explicitly requests this placement enabled. Later segments use B2. Small initialization segments remain on R2; a packed chunk crossing the boundary stays on R2 as a whole. The original complete file is untouched.
+
+The existing heavy worker runs `object.startup-tier` after cuts and lazily when a master playlist is read. A public read does not wait for a copy. Lazy requests are capped by `MEDIA_HLS_BACKFILL_PER_HOUR` (20 by default), the queue's tenant limit, a ten-minute in-process throttle and a source/policy idempotency key. First-time access to an old recording can therefore still use B2 until the job completes. Each job copies at most 512 MiB and takes at most 30 minutes; unavailable providers, holds and verification failures retain the current routing.
+
+A joined recording uses its complete timeline: a later part does not get another three hot minutes. Clips do not retier their sources. Every move reads back the target's size and SHA-256, then commits under the same advisory locks as recuts/packs with a source snapshot check. Cancelled, changed or unverified copies are never published. Both R2 and B2 must be healthy. The read router still filters unhealthy providers; the segment route then uses the selected durable tier ahead of a retained local cache. `X-OV-Media-Provider` reports the actual provider on the Media response (including redirects and packed slices).
+
+A durable, internal-only `storage.startup.cleanup` job keeps old copies for 360 seconds: the presigned URL lifetime is 300 seconds and the joined timeline cache is 20 seconds. It refuses held objects and removes a copy only if no timeline row still names it. Failed deletes retry in the light worker. Packing keeps provider boundaries and preserves the current provider for each chunk.
+
+Validation: `ov test test/startup-tier.test.js` and `ov test test/media-timeline.test.js`, then the full suite. Production baseline and after-deploy measurements are in the PR and `~/openvibe/research-2026-11-10/player-starts.md`. The baseline recording (`/o/med_01M729VRTW6E3NPTTM008H0GAA`) used B2 at both `source/p1-000001.m4s` (0 s) and `source/p2-000019.m4s` (217.297 s). Once tiered, the first should report `X-OV-Media-Provider: r2` and the latter `b2`.
+
+Next for the design: none; this is storage, jobs and diagnostic headers only.
@@ -103,6 +103,8 @@ const config = {
@@ -103,6 +103,8 @@ const config = {
// 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
// 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.
// media_timeline (/o/:id/master.m3u8, /o/:id/source/…). Off by default: off, the job is refused and the routes 404.
hls: {
hls: {
+ // Ask 288 explicitly enables the opening tier. Zero is the owner's off switch.
+ startupSeconds: Math.min(300, Math.max(0, intEnv('MEDIA_HLS_STARTUP_SECONDS', 180))),
enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
enabled: ['1', 'true', 'on'].includes(String(process.env.MEDIA_HLS_ENABLED || '').toLowerCase()),
transcode: ['1', 'true', 'on'].includes(String(process.env.MEDIA_CMAF_TRANSCODE || '').toLowerCase()),
transcode: ['1', 'true', 'on'].includes(String(process.env.MEDIA_CMAF_TRANSCODE || '').toLowerCase()),
transcodeMaxSeconds: Math.max(0, intEnv('MEDIA_CMAF_TRANSCODE_MAX_SECONDS', 14400)),
transcodeMaxSeconds: Math.max(0, intEnv('MEDIA_CMAF_TRANSCODE_MAX_SECONDS', 14400)),
@@ -65,7 +65,8 @@ function plan(rows, targetMs) {
@@ -65,7 +65,8 @@ function plan(rows, targetMs) {
let run = [];
let run = [];
for (const r of segs) {
for (const r of segs) {
const fits = !r.packed_object_id && r.durability === 'durable' && r.durable_provider;
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;
+ const next = run.length && Number(r.seq) === Number(run[run.length - 1].seq) + 1
+ && r.durable_provider === run[run.length - 1].durable_provider;
if (fits && (next || !run.length)) { run.push(r); continue; }
if (fits && (next || !run.length)) { run.push(r); continue; }
if (run.length) runs.push(run);
if (run.length) runs.push(run);
run = fits ? [r] : [];
run = fits ? [r] : [];
@@ -136,6 +137,8 @@ async function run(job, ctx) {
@@ -136,6 +137,8 @@ async function run(job, ctx) {
const dir = derive.workDir(job.id);
const dir = derive.workDir(job.id);
try {
try {
for (const chunk of chunks) {
for (const chunk of chunks) {
+ // Keep an R2 opening hot when packing; plan never joins two provider tiers.
+ const provider = chunk[0].durable_provider;
aborted(ctx);
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 });
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.
// Concatenate, recording where each segment starts in the chunk.
@@ -194,6 +197,8 @@ async function run(job, ctx) {
@@ -194,6 +197,8 @@ async function run(job, ctx) {
if (kept.length) console.warn(`[Pack] ${job.id}: ${kept.length} packed segment location(s) kept — another object's rows name them`);
if (kept.length) console.warn(`[Pack] ${job.id}: ${kept.length} packed segment location(s) kept — another object's rows name them`);
out.left_behind += failed.length;
out.left_behind += failed.length;
out.removed_segments += chunk.length - failed.length;
out.removed_segments += chunk.length - failed.length;
+ out.providers = [...new Set([...(out.providers || []), provider])];
+ out.provider = out.providers.length === 1 ? provider : 'mixed';
out.packs++;
out.packs++;
out.segments += chunk.length;
out.segments += chunk.length;
out.rows_packed += chunk.length;
out.rows_packed += chunk.length;
@@ -222,6 +222,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
@@ -222,6 +222,8 @@ async function publishRendition({ job, ctx, src, rendition, out, list, label = '
// its retry.
// its retry.
try { await require('../objects/timeline-queue').queuePack(job.app_id, src.id, { rendition }); }
try { await require('../objects/timeline-queue').queuePack(job.app_id, src.id, { rendition }); }
catch (err) { console.warn(`[${label}] ${job.id}: pack queue failed: ${err.message}`); }
catch (err) { console.warn(`[${label}] ${job.id}: pack queue failed: ${err.message}`); }
+ try { await require('./startup-tier').queueOnce(src); }
+ catch (err) { console.warn(`[${label}] ${job.id}: startup queue failed: ${err.message}`); }
return {
return {
rows: written, uploaded,
rows: written, uploaded,
durable: rows.filter((x) => x.durability === 'durable').length,
durable: rows.filter((x) => x.durability === 'durable').length,
new file mode 100644
new file mode 100644
@@ -0,0 +1,153 @@
@@ -0,0 +1,153 @@
+/** Ask 288: the first three minutes of a recording on R2, later segments on B2. */
+'use strict';
+const crypto = require('crypto');
+const config = require('../config');
+const db = require('../db/database');
+const timeline = require('../objects/timeline');
+const model = require('../objects/model');
+const { JobError } = require('./queue');
+const MAX_BYTES = 512 * 1024 * 1024;
+const recent = new Map();
+const queuedAt = [];
+let pending = 0;
+const identity = (r) => `${r.object_id}\n${r.rendition}\n${r.key}`;
+
+// Packed chunks that straddle the boundary stay hot as a whole. Inits are small and always hot, including later
+// discontinuities. Joined recording rows carry recording time and retain their original object_id/key.
+function plan(rows, seconds = config.hls.startupSeconds) {
+ const groups = new Map();
+ for (const row of rows) {
+ const id = identity(row);
+ const hot = Number(row.seq) <= 0 || Number(row.start_ms) < seconds * 1000;
+ const previous = groups.get(id);
+ groups.set(id, { row, provider: hot || (previous && previous.provider === 'r2') ? 'r2' : 'b2' });
+ }
+ return groups;
+}
+// The shared read router applies this preference only after its health and capability checks.
+function playbackPreference(row) {
+ return config.hls.startupSeconds && row.durability === 'durable' ? row.durable_provider : null;
+}
+
+async function subject(obj) {
+ if (!obj || obj.kind === 'clip') return null; // A clip must not retier its source around a different zero point.
+ const parent = await db.get('SELECT p.recording_object_id FROM media_recording_parts p JOIN vods v ON v.id = p.vod_id WHERE v.object_id = ? LIMIT 1', [obj.id]);
+ return parent ? await model.getObject(parent.recording_object_id) : obj;
+}
+async function rowsOf(obj) {
+ if (require('../vod/recordings').isRecording(obj)) return (await require('../objects/hls').timelineOf(obj)).rows;
+ return db.all('SELECT * FROM media_timeline WHERE object_id = ? ORDER BY rendition, seq', [obj.id]);
+}
+function providersReady() {
+ const storage = require('../vod/vod-storage');
+ const providers = require('../placement/providers');
+ return storage.providerAvailable('r2') && storage.providerAvailable('b2')
+ && providers.placementGate('r2','online-hot').eligible
+ && providers.placementGate('b2','online-canonical').eligible;
+}
+async function queueOnce(obj) {
+ if (!config.hls.enabled || !config.hls.startupSeconds) return null;
+ if (!providersReady()) return null;
+ obj = await subject(obj);
+ if (!obj || obj.lifecycle_status !== 'ready') return null;
+ const now = Date.now();
+ if (recent.has(obj.id) && now - recent.get(obj.id) < 600000) return null;
+ const rows = await rowsOf(obj);
+ if (!rows.some(r => Number(r.seq) > 0)) return null;
+ const groups = plan(rows);
+ if (![...groups.values()].some(({row,provider}) => row.durable_provider !== provider || row.durability !== 'durable')) return null;
+ const hash = crypto.createHash('sha256').update(String(config.hls.startupSeconds));
+ for (const [key, {row,provider}] of groups) hash.update(`${key}:${row.sha256}:${provider}\n`);
+ const recentJobs = await db.get("SELECT COUNT(*) AS n FROM media_jobs WHERE job_type = 'object.startup-tier' AND created_at >= ov_now_iso('-1 hour')");
+ while (queuedAt.length && queuedAt[0] <= now - 3600000) queuedAt.shift();
+ if (Math.max(Number(recentJobs.n),queuedAt.length) + pending >= config.hls.backfillPerHour) return null;
+ pending++;
+ let result;
+ try { result = await require('./queue').enqueue({appId:obj.app_id,type:'object.startup-tier',objectId:obj.id,params:{},
+ dedupeActive:true,idempotencyKey:`startup-tier:${obj.id}:${hash.digest('hex').slice(0,20)}`,createdBy:'system:startup-tier'});
+ if (result.created) queuedAt.push(now);
+ } finally { pending--; }
+ if (recent.size >= 1000) recent.delete(recent.keys().next().value);
+ recent.set(obj.id,now);
+ return result;
+}
+function validate({obj,params}) {
+ if (!obj || obj.kind === 'clip') throw new JobError('media.job.invalid','Startup placement requires a recording or media object',{permanent:true});
+ if (Object.keys(params || {}).length) throw new JobError('media.job.invalid','Startup placement takes no parameters',{permanent:true});
+ return {};
+}
+async function run(job, ctx) {
+ if (!config.hls.enabled || !config.hls.startupSeconds) return {skipped:'startup placement disabled'};
+ const obj = await model.getObject(job.object_id);
+ if (!obj || obj.app_id !== job.app_id || obj.lifecycle_status !== 'ready') throw new JobError('not_ready','Recording is not ready',{permanent:true});
+ const rows = await rowsOf(obj), groups = plan(rows);
+ const vs = require('../vod/vod-storage');
+ if (!providersReady()) throw new JobError('provider_unavailable','R2 and B2 must both be healthy and eligible',{retryAfterS:300});
+ const ids = [...new Set([obj.id,...rows.map(r=>r.object_id)])].sort();
+ for (const id of ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Held recording stays in place',{retryAfterS:21600});
+ const originals = [];
+ for (const id of ids) originals.push(...await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [id]));
+ const moving = originals.filter(r=>groups.has(identity(r)) && (r.durable_provider !== groups.get(identity(r)).provider || r.durability !== 'durable'));
+ if (!moving.length) return {moved:0,bytes:0};
+ const copies = new Map(), staged = [], updated = [];
+ let bytes = 0, committed = false;
+ try {
+ for (const row of moving) {
+ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('Cancelled');
+ const provider = groups.get(identity(row)).provider;
+ const copyId = `${provider}\n${row.key}`;
+ if (!copies.has(copyId)) {
+ const local = row.local_path && require('fs').existsSync(row.local_path);
+ const size = local ? require('fs').statSync(row.local_path).size
+ : Number((await vs.headObject(row.durable_provider,row.key) || {}).size);
+ if (!(size > 0) || bytes + size > MAX_BYTES) throw new JobError('startup_budget','Startup move exceeds 512 MiB or has no readable source',{permanent:true});
+ bytes += size;
+ staged.push({...row,local_path:null,durable_provider:provider});
+ if (local) await vs.uploadFile(provider,row.key,row.local_path,Number(row.seq)<=0?'video/mp4':'video/iso.segment',{signal:ctx.signal});
+ else if (row.durability === 'durable') await vs.copyBetweenProviders(row.durable_provider,provider,row.key);
+ else throw new JobError('segment_unreadable','No durable source',{retryAfterS:300});
+ const back = await vs.sha256Object(provider,row.key);
+ if (!back || back.sha256 !== (row.packed_object_id || row.sha256) || Number(back.size) !== size)
+ throw new JobError('copy_unverified','Startup copy failed size/SHA256 verification',{retryAfterS:300});
+ copies.set(copyId,true);
+ }
+ updated.push({...row,durable_provider:provider,durability:'durable'});
+ }
+ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('Cancelled');
+ await db.getDb().tx(async()=>{
+ // Same locks as timeline.replace and markPacked; the complete snapshot is checked before any row moves.
+ for (const id of ids) await db.get('SELECT pg_advisory_xact_lock(hashtext(?)) AS locked',[`media.timeline:${id}`]);
+ for (const id of ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Hold added during copy',{retryAfterS:21600});
+ for (const row of moving) {
+ const current = await timeline.byName(row.object_id,row.rendition,row.name);
+ if (!current || ['key','sha256','durable_provider','packed_object_id','local_path'].some(k=>current[k]!==row[k]))
+ throw new JobError('media.timeline.changed','Timeline changed during startup copy',{retryAfterS:60});
+ }
+ const queue = require('./queue');
+ await queue.enqueue({appId:queue.SYSTEM_APP,type:'storage.startup.cleanup',params:{rows:moving.map(r=>({object_id:r.object_id,durable_provider:r.durable_provider,key:r.key})),object_ids:ids},runAfterS:360,idempotencyKey:`startup-cleanup:${job.id}`,createdBy:'system:startup-tier'});
+ for (const row of updated) await db.run("UPDATE media_timeline SET durable_provider = ?, durability = 'durable', updated_at = ov_now() WHERE object_id = ? AND rendition = ? AND seq = ?",[row.durable_provider,row.object_id,row.rendition,row.seq]);
+ });
+ committed = true;
+ // Old presigned URLs (300 s) and joined-row caches (20 s) stay valid through a durable 360 s cleanup job.
+ return {moved:updated.length,bytes,hot_seconds:config.hls.startupSeconds,cleanup_after_s:360};
+ } finally {
+ if (!committed) await timeline.deleteBytes(staged).catch(()=>{});
+ }
+}
+const cleanup = {
+ lane:'light',maxAttempts:6,timeoutMs:300000,needsObject:false,
+ validate({appId,params}) {
+ if (appId !== require('./queue').SYSTEM_APP) throw new JobError('media.job.forbidden','Internal cleanup only',{permanent:true});
+ if (!params || !Array.isArray(params.rows) || params.rows.length > 10000 || !Array.isArray(params.object_ids))
+ throw new JobError('media.job.invalid','Cleanup rows required',{permanent:true});
+ return params;
+ },
+ async run(job) {
+ cleanup.validate({appId:job.app_id,params:job.params});
+ for (const id of job.params.object_ids) if (await model.isHeld(id)) throw new JobError('media.object.held','Held copies stay in place',{retryAfterS:21600});
+ const result=await timeline.deleteBytes(job.params.rows.map(r=>({durable_provider:r.durable_provider,key:r.key,local_path:null})));
+ if (result.failed.length) throw new JobError('cleanup_failed','Old segment copy deletion failed',{retryAfterS:600});
+ return {kept:result.kept.length,removed:job.params.rows.length-result.kept.length};
+ },
+};
+module.exports = {plan,queueOnce,playbackPreference,cleanup, spec:{lane:'heavy',maxAttempts:3,timeoutMs:30*60*1000,needsObject:true,validate,run}};
@@ -29,6 +29,8 @@ queue.register('object.waveform', require('./previews').waveform);
@@ -29,6 +29,8 @@ queue.register('object.waveform', require('./previews').waveform);
queue.register('object.sprite', require('./previews').sprite);
queue.register('object.sprite', require('./previews').sprite);
queue.register('object.cmaf', require('./cmaf').spec);
queue.register('object.cmaf', require('./cmaf').spec);
queue.register('object.pack', require('./pack').spec);
queue.register('object.pack', require('./pack').spec);
+queue.register('object.startup-tier', require('./startup-tier').spec);
+queue.register('storage.startup.cleanup', require('./startup-tier').cleanup);
queue.register('rendition.create', require('./rendition').spec);
queue.register('rendition.create', require('./rendition').spec);
queue.register('vod.finalize', require('./vod-finalize').spec);
queue.register('vod.finalize', require('./vod-finalize').spec);
queue.register('clip.cut', require('../vod/clip-jobs').spec);
queue.register('clip.cut', require('../vod/clip-jobs').spec);
@@ -1014,6 +1014,8 @@ publicRouter.get('/:id/master.m3u8', hlsRoute(async (req, res, { obj, query }) =
@@ -1014,6 +1014,8 @@ publicRouter.get('/:id/master.m3u8', hlsRoute(async (req, res, { obj, query }) =
// A missing rendition (F4) is queued lazily; only the ones whose rows exist are listed, so a client never follows
// A missing rendition (F4) is queued lazily; only the ones whose rows exist are listed, so a client never follows
// a variant that is not there yet.
// a variant that is not there yet.
await hls.queueRenditionsOnce(obj);
await hls.queueRenditionsOnce(obj);
+ // Existing recordings are warmed in the bounded worker queue; playlist delivery never waits for a copy.
+ require('../jobs/startup-tier').queueOnce(obj).catch(err => console.warn('[HLS] startup queue:', err.message));
res.type(HLS_TYPE).send(timeline.masterPlaylist(await hls.renditionsFor(obj, rows), { query }));
res.type(HLS_TYPE).send(timeline.masterPlaylist(await hls.renditionsFor(obj, rows), { query }));
}));
}));
@@ -1086,8 +1088,10 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
@@ -1086,8 +1088,10 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
const router = require('../placement/router');
const router = require('../placement/router');
const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: row.name,
const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: row.name,
purpose: 'playback', session: router.sessionFor(req, `${obj.id}/${rendition}`),
purpose: 'playback', session: router.sessionFor(req, `${obj.id}/${rendition}`),
- contentType: mime, expiresIn: 300, presign: !packed });
+ contentType: mime, expiresIn: 300, presign: !packed,
+ preferredProvider: require('../jobs/startup-tier').playbackPreference(row) });
for (const loc of decision.candidates || []) {
for (const loc of decision.candidates || []) {
+ headers['X-OV-Media-Provider'] = loc.provider;
if (loc.provider === 'local') {
if (loc.provider === 'local') {
if (!fs.existsSync(loc.key)) continue;
if (!fs.existsSync(loc.key)) continue;
require('./timeline-cache').served(loc.key);
require('./timeline-cache').served(loc.key);
@@ -1102,6 +1106,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
@@ -1102,6 +1106,7 @@ async function sendSegmentRow(req, res, { obj, closed }, row, rendition) {
continue;
continue;
}
}
res.set('Cache-Control', 'private, max-age=0');
res.set('Cache-Control', 'private, max-age=0');
+ res.set('X-OV-Media-Provider', loc.provider);
return res.redirect(302, url);
return res.redirect(302, url);
}
}
res.status(404).json({ error: 'Segment bytes unavailable' });
res.status(404).json({ error: 'Segment bytes unavailable' });
@@ -167,8 +167,10 @@ function tierOf(candidate, purpose) {
@@ -167,8 +167,10 @@ function tierOf(candidate, purpose) {
}
}
/** Rank the candidates: by tier, with a measured latency advantage (> TIE_BAND) allowed to promote. */
/** Rank the candidates: by tier, with a measured latency advantage (> TIE_BAND) allowed to promote. */
-function rankCandidates(candidates, { purpose = 'playback' } = {}) {
+function rankCandidates(candidates, { purpose = 'playback', preferredProvider = null } = {}) {
return [...candidates].sort((a, b) => {
return [...candidates].sort((a, b) => {
+ const ap = a.provider === preferredProvider, bp = b.provider === preferredProvider;
+ if (ap !== bp) return ap ? -1 : 1;
const ta = tierOf(a, purpose), tb = tierOf(b, purpose);
const ta = tierOf(a, purpose), tb = tierOf(b, purpose);
if (ta === tb) {
if (ta === tb) {
const ma = measuredMs(a.provider), mb = measuredMs(b.provider);
const ma = measuredMs(a.provider), mb = measuredMs(b.provider);
@@ -217,6 +219,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);
@@ -217,6 +219,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);
async function route(opts = {}) {
async function route(opts = {}) {
const purpose = opts.purpose || 'playback';
const purpose = opts.purpose || 'playback';
const session = opts.session || null;
const session = opts.session || null;
+ // A verified timeline names its opening/tail tier. This never admits a missing or unhealthy copy.
+ const preferredProvider = purpose === 'playback' && (opts.preferredProvider === 'r2' || opts.preferredProvider === 'b2') ? opts.preferredProvider : null;
const locs = (await locationsFor(opts)).filter((l) => l && l.state === 'present' && l.key);
const locs = (await locationsFor(opts)).filter((l) => l && l.state === 'present' && l.key);
// One candidate per provider: the first present location it has. A copy on an unhealthy (or avoided) provider
// One candidate per provider: the first present location it has. A copy on an unhealthy (or avoided) provider
@@ -234,7 +238,7 @@ async function route(opts = {}) {
@@ -234,7 +238,7 @@ async function route(opts = {}) {
}
}
const lastResort = candidates.length === 0 && degraded.length > 0;
const lastResort = candidates.length === 0 && degraded.length > 0;
if (lastResort) candidates = degraded;
if (lastResort) candidates = degraded;
- const ranked = rankCandidates(candidates, { purpose });
+ const ranked = rankCandidates(candidates, { purpose, preferredProvider });
// Sticky: same session -> same provider for STICKY_TTL_MS unless its copy/breaker is gone, or a better tier
// Sticky: same session -> same provider for STICKY_TTL_MS unless its copy/breaker is gone, or a better tier
// is available again. Stickiness stops latency flapping between copies of one tier; it never holds a viewer
// is available again. Stickiness stops latency flapping between copies of one tier; it never holds a viewer
@@ -244,7 +248,7 @@ async function route(opts = {}) {
@@ -244,7 +248,7 @@ async function route(opts = {}) {
let stickyHit = false;
let stickyHit = false;
if (sticky) {
if (sticky) {
const at = ranked.findIndex((c) => c.provider === sticky.provider);
const at = ranked.findIndex((c) => c.provider === sticky.provider);
- if (at >= 0 && tierOf(ranked[at], purpose) <= tierOf(ranked[0], purpose)) { ordered = [ranked[at], ...ranked.slice(0, at), ...ranked.slice(at + 1)]; stickyHit = true; }
+ if (at >= 0 && (!preferredProvider || ranked[0].provider !== preferredProvider || ranked[at].provider === preferredProvider) && tierOf(ranked[at], purpose) <= tierOf(ranked[0], purpose)) { ordered = [ranked[at], ...ranked.slice(0, at), ...ranked.slice(at + 1)]; stickyHit = true; }
else stickyByKey.delete(session);
else stickyByKey.delete(session);
}
}
@@ -261,7 +265,7 @@ async function route(opts = {}) {
@@ -261,7 +265,7 @@ async function route(opts = {}) {
const fallbacks = ordered.filter((c) => c !== primary).map((c) => c.provider);
const fallbacks = ordered.filter((c) => c !== primary).map((c) => c.provider);
const reason = primary
const reason = primary
- ? lastResort ? 'last-resort' : (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy')
+ ? lastResort ? 'last-resort' : primary.provider === preferredProvider ? 'timeline-tier' : (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy')
: 'no-healthy-copy';
: 'no-healthy-copy';
if (primary) stickyPut(session, primary.provider);
if (primary) stickyPut(session, primary.provider);
@@ -1,4 +1,6 @@
@@ -1,4 +1,6 @@
'use strict';
'use strict';
+// This fixture verifies local/remote packing explicitly; startup placement has its own move and routing tests.
+process.env.MEDIA_HLS_STARTUP_SECONDS = '0';
// Segment-native video, first slice (F3.1; docs/media-fabric.md §3): object.cmaf stream-copies a finished video into
// Segment-native video, first slice (F3.1; docs/media-fabric.md §3): object.cmaf stream-copies a finished video into
// CMAF segments on its own keyframes and writes one media_timeline row per segment; the HLS playlists come from the
// CMAF segments on its own keyframes and writes one media_timeline row per segment; the HLS playlists come from the
// rows (ENDLIST, the rows' durations); segmentAt answers the segment at an instant, at the boundaries too; a rerun is a
// rows (ENDLIST, the rows' durations); segmentAt answers the segment at an instant, at the boundaries too; a rerun is a
@@ -179,6 +179,20 @@ const assert = require('assert');
@@ -179,6 +179,20 @@ const assert = require('assert');
assert.strictEqual(decision.provider, null, 'never an unconfigured provider, even as a last resort');
assert.strictEqual(decision.provider, null, 'never an unconfigured provider, even as a last resort');
signals.reset(); router.reset();
signals.reset(); router.reset();
+ // A timeline's selected cloud tier beats its retained local cache and the old sticky choice.
+ providers.reset(); signals.reset(); router.reset();
+ providers.isConfigured = () => true;
+ const startupCopies = [{provider:'local',state:'present',key:'/tmp/cache'},{provider:'b2',state:'present',key:'tail'}];
+ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false});
+ assert.strictEqual(decision.provider,'local');
+ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false,preferredProvider:'b2'});
+ assert.strictEqual(decision.provider,'b2'); assert.strictEqual(decision.reason,'timeline-tier');
+ assert.strictEqual(decision.candidates[0].provider,'b2','metrics, sticky choice and served order agree');
+ for (let i=0;i<10;i++) signals.recordFailure('b2','presign',50);
+ decision = await router.route({locations:startupCopies,session:'startup-policy',presign:false,preferredProvider:'b2'});
+ assert.strictEqual(decision.provider,'local','an unhealthy preferred provider is still filtered out');
+ signals.reset(); router.reset();
+
providers.isConfigured = realIsConfigured;
providers.isConfigured = realIsConfigured;
console.log('placement-router: all checks passed');
console.log('placement-router: all checks passed');
process.exit(0);
process.exit(0);
new file mode 100644
new file mode 100644
@@ -0,0 +1,88 @@
@@ -0,0 +1,88 @@
+'use strict';
+const assert = require('assert');
+const crypto = require('crypto');
+process.env.MEDIA_HLS_ENABLED='1';
+process.env.MEDIA_HLS_STARTUP_SECONDS='180';
+(async()=>{
+ const db=require('../server/db/database'), timeline=require('../server/objects/timeline');
+ const model=require('../server/objects/model'), storage=require('../server/vod/vod-storage');
+ const startup=require('../server/jobs/startup-tier'), config=require('../server/config');
+ const obj={id:'startup-test',app_id:'live',kind:'vod',lifecycle_status:'ready'};
+ const queue=require('../server/jobs/queue');
+ const original={get:model.getObject,held:model.isHeld,enqueue:queue.enqueue};
+ model.getObject=async()=>obj; model.isHeld=async()=>false;
+ const blobs=new Map(), sha=b=>crypto.createHash('sha256').update(b).digest('hex');
+ const source=(seq,start)=>{
+ const bytes=Buffer.from(`segment-${seq}`), key=`startup/${seq}.m4s`; blobs.set(`b2:${key}`,bytes);
+ return {seq,name:seq?`${seq}.m4s`:'init.mp4',start_ms:start,end_ms:seq?start+4000:0,keyframe_ms:seq?start:null,
+ key,local_path:null,durable_provider:'b2',packed_object_id:null,byte_offset:null,byte_length:bytes.length,
+ sha256:sha(bytes),durability:'durable',init_name:null};
+ };
+ const rows=[source(0,0),source(1,176000),source(2,180000),source(3,220000)];
+ await timeline.replace(obj.id,'source',rows);
+ const native={};
+ for(const k of ['providerAvailable','providerConfigured','headObject','copyBetweenProviders','sha256Object','deleteObject'])native[k]=storage[k];
+ storage.providerAvailable=storage.providerConfigured=()=>true;
+ storage.headObject=async(p,k)=>blobs.has(`${p}:${k}`)?{size:blobs.get(`${p}:${k}`).length}:null;
+ storage.copyBetweenProviders=async(a,b,k)=>{blobs.set(`${b}:${k}`,blobs.get(`${a}:${k}`));return true;};
+ storage.sha256Object=async(p,k)=>{const b=blobs.get(`${p}:${k}`);return b?{size:b.length,sha256:sha(b)}:null;};
+ storage.deleteObject=async(p,k)=>{blobs.delete(`${p}:${k}`);return true;};
+ require('../server/jobs/queue').register('storage.startup.cleanup',startup.cleanup);
+ const ctx={signal:new AbortController().signal};
+ try {
+ let queuedByRead=null; queue.enqueue=async request=>{queuedByRead=request;return request;};
+ await startup.queueOnce(obj);
+ assert.strictEqual(queuedByRead.objectId,obj.id,'master queues its existing source through the real recording-part lookup');
+ queue.enqueue=original.enqueue;
+ const read=storage.sha256Object;
+ storage.sha256Object=async()=>({size:9,sha256:'wrong'});
+ await assert.rejects(startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx),/verification/);
+ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'failed copy leaves every original route');
+ assert(blobs.has('b2:startup/0.m4s')); assert(!blobs.has('r2:startup/0.m4s'),'uncommitted copy cleaned');
+ storage.sha256Object=read;
+ const copy=storage.copyBetweenProviders;
+ const cancelled=new AbortController();
+ storage.copyBetweenProviders=async(a,b,k)=>{await copy(a,b,k);cancelled.abort(new Error('Test cancelled'));};
+ await assert.rejects(startup.spec.run({id:'cancelled-job',app_id:'live',object_id:obj.id},{signal:cancelled.signal}),/Test cancelled/);
+ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'cancellation keeps original routes');
+ assert(!blobs.has('r2:startup/0.m4s'),'cancelled copies cleaned');
+ let raced=false;
+ storage.copyBetweenProviders=async(a,b,k)=>{
+ await copy(a,b,k);
+ if(!raced){raced=true;await db.run('UPDATE media_timeline SET packed_object_id = ? WHERE object_id = ? AND seq = 1',['concurrent-pack',obj.id]);}
+ };
+ await assert.rejects(startup.spec.run({id:'raced-job',app_id:'live',object_id:obj.id},ctx),/Timeline changed/);
+ assert((await timeline.list(obj.id)).every(r=>r.durable_provider==='b2'),'a concurrent pack is never overwritten');
+ await db.run('UPDATE media_timeline SET packed_object_id = NULL WHERE object_id = ?',[obj.id]);
+ storage.copyBetweenProviders=copy;
+ await timeline.replace('startup-alias','source',[rows[1]]);
+ const result=await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx);
+ assert.strictEqual(result.moved,2);
+ assert(blobs.has('b2:startup/0.m4s'),'old signed URLs stay readable until cleanup');
+ const queued=await db.get("SELECT * FROM media_jobs WHERE job_type = 'storage.startup.cleanup'");
+ assert(queued.run_after,'cleanup is durably delayed');
+ assert.throws(()=>startup.cleanup.validate({appId:'live',params:{rows:[],object_ids:[]}}),/Internal/);
+ const cleaned=await startup.cleanup.run({...queued,params:JSON.parse(queued.params)});
+ assert.strictEqual(cleaned.kept,1,'another timeline retains its old copy');
+ assert(blobs.has('b2:startup/1.m4s'));
+ assert(!blobs.has('b2:startup/0.m4s'));
+ assert(blobs.has('r2:startup/0.m4s'));
+ assert.deepStrictEqual((await timeline.list(obj.id)).map(r=>r.durable_provider),['r2','r2','b2','b2']);
+ assert.strictEqual((await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx)).moved,0,'idempotent');
+ const joined=[{...rows[1],object_id:'part1',rendition:'source',start_ms:179000,key:'chunk'},
+ {...rows[2],object_id:'part1',rendition:'source',start_ms:181000,key:'chunk'},
+ {...rows[1],object_id:'part2',rendition:'source',start_ms:220000,key:'later'}];
+ assert.deepStrictEqual([...startup.plan(joined).values()].map(x=>x.provider),['r2','b2'],'whole crossing chunk stays hot; part2 uses recording time');
+ assert.strictEqual(startup.playbackPreference({durability:'durable',durable_provider:'b2'}),'b2');
+ assert.strictEqual(startup.playbackPreference({durability:'local',durable_provider:null}),null);
+ config.hls.startupSeconds=0;
+ assert.strictEqual((await startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx)).skipped,'startup placement disabled');
+ config.hls.startupSeconds=180;
+ model.isHeld=async()=>true;
+ await assert.rejects(startup.spec.run({id:'startup-job-test',app_id:'live',object_id:obj.id},ctx),/Held/);
+ console.log('startup tier: verified R2 prefix/B2 tail, failed-copy rollback, idempotency, packed boundary, recording time, hold and off switch passed');
+ } finally {
+ Object.assign(storage,native);model.getObject=original.get;model.isHeld=original.held;queue.enqueue=original.enqueue;
+ await db.run('DELETE FROM media_timeline WHERE object_id IN (?, ?)',[obj.id,'startup-alias']); await db.close();
+ }
+})().catch(e=>{console.error(e);process.exitCode=1});
@@ -119,6 +119,7 @@ const { EventEmitter } = require('events');
@@ -119,6 +119,7 @@ const { EventEmitter } = require('events');
const res = { destroyed: false, writeHead(code, headers) { this.code = code; this.headers = headers; }, end(body) { this.body = body; } };
const res = { destroyed: false, writeHead(code, headers) { this.code = code; this.headers = headers; }, end(body) { this.body = body; } };
await require('../server/objects/routes').sendSegmentRow(req, res, { obj: { id: 'cache-packed' }, closed: false }, packedRow, 'source');
await require('../server/objects/routes').sendSegmentRow(req, res, { obj: { id: 'cache-packed' }, closed: false }, packedRow, 'source');
assert.strictEqual(res.code, 200);
assert.strictEqual(res.code, 200);
+ assert.strictEqual(res.headers['X-OV-Media-Provider'], 'b2', 'packed responses identify their backing tier');
assert.deepStrictEqual(res.body, Buffer.from('cde'), 'the segment route reads the packed range from durable storage');
assert.deepStrictEqual(res.body, Buffer.from('cde'), 'the segment route reads the packed range from durable storage');
} finally {
} finally {
router.route = realRoute; router.sessionFor = realSession; storage.presignGet = realPresign; global.fetch = realFetch;
router.route = realRoute; router.sessionFor = realSession; storage.presignGet = realPresign; global.fetch = realFetch;