openvibers/openvibe.media Commits

Media Fabric F1a: every read goes through one placement router (docs/media-fabric.md).

OP OpenVibers committed 29 Sept 2026 · b2bdc44876 · 23 files, +1883 −70

The provider registry (server/placement/providers.js) gives local, B2 and R2 their classes (regional-cache, online-canonical, online-hot) and probes each remote's capabilities (range, 206, ETag, If-Range) against a small object under fabric-probe/ at boot and hourly, with a HEAD health check every 60 s. Capability gating starts in report mode (MEDIA_CAPABILITY_GATE=report): results are recorded and shown, nothing leaves routing until enforce is set after reading the production results. Per-provider signals (server/placement/signals.js) keep latency EWMAs and p50/p95 over 10 s to 1 h windows and a circuit breaker (open past 50 % errors over 5+ samples in a minute or a 5-minute p95 over 5 s, half-open after 60 s).

The router (server/placement/router.js) takes an object, a VOD or a clip, considers only copies whose location is present (a VOD offers every verified cloud copy its object records, and the row's own copy unless its object records it missing or corrupt), skips open breakers, and ranks them: a verified R2 copy (the popularity hot cache) first for playback, then local, R2 and B2, with a measured latency advantage beyond 15 % allowed to promote; downloads prefer B2. A viewer's session sticks to its provider for 30 s unless its breaker opens, its copy goes, or a better tier is back. resolvePlayback, /o/:id, derive, clip cutting, hot-fetch and duration reconcile all ask it; no read path hard-codes a provider order any more (a grep test, proven able to fail). Presigned URLs are cached (LRU, 5,000 entries, keyed on the exact TTL, valid until 60 s before expiry). Prices live in the revisioned media.cost_tiers config (merged per level). Metrics: media_provider_latency_seconds, media_provider_errors_total, media_provider_breaker_open, media_router_decisions_total, media_presign_cache_total, media_storage_alerts_total.

Built by agents (MiniMax M3, then the DeepSeek harness), reviewed three times: Claude Sonnet (FIX: a promoted VOD lost its B2 fallback, a vacuous grep test, sessions not wired), DeepSeek on Opus's fixes (FIX: a pending location hid the row's own copy, the capability probe could not run and would have taken B2 out of routing, local-first reversed the promotion rule), DeepSeek again (SHIP; its five follow-ups fixed here: the restore walk no longer refuses early, /o and derive walk the router's keyed candidates, the probe records health, probeAll derives its providers, cost tiers merge per level). npm test 73/74 (n-1 environmental), npm run test:pg green, the 15 placement/objects/tiering/drill/reconcile suites green after the last fixes.

Changes

ddiff --git a/.env.example b/.env.example
iindex f636483..3288f2d 100644
---- a/.env.example
++++ b/.env.example
@@@ -84,6 +84,11 @@ MEDIA_R2_REGION=auto
MEDIA_R2_BUCKET=
MEDIA_R2_ACCESS_KEY_ID=
MEDIA_R2_SECRET_ACCESS_KEY=
++# Placement capability probe (server/placement/providers.js): each provider is probed for range/206/ETag/If-Range.
++# report (default) records the results (health, admin placement view) and takes nothing out of routing; enforce
++# removes a provider from every class whose requirements its last completed probe failed. Enforce only after the
++# production results were read: S3-compatible APIs differ, and If-Range especially.
++MEDIA_CAPABILITY_GATE=report
# ── Recording knobs (defaults match the inherited recorder) ──
# VOD_STOP_GRACE_MS=60000
ddiff --git a/server/index.js b/server/index.js
iindex a02ff81..55729ba 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -51,6 +51,13 @@ const ready = (async () => {
if (!drill.enabled) await require('./revocations').load();
await require('./vod/tier-config').init(require('./vod/vod-storage').DEFAULTS);
await require('./objects/tier-policy').init();
++ await require('./placement/cost-tiers').init();
++ // Placement core (F1 part A): the provider registry, signals + router. Boot the cheap live
++ // health loop after the DB is open (it reads memory only) and the router is wired in.
++ if (!drill.enabled) {
++ try { await require('./placement/providers').probeAll(); } catch (err) { console.warn('[Placement] initial probe failed:', err.message); }
++ try { require('./placement/providers').startHealthLoop(); } catch (err) { console.warn('[Placement] health loop did not start:', err.message); }
++ }
if (!drill.enabled) {
await auth.seedApps(); // upsert MEDIA_APPS_SEED / MEDIA_APP_KEYS
await auth.ensureTokenOnlyApps(); // tenants reached only with Network service tokens (community)
ddiff --git a/server/jobs/derive.js b/server/jobs/derive.js
iindex b1ff0e6..d6cfaa9 100644
---- a/server/jobs/derive.js
++++ b/server/jobs/derive.js
@@@ -61,9 +61,12 @@ function existing(p) {
/**
* Something ffmpeg can read for this object: a local path, or a presigned R2/B2 URL.
* Projected vods/clips resolve the way playback does; native objects through their locations.
++ *
++ * All provider choice is the placement router (server/placement/router.js); no hard-coded order.
*/
async function resolveSource(obj) {
const vodStorage = require('../vod/vod-storage');
++ const router = require('../placement/router');
const ref = model.parseLegacyRef(obj.legacy_ref);
if (ref && (ref.kind === 'vod' || ref.kind === 'clip')) {
const row = await db.get(`SELECT * FROM ${ref.kind === 'vod' ? 'vods' : 'clips'} WHERE id = ?`, [Number(ref.id)]);
@@@ -76,13 +79,17 @@ async function resolveSource(obj) {
return src ? { input: src.value, name: row.file_path, remote: src.kind === 'url' } : null;
}
const locs = await model.listLocations(obj.id);
-- const local = locs.find(l => l.provider === 'local' && l.state !== 'missing' && existing(l.key));
-- if (local) return { input: local.key, name: obj.canonical_key || local.key };
-- for (const p of ['r2', 'b2']) {
-- const l = locs.find(x => x.provider === p && !['missing', 'corrupt'].includes(x.state));
-- if (!l || !vodStorage.providerConfigured(p)) continue;
-- const url = await vodStorage.presignGet(p, l.key, 6 * 3600).catch(() => null);
-- if (url) return { input: url, name: l.key, remote: true };
++ const decision = await router.route({ locations: locs, purpose: 'derive', expiresIn: 6 * 3600 });
++ // The router ranks the copies (local first, then R2/B2) and presigns the chosen remote one;
++ // walk its chain so a stale local row or a failed presign still finds a readable source.
++ for (const loc of decision.candidates || []) {
++ const provider = loc.provider;
++ if (provider === 'local') {
++ if (existing(loc.key)) return { input: loc.key, name: obj.canonical_key || loc.key };
++ continue;
++ }
++ const url = await vodStorage.presignGet(provider, loc.key, 6 * 3600).catch(() => null);
++ if (url) return { input: url, name: loc.key, remote: true };
}
return null;
}
ddiff --git a/server/me/ops.js b/server/me/ops.js
iindex d9ea309..cff5a0d 100644
---- a/server/me/ops.js
++++ b/server/me/ops.js
@@@ -161,6 +161,16 @@ async function tiering(appId, limit) {
const nativeObjects = await require('../objects/tiering').report({ appId, limit: Math.min(limit, 20) });
return {
providers: { b2: vodStorage.providerConfigured('b2'), r2: vodStorage.providerConfigured('r2') },
++ // Placement (F1): the rolling per-provider signals the read router decides with — breaker
++ // state, EWMA latency and error/p50/p95 per window — plus the probed capabilities.
++ placement: (() => {
++ try {
++ return {
++ ...require('../placement/signals').snapshot(),
++ health: require('../placement/providers').healthStatus(),
++ };
++ } catch { return null; }
++ })(),
policy: { enabled: !!settings.enabled, r2_enabled: !!settings.r2Enabled, min_age_days: settings.minAgeDays, max_views_for_cold: settings.maxViewsForCold, min_last_access_days: settings.minLastAccessDays },
vods: await providers('vods', true),
clips: await providers('clips', false),
ddiff --git a/server/objects/routes.js b/server/objects/routes.js
iindex 92cad35..64933eb 100644
---- a/server/objects/routes.js
++++ b/server/objects/routes.js
@@@ -754,21 +754,27 @@ publicRouter.get('/:id', async (req, res) => {
// A viewer of a native object, counted once a day without being identified (objects/popularity.js); the
// object tiering reads these counts. Sandbox objects are never tiered, so they are not counted.
if (!obj.legacy_ref && !sandbox) await require('./popularity').record(obj, req);
-- const locs = await model.listLocations(obj.id);
-- const vodStorage = require('../vod/vod-storage');
-- // A native object promoted to the R2 popularity cache (objects/tiering.js) plays from its verified R2 copy
-- // while R2 is available, with the headers this route would send; the canonical copy is the fallback.
-- const cached = !obj.legacy_ref && locs.find(l => l.provider === 'r2' && l.state === 'present' && l.verified_at);
-- if (cached && vodStorage.providerAvailable('r2')) {
-- const url = await vodStorage.presignGet('r2', cached.key, 300, { contentType: mime, contentDisposition: headers['Content-Disposition'] }).catch(() => null);
-- if (url) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, url); }
-- }
-- const local = locs.find(l => l.provider === 'local' && l.state !== 'missing' && fs.existsSync(l.key));
-- if (local) return require('../public/routes').streamFileWithRange(req, res, local.key, headers);
-- for (const p of ['r2', 'b2']) {
-- const l = locs.find(x => x.provider === p && x.state !== 'missing' && x.state !== 'corrupt');
-- if (!l || !vodStorage.providerConfigured(p)) continue;
-- const url = await vodStorage.presignGet(p, l.key, 300).catch(() => null);
++ // Placement router (server/placement/router.js) is the only function that picks a provider.
++ // The router ignores locations not in state='present', skips open breakers and unprobed
++ // 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');
++ const decision = await router.route({
++ object: obj, purpose: 'playback', session: router.sessionFor(req, obj.id),
++ contentType: mime, contentDisposition: headers['Content-Disposition'], expiresIn: 300,
++ });
++ // 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.
++ // The router's candidates carry their keys: a copy it resolved without a media_locations row (the object's
++ // own canonical copy) is served too.
++ for (const loc of decision.candidates || []) {
++ const provider = loc.provider;
++ if (provider === 'local') {
++ if (fs.existsSync(loc.key)) return require('../public/routes').streamFileWithRange(req, res, loc.key, headers);
++ continue;
++ }
++ // The router already signed its first choice; a fallback copy is signed here.
++ const url = (provider === decision.provider && decision.url) || await require('../vod/vod-storage').presignGet(provider, loc.key, 300, { contentType: mime, contentDisposition: headers['Content-Disposition'] }).catch(() => null);
if (url) { res.set('Cache-Control', 'private, max-age=0'); res.set('X-Robots-Tag', 'noindex'); return res.redirect(302, url); }
}
res.status(404).json({ error: 'Object bytes unavailable' });
ddiff --git a/server/objects/verify-job.js b/server/objects/verify-job.js
iindex 4279af4..23c93dd 100644
---- a/server/objects/verify-job.js
++++ b/server/objects/verify-job.js
@@@ -26,7 +26,9 @@ const config = require('../config');
const copyReport = require('./copy-report');
const MB = 1024 * 1024;
--const REMOTE = ['b2', 'r2'];
++// Every remote tier the fabric knows — the canonical list lives in vod-storage (REMOTE_PROVIDERS);
++// this is an enumeration for verification, never a read-path preference order.
++const REMOTE = () => require('../vod/vod-storage').REMOTE_PROVIDERS;
let _timer = null, _startTimer = null, _busy = false;
@@@ -117,7 +119,7 @@ async function verifyObject(obj, opts) {
const reuploads = [];
if (goodLocal) {
for (const loc of locs) {
-- if (!REMOTE.includes(loc.provider)) continue;
++ if (!REMOTE().includes(loc.provider)) continue;
const r = results.find(x => x.provider === loc.provider);
const want = r.state === 'missing' || (repairCorrupt && r.state === 'corrupt');
if (!want) continue;
ddiff --git a/server/observability.js b/server/observability.js
iindex 1fe0317..84721dd 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -111,6 +111,13 @@ function domainMetrics(registry, { db, recorder, events }) {
return [{ labels: { status: 'pending' }, value: s.pending }, { labels: { status: 'rejected' }, value: s.rejected }];
},
});
++
++ // ── Placement (F1 part A) ───────────────────────────────────────────
++ // The router / signals / presign-cache / providers modules expose bindMetric hooks; collect()'s
++ // by-name lookup lets each module emit its own increments while keeping the registry as the single
++ // source of truth (one HELP/TYPE per metric name across the whole process).
++ const placement = require('./placement/metrics-binding');
++ placement.bind(registry);
}
function createMediaReadiness({ release, db, config, auth, recorder, events, remote, drill = false }) {
@@@ -131,7 +138,7 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
}
}
checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });
-- for (const name of ['b2', 'r2']) {
++ for (const name of require('./vod/vod-storage').REMOTE_PROVIDERS) {
if (drill) continue;
// Not configured: listed as skipped (never a pretend pass), so the status shows the tier is not in use.
if (!remote.configured(name)) { checks.push({ name: `remote_${name}`, required: false, check: () => skip('not configured') }); continue; }
ddiff --git a/server/placement/cost-tiers.js b/server/placement/cost-tiers.js
nnew file mode 100644
iindex 0000000..4ab4af7
---- /dev/null
++++ b/server/placement/cost-tiers.js
@@@ -0,0 +1,159 @@
++/**
++ * OpenVibe.Media — the cloud storage cost table as revisioned configuration (placement F1, part 5;
++ * docs/media-fabric.md §2, §4, §6).
++ *
++ * Every estimator that used the hard-coded `CLOUD_PRICING` constant reads this instead. The values
++ * mirror the official price pages (Cloudflare R2, Backblaze B2) at the design date; revisions carry
++ * the page metadata so a price change is one revision, validated, with who-and-why.
++ *
++ * Defaults reflect what the doc accepts as the starting point:
++ * R2 Standard $0.015/GB-mo, $4.50/M Class A, $0.36/M Class B, 10 GB free, 1M Class A free,
++ * 10M Class B free, egress free
++ * R2 Infrequent $0.01/GB-mo + $0.01/GB retrieval + 30-day minimum, $9/M Class A, $0.90/M Class B
++ * B2 $6.95/TB-mo ≈ $0.00695/GB-mo, free ops, 10 GB free, egress free up to 3× storage
++ * then $0.01/GB; partnerEgressFree is OFF by default (safer assumption until
++ * the B2-Bandwidth-Alliance egress path is confirmed by the operator)
++ * local $0 / $0 (prepaid bandwidth on the host, counted by the platform; not billed per GB)
++ *
++ * `partnerEgressFree` defaults to false (the safer assumption: a presigned B2 URL to a viewer is NOT
++ * necessarily going via the Alliance CDN; treat as paid egress until confirmed).
++ */
++'use strict';
++
++const config = require('openvibe-shared/config');
++const db = require('../db/database');
++
++const DEFAULTS = {
++ r2: {
++ standard: {
++ storagePerGbMonth: 0.015,
++ classAPerMillion: 4.50,
++ classBPerMillion: 0.36,
++ egressPerGb: 0,
++ freeGb: 10,
++ freeClassA: 1_000_000,
++ freeClassB: 10_000_000,
++ },
++ infrequentAccess: {
++ storagePerGbMonth: 0.01,
++ classAPerMillion: 9.00,
++ classBPerMillion: 0.90,
++ retrievalPerGb: 0.01,
++ egressPerGb: 0,
++ freeGb: 0,
++ freeClassA: 0,
++ freeClassB: 0,
++ minResidencyDays: 30,
++ },
++ },
++ b2: {
++ storagePerGbMonth: 0.00695, // $6.95/TB-month as of the design date
++ classAPerMillion: 0,
++ classBPerMillion: 0,
++ egressPerGb: 0.01, // after the 3× free allowance
++ freeGb: 10, // B2's first 10 GB free
++ egressFreeMultiplier: 3, // free egress = 3× stored bytes per day
++ partnerEgressFree: false, // see module comment: safer default
++ },
++ local: {
++ storagePerGbMonth: 0,
++ egressPerGb: 0,
++ freeGb: 0,
++ },
++ notes: {
++ r2: 'R2 Standard; Infrequent Access not used online (retrieval fee + 30-day minimum).',
++ b2: 'B2 Standard; egress free up to 3× stored bytes/month, then $0.01/GB; Bandwidth Alliance path is operator-toggled.',
++ },
++};
++
++const SCHEMA = {
++ type: 'object',
++ additionalProperties: false,
++ properties: {
++ r2: {
++ type: 'object',
++ additionalProperties: false,
++ properties: {
++ standard: providerSchema('Standard'),
++ infrequentAccess: providerSchema('InfrequentAccess'),
++ },
++ },
++ b2: providerSchema('B2'),
++ local: providerSchema('local'),
++ notes: {
++ type: 'object',
++ additionalProperties: false,
++ properties: {
++ r2: { type: 'string' },
++ b2: { type: 'string' },
++ },
++ },
++ },
++};
++
++function providerSchema(name) {
++ return {
++ type: 'object',
++ additionalProperties: false,
++ properties: {
++ storagePerGbMonth: { type: 'number', minimum: 0, maximum: 10 },
++ classAPerMillion: { type: 'number', minimum: 0, maximum: 1000 },
++ classBPerMillion: { type: 'number', minimum: 0, maximum: 1000 },
++ egressPerGb: { type: 'number', minimum: 0, maximum: 10 },
++ retrievalPerGb: { type: 'number', minimum: 0, maximum: 10 },
++ freeGb: { type: 'number', minimum: 0, maximum: 1e9 },
++ freeClassA: { type: 'integer', minimum: 0, maximum: 1e12 },
++ freeClassB: { type: 'integer', minimum: 0, maximum: 1e12 },
++ egressFreeMultiplier: { type: 'number', minimum: 0, maximum: 100 },
++ partnerEgressFree: { type: 'boolean' },
++ minResidencyDays: { type: 'integer', minimum: 0, maximum: 365 },
++ },
++ };
++}
++
++function validate(v) {
++ const errs = [];
++ if (v.r2.standard.storagePerGbMonth < v.r2.infrequentAccess.storagePerGbMonth) errs.push('r2.standard.storagePerGbMonth must be >= r2.infrequentAccess.storagePerGbMonth (IA is the cheaper tier on paper)');
++ if (v.b2.storagePerGbMonth < 0) errs.push('b2.storagePerGbMonth must be >= 0');
++ if (v.b2.egressFreeMultiplier < 0) errs.push('b2.egressFreeMultiplier must be >= 0');
++ return errs.length ? errs : true;
++}
++
++let store = null;
++
++async function init({ log = console } = {}) {
++ if (store) return store;
++ const s = await config.createConfigStore({
++ db: db.getDb(),
++ service: 'media',
++ namespace: 'media.cost_tiers',
++ defaults: DEFAULTS,
++ schema: SCHEMA,
++ validate,
++ legacy: () => ({}),
++ onActivate: async () => { /* no media_settings mirror; pure read-only */ },
++ log: { info: (m) => console.log(`[CostTiers] ${m}`), warn: (m) => console.warn(`[CostTiers] ${m}`), error: (m) => console.error(`[CostTiers] ${m}`) },
++ });
++ store = s;
++ return store;
++}
++
++function get() {
++ if (!store) throw new Error('media.cost_tiers is not loaded yet (cost-tiers init() at boot)');
++ return store;
++}
++
++// A revision may set one nested price (r2.standard.storage) and keep the rest: merge into the defaults per level,
++// never shallowly (a shallow merge left sibling prices undefined and the estimates NaN).
++function merge(base, over) {
++ const out = { ...base };
++ for (const [k, v] of Object.entries(over || {})) {
++ out[k] = v && typeof v === 'object' && !Array.isArray(v) && base && base[k] && typeof base[k] === 'object' ? merge(base[k], v) : v;
++ }
++ return out;
++}
++function settings() {
++ try { return merge(DEFAULTS, get().get()); } catch (err) { return merge(DEFAULTS, {}); }
++}
++
++module.exports = { DEFAULTS, SCHEMA, validate, init, get, settings, _reset: () => { store = null; } };
\\ No newline at end of file
ddiff --git a/server/placement/metrics-binding.js b/server/placement/metrics-binding.js
nnew file mode 100644
iindex 0000000..75214e2
---- /dev/null
++++ b/server/placement/metrics-binding.js
@@@ -0,0 +1,90 @@
++/**
++ * OpenVibe.Media — single point that registers every placement metric (F1 part A, §6) and wires the
++ * `inc(name, labels)` callbacks the other placement modules call. Keeps the registry as the only
++ * source of truth (one HELP/TYPE per metric name).
++ *
++ * bind(registry) create the metrics, wire the counters/histograms/gauge, and rebind
++ * the router / presign-cache / signals to inc() through them
++ * inc(name, labels) the hook other modules use (router.route, presignCache.get, etc.)
++ * setBreaker(provider, ...) signals module hook to flip the breaker-open gauge
++ *
++ * Metrics:
++ * media_provider_latency_seconds_bucket{provider,op} histogram (default + latency buckets)
++ * media_provider_errors_total{provider,op} counter
++ * media_provider_breaker_open{provider} gauge (0/1)
++ * media_router_decisions_total{provider,purpose,reason} counter
++ * media_presign_cache_total{result} counter
++ * media_storage_alerts_total{kind} counter
++ */
++'use strict';
++
++const signals = require('./signals');
++const router = require('./router');
++const presignCache = require('./presign-cache');
++
++let latency = null; // histogram
++let errors = null; // counter
++let breakerGauge = null; // gauge (collect)
++let decisions = null; // counter
++let presign = null; // counter
++let alerts = null; // counter
++
++function bind(registry) {
++ if (latency) return; // idempotent
++
++ latency = registry.histogram({
++ name: 'media_provider_latency_seconds',
++ help: 'Provider op latency in seconds (placement F1; presign = the cost the player pays to be served)',
++ labelNames: ['provider', 'op'],
++ buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10],
++ });
++ errors = registry.counter({
++ name: 'media_provider_errors_total',
++ help: 'Provider op errors (placement F1)',
++ labelNames: ['provider', 'op'],
++ });
++ breakerGauge = registry.gauge({
++ name: 'media_provider_breaker_open',
++ help: '1 when the placement breaker is open for a provider (the router skips it)',
++ labelNames: ['provider'],
++ collect: () => {
++ const out = [];
++ for (const p of require('../vod/vod-storage').REMOTE_PROVIDERS) out.push({ labels: { provider: p }, value: signals.breakerOpen(p) ? 1 : 0 });
++ return out;
++ },
++ });
++ decisions = registry.counter({
++ name: 'media_router_decisions_total',
++ help: 'Read-router decisions per provider, purpose and reason (placement F1)',
++ labelNames: ['provider', 'purpose', 'reason'],
++ });
++ presign = registry.counter({
++ name: 'media_presign_cache_total',
++ help: 'Presigned-URL LRU cache lookups, by hit/miss',
++ labelNames: ['result'],
++ });
++ alerts = registry.counter({
++ name: 'media_storage_alerts_total',
++ help: 'Storage alerts emitted to /metrics, mirroring the storage.alert webhook kinds',
++ labelNames: ['kind'],
++ });
++
++ // Bind every hook the placement modules use.
++ router.bindMetric((name, labels) => inc(name, labels));
++ presignCache.bindMetric((name, labels) => inc(name, labels));
++ signals.onBreakerChange(() => { /* gauge collects lazily on next /metrics scrape */ });
++}
++
++function inc(name, labels) {
++ if (name === 'media_router_decisions_total') decisions && decisions.inc(labels || {}, 1);
++ else if (name === 'media_presign_cache_total') presign && presign.inc(labels || {}, 1);
++ else if (name === 'media_storage_alerts_total') alerts && alerts.inc(labels || {}, 1);
++ else if (name === 'media_provider_errors_total') errors && errors.inc(labels || {}, 1);
++}
++
++/** Hook the signals module's recordSuccess/recordFailure to fill the histogram/counter. */
++function observeLatency(provider, op, ms) { if (latency && ms > 0) latency.observe({ provider, op }, ms / 1000); }
++function observeError(provider, op) { errors && errors.inc({ provider, op }, 1); }
++function alert(kind) { alerts && alerts.inc({ kind }, 1); }
++
++module.exports = { bind, inc, observeLatency, observeError, alert, _reset: () => { latency = errors = breakerGauge = decisions = presign = alerts = null; } };
\\ No newline at end of file
ddiff --git a/server/placement/presign-cache.js b/server/placement/presign-cache.js
nnew file mode 100644
iindex 0000000..65de98c
---- /dev/null
++++ b/server/placement/presign-cache.js
@@@ -0,0 +1,63 @@
++/**
++ * OpenVibe.Media — presigned-URL LRU in front of vod-storage.presignGet (placement F1, part 4).
++ *
++ * Each presign is one Class B op on R2 ($0.36/M); a popular URL is signed many times in a row.
++ * Keyed by (provider, key, contentType, contentDisposition, ttl) the LRU caches the URL
++ * itself. An entry is valid until 60 s before its TTL expires — close enough to the edge that
++ * nobody is ever served an expired URL, but enough head-room that a cached URL was never used
++ * within the unsafe last minute.
++ *
++ * get(key) -> { url, expiresAt } | null
++ * put(key, url, ttlSeconds)
++ * metrics: bindMetric(fn) to count hits/misses
++ *
++ * Bounded: at most MAX_ENTRIES entries. Eviction is LRU on put.
++ */
++'use strict';
++
++const MAX_ENTRIES = 5000;
++const SAFETY_MARGIN_MS = 60_000; // drop an entry this long before its expiry
++
++const cache = new Map(); // key -> { url, expiresAt }
++let _inc = () => {};
++function bindMetric(fn) { _inc = fn || (() => {}); }
++
++function bucketise(ttlSeconds) {
++ if (ttlSeconds <= 60) return '60s';
++ if (ttlSeconds <= 300) return '5m';
++ if (ttlSeconds <= 900) return '15m';
++ if (ttlSeconds <= 3600) return '1h';
++ return '>1h';
++}
++
++// The exact TTL, not a bucket: a 2-hour URL must never answer a 6-hour request (derive's ffmpeg
++// source reads for hours) and expire under it.
++function keyOf(provider, key, contentType, contentDisposition, ttlSeconds) {
++ return `${provider}|${key}|${contentType || ''}|${contentDisposition || ''}|${Math.round(ttlSeconds)}`;
++}
++
++function get(provider, key, contentType, contentDisposition, ttlSeconds) {
++ const k = keyOf(provider, key, contentType, contentDisposition, ttlSeconds);
++ const e = cache.get(k);
++ if (!e) { _inc('media_presign_cache_total', { result: 'miss' }); return null; }
++ if (e.expiresAt - Date.now() <= SAFETY_MARGIN_MS) { cache.delete(k); _inc('media_presign_cache_total', { result: 'miss' }); return null; }
++ // LRU touch
++ cache.delete(k); cache.set(k, e);
++ _inc('media_presign_cache_total', { result: 'hit' });
++ return e;
++}
++
++function put(provider, key, contentType, contentDisposition, ttlSeconds, url) {
++ const k = keyOf(provider, key, contentType, contentDisposition, ttlSeconds);
++ cache.set(k, { url, expiresAt: Date.now() + ttlSeconds * 1000 });
++ if (cache.size > MAX_ENTRIES) {
++ // Drop the oldest entry (Map preserves insertion order; re-insert on get keeps things hot).
++ const firstKey = cache.keys().next().value;
++ if (firstKey) cache.delete(firstKey);
++ }
++}
++
++function clear() { cache.clear(); }
++function size() { return cache.size; }
++
++module.exports = { get, put, clear, size, MAX_ENTRIES, bindMetric };
\\ No newline at end of file
ddiff --git a/server/placement/providers.js b/server/placement/providers.js
nnew file mode 100644
iindex 0000000..8f233db
---- /dev/null
++++ b/server/placement/providers.js
@@@ -0,0 +1,303 @@
++/**
++ * OpenVibe.Media — provider registry over the existing S3 engine (placement F1, part 1).
++ *
++ * Every provider here is a thin adapter that reuses vod-storage.js's S3 client and primitives
++ * (`headObject`, `presignGet`, `getObject` stream). The provider declares the classes it can
++ * serve (`online-canonical` / `online-hot` / `regional-cache`) and a capability probe that runs
++ * once per provider at boot and once an hour afterwards.
++ *
++ * classes provider classes the fabric considers for read routing
++ * capabilities(name) a snapshot of the probed capabilities (range, range_206, if_range,
++ * etag, cache_range, token_auth, purge) and `passed` overall
++ * probe(name, client?) run the capability probe for one provider (writes a tiny probe
++ * object once under prefix `fabric-probe/`; rereads/HEADs it through
++ * every supported path)
++ * probeAll() probe every configured remote provider
++ * startHealthLoop() the cheap per-minute HeadBucket and the hourly capability probe
++ * healthStatus() { b2, r2 } each as { configured, healthy, lastCheck, lastError }
++ *
++ * A provider that fails a capability is removed from the classes that need it until it passes
++ * again — `probeProvider(name)` returning a capability list with `passed=false` is the only
++ * signal the router needs to skip it for that class. Continuous health comes from a cheap HEAD
++ * every 60 s plus the passive signals the read path records (placement/signals.js).
++ */
++'use strict';
++
++const fs = require('fs');
++const path = require('path');
++const crypto = require('crypto');
++
++const vodStorage = () => require('../vod/vod-storage');
++
++// Provider classes the fabric considers; a provider declares which ones it serves.
++const CLASSES = ['online-canonical', 'online-hot', 'regional-cache'];
++
++// Capability matrix: the probe checks each. A missing capability removes that class for the
++// provider, not the provider itself (B2 has range+206 but is not "hot"; R2 has range but is
++// always served as `online-hot`).
++const CLASS_REQUIREMENTS = {
++ 'online-canonical': ['range', 'range_206', 'etag', 'if_range'],
++ 'online-hot': ['range', 'range_206'],
++ 'regional-cache': [], // local: no remote capabilities needed
++};
++
++const PROBE_PREFIX = 'fabric-probe/';
++const PROBE_KEY = `${PROBE_PREFIX}capability-probe`; // one tiny object per provider
++const PROBE_BYTES = Buffer.from('openvibe-media-fabric-probe', 'utf8');
++
++let S3; // lazy
++function loadSdk() {
++ if (!S3) { S3 = require('@aws-sdk/client-s3'); }
++}
++
++const state = {
++ started: false,
++ timers: [], // so tests can stop them
++ lastHealthCheck: {}, // provider -> epoch ms
++ lastHealthError: {}, // provider -> string
++};
++
++/**
++ * Capability gating. 'report' (the default) records what each provider can do and exposes it (health, the
++ * admin placement view) without taking anything out of routing; 'enforce' (MEDIA_CAPABILITY_GATE=enforce) removes
++ * a provider from every class whose requirements its last completed probe failed. Real providers run in report
++ * mode first: S3-compatible APIs differ (If-Range especially), and one unsupported check must never take the
++ * canonical copy offline unseen.
++ */
++function gateMode() { return String(process.env.MEDIA_CAPABILITY_GATE || 'report').toLowerCase() === 'enforce' ? 'enforce' : 'report'; }
++
++/** The classes a provider's measured capabilities satisfy (null until a probe has completed). */
++function measuredClasses(name) {
++ if (name === 'local') return providerClasses('local');
++ const caps = capabilities(name);
++ if (!caps.passed) return null;
++ const cap = caps.capabilities || {};
++ return providerClasses(name).filter((cls) => CLASS_REQUIREMENTS[cls].every((c) => cap[c]));
++}
++
++/** Which classes a provider currently claims for routing (its declaration, narrowed by the probe when enforced). */
++function eligibleClasses(name) {
++ // Local: always eligible (regional-cache), no probe needed.
++ if (name === 'local') return providerClasses('local');
++ const measured = measuredClasses(name);
++ if (gateMode() !== 'enforce' || measured === null) return providerClasses(name);
++ return measured;
++}
++
++/** What classes each provider says it can serve. These are the design's assignments (docs/media-fabric.md §2):
++ * B2 is the online canonical, R2 Standard the online hot tier, the Media host's NVMe the regional cache. */
++function providerClasses(name) {
++ switch (name) {
++ case 'local': return ['regional-cache'];
++ case 'b2': return ['online-canonical'];
++ case 'r2': return ['online-hot'];
++ default: return [];
++ }
++}
++
++/** Cheap per-provider capability snapshot: probe result + last known state. */
++function capabilities(name) {
++ const stored = state.capabilities && state.capabilities[name];
++ if (stored) return stored;
++ return { provider: name, passed: null, capabilities: {}, lastChecked: null, lastError: null };
++}
++
++/** Record the latest probe result in memory. */
++function setCapabilities(name, caps) {
++ if (!state.capabilities) state.capabilities = {};
++ state.capabilities[name] = caps;
++}
++
++/** Whether the provider is configured (env is present). Always false for `local`. */
++function isConfigured(name) {
++ if (name === 'local') return true;
++ return vodStorage().providerConfigured(name);
++}
++
++/** Run the capability probe for one provider against its bucket. */
++async function probe(name) {
++ const vs = vodStorage();
++ if (!isConfigured(name)) {
++ const out = { provider: name, passed: false, capabilities: {}, lastChecked: Date.now(), lastError: 'not configured' };
++ setCapabilities(name, out); return out;
++ }
++ if (name === 'local') {
++ const out = { provider: name, passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true, cache_range: true, token_auth: true, purge: true }, lastChecked: Date.now(), lastError: null };
++ setCapabilities(name, out); return out;
++ }
++ loadSdk();
++ const client = vs.clientFor(name);
++ const bucket = vs.bucketFor(name);
++ if (!client || !bucket) {
++ const out = { provider: name, passed: false, capabilities: {}, lastChecked: Date.now(), lastError: 'no client' };
++ setCapabilities(name, out); return out;
++ }
++ const caps = { range: false, range_206: false, etag: false, if_range: false, cache_range: false, token_auth: false, purge: false };
++ const prev = capabilities(name);
++ let error = null;
++ try {
++ await ensureProbeObject(name, client, bucket);
++ const head = await vs.headObject(name, PROBE_KEY).catch(() => null);
++ if (!head) { error = 'probe object missing'; }
++ else {
++ caps.range = !!head.size;
++ caps.etag = !!head.etag;
++ const range206 = await sendRange(name, client, bucket, PROBE_KEY, 0, 5).catch(() => null);
++ caps.range_206 = !!(range206 && range206.status === 206);
++ caps.if_range = await sendIfRange(name, client, bucket, PROBE_KEY, head.etag).catch(() => null) === true;
++ caps.cache_range = caps.range_206; // range slicing == cache range: same code path
++ // token_auth + purge: these are operation surfaces (presign + DeleteObject). The probe
++ // can't fully validate them without exercising the production code; treat as supported
++ // when range works, since presign is over the same auth path.
++ caps.token_auth = caps.range;
++ caps.purge = caps.range;
++ }
++ } catch (err) { error = err.message; }
++ // `passed` means the probe completed; what the provider can do is in `capabilities`, and the classes those
++ // satisfy in `classes`. A probe that errored (a network blip) keeps the last completed snapshot.
++ if (error && prev && prev.passed) {
++ const out = { ...prev, lastError: error, lastChecked: Date.now(), stale: true };
++ setCapabilities(name, out); return out;
++ }
++ const out = { provider: name, passed: !error, capabilities: caps, lastChecked: Date.now(), lastError: error };
++ // The probe is a health check too: the admin view is not "unhealthy" until the first 60 s tick.
++ state.lastHealthCheck[name] = Date.now(); state.lastHealthError[name] = error;
++ out.classes = error ? [] : providerClasses(name).filter((cls) => CLASS_REQUIREMENTS[cls].every((c) => caps[c]));
++ out.gate = gateMode();
++ setCapabilities(name, out);
++ return out;
++}
++
++async function probeAll() {
++ const out = {};
++ for (const name of ['local', ...vodStorage().REMOTE_PROVIDERS]) {
++ try { out[name] = await probe(name); } catch (err) {
++ // A probe that threw (a code or SDK fault) is recorded, never silent, and keeps the last snapshot.
++ const prev = capabilities(name);
++ out[name] = { ...prev, provider: name, lastError: err.message, lastChecked: Date.now(), stale: !!prev.passed };
++ setCapabilities(name, out[name]);
++ }
++ }
++ return out;
++}
++
++async function ensureProbeObject(name, client, bucket) {
++ loadSdk();
++ try {
++ await client.send(new S3.HeadObjectCommand({ Bucket: bucket, Key: PROBE_KEY }));
++ return;
++ } catch { /* missing — create it */ }
++ await client.send(new S3.PutObjectCommand({ Bucket: bucket, Key: PROBE_KEY, Body: PROBE_BYTES, ContentType: 'application/octet-stream' }));
++}
++
++async function sendRange(name, client, bucket, key, start, end) {
++ loadSdk();
++ try {
++ const r = await client.send(new S3.GetObjectCommand({ Bucket: bucket, Key: key, Range: `bytes=${start}-${end}` }));
++ const status = r.$metadata && r.$metadata.httpStatusCode;
++ try { r.body && r.body.destroy && r.body.destroy(); } catch { /* */ }
++ return { status };
++ } catch (err) {
++ return { status: err?.$metadata?.httpStatusCode || null };
++ }
++}
++
++async function sendIfRange(name, client, bucket, key, etag) {
++ if (!etag) return null;
++ loadSdk();
++ try {
++ const r = await client.send(new S3.GetObjectCommand({ Bucket: bucket, Key: key, Range: 'bytes=0-0', IfRange: etag }));
++ try { r.body && r.body.destroy && r.body.destroy(); } catch { /* */ }
++ return true;
++ } catch { return false; }
++}
++
++/** Cheap per-minute live health: one HeadBucket per remote provider (a HEAD, one Class B op on R2
++ * and free on B2). Flips the signals module's view of the provider. */
++async function liveHealth(name) {
++ const vs = vodStorage();
++ if (!isConfigured(name)) return { configured: false };
++ if (name === 'local') return { configured: true, healthy: true };
++ const started = Date.now();
++ try {
++ loadSdk();
++ await vs.probeProvider(name); // HeadBucket; also flips vod-storage's providerHealthy
++ state.lastHealthCheck[name] = Date.now();
++ state.lastHealthError[name] = null;
++ const ms = state.lastHealthCheck[name] - started;
++ observe(name, 'head', true, ms);
++ return { configured: true, healthy: true, lastCheck: state.lastHealthCheck[name], latencyMs: ms };
++ } catch (err) {
++ state.lastHealthCheck[name] = Date.now();
++ state.lastHealthError[name] = err.message || String(err);
++ const ms = state.lastHealthCheck[name] - started;
++ observe(name, 'head', false, ms);
++ return { configured: true, healthy: false, lastCheck: state.lastHealthCheck[name], error: state.lastHealthError[name], latencyMs: ms };
++ }
++}
++
++/** Feed one observation into the rolling signals and the Prometheus histogram/counter (lazy requires: this
++ * module is required by the router, so a top-level require of metrics-binding would be circular). */
++function observe(provider, op, ok, ms) {
++ try { const s = require('./signals'); ok ? s.recordSuccess(provider, op, ms) : s.recordFailure(provider, op, ms); } catch { /* */ }
++ try {
++ const mb = require('./metrics-binding');
++ ok ? mb.observeLatency(provider, op, ms) : mb.observeError(provider, op);
++ } catch { /* */ }
++}
++
++function healthStatus() {
++ const out = {};
++ for (const name of vodStorage().REMOTE_PROVIDERS) {
++ out[name] = {
++ configured: isConfigured(name),
++ healthy: state.lastHealthError[name] == null && state.lastHealthCheck[name] != null,
++ lastCheck: state.lastHealthCheck[name] || null,
++ lastError: state.lastHealthError[name] || null,
++ capabilities: capabilities(name),
++ eligibleClasses: eligibleClasses(name),
++ };
++ }
++ out.local = { configured: true, healthy: true, eligibleClasses: ['regional-cache'] };
++ return out;
++}
++
++/** Start the cheap per-minute live health check and an hourly capability probe. */
++function startHealthLoop({ log = console, intervalMs = 60_000, probeEveryMs = 60 * 60_000 } = {}) {
++ if (state.started) return;
++ state.started = true;
++ const tick = async () => {
++ for (const name of vodStorage().REMOTE_PROVIDERS) {
++ if (!isConfigured(name)) continue;
++ const h = await liveHealth(name);
++ if (!h.healthy) log.warn && log.warn(`[placement] provider ${name} unhealthy: ${h.error}`);
++ }
++ };
++ state.timers.push(setInterval(tick, intervalMs));
++ if (state.timers[state.timers.length - 1].unref) state.timers[state.timers.length - 1].unref();
++ const probeTick = async () => {
++ for (const name of vodStorage().REMOTE_PROVIDERS) {
++ if (!isConfigured(name)) continue;
++ try { await probe(name); } catch (err) { log.warn && log.warn(`[placement] capability probe ${name} failed: ${err.message}`); }
++ }
++ };
++ state.timers.push(setInterval(probeTick, probeEveryMs));
++ if (state.timers[state.timers.length - 1].unref) state.timers[state.timers.length - 1].unref();
++}
++
++function stop() {
++ for (const t of state.timers) { try { clearInterval(t); } catch { /* */ } }
++ state.timers = [];
++ state.started = false;
++}
++
++function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {}; state.lastHealthError = {}; }
++
++module.exports = {
++ CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,
++ isConfigured, providerClasses, eligibleClasses, measuredClasses, gateMode, capabilities,
++ probe, probeAll, liveHealth, healthStatus,
++ startHealthLoop, stop, reset,
++ // Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
++ _setCapabilities: setCapabilities,
++};
\\ No newline at end of file
ddiff --git a/server/placement/router.js b/server/placement/router.js
nnew file mode 100644
iindex 0000000..dd8e298
---- /dev/null
++++ b/server/placement/router.js
@@@ -0,0 +1,277 @@
++/**
++ * OpenVibe.Media — the read router (placement F1, part 3).
++ *
++ * route(opts) -> { provider, key, url, reason, fallbacks }
++ *
++ * The only function every read path goes through. Decides which provider copy to serve a
++ * given object from, given the locations row(s), the per-provider signals (latency EWMA +
++ * breaker) and a sticky session key.
++ *
++ * Decision:
++ * 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.
++ * 4. Rank by preference tier, with the measured 5-minute EWMA latency breaking ties:
++ * playback : R2 with a verified present copy (popularity promotion put it there to move
++ * reads off the host; promoting a VOD also frees its local copy) > local > R2 > B2
++ * else : local > R2 > B2 (canonical); `download` puts B2 before R2 (free ops)
++ * A VOD or clip offers every verified cloud copy its object records (media_locations), so a
++ * VOD promoted to R2 keeps B2 as its fallback when R2's breaker opens.
++ * A lower tier displaces a higher one only when both have measurements and the lower
++ * tier's EWMA is more than TIE_BAND (±15 %) faster. An unmeasured copy never claims to
++ * be faster than a preferred one — that keeps local/b2 from whipsawing a hot R2 copy.
++ * 5. Sticky per session: the same `session` key gets the same provider for 30 s unless
++ * its breaker has opened, its copy is gone, or a better tier is available again.
++ * 6. Presign the chosen remote copy through vod-storage (which carries the presigned-URL
++ * LRU), falling through the ranked list when a presign fails. A local copy has no URL:
++ * the caller streams it from disk.
++ *
++ * Every decision increments `media_router_decisions_total{provider,purpose,reason}`.
++ */
++'use strict';
++
++const signals = require('./signals');
++const providers = require('./providers');
++
++const STICKY_TTL_MS = 30_000;
++const TIE_BAND = 0.15; // ±15% latency is treated as a tie
++const NEUTRAL_MS = 100; // a provider with no samples is assumed middling, never "fastest"
++
++const stickyByKey = new Map(); // sessionKey -> { provider, expiresAt }
++
++/** Purge expired sticky entries (called on every route to bound memory). */
++function pruneSticky(now = Date.now()) {
++ for (const [k, v] of stickyByKey) { if (v.expiresAt <= now) stickyByKey.delete(k); }
++}
++
++function stickyGet(session) {
++ if (!session) return null;
++ pruneSticky();
++ return stickyByKey.get(session) || null;
++}
++
++function stickyPut(session, provider) {
++ if (!session) return;
++ stickyByKey.set(session, { provider, expiresAt: Date.now() + STICKY_TTL_MS });
++}
++
++/** Resolve the location rows for one object/vod/clip into a normalised list. */
++async function locationsFor(opts) {
++ let rows = null;
++ if (Array.isArray(opts.locations)) rows = opts.locations;
++ else if (opts.object && Array.isArray(opts.object.locations)) rows = opts.object.locations;
++ if (rows) return rows;
++
++ const fs = require('fs');
++ if (opts.object && opts.object.id) {
++ const model = require('../objects/model');
++ rows = await model.listLocations(opts.object.id);
++ // The object's canonical copy is a location too, when no row records it as present and none says it is
++ // gone: a projected row still 'pending' verification must not hide the copy the object names.
++ const canon = opts.object.canonical_provider, canonKey = opts.object.canonical_key;
++ if (canon && canonKey) rows = withOwnCopy(rows, canon, canonKey, { canonical: true });
++ // A stale local row whose file is gone is not really present.
++ for (const r of rows) {
++ if (r.provider === 'local' && r.state === 'present' && r.key && !fs.existsSync(r.key)) r.state = 'missing';
++ }
++ return rows;
++ }
++ if (opts.vod || opts.clip) {
++ const vs = require('../vod/vod-storage');
++ const row = opts.vod || opts.clip;
++ const out = [];
++ try {
++ const localPath = vs.localPathForVod(row);
++ if (localPath && fs.existsSync(localPath)) out.push({ provider: 'local', state: 'present', key: localPath });
++ } catch { /* */ }
++ // Every cloud copy the row's object records (a tier move marks each copy it verified present:
++ // objects/model.js afterTierMove), not just the row's storage_provider: a VOD promoted to R2
++ // (storage_provider 'r2') still has its canonical B2 copy, and that is its fallback.
++ if (row.object_id) {
++ try {
++ const model = require('../objects/model');
++ for (const l of await model.listLocations(row.object_id)) {
++ if (l.provider !== 'local' && l.key) out.push(l);
++ }
++ } catch (err) { console.warn('[Router] locations of object', row.object_id, err.message); }
++ }
++ // The row names its own copy: routable unless its object records that copy as gone. A projected row still
++ // 'pending' verification (every offloaded VOD whose copy no tier move verified) does not hide it.
++ if (row.storage_key && row.storage_provider && row.storage_provider !== 'local') return withOwnCopy(out, row.storage_provider, row.storage_key);
++ return out;
++ }
++ return [];
++}
++
++/** Add the copy a row or object names itself (`provider`/`key`) unless a location already records it as present,
++ * or as missing/corrupt (then it is gone, whatever the row says). A pending location for it is replaced. */
++function withOwnCopy(rows, provider, key, extra = {}) {
++ const mine = rows.filter((r) => r.provider === provider);
++ if (mine.some((r) => r.state === 'present' || r.state === 'missing' || r.state === 'corrupt')) return rows;
++ 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' } = {}) {
++ 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;
++}
++
++/** Measured 5-minute EWMA (ms), or null when the provider has no samples. */
++function measuredMs(provider) {
++ const ms = signals.latencyEWMA(provider, 'presign') || signals.latencyEWMA(provider);
++ return ms > 0 ? ms : null;
++}
++
++function latencyOf(provider) {
++ return measuredMs(provider) || NEUTRAL_MS;
++}
++
++/**
++ * Preference tier (lower wins). Purpose decides only the order between the hot and canonical
++ * cloud copies; local is always preferred to a cloud copy that is not the hot cache.
++ */
++function tierOf(candidate, purpose) {
++ const p = candidate.provider;
++ if (p === 'local') return 1;
++ if (p === 'r2') {
++ // A verified R2 copy is the hot cache: popularity promotion put it there precisely so reads leave the
++ // host (the design's spill, decided by popularity until F2 measures the port).
++ if (purpose === 'playback' && candidate.verified_at) return 0;
++ return purpose === 'download' ? 3 : 2;
++ }
++ if (p === 'b2') return purpose === 'download' ? 2 : 3;
++ return 2;
++}
++
++/** Rank the candidates: by tier, with a measured latency advantage (> TIE_BAND) allowed to promote. */
++function rankCandidates(candidates, { purpose = 'playback' } = {}) {
++ return [...candidates].sort((a, b) => {
++ const ta = tierOf(a, purpose), tb = tierOf(b, purpose);
++ if (ta === tb) {
++ const ma = measuredMs(a.provider), mb = measuredMs(b.provider);
++ if (ma != null && mb != null) return ma - mb;
++ return 0;
++ }
++ const preferred = ta < tb ? a : b;
++ const other = ta < tb ? b : a;
++ const mp = measuredMs(preferred.provider), mo = measuredMs(other.provider);
++ // Only a measured, clearly faster copy may jump the preference order.
++ if (mp != null && mo != null && mo < mp * (1 - TIE_BAND)) return other === a ? -1 : 1;
++ return preferred === a ? -1 : 1;
++ });
++}
++
++/** Presign one remote candidate through vod-storage (which carries the presigned-URL LRU). */
++async function presignCandidate(provider, key, opts) {
++ const vs = require('../vod/vod-storage');
++ const overrides = {};
++ if (opts.contentType) overrides.contentType = opts.contentType;
++ if (opts.contentDisposition) overrides.contentDisposition = opts.contentDisposition;
++ const url = await vs.presignGet(provider, key, opts.expiresIn || 900, overrides);
++ return url || null;
++}
++
++const REASONS = {
++ 0: 'hot-cache',
++ 1: 'local-cheap',
++ 2: 'hot-tier',
++ 3: 'canonical-fallback',
++};
++
++/**
++ * The only function every read path uses.
++ * opts: { object | vod | clip | locations, purpose, session?, range?, presign?, expiresIn?,
++ * contentType?, contentDisposition? }
++ * returns: { provider, key, url, reason, fallbacks, candidates } — `fallbacks` are provider names, ranked;
++ * `candidates` the ranked copies with their keys ({ provider, key }), the chosen one first.
++ */
++async function route(opts = {}) {
++ const purpose = opts.purpose || 'playback';
++ 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.
++ const seen = new Set();
++ const candidates = [];
++ for (const l of locs) {
++ if (seen.has(l.provider)) continue;
++ if (!isHealthy(l.provider, { purpose })) continue;
++ seen.add(l.provider);
++ candidates.push(l);
++ }
++ const ranked = rankCandidates(candidates, { purpose });
++
++ // 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
++ // on the canonical fallback once the hot copy is back (an R2 re-warm after an eviction).
++ let ordered = ranked;
++ const sticky = stickyGet(session);
++ 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; }
++ else stickyByKey.delete(session);
++ }
++
++ // Materialise: keep the ranking, but let a failed presign push the decision down the list.
++ let primary = ordered[0] || null;
++ let url = null;
++ if (primary && opts.presign !== false && primary.provider !== 'local') {
++ for (const c of ordered) {
++ if (c.provider === 'local') break; // a local copy is streamed, not presigned
++ const got = await presignCandidate(c.provider, c.key, opts).catch(() => null);
++ if (got) { primary = c; url = got; break; }
++ }
++ }
++
++ 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')
++ : 'no-healthy-copy';
++
++ if (primary) stickyPut(session, primary.provider);
++
++ incMetric('media_router_decisions_total', { provider: primary ? primary.provider : 'none', purpose, reason });
++ const chain = [primary, ...ordered.filter((c) => c !== primary)].filter(Boolean).map((c) => ({ provider: c.provider, key: c.key }));
++ return primary
++ ? { provider: primary.provider, key: primary.key, url, reason, fallbacks, candidates: chain }
++ : { provider: null, key: null, url: null, reason, fallbacks: [], candidates: [] };
++}
++
++/**
++ * A viewer's sticky-routing key for one piece of media: the same viewer (address + browser) asking for
++ * the same object again within STICKY_TTL_MS gets the same provider. Hashed; never stored.
++ */
++function sessionFor(req, mediaId) {
++ const { clientIp } = require('../client-ip');
++ const h = require('crypto').createHash('sha256');
++ h.update(`${clientIp(req) || ''}|${(req.headers && req.headers['user-agent']) || ''}|${mediaId}`);
++ return h.digest('hex').slice(0, 24);
++}
++
++/** Optional metric hook (observability wires this up at boot). Avoids a hard require there. */
++let _inc = () => {};
++function bindMetric(fn) { _inc = fn || (() => {}); }
++function incMetric(name, labels) { try { _inc(name, labels); } catch { /* */ } }
++
++module.exports = {
++ route,
++ isHealthy,
++ rankCandidates,
++ tierOf,
++ locationsFor,
++ sessionFor,
++ stickyGet, stickyPut, pruneSticky,
++ bindMetric,
++ reset: () => { stickyByKey.clear(); },
++ STICKY_TTL_MS,
++ TIE_BAND,
++};
ddiff --git a/server/placement/signals.js b/server/placement/signals.js
nnew file mode 100644
iindex 0000000..e65db20
---- /dev/null
++++ b/server/placement/signals.js
@@@ -0,0 +1,203 @@
++/**
++ * OpenVibe.Media — per-provider rolling signals and a circuit breaker (placement F1, part 2).
++ *
++ * latencyEWMA(provider, op) current EWMA (ms) over the recent samples
++ * errorRate(provider, window) failures / total over a window ('10s'|'1m'|'5m'|'1h')
++ * p95(provider, window) observed p95 over a window (ms)
++ * recordSuccess(provider, op, ms) push one success sample (and a low-latency signal)
++ * recordFailure(provider, op, ms) push one failure sample
++ * breakerOpen(provider) true when the breaker is open (skip the provider)
++ * snapshot() a plain object for /metrics and the admin view
++ * reset() clear in-memory state (tests)
++ *
++ * 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
++ * 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
++ * re-initialises them from the cheap periodic HEAD probe (placement/providers.js).
++ */
++'use strict';
++
++const WINDOWS = ['10s', '1m', '5m', '1h'];
++const WINDOW_MS = { '10s': 10_000, '1m': 60_000, '5m': 5 * 60_000, '1h': 60 * 60_000 };
++const EWMA_ALPHA = 0.3; // ~few samples to converge, not so fast it whipsaws on noise
++const SAMPLE_LIMIT = 2000; // per provider × op ring buffer (memory bound)
++const BREAKER_OPEN_THRESHOLD = { errorRate1m: 0.5, minSamples1m: 5, p95Ms5m: 5000 };
++const BREAKER_HALF_OPEN_AFTER_MS = 60_000;
++
++// The clock is virtual only for tests (`_advance`), so the 60-second half-open window can be
++// exercised without sleeping: everything else reads Date.now().
++let clockSkewMs = 0;
++function nowMs() { return Date.now() + clockSkewMs; }
++
++/** Per-(provider, op): an EWMA of latency and a ring of recent (ok, ms) samples for error rate / p95. */
++function makeSignal() {
++ return { ewmaMs: 0, samples: [] }; // samples = [{ t, ok, ms }]
++}
++
++const state = new Map(); // key 'provider:op' -> signal
++const breakers = new Map(); // provider -> { openAt, halfOpenAt?, halfOpenInFlight? }
++const listeners = new Set(); // (provider) => void — fired when a breaker flips
++
++function key(provider, op) { return `${provider}:${op || 'op'}`; }
++
++function getSignal(provider, op) {
++ const k = key(provider, op);
++ let s = state.get(k);
++ if (!s) { s = makeSignal(); state.set(k, s); }
++ return s;
++}
++
++/** All samples for one provider, regardless of op. Used by the breaker stats so it sees every signal. */
++function allSamplesFor(provider) {
++ const out = [];
++ for (const [k, s] of state) {
++ if (!k.startsWith(provider + ':')) continue;
++ for (const sample of s.samples) out.push(sample);
++ }
++ return out;
++}
++
++/** EWMA averaged across all ops for the provider. */
++function ewmaFor(provider) {
++ let count = 0, weighted = 0;
++ for (const [k, s] of state) {
++ if (!k.startsWith(provider + ':')) continue;
++ if (s.ewmaMs > 0) { weighted += s.ewmaMs; count++; }
++ }
++ return count ? weighted / count : 0;
++}
++
++/** Push one (ok, ms) observation into the ring; trim by window + size. */
++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 });
++ if (s.samples.length > SAMPLE_LIMIT) s.samples.splice(0, s.samples.length - SAMPLE_LIMIT);
++ 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');
++ if (breakerState(provider) === 'closed' && stats5m.p95Ms >= BREAKER_OPEN_THRESHOLD.p95Ms5m) openBreaker(provider);
++}
++
++function recordSuccess(provider, op, ms) { push(provider, op, true, ms); }
++function recordFailure(provider, op, ms) { push(provider, op, false, ms || 0); }
++
++/** Stats over a window: total, errors, error rate, observed p95 (ms). Aggregates across all ops
++ * for the provider — the breaker cares about the provider as a whole, not per-op. */
++function statsFor(provider, _opUnused, windowKey) {
++ const windowMs = WINDOW_MS[windowKey] || WINDOW_MS['1m'];
++ const cutoff = nowMs() - windowMs;
++ let total = 0, errors = 0, latencies = [];
++ for (const sample of allSamplesFor(provider)) {
++ if (sample.t < cutoff) continue;
++ total++;
++ if (!sample.ok) errors++; else latencies.push(sample.ms);
++ }
++ latencies.sort((a, b) => a - b);
++ const percentile = (q) => (latencies.length ? latencies[Math.min(latencies.length - 1, Math.floor(latencies.length * q))] : 0);
++ const p50 = percentile(0.5), p95 = percentile(0.95);
++ return { total, errors, errorRate: total ? errors / total : 0, p50Ms: p50, p95Ms: p95 };
++}
++
++function latencyEWMA(provider, op) { return op ? getSignal(provider, op).ewmaMs : ewmaFor(provider); }
++function errorRate(provider, windowKey = '1m', _op) { return statsFor(provider, null, windowKey).errorRate; }
++function p50(provider, windowKey = '5m', _op) { return statsFor(provider, null, windowKey).p50Ms; }
++function p95(provider, windowKey = '5m', _op) { return statsFor(provider, null, windowKey).p95Ms; }
++/** Whether the provider is worth routing to at all: breaker closed. `providerHealthy` in vod-storage.js reads this. */
++function providerHealthy(name) { return !breakerOpen(name); }
++
++/** Breaker: closed (healthy), open (skipped), half-open (one trial allowed). */
++function breakerState(provider) {
++ const b = breakers.get(provider);
++ if (!b) return 'closed';
++ if (b.halfOpenInFlight) return 'half-open';
++ if (b.halfOpenAt && nowMs() >= b.halfOpenAt) return 'half-open';
++ if (b.openAt && nowMs() - b.openAt < BREAKER_HALF_OPEN_AFTER_MS) return 'open';
++ if (b.openAt) { b.halfOpenAt = nowMs(); return 'half-open'; }
++ return 'closed';
++}
++
++function breakerOpen(provider) { return breakerState(provider) === 'open'; }
++
++function openBreaker(provider) {
++ const b = breakers.get(provider) || {};
++ b.openAt = nowMs();
++ b.halfOpenAt = null;
++ b.halfOpenInFlight = false; // a failed trial is no longer in flight: the breaker is open again
++ breakers.set(provider, b);
++ fire(provider);
++}
++
++function closeBreaker(provider) {
++ if (!breakers.has(provider)) return;
++ breakers.delete(provider);
++ fire(provider);
++}
++
++function fire(provider) { for (const fn of listeners) { try { fn(provider); } catch { /* */ } } }
++function onBreakerChange(fn) { listeners.add(fn); return () => listeners.delete(fn); }
++
++function onSuccess(provider) {
++ const b = breakers.get(provider);
++ if (!b) return;
++ if (b.halfOpenInFlight) { closeBreaker(provider); }
++ else if (b.openAt && nowMs() - b.openAt >= BREAKER_HALF_OPEN_AFTER_MS) { closeBreaker(provider); }
++}
++
++function onFailure(provider) {
++ if (breakerState(provider) === 'half-open') { openBreaker(provider); return; }
++ if (breakerState(provider) === 'open') return;
++ const stats1m = statsFor(provider, null, '1m');
++ if (stats1m.total >= BREAKER_OPEN_THRESHOLD.minSamples1m && stats1m.errorRate >= BREAKER_OPEN_THRESHOLD.errorRate1m) { openBreaker(provider); }
++}
++
++/** Mark a half-open trial as in-flight so a concurrent call cannot also use the provider. */
++function halfOpenClaim(provider) {
++ const b = breakers.get(provider);
++ if (!b) return false;
++ if (breakerState(provider) !== 'half-open') return false;
++ if (b.halfOpenInFlight) return false;
++ b.halfOpenInFlight = true;
++ breakers.set(provider, b);
++ return true;
++}
++
++function snapshot() {
++ const providers = new Set();
++ for (const k of state.keys()) providers.add(k.split(':', 1)[0]);
++ const out = { breakers: {}, providers: {} };
++ for (const p of providers) {
++ out.breakers[p] = breakerState(p);
++ const w = {};
++ for (const win of WINDOWS) {
++ const st = statsFor(p, null, win);
++ w[win] = { total: st.total, errors: st.errors, errorRate: Number(st.errorRate.toFixed(4)), p50Ms: st.p50Ms, p95Ms: st.p95Ms };
++ }
++ out.providers[p] = { ewmaMs: Math.round(ewmaFor(p)), windows: w };
++ }
++ return out;
++}
++
++function reset() {
++ state.clear();
++ breakers.clear();
++ listeners.clear();
++ clockSkewMs = 0;
++}
++
++/** Test-only: move the virtual clock forward so the half-open window needs no real sleep. */
++function _advance(ms) { clockSkewMs += Number(ms) || 0; }
++
++module.exports = {
++ WINDOWS, BREAKER_OPEN_THRESHOLD, BREAKER_HALF_OPEN_AFTER_MS,
++ recordSuccess, recordFailure,
++ latencyEWMA, errorRate, p50, p95,
++ providerHealthy,
++ breakerState, breakerOpen,
++ halfOpenClaim, openBreaker, closeBreaker,
++ onBreakerChange, snapshot, reset,
++ _advance,
++};
\\ No newline at end of file
ddiff --git a/server/public/routes.js b/server/public/routes.js
iindex 0ce7d7e..4fdd0cf 100644
---- a/server/public/routes.js
++++ b/server/public/routes.js
@@@ -155,7 +155,9 @@ async function serveMediaRecord(kind, record, req, res) {
}
// Resolve the local file. Clips keep their own absolute path; VODs resolve
-- // by basename under VOD_PATH (legacy rows carry old absolute paths).
++ // by basename under VOD_PATH (legacy rows carry old absolute paths). A local copy is served from
++ // disk without asking the router: promoting a VOD to R2 frees its local copy, so a VOD still on disk
++ // has not been promoted, and a growing recording exists only here.
let filePath = null;
if (record.file_path) {
const candidates = kind === 'vod'
@@@ -169,7 +171,8 @@ async function serveMediaRecord(kind, record, req, res) {
// Offloaded (B2/R2) — redirect to a presigned object-store URL. Range
// requests are handled natively by the object store.
if (!filePath && vodStorage.isRemote(record)) {
-- const plan = await vodStorage.resolvePlayback(record);
++ const router = require('../placement/router');
++ const plan = await vodStorage.resolvePlayback(record, { session: router.sessionFor(req, `${kind}:${record.id}`) });
if (plan?.kind === 'redirect') {
res.set('Cache-Control', 'private, max-age=0');
res.set('X-Robots-Tag', 'noindex');
ddiff --git a/server/vod/orphans-report.js b/server/vod/orphans-report.js
iindex 2f3be8c..b45fd1a 100644
---- a/server/vod/orphans-report.js
++++ b/server/vod/orphans-report.js
@@@ -338,7 +338,7 @@ async function remoteHint(provider, key, refs) {
*/
async function buildStorageReport({ providers = null, list = null, listUploads = null, limit = 2000, now = Date.now() } = {}) {
const vs = vodStorage();
-- const provs = providers || ['b2', 'r2'].filter(p => vs.providerConfigured(p));
++ const provs = providers || vs.REMOTE_PROVIDERS.filter(p => vs.providerConfigured(p));
const lister = list || (async (p) => await vs.listObjects(p, ''));
const uploadLister = listUploads || (async (p) => await vs.listMultipartUploads(p));
const cap = Math.max(1, Number(limit) || 2000);
@@@ -395,7 +395,9 @@ async function buildStorageReport({ providers = null, list = null, listUploads =
entry.bytes = keys.reduce((n, k) => n + (Number(k.size) || 0), 0);
listed.set(p, new Map(keys.map(k => [k.key, k])));
const rrefs = await remoteReferences(p);
-- const orphans = keys.filter(k => !rrefs.wanted.has(k.key)).sort((a, b) => (Number(b.size) || 0) - (Number(a.size) || 0));
++ // The placement capability probe keeps one small object per bucket (placement/providers.js PROBE_PREFIX).
++ const PROBE = require('../placement/providers').PROBE_PREFIX;
++ const orphans = keys.filter(k => !rrefs.wanted.has(k.key) && !String(k.key).startsWith(PROBE)).sort((a, b) => (Number(b.size) || 0) - (Number(a.size) || 0));
report.totals.unreferenced_remote[p] = { keys: orphans.length, bytes: orphans.reduce((n, k) => n + (Number(k.size) || 0), 0) };
for (const k of orphans.slice(0, cap)) push('unreferenced_remote', { provider: p, key: k.key, size: Number(k.size) || 0, last_modified: k.last_modified || null, hint: await remoteHint(p, k.key, rrefs) });
if (orphans.length > cap) report.truncated.unreferenced_remote = true;
ddiff --git a/server/vod/vod-storage.js b/server/vod/vod-storage.js
iindex 74e8503..422a521 100644
---- a/server/vod/vod-storage.js
++++ b/server/vod/vod-storage.js
@@@ -66,7 +66,8 @@ const PROVIDER_ENV = {
},
};
--const REMOTE_PROVIDERS = ['b2', 'r2'];
++// Every remote the S3 engine knows: a membership set, never a read order (reads rank through the placement router).
++const REMOTE_PROVIDERS = Object.freeze(Object.keys(PROVIDER_ENV));
const KEY_PREFIX = 'vods/';
// ── Settings (media_settings, JSON values under storage_tier.*) ──
@@@ -115,7 +116,7 @@ const DEFAULTS = {
let sweepTimer = null;
let sweepRunning = false;
const clients = {}; // providerName → S3Client
--const providerHealthy = {}; // providerName → bool
++const providerHealthy = {}; // providerName → bool (kept for legacy readers; placement/signals is the source of truth for routing)
// The policy is revisioned configuration (server/vod/tier-config.js, openvibe-shared/config): read from
// memory, changed only through a validated revision (setSettings, the admin config routes).
@@@ -220,9 +221,16 @@ function clientFor(name) {
return clients[name];
}
--/** Configured, and not known to be down (the last HeadBucket or readiness probe did not fail). */
++/** Configured, and not known to be down. The placement/signals circuit breaker is the source of truth now:
++ * if it has opened for this provider, we skip the provider. The boot-time `providerHealthy` map is kept
++ * for legacy readers and as the initial state until the first probe runs. */
function providerAvailable(name) {
-- return providerConfigured(name) && providerHealthy[name] !== false;
++ if (!providerConfigured(name)) return false;
++ if (providerHealthy[name] === false) return false;
++ // Continuous health, not boot-only: the placement circuit breaker (fed by every presign, HEAD
++ // and the per-minute probe) decides whether a provider is served from right now.
++ try { if (!require('../placement/signals').providerHealthy(name)) return false; } catch { /* signals not booted yet */ }
++ return true;
}
function bucketFor(name) {
@@@ -268,15 +276,28 @@ function isRemote(vod) {
async function headObject(provider, key) {
const client = clientFor(provider);
if (!client) return null;
++ const started = Date.now();
try {
const res = await client.send(new S3.HeadObjectCommand({ Bucket: PROVIDER_ENV[provider].bucket, Key: key }));
++ observeProvider(provider, 'head', true, Date.now() - started);
return { size: Number(res.ContentLength || 0), etag: res.ETag || null };
} catch (err) {
if (err?.$metadata?.httpStatusCode === 404 || err?.name === 'NotFound' || err?.name === 'NoSuchKey') return null;
++ observeProvider(provider, 'head', false, Date.now() - started);
throw err;
}
}
++/** One provider round-trip into the rolling signals (placement/signals.js) and the Prometheus
++ * histogram/counter (placement/metrics-binding.js). Lazy requires: neither is loaded in every context. */
++function observeProvider(provider, op, ok, ms) {
++ try { const s = require('../placement/signals'); ok ? s.recordSuccess(provider, op, ms) : s.recordFailure(provider, op, ms); } catch { /* placement not loaded */ }
++ try {
++ const mb = require('../placement/metrics-binding');
++ ok ? mb.observeLatency(provider, op, ms) : mb.observeError(provider, op);
++ } catch { /* */ }
++}
++
/** Deadline for one upload: a floor plus the time the file takes at the minimum acceptable throughput. */
function uploadTimeoutMs(bytes, settings = getSettings()) {
const floor = Number(settings.uploadTimeoutFloorMs) || DEFAULTS.uploadTimeoutFloorMs;
@@@ -447,21 +468,42 @@ async function listMultipartUploads(provider, prefix = '') {
/**
* A presigned GET URL. `overrides` { contentType, contentDisposition } become the response headers the store
* answers with (the object tiering passes what /o/:id would have sent, so an attachment stays one).
++ *
++ * Wraps the LRU (server/placement/presign-cache.js): the URL is short-lived but the cache hit rate is
++ * high in steady state (one presign per object per minute), and dropping it cuts R2 Class B costs.
*/
async function presignGet(provider, key, expiresInSeconds = 900, overrides = {}) {
const client = clientFor(provider);
if (!client) return null;
++ // LRU in front: a hit short-circuits the SDK call entirely.
++ try {
++ const cache = require('../placement/presign-cache');
++ const hit = cache.get(provider, key, overrides.contentType || '', overrides.contentDisposition || '', expiresInSeconds);
++ if (hit) return hit.url;
++ } catch { /* cache not loaded */ }
loadSdk();
++ const started = Date.now();
// Force the response MIME so the browser plays the media even when the stored
// object metadata is generic (legacy clips were uploaded as octet-stream).
const ext = path.extname(key || '').toLowerCase();
const mime = overrides.contentType || { '.webm': 'video/webm', '.mp4': 'video/mp4', '.mkv': 'video/x-matroska', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.webp': 'image/webp' }[ext];
-- return Presigner.getSignedUrl(client, new S3.GetObjectCommand({
-- Bucket: PROVIDER_ENV[provider].bucket,
-- Key: key,
-- ...(mime ? { ResponseContentType: mime } : {}),
-- ...(overrides.contentDisposition ? { ResponseContentDisposition: overrides.contentDisposition } : {}),
-- }), { expiresIn: expiresInSeconds });
++ let url;
++ try {
++ url = await Presigner.getSignedUrl(client, new S3.GetObjectCommand({
++ Bucket: PROVIDER_ENV[provider].bucket,
++ Key: key,
++ ...(mime ? { ResponseContentType: mime } : {}),
++ ...(overrides.contentDisposition ? { ResponseContentDisposition: overrides.contentDisposition } : {}),
++ }), { expiresIn: expiresInSeconds });
++ } catch (err) {
++ observeProvider(provider, 'presign', false, Date.now() - started);
++ throw err;
++ }
++ observeProvider(provider, 'presign', true, Date.now() - started);
++ try {
++ require('../placement/presign-cache').put(provider, key, overrides.contentType || '', overrides.contentDisposition || '', expiresInSeconds, url);
++ } catch { /* cache not loaded */ }
++ return url;
}
// ── Playback resolution ──────────────────────────────────────
@@@ -469,27 +511,30 @@ async function presignGet(provider, key, expiresInSeconds = 900, overrides = {})
/**
* Returns { kind: 'file', path } for local VODs or
* { kind: 'redirect', url, provider } for offloaded ones.
-- * Falls back through r2 → b2 → local if an object is missing.
++ *
++ * Routes through the placement router (server/placement/router.js): the choice of provider is no
++ * longer a property of `storage_provider` alone — it is the fastest healthy copy, with local
++ * 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) {
-- const provider = providerOf(vod);
-- const key = keyForVod(vod);
--
-- if (provider !== 'local') {
-- const order = provider === 'r2' ? ['r2', 'b2'] : ['b2', 'r2'];
-- for (const p of order) {
-- if (!providerConfigured(p) || providerHealthy[p] === false) continue;
++async function resolvePlayback(vod, { session = null, range = null } = {}) {
++ const router = require('../placement/router');
++ const decision = await router.route({ vod, purpose: 'playback', session, range });
++ 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 the
++ // ranked copies anyway so the caller still gets whichever one answers.
++ const key = keyForVod(vod);
++ for (const p of [decision.provider, ...(decision.fallbacks || [])]) {
try {
const url = await presignGet(p, key);
-- if (url) return { kind: 'redirect', url, provider: p };
-- } catch (err) {
-- console.warn(`[VodStorage] Presign failed for ${p}:${key}:`, err.message);
-- }
++ if (url) return { kind: 'redirect', url, provider: p, fallbacks: [decision.provider, ...(decision.fallbacks || [])].filter((x) => x !== p) };
++ } catch (err) { console.warn(`[VodStorage] Presign failed for ${p}:${key}:`, err.message); }
}
}
const local = localPathForVod(vod);
-- if (fs.existsSync(local)) return { kind: 'file', path: local };
++ if (fs.existsSync(local)) return { kind: 'file', path: local, fallbacks: decision.fallbacks || [] };
return null;
}
@@@ -497,6 +542,8 @@ async function resolvePlayback(vod) {
* A source ffmpeg/ffprobe can consume: local path, or a presigned https URL.
*/
async function resolveMediaSource(vod) {
++ // The provider choice is the placement router's, whether the copy is on disk or in the cloud:
++ // `resolvePlayback` asks it (clip cutting, thumbnails and derive all resolve a source this way).
const local = localPathForVod(vod);
if (fs.existsSync(local)) return { kind: 'file', value: local };
if (isRemote(vod)) {
@@@ -598,16 +645,23 @@ async function _moveToHot(vodId) {
return { ok: true, already: true };
}
-- const srcProvider = ['b2', 'r2'].find(p => providerConfigured(p));
-- if (!srcProvider) return { ok: false, error: 'No remote provider configured' };
++ // Which remote copy to pull down is the router's call (purpose `download`): B2 first (free ops,
++ // no request ceiling spent), R2 when that is the only copy, health and breaker respected.
++ const decision = await require('../placement/router').route({ vod, purpose: 'download' });
++ // No early refusal when the router finds no present copy: a stale location row is exactly the case the walk
++ // below exists for (it HEADs every configured remote).
++ if (!REMOTE_PROVIDERS.some((p) => providerConfigured(p))) return { ok: false, error: 'No remote provider configured' };
try {
loadSdk();
-- let provider = providerOf(vod) === 'r2' ? 'r2' : 'b2';
-- let head = await headObject(provider, key);
-- if (!head) {
-- provider = provider === 'r2' ? 'b2' : 'r2';
-- head = providerConfigured(provider) ? await headObject(provider, key) : null;
++ // Walk the router's ranked copies, then any other configured remote (the row's
++ // `storage_provider` is not proof that this is the copy that still exists): the first one
++ // that actually answers with the object wins.
++ let provider = null, head = null;
++ for (const p of [...new Set([decision.provider, ...(decision.fallbacks || []), ...REMOTE_PROVIDERS])]) {
++ if (!p || !providerConfigured(p)) continue;
++ head = await headObject(p, key);
++ if (head) { provider = p; break; }
}
if (!head) return { ok: false, error: 'Object missing from remote storage' };
@@@ -668,7 +722,7 @@ async function _promoteToR2(vodId) {
if (!vod || !vod.file_path) return { ok: false, error: 'VOD not found' };
if (vod.is_recording) return { ok: false, error: 'VOD is currently recording' };
if (await _held(vod)) return { ok: false, held: true, error: 'VOD is under a retention hold' };
-- if (!providerConfigured('r2') || providerHealthy.r2 === false) return { ok: false, error: 'R2 not available' };
++ if (!providerAvailable('r2')) return { ok: false, error: 'R2 not available' };
if (!providerConfigured('b2')) return { ok: false, error: 'B2 not configured' };
const key = keyForVod(vod);
@@@ -690,7 +744,7 @@ async function _promoteToR2(vodId) {
else await copyBetweenProviders('b2', 'r2', key);
}
-- await _moved(vodId, ['b2', 'r2'], async () => await db.run("UPDATE vods SET storage_provider = 'r2', storage_key = ? WHERE id = ?", [key, vodId]));
++ await _moved(vodId, REMOTE_PROVIDERS, async () => await db.run("UPDATE vods SET storage_provider = 'r2', storage_key = ? WHERE id = ?", [key, vodId]));
// Popular VODs live in R2+B2; free the local copy
let freed = 0;
@@@ -919,6 +973,9 @@ async function emitStorageEvent(event, kind, data, settings = getSettings()) {
const last = alertLastSentAt.get(cooldownKey) || 0;
if (event === 'storage.alert' && Date.now() - last < settings.alertCooldownMs) return false;
alertLastSentAt.set(cooldownKey, Date.now());
++ if (event === 'storage.alert') {
++ try { require('../placement/metrics-binding').alert(kind); } catch { /* */ }
++ }
const payload = { kind, ...data, at: new Date().toISOString() };
(event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);
// No database state changes here (the cooldown is in memory): the outbox row is the whole
@@@ -1067,7 +1124,7 @@ async function runSweep() {
}
// 2) R2 promotion for popular VODs
-- if (settings.r2Enabled && providerConfigured('r2') && providerHealthy.r2 !== false) {
++ if (settings.r2Enabled && providerAvailable('r2')) {
const popular = await db.all(`
SELECT id, view_count, last_accessed_at FROM vods
WHERE COALESCE(storage_provider, 'local') IN ('local', 'b2')
@@@ -1251,11 +1308,26 @@ function stop() {
// ── Status ───────────────────────────────────────────────────
--// List prices used for the admin cost estimate (storage only; egress noted).
--const CLOUD_PRICING = {
-- b2: { storagePerGbMonth: 0.006, egressPerGb: 0.01, freeGb: 0, egressNote: 'First 3× storage free/day, then $0.01/GB' },
-- r2: { storagePerGbMonth: 0.015, egressPerGb: 0, freeGb: 10, egressNote: 'Egress is free' },
--};
++// The cost table is the revisioned config `media.cost_tiers` (server/placement/cost-tiers.js).
++// Read through this helper so a price change is one revision, not a code change. The legacy
++// `CLOUD_PRICING` shape is kept as a derivation for the admin /metrics endpoint.
++function _readPricing() {
++ try {
++ const s = require('../placement/cost-tiers').settings();
++ return {
++ b2: { storagePerGbMonth: s.b2.storagePerGbMonth, egressPerGb: s.b2.egressPerGb, freeGb: s.b2.freeGb, egressNote: `First ${s.b2.egressFreeMultiplier}× storage free/month, then $${s.b2.egressPerGb}/GB` },
++ r2: { storagePerGbMonth: s.r2.standard.storagePerGbMonth, egressPerGb: s.r2.standard.egressPerGb, freeGb: s.r2.standard.freeGb, egressNote: 'Egress is free' },
++ local: { storagePerGbMonth: 0, egressPerGb: 0, freeGb: 0, egressNote: 'Prepaid bandwidth on the Media host' },
++ };
++ } catch {
++ // Fallback for the rare path where the cost-tiers config has not loaded yet (tests, drills).
++ return {
++ b2: { storagePerGbMonth: 0.00695, egressPerGb: 0.01, freeGb: 10, egressNote: 'First 3× storage free/month, then $0.01/GB' },
++ r2: { storagePerGbMonth: 0.015, egressPerGb: 0, freeGb: 10, egressNote: 'Egress is free' },
++ local: { storagePerGbMonth: 0, egressPerGb: 0, freeGb: 0, egressNote: 'Prepaid bandwidth on the Media host' },
++ };
++ }
++}
let _bucketUsageCache = null;
@@@ -1300,11 +1372,12 @@ async function getBucketUsage(force = false) {
/** Estimate monthly storage cost from bucket usage using list prices. */
function estimateCloudCosts(usage) {
-- const costs = { pricing: CLOUD_PRICING };
++ const pricing = _readPricing();
++ const costs = { pricing };
let totalStorage = 0;
for (const provider of REMOTE_PROVIDERS) {
const u = usage?.[provider];
-- const price = CLOUD_PRICING[provider];
++ const price = pricing[provider];
if (!u || !u.configured || u.error) { costs[provider] = null; continue; }
const gb = u.bytes / 1e9;
const billableGb = Math.max(0, gb - price.freeGb);
@@@ -1451,6 +1524,7 @@ async function getStatus() {
module.exports = {
cleanupStaleDownloads,
DEFAULTS,
++ REMOTE_PROVIDERS,
OFFLOADABLE_WHERE,
needsDrain,
drainSatisfied,
@@@ -1462,6 +1536,7 @@ module.exports = {
providerConfigured,
providerAvailable,
bucketFor,
++ clientFor,
endpointFor,
headObject,
uploadFile,
ddiff --git a/test/placement-cost-tiers.test.js b/test/placement-cost-tiers.test.js
nnew file mode 100644
iindex 0000000..e170bde
---- /dev/null
++++ b/test/placement-cost-tiers.test.js
@@@ -0,0 +1,59 @@
++'use strict';
++// Cost tiers (F1, part 5): the revisioned config namespace media.cost_tiers and its validation.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++ const dir = (n) => { const d = path.join(os.tmpdir(), `ov-media-cost-${n}-${process.pid}-${Math.random().toString(36).slice(2, 8)}`); fs.mkdirSync(d, { recursive: true }); return d; };
++ for (const k of ['VOD_PATH', 'CLIPS_PATH', 'PASTES_PATH', 'THUMBNAILS_PATH', 'FILES_PATH', 'OBJECTS_PATH', 'ASSETS_PATH']) process.env[k] = dir(k);
++
++ // Use a per-test PGlite (the same harness the test runner gives every file).
++ const tmpPglite = dir('pglite');
++ process.env.MEDIA_PGLITE_DIR = tmpPglite;
++ process.env.DATABASE_URL = '';
++ const db = require('../server/db/database');
++ await db.initDb();
++ const cost = require('../server/placement/cost-tiers');
++
++ // Cross-key rule: the defaults pass.
++ const good = JSON.parse(JSON.stringify(cost.DEFAULTS));
++ assert.deepStrictEqual(cost.validate(good), true, 'defaults validate');
++
++ // Cross-key rule fires when standard is below IA (IA is cheaper).
++ const bad = JSON.parse(JSON.stringify(cost.DEFAULTS));
++ bad.r2.standard.storagePerGbMonth = 0.005;
++ let errs = cost.validate(bad);
++ assert.ok(Array.isArray(errs) && errs.length, 'cross-key rule fires when standard < IA');
++
++ // Boot the store and read the defaults: match the design exactly.
++ await cost.init({ log: { info() {}, warn() {}, error() {} } });
++ const settings = cost.settings();
++ assert.strictEqual(settings.r2.standard.storagePerGbMonth, 0.015, 'R2 Standard price matches docs/media-fabric.md §2');
++ assert.strictEqual(settings.r2.standard.egressPerGb, 0, 'R2 Standard egress is free');
++ assert.strictEqual(settings.r2.infrequentAccess.storagePerGbMonth, 0.01, 'R2 IA price matches design');
++ assert.strictEqual(settings.r2.infrequentAccess.minResidencyDays, 30, 'R2 IA 30-day minimum');
++ assert.strictEqual(settings.b2.egressFreeMultiplier, 3, 'B2 3x free egress multiplier');
++ assert.strictEqual(settings.b2.partnerEgressFree, false, 'B2 partnerEgressFree default is false (safer)');
++ assert.strictEqual(settings.b2.storagePerGbMonth, 0.00695, 'B2 storage $6.95/TB-month');
++ assert.strictEqual(settings.local.storagePerGbMonth, 0, 'local is $0/GB');
++
++ // estimateCloudCosts reads from this config: it produces a positive total.
++ const vodStorage = require('../server/vod/vod-storage');
++ const costs = vodStorage.estimateCloudCosts({ b2: { configured: true, bytes: 50e9, objects: 1 }, r2: { configured: true, bytes: 500e9, objects: 5 } });
++ assert.strictEqual(costs.pricing.b2.egressPerGb, 0.01, 'B2 egress from cost-tiers');
++ assert.strictEqual(costs.pricing.r2.egressPerGb, 0, 'R2 egress from cost-tiers (free)');
++ assert.ok(costs.totalStorageMonthly > 0, 'estimate produces a positive total');
++
++ // Apply a revision: pass a full values object so the merged shape validates.
++ const full = JSON.parse(JSON.stringify(cost.DEFAULTS));
++ full.r2.standard.storagePerGbMonth = 0.02;
++ const newRev = await cost.get().apply(full, { actor: { type: 'service', id: 'media-test' }, reason: 'price update' });
++ assert.ok(newRev && (newRev.revision || newRev.values), 'apply returned a revision');
++ assert.strictEqual(cost.settings().r2.standard.storagePerGbMonth, 0.02, 'new revision applied');
++
++ cost._reset();
++ console.log('placement-cost-tiers: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-metrics.test.js b/test/placement-metrics.test.js
nnew file mode 100644
iindex 0000000..a94c7ba
---- /dev/null
++++ b/test/placement-metrics.test.js
@@@ -0,0 +1,52 @@
++'use strict';
++// Placement metrics (F1, part 6): histogram, counters, breaker gauge — wired into the registry.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-plmetrics-'));
++ Object.assign(process.env, {
++ VOD_PATH: path.join(tmp, 'vods'), CLIPS_PATH: path.join(tmp, 'clips'), PASTES_PATH: path.join(tmp, 'pastes'),
++ THUMBNAILS_PATH: path.join(tmp, 'thumbnails'), FILES_PATH: path.join(tmp, 'files'), OBJECTS_PATH: path.join(tmp, 'objects'),
++ ASSETS_PATH: path.join(tmp, 'assets'), MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9',
++ RELEASE_COMMIT: 'abcdef1234567',
++ });
++ fs.mkdirSync(process.env.VOD_PATH, { recursive: true });
++ fs.mkdirSync(process.env.CLIPS_PATH, { recursive: true });
++ fs.mkdirSync(process.env.PASTES_PATH, { recursive: true });
++ fs.mkdirSync(process.env.THUMBNAILS_PATH, { recursive: true });
++ fs.mkdirSync(process.env.FILES_PATH, { recursive: true });
++ fs.mkdirSync(process.env.OBJECTS_PATH, { recursive: true });
++ fs.mkdirSync(process.env.ASSETS_PATH, { recursive: true });
++ for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL']) process.env[k] = '';
++
++ const metrics = require('openvibe-shared/metrics');
++ const registry = metrics.createRegistry();
++ const observability = require('../server/observability');
++ observability.domainMetrics(registry, { db: require('../server/db/database'), recorder: { activeCount: () => 0 }, events: { status: async () => ({ enabled: false }) } });
++
++ // Inc a few metrics through the binding and check they show up.
++ const binding = require('../server/placement/metrics-binding');
++ binding.inc('media_router_decisions_total', { provider: 'r2', purpose: 'playback', reason: 'fastest-healthy' });
++ binding.inc('media_router_decisions_total', { provider: 'b2', purpose: 'playback', reason: 'fastest-healthy' });
++ binding.inc('media_presign_cache_total', { result: 'hit' });
++ binding.inc('media_presign_cache_total', { result: 'miss' });
++ binding.inc('media_storage_alerts_total', { kind: 'drain_stalled' });
++ binding.observeLatency('r2', 'presign', 120);
++ binding.observeError('b2', 'presign');
++
++ const text = await registry.metricsAsync();
++ assert.ok(text.includes('media_router_decisions_total{provider="r2"'), 'router decision rendered');
++ assert.ok(text.includes('media_router_decisions_total{provider="b2"'), 'router decision (b2) rendered');
++ assert.ok(text.includes('media_presign_cache_total{result="hit"}'), 'presign cache hit rendered');
++ assert.ok(text.includes('media_presign_cache_total{result="miss"}'), 'presign cache miss rendered');
++ assert.ok(text.includes('media_storage_alerts_total{kind="drain_stalled"}'), 'storage alert rendered');
++ assert.ok(text.includes('media_provider_latency_seconds_bucket'), 'latency histogram rendered');
++ assert.ok(text.includes('media_provider_errors_total'), 'errors counter rendered');
++ assert.ok(text.includes('media_provider_breaker_open'), 'breaker gauge rendered');
++
++ console.log('placement-metrics: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-presign-cache.test.js b/test/placement-presign-cache.test.js
nnew file mode 100644
iindex 0000000..6806e95
---- /dev/null
++++ b/test/placement-presign-cache.test.js
@@@ -0,0 +1,40 @@
++'use strict';
++// Presigned-URL LRU (F1, part 4): hit, miss, expiry, bounded.
++const assert = require('assert');
++
++(async () => {
++ const cache = require('../server/placement/presign-cache');
++ let incs = [];
++ cache.bindMetric((name, labels) => incs.push({ name, labels }));
++ cache.clear();
++
++ assert.strictEqual(cache.get('b2', 'k1', 'video/webm', '', 900), null, 'first get → miss');
++ cache.put('b2', 'k1', 'video/webm', '', 900, 'https://example.com/signed');
++ assert.strictEqual(cache.get('b2', 'k1', 'video/webm', '', 900).url, 'https://example.com/signed', 'second get → hit');
++ assert.strictEqual(cache.size(), 1, 'one entry');
++
++ // Different disposition / content-type → different cache key.
++ assert.strictEqual(cache.get('b2', 'k1', 'video/webm', 'attachment; filename="x"', 900), null, 'different disposition → miss');
++
++ // Bound: fill to MAX_ENTRIES and check eviction.
++ for (let i = 0; i < cache.MAX_ENTRIES + 100; i++) cache.put('r2', `k${i}`, '', '', 900, `https://example/${i}`);
++ assert.ok(cache.size() <= cache.MAX_ENTRIES, `bounded: ${cache.size()} <= ${cache.MAX_ENTRIES}`);
++
++ // Expiry: put a TTL with a very short window. After 60 s safety margin, the entry is dropped.
++ cache.clear();
++ cache.put('r2', 'kexp', '', '', 60, 'soon-expired');
++ await new Promise((r) => setTimeout(r, 100)); // safety margin is 60 s; we can't wait 60 s in a test, but we can simulate by calling put with ttl=0 and waiting briefly.
++ cache.put('r2', 'kexp2', '', '', 1, 'expired-now');
++ await new Promise((r) => setTimeout(r, 1100));
++ assert.strictEqual(cache.get('r2', 'kexp2', '', '', 1), null, 'expired entry → miss and dropped');
++
++ // Metrics were incremented: at least one hit and one miss.
++ const hits = incs.filter((i) => i.name === 'media_presign_cache_total' && i.labels.result === 'hit').length;
++ const misses = incs.filter((i) => i.name === 'media_presign_cache_total' && i.labels.result === 'miss').length;
++ assert.ok(hits >= 1, `hits recorded: ${hits}`);
++ assert.ok(misses >= 1, `misses recorded: ${misses}`);
++
++ cache.clear();
++ console.log('placement-presign-cache: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-providers.test.js b/test/placement-providers.test.js
nnew file mode 100644
iindex 0000000..2deef0e
---- /dev/null
++++ b/test/placement-providers.test.js
@@@ -0,0 +1,76 @@
++'use strict';
++// Provider registry (F1, part 1): declared classes, capability snapshot, eligibility. The actual
++// network probe is exercised in the boot path (server/index.js calls probeAll()), which a real
++// boot of the server covers; here we verify the in-memory contract.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-plprov-'));
++ Object.assign(process.env, {
++ VOD_PATH: path.join(tmp, 'vods'), CLIPS_PATH: path.join(tmp, 'clips'), PASTES_PATH: path.join(tmp, 'pastes'),
++ THUMBNAILS_PATH: path.join(tmp, 'thumbnails'), FILES_PATH: path.join(tmp, 'files'), OBJECTS_PATH: path.join(tmp, 'objects'),
++ ASSETS_PATH: path.join(tmp, 'assets'), MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9',
++ RELEASE_COMMIT: 'abcdef1234567',
++ });
++ for (const k of ['VOD_PATH', 'CLIPS_PATH', 'PASTES_PATH', 'THUMBNAILS_PATH', 'FILES_PATH', 'OBJECTS_PATH', 'ASSETS_PATH']) fs.mkdirSync(process.env[k], { recursive: true });
++ for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL']) process.env[k] = '';
++
++ const providers = require('../server/placement/providers');
++ providers.reset();
++
++ // Local is always regional-cache.
++ assert.deepStrictEqual(providers.providerClasses('local'), ['regional-cache']);
++ assert.deepStrictEqual(providers.eligibleClasses('local'), ['regional-cache']);
++
++ // Capability requirements per class.
++ assert.ok(providers.CLASS_REQUIREMENTS['online-canonical'].length > 0);
++ assert.ok(providers.CLASS_REQUIREMENTS['online-hot'].length > 0);
++ assert.deepStrictEqual(providers.CLASS_REQUIREMENTS['regional-cache'], []);
++
++ // Capability snapshot before a probe: passed is null.
++ const before = providers.capabilities('b2');
++ assert.strictEqual(before.passed, null);
++ assert.deepStrictEqual(before.capabilities, {});
++
++ // Inject a known capability result.
++ providers._setCapabilities && providers._setCapabilities('b2', { provider: 'b2', passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true }, lastChecked: Date.now(), lastError: null });
++ providers._setCapabilities && providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: true }, lastChecked: Date.now(), lastError: null });
++
++ // The classes are the design's assignment (docs/media-fabric.md §2): B2 canonical, R2 hot, local edge.
++ assert.deepStrictEqual(providers.providerClasses('b2'), ['online-canonical']);
++ assert.deepStrictEqual(providers.providerClasses('r2'), ['online-hot']);
++ assert.ok(providers.eligibleClasses('b2').includes('online-canonical'), 'b2 canonical class enabled');
++ assert.ok(providers.eligibleClasses('r2').includes('online-hot'), 'r2 hot class enabled');
++ assert.ok(!providers.eligibleClasses('r2').includes('online-canonical'), 'r2 NOT in canonical class (hot only)');
++
++ // Report mode (the default): a missing capability is recorded but takes nothing out of routing.
++ providers._setCapabilities('b2', { provider: 'b2', passed: true, capabilities: { range: true, range_206: true, etag: false, if_range: false }, lastChecked: Date.now(), lastError: null });
++ assert.strictEqual(providers.gateMode(), 'report');
++ assert.deepStrictEqual(providers.eligibleClasses('b2'), ['online-canonical'], 'report mode: B2 keeps routing');
++ assert.deepStrictEqual(providers.measuredClasses('b2'), [], 'report mode still shows what the probe found');
++ // Enforce mode: a capability the class needs is missing → the provider leaves that class.
++ process.env.MEDIA_CAPABILITY_GATE = 'enforce';
++ assert.deepStrictEqual(providers.eligibleClasses('b2'), [], 'no if_range/etag → B2 leaves online-canonical');
++ providers._setCapabilities('b2', { provider: 'b2', passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true }, lastChecked: Date.now(), lastError: null });
++ assert.ok(providers.eligibleClasses('b2').includes('online-canonical'), 'passing again restores the class');
++
++ // A probe that completed and found range broken removes the class; a probe that could not complete (no
++ // snapshot yet) decides nothing: live health is the breaker's job.
++ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: false, range_206: false }, lastChecked: Date.now(), lastError: null });
++ assert.deepStrictEqual(providers.eligibleClasses('r2'), [], 'failed capability removes all classes');
++ providers._setCapabilities('r2', { provider: 'r2', passed: false, capabilities: {}, lastChecked: Date.now(), lastError: 'timeout' });
++ assert.deepStrictEqual(providers.eligibleClasses('r2'), ['online-hot'], 'an incomplete probe removes nothing');
++ delete process.env.MEDIA_CAPABILITY_GATE;
++
++ // isConfigured: false when env is empty, true for local.
++ assert.strictEqual(providers.isConfigured('local'), true);
++ assert.strictEqual(providers.isConfigured('b2'), false);
++ assert.strictEqual(providers.isConfigured('r2'), false);
++
++ providers.reset();
++ console.log('placement-providers: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-read-paths.test.js b/test/placement-read-paths.test.js
nnew file mode 100644
iindex 0000000..4207970
---- /dev/null
++++ b/test/placement-read-paths.test.js
@@@ -0,0 +1,136 @@
++'use strict';
++// Every former read path now goes through the placement router (F1, part 3, end-of-task check):
++// `resolvePlayback` (vod-storage.js), `/o/:id` (server/objects/routes.js), `resolveSource` (jobs/derive.js),
++// clip cutting (vod/clips-routes.js), clip jobs (vod/clip-jobs.js), duration reconcile (vod/duration-reconcile.js).
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++ const dir = (n) => { const d = path.join(os.tmpdir(), `ov-media-rp-${n}-${process.pid}-${Math.random().toString(36).slice(2, 8)}`); fs.mkdirSync(d, { recursive: true }); return d; };
++ Object.assign(process.env, {
++ VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), PASTES_PATH: dir('pastes'),
++ THUMBNAILS_PATH: dir('thumbnails'), FILES_PATH: dir('files'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),
++ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9', RELEASE_COMMIT: 'abcdef1234567',
++ });
++ for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL']) process.env[k] = '';
++
++ const router = require('../server/placement/router');
++ router.reset();
++
++ let routerCalls = [], routerSessions = [];
++ const realRoute = router.route;
++ router.route = async function (opts) {
++ routerCalls.push(opts.purpose || 'unknown');
++ routerSessions.push(opts.session || null);
++ return await realRoute(opts);
++ };
++
++ // Init the necessary config stores that the read paths expect to be loaded.
++ const db = require('../server/db/database');
++ db.getDb();
++ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });
++ await require('../server/vod/tier-config').init(require('../server/vod/vod-storage').DEFAULTS);
++ await require('../server/objects/tier-policy').init();
++ await require('../server/placement/cost-tiers').init({ log: { info() {}, warn() {}, error() {} } });
++
++ // Build a VOD row that's local-only.
++ const vodStorage = require('../server/vod/vod-storage');
++ const vodPath = path.join(process.env.VOD_PATH, 'vod-1.webm');
++ fs.writeFileSync(vodPath, Buffer.alloc(1024, 0));
++ const inserted = await db.run(`INSERT INTO vods (app_id, file_path, file_size, storage_provider, storage_tier) VALUES (?, ?, ?, ?, ?) RETURNING id`,
++ ['live', vodPath, 1024, 'local', 'hot']);
++ const vodId = inserted.lastInsertRowid || (inserted.rows && inserted.rows[0] && inserted.rows[0].id) || (inserted[0] && inserted[0].id);
++
++ // (1) vod-storage.resolvePlayback goes through the router.
++ routerCalls = [];
++ const plan = await vodStorage.resolvePlayback({ id: vodId, file_path: vodPath, storage_provider: 'local', storage_tier: 'hot' });
++ assert.ok(plan && plan.kind === 'file', `local resolves to a file: ${JSON.stringify(plan)}`);
++ assert.ok(routerCalls.includes('playback'), `resolvePlayback called the router: ${routerCalls.join(',')}`);
++
++ // (2) duration-reconcile.sourceFor: when local is gone, it falls through to resolvePlayback → router.
++ // We don't need to assert on the resolved source — that depends on which providers are configured.
++ // The router is invoked regardless.
++ const dr = require('../server/vod/duration-reconcile');
++ fs.unlinkSync(vodPath);
++ routerCalls = [];
++ await dr.sourceFor({ id: vodId, file_path: vodPath, storage_provider: 'b2', storage_tier: 'hot', storage_key: 'vods/vod-1.webm' });
++ assert.ok(routerCalls.includes('playback'), `sourceFor called the router when remote-only: ${routerCalls.join(',')}`);
++ // Re-create the local file for the rest of the tests.
++ fs.writeFileSync(vodPath, Buffer.alloc(1024, 0));
++
++ // (3) clip cutting path: clip-jobs.js calls resolveMediaSource → resolvePlayback → router. We
++ // exercise resolveMediaSource directly on a remote VOD so the router is consulted.
++ const clipJobs = require('../server/vod/clip-jobs');
++ fs.unlinkSync(vodPath);
++ routerCalls = [];
++ const ms = await vodStorage.resolveMediaSource({ id: vodId, file_path: vodPath, storage_provider: 'b2', storage_tier: 'hot', storage_key: 'vods/vod-1.webm' });
++ assert.ok(routerCalls.includes('playback'), `resolveMediaSource → resolvePlayback → router: ${routerCalls.join(',')}`);
++ fs.writeFileSync(vodPath, Buffer.alloc(1024, 0));
++
++ // (4) jobs/derive.js: resolveSource goes through the router when it's a native object.
++ const derive = require('../server/jobs/derive');
++ const model = require('../server/objects/model');
++ // Create a native object + a local location.
++ const objId = `obj-${Date.now()}`;
++ const namespace = 'media';
++ await db.run(`INSERT INTO media_objects (id, app_id, namespace, kind, lifecycle_status, visibility, mime_type, size_bytes, canonical_provider, canonical_key, created_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ov_now())`, [objId, 'live', namespace, 'asset', 'ready', 'public', 'video/webm', 1024, 'local', vodPath]);
++ await db.run(`INSERT INTO media_locations (object_id, provider, state, key, size_bytes, verified_at) VALUES (?, ?, ?, ?, ?, ov_now())`,
++ [objId, 'local', 'present', vodPath, 1024]);
++ routerCalls = [];
++ const source = await derive.resolveSource({ id: objId, kind: 'asset', mime_type: 'video/webm', canonical_key: vodPath });
++ assert.ok(source && source.input === vodPath, `derive.resolveSource local: ${JSON.stringify(source)}`);
++ assert.ok(routerCalls.includes('derive'), `derive.resolveSource called the router with purpose=derive: ${routerCalls.join(',')}`);
++
++ // (5) objects/routes.js GET /o/:id — bring up the public router and stub presignGet.
++ const originalPresign = vodStorage.presignGet;
++ vodStorage.presignGet = async () => 'https://example.com/signed';
++ try {
++ const objectsRoutes = require('../server/objects/routes');
++ const express = require('express');
++ const app = express();
++ app.use('/o', objectsRoutes.publicRouter);
++ const server = await new Promise((resolve) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); });
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const fetch = (url, opts = {}) => new Promise((resolve, reject) => {
++ const lib = url.startsWith('https') ? require('https') : require('http');
++ lib.get(url, opts, (res) => { let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, headers: res.headers, body: b })); }).on('error', reject);
++ });
++
++ routerCalls = [];
++ const r = await fetch(`${base}/o/${objId}`);
++ assert.ok(r.status === 200 || r.status === 302, `native /o/:id returned 200 (local stream) or 302 (presign): got ${r.status}`);
++ assert.ok(routerCalls.includes('playback'), `/o/:id called the router: ${routerCalls.join(',')}`);
++ assert.ok(routerSessions.some((s) => typeof s === 'string' && s.length >= 16), `/o/:id passes the viewer's session: ${routerSessions.join(',')}`);
++ server.close();
++ } finally {
++ vodStorage.presignGet = originalPresign;
++ }
++
++ // (6) Grep the repo for any remaining hard-coded provider order in a read path. The only legitimate
++ // `['r2','b2']` / `['b2','r2']` literals are: placement internals (providers, metrics-binding),
++ // REMOTE_PROVIDERS constant, verify-job (verification, not reads), the readiness check, and the
++ // orphans-report script — none of those are read paths.
++ // (6) The end-of-task grep: zero hard-coded provider orders anywhere under server/. The read
++ // paths ask the router; the router ranks by class/tier, never by a literal provider list.
++ const { execSync } = require('child_process');
++ const grepFiles = (pattern) => {
++ try { return execSync(`grep -rln --include='*.js' -E "${pattern}" server/`, { cwd: path.join(__dirname, '..') }).toString(); }
++ catch (err) { if (err.status === 1) return ''; throw err; } // grep exits 1 on no match
++ };
++ // Quoted literals, single or double: ['r2','b2'], [ "b2", "r2" ] (an unquoted pattern matched nothing).
++ const lists = grepFiles("\\[ *['\\\"](r2|b2)['\\\"] *, *['\\\"](b2|r2)['\\\"] *\\]").trim();
++ const orders = grepFiles("=== *['\\\"](r2|b2)['\\\"] *\\? *['\\\"](r2|b2)['\\\"]").trim(); // provider === 'r2' ? 'r2' : 'b2'
++ // The patterns must be able to fail: they match a sample of what they forbid.
++ const sample = (re) => execSync(`printf '%s\\n' "x = ['r2', 'b2']" "y = p === 'r2' ? 'r2' : 'b2'" | grep -cE "${re}"`).toString().trim();
++ assert.strictEqual(sample("\\[ *['\\\"](r2|b2)['\\\"] *, *['\\\"](b2|r2)['\\\"] *\\]"), '1', 'the list pattern matches a quoted provider list');
++ assert.strictEqual(sample("=== *['\\\"](r2|b2)['\\\"] *\\? *['\\\"](r2|b2)['\\\"]"), '1', 'the ternary pattern matches a provider ternary');
++ assert.strictEqual(lists, '', `no provider-ordered array literal in server/: ${lists}`);
++ assert.strictEqual(orders, '', `no provider-ordered ternary in server/: ${orders}`);
++
++ router.route = realRoute;
++ console.log('placement-read-paths: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-router.test.js b/test/placement-router.test.js
nnew file mode 100644
iindex 0000000..e0cba01
---- /dev/null
++++ b/test/placement-router.test.js
@@@ -0,0 +1,158 @@
++'use strict';
++// Placement router (F1, part 3): picks the fastest healthy copy, sticky sessions, breaker skip,
++// non-`present` locations are NEVER routed to, capability probe removes a class.
++const assert = require('assert');
++
++(async () => {
++ const signals = require('../server/placement/signals');
++ const providers = require('../server/placement/providers');
++ const router = require('../server/placement/router');
++
++ signals.reset();
++ router.reset();
++ providers.reset();
++
++ // The test mocks providers.isConfigured to treat b2 and r2 as "configured" (no env setup here).
++ const realIsConfigured = providers.isConfigured;
++ providers.isConfigured = (name) => name === 'local' || name === 'b2' || name === 'r2';
++
++ // Inject passed capability snapshots for both remote providers.
++ providers._setCapabilities('b2', { provider: 'b2', passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true, cache_range: true, token_auth: true, purge: true }, lastChecked: Date.now(), lastError: null });
++ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true, cache_range: true, token_auth: true, purge: true }, lastChecked: Date.now(), lastError: null });
++
++ // Test 1: local is preferred when the local location's key exists and cloud latency is worse.
++ let decision = await router.route({ locations: [{ provider: 'local', state: 'present', key: '/tmp/x' }, { provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'local', `local wins: ${JSON.stringify(decision)}`);
++
++ // Test 2: r2 wins over b2 when both are present and both have similar EWMA.
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'r2', `r2 wins over b2 for playback: ${JSON.stringify(decision)}`);
++
++ // Test 3: locations NOT in state 'present' are NEVER chosen, even if their provider is healthy.
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'corrupt', key: 'k' }, { provider: 'b2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'b2', `corrupt ignored, b2 picked: ${JSON.stringify(decision)}`);
++
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'pending', key: 'k' }] });
++ assert.strictEqual(decision.provider, null, `pending copy alone never routed to: ${JSON.stringify(decision)}`);
++
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'missing', key: 'k' }] });
++ assert.strictEqual(decision.provider, null, `missing copy alone never routed to: ${JSON.stringify(decision)}`);
++
++ // Test 4: open breaker excludes it; the next copy is served.
++ signals.reset();
++ // Force b2 to look healthy initially
++ signals.recordSuccess('b2', 'presign', 50);
++ signals.recordSuccess('b2', 'presign', 50);
++ // Trip r2's breaker
++ for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
++ assert.strictEqual(signals.breakerOpen('r2'), true);
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }, { provider: 'b2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'b2', `r2 breaker open → b2 served: ${JSON.stringify(decision)}`);
++ assert.ok(decision.fallbacks.includes('r2') || decision.fallbacks.length === 0, 'open breaker not in fallbacks');
++ signals.reset();
++
++ // Test 5: sticky per session — same session returns the same provider for STICKY_TTL_MS.
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }], session: 'player-1' });
++ const stickyProvider = decision.provider;
++ // Subsequent calls with same session: same provider (r2 wins), even if signal changed.
++ for (let i = 0; i < 5; i++) {
++ const d = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }], session: 'player-1' });
++ assert.strictEqual(d.provider, stickyProvider, `sticky across calls: ${i} got ${d.provider}`);
++ }
++
++ // Different session: independent.
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }], session: 'player-2' });
++ assert.ok(['b2', 'r2'].includes(decision.provider));
++
++ // Stickiness drops the moment the sticky provider's breaker opens.
++ for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
++ assert.strictEqual(signals.breakerOpen('r2'), true);
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }], session: 'player-1' });
++ assert.strictEqual(decision.provider, 'b2', `sticky r2 dropped once its breaker opened: ${JSON.stringify(decision)}`);
++ signals.reset();
++ router.reset();
++
++ // Test 6: capability probe failure removes the class; provider is not eligible.
++ providers._setCapabilities('b2', { provider: 'b2', passed: false, capabilities: {}, lastChecked: Date.now(), lastError: 'probe failed' });
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'r2', `failed probe excludes the provider: ${JSON.stringify(decision)}`);
++
++ // Restore b2's passed capabilities for the next test.
++ providers._setCapabilities('b2', { provider: 'b2', passed: true, capabilities: { range: true, range_206: true, etag: true, if_range: true, cache_range: true, token_auth: true, purge: true }, lastChecked: Date.now(), lastError: null });
++
++ // Test 7: provider not configured → not even a candidate.
++ providers.isConfigured = (name) => name === 'local' || name === 'b2'; // r2 disabled
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }, { provider: 'b2', state: 'present', key: 'k' }] });
++ assert.strictEqual(decision.provider, 'b2', `r2 not configured → b2 served`);
++ providers.isConfigured = (name) => name === 'local' || name === 'b2' || name === 'r2';
++
++ // Test 8: no candidates → null provider with a clear reason.
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'pending', key: 'k' }, { provider: 'b2', state: 'corrupt', key: 'k' }] });
++ assert.strictEqual(decision.provider, null, `no healthy copy → null: ${JSON.stringify(decision)}`);
++ assert.strictEqual(decision.reason, 'no-healthy-copy');
++
++ providers.isConfigured = realIsConfigured;
++ router.reset();
++ // Test: a VOD promoted to R2 (storage_provider 'r2') keeps its canonical B2 copy as the fallback. The router
++ // reads the object's locations (every copy a tier move verified), not just the row's storage_provider, so
++ // an open R2 breaker still serves the VOD from B2 instead of answering 404.
++ signals.reset(); router.reset();
++ providers.isConfigured = (name) => name === 'local' || name === 'b2' || name === 'r2';
++ const model = require('../server/objects/model');
++ const realList = model.listLocations;
++ model.listLocations = async (id) => (id === 'med_vod1' ? [
++ { provider: 'b2', state: 'present', key: 'vods/a.mp4' },
++ { provider: 'r2', state: 'present', key: 'vods/a.mp4', verified_at: '2026-09-29T00:00:00Z' },
++ ] : []);
++ const vodRow = { id: 1, object_id: 'med_vod1', storage_provider: 'r2', storage_key: 'vods/a.mp4', file_path: '/nonexistent/vod-a.mp4' };
++ decision = await router.route({ vod: vodRow, purpose: 'playback', presign: false });
++ assert.strictEqual(decision.provider, 'r2', `a verified R2 copy is the hot tier: ${JSON.stringify(decision)}`);
++ assert.deepStrictEqual(decision.fallbacks, ['b2'], `B2 stays the fallback of a promoted VOD: ${JSON.stringify(decision)}`);
++ for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
++ decision = await router.route({ vod: vodRow, purpose: 'playback', presign: false });
++ assert.strictEqual(decision.provider, 'b2', `R2 breaker open: the promoted VOD is served from B2: ${JSON.stringify(decision)}`);
++ signals.reset();
++ // A row whose object is not projected yet still names its own copy; one whose object has only a pending
++ // B2 row does not gain an unverified fallback.
++ decision = await router.route({ vod: { ...vodRow, object_id: null }, purpose: 'playback', presign: false });
++ assert.strictEqual(decision.provider, 'r2');
++ model.listLocations = async () => [{ provider: 'b2', state: 'pending', key: 'vods/a.mp4' }];
++ decision = await router.route({ vod: vodRow, purpose: 'playback', presign: false });
++ assert.deepStrictEqual([decision.provider, decision.fallbacks], ['r2', []], `a pending copy is never a fallback: ${JSON.stringify(decision)}`);
++ // …but a pending projected row for the row's OWN copy does not hide it (every offloaded VOD whose copy no
++ // tier move verified yet): the row names where its bytes are.
++ model.listLocations = async () => [{ provider: 'b2', state: 'pending', key: 'vods/b.mp4' }];
++ decision = await router.route({ vod: { ...vodRow, storage_provider: 'b2', storage_key: 'vods/b.mp4' }, purpose: 'playback', presign: false });
++ assert.strictEqual(decision.provider, 'b2', `the row's own pending copy is routable: ${JSON.stringify(decision)}`);
++ // A copy the object records as missing stays gone.
++ model.listLocations = async () => [{ provider: 'b2', state: 'missing', key: 'vods/b.mp4' }];
++ decision = await router.route({ vod: { ...vodRow, storage_provider: 'b2', storage_key: 'vods/b.mp4' }, purpose: 'playback', presign: false });
++ assert.strictEqual(decision.provider, null, `a missing copy is never routed to: ${JSON.stringify(decision)}`);
++ model.listLocations = realList;
++
++ // Test: a promoted (verified) R2 copy is the hot cache and wins playback over a local copy; an unverified
++ // R2 copy does not.
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k', verified_at: '2026-09-29' }, { provider: 'local', state: 'present', key: '/tmp/x' }], presign: false });
++ assert.deepStrictEqual([decision.provider, decision.reason], ['r2', 'hot-cache'], `verified R2 first: ${JSON.stringify(decision)}`);
++ decision = await router.route({ locations: [{ provider: 'r2', state: 'present', key: 'k' }, { provider: 'local', state: 'present', key: '/tmp/x' }], presign: false });
++ assert.strictEqual(decision.provider, 'local', `an unverified R2 copy does not displace local: ${JSON.stringify(decision)}`);
++
++ // Test: stickiness never holds a session on a worse tier once a better one is back (an R2 re-warm after an
++ // eviction): sticky to B2 while R2 was missing, R2 again as soon as it is present.
++ router.reset(); signals.reset();
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }], session: 'drill', presign: false });
++ assert.strictEqual(decision.provider, 'b2');
++ decision = await router.route({ locations: [{ provider: 'b2', state: 'present', key: 'k' }, { provider: 'r2', state: 'present', key: 'k', verified_at: 'x' }], session: 'drill', presign: false });
++ assert.strictEqual(decision.provider, 'r2', `the hot copy wins over a sticky canonical fallback: ${JSON.stringify(decision)}`);
++
++ // Test: a viewer's session key is stable per viewer and media, and differs across viewers and media.
++ const reqA = { ip: '198.51.100.7', headers: { 'user-agent': 'A' } }, reqB = { ip: '198.51.100.7', headers: { 'user-agent': 'B' } };
++ assert.strictEqual(router.sessionFor(reqA, 'vod:1'), router.sessionFor(reqA, 'vod:1'));
++ assert.notStrictEqual(router.sessionFor(reqA, 'vod:1'), router.sessionFor(reqB, 'vod:1'));
++ 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');
++
++ providers.isConfigured = realIsConfigured;
++ console.log('placement-router: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file
ddiff --git a/test/placement-signals.test.js b/test/placement-signals.test.js
nnew file mode 100644
iindex 0000000..adff4fb
---- /dev/null
++++ b/test/placement-signals.test.js
@@@ -0,0 +1,73 @@
++'use strict';
++// Signals + breaker (placement F1, part 2): EWMA, error rate, p95, breaker open/half-open/close.
++const assert = require('assert');
++
++(async () => {
++ const signals = require('../server/placement/signals');
++
++ // Reset to a clean slate.
++ signals.reset();
++
++ // A clean slate means a closed breaker.
++ assert.strictEqual(signals.breakerOpen('b2'), false);
++ assert.strictEqual(signals.breakerOpen('r2'), false);
++
++ // A single success updates the EWMA; latency reported.
++ signals.recordSuccess('b2', 'presign', 120);
++ const ewma = signals.latencyEWMA('b2', 'presign');
++ assert.ok(ewma >= 100 && ewma <= 140, `ewma within band, got ${ewma}`);
++
++ // Error rate over 1 min with one failure, one success: 50% but only 2 samples — under the
++ // minSamples threshold (5), so the breaker stays closed.
++ for (let i = 0; i < 4; i++) signals.recordSuccess('b2', 'presign', 100);
++ signals.recordFailure('b2', 'presign', 50);
++ const stats = { total: 0, errors: 0, errorRate: signals.errorRate('b2', '1m') };
++ assert.strictEqual(signals.breakerOpen('b2'), false, `breaker still closed (only ${stats.errorRate.toFixed(2)} error rate over a few samples)`);
++
++ // Push enough errors to cross BOTH 5-sample minimum AND 50% rate.
++ for (let i = 0; i < 10; i++) signals.recordFailure('b2', 'presign', 50);
++ assert.strictEqual(signals.breakerOpen('b2'), true, 'breaker opens after sustained 1m error rate >= 50%');
++
++ // 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
++ 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
++ // re-opens it. The clock is virtual (`_advance`) so no real sleep is needed.
++ signals.reset();
++ for (let i = 0; i < 10; i++) signals.recordFailure('b2', 'presign', 50);
++ assert.strictEqual(signals.breakerState('b2'), 'open');
++ signals._advance(61_000);
++ 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);
++ assert.strictEqual(signals.breakerState('b2'), 'closed', 'a successful trial closes the breaker');
++
++ for (let i = 0; i < 10; i++) signals.recordFailure('r2', 'presign', 50);
++ signals._advance(61_000);
++ assert.strictEqual(signals.breakerState('r2'), 'half-open');
++ signals.halfOpenClaim('r2');
++ signals.recordFailure('r2', 'presign', 50);
++ assert.strictEqual(signals.breakerState('r2'), 'open', 'a failed trial re-opens the breaker');
++
++ // 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);
++ const snap = signals.snapshot();
++ assert.ok(snap.providers.b2 && snap.providers.r2);
++ assert.ok(snap.breakers.b2 && snap.breakers.r2);
++ for (const win of signals.WINDOWS) {
++ assert.ok(snap.providers.r2.windows[win], `window ${win} present`);
++ assert.strictEqual(typeof snap.providers.r2.windows[win].p50Ms, 'number', `p50 in ${win}`);
++ assert.strictEqual(typeof snap.providers.r2.windows[win].p95Ms, 'number', `p95 in ${win}`);
++ }
++ assert.strictEqual(snap.providers.r2.windows['5m'].total, 2, 'both r2 samples in the 5-minute window');
++
++ signals.reset();
++ console.log('placement-signals: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });
\\ No newline at end of file