openvibers/openvibe.media Commits
Media: losing a provider fails over (presigns no longer mask an outage, failed health checks and shield fetches move reads to another copy); DVR seek proven with real ffmpeg (#106)
Changes
@@@ -40,6 +40,10 @@ server {proxy_cache_lock_age 10s;proxy_cache_use_stale error timeout updating http_500 http_502 http_503 http_504;proxy_cache_background_update on;++ # Losing B2 is not a 502 for the viewer: a first slice B2 cannot serve, with nothing stale cached, goes back to++ # Node, which records the failure and sends the read to another copy (server/placement/shield.js failedOf).++ proxy_intercept_errors on;++ error_page 502 503 504 = @shield_failed_b2;# Node's answer owns the viewer-facing headers (Content-Type, Content-Disposition, Cache-Control survive the# X-Accel redirect): B2's copies, and its request ids, are dropped. The security headers Node sets do not survive# the redirect, so they are set here.@@@ -60,6 +64,20 @@ server {include snippets/media-shield-b2-upstream.conf;}++ # The read the shield could not fetch, back to Node as the viewer sent it, marked with the provider that failed.++ location @shield_failed_b2 {++ proxy_pass http://127.0.0.1:4100$request_uri;++ proxy_http_version 1.1;++ proxy_set_header Host $host;++ proxy_set_header X-Media-Shield-Host $server_name;++ proxy_set_header X-Media-Shield-Failed b2;++ proxy_set_header X-Forwarded-Host "";++ proxy_set_header X-Real-IP $remote_addr;++ proxy_set_header X-Forwarded-For $remote_addr;++ proxy_set_header CF-Connecting-IP $remote_addr;++ proxy_set_header X-Forwarded-Proto $scheme;++ }++# Everything else is Node, exactly as on openvibe.media (client-IP headers from $remote_addr only), plus the one# header that tells Node this is the edge: nginx sets it from its own server name, so a client cannot forge it.location / {@@@ -67,6 +85,7 @@ server {proxy_http_version 1.1;proxy_set_header Host $host;proxy_set_header X-Media-Shield-Host $server_name;++ proxy_set_header X-Media-Shield-Failed "";proxy_set_header X-Forwarded-Host "";proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $remote_addr;@@@ -28,6 +28,7 @@ server {# Only the edge host's block may say "edge" (placement/shield.js): cleared here so a client cannot send it# through Cloudflare; X-Forwarded-Host is cleared too (Express's req.hostname would honour it).proxy_set_header X-Media-Shield-Host "";++ proxy_set_header X-Media-Shield-Failed "";proxy_set_header X-Forwarded-Host "";proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $remote_addr;@@@ -483,6 +483,14 @@ Cache-Control, nosniff and noindex, and no B2 header. VOD bytes carry no CORS hethrough the shared cache (anyone with the link may read them; `/o` answers `public, max-age=3600` for them), unlistedVODs answer `private, max-age=0`.- **Private and sandbox bytes never enter the shield.**++- **Losing a provider fails over (2026-10-14).** A failed live health check (the HEAD each minute) moves reads to the++ next copy at once; a presign is local signing, so its successes no longer dilute the breaker's error rate or close it++ (during an outage every viewer's presign succeeded and the breaker never opened). A first slice the shield cannot++ fetch from B2, with nothing stale cached, goes back to Node through the edge's `@shield_failed_b2` location++ (`X-Media-Shield-Failed: b2`, cleared on every location a client reaches); Node records the failure and serves another++ copy. A copy whose only provider is unhealthy is still served as the last resort (`reason: last-resort`): an outage++ never makes a one-copy object vanish, and a corrupt, missing or unverified copy is never served at all. A second CDN++ path is F5's (Bunny needs an account).- **Sizing:** one disk with recordings: `max_size=5g`, `min_free=20g` (the recording guardian warns at 15 GB), `inactive=24h`.- **Purge:** deleting a shielded object refreshes each of its slices (four at a time) through a loopback-only listener(`127.0.0.1:8479`, `allow 127.0.0.1; deny all`, `proxy_cache_bypass`), so the provider's 404 replaces the cached bytes@@@ -994,9 +994,12 @@ publicRouter.get('/:id', async (req, res) => {// providers, and ranks by preference tier with the measured 5-minute EWMA latency and the// design's ±15 % tie band: a verified R2 copy (the hot cache) for playback, then local, R2, B2.const router = require('../placement/router');++ // The shield could not fetch this read (nginx handed it back): another copy, and no shield for this one.++ const shieldFailed = require('../placement/shield').failedOf(req);const decision = await router.route({object: obj, purpose: 'playback', session: router.sessionFor(req, obj.id),contentType: mime, contentDisposition: headers['Content-Disposition'], expiresIn: 300,++ exclude: shieldFailed ? [shieldFailed] : [],});// Walk the router's ranked copies: stream a local one, redirect to a presigned remote one.// A stale `present` local row whose file is gone falls through to the next copy.@@@ -1011,11 +1014,12 @@ publicRouter.get('/:id', async (req, res) => {// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through its// slice cache, with a long internal presign that never reaches the viewer (only on the DNS-only edge host).const shield = require('../placement/shield');-- if (shield.enabled(provider) && !sandbox && obj.visibility !== 'private' && !shield.onEdge(req)) {++ const shielded = shield.enabled(provider) && provider !== shieldFailed;++ if (shielded && !sandbox && obj.visibility !== 'private' && !shield.onEdge(req)) {const edge = shield.toEdge(req, provider);if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }}-- if (shield.enabled(provider) && shield.onEdge(req) && !sandbox && obj.visibility !== 'private') {++ if (shielded && shield.onEdge(req) && !sandbox && obj.visibility !== 'private') {const inner = await require('../vod/vod-storage').presignGet(provider, loc.key, shield.SHIELD_TTL_SECONDS, { contentType: mime, contentDisposition: headers['Content-Disposition'] }).catch(() => null);const t = inner && shield.target({ provider, url: inner, visibility: obj.visibility, sandbox, edge: shield.edgeOf(req) });if (t) return shield.send(res, t, headers);@@@ -290,6 +290,14 @@ function observe(provider, op, ok, ms) {} catch { /* */ }}++/** True while the provider's last live health check (or capability probe) failed and is recent: the router routes++ * reads to another copy until a check passes again. A check older than LIVE_HEALTH_FRESH_MS says nothing. */++const LIVE_HEALTH_FRESH_MS = 5 * 60_000;++function liveUnhealthy(name) {++ const at = state.lastHealthCheck[name];++ return state.lastHealthError[name] != null && at != null && Date.now() - at < LIVE_HEALTH_FRESH_MS;++}++function healthStatus() {const out = {};for (const name of vodStorage().REMOTE_PROVIDERS) {@@@ -340,8 +348,9 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};module.exports = {CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,isConfigured, providerClasses, eligibleClasses, measuredClasses, placementGate, gateMode, capabilities,-- probe, probeAll, liveHealth, healthStatus,++ probe, probeAll, liveHealth, liveUnhealthy, healthStatus,startHealthLoop, stop, reset,// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK)._setCapabilities: setCapabilities,++ _setHealth: (name, error) => { state.lastHealthCheck[name] = Date.now(); state.lastHealthError[name] = error || null; },};@@@ -11,8 +11,10 @@* 1. Resolve locations for the object (`locationsFor(opts)` accepts an `object`,* a `vod` / `clip` row, or an explicit list).* 2. Filter to `state === 'present'` rows only. Non-present rows are NEVER routed to.-- * 3. Filter to providers that are configured, have passed their capability probe and are-- * not breaker-open.++ * 3. Filter to providers that are configured and have passed their capability probe. Of those, a provider++ * that is breaker-open, whose last live health check failed, or that this read was told to avoid (`exclude`:++ * the origin shield could not fetch from it) is only a last resort: served when no other copy is left, so++ * losing a provider moves reads to the next copy and never makes a one-copy object vanish.* 4. Rank by preference tier, with the measured 5-minute EWMA latency breaking ties:* playback : R2 with a verified present copy (popularity promotion put it there to move* reads off the host; promoting a VOD also frees its local copy) > local > R2 > B2@@@ -113,17 +115,25 @@ function withOwnCopy(rows, provider, key, extra = {}) {return [...rows.filter((r) => r.provider !== provider), { provider, state: 'present', key, ...extra }];}--/** A copy is healthy when: configured, breaker closed, and eligible for at least one provider class-- * (a failed capability probe removes every class it needed). `passed === null` = probe not run yet. */--function isHealthy(provider, { purpose = 'playback' } = {}) {++/** A copy can be served at all when: configured and eligible for at least one provider class (a failed capability++ * probe removes every class it needed). `passed === null` = probe not run yet. */++function isEligible(provider) {if (provider === 'local') return true;if (!providers.isConfigured(provider)) return false;-- if (signals.breakerOpen(provider)) return false;const caps = providers.capabilities(provider);if (caps.passed === null) return providers.providerClasses(provider).length > 0;return providers.eligibleClasses(provider).length > 0;}++/** A copy is healthy when eligible, its breaker is closed and its last live health check did not fail. */++function isHealthy(provider, { purpose = 'playback' } = {}) { // eslint-disable-line no-unused-vars++ if (!isEligible(provider)) return false;++ if (provider === 'local') return true;++ if (signals.breakerOpen(provider)) return false;++ try { if (providers.liveUnhealthy(provider)) return false; } catch { /* a providers stub without health */ }++ return true;++}++/** Measured 5-minute EWMA (ms) of a network round-trip op (head, then get), or null when the* provider has no such samples. Presign is excluded: it is a local SigV4 operation, so its* EWMA is host CPU noise, not a network signal, and must not rank providers. */@@@ -197,7 +207,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);/*** The only function every read path uses.* opts: { object | vod | clip | locations, objectId?, purpose, session?, segment?, range?, presign?, expiresIn?,-- * contentType?, contentDisposition? }++ * contentType?, contentDisposition?, exclude? }++ * `exclude` names providers this read should avoid (the shield just failed to fetch from one): last resort only.* `objectId` names the object a demand hit belongs to when the caller passes no object/vod/clip row (a segment* read: the route already resolved its row and passes explicit locations); `segment` is the timeline segment served.* returns: { provider, key, url, reason, fallbacks, candidates } — `fallbacks` are provider names, ranked;@@@ -208,15 +219,21 @@ async function route(opts = {}) {const session = opts.session || null;const locs = (await locationsFor(opts)).filter((l) => l && l.state === 'present' && l.key);-- // One candidate per provider: the first present location it has.++ // One candidate per provider: the first present location it has. A copy on an unhealthy (or avoided) provider++ // waits as a last resort: served only when no healthy copy is left.++ const avoid = new Set(Array.isArray(opts.exclude) ? opts.exclude : []);const seen = new Set();-- const candidates = [];++ let candidates = [];++ const degraded = [];for (const l of locs) {if (seen.has(l.provider)) continue;-- if (!isHealthy(l.provider, { purpose })) continue;++ if (!isEligible(l.provider)) continue;seen.add(l.provider);-- candidates.push(l);++ if (isHealthy(l.provider, { purpose }) && !avoid.has(l.provider)) candidates.push(l);++ else degraded.push(l);}++ const lastResort = candidates.length === 0 && degraded.length > 0;++ if (lastResort) candidates = degraded;const ranked = rankCandidates(candidates, { purpose });// Sticky: same session -> same provider for STICKY_TTL_MS unless its copy/breaker is gone, or a better tier@@@ -244,7 +261,7 @@ async function route(opts = {}) {const fallbacks = ordered.filter((c) => c !== primary).map((c) => c.provider);const reason = primary-- ? (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy')++ ? lastResort ? 'last-resort' : (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy'): 'no-healthy-copy';if (primary) stickyPut(session, primary.provider);@@@ -284,6 +301,7 @@ function incMetric(name, labels) { try { _inc(name, labels); } catch { /* */ } }module.exports = {route,isHealthy,++ isEligible,rankCandidates,tierOf,locationsFor,@@@ -32,6 +32,13 @@* edge, where the bytes come from the shield* target({ provider, url, visibility, sandbox, edge }) → '/_media_shield/b2/<bucket>/<key>?<signed query>' | null* send(res, target, headers) → the 200 + X-Accel-Redirect answer++ * failedOf(req) → the provider the shield could not fetch this read from, or null:++ * nginx hands a failed first slice (502/503/504 with nothing stale++ * cached) back to Node with `X-Media-Shield-Failed`, set only by the++ * edge's @shield_failed location (every other location clears it).++ * The route records the failure, skips the shield and asks the++ * router for another copy (`exclude`), so losing the provider moves++ * the read to the next copy instead of answering 502.* await purge({ provider, key, size, presign }) → refreshes every cached slice of a deleted or replaced object*/'use strict';@@@ -62,6 +69,16 @@ function toEdge(req, provider) {return p.startsWith('/') && !p.startsWith('//') ? `https://${edge}${p}` : null;}++const FAILED_HEADER = 'x-media-shield-failed'; // set by the edge's @shield_failed location only (nginx)++/** The shielded provider whose fetch failed for this read, on the edge only (a client cannot reach it with the header). */++function failedOf(req) {++ if (!onEdge(req)) return null;++ const v = String((typeof req.get === 'function' && req.get(FAILED_HEADER)) || '').trim().toLowerCase();++ if (!v || !enabled(v)) return null;++ try { require('./signals').recordFailure(v, 'get', 0); } catch { /* */ }++ return v;++}++/** The internal location for one presigned URL, or null when this read must not go through the shield. `edge` is* edgeOf(req): the shield answers only on MEDIA_SHIELD_HOST (a DNS-only edge name), never through Cloudflare. */function target({ provider, url, visibility = 'public', sandbox = false, edge = null }) {@@@ -113,4 +130,4 @@ async function purge({ provider, key, size, presign, fetchImpl = globalThis.fetcreturn { purged, failed, slices };}--module.exports = { target, send, purge, enabled, providers, onEdge, edgeOf, toEdge, SLICE_BYTES, SHIELD_TTL_SECONDS, B2_UPSTREAM_HOST, EDGE_HEADER };++module.exports = { target, send, purge, enabled, providers, onEdge, edgeOf, toEdge, failedOf, SLICE_BYTES, SHIELD_TTL_SECONDS, B2_UPSTREAM_HOST, EDGE_HEADER, FAILED_HEADER };@@@ -12,7 +12,10 @@** The breaker opens when the 1-minute error rate exceeds 50% over at least 5 samples,* or the 5-minute p95 exceeds 5 s. Half-open after 60 s: one trial; on success it-- * closes, on failure it re-opens. `providerHealthy` in vod-storage.js reads this so a++ * closes, on failure it re-opens. A presign is local SigV4 signing: its successes say nothing about the++ * provider, so they never count toward the error rate or p95 and never close a breaker (otherwise every++ * viewer's presign during an outage diluted the failing HEADs, and the breaker never opened).++ * A failed presign still counts: a copy that cannot be signed cannot be served. `providerHealthy` in vod-storage.js reads this so a* dead provider is skipped without waiting for the next boot probe.** In-memory only: signals are observed on the actual read path and the next boot@@@ -43,6 +46,9 @@ const listeners = new Set(); // (provider) => void — fired when a breakerfunction key(provider, op) { return `${provider}:${op || 'op'}`; }++// Ops that never touch the provider: their successes are not evidence of its health.++const LOCAL_OPS = new Set(['presign']);++function getSignal(provider, op) {const k = key(provider, op);let s = state.get(k);@@@ -74,8 +80,10 @@ function ewmaFor(provider) {function push(provider, op, ok, ms) {const s = getSignal(provider, op);if (ok && ms > 0) s.ewmaMs = s.ewmaMs ? (EWMA_ALPHA * ms + (1 - EWMA_ALPHA) * s.ewmaMs) : ms;-- s.samples.push({ t: nowMs(), ok: !!ok, ms: Number(ms) || 0 });++ const local = !!ok && LOCAL_OPS.has(op);++ s.samples.push({ t: nowMs(), ok: !!ok, ms: Number(ms) || 0, local });if (s.samples.length > SAMPLE_LIMIT) s.samples.splice(0, s.samples.length - SAMPLE_LIMIT);++ if (local) return;if (ok) onSuccess(provider); else onFailure(provider);// Slow success (high p95) is itself a degradation signal: check every push.const stats5m = statsFor(provider, null, '5m');@@@ -92,7 +100,7 @@ function statsFor(provider, _opUnused, windowKey) {const cutoff = nowMs() - windowMs;let total = 0, errors = 0, latencies = [];for (const sample of allSamplesFor(provider)) {-- if (sample.t < cutoff) continue;++ if (sample.t < cutoff || sample.local) continue;total++;if (!sample.ok) errors++; else latencies.push(sample.ms);}@@@ -192,13 +192,15 @@ async function serveMediaRecord(kind, record, req, res) {// requests are handled natively by the object store.if (!filePath && vodStorage.isRemote(record)) {const router = require('../placement/router');-- const plan = await vodStorage.resolvePlayback(record, { session: router.sessionFor(req, `${kind}:${record.id}`) });++ // The shield could not fetch this read (nginx handed it back): another copy, and no shield for this one.++ const shieldFailed = require('../placement/shield').failedOf(req);++ const plan = await vodStorage.resolvePlayback(record, { session: router.sessionFor(req, `${kind}:${record.id}`), exclude: shieldFailed ? [shieldFailed] : [] });if (plan?.kind === 'redirect') {// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through// its slice cache with a long internal presign (only on the DNS-only edge host); everything else keeps the// short-lived 302.const shield = require('../placement/shield');-- const shieldable = shield.enabled(plan.provider) && visibility !== 'private' && !record.is_recording;++ const shieldable = shield.enabled(plan.provider) && plan.provider !== shieldFailed && visibility !== 'private' && !record.is_recording;const edge = shieldable && shield.toEdge(req, plan.provider);if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }if (shieldable && shield.onEdge(req)) {@@@ -525,9 +525,9 @@ async function presignGet(provider, key, expiresInSeconds = 900, overrides = {})* preferred when the file is on disk, R2 over B2 for playback when both are present, and B2* as the canonical fallback. The router returns the chosen provider; we presign it and answer.*/--async function resolvePlayback(vod, { session = null, range = null } = {}) {++async function resolvePlayback(vod, { session = null, range = null, exclude = [] } = {}) {const router = require('../placement/router');-- const decision = await router.route({ vod, purpose: 'playback', session, range });++ const decision = await router.route({ vod, purpose: 'playback', session, range, exclude });if (decision.provider && decision.provider !== 'local') {if (decision.url) return { kind: 'redirect', url: decision.url, provider: decision.provider, fallbacks: decision.fallbacks };// The router met a provider it could not rank-presign (the client is unavailable). Try thennew file mode 100644@@@ -0,0 +1,123 @@++'use strict';++// Plan §9 gate: seek works during an active DVR. A recording still being written is a fragmented MP4++// (frag_keyframe+empty_moov, as the recorder writes it); Live's DVR player reads it through /v/:id, which++// serves the live seek snapshot (remuxForLiveSeeking: a copy-remux with the whole index at the front). Here++// with real ffmpeg: the snapshot of a recording mid-stream answers ranged reads from any offset with no-cache,++// carries its index before its media, and a player can seek to a late moment in it; after the recording grows,++// the next snapshot reaches the new live edge; once finalized, /v/:id serves the recording itself again.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const { spawnSync } = require('child_process');++++(async () => {++ // Like cmaf-transcode.test.js: real encodes need ffmpeg/ffprobe, which CI may not have.++ if (spawnSync('ffmpeg', ['-version']).status !== 0 || spawnSync('ffprobe', ['-version']).status !== 0) {++ console.log('dvr seek: skipped (ffmpeg not found)');++ process.exit(0);++ }++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-dvr-seek-'));++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++ process.env.VOD_PATH = dir('vods');++ process.env.CLIPS_PATH = dir('clips');++ process.env.FILES_PATH = dir('files');++ process.env.OBJECTS_PATH = dir('objects');++ process.env.THUMBNAILS_PATH = dir('thumbnails');++ process.env.PASTES_PATH = dir('pastes');++ process.env.MEDIA_PUBLIC_URL = 'https://media.test';++++ const ffmpeg = (...args) => {++ const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args], { encoding: 'utf8' });++ assert.strictEqual(r.status, 0, r.stderr);++ };++ const duration = (file) => Number(spawnSync('ffprobe', ['-v', 'error', '-show_entries', 'format=duration', '-of', 'csv=p=0', file], { encoding: 'utf8' }).stdout.trim());++ // A recording `seconds` long, written the way the recorder writes a growing MP4 (a keyframe every second).++ const record = (file, seconds) => ffmpeg('-f', 'lavfi', '-i', `testsrc=duration=${seconds}:size=160x90:rate=10`,++ '-f', 'lavfi', '-i', `sine=duration=${seconds}`, '-c:v', 'mpeg4', '-g', '10', '-c:a', 'aac',++ '-movflags', '+frag_keyframe+empty_moov+default_base_moof', '-f', 'mp4', file);++++ const db = require('../server/db/database');++ const tools = require('../server/vod/media-tools');++ const express = require('express');++++ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });++ const rec = path.join(process.env.VOD_PATH, 'vod-live-7.mp4');++ record(rec, 4);++ await db.run(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, visibility, is_recording) OVERRIDING SYSTEM VALUE++ VALUES (7, 'live', 5, 'Still live', ?, 1, 'public', 1)`, [rec]);++++ const app = express();++ app.use('/', require('../server/public/routes'));++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${server.address().port}`;++ const get = async (range) => {++ const res = await fetch(`${base}/v/7`, { headers: range ? { range } : {}, redirect: 'manual' });++ return { status: res.status, headers: res.headers, body: Buffer.from(await res.arrayBuffer()) };++ };++ // A pass is paced (the next may start only after 9× the last one's run time): retry until one runs.++ const snapshot = async () => {++ for (let i = 0; i < 200; i++) {++ if (await tools.remuxForLiveSeeking(rec)) return true;++ await new Promise((r) => setTimeout(r, 100));++ }++ return false;++ };++ const ok = [];++ const check = async (name, fn) => { await fn(); ok.push(name); };++++ try {++ const sidecar = tools.seekableSidecarPath(rec);++ await check('mid-recording, /v/:id serves the seek snapshot: ranged, uncached, its index before its media', async () => {++ assert.ok(await snapshot(), 'the first snapshot ran');++ const size = fs.statSync(sidecar).size;++ const whole = await get('bytes=0-');++ assert.strictEqual(whole.status, 206);++ assert.match(whole.headers.get('cache-control'), /no-cache/);++ assert.strictEqual(whole.headers.get('content-range'), `bytes 0-${size - 1}/${size}`);++ assert.ok(whole.body.equals(fs.readFileSync(sidecar)), 'the snapshot, not the growing file');++ const moov = whole.body.indexOf('moov'), mdat = whole.body.indexOf('mdat');++ assert.ok(moov > 0 && mdat > moov, `the index sits before the media (moov ${moov}, mdat ${mdat})`);++ assert.strictEqual(whole.body.indexOf('moof'), -1, 'one whole index, not fragments a player must walk');++ // A seek lands mid-file: the player asks for a range starting there.++ const mid = Math.floor(size / 2);++ const part = await get(`bytes=${mid}-${mid + 999}`);++ assert.strictEqual(part.status, 206);++ assert.strictEqual(part.headers.get('content-range'), `bytes ${mid}-${mid + 999}/${size}`);++ assert.ok(part.body.equals(whole.body.subarray(mid, mid + 1000)));++ });++++ await check('a player seeks to a late moment of the snapshot it was served', async () => {++ const served = path.join(tmp, 'served.mp4');++ fs.writeFileSync(served, (await get()).body);++ assert.ok(Math.abs(duration(served) - 4) < 0.5, `the snapshot covers what was recorded: ${duration(served)} s`);++ const r = spawnSync('ffprobe', ['-v', 'error', '-read_intervals', '3.5%+0.5', '-select_streams', 'v', '-show_entries', 'frame=pts_time', '-of', 'csv=p=0', served], { encoding: 'utf8' });++ assert.strictEqual(r.status, 0, r.stderr);++ const times = r.stdout.trim().split('\n').map(Number).filter(Number.isFinite);++ assert.ok(times.length > 0 && times[0] >= 2.9 && times[0] < 3.6, `seeking to 3.5 s decodes from its keyframe (3 s), not the start: ${times.slice(0, 3)}`);++ });++++ await check('as the recording grows, the next snapshot reaches the new live edge', async () => {++ record(rec, 8);++ assert.ok(await snapshot(), 'the next snapshot ran');++ const served = path.join(tmp, 'served-later.mp4');++ fs.writeFileSync(served, (await get()).body);++ assert.ok(Math.abs(duration(served) - 8) < 0.5, `the later snapshot covers 8 s: ${duration(served)} s`);++ });++++ await check('once finalized, /v/:id serves the recording itself, cacheable again', async () => {++ await db.run('UPDATE vods SET is_recording = 0 WHERE id = 7');++ const res = await get('bytes=0-');++ assert.strictEqual(res.status, 206);++ assert.ok(res.body.equals(fs.readFileSync(rec)), 'the recording, not the snapshot');++ assert.doesNotMatch(res.headers.get('cache-control') || '', /no-cache/);++ });++ } finally {++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ }++ console.log(`dvr seek: ${ok.length} checks passed`);++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -152,6 +152,33 @@ const assert = require('assert');assert.notStrictEqual(router.sessionFor(reqA, 'vod:1'), router.sessionFor(reqA, 'vod:2'));assert.ok(!router.sessionFor(reqA, 'vod:1').includes('198.51'), 'the key is a hash, not the address');++ // Losing a provider fails over (plan §9 gate). Its failed live health check (a HEAD each minute) moves reads to++ // the next copy at once; a read the origin shield could not fetch avoids that provider; and a copy whose only++ // provider is unhealthy is still served as the last resort: an outage never makes a one-copy object vanish.++ router.reset(); signals.reset();++ const both = () => [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }];++ providers._setHealth('r2', 'r2 HeadBucket failed: TimeoutError');++ decision = await router.route({ locations: both(), presign: false });++ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'canonical-fallback'], `r2 failed its health check → b2: ${JSON.stringify(decision)}`);++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }], presign: false });++ assert.deepStrictEqual([decision.provider, decision.reason], ['r2', 'last-resort'], `the only copy is still served: ${JSON.stringify(decision)}`);++ providers._setHealth('r2', null);++ decision = await router.route({ locations: both(), presign: false });++ assert.strictEqual(decision.provider, 'r2', 'a passing check puts it back');++ decision = await router.route({ locations: both(), presign: false, exclude: ['r2'] });++ assert.deepStrictEqual([decision.provider, decision.fallbacks], ['b2', []], `a provider the shield could not fetch from is avoided: ${JSON.stringify(decision)}`);++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }], presign: false, exclude: ['b2'] });++ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'last-resort'], 'avoided, but the last copy');++ for (let i = 0; i < 10; i++) signals.recordFailure('b2', 'head', 50);++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }], presign: false });++ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'last-resort'], 'an open breaker on the only copy: last resort, not a 404');++ decision = await router.route({ locations: [{ provider: 'b2', state: 'corrupt', key: 'k' }, { provider: 'r2', state: 'pending', key: 'k' }], presign: false });++ assert.strictEqual(decision.provider, null, 'never a corrupt or unverified copy, even as a last resort');++ providers.isConfigured = (name) => name === 'local' || name === 'b2';++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }], presign: false });++ assert.strictEqual(decision.provider, null, 'never an unconfigured provider, even as a last resort');++ signals.reset(); router.reset();++providers.isConfigured = realIsConfigured;console.log('placement-router: all checks passed');process.exit(0);@@@ -47,6 +47,26 @@ const assert = require('assert');for (const c of [edgeConf, mainConf]) assert.match(c, /proxy_set_header X-Forwarded-Host "";/);for (const d of ['client_max_body_size 4096m;', 'proxy_read_timeout 3600s;', 'location ~* ^/metrics(/|$) { return 404; }']) assert.ok(edgeConf.includes(d), `the edge block has ${d}`);assert.match(conf('media-shield.http.conf'), /allow 127\.0\.0\.1;\s*deny all;/, 'only loopback may refresh the cache');++ // Losing B2 fails over: a first slice B2 cannot serve (nothing stale cached) goes back to Node, marked by a header++ // only that named location sets; every location a client reaches clears it.++ const shieldBlock = edgeConf.slice(edgeConf.indexOf('location /_media_shield/b2/'), edgeConf.indexOf('location @shield_failed_b2'));++ assert.match(shieldBlock, /proxy_intercept_errors on;/);++ assert.match(shieldBlock, /error_page 502 503 504 = @shield_failed_b2;/);++ const failedBlock = edgeConf.slice(edgeConf.indexOf('location @shield_failed_b2'));++ assert.match(failedBlock, /proxy_pass http:\/\/127\.0\.0\.1:4100\$request_uri;/, 'the original request goes back to Node');++ assert.match(failedBlock, /proxy_set_header X-Media-Shield-Failed b2;/);++ assert.match(failedBlock, /proxy_set_header X-Media-Shield-Host \$server_name;/);++ assert.match(failedBlock, /proxy_set_header X-Real-IP \$remote_addr;/, 'client-IP headers from $remote_addr only');++ for (const c of [edgeConf, mainConf]) assert.match(c.slice(c.indexOf(' location / {')), /proxy_set_header X-Media-Shield-Failed "";/, 'a client cannot claim a shield failure');++ // failedOf(): only on the edge, only a shielded provider, and the failure feeds the provider's signals.++ const signals = require('../server/placement/signals');++ signals.reset();++ const hdr = (h) => ({ get: (k) => h[k.toLowerCase()] });++ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-host': 'edge.openvibe.media', 'x-media-shield-failed': 'b2' })), 'b2');++ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-failed': 'b2' })), null, 'not on the edge: ignored');++ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-host': 'edge.openvibe.media', 'x-media-shield-failed': 'r2' })), null, 'r2 is not shielded');++ assert.strictEqual(signals.snapshot().providers.b2.windows['1m'].errors, 1, 'one failed get recorded');++ signals.reset();// send(): the route's headers plus the internal redirect, an empty 200.const res = { h: {}, set(k, v) { this.h[k] = v; }, status(c) { this.code = c; return this; }, end() { this.ended = true; return this; } };@@@ -30,7 +30,7 @@ const assert = require('assert');// p95 latency on a different provider drives the second open rule.signals.reset();-- for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'presign', 6000); // 6 s, over the 5 s threshold++ for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'head', 6000); // 6 s, over the 5 s thresholdassert.strictEqual(signals.breakerOpen('r2'), true, 'breaker opens when 5m p95 > 5 s');// Half-open: after 60 s exactly one trial is allowed; a success closes the breaker, a failure@@@ -42,7 +42,7 @@ const assert = require('assert');assert.strictEqual(signals.breakerState('b2'), 'half-open', 'half-open 60 s after opening');assert.strictEqual(signals.halfOpenClaim('b2'), true, 'the one trial is claimed');assert.strictEqual(signals.halfOpenClaim('b2'), false, 'a concurrent second trial is refused');-- signals.recordSuccess('b2', 'presign', 80);++ signals.recordSuccess('b2', 'head', 80);assert.strictEqual(signals.breakerState('b2'), 'closed', 'a successful trial closes the breaker');for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);@@@ -54,9 +54,9 @@ const assert = require('assert');// Snapshot exposes both providers' state, with p50 and p95 per window (the admin view and /metrics).signals.reset();-- signals.recordSuccess('b2', 'presign', 100);-- signals.recordSuccess('r2', 'presign', 100);-- signals.recordSuccess('r2', 'presign', 400);++ signals.recordSuccess('b2', 'head', 100);++ signals.recordSuccess('r2', 'head', 100);++ signals.recordSuccess('r2', 'get', 400);const snap = signals.snapshot();assert.ok(snap.providers.b2 && snap.providers.r2);assert.ok(snap.breakers.b2 && snap.breakers.r2);@@@ -67,6 +67,23 @@ const assert = require('assert');}assert.strictEqual(snap.providers.r2.windows['5m'].total, 2, 'both r2 samples in the 5-minute window');++ // A presign is local signing, not a provider round-trip: during an outage every viewer's presign succeeds while++ // the HEADs fail. Those successes must not dilute the error rate, feed the p95, or close an open breaker.++ signals.reset();++ for (let i = 0; i < 50; i++) signals.recordSuccess('b2', 'presign', 2);++ for (let i = 0; i < 5; i++) signals.recordFailure('b2', 'head', 10_000);++ assert.strictEqual(signals.breakerState('b2'), 'open', 'five failed HEADs open it, whatever the presigns say');++ signals._advance(61_000);++ signals.recordSuccess('b2', 'presign', 2);++ assert.notStrictEqual(signals.breakerState('b2'), 'closed', 'a presign is no proof the provider is back');++ signals.recordSuccess('b2', 'head', 90);++ assert.strictEqual(signals.breakerState('b2'), 'closed', 'a HEAD that answers is');++ for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'presign', 6000);++ assert.strictEqual(signals.breakerOpen('r2'), false, 'slow signing (host CPU) is not a slow provider');++ assert.ok(signals.latencyEWMA('r2', 'presign') > 0, 'presign latency is still measured');++ for (let i = 0; i < 6; i++) signals.recordFailure('r2', 'presign', 1);++ assert.strictEqual(signals.breakerOpen('r2'), true, 'a copy that cannot be signed cannot be served: failed presigns still count');++signals.reset();console.log('placement-signals: all checks passed');process.exit(0);
@@ -40,6 +40,10 @@ server {
@@ -40,6 +40,10 @@ server {
proxy_cache_lock_age 10s;
proxy_cache_lock_age 10s;
proxy_cache_use_stale error timeout updating http_500 http_502 http_503 http_504;
proxy_cache_use_stale error timeout updating http_500 http_502 http_503 http_504;
proxy_cache_background_update on;
proxy_cache_background_update on;
+ # Losing B2 is not a 502 for the viewer: a first slice B2 cannot serve, with nothing stale cached, goes back to
+ # Node, which records the failure and sends the read to another copy (server/placement/shield.js failedOf).
+ proxy_intercept_errors on;
+ error_page 502 503 504 = @shield_failed_b2;
# Node's answer owns the viewer-facing headers (Content-Type, Content-Disposition, Cache-Control survive the
# Node's answer owns the viewer-facing headers (Content-Type, Content-Disposition, Cache-Control survive the
# X-Accel redirect): B2's copies, and its request ids, are dropped. The security headers Node sets do not survive
# X-Accel redirect): B2's copies, and its request ids, are dropped. The security headers Node sets do not survive
# the redirect, so they are set here.
# the redirect, so they are set here.
@@ -60,6 +64,20 @@ server {
@@ -60,6 +64,20 @@ server {
include snippets/media-shield-b2-upstream.conf;
include snippets/media-shield-b2-upstream.conf;
}
}
+ # The read the shield could not fetch, back to Node as the viewer sent it, marked with the provider that failed.
+ location @shield_failed_b2 {
+ proxy_pass http://127.0.0.1:4100$request_uri;
+ proxy_http_version 1.1;
+ proxy_set_header Host $host;
+ proxy_set_header X-Media-Shield-Host $server_name;
+ proxy_set_header X-Media-Shield-Failed b2;
+ proxy_set_header X-Forwarded-Host "";
+ proxy_set_header X-Real-IP $remote_addr;
+ proxy_set_header X-Forwarded-For $remote_addr;
+ proxy_set_header CF-Connecting-IP $remote_addr;
+ proxy_set_header X-Forwarded-Proto $scheme;
+ }
+
# Everything else is Node, exactly as on openvibe.media (client-IP headers from $remote_addr only), plus the one
# Everything else is Node, exactly as on openvibe.media (client-IP headers from $remote_addr only), plus the one
# header that tells Node this is the edge: nginx sets it from its own server name, so a client cannot forge it.
# header that tells Node this is the edge: nginx sets it from its own server name, so a client cannot forge it.
location / {
location / {
@@ -67,6 +85,7 @@ server {
@@ -67,6 +85,7 @@ server {
proxy_http_version 1.1;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header Host $host;
proxy_set_header X-Media-Shield-Host $server_name;
proxy_set_header X-Media-Shield-Host $server_name;
+ proxy_set_header X-Media-Shield-Failed "";
proxy_set_header X-Forwarded-Host "";
proxy_set_header X-Forwarded-Host "";
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr;
@@ -28,6 +28,7 @@ server {
@@ -28,6 +28,7 @@ server {
# Only the edge host's block may say "edge" (placement/shield.js): cleared here so a client cannot send it
# Only the edge host's block may say "edge" (placement/shield.js): cleared here so a client cannot send it
# through Cloudflare; X-Forwarded-Host is cleared too (Express's req.hostname would honour it).
# through Cloudflare; X-Forwarded-Host is cleared too (Express's req.hostname would honour it).
proxy_set_header X-Media-Shield-Host "";
proxy_set_header X-Media-Shield-Host "";
+ proxy_set_header X-Media-Shield-Failed "";
proxy_set_header X-Forwarded-Host "";
proxy_set_header X-Forwarded-Host "";
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-For $remote_addr;
@@ -483,6 +483,14 @@ Cache-Control, nosniff and noindex, and no B2 header. VOD bytes carry no CORS he
@@ -483,6 +483,14 @@ Cache-Control, nosniff and noindex, and no B2 header. VOD bytes carry no CORS he
through the shared cache (anyone with the link may read them; `/o` answers `public, max-age=3600` for them), unlisted
through the shared cache (anyone with the link may read them; `/o` answers `public, max-age=3600` for them), unlisted
VODs answer `private, max-age=0`.
VODs answer `private, max-age=0`.
- **Private and sandbox bytes never enter the shield.**
- **Private and sandbox bytes never enter the shield.**
+- **Losing a provider fails over (2026-10-14).** A failed live health check (the HEAD each minute) moves reads to the
+ next copy at once; a presign is local signing, so its successes no longer dilute the breaker's error rate or close it
+ (during an outage every viewer's presign succeeded and the breaker never opened). A first slice the shield cannot
+ fetch from B2, with nothing stale cached, goes back to Node through the edge's `@shield_failed_b2` location
+ (`X-Media-Shield-Failed: b2`, cleared on every location a client reaches); Node records the failure and serves another
+ copy. A copy whose only provider is unhealthy is still served as the last resort (`reason: last-resort`): an outage
+ never makes a one-copy object vanish, and a corrupt, missing or unverified copy is never served at all. A second CDN
+ path is F5's (Bunny needs an account).
- **Sizing:** one disk with recordings: `max_size=5g`, `min_free=20g` (the recording guardian warns at 15 GB), `inactive=24h`.
- **Sizing:** one disk with recordings: `max_size=5g`, `min_free=20g` (the recording guardian warns at 15 GB), `inactive=24h`.
- **Purge:** deleting a shielded object refreshes each of its slices (four at a time) through a loopback-only listener
- **Purge:** deleting a shielded object refreshes each of its slices (four at a time) through a loopback-only listener
(`127.0.0.1:8479`, `allow 127.0.0.1; deny all`, `proxy_cache_bypass`), so the provider's 404 replaces the cached bytes
(`127.0.0.1:8479`, `allow 127.0.0.1; deny all`, `proxy_cache_bypass`), so the provider's 404 replaces the cached bytes
@@ -994,9 +994,12 @@ publicRouter.get('/:id', async (req, res) => {
@@ -994,9 +994,12 @@ publicRouter.get('/:id', async (req, res) => {
// providers, and ranks by preference tier with the measured 5-minute EWMA latency and the
// providers, and ranks by preference tier with the measured 5-minute EWMA latency and the
// design's ±15 % tie band: a verified R2 copy (the hot cache) for playback, then local, R2, B2.
// design's ±15 % tie band: a verified R2 copy (the hot cache) for playback, then local, R2, B2.
const router = require('../placement/router');
const router = require('../placement/router');
+ // The shield could not fetch this read (nginx handed it back): another copy, and no shield for this one.
+ const shieldFailed = require('../placement/shield').failedOf(req);
const decision = await router.route({
const decision = await router.route({
object: obj, purpose: 'playback', session: router.sessionFor(req, obj.id),
object: obj, purpose: 'playback', session: router.sessionFor(req, obj.id),
contentType: mime, contentDisposition: headers['Content-Disposition'], expiresIn: 300,
contentType: mime, contentDisposition: headers['Content-Disposition'], expiresIn: 300,
+ exclude: shieldFailed ? [shieldFailed] : [],
});
});
// Walk the router's ranked copies: stream a local one, redirect to a presigned remote one.
// Walk the router's ranked copies: stream a local one, redirect to a presigned remote one.
// A stale `present` local row whose file is gone falls through to the next copy.
// A stale `present` local row whose file is gone falls through to the next copy.
@@ -1011,11 +1014,12 @@ publicRouter.get('/:id', async (req, res) => {
@@ -1011,11 +1014,12 @@ publicRouter.get('/:id', async (req, res) => {
// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through its
// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through its
// slice cache, with a long internal presign that never reaches the viewer (only on the DNS-only edge host).
// slice cache, with a long internal presign that never reaches the viewer (only on the DNS-only edge host).
const shield = require('../placement/shield');
const shield = require('../placement/shield');
- if (shield.enabled(provider) && !sandbox && obj.visibility !== 'private' && !shield.onEdge(req)) {
+ const shielded = shield.enabled(provider) && provider !== shieldFailed;
+ if (shielded && !sandbox && obj.visibility !== 'private' && !shield.onEdge(req)) {
const edge = shield.toEdge(req, provider);
const edge = shield.toEdge(req, provider);
if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }
if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }
}
}
- if (shield.enabled(provider) && shield.onEdge(req) && !sandbox && obj.visibility !== 'private') {
+ if (shielded && shield.onEdge(req) && !sandbox && obj.visibility !== 'private') {
const inner = await require('../vod/vod-storage').presignGet(provider, loc.key, shield.SHIELD_TTL_SECONDS, { contentType: mime, contentDisposition: headers['Content-Disposition'] }).catch(() => null);
const inner = await require('../vod/vod-storage').presignGet(provider, loc.key, shield.SHIELD_TTL_SECONDS, { contentType: mime, contentDisposition: headers['Content-Disposition'] }).catch(() => null);
const t = inner && shield.target({ provider, url: inner, visibility: obj.visibility, sandbox, edge: shield.edgeOf(req) });
const t = inner && shield.target({ provider, url: inner, visibility: obj.visibility, sandbox, edge: shield.edgeOf(req) });
if (t) return shield.send(res, t, headers);
if (t) return shield.send(res, t, headers);
@@ -290,6 +290,14 @@ function observe(provider, op, ok, ms) {
@@ -290,6 +290,14 @@ function observe(provider, op, ok, ms) {
} catch { /* */ }
} catch { /* */ }
}
}
+/** True while the provider's last live health check (or capability probe) failed and is recent: the router routes
+ * reads to another copy until a check passes again. A check older than LIVE_HEALTH_FRESH_MS says nothing. */
+const LIVE_HEALTH_FRESH_MS = 5 * 60_000;
+function liveUnhealthy(name) {
+ const at = state.lastHealthCheck[name];
+ return state.lastHealthError[name] != null && at != null && Date.now() - at < LIVE_HEALTH_FRESH_MS;
+}
+
function healthStatus() {
function healthStatus() {
const out = {};
const out = {};
for (const name of vodStorage().REMOTE_PROVIDERS) {
for (const name of vodStorage().REMOTE_PROVIDERS) {
@@ -340,8 +348,9 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};
@@ -340,8 +348,9 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};
module.exports = {
module.exports = {
CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,
CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,
isConfigured, providerClasses, eligibleClasses, measuredClasses, placementGate, gateMode, capabilities,
isConfigured, providerClasses, eligibleClasses, measuredClasses, placementGate, gateMode, capabilities,
- probe, probeAll, liveHealth, healthStatus,
+ probe, probeAll, liveHealth, liveUnhealthy, healthStatus,
startHealthLoop, stop, reset,
startHealthLoop, stop, reset,
// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
_setCapabilities: setCapabilities,
_setCapabilities: setCapabilities,
+ _setHealth: (name, error) => { state.lastHealthCheck[name] = Date.now(); state.lastHealthError[name] = error || null; },
};
};
@@ -11,8 +11,10 @@
@@ -11,8 +11,10 @@
* 1. Resolve locations for the object (`locationsFor(opts)` accepts an `object`,
* 1. Resolve locations for the object (`locationsFor(opts)` accepts an `object`,
* a `vod` / `clip` row, or an explicit list).
* a `vod` / `clip` row, or an explicit list).
* 2. Filter to `state === 'present'` rows only. Non-present rows are NEVER routed to.
* 2. Filter to `state === 'present'` rows only. Non-present rows are NEVER routed to.
- * 3. Filter to providers that are configured, have passed their capability probe and are
- * not breaker-open.
+ * 3. Filter to providers that are configured and have passed their capability probe. Of those, a provider
+ * that is breaker-open, whose last live health check failed, or that this read was told to avoid (`exclude`:
+ * the origin shield could not fetch from it) is only a last resort: served when no other copy is left, so
+ * losing a provider moves reads to the next copy and never makes a one-copy object vanish.
* 4. Rank by preference tier, with the measured 5-minute EWMA latency breaking ties:
* 4. Rank by preference tier, with the measured 5-minute EWMA latency breaking ties:
* playback : R2 with a verified present copy (popularity promotion put it there to move
* playback : R2 with a verified present copy (popularity promotion put it there to move
* reads off the host; promoting a VOD also frees its local copy) > local > R2 > B2
* reads off the host; promoting a VOD also frees its local copy) > local > R2 > B2
@@ -113,17 +115,25 @@ function withOwnCopy(rows, provider, key, extra = {}) {
@@ -113,17 +115,25 @@ function withOwnCopy(rows, provider, key, extra = {}) {
return [...rows.filter((r) => r.provider !== provider), { provider, state: 'present', key, ...extra }];
return [...rows.filter((r) => r.provider !== provider), { provider, state: 'present', key, ...extra }];
}
}
-/** A copy is healthy when: configured, breaker closed, and eligible for at least one provider class
- * (a failed capability probe removes every class it needed). `passed === null` = probe not run yet. */
-function isHealthy(provider, { purpose = 'playback' } = {}) {
+/** A copy can be served at all when: configured and eligible for at least one provider class (a failed capability
+ * probe removes every class it needed). `passed === null` = probe not run yet. */
+function isEligible(provider) {
if (provider === 'local') return true;
if (provider === 'local') return true;
if (!providers.isConfigured(provider)) return false;
if (!providers.isConfigured(provider)) return false;
- if (signals.breakerOpen(provider)) return false;
const caps = providers.capabilities(provider);
const caps = providers.capabilities(provider);
if (caps.passed === null) return providers.providerClasses(provider).length > 0;
if (caps.passed === null) return providers.providerClasses(provider).length > 0;
return providers.eligibleClasses(provider).length > 0;
return providers.eligibleClasses(provider).length > 0;
}
}
+/** A copy is healthy when eligible, its breaker is closed and its last live health check did not fail. */
+function isHealthy(provider, { purpose = 'playback' } = {}) { // eslint-disable-line no-unused-vars
+ if (!isEligible(provider)) return false;
+ if (provider === 'local') return true;
+ if (signals.breakerOpen(provider)) return false;
+ try { if (providers.liveUnhealthy(provider)) return false; } catch { /* a providers stub without health */ }
+ return true;
+}
+
/** Measured 5-minute EWMA (ms) of a network round-trip op (head, then get), or null when the
/** Measured 5-minute EWMA (ms) of a network round-trip op (head, then get), or null when the
* provider has no such samples. Presign is excluded: it is a local SigV4 operation, so its
* provider has no such samples. Presign is excluded: it is a local SigV4 operation, so its
* EWMA is host CPU noise, not a network signal, and must not rank providers. */
* EWMA is host CPU noise, not a network signal, and must not rank providers. */
@@ -197,7 +207,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);
@@ -197,7 +207,8 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);
/**
/**
* The only function every read path uses.
* The only function every read path uses.
* opts: { object | vod | clip | locations, objectId?, purpose, session?, segment?, range?, presign?, expiresIn?,
* opts: { object | vod | clip | locations, objectId?, purpose, session?, segment?, range?, presign?, expiresIn?,
- * contentType?, contentDisposition? }
+ * contentType?, contentDisposition?, exclude? }
+ * `exclude` names providers this read should avoid (the shield just failed to fetch from one): last resort only.
* `objectId` names the object a demand hit belongs to when the caller passes no object/vod/clip row (a segment
* `objectId` names the object a demand hit belongs to when the caller passes no object/vod/clip row (a segment
* read: the route already resolved its row and passes explicit locations); `segment` is the timeline segment served.
* read: the route already resolved its row and passes explicit locations); `segment` is the timeline segment served.
* returns: { provider, key, url, reason, fallbacks, candidates } — `fallbacks` are provider names, ranked;
* returns: { provider, key, url, reason, fallbacks, candidates } — `fallbacks` are provider names, ranked;
@@ -208,15 +219,21 @@ async function route(opts = {}) {
@@ -208,15 +219,21 @@ async function route(opts = {}) {
const session = opts.session || null;
const session = opts.session || 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.
+ // One candidate per provider: the first present location it has. A copy on an unhealthy (or avoided) provider
+ // waits as a last resort: served only when no healthy copy is left.
+ const avoid = new Set(Array.isArray(opts.exclude) ? opts.exclude : []);
const seen = new Set();
const seen = new Set();
- const candidates = [];
+ let candidates = [];
+ const degraded = [];
for (const l of locs) {
for (const l of locs) {
if (seen.has(l.provider)) continue;
if (seen.has(l.provider)) continue;
- if (!isHealthy(l.provider, { purpose })) continue;
+ if (!isEligible(l.provider)) continue;
seen.add(l.provider);
seen.add(l.provider);
- candidates.push(l);
+ if (isHealthy(l.provider, { purpose }) && !avoid.has(l.provider)) candidates.push(l);
+ else degraded.push(l);
}
}
+ const lastResort = candidates.length === 0 && degraded.length > 0;
+ if (lastResort) candidates = degraded;
const ranked = rankCandidates(candidates, { purpose });
const ranked = rankCandidates(candidates, { purpose });
// 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
@@ -244,7 +261,7 @@ async function route(opts = {}) {
@@ -244,7 +261,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
- ? (stickyHit && primary === ordered[0] && primary.provider === sticky.provider ? 'sticky' : REASONS[tierOf(primary, purpose)] || 'fastest-healthy')
+ ? lastResort ? 'last-resort' : (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);
@@ -284,6 +301,7 @@ function incMetric(name, labels) { try { _inc(name, labels); } catch { /* */ } }
@@ -284,6 +301,7 @@ function incMetric(name, labels) { try { _inc(name, labels); } catch { /* */ } }
module.exports = {
module.exports = {
route,
route,
isHealthy,
isHealthy,
+ isEligible,
rankCandidates,
rankCandidates,
tierOf,
tierOf,
locationsFor,
locationsFor,
@@ -32,6 +32,13 @@
@@ -32,6 +32,13 @@
* edge, where the bytes come from the shield
* edge, where the bytes come from the shield
* target({ provider, url, visibility, sandbox, edge }) → '/_media_shield/b2/<bucket>/<key>?<signed query>' | null
* target({ provider, url, visibility, sandbox, edge }) → '/_media_shield/b2/<bucket>/<key>?<signed query>' | null
* send(res, target, headers) → the 200 + X-Accel-Redirect answer
* send(res, target, headers) → the 200 + X-Accel-Redirect answer
+ * failedOf(req) → the provider the shield could not fetch this read from, or null:
+ * nginx hands a failed first slice (502/503/504 with nothing stale
+ * cached) back to Node with `X-Media-Shield-Failed`, set only by the
+ * edge's @shield_failed location (every other location clears it).
+ * The route records the failure, skips the shield and asks the
+ * router for another copy (`exclude`), so losing the provider moves
+ * the read to the next copy instead of answering 502.
* await purge({ provider, key, size, presign }) → refreshes every cached slice of a deleted or replaced object
* await purge({ provider, key, size, presign }) → refreshes every cached slice of a deleted or replaced object
*/
*/
'use strict';
'use strict';
@@ -62,6 +69,16 @@ function toEdge(req, provider) {
@@ -62,6 +69,16 @@ function toEdge(req, provider) {
return p.startsWith('/') && !p.startsWith('//') ? `https://${edge}${p}` : null;
return p.startsWith('/') && !p.startsWith('//') ? `https://${edge}${p}` : null;
}
}
+const FAILED_HEADER = 'x-media-shield-failed'; // set by the edge's @shield_failed location only (nginx)
+/** The shielded provider whose fetch failed for this read, on the edge only (a client cannot reach it with the header). */
+function failedOf(req) {
+ if (!onEdge(req)) return null;
+ const v = String((typeof req.get === 'function' && req.get(FAILED_HEADER)) || '').trim().toLowerCase();
+ if (!v || !enabled(v)) return null;
+ try { require('./signals').recordFailure(v, 'get', 0); } catch { /* */ }
+ return v;
+}
+
/** The internal location for one presigned URL, or null when this read must not go through the shield. `edge` is
/** The internal location for one presigned URL, or null when this read must not go through the shield. `edge` is
* edgeOf(req): the shield answers only on MEDIA_SHIELD_HOST (a DNS-only edge name), never through Cloudflare. */
* edgeOf(req): the shield answers only on MEDIA_SHIELD_HOST (a DNS-only edge name), never through Cloudflare. */
function target({ provider, url, visibility = 'public', sandbox = false, edge = null }) {
function target({ provider, url, visibility = 'public', sandbox = false, edge = null }) {
@@ -113,4 +130,4 @@ async function purge({ provider, key, size, presign, fetchImpl = globalThis.fetc
@@ -113,4 +130,4 @@ async function purge({ provider, key, size, presign, fetchImpl = globalThis.fetc
return { purged, failed, slices };
return { purged, failed, slices };
}
}
-module.exports = { target, send, purge, enabled, providers, onEdge, edgeOf, toEdge, SLICE_BYTES, SHIELD_TTL_SECONDS, B2_UPSTREAM_HOST, EDGE_HEADER };
+module.exports = { target, send, purge, enabled, providers, onEdge, edgeOf, toEdge, failedOf, SLICE_BYTES, SHIELD_TTL_SECONDS, B2_UPSTREAM_HOST, EDGE_HEADER, FAILED_HEADER };
@@ -12,7 +12,10 @@
@@ -12,7 +12,10 @@
*
*
* The breaker opens when the 1-minute error rate exceeds 50% over at least 5 samples,
* The breaker opens when the 1-minute error rate exceeds 50% over at least 5 samples,
* or the 5-minute p95 exceeds 5 s. Half-open after 60 s: one trial; on success it
* or the 5-minute p95 exceeds 5 s. Half-open after 60 s: one trial; on success it
- * closes, on failure it re-opens. `providerHealthy` in vod-storage.js reads this so a
+ * closes, on failure it re-opens. A presign is local SigV4 signing: its successes say nothing about the
+ * provider, so they never count toward the error rate or p95 and never close a breaker (otherwise every
+ * viewer's presign during an outage diluted the failing HEADs, and the breaker never opened).
+ * A failed presign still counts: a copy that cannot be signed cannot be served. `providerHealthy` in vod-storage.js reads this so a
* dead provider is skipped without waiting for the next boot probe.
* dead provider is skipped without waiting for the next boot probe.
*
*
* In-memory only: signals are observed on the actual read path and the next boot
* In-memory only: signals are observed on the actual read path and the next boot
@@ -43,6 +46,9 @@ const listeners = new Set(); // (provider) => void — fired when a breaker
@@ -43,6 +46,9 @@ const listeners = new Set(); // (provider) => void — fired when a breaker
function key(provider, op) { return `${provider}:${op || 'op'}`; }
function key(provider, op) { return `${provider}:${op || 'op'}`; }
+// Ops that never touch the provider: their successes are not evidence of its health.
+const LOCAL_OPS = new Set(['presign']);
+
function getSignal(provider, op) {
function getSignal(provider, op) {
const k = key(provider, op);
const k = key(provider, op);
let s = state.get(k);
let s = state.get(k);
@@ -74,8 +80,10 @@ function ewmaFor(provider) {
@@ -74,8 +80,10 @@ function ewmaFor(provider) {
function push(provider, op, ok, ms) {
function push(provider, op, ok, ms) {
const s = getSignal(provider, op);
const s = getSignal(provider, op);
if (ok && ms > 0) s.ewmaMs = s.ewmaMs ? (EWMA_ALPHA * ms + (1 - EWMA_ALPHA) * s.ewmaMs) : ms;
if (ok && ms > 0) s.ewmaMs = s.ewmaMs ? (EWMA_ALPHA * ms + (1 - EWMA_ALPHA) * s.ewmaMs) : ms;
- s.samples.push({ t: nowMs(), ok: !!ok, ms: Number(ms) || 0 });
+ const local = !!ok && LOCAL_OPS.has(op);
+ s.samples.push({ t: nowMs(), ok: !!ok, ms: Number(ms) || 0, local });
if (s.samples.length > SAMPLE_LIMIT) s.samples.splice(0, s.samples.length - SAMPLE_LIMIT);
if (s.samples.length > SAMPLE_LIMIT) s.samples.splice(0, s.samples.length - SAMPLE_LIMIT);
+ if (local) return;
if (ok) onSuccess(provider); else onFailure(provider);
if (ok) onSuccess(provider); else onFailure(provider);
// Slow success (high p95) is itself a degradation signal: check every push.
// Slow success (high p95) is itself a degradation signal: check every push.
const stats5m = statsFor(provider, null, '5m');
const stats5m = statsFor(provider, null, '5m');
@@ -92,7 +100,7 @@ function statsFor(provider, _opUnused, windowKey) {
@@ -92,7 +100,7 @@ function statsFor(provider, _opUnused, windowKey) {
const cutoff = nowMs() - windowMs;
const cutoff = nowMs() - windowMs;
let total = 0, errors = 0, latencies = [];
let total = 0, errors = 0, latencies = [];
for (const sample of allSamplesFor(provider)) {
for (const sample of allSamplesFor(provider)) {
- if (sample.t < cutoff) continue;
+ if (sample.t < cutoff || sample.local) continue;
total++;
total++;
if (!sample.ok) errors++; else latencies.push(sample.ms);
if (!sample.ok) errors++; else latencies.push(sample.ms);
}
}
@@ -192,13 +192,15 @@ async function serveMediaRecord(kind, record, req, res) {
@@ -192,13 +192,15 @@ async function serveMediaRecord(kind, record, req, res) {
// requests are handled natively by the object store.
// requests are handled natively by the object store.
if (!filePath && vodStorage.isRemote(record)) {
if (!filePath && vodStorage.isRemote(record)) {
const router = require('../placement/router');
const router = require('../placement/router');
- const plan = await vodStorage.resolvePlayback(record, { session: router.sessionFor(req, `${kind}:${record.id}`) });
+ // The shield could not fetch this read (nginx handed it back): another copy, and no shield for this one.
+ const shieldFailed = require('../placement/shield').failedOf(req);
+ const plan = await vodStorage.resolvePlayback(record, { session: router.sessionFor(req, `${kind}:${record.id}`), exclude: shieldFailed ? [shieldFailed] : [] });
if (plan?.kind === 'redirect') {
if (plan?.kind === 'redirect') {
// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through
// The origin shield (placement/shield.js): a public copy on a shielded provider is served by nginx through
// its slice cache with a long internal presign (only on the DNS-only edge host); everything else keeps the
// its slice cache with a long internal presign (only on the DNS-only edge host); everything else keeps the
// short-lived 302.
// short-lived 302.
const shield = require('../placement/shield');
const shield = require('../placement/shield');
- const shieldable = shield.enabled(plan.provider) && visibility !== 'private' && !record.is_recording;
+ const shieldable = shield.enabled(plan.provider) && plan.provider !== shieldFailed && visibility !== 'private' && !record.is_recording;
const edge = shieldable && shield.toEdge(req, plan.provider);
const edge = shieldable && shield.toEdge(req, plan.provider);
if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }
if (edge) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, edge); }
if (shieldable && shield.onEdge(req)) {
if (shieldable && shield.onEdge(req)) {
@@ -525,9 +525,9 @@ async function presignGet(provider, key, expiresInSeconds = 900, overrides = {})
@@ -525,9 +525,9 @@ async function presignGet(provider, key, expiresInSeconds = 900, overrides = {})
* preferred when the file is on disk, R2 over B2 for playback when both are present, and B2
* preferred when the file is on disk, R2 over B2 for playback when both are present, and B2
* as the canonical fallback. The router returns the chosen provider; we presign it and answer.
* as the canonical fallback. The router returns the chosen provider; we presign it and answer.
*/
*/
-async function resolvePlayback(vod, { session = null, range = null } = {}) {
+async function resolvePlayback(vod, { session = null, range = null, exclude = [] } = {}) {
const router = require('../placement/router');
const router = require('../placement/router');
- const decision = await router.route({ vod, purpose: 'playback', session, range });
+ const decision = await router.route({ vod, purpose: 'playback', session, range, exclude });
if (decision.provider && decision.provider !== 'local') {
if (decision.provider && decision.provider !== 'local') {
if (decision.url) return { kind: 'redirect', url: decision.url, provider: decision.provider, fallbacks: decision.fallbacks };
if (decision.url) return { kind: 'redirect', url: decision.url, provider: decision.provider, fallbacks: decision.fallbacks };
// The router met a provider it could not rank-presign (the client is unavailable). Try the
// The router met a provider it could not rank-presign (the client is unavailable). Try the
new file mode 100644
new file mode 100644
@@ -0,0 +1,123 @@
@@ -0,0 +1,123 @@
+'use strict';
+// Plan §9 gate: seek works during an active DVR. A recording still being written is a fragmented MP4
+// (frag_keyframe+empty_moov, as the recorder writes it); Live's DVR player reads it through /v/:id, which
+// serves the live seek snapshot (remuxForLiveSeeking: a copy-remux with the whole index at the front). Here
+// with real ffmpeg: the snapshot of a recording mid-stream answers ranged reads from any offset with no-cache,
+// carries its index before its media, and a player can seek to a late moment in it; after the recording grows,
+// the next snapshot reaches the new live edge; once finalized, /v/:id serves the recording itself again.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const { spawnSync } = require('child_process');
+
+(async () => {
+ // Like cmaf-transcode.test.js: real encodes need ffmpeg/ffprobe, which CI may not have.
+ if (spawnSync('ffmpeg', ['-version']).status !== 0 || spawnSync('ffprobe', ['-version']).status !== 0) {
+ console.log('dvr seek: skipped (ffmpeg not found)');
+ process.exit(0);
+ }
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-dvr-seek-'));
+ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+ process.env.VOD_PATH = dir('vods');
+ process.env.CLIPS_PATH = dir('clips');
+ process.env.FILES_PATH = dir('files');
+ process.env.OBJECTS_PATH = dir('objects');
+ process.env.THUMBNAILS_PATH = dir('thumbnails');
+ process.env.PASTES_PATH = dir('pastes');
+ process.env.MEDIA_PUBLIC_URL = 'https://media.test';
+
+ const ffmpeg = (...args) => {
+ const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args], { encoding: 'utf8' });
+ assert.strictEqual(r.status, 0, r.stderr);
+ };
+ const duration = (file) => Number(spawnSync('ffprobe', ['-v', 'error', '-show_entries', 'format=duration', '-of', 'csv=p=0', file], { encoding: 'utf8' }).stdout.trim());
+ // A recording `seconds` long, written the way the recorder writes a growing MP4 (a keyframe every second).
+ const record = (file, seconds) => ffmpeg('-f', 'lavfi', '-i', `testsrc=duration=${seconds}:size=160x90:rate=10`,
+ '-f', 'lavfi', '-i', `sine=duration=${seconds}`, '-c:v', 'mpeg4', '-g', '10', '-c:a', 'aac',
+ '-movflags', '+frag_keyframe+empty_moov+default_base_moof', '-f', 'mp4', file);
+
+ const db = require('../server/db/database');
+ const tools = require('../server/vod/media-tools');
+ const express = require('express');
+
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+ const rec = path.join(process.env.VOD_PATH, 'vod-live-7.mp4');
+ record(rec, 4);
+ await db.run(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, visibility, is_recording) OVERRIDING SYSTEM VALUE
+ VALUES (7, 'live', 5, 'Still live', ?, 1, 'public', 1)`, [rec]);
+
+ const app = express();
+ app.use('/', require('../server/public/routes'));
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const get = async (range) => {
+ const res = await fetch(`${base}/v/7`, { headers: range ? { range } : {}, redirect: 'manual' });
+ return { status: res.status, headers: res.headers, body: Buffer.from(await res.arrayBuffer()) };
+ };
+ // A pass is paced (the next may start only after 9× the last one's run time): retry until one runs.
+ const snapshot = async () => {
+ for (let i = 0; i < 200; i++) {
+ if (await tools.remuxForLiveSeeking(rec)) return true;
+ await new Promise((r) => setTimeout(r, 100));
+ }
+ return false;
+ };
+ const ok = [];
+ const check = async (name, fn) => { await fn(); ok.push(name); };
+
+ try {
+ const sidecar = tools.seekableSidecarPath(rec);
+ await check('mid-recording, /v/:id serves the seek snapshot: ranged, uncached, its index before its media', async () => {
+ assert.ok(await snapshot(), 'the first snapshot ran');
+ const size = fs.statSync(sidecar).size;
+ const whole = await get('bytes=0-');
+ assert.strictEqual(whole.status, 206);
+ assert.match(whole.headers.get('cache-control'), /no-cache/);
+ assert.strictEqual(whole.headers.get('content-range'), `bytes 0-${size - 1}/${size}`);
+ assert.ok(whole.body.equals(fs.readFileSync(sidecar)), 'the snapshot, not the growing file');
+ const moov = whole.body.indexOf('moov'), mdat = whole.body.indexOf('mdat');
+ assert.ok(moov > 0 && mdat > moov, `the index sits before the media (moov ${moov}, mdat ${mdat})`);
+ assert.strictEqual(whole.body.indexOf('moof'), -1, 'one whole index, not fragments a player must walk');
+ // A seek lands mid-file: the player asks for a range starting there.
+ const mid = Math.floor(size / 2);
+ const part = await get(`bytes=${mid}-${mid + 999}`);
+ assert.strictEqual(part.status, 206);
+ assert.strictEqual(part.headers.get('content-range'), `bytes ${mid}-${mid + 999}/${size}`);
+ assert.ok(part.body.equals(whole.body.subarray(mid, mid + 1000)));
+ });
+
+ await check('a player seeks to a late moment of the snapshot it was served', async () => {
+ const served = path.join(tmp, 'served.mp4');
+ fs.writeFileSync(served, (await get()).body);
+ assert.ok(Math.abs(duration(served) - 4) < 0.5, `the snapshot covers what was recorded: ${duration(served)} s`);
+ const r = spawnSync('ffprobe', ['-v', 'error', '-read_intervals', '3.5%+0.5', '-select_streams', 'v', '-show_entries', 'frame=pts_time', '-of', 'csv=p=0', served], { encoding: 'utf8' });
+ assert.strictEqual(r.status, 0, r.stderr);
+ const times = r.stdout.trim().split('\n').map(Number).filter(Number.isFinite);
+ assert.ok(times.length > 0 && times[0] >= 2.9 && times[0] < 3.6, `seeking to 3.5 s decodes from its keyframe (3 s), not the start: ${times.slice(0, 3)}`);
+ });
+
+ await check('as the recording grows, the next snapshot reaches the new live edge', async () => {
+ record(rec, 8);
+ assert.ok(await snapshot(), 'the next snapshot ran');
+ const served = path.join(tmp, 'served-later.mp4');
+ fs.writeFileSync(served, (await get()).body);
+ assert.ok(Math.abs(duration(served) - 8) < 0.5, `the later snapshot covers 8 s: ${duration(served)} s`);
+ });
+
+ await check('once finalized, /v/:id serves the recording itself, cacheable again', async () => {
+ await db.run('UPDATE vods SET is_recording = 0 WHERE id = 7');
+ const res = await get('bytes=0-');
+ assert.strictEqual(res.status, 206);
+ assert.ok(res.body.equals(fs.readFileSync(rec)), 'the recording, not the snapshot');
+ assert.doesNotMatch(res.headers.get('cache-control') || '', /no-cache/);
+ });
+ } finally {
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ }
+ console.log(`dvr seek: ${ok.length} checks passed`);
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -152,6 +152,33 @@ const assert = require('assert');
@@ -152,6 +152,33 @@ const assert = require('assert');
assert.notStrictEqual(router.sessionFor(reqA, 'vod:1'), router.sessionFor(reqA, 'vod:2'));
assert.notStrictEqual(router.sessionFor(reqA, 'vod:1'), router.sessionFor(reqA, 'vod:2'));
assert.ok(!router.sessionFor(reqA, 'vod:1').includes('198.51'), 'the key is a hash, not the address');
assert.ok(!router.sessionFor(reqA, 'vod:1').includes('198.51'), 'the key is a hash, not the address');
+ // Losing a provider fails over (plan §9 gate). Its failed live health check (a HEAD each minute) moves reads to
+ // the next copy at once; a read the origin shield could not fetch avoids that provider; and a copy whose only
+ // provider is unhealthy is still served as the last resort: an outage never makes a one-copy object vanish.
+ router.reset(); signals.reset();
+ const both = () => [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }];
+ providers._setHealth('r2', 'r2 HeadBucket failed: TimeoutError');
+ decision = await router.route({ locations: both(), presign: false });
+ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'canonical-fallback'], `r2 failed its health check → b2: ${JSON.stringify(decision)}`);
+ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }], presign: false });
+ assert.deepStrictEqual([decision.provider, decision.reason], ['r2', 'last-resort'], `the only copy is still served: ${JSON.stringify(decision)}`);
+ providers._setHealth('r2', null);
+ decision = await router.route({ locations: both(), presign: false });
+ assert.strictEqual(decision.provider, 'r2', 'a passing check puts it back');
+ decision = await router.route({ locations: both(), presign: false, exclude: ['r2'] });
+ assert.deepStrictEqual([decision.provider, decision.fallbacks], ['b2', []], `a provider the shield could not fetch from is avoided: ${JSON.stringify(decision)}`);
+ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }], presign: false, exclude: ['b2'] });
+ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'last-resort'], 'avoided, but the last copy');
+ for (let i = 0; i < 10; i++) signals.recordFailure('b2', 'head', 50);
+ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }], presign: false });
+ assert.deepStrictEqual([decision.provider, decision.reason], ['b2', 'last-resort'], 'an open breaker on the only copy: last resort, not a 404');
+ decision = await router.route({ locations: [{ provider: 'b2', state: 'corrupt', key: 'k' }, { provider: 'r2', state: 'pending', key: 'k' }], presign: false });
+ assert.strictEqual(decision.provider, null, 'never a corrupt or unverified copy, even as a last resort');
+ providers.isConfigured = (name) => name === 'local' || name === 'b2';
+ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }], presign: false });
+ assert.strictEqual(decision.provider, null, 'never an unconfigured provider, even as a last resort');
+ 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);
@@ -47,6 +47,26 @@ const assert = require('assert');
@@ -47,6 +47,26 @@ const assert = require('assert');
for (const c of [edgeConf, mainConf]) assert.match(c, /proxy_set_header X-Forwarded-Host "";/);
for (const c of [edgeConf, mainConf]) assert.match(c, /proxy_set_header X-Forwarded-Host "";/);
for (const d of ['client_max_body_size 4096m;', 'proxy_read_timeout 3600s;', 'location ~* ^/metrics(/|$) { return 404; }']) assert.ok(edgeConf.includes(d), `the edge block has ${d}`);
for (const d of ['client_max_body_size 4096m;', 'proxy_read_timeout 3600s;', 'location ~* ^/metrics(/|$) { return 404; }']) assert.ok(edgeConf.includes(d), `the edge block has ${d}`);
assert.match(conf('media-shield.http.conf'), /allow 127\.0\.0\.1;\s*deny all;/, 'only loopback may refresh the cache');
assert.match(conf('media-shield.http.conf'), /allow 127\.0\.0\.1;\s*deny all;/, 'only loopback may refresh the cache');
+ // Losing B2 fails over: a first slice B2 cannot serve (nothing stale cached) goes back to Node, marked by a header
+ // only that named location sets; every location a client reaches clears it.
+ const shieldBlock = edgeConf.slice(edgeConf.indexOf('location /_media_shield/b2/'), edgeConf.indexOf('location @shield_failed_b2'));
+ assert.match(shieldBlock, /proxy_intercept_errors on;/);
+ assert.match(shieldBlock, /error_page 502 503 504 = @shield_failed_b2;/);
+ const failedBlock = edgeConf.slice(edgeConf.indexOf('location @shield_failed_b2'));
+ assert.match(failedBlock, /proxy_pass http:\/\/127\.0\.0\.1:4100\$request_uri;/, 'the original request goes back to Node');
+ assert.match(failedBlock, /proxy_set_header X-Media-Shield-Failed b2;/);
+ assert.match(failedBlock, /proxy_set_header X-Media-Shield-Host \$server_name;/);
+ assert.match(failedBlock, /proxy_set_header X-Real-IP \$remote_addr;/, 'client-IP headers from $remote_addr only');
+ for (const c of [edgeConf, mainConf]) assert.match(c.slice(c.indexOf(' location / {')), /proxy_set_header X-Media-Shield-Failed "";/, 'a client cannot claim a shield failure');
+ // failedOf(): only on the edge, only a shielded provider, and the failure feeds the provider's signals.
+ const signals = require('../server/placement/signals');
+ signals.reset();
+ const hdr = (h) => ({ get: (k) => h[k.toLowerCase()] });
+ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-host': 'edge.openvibe.media', 'x-media-shield-failed': 'b2' })), 'b2');
+ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-failed': 'b2' })), null, 'not on the edge: ignored');
+ assert.strictEqual(shield.failedOf(hdr({ 'x-media-shield-host': 'edge.openvibe.media', 'x-media-shield-failed': 'r2' })), null, 'r2 is not shielded');
+ assert.strictEqual(signals.snapshot().providers.b2.windows['1m'].errors, 1, 'one failed get recorded');
+ signals.reset();
// send(): the route's headers plus the internal redirect, an empty 200.
// send(): the route's headers plus the internal redirect, an empty 200.
const res = { h: {}, set(k, v) { this.h[k] = v; }, status(c) { this.code = c; return this; }, end() { this.ended = true; return this; } };
const res = { h: {}, set(k, v) { this.h[k] = v; }, status(c) { this.code = c; return this; }, end() { this.ended = true; return this; } };
@@ -30,7 +30,7 @@ const assert = require('assert');
@@ -30,7 +30,7 @@ const assert = require('assert');
// p95 latency on a different provider drives the second open rule.
// p95 latency on a different provider drives the second open rule.
signals.reset();
signals.reset();
- for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'presign', 6000); // 6 s, over the 5 s threshold
+ for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'head', 6000); // 6 s, over the 5 s threshold
assert.strictEqual(signals.breakerOpen('r2'), true, 'breaker opens when 5m p95 > 5 s');
assert.strictEqual(signals.breakerOpen('r2'), true, 'breaker opens when 5m p95 > 5 s');
// Half-open: after 60 s exactly one trial is allowed; a success closes the breaker, a failure
// Half-open: after 60 s exactly one trial is allowed; a success closes the breaker, a failure
@@ -42,7 +42,7 @@ const assert = require('assert');
@@ -42,7 +42,7 @@ const assert = require('assert');
assert.strictEqual(signals.breakerState('b2'), 'half-open', 'half-open 60 s after opening');
assert.strictEqual(signals.breakerState('b2'), 'half-open', 'half-open 60 s after opening');
assert.strictEqual(signals.halfOpenClaim('b2'), true, 'the one trial is claimed');
assert.strictEqual(signals.halfOpenClaim('b2'), true, 'the one trial is claimed');
assert.strictEqual(signals.halfOpenClaim('b2'), false, 'a concurrent second trial is refused');
assert.strictEqual(signals.halfOpenClaim('b2'), false, 'a concurrent second trial is refused');
- signals.recordSuccess('b2', 'presign', 80);
+ signals.recordSuccess('b2', 'head', 80);
assert.strictEqual(signals.breakerState('b2'), 'closed', 'a successful trial closes the breaker');
assert.strictEqual(signals.breakerState('b2'), 'closed', 'a successful trial closes the breaker');
for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
@@ -54,9 +54,9 @@ const assert = require('assert');
@@ -54,9 +54,9 @@ const assert = require('assert');
// Snapshot exposes both providers' state, with p50 and p95 per window (the admin view and /metrics).
// Snapshot exposes both providers' state, with p50 and p95 per window (the admin view and /metrics).
signals.reset();
signals.reset();
- signals.recordSuccess('b2', 'presign', 100);
- signals.recordSuccess('r2', 'presign', 100);
- signals.recordSuccess('r2', 'presign', 400);
+ signals.recordSuccess('b2', 'head', 100);
+ signals.recordSuccess('r2', 'head', 100);
+ signals.recordSuccess('r2', 'get', 400);
const snap = signals.snapshot();
const snap = signals.snapshot();
assert.ok(snap.providers.b2 && snap.providers.r2);
assert.ok(snap.providers.b2 && snap.providers.r2);
assert.ok(snap.breakers.b2 && snap.breakers.r2);
assert.ok(snap.breakers.b2 && snap.breakers.r2);
@@ -67,6 +67,23 @@ const assert = require('assert');
@@ -67,6 +67,23 @@ const assert = require('assert');
}
}
assert.strictEqual(snap.providers.r2.windows['5m'].total, 2, 'both r2 samples in the 5-minute window');
assert.strictEqual(snap.providers.r2.windows['5m'].total, 2, 'both r2 samples in the 5-minute window');
+ // A presign is local signing, not a provider round-trip: during an outage every viewer's presign succeeds while
+ // the HEADs fail. Those successes must not dilute the error rate, feed the p95, or close an open breaker.
+ signals.reset();
+ for (let i = 0; i < 50; i++) signals.recordSuccess('b2', 'presign', 2);
+ for (let i = 0; i < 5; i++) signals.recordFailure('b2', 'head', 10_000);
+ assert.strictEqual(signals.breakerState('b2'), 'open', 'five failed HEADs open it, whatever the presigns say');
+ signals._advance(61_000);
+ signals.recordSuccess('b2', 'presign', 2);
+ assert.notStrictEqual(signals.breakerState('b2'), 'closed', 'a presign is no proof the provider is back');
+ signals.recordSuccess('b2', 'head', 90);
+ assert.strictEqual(signals.breakerState('b2'), 'closed', 'a HEAD that answers is');
+ for (let i = 0; i < 20; i++) signals.recordSuccess('r2', 'presign', 6000);
+ assert.strictEqual(signals.breakerOpen('r2'), false, 'slow signing (host CPU) is not a slow provider');
+ assert.ok(signals.latencyEWMA('r2', 'presign') > 0, 'presign latency is still measured');
+ for (let i = 0; i < 6; i++) signals.recordFailure('r2', 'presign', 1);
+ assert.strictEqual(signals.breakerOpen('r2'), true, 'a copy that cannot be signed cannot be served: failed presigns still count');
+
signals.reset();
signals.reset();
console.log('placement-signals: all checks passed');
console.log('placement-signals: all checks passed');
process.exit(0);
process.exit(0);