openvibers/openvibe.media Commits

Record per-object demand in Valkey and stage hourly media.object.hot

OP OpenVibers committed 3 Oct 2026 · 7970a3a6a0 · 9 files, +424 −2

Asked: T4 F2.4 — Valkey demand rollups (object × region × 5-minute bucket) in OpenVibe.Media. ONE pull request from current origin/main (git fetch first; F2.1, F2.2 and F2.3 'Record placement decisions for replicas and provider transitions' are merged, so line numbers in the dossier drift). The full spec …
Harness job media-t4-f2-4-valkey-demand-164039 (plan T4).

Changes

ddiff --git a/.env.example b/.env.example
iindex 1bdb573..84e6dd9 100644
---- a/.env.example
++++ b/.env.example
@@@ -95,6 +95,9 @@ MEDIA_R2_SECRET_ACCESS_KEY=
# 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
++# Placement demand (server/placement/demand.js): the region tag on the Valkey demand counters and the hot sorted
++# set (object × region × 5-minute bucket). One Media host is one region; default local.
++#MEDIA_DEMAND_REGION=local
# Media shield (server/placement/shield.js): hand eligible public reads to the nginx edge cache
# (X-Accel-Redirect) instead of streaming them from this process, for the comma-separated providers
# listed. It answers only on MEDIA_SHIELD_HOST (a DNS-only edge name, never through Cloudflare's proxy);
ddiff --git a/STATUS.json b/STATUS.json
iindex 0583c5e..d1e8278 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -33,7 +33,8 @@
"the object explorer (/me) and operator views",
"media.* events through the SDK outbox, webhooks during the transition",
"restore-drill mode and N-1 compatibility tests",
-- "per-actor limits (openvibe-sdk/limits)"
++ "per-actor limits (openvibe-sdk/limits)",
++ "placement demand rollups on Valkey (F2.4): reads per object × region × 5-minute bucket, hotness over the hour, media.object.hot once per object per hour"
],
-- "updated": "2026-09-28"
++ "updated": "2026-10-03"
}
ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex ac22502..2db5572 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -126,6 +126,14 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
at bytes that do not exist.
- **Dry-run and simulator:** every policy can run in dry-run mode (logs what it would do and what it would cost), and a
simulator replays recorded telemetry against a candidate policy before it is enabled.
++- **Demand rollups (F2.4, shipped):** every served viewer read (`router.route()` for playback or download, not a derive
++ job) adds one best-effort hit in Valkey (`server/placement/demand.js`): a counter `demand:<region>:<bucket>:<object_id>`
++ and the region's hot sorted set `hot:<region>:<bucket>` (object id → reads), both under `VALKEY_PREFIX` with a 2 h TTL;
++ `<bucket>` is the 5-minute bucket number (`floor(epoch_ms / 300000)`). The region is one per deployment
++ (`MEDIA_DEMAND_REGION`, default `local`); per-viewer regions wait for an edge-provided header. `hotness()` sums the last
++ 12 buckets (one hour) per object. The writes are fire-and-forget: a slow, failing or absent Valkey never delays or fails
++ a read (without Valkey the counts stay in-process). Object × region only for now; the segment bucket comes with F3.
++ The sweep's eligibility still reads the daily PostgreSQL view counts; moving it to `hotness()` is F2.5.
## 7. Delivery: sticky, measured, canaried
@@@ -160,6 +168,11 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines
- **Loops cooperate by time constant and precedence:** player ABR (seconds) ⊂ delivery routing (30 s–5 min) ⊂ placement
(minutes–hours) ⊂ cost (hours–days). Each outer loop only constrains the inner (budgets bound placement; placement
bounds routes); every loop uses hysteresis and rate limits.
++- **Shipped (F2.4):** the Valkey counters and per-region hot sorted sets of §6. Each tiering sweep calls
++ `demand.rollup()`, which stages `media.object.hot` (`{ object_id, app_id, region, reads, window_s, threshold, bucket,
++ since }`) through the placement outbox for objects with at least 100 reads in the last hour, at most once per object per
++ hour (`hot-announced:<region>:<object_id>`, `SET NX EX 3600`). Not yet: nginx logs, player beacons, EWMA/percentiles,
++ HyperLogLog, the PostgreSQL 5-minute rollup and `media.region.hot`.
- **Events emitted:** `media.object.hot`, `media.replica.requested|ready|draining|evicted`, `media.variant.requested|ready`,
`media.region.hot`, `media.delivery.surge|degraded`, `media.provider.health_degraded`, `media.provider.capacity_warning`,
`media.provider.cost_threshold`.
ddiff --git a/server/config.js b/server/config.js
iindex c34a69b..7805e53 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -30,6 +30,8 @@ const config = {
dataDir: process.env.MEDIA_DATA_DIR || './data',
// Valkey (ADR-035): per-actor limit counters shared across processes; without it they count in this process.
valkey: { url: process.env.VALKEY_URL || '', prefix: process.env.VALKEY_PREFIX || 'ov:media:' },
++ // Placement demand (F2.4): the region tag on hotness counters. One Media host is one region today.
++ demand: { region: process.env.MEDIA_DEMAND_REGION || 'local' },
vod: {
path: process.env.VOD_PATH || './data/vods',
ddiff --git a/server/index.js b/server/index.js
iindex fa3a266..d3e0ec6 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -48,6 +48,8 @@ const ready = (async () => {
// Valkey (ADR-035): per-actor limit counters shared across processes (not in a drill: it connects nowhere).
const valkey = !drill.enabled && config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix }) : null;
require('./actor-limits').useValkey(valkey);
++ // Demand rollups (F2.4, placement/demand.js): reads per object × region × 5-minute bucket on the same handle.
++ require('./placement/demand').useValkey(valkey);
if (!drill.enabled) await require('./revocations').load();
await require('./vod/tier-config').init(require('./vod/vod-storage').DEFAULTS);
await require('./objects/tier-policy').init();
ddiff --git a/server/objects/tiering.js b/server/objects/tiering.js
iindex f9acf41..43d8b34 100644
---- a/server/objects/tiering.js
++++ b/server/objects/tiering.js
@@@ -511,6 +511,10 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.promote[cls] = (spent.promote[cls] || 0) + 1;
}
++ // Demand (F2.4): stage media.object.hot for objects over the hourly read threshold. Eligibility above
++ // still reads the daily PostgreSQL counts; moving it to hotness() is F2.5.
++ out.demand = await require('../placement/demand').rollup({ now });
++
if (out.promoted || out.demoted || out.failed) {
console.log(`[ObjectTiers] Sweep: ${out.promoted} promoted to R2, ${out.demoted} demoted, ${out.failed} failed, ${out.refused} refused`);
}
ddiff --git a/server/placement/demand.js b/server/placement/demand.js
nnew file mode 100644
iindex 0000000..b77f15b
---- /dev/null
++++ b/server/placement/demand.js
@@@ -0,0 +1,204 @@
++'use strict';
++/**
++ * Demand rollups (F2.4, docs/media-fabric.md §6, §8): reads per object × region × 5-minute bucket, folded into
++ * Valkey counters plus one hot sorted set per region and bucket, with an in-process fallback when Valkey is off.
++ * Valkey is never authoritative (ADR-007): a lost counter only delays a promotion, never loses bytes, and a slow
++ * or absent Valkey never delays or fails a read (record() is fire-and-forget).
++ *
++ * Keys (inside the service prefix, VALKEY_PREFIX=ov:media:):
++ * demand:<region>:<bucket>:<object_id> counter, one per served read TTL KEY_TTL_S
++ * hot:<region>:<bucket> sorted set, object_id -> reads in that bucket TTL KEY_TTL_S
++ * hot-announced:<region>:<object_id> media.object.hot staged within the hour (NX) TTL ANNOUNCE_S
++ * <bucket> = floor(epoch_ms / BUCKET_MS). The region is one per deployment (MEDIA_DEMAND_REGION, default 'local');
++ * a per-viewer region needs an edge-provided header that does not exist yet.
++ *
++ * Not yet read by the sweep's eligibility (that stays on the daily PostgreSQL counts, objects/popularity.js, until
++ * F2.5): the sweep only calls rollup(), which stages media.object.hot for objects over HOT_READS in the window.
++ */
++const config = require('../config');
++
++const BUCKET_MS = 5 * 60 * 1000; // the design's video bucket
++const WINDOW_BUCKETS = 12; // one hour of buckets hotness() and rollup() read
++const KEY_TTL_S = 2 * 3600; // longer than the window, so a slow sweep still reads a full hour
++const TOP_N = 500; // in-process sorted set cap per bucket; most objects one rollup announces
++const HOT_READS = 100; // reads in the window that make an object hot
++const ANNOUNCE_S = 3600; // media.object.hot at most once per object per hour
++const READ_TIMEOUT_MS = 2000; // a rollup read never waits on a stuck Valkey longer than this
++const FALLBACK_MAX = 50000; // in-process counters before expired ones are pruned on write
++
++let vk = null; // openvibe-sdk/valkey handle, or null: count in this process
++const counters = new Map(); // fallback: counter key -> { n, expiresAt }
++const sets = new Map(); // fallback: hot key -> { members: Map(object_id -> n), expiresAt }
++const announced = new Map(); // fallback: announce key -> expiresAt
++
++function useValkey(handle) { vk = handle && handle.client ? handle : null; }
++function region() { return String((config.demand && config.demand.region) || 'local').replace(/[^A-Za-z0-9_.-]/g, '_') || 'local'; }
++function bucketOf(now = Date.now()) { return Math.floor(now / BUCKET_MS); }
++const name = (...parts) => (vk ? vk.key(...parts) : parts.join(':'));
++function counterKey(r, b, id) { return name('demand', r, String(b), String(id)); }
++function hotKey(r, b) { return name('hot', r, String(b)); }
++function announceKey(r, id) { return name('hot-announced', r, String(id)); }
++const windowOf = (now, buckets) => Array.from({ length: buckets }, (_, i) => bucketOf(now) - i);
++const quiet = (p) => { if (p && typeof p.catch === 'function') p.catch(() => {}); return p; };
++const timed = (p) => Promise.race([p, new Promise((_, rej) => setTimeout(() => rej(new Error('valkey read timeout')), READ_TIMEOUT_MS).unref())]);
++
++function pruneFallback(now = Date.now()) {
++ for (const [k, c] of counters) if (c.expiresAt <= now) counters.delete(k);
++ for (const [k, s] of sets) if (s.expiresAt <= now) sets.delete(k);
++ for (const [k, at] of announced) if (at <= now) announced.delete(k);
++}
++
++/** One served read of an object. Best-effort: never throws, never awaits Valkey. */
++function record({ objectId, region: r = region(), now = Date.now() } = {}) {
++ if (objectId == null || objectId === '') return;
++ const id = String(objectId), b = bucketOf(now);
++ try {
++ if (vk) {
++ const ck = counterKey(r, b, id), hk = hotKey(r, b);
++ quiet(vk.client.incr(ck).then((n) => (n === 1 ? vk.client.expire(ck, KEY_TTL_S) : null)));
++ quiet(vk.client.zincrby(hk, 1, id).then((n) => (Number(n) === 1 ? vk.client.expire(hk, KEY_TTL_S) : null)));
++ return;
++ }
++ const expiresAt = now + KEY_TTL_S * 1000;
++ if (counters.size >= FALLBACK_MAX) pruneFallback(now);
++ const ck = counterKey(r, b, id);
++ const c = counters.get(ck) || { n: 0, expiresAt };
++ c.n++;
++ counters.set(ck, c);
++ const hk = hotKey(r, b);
++ const s = sets.get(hk) || { members: new Map(), expiresAt };
++ s.members.set(id, (s.members.get(id) || 0) + 1);
++ if (s.members.size > TOP_N) {
++ let lo = null;
++ for (const e of s.members) if (e[0] !== id && (!lo || e[1] < lo[1])) lo = e;
++ if (lo) s.members.delete(lo[0]);
++ }
++ sets.set(hk, s);
++ } catch { /* demand is a hint */ }
++}
++
++/**
++ * Reads per object over the last `buckets` buckets (summed), newest bucket included. With `objectIds`, the counters
++ * of those objects (all of them, zero when unread); without, every object in the region's hot sets.
++ * A Valkey error answers an empty map: demand is a hint. → Map(object_id -> reads)
++ */
++async function hotness({ region: r = region(), now = Date.now(), buckets = WINDOW_BUCKETS, objectIds = null } = {}) {
++ const out = new Map();
++ const window = windowOf(now, buckets);
++ const ids = objectIds ? [...new Set(objectIds.filter((x) => x != null).map(String))] : null;
++ if (ids) for (const id of ids) out.set(id, 0);
++ try {
++ if (vk) {
++ if (ids) {
++ if (!ids.length) return out;
++ const keys = ids.flatMap((id) => window.map((b) => counterKey(r, b, id)));
++ const vals = await timed(vk.client.mget(...keys));
++ ids.forEach((id, i) => {
++ let n = 0;
++ for (let j = 0; j < window.length; j++) n += Number(vals[i * window.length + j]) || 0;
++ out.set(id, n);
++ });
++ return out;
++ }
++ const all = await timed(Promise.all(window.map((b) => vk.client.zrange(hotKey(r, b), 0, -1, 'WITHSCORES'))));
++ for (const flat of all) {
++ for (let i = 0; i + 1 < flat.length; i += 2) out.set(flat[i], (out.get(flat[i]) || 0) + Number(flat[i + 1]));
++ }
++ return out;
++ }
++ pruneFallback(now);
++ if (ids) {
++ for (const id of ids) {
++ let n = 0;
++ for (const b of window) { const c = counters.get(counterKey(r, b, id)); if (c) n += c.n; }
++ out.set(id, n);
++ }
++ return out;
++ }
++ for (const b of window) {
++ const s = sets.get(hotKey(r, b));
++ if (s) for (const [id, n] of s.members) out.set(id, (out.get(id) || 0) + n);
++ }
++ return out;
++ } catch (err) {
++ console.warn('[Demand] hotness not read:', err.message);
++ return ids ? new Map(ids.map((id) => [id, 0])) : new Map();
++ }
++}
++
++/** The hottest objects of a region over the last `buckets` buckets, hottest first. → [{ object_id, reads }] */
++async function top({ region: r = region(), now = Date.now(), buckets = 1, limit = 100 } = {}) {
++ return [...(await hotness({ region: r, now, buckets }))]
++ .sort((a, b) => b[1] - a[1] || (a[0] < b[0] ? -1 : 1))
++ .slice(0, limit)
++ .map(([object_id, reads]) => ({ object_id, reads }));
++}
++
++/** Claim this hour's media.object.hot for an object: true once per ANNOUNCE_S (across processes on Valkey). */
++async function claimAnnounce(r, id, now) {
++ const k = announceKey(r, id);
++ if (vk) return (await timed(vk.client.set(k, String(now), 'EX', ANNOUNCE_S, 'NX'))) === 'OK';
++ const at = announced.get(k);
++ if (at && at > now) return false;
++ announced.set(k, now + ANNOUNCE_S * 1000);
++ return true;
++}
++
++async function releaseAnnounce(r, id) {
++ const k = announceKey(r, id);
++ try { if (vk) await timed(vk.client.del(k)); else announced.delete(k); } catch { /* the claim lapses with its TTL */ }
++}
++
++/**
++ * The sweep's demand pass: stage media.object.hot (through the placement outbox, one transaction per object) for
++ * every object with at least `threshold` reads in the window, at most once per object per hour. Objects no
++ * longer in media_objects are skipped; a sandbox tenant's event is dropped by the outbox. Never throws.
++ * → { region, window_s, threshold, objects, hot, announced }
++ */
++async function rollup({ region: r = region(), now = Date.now(), threshold = HOT_READS, buckets = WINDOW_BUCKETS } = {}) {
++ const out = { region: r, window_s: (buckets * BUCKET_MS) / 1000, threshold, objects: 0, hot: 0, announced: 0 };
++ try {
++ const heat = await hotness({ region: r, now, buckets });
++ out.objects = heat.size;
++ const over = [...heat].filter(([, n]) => n >= threshold).sort((a, b) => b[1] - a[1]).slice(0, TOP_N);
++ out.hot = over.length;
++ if (!over.length) return out;
++ const db = require('../db/database');
++ const events = require('../events');
++ const rows = await db.all(`SELECT id, app_id FROM media_objects WHERE id IN (${over.map(() => '?').join(', ')})`, over.map(([id]) => id));
++ const appOf = new Map(rows.map((row) => [String(row.id), row.app_id]));
++ for (const [id, reads] of over) {
++ if (!appOf.has(id)) continue;
++ try {
++ if (!(await claimAnnounce(r, id, now))) continue;
++ } catch (err) { console.warn('[Demand] hot claim failed:', err.message); continue; }
++ const appId = appOf.get(id) || null;
++ try {
++ let env = null;
++ await db.getDb().tx(async () => {
++ env = await events.recordPlacement('media.object.hot', appId, { type: 'object', id }, {
++ object_id: id, app_id: appId, region: r, reads, window_s: out.window_s, threshold,
++ bucket: bucketOf(now), since: new Date(now - out.window_s * 1000).toISOString(),
++ });
++ });
++ if (env) out.announced++;
++ } catch (err) {
++ console.warn(`[Demand] media.object.hot for ${id} not staged: ${err.message}`);
++ await releaseAnnounce(r, id);
++ }
++ }
++ if (out.announced) events.kick();
++ } catch (err) {
++ console.warn('[Demand] rollup failed:', err.message);
++ out.error = err.message;
++ }
++ return out;
++}
++
++/** Forget everything in this process (tests). */
++function _reset() { counters.clear(); sets.clear(); announced.clear(); }
++
++module.exports = {
++ BUCKET_MS, WINDOW_BUCKETS, KEY_TTL_S, TOP_N, HOT_READS, ANNOUNCE_S,
++ useValkey, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,
++};
ddiff --git a/server/placement/router.js b/server/placement/router.js
iindex 943eee7..04c3c59 100644
---- a/server/placement/router.js
++++ b/server/placement/router.js
@@@ -188,6 +188,9 @@ const REASONS = {
3: 'canonical-fallback',
};
++// Reads that count as demand: a viewer playing or downloading, not a job deriving from the copy.
++const DEMAND_PURPOSES = new Set(['playback', 'download']);
++
/**
* The only function every read path uses.
* opts: { object | vod | clip | locations, purpose, session?, range?, presign?, expiresIn?,
@@@ -242,6 +245,14 @@ async function route(opts = {}) {
if (primary) stickyPut(session, primary.provider);
incMetric('media_router_decisions_total', { provider: primary ? primary.provider : 'none', purpose, reason });
++ // Demand (F2.4): one hit per served viewer read (not a derive job), object × region × 5-minute bucket.
++ // Best-effort and fire-and-forget: a Valkey error or absence never delays or fails the read.
++ if (primary && DEMAND_PURPOSES.has(purpose)) {
++ try {
++ const id = opts.object ? opts.object.id : (opts.vod || opts.clip || {}).object_id;
++ if (id) require('./demand').record({ objectId: id });
++ } catch { /* demand is a hint */ }
++ }
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 }
ddiff --git a/test/placement-demand.test.js b/test/placement-demand.test.js
nnew file mode 100644
iindex 0000000..57e1336
---- /dev/null
++++ b/test/placement-demand.test.js
@@@ -0,0 +1,182 @@
++'use strict';
++// Demand rollups (F2.4): reads per object × region × 5-minute bucket, on Valkey or in this process; hotness over
++// the window; media.object.hot staged once per object per hour; a Valkey failure never touches the read path.
++const assert = require('assert');
++const crypto = require('crypto');
++const fs = require('fs');
++const http = require('http');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-placement-demand-'));
++ const dir = (name) => { const p = path.join(tmp, name); fs.mkdirSync(p, { recursive: true }); return p; };
++ Object.assign(process.env, {
++ VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
++ });
++ const demand = require('../server/placement/demand');
++ const router = require('../server/placement/router');
++ const B = demand.BUCKET_MS;
++ const T0 = 1_800_000_000_000; // a fixed instant, the start of a bucket
++ assert.strictEqual(T0 % B, 0);
++
++ // 1. Bucket math and key layout (in-process: no Valkey handle).
++ assert.strictEqual(B, 300000);
++ assert.strictEqual(demand.bucketOf(T0), T0 / B);
++ assert.strictEqual(demand.bucketOf(T0 + B - 1), T0 / B);
++ assert.strictEqual(demand.bucketOf(T0 + B), T0 / B + 1);
++ assert.strictEqual(demand.region(), 'local');
++ assert.strictEqual(demand.counterKey('local', 7, 'obj-a'), 'demand:local:7:obj-a');
++ assert.strictEqual(demand.hotKey('local', 7), 'hot:local:7');
++ assert.strictEqual(demand.KEY_TTL_S, 7200);
++ assert.ok(demand.KEY_TTL_S * 1000 > demand.WINDOW_BUCKETS * B, 'a key outlives the window it is read in');
++ console.log('✅ bucket math and key layout');
++
++ // 2. hotness() sums across the window's buckets; a bucket older than the window is out; top() orders.
++ demand._reset();
++ for (let i = 0; i < 3; i++) demand.record({ objectId: 'obj-a', now: T0 });
++ demand.record({ objectId: 'obj-a', now: T0 - 4 * B }); // an earlier bucket, inside the hour
++ demand.record({ objectId: 'obj-b', now: T0 + 1000 });
++ demand.record({ objectId: 'obj-c', now: T0 - demand.WINDOW_BUCKETS * B }); // 13th bucket back: outside
++ demand.record({ objectId: 'obj-a', region: 'eu', now: T0 }); // another region
++ let heat = await demand.hotness({ now: T0 + 2000 });
++ assert.deepStrictEqual([...heat].sort(), [['obj-a', 4], ['obj-b', 1]]);
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-a', 'obj-c', 'obj-z'] })), { 'obj-a': 4, 'obj-c': 0, 'obj-z': 0 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, buckets: 1 })), { 'obj-a': 3, 'obj-b': 1 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ region: 'eu', now: T0 })), { 'obj-a': 1 });
++ assert.deepStrictEqual(await demand.top({ now: T0 }), [{ object_id: 'obj-a', reads: 3 }, { object_id: 'obj-b', reads: 1 }]);
++ // Past the TTL, the in-process counters are gone as Valkey's would be.
++ assert.strictEqual((await demand.hotness({ now: T0 + demand.KEY_TTL_S * 1000 + B, buckets: 100 })).size, 0);
++ console.log('✅ hotness() sums the window per object and region; top() orders the bucket; TTL expires counts');
++
++ // 3. On Valkey: the exact keys inside the service prefix, and a TTL set once per key.
++ const calls = [];
++ const store = new Map(), zsets = new Map(), ttls = new Map();
++ const fake = {
++ prefix: 'ov:media:',
++ key: (...parts) => 'ov:media:' + parts.join(':'),
++ client: {
++ async incr(k) { calls.push(['incr', k]); store.set(k, (Number(store.get(k)) || 0) + 1); return store.get(k); },
++ async expire(k, s) { calls.push(['expire', k, s]); ttls.set(k, s); return 1; },
++ async zincrby(k, by, m) { calls.push(['zincrby', k, by, m]); const z = zsets.get(k) || new Map(); z.set(m, (z.get(m) || 0) + by); zsets.set(k, z); return String(z.get(m)); },
++ async zrange(k, start, stop, ws) { assert.deepStrictEqual([start, stop, ws], [0, -1, 'WITHSCORES']); return [...(zsets.get(k) || new Map())].flatMap(([m, n]) => [m, String(n)]); },
++ async mget(...keys) { return keys.map((k) => (store.has(k) ? String(store.get(k)) : null)); },
++ async set(k, v, ex, s, nx) { assert.deepStrictEqual([ex, s, nx], ['EX', demand.ANNOUNCE_S, 'NX']); if (store.has(k)) return null; store.set(k, v); ttls.set(k, s); return 'OK'; },
++ async del(k) { store.delete(k); return 1; },
++ },
++ };
++ demand.useValkey(fake);
++ const b = demand.bucketOf(T0);
++ demand.record({ objectId: 'obj-a', now: T0 });
++ demand.record({ objectId: 'obj-a', now: T0 });
++ demand.record({ objectId: 'obj-b', now: T0 - B });
++ await new Promise((resolve) => setImmediate(resolve));
++ const ck = `ov:media:demand:local:${b}:obj-a`, hk = `ov:media:hot:local:${b}`;
++ assert.strictEqual(store.get(ck), 2);
++ assert.deepStrictEqual([...zsets.get(hk)], [['obj-a', 2]]);
++ assert.strictEqual(ttls.get(ck), 7200);
++ assert.strictEqual(ttls.get(hk), 7200);
++ assert.strictEqual(ttls.get(`ov:media:hot:local:${b - 1}`), 7200);
++ assert.strictEqual(calls.filter((c) => c[0] === 'expire' && c[1] === ck).length, 1, 'TTL set on the first hit only');
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0 })), { 'obj-a': 2, 'obj-b': 1 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-a', 'obj-b'] })), { 'obj-a': 2, 'obj-b': 1 });
++ console.log('✅ Valkey keys ov:media:demand:<region>:<bucket>:<id> and ov:media:hot:<region>:<bucket>, TTL 7200 s');
++
++ // 4. The router records one hit per served viewer read; none for a derive job or when nothing is served.
++ const file = path.join(tmp, 'objects', 'obj-r');
++ fs.writeFileSync(file, 'bytes');
++ const served = { id: 'obj-r', locations: [{ provider: 'local', state: 'present', key: file }] };
++ store.clear(); zsets.clear(); calls.length = 0;
++ let decision = await router.route({ object: served, purpose: 'playback' });
++ assert.strictEqual(decision.provider, 'local');
++ await router.route({ vod: { id: 9, object_id: 'obj-v' }, locations: served.locations, purpose: 'download' });
++ await router.route({ object: served, purpose: 'derive' });
++ await router.route({ object: { id: 'obj-none', locations: [] }, purpose: 'playback' });
++ await new Promise((resolve) => setImmediate(resolve));
++ assert.deepStrictEqual(calls.filter((c) => c[0] === 'incr').map((c) => c[1]).map((k) => k.split(':').pop()), ['obj-r', 'obj-v']);
++ console.log('✅ router: one demand hit per served playback/download, none for derive or no copy');
++
++ // 5. A failing Valkey (rejecting, throwing, hanging) leaves the read path untouched.
++ const unhandled = [];
++ process.on('unhandledRejection', (err) => unhandled.push(err));
++ const broken = {
++ key: fake.key,
++ client: {
++ incr: () => Promise.reject(new Error('ECONNREFUSED')),
++ zincrby: () => { throw new Error('socket closed'); },
++ mget: () => Promise.reject(new Error('ECONNREFUSED')),
++ zrange: () => new Promise(() => {}), // never answers
++ },
++ };
++ demand.useValkey(broken);
++ const before = Date.now();
++ decision = await router.route({ object: served, purpose: 'playback' });
++ assert.strictEqual(decision.provider, 'local');
++ assert.strictEqual(decision.key, file);
++ assert.doesNotThrow(() => demand.record({ objectId: 'obj-a' }));
++ assert.doesNotThrow(() => demand.record({ objectId: null }));
++ assert.doesNotThrow(() => demand.record());
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ objectIds: ['obj-a'] })), { 'obj-a': 0 });
++ assert.strictEqual((await demand.hotness()).size, 0); // the hung read times out
++ assert.ok(Date.now() - before < 5000);
++ assert.strictEqual((await demand.rollup()).announced, 0);
++ await new Promise((resolve) => setImmediate(resolve));
++ assert.deepStrictEqual(unhandled, []);
++ console.log('✅ a Valkey failure never throws into a read; reads answer no demand');
++
++ // 6. rollup(): media.object.hot through the placement outbox, at most once per object per hour.
++ demand.useValkey(null);
++ demand._reset();
++ const stub = http.createServer((req, res) => {
++ req.resume();
++ req.on('end', () => {
++ res.setHeader('Content-Type', 'application/json');
++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
++ res.statusCode = 503; res.end('{}'); // keep every envelope in the outbox for inspection
++ });
++ });
++ await new Promise((resolve) => stub.listen(0, '127.0.0.1', resolve));
++ const base = `http://127.0.0.1:${stub.address().port}`;
++ const db = require('../server/db/database');
++ const model = require('../server/objects/model');
++ const events = require('../server/events');
++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-placement-demand' });
++ const addObject = async () => model.createObject({ app_id: 'live', kind: 'file', visibility: 'public', lifecycle_status: 'ready',
++ mime_type: 'video/mp4', size_bytes: 5, content_hash: crypto.randomBytes(32).toString('hex'), canonical_provider: 'local' });
++ const hotId = String(await addObject()), coolId = String(await addObject());
++ assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 3600000 }));
++ const hotRows = async () => (await db.getDb().prepare('SELECT envelope FROM event_outbox ORDER BY id').all())
++ .map((row) => (typeof row.envelope === 'string' ? JSON.parse(row.envelope) : row.envelope))
++ .filter((e) => e.event_type === 'media.object.hot');
++
++ const now = Date.now();
++ for (let i = 0; i < demand.HOT_READS; i++) demand.record({ objectId: hotId, now: now - (i % 6) * B });
++ for (let i = 0; i < 5; i++) demand.record({ objectId: coolId, now });
++ demand.record({ objectId: 'gone-object', now });
++ let r = await demand.rollup({ now, threshold: demand.HOT_READS });
++ assert.deepStrictEqual(r, { region: 'local', window_s: 3600, threshold: 100, objects: 3, hot: 1, announced: 1 });
++ let rows = await hotRows();
++ assert.strictEqual(rows.length, 1);
++ assert.deepStrictEqual(rows[0].subject, { type: 'object', id: hotId });
++ assert.deepStrictEqual(Object.keys(rows[0].payload).sort(), ['app_id', 'bucket', 'object_id', 'reads', 'region', 'since', 'threshold', 'window_s']);
++ assert.deepStrictEqual([rows[0].payload.object_id, rows[0].payload.app_id, rows[0].payload.region, rows[0].payload.reads], [hotId, 'live', 'local', 100]);
++ // Still hot later in the same hour: not announced again.
++ for (let i = 0; i < 50; i++) demand.record({ objectId: hotId, now: now + 20 * 60 * 1000 });
++ r = await demand.rollup({ now: now + 20 * 60 * 1000 });
++ assert.strictEqual(r.announced, 0);
++ assert.strictEqual((await hotRows()).length, 1);
++ // An hour after the first announcement, still hot: announced once more.
++ for (let i = 0; i < demand.HOT_READS; i++) demand.record({ objectId: hotId, now: now + 61 * 60 * 1000 });
++ r = await demand.rollup({ now: now + 61 * 60 * 1000 });
++ assert.strictEqual(r.announced, 1);
++ assert.strictEqual((await hotRows()).length, 2);
++ console.log('✅ rollup(): media.object.hot for objects over the threshold, once per object per hour');
++
++ events._reset();
++ stub.close();
++ await db.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('placement-demand: all six cases passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });