openvibers/openvibe.media Commits

Bucket placement demand per segment, not just per object (#39)

OP OpenVibers committed 5 Oct 2026 · 3e409b9835 · 7 files, +211 −52

* Bucket placement demand per segment, not just per object

Implementation complete. All tests green on this exact code.

## Files changed

- `server/placement/demand.js` — `counterKey(r,b,id,segment)` now appends the segment (`demand:<region>:<bucket>:<object>:<segment>`); added `segKey` (per-object segment hot set `hot:<region>:<bucket>:<object>`). `record({ segment })` bumps the object total/set on every read and, with a segment, that segment's counter + set — so the object stays the sum over its segments. `hotness({ segment })` reads the segment axis (per-object counters via the one-MGET path with `objectIds`, or each object's segment set without); the object-level path is byte-for-byte the old one, so the tiering sweep's eligibility and batching are unchanged. In-process fallback keeps both axes.
- `server/placement/router.js` — `route()` accepts `objectId`/`segment` and passes `segment` into `demand.record()`; docblock updated. Segment reads previously recorded nothing (no object/vod/clip row was passed).
- `server/objects/routes.js` — `GET /o/:id/source/:name` now passes `objectId: obj.id, segment: name` into `router.route(...)`; stale "segment-bucket demand comes later" comment replaced.
- `test/placement-demand.test.js` — extended: key layout, in-process segment read moves only its own bucket / aggregates to the object / segment axis without an object list, Valkey segment keys + TTL, and the router naming the object and segment.
- `test/clip-timeline.test.js` — end-to-end check that signed segment GETs through the real route record per-segment demand (object total 3, each segment 1, an unserved segment 0).
- `docs/media-fabric.md` — §6 hotness bullet and the F2.4 demand-rollups bullet describe both axes; removed "segment-bucket demand" from the two "still open" lists.
- `STATUS.json` — F2.4 feature line now names the per-segment dimension.

## Tests run

- `ov test test/placement-demand.test.js` → 1 passed (15.0 s)
- `ov test test/placement-metrics.test.js test/clip-timeline.test.js test/object-tiering.test.js` → 3 passed (20 s)
- Final re-run of the three directly-reaching files → all green on this exact code.

## Notes

- `test/object-tiering.test.js` passed unchanged, confirming the sweep's object-level eligibility still reads the same keys.
- Could not update a CHANGELOG: this repo has none (only `docs/media-fabric.md` and `docs/object-model.md`).
- Per instructions, nothing was committed, pushed, deployed, or installed.

* Segment demand lives in one capped zset per object; no per-segment keys; bounded names and reads (review)

**demand.js**
- `server/placement/demand.js:48-49` — `counterKey(r,b,id)` lost its segment parameter; the per-segment `demand:<r>:<b>:<id>:<seg>` counter key is gone.
- `server/placement/demand.js:94-100` — `record()`'s Valkey path no longer `INCR`s a segment counter; a segment read only `ZINCRBY`s its member in the per-object set, sets the set's TTL on the first member, and caps it with `ZREMRANGEBYRANK(gk, 0, -(TOP_N+1))` when a new member grows it.
- `server/placement/demand.js:104-108` — in-process fallback drops the segment counter too; the TOP_N-capped segment set is the only segment state, so `pruneFallback` can't evict object counters on segment writes.
- `server/placement/demand.js:130-142` — `hotness({objectIds, segment})` reads segment hotness from each per-object zset via `ZSCORE`, issued together (auto-pipelined); the object axis still uses one `MGET` of counters.
- `server/placement/demand.js:144-157` — segment axis without `objectIds` is bounded to the region's hottest `TOP_N` objects before its per-object reads.
- `server/placement/demand.js:166-186` — fallback segment axis bounded the same way (`slice(0, TOP_N)`).

**routes.js**
- `server/objects/routes.js:784` — `SEGMENT_NAME` capped at `/^(init\.mp4|\d{6,9}\.m4s)$/`.

**docs / tests**
- `docs/media-fabric.md:234-244` — key layout now documents the capped per-object segment set, no per-segment counter.
- `test/placement-demand.test.js:77-79, 111-121, 124-151` — fake Valkey gains `zscore`/`zremrangebyrank`; assertions drop the per-segment counter key; new case 3b: 1000 segment reads leave one object counter (exact total 1000), one segment zset capped at `TOP_N` and no per-segment counter keys, with the same bound checked in the in-process fallback.
- `test/placement-demand.test.js:163-169` — router case asserts the segment read writes only an object counter (no `000002.m4s` counter) and names the segment in its per-object set.

**Test results**
- `ov test test/placement-demand.test.js test/placement-metrics.test.js` → both green (`placement-demand` all seven cases; `placement-metrics` unchanged-since-green).
- Also ran the other directly-reaching files: `test/clip-timeline.test.js` and `test/object-tiering.test.js` → both passed.

Changes

ddiff --git a/STATUS.json b/STATUS.json
iindex e9d76f1..091a0fe 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -34,7 +34,7 @@
"media.* events through the SDK outbox, webhooks during the transition",
"restore-drill mode and N-1 compatibility tests",
"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",
++ "placement demand rollups on Valkey (F2.4): reads per object × region × 5-minute bucket and, for a timeline segment, per segment too (object total is the sum over its segments; hotness over the hour, media.object.hot once per object per hour)",
"tiering on demand (F2.5): the object sweep promotes and demotes on the hour's Valkey reads against per-class hysteresis bands, one MGET per page, falling back to the daily PostgreSQL counts for a whole sweep when Valkey fails",
"placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs",
"every storage class in the native-object sweep: kind and mime type classify video/image/download, an app may declare game-asset/attachment/backup (or any of the six) in the object's metadata (metadata.class, or media_class/placement_class), and each class spends its own per-sweep budget, hysteresis band and R2 storage ceiling",
ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex 7ec215f..46a5e3c 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -177,8 +177,9 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
`MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.
**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for
live segments and its upload-lag metric, a materialized clip that references the source's chunks (reference counts
-- shipped, F3.4; its own persisted rows and boundary-only re-encode are open), captions on the timeline,
-- segment-bucket demand, and Live's player moving to HLS. Renditions are F4.
++ shipped, F3.4; its own persisted rows and boundary-only re-encode are open: docs/materialized-clips.md), captions on
++ the timeline, and Live's player moving to HLS. Segment-bucket demand (F2.4) now records reads per timeline segment;
++ deciding placement per segment stays F5's. Renditions are F4.
## 4. Clips reuse the source
@@@ -237,7 +238,9 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
- **Hotness is per object × region × segment bucket** (5-minute buckets for video: beginnings are hotter than middles).
A two-hour VOD can have its first 10 minutes on R2 + CDN + edge, minutes 10–40 on R2, the rest only on B2, and
-- promote minute 60 when viewers start landing there.
++ promote minute 60 when viewers start landing there. The reads record both axes (F2.4, below): the object total the
++ tiering sweep's eligibility reads, and each timeline segment's own count; the per-segment placement decision
++ itself stays open.
- **Decision rule:** value-per-dollar greedy under per-class budgets (the knapsack is near-optimal when costs are
monotone): expected requests and bytes saved × latency value, against storage + requests + retrieval + minimum
residency, with hysteresis (promote above P, demote below D < P), minimum residency, and rate limits on moves.
@@@ -257,10 +260,17 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
- **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 later in F3 (F3.1 shipped the timeline it will bucket).
++ `<bucket>` is the 5-minute bucket number (`floor(epoch_ms / 300000)`). A read of a timeline segment (`GET
++ /o/:id/source/:name`) carries the segment name, moving that segment's member in the per-object segment set
++ `hot:<region>:<bucket>:<object_id>` (segment → reads; capped at `TOP_N` members, 2 h TTL), so a hot beginning reads
++ apart from a cold middle without a key per segment; the object counter moves on every read either
++ way, so the object is the sum over its segments (plus whole-object reads) and `hotness()` — the sweep's eligibility —
++ is unchanged. `hotness({ segment })` reads one segment (per object, or across the region's hottest `TOP_N` objects).
++ 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). 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, both axes). The per-segment placement decision
++ (which minutes of a VOD sit on R2/edge) stays open; F2 records the demand it will read.
- **The sweep reads demand (F2.5, shipped):** the object tiering sweep's promote/demote eligibility is `hotness()` for the
deployment's region, against each class's hysteresis band in `media.storage_policy` (`promoteReadsPerHour` /
`demoteReadsPerHour`, demote < promote, validated per revision): video 60 / 6, image 300 / 30, download 30 / 3, the
@@@ -377,7 +387,7 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
materialized clips (F3.5, shipped; §4) and sprite sheets cut from the timeline's rows (F3.1, shipped; §3); still
open: live DVR from OpenRe (CMAF/HLS recording with the growing
playlist), materialized clips referencing the source's chunks, captions on the
-- timeline, segment-bucket demand; Live's player moves to HLS.
++ timeline (the demand rollups already record reads per timeline segment, F2.4); Live's player moves to HLS.
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
VODs, compute placement.
- **F5 multi-CDN delivery:** Bunny path (B2 origin), R2 custom-domain path, OpenVibe edge first up to capacity, route
ddiff --git a/server/objects/routes.js b/server/objects/routes.js
iindex 592e6b9..90bff35 100644
---- a/server/objects/routes.js
++++ b/server/objects/routes.js
@@@ -778,7 +778,7 @@ const publicRouter = express.Router();
// Like GET /o/:id, a private or sandbox answer carries no Access-Control-Allow-Origin; a public one may be read cross-origin.
// A segment's copy is the placement router's choice; a packed one (F3.3) is a ranged read of its ~60 s chunk.
const HLS_TYPE = 'application/vnd.apple.mpegurl';
--const SEGMENT_NAME = /^(init\.mp4|\d{6,}\.m4s)$/;
++const SEGMENT_NAME = /^(init\.mp4|\d{6,9}\.m4s)$/;
const SLICE_FETCH_MS = 15000; // a ranged read of one packed segment from B2/R2
// clipWindow/timelineOf (a clip's window over its source, and the rows its playlists are written from) live in
// objects/hls.js, shared with the v1 VOD/clip discovery.
@@@ -883,11 +883,13 @@ publicRouter.get('/:id/source/:name', hlsRoute(async (req, res, { obj, closed })
const headers = { 'Content-Type': mime, 'Cache-Control': closed ? 'private, no-store' : 'public, max-age=3600' };
const packed = !!row.packed_object_id;
const slice = { offset: Number(row.byte_offset), length: Number(row.byte_length), headers };
-- // The router ranks this node's copy and the durable one (no viewer demand counted per segment: segment-bucket demand
-- // comes later). A local file is streamed; a remote unpacked segment is a redirect to a short presigned URL; a remote
-- // packed one is read as a byte range of its chunk here, since a redirect cannot carry the segment's range.
++ // The router ranks this node's copy and the durable one, and the served read counts as demand for the object and
++ // its segment (the timeline's own bucket): a hot beginning is visible apart from a cold middle. A local file is
++ // streamed; a remote unpacked segment is a redirect to a short presigned URL; a remote packed one is read as a byte
++ // range of its chunk here, since a redirect cannot carry the segment's range.
const router = require('../placement/router');
-- const decision = await router.route({ locations: timeline.locationsOf(row), purpose: 'playback', session: router.sessionFor(req, `${obj.id}/source`),
++ const decision = await router.route({ locations: timeline.locationsOf(row), objectId: obj.id, segment: name,
++ purpose: 'playback', session: router.sessionFor(req, `${obj.id}/source`),
contentType: mime, expiresIn: 300, presign: !packed });
for (const loc of decision.candidates || []) {
if (loc.provider === 'local') {
ddiff --git a/server/placement/demand.js b/server/placement/demand.js
iindex 5b4be99..4c37344 100644
---- a/server/placement/demand.js
++++ b/server/placement/demand.js
@@@ -1,19 +1,27 @@
'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.
++ * Demand rollups (F2.4, docs/media-fabric.md §6, §8): reads per object × region × 5-minute bucket, each carrying the
++ * timeline segment it served (a beginning is hotter than a middle), folded into Valkey counters plus one hot sorted
++ * set per region and bucket and one per object 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
++ * demand:<region>:<bucket>:<object_id> counter, one per served read (the object total) TTL KEY_TTL_S
++ * hot:<region>:<bucket> sorted set, object_id -> reads in that bucket TTL KEY_TTL_S
++ * hot:<region>:<bucket>:<object_id> sorted set, segment -> 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.
++ * a per-viewer region needs an edge-provided header that does not exist yet. The per-object segment sorted set is
++ * capped at TOP_N members (like the in-process set), so a hot VOD with thousands of segments keeps one bounded key
++ * per object × bucket, not one counter key per segment.
*
-- * The tiering sweep (objects/tiering.js, F2.5) reads hotness() with `strict` for its promote/demote eligibility, one
-- * MGET per page of objects, and calls rollup(), which stages media.object.hot for objects over HOT_READS in the window.
++ * Every read bumps the object counter, so the object is the sum over its segments (plus any whole-object reads): the
++ * object-level hotness the tiering sweep's eligibility reads (objects/tiering.js, F2.5) is unchanged, and a segment
++ * read additionally bumps only that segment's member in the per-object segment set (its own read count lives there).
++ *
++ * The tiering sweep reads hotness() with `strict` for its promote/demote eligibility, one MGET per page of objects, and
++ * calls rollup(), which stages media.object.hot for objects over HOT_READS in the window.
*/
const config = require('../config');
@@@ -37,8 +45,11 @@ function available() { return !!vk; }
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(':'));
++/** The object total's counter key (the object is the only counter; segments live in their per-object sorted set). */
function counterKey(r, b, id) { return name('demand', r, String(b), String(id)); }
function hotKey(r, b) { return name('hot', r, String(b)); }
++/** The per-object segment hot set: segment name -> reads in that bucket. */
++function segKey(r, b, id) { return name('hot', r, String(b), String(id)); }
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; };
@@@ -50,43 +61,66 @@ function pruneFallback(now = Date.now()) {
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() } = {}) {
++/** Bump one in-process hot set (member -> reads), evicting the coldest member over TOP_N. */
++function bumpSet(hk, member, expiresAt) {
++ const s = sets.get(hk) || { members: new Map(), expiresAt };
++ s.members.set(member, (s.members.get(member) || 0) + 1);
++ if (s.members.size > TOP_N) {
++ let lo = null;
++ for (const e of s.members) if (e[0] !== member && (!lo || e[1] < lo[1])) lo = e;
++ if (lo) s.members.delete(lo[0]);
++ }
++ sets.set(hk, s);
++}
++
++function bumpCounter(ck, expiresAt) {
++ const c = counters.get(ck) || { n: 0, expiresAt };
++ c.n++;
++ counters.set(ck, c);
++}
++
++/**
++ * One served read of an object, optionally of one of its timeline segments. Best-effort: never throws, never awaits
++ * Valkey. The object total is bumped on every read; with a segment, that segment's member in its per-object hot set
++ * moves too, so hotness({ segment }) can tell a hot beginning from a cold middle.
++ */
++function record({ objectId, segment = null, region: r = region(), now = Date.now() } = {}) {
if (objectId == null || objectId === '') return;
-- const id = String(objectId), b = bucketOf(now);
++ const id = String(objectId), b = bucketOf(now), seg = segment == null || segment === '' ? null : String(segment);
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)));
++ if (seg) {
++ const gk = segKey(r, b, id);
++ quiet(vk.client.zincrby(gk, 1, seg).then((n) => {
++ if (Number(n) !== 1) return null; // not a new member: the set cannot have grown past TOP_N
++ return vk.client.expire(gk, KEY_TTL_S).then(() => vk.client.zremrangebyrank(gk, 0, -(TOP_N + 1)));
++ }));
++ }
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);
++ bumpCounter(counterKey(r, b, id), expiresAt);
++ bumpSet(hotKey(r, b), id, expiresAt);
++ if (seg) bumpSet(segKey(r, b, id), seg, expiresAt);
} 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.
++ * `segment` narrows the read to one timeline segment: with `objectIds`, the segment's score in each object's
++ * per-object hot set (one pipelined ZSCORE per object per bucket); without, the region's hottest TOP_N objects, read
++ * out of each object's per-object hot set (the unbounded walk is capped so it stays cheap).
* A Valkey error answers an empty map: demand is a hint. With `strict`, an error, a timeout or no Valkey at all throws
* instead (a caller that would act on zeros falls back to another signal). → Map(object_id -> reads)
*/
--async function hotness({ region: r = region(), now = Date.now(), buckets = WINDOW_BUCKETS, objectIds = null, strict = false } = {}) {
++async function hotness({ region: r = region(), now = Date.now(), buckets = WINDOW_BUCKETS, objectIds = null, strict = false, segment = null } = {}) {
if (strict && !vk) throw new Error('Valkey is not configured');
++ const seg = segment == null || segment === '' ? null : String(segment);
const out = new Map();
const window = windowOf(now, buckets);
const ids = objectIds ? [...new Set(objectIds.filter((x) => x != null).map(String))] : null;
@@@ -95,8 +129,11 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO
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));
++ // The segment axis reads each object's per-object hot set; issued together, so the auto-pipeline keeps
++ // it one round trip. The object axis reads the object counters, one MGET.
++ const vals = seg
++ ? await timed(Promise.all(ids.flatMap((id) => window.map((b) => vk.client.zscore(segKey(r, b, id), seg)))))
++ : await timed(vk.client.mget(...ids.flatMap((id) => window.map((b) => counterKey(r, b, id)))));
ids.forEach((id, i) => {
let n = 0;
for (let j = 0; j < window.length; j++) n += Number(vals[i * window.length + j]) || 0;
@@@ -104,6 +141,21 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO
});
return out;
}
++ if (seg) {
++ // The segment axis with no object list: the region's hottest TOP_N objects, then that segment out of
++ // each one's set. Not the tiering sweep's path (that always names its object ids, one MGET per page).
++ const all = await timed(Promise.all(window.map((b) => vk.client.zrange(hotKey(r, b), 0, -1, 'WITHSCORES'))));
++ const heat = new Map();
++ for (const flat of all) for (let i = 0; i + 1 < flat.length; i += 2) heat.set(flat[i], (heat.get(flat[i]) || 0) + Number(flat[i + 1]));
++ const list = [...heat].sort((a, b) => b[1] - a[1] || (a[0] < b[0] ? -1 : 1)).slice(0, TOP_N).map(([id]) => id);
++ const groups = await timed(Promise.all(list.flatMap((id) => window.map((b) => vk.client.zscore(segKey(r, b, id), seg)))));
++ list.forEach((id, i) => {
++ let n = 0;
++ for (let j = 0; j < window.length; j++) n += Number(groups[i * window.length + j]) || 0;
++ if (n) 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]));
@@@ -114,11 +166,24 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO
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; }
++ for (const b of window) {
++ const c = seg ? sets.get(segKey(r, b, id)) : counters.get(counterKey(r, b, id));
++ if (c) n += seg ? (c.members.get(seg) || 0) : c.n;
++ }
out.set(id, n);
}
return out;
}
++ if (seg) {
++ const objs = new Set();
++ for (const b of window) { const s = sets.get(hotKey(r, b)); if (s) for (const id of s.members.keys()) objs.add(id); }
++ for (const id of [...objs].slice(0, TOP_N)) {
++ let n = 0;
++ for (const b of window) { const s = sets.get(segKey(r, b, id)); if (s) n += s.members.get(seg) || 0; }
++ if (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);
@@@ -205,5 +270,5 @@ function _reset() { counters.clear(); sets.clear(); announced.clear(); }
module.exports = {
BUCKET_MS, WINDOW_BUCKETS, KEY_TTL_S, TOP_N, HOT_READS, ANNOUNCE_S,
-- useValkey, available, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,
++ useValkey, available, region, bucketOf, counterKey, hotKey, segKey, announceKey, record, hotness, top, rollup, _reset,
};
ddiff --git a/server/placement/router.js b/server/placement/router.js
iindex 704ebd8..3ea272a 100644
---- a/server/placement/router.js
++++ b/server/placement/router.js
@@@ -196,8 +196,10 @@ const DEMAND_PURPOSES = new Set(['playback', 'download']);
/**
* The only function every read path uses.
-- * opts: { object | vod | clip | locations, purpose, session?, range?, presign?, expiresIn?,
++ * opts: { object | vod | clip | locations, objectId?, purpose, session?, segment?, range?, presign?, expiresIn?,
* contentType?, contentDisposition? }
++ * `objectId` names the object a demand hit belongs to when the caller passes no object/vod/clip row (a segment
++ * read: the route already resolved its row and passes explicit locations); `segment` is the timeline segment served.
* returns: { provider, key, url, reason, fallbacks, candidates } — `fallbacks` are provider names, ranked;
* `candidates` the ranked copies with their keys ({ provider, key }), the chosen one first.
*/
@@@ -248,12 +250,13 @@ 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.
++ // Demand (F2.4): one hit per served viewer read (not a derive job), object × region × 5-minute bucket, carrying the
++ // timeline segment the read served when the caller names one (a segment read, whose route passes `objectId`).
// 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 });
++ const id = opts.objectId || (opts.object ? opts.object.id : (opts.vod || opts.clip || {}).object_id);
++ if (id) require('./demand').record({ objectId: id, segment: opts.segment });
} catch { /* demand is a hint */ }
}
const chain = [primary, ...ordered.filter((c) => c !== primary)].filter(Boolean).map((c) => ({ provider: c.provider, key: c.key }));
ddiff --git a/test/clip-timeline.test.js b/test/clip-timeline.test.js
iindex c626daf..bf5e75a 100644
---- a/test/clip-timeline.test.js
++++ b/test/clip-timeline.test.js
@@@ -251,11 +251,19 @@ const express = require('express');
const index = (await get(`/o/${clipObj}/source/index.m3u8${tok}`)).buf.toString();
assert.ok(index.includes(`#EXT-X-MAP:URI="init.mp4${tok}"`) && index.includes(`#EXTINF:3.000,\n000002.m4s${tok}`) && index.includes(`#EXTINF:3.000,\n000003.m4s${tok}`), index);
assert.ok(!/00000[145]\.m4s/.test(index), 'nothing outside the window is listed');
++ const demand = require('../server/placement/demand');
++ demand._reset(); // only the segment reads below count
for (const n of ['init.mp4', '000002.m4s', '000003.m4s']) {
const s = await get(`/o/${clipObj}/source/${n}${tok}`);
assert.strictEqual(s.status, 200, n);
assert.deepStrictEqual(s.buf, bytes[n], `${n}: the source's bytes${n === '000003.m4s' ? ' (a slice of its packed chunk)' : ''}`);
}
++ // The segment route threads the segment into the demand rollup (F2.4): the object total is what it served,
++ // and each segment reads apart from the others.
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ objectIds: [clipObj] })), { [clipObj]: 3 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ objectIds: [clipObj], segment: '000002.m4s' })), { [clipObj]: 1 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ objectIds: [clipObj], segment: '000003.m4s' })), { [clipObj]: 1 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ objectIds: [clipObj], segment: '000001.m4s' })), { [clipObj]: 0 }, 'a segment outside the window was never served');
console.log('✅ a private virtual clip: 404 unsigned, signed playlists over the source window, source bytes and packed slices');
// ── Token scope: the clip's window only ──
ddiff --git a/test/placement-demand.test.js b/test/placement-demand.test.js
iindex 66aa9a9..cd35a6b 100644
---- a/test/placement-demand.test.js
++++ b/test/placement-demand.test.js
@@@ -28,7 +28,9 @@ const path = require('path');
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.counterKey('local', 7, 'obj-a', '000001.m4s'), 'demand:local:7:obj-a', 'counterKey has no segment axis');
assert.strictEqual(demand.hotKey('local', 7), 'hot:local:7');
++ assert.strictEqual(demand.segKey('local', 7, 'obj-a'), 'hot:local:7:obj-a');
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');
@@@ -46,9 +48,21 @@ const path = require('path');
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 }]);
++ // A segment read moves the object total and that segment's own counter, never another segment's; hotness()
++ // aggregates the segments to the object, and the segment axis reads each object's per-segment set.
++ demand.record({ objectId: 'obj-s', segment: 'init.mp4', now: T0 });
++ for (let i = 0; i < 3; i++) demand.record({ objectId: 'obj-s', segment: '000001.m4s', now: T0 });
++ demand.record({ objectId: 'obj-s', segment: '000001.m4s', now: T0 - B }); // inside the hour
++ demand.record({ objectId: 'obj-s', segment: '000002.m4s', now: T0 - demand.WINDOW_BUCKETS * B }); // 13th bucket back: outside
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-s'] })), { 'obj-s': 5 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-s'], segment: '000001.m4s' })), { 'obj-s': 4 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-s'], segment: 'init.mp4' })), { 'obj-s': 1 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-s'], segment: '000002.m4s' })), { 'obj-s': 0 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-s'], segment: '000009.m4s' })), { 'obj-s': 0 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, segment: '000001.m4s' })), { 'obj-s': 4 }, 'the segment axis without an object list');
// 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');
++ console.log('✅ hotness() sums the window per object and region; segments aggregate to the object and read apart; 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 = [];
@@@ -61,6 +75,16 @@ const path = require('path');
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 zscore(k, m) { calls.push(['zscore', k, m]); const z = zsets.get(k); return z && z.has(m) ? String(z.get(m)) : null; },
++ async zremrangebyrank(k, start, stop) {
++ calls.push(['zremrangebyrank', k, start, stop]);
++ const z = zsets.get(k); if (!z) return 0;
++ const sorted = [...z.entries()].sort((a, b) => a[1] - b[1] || (a[0] < b[0] ? -1 : 1));
++ const len = sorted.length, s = start < 0 ? Math.max(0, len + start) : start, e = stop < 0 ? len + stop : stop;
++ let removed = 0;
++ for (let i = s; i <= Math.min(e, len - 1); i++) { z.delete(sorted[i][0]); removed++; }
++ return removed;
++ },
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; },
@@@ -81,7 +105,50 @@ const path = require('path');
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');
++ // A segment read moves the object's per-segment set and the object total with it — no per-segment counter key.
++ demand.record({ objectId: 'obj-a', segment: '000001.m4s', now: T0 });
++ demand.record({ objectId: 'obj-a', segment: '000001.m4s', now: T0 });
++ demand.record({ objectId: 'obj-a', segment: '000002.m4s', now: T0 });
++ await new Promise((resolve) => setImmediate(resolve));
++ const gk = `ov:media:hot:local:${b}:obj-a`;
++ assert.ok(!store.has(`ov:media:demand:local:${b}:obj-a:000001.m4s`), 'no per-segment counter key');
++ assert.deepStrictEqual([...zsets.get(gk)], [['000001.m4s', 2], ['000002.m4s', 1]]);
++ assert.strictEqual(ttls.get(gk), 7200);
++ assert.strictEqual(calls.filter((c) => c[0] === 'expire' && c[1] === gk).length, 2, 'TTL set once per new segment member');
++ assert.strictEqual(calls.filter((c) => c[0] === 'zremrangebyrank' && c[1] === gk && c[3] === -(demand.TOP_N + 1)).length, 2, 'the capped set is trimmed when a new member arrives');
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0 })), { 'obj-a': 5, 'obj-b': 1 }, 'the object total grew by its segments');
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-a'], segment: '000001.m4s' })), { 'obj-a': 2 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, segment: '000001.m4s' })), { 'obj-a': 2 }, 'the segment axis without an object list');
++ console.log('✅ Valkey keys ov:media:demand:<region>:<bucket>:<id> and ov:media:hot:<region>:<bucket>[:<id>], TTL 7200 s');
++
++ // 3b. Bounded keys: 1000 segment reads of one object leave one per-object set of at most TOP_N members and no
++ // per-segment counter keys, and the object total stays exact (every read bumps it).
++ calls.length = 0; store.clear(); zsets.clear();
++ const hotB = demand.bucketOf(T0), hotGk = `ov:media:hot:local:${hotB}:obj-hot`, hotCk = `ov:media:demand:local:${hotB}:obj-hot`;
++ for (let i = 0; i < 1000; i++) demand.record({ objectId: 'obj-hot', segment: `${String(i + 1).padStart(6, '0')}.m4s`, now: T0 });
++ await new Promise((resolve) => setImmediate(resolve));
++ assert.strictEqual(zsets.size, 2, 'the region object set and one per-object segment set, nothing per segment');
++ assert.ok(zsets.has(hotGk));
++ assert.strictEqual(zsets.get(`ov:media:hot:local:${hotB}`).size, 1, 'the region object set holds the one object');
++ assert.strictEqual(zsets.get(hotGk).size, demand.TOP_N, 'the segment set is capped at TOP_N');
++ assert.strictEqual([...store.keys()].filter((k) => k.startsWith('ov:media:demand:')).length, 1, 'only the object counter key');
++ assert.strictEqual(store.get(hotCk), 1000, 'the object total stays exact');
++ assert.ok([...store.keys()].every((k) => !/\.m4s$/.test(k)), 'no per-segment counter key');
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-hot'] })), { 'obj-hot': 1000 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-hot'], segment: '001000.m4s' })), { 'obj-hot': 1 }, 'a surviving segment reads from the set');
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-hot'], segment: '000001.m4s' })), { 'obj-hot': 0 }, 'an evicted segment is gone');
++ // The in-process fallback is bounded the same way: one object counter and one capped segment set.
++ demand.useValkey(null);
++ demand._reset();
++ for (let i = 0; i < 1000; i++) demand.record({ objectId: 'obj-hot', segment: `${String(i + 1).padStart(6, '0')}.m4s`, now: T0 });
++ assert.deepStrictEqual(Object.fromEntries(await demand.hotness({ now: T0, objectIds: ['obj-hot'] })), { 'obj-hot': 1000 });
++ let present = 0;
++ for (let i = 1; i <= 1000; i++) {
++ if ((await demand.hotness({ now: T0, objectIds: ['obj-hot'], segment: `${String(i).padStart(6, '0')}.m4s` })).get('obj-hot')) present++;
++ }
++ assert.strictEqual(present, demand.TOP_N, 'the fallback keeps only TOP_N segments');
++ console.log('✅ 1000 segment reads stay bounded: one capped set per object, object total exact, fallback bounded');
++ demand.useValkey(fake);
// 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');
@@@ -93,9 +160,13 @@ const path = require('path');
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' });
++ // A segment read names its object through `objectId` and the segment through `segment` (routes.js /o/:id/source/:name).
++ await router.route({ locations: served.locations, objectId: 'obj-sg', segment: '000002.m4s', 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');
++ const incrKeys = calls.filter((c) => c[0] === 'incr').map((c) => c[1]);
++ assert.deepStrictEqual(incrKeys.map((k) => k.split(':').pop()), ['obj-r', 'obj-v', 'obj-sg'], 'one object counter per served read, no per-segment counter');
++ assert.ok(calls.some((c) => c[0] === 'zincrby' && c[1].endsWith(':obj-sg') && c[3] === '000002.m4s'), 'the segment read names the object and the segment in its per-object set');
++ console.log('✅ router: one demand hit per served playback/download, none for derive or no copy, segments named');
// 5. A failing Valkey (rejecting, throwing, hanging) leaves the read path untouched.
const unhandled = [];
@@@ -182,6 +253,6 @@ const path = require('path');
stub.close();
await db.close();
fs.rmSync(tmp, { recursive: true, force: true });
-- console.log('placement-demand: all six cases passed');
++ console.log('placement-demand: all seven cases passed');
process.exit(0);
})().catch((err) => { console.error(err); process.exit(1); });