openvibers/openvibe.media Commits
T4 F2.5 — the tiering sweep's eligibility reads Valkey demand in… (#22)
Changes
@@@ -34,7 +34,8 @@"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, 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"],"updated": "2026-10-03"}@@@ -133,7 +133,19 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:(`MEDIA_DEMAND_REGION`, default `local`); per-viewer regions wait for an edge-provided header. `hotness()` sums the last12 buckets (one hour) per object. The writes are fire-and-forget: a slow, failing or absent Valkey never delays or failsa 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.++- **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++ other classes 60 / 6. At or above P an object may promote, below D its R2 copy may go (still in R2 at least++ `demoteIdleDays`), in between nothing moves. Size bounds, per-class budgets, minimum residency, holds and back-off are++ unchanged; value per dollar ranks at the hour's rate (reads × 24 × 30 a month). The reads are batched: one `MGET` per++ page of 500 objects (the promotion scan pages by id; the R2 copies are read in pages of 500), never a round trip per++ object. Valkey absent, erroring or timing out (2 s) anywhere in the scan sends the **whole** sweep back to the daily++ PostgreSQL view counts (`media.object_tier` thresholds, today's rules), with one warning per sweep and++ `media_sweep_demand_source_total{source="valkey"|"pg"}`; both candidate lists are read before anything moves. Every++ decision's `inputs.demand` (`{ source, metric, value, threshold[, region, window_s] }`) and its reason say which source++ and value drove it, dry runs included; a sweep's `media.replica.requested` and `media.replica.draining` carry the same++ `demand`, and the sweep result carries `demand_source` (and `demand_fallback`).## 7. Delivery: sticky, measured, canaried@@@ -173,6 +185,9 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} enginessince }`) through the placement outbox for objects with at least 100 reads in the last hour, at most once per object perhour (`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`.++- **Shipped (F2.5):** the placement loop reads those counters: the tiering sweep promotes and demotes on the hour's reads++ per object with per-class hysteresis (§6), and falls back to the daily PostgreSQL counts for a whole sweep when Valkey++ does not answer.- **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`.@@@ -191,7 +206,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thrmoves with a cleanup job; per-class hysteresis bands; `media_storage_alerts_total`, provider latency histograms; thecorrected price table; the nginx shield (slice + cache lock) on the Media host.- **F2 one placement engine:** every class through one sweep, value-per-dollar under budgets, dry-run + simulator,-- decision log with class and reason, the events above, Valkey rollups of demand.++ decision log with class and reason, the events above, Valkey rollups of demand (F2.4, shipped) and the sweep's++ eligibility on them with per-class hysteresis and a PostgreSQL fallback (F2.5, shipped).- **F3 segment-native video:** CMAF/HLS recording with the growing DVR playlist, the timeline index, packing afterfinalize, virtual clips, sprites and captions on the timeline, MP4 fallback; Live's player moves to HLS.- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral@@@ -10,6 +10,12 @@* The thresholds say who may move; value per dollar (../placement/value.js, R2 Standard prices from* ../placement/cost-tiers.js) says who moves first: each class spends its per-sweep budget* (../placement/storage-policy.js) on its highest-scoring promotions and its lowest-scoring demotions.++ * Demand (F2.5): the sweep's eligibility reads each object's reads in the last hour for this deployment's region++ * (../placement/demand.js hotness(), one MGET per page of objects) against its class's hysteresis band++ * (promoteReadsPerHour / demoteReadsPerHour, media.storage_policy): at or above the first it may promote, below the++ * second it may demote, in between nothing moves. Without Valkey, or when a read fails or times out, the whole sweep++ * reads the daily PostgreSQL counts instead (the thresholds above), logged once and counted in++ * media_sweep_demand_source_total{source}. Each decision's inputs.demand says which source and value drove it.** runSweep({ trigger }) one pass: rotate the popularity counts, then promote and demote by the policy.* Step 4 of the storage sweep (server/vod/vod-storage.js runSweep), which a restore@@@ -187,6 +193,7 @@ async function frame(objectId, action, ctx) {const held = await model().isHeld(obj.id);const inputs = await inputsOf(obj, { locs, pop, held, settings });if (ctx.valuePerDollar != null) inputs.value_per_dollar = ctx.valuePerDollar;++ if (ctx.demand) inputs.demand = ctx.demand;const canon = canonicalLocation(obj, locs);const trigger = ctx.trigger || 'manual';const why = ctx.reason || 'requested';@@@ -232,7 +239,7 @@ async function promote(objectId, ctx = {}) {await db.getDb().tx(async () => {await events().recordPlacement('media.replica.requested', obj.app_id, subjectOf(obj), {object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'promote',-- from: canon ? canon.provider : null, to: 'r2', provider: 'r2',++ from: canon ? canon.provider : null, to: 'r2', provider: 'r2', ...(ctx.demand ? { demand: ctx.demand } : {}),});});events().kick();@@@ -299,7 +306,7 @@ async function demote(objectId, ctx = {}) {await db.getDb().tx(async () => {await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',-- provider: 'r2', key: r2.key,++ provider: 'r2', key: r2.key, ...(ctx.demand ? { demand: ctx.demand } : {}),});});events().kick();@@@ -326,8 +333,8 @@ async function demote(objectId, ctx = {}) {// ── Candidates ───────────────────────────────────────────────/**-- * Value per dollar of an R2 copy of this row (../placement/value.js): its 7-day unique viewers scaled to a month-- * of requests, against R2 Standard's storage, read and retrieval prices (media.cost_tiers) and, for a promotion,++ * Value per dollar of an R2 copy of this row (../placement/value.js): its reads in the last hour (Valkey demand)++ * or else its 7-day unique viewers, scaled to a month of requests, against R2 Standard's storage, read and retrieval prices (media.cost_tiers) and, for a promotion,* the class's minimum residency beyond that month (media.storage_policy), which the copy commits to. Free* allowances are not counted: a promotion is priced at the margin.*/@@@ -338,7 +345,8 @@ function scoreOf(row, cls, { committing = false } = {}) {return value.score({// A row that carries a month of counts (the demotion scan) is read as is: an idle copy has no viewer in the// 7-day window, so only the longer count tells a copy last viewed 20 days ago from one never viewed.-- requests: row.unique_viewers_30d != null ? Number(row.unique_viewers_30d) || 0++ requests: row.reads_1h != null ? (Number(row.reads_1h) || 0) * 24 * MONTH_DAYS++ : row.unique_viewers_30d != null ? Number(row.unique_viewers_30d) || 0: (Number(row.unique_viewers_7d) || 0) * MONTH_DAYS / popularity.WINDOW_DAYS,bytes: Number(row.size_bytes) || 0,storageCost: storage,@@@ -360,8 +368,10 @@ const SWEEP_KEEP = 500; // top-ranked promotion candidates a sweep* before the limit applies (a most-viewed prefix would drop smaller objects that are worth more per dollar), but* the scan reads SCAN_PAGE rows at a time and holds only the best `limit`, so memory stays bounded.* excludeSandbox drops developer-sandbox tenants in the query (their objects are never tiered).++ * With `demand` (the sweep's Valkey reader), the hourly reads replace the daily counts: see hotPromotionCandidates.*/--async function promotionCandidates(settings, { now = Date.now(), appId = null, limit = 200, excludeSandbox = false } = {}) {++async function promotionCandidates(settings, { now = Date.now(), appId = null, limit = 200, excludeSandbox = false, demand = null } = {}) {++ if (demand) return await hotPromotionCandidates(settings, demand, { appId, limit, excludeSandbox });const today = popularity.dayOf(now);const args = [popularity.windowStart(today, popularity.WINDOW_DAYS)];const rank = (a, b) => b.value_per_dollar - a.value_per_dollar || Number(b.unique_viewers_7d) - Number(a.unique_viewers_7d) || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0);@@@ -381,6 +391,7 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, lfor (const r of rows) {r.class = storagePolicy.classOf(r);r.value_per_dollar = scoreOf(r, r.class, { committing: true });++ r.demand = { source: 'pg', metric: 'unique_viewers_7d', value: Number(r.unique_viewers_7d) || 0, threshold: settings.promoteMinUniqueViewers7d };}best = best.concat(rows).sort(rank).slice(0, limit);if (rows.length < SCAN_PAGE) return best;@@@ -388,12 +399,51 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, l}}++/** The demand fields a decision records for a Valkey reading: source, the hourly reads and the class's threshold. */++function hotDemand(demand, reads, threshold) {++ return { source: 'valkey', metric: 'reads_1h', value: reads, threshold, region: demand.region, window_s: demand.window_s };++}++++/**++ * The Valkey form of promotionCandidates: ready native objects in the size bounds with no R2 copy, paged by id++ * (SCAN_PAGE rows, one hotness MGET per page), kept when their reads in the last hour reach their class's++ * promoteReadsPerHour; ranked by value per dollar at that rate, then reads. A failed read throws (the sweep falls back).++ */++async function hotPromotionCandidates(settings, demand, { appId = null, limit = 200, excludeSandbox = false } = {}) {++ const rank = (a, b) => b.value_per_dollar - a.value_per_dollar || b.reads_1h - a.reads_1h || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0);++ let best = [];++ for (let after = ''; ;) {++ const rows = await db.all(`SELECT o.* FROM media_objects o++ WHERE o.id > ? AND o.legacy_ref IS NULL AND o.lifecycle_status = 'ready' AND o.size_bytes >= ? AND o.size_bytes <= ?++ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.provider = 'r2' AND l.state = 'present')${appId ? ' AND o.app_id = ?' : ''}${excludeSandbox++ ? " AND NOT EXISTS (SELECT 1 FROM apps a WHERE a.app_id = o.app_id AND a.env = 'sandbox')" : ''}++ ORDER BY o.id LIMIT ?`,++ [after, Math.ceil(settings.promoteMinSizeMb * MB), Math.floor(settings.promoteMaxSizeMb * MB), ...(appId ? [appId] : []), SCAN_PAGE]);++ const heat = rows.length ? await demand.read(rows.map((r) => r.id)) : new Map();++ const hot = [];++ for (const r of rows) {++ r.class = storagePolicy.classOf(r);++ r.reads_1h = heat.get(String(r.id)) || 0;++ const { promoteReadsPerHour } = storagePolicy.bandFor(r.class);++ if (r.reads_1h < promoteReadsPerHour) continue;++ r.value_per_dollar = scoreOf(r, r.class, { committing: true });++ r.demand = hotDemand(demand, r.reads_1h, promoteReadsPerHour);++ hot.push(r);++ }++ best = best.concat(hot).sort(rank).slice(0, limit);++ if (rows.length < SCAN_PAGE) return best;++ after = rows[rows.length - 1].id;++ }++}++/*** Native objects with an R2 copy that should lose it: not ready any more (first, whatever their score), or idle,* lowest value per dollar of keeping the copy first (its viewers over the kept month, since an idle copy has none-- * in the last 7 days), then the largest. → [{ row, reason, class, value_per_dollar }]++ * in the last 7 days), then the largest. With `demand` (the sweep's Valkey reader, one MGET per SCAN_PAGE copies), a++ * ready object is cold when its reads in the last hour fall below its class's demoteReadsPerHour instead of idle by++ * the daily counts (still in R2 at least demoteIdleDays either way). → [{ row, reason, class, value_per_dollar, demand }]*/--async function demotionCandidates(settings, { now = Date.now(), appId = null } = {}) {++async function demotionCandidates(settings, { now = Date.now(), appId = null, demand = null } = {}) {const idleFrom = popularity.windowStart(popularity.dayOf(now), settings.demoteIdleDays);const monthFrom = popularity.windowStart(popularity.dayOf(now), popularity.KEEP_DAYS);const rows = await db.all(`SELECT o.*, l.created_at AS r2_since,@@@ -402,15 +452,29 @@ async function demotionCandidates(settings, { now = Date.now(), appId = null } =FROM media_locations l JOIN media_objects o ON o.id = l.object_idWHERE l.provider = 'r2' AND o.legacy_ref IS NULL${appId ? ' AND o.app_id = ?' : ''}ORDER BY (o.lifecycle_status = 'ready'), o.id`, appId ? [monthFrom, appId] : [monthFrom]);++ if (demand) {++ for (let i = 0; i < rows.length; i += SCAN_PAGE) {++ const page = rows.slice(i, i + SCAN_PAGE);++ const heat = await demand.read(page.map((r) => r.id));++ for (const r of page) r.reads_1h = heat.get(String(r.id)) || 0;++ }++ }const out = [];for (const r of rows) {const cls = storagePolicy.classOf(r);-- const scored = { row: r, class: cls, value_per_dollar: scoreOf(r, cls) };++ const { demoteReadsPerHour } = storagePolicy.bandFor(cls);++ const scored = { row: r, class: cls, value_per_dollar: scoreOf(r, cls),++ demand: demand ? hotDemand(demand, r.reads_1h, demoteReadsPerHour) : { source: 'pg', metric: 'last_viewed_day', value: r.last_viewed_day || null, threshold: idleFrom } };if (r.lifecycle_status !== 'ready') { out.push({ ...scored, reason: `the object is ${r.lifecycle_status}` }); continue; }-- const idle = !r.last_viewed_day || r.last_viewed_day < idleFrom;const since = msOf(r.r2_since);const settled = !Number.isFinite(since) || now - since >= settings.demoteIdleDays * DAY_MS;-- if (idle && settled) {++ if (!settled) continue;++ if (demand) {++ if (r.reads_1h < demoteReadsPerHour) {++ out.push({ ...scored, reason: `${r.reads_1h} reads in the last hour (demand valkey, region ${demand.region}) < demoteReadsPerHour ${demoteReadsPerHour} (class ${cls}); `++ + `in R2 since ${r.r2_since || 'unknown'} (at least demoteIdleDays ${settings.demoteIdleDays})` });++ }++ } else if (!r.last_viewed_day || r.last_viewed_day < idleFrom) {out.push({ ...scored, reason: `no viewer since ${r.last_viewed_day || 'the kept counts began'} (idle for demoteIdleDays ${settings.demoteIdleDays}); in R2 since ${r.r2_since || 'unknown'}` });}}@@@ -426,6 +490,11 @@ function rankNote(cls, score, slot, budgetKey) {}function promoteReason(c, s, slot) {++ if (c.demand && c.demand.source === 'valkey') {++ return `${c.reads_1h} reads in the last hour (demand valkey, region ${c.demand.region}) >= promoteReadsPerHour ${c.demand.threshold} (class ${c.class}); `++ + `${(Number(c.size_bytes) / MB).toFixed(1)} MB (promoteMinSizeMb ${s.promoteMinSizeMb} to promoteMaxSizeMb ${s.promoteMaxSizeMb}); `++ + rankNote(c.class, c.value_per_dollar, slot, 'maxPromotionsPerSweep');++ }return `${c.unique_viewers_7d} unique viewers in 7 days >= promoteMinUniqueViewers7d ${s.promoteMinUniqueViewers7d}; last viewed ${c.last_viewed_day} `+ `(within promoteRecentAccessDays ${s.promoteRecentAccessDays}); ${(Number(c.size_bytes) / MB).toFixed(1)} MB (promoteMinSizeMb ${s.promoteMinSizeMb} to promoteMaxSizeMb ${s.promoteMaxSizeMb}); `+ rankNote(c.class, c.value_per_dollar, slot, 'maxPromotionsPerSweep');@@@ -465,8 +534,36 @@ async function residencyRefusal(obj, cls, action, now) {return `minimum residency of ${minResidencyMs} ms for class ${cls} (media.storage_policy) not reached since the last move at ${lastAt}; ${action} suppressed`;}++/** The sweep's Valkey demand reader, or null when Valkey is not configured. read(ids) throws on any failure. */++function valkeyDemand(now) {++ const demand = require('../placement/demand');++ if (!demand.available()) return null;++ return { region: demand.region(), window_s: (demand.WINDOW_BUCKETS * demand.BUCKET_MS) / 1000,++ read: (ids) => demand.hotness({ objectIds: ids, now, strict: true }) };++}++/**-- * One pass over native objects. → { gate, promoted, demoted, would_promote, would_demote, already, refused, failed,++ * Both candidate lists, from one demand source for the whole sweep: Valkey's hourly reads, or (Valkey absent, or a++ * read failed or timed out anywhere in the scan) the daily PostgreSQL counts. Read before anything moves, so a++ * failure part-way never leaves a sweep acting on two signals. → { source, fallback, demote, promote }++ */++async function sweepCandidates(settings, now) {++ const heat = valkeyDemand(now);++ let fallback = heat ? null : 'Valkey is not configured';++ if (heat) {++ try {++ return { source: 'valkey', fallback, demote: await demotionCandidates(settings, { now, demand: heat }),++ promote: await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true, demand: heat }) };++ } catch (err) {++ fallback = `Valkey read failed: ${err.message}`;++ }++ }++ return { source: 'pg', fallback, demote: await demotionCandidates(settings, { now }),++ promote: await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true }) };++}++++/**++ * One pass over native objects. → { gate, demand_source, demand_fallback?, promoted, demoted, would_promote, would_demote, already, refused, failed,* repeats, skipped_backoff, candidates: { promote, demote }, rotated, errors }*/async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {@@@ -478,27 +575,35 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {const out = { gate: !!settings.active, promoted: 0, demoted: 0, would_promote: 0, would_demote: 0, already: 0, refused: 0, failed: 0,repeats: 0, skipped_backoff: 0, candidates: { promote: 0, demote: 0 }, rotated: await popularity.rotate({ now }), errors: [] };++ const picked = await sweepCandidates(settings, now);++ out.demand_source = picked.source;++ if (picked.fallback) {++ out.demand_fallback = picked.fallback;++ console.warn(`[ObjectTiers] Sweep eligibility reads the daily PostgreSQL counts: ${picked.fallback}`);++ }++ require('../placement/metrics-binding').inc('media_sweep_demand_source_total', { source: picked.source });++// Demotions first: they free the R2 cache, and a deleted object's copy should not wait behind promotions.-- const demote_ = await demotionCandidates(settings, { now });++ const demote_ = picked.demote;out.candidates.demote = demote_.length;// Budgets and residency are per class (media.storage_policy); a class spends only its own budget, in the// candidates' value-per-dollar order. A refusal (residency, hold) or a back-off spends nothing, so the next// in line takes the slot.const spent = { promote: {}, demote: {} };const room = (action, cls) => (spent[action][cls] || 0) < storagePolicy.budgetFor(cls)[action === 'promote' ? 'maxPromotionsPerSweep' : 'maxDemotionsPerSweep'];-- for (const { row, reason, class: cls, value_per_dollar: score } of demote_) {++ for (const { row, reason, class: cls, value_per_dollar: score, demand } of demote_) {if (!room('demote', cls)) continue;if (await inBackoff(row.id, 'demote')) { out.skipped_backoff++; continue; }// A copy of an object that is no longer ready is removed regardless of residency.const residency = row.lifecycle_status === 'ready' ? await residencyRefusal(row, cls, 'demote', now) : null;-- const r = await demote(row.id, { trigger, settings, now, residency, valuePerDollar: score,++ const r = await demote(row.id, { trigger, settings, now, residency, valuePerDollar: score, demand,reason: `${reason}; ${rankNote(cls, score, (spent.demote[cls] || 0) + 1, 'maxDemotionsPerSweep')}` });tally(out, { ...r, action: 'demote' });if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.demote[cls] = (spent.demote[cls] || 0) + 1;}// Every eligible candidate is ranked (not a most-viewed prefix); the best SWEEP_KEEP are kept, so no class is crowded out of its budget.-- const promote_ = await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true });++ const promote_ = picked.promote;out.candidates.promote = promote_.length;for (const c of promote_) {const cls = c.class;@@@ -506,13 +611,12 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }const residency = await residencyRefusal(c, cls, 'promote', now);const r = await promote(c.id, { trigger, reason: promoteReason(c, settings, (spent.promote[cls] || 0) + 1), settings, now, residency, valuePerDollar: c.value_per_dollar,-- pop: { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } });++ demand: c.demand, pop: picked.source === 'pg' ? { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } : undefined });tally(out, { ...r, action: 'promote' });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.++ // Demand (F2.4): stage media.object.hot for objects over the hourly read threshold.out.demand = await require('../placement/demand').rollup({ now });if (out.promoted || out.demoted || out.failed) {@@@ -12,8 +12,8 @@* <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.++ * 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.*/const config = require('../config');@@@ -32,6 +32,8 @@ const sets = new Map(); // fallback: hot key -> { members: Map(objecconst announced = new Map(); // fallback: announce key -> expiresAtfunction useValkey(handle) { vk = handle && handle.client ? handle : null; }++/** Valkey is configured: counts are shared across processes (the in-process fallback only sees this one). */++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(':'));@@@ -80,9 +82,11 @@ function record({ objectId, region: r = region(), now = Date.now() } = {}) {/*** 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)++ * 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 } = {}) {++async function hotness({ region: r = region(), now = Date.now(), buckets = WINDOW_BUCKETS, objectIds = null, strict = false } = {}) {++ if (strict && !vk) throw new Error('Valkey is not configured');const out = new Map();const window = windowOf(now, buckets);const ids = objectIds ? [...new Set(objectIds.filter((x) => x != null).map(String))] : null;@@@ -121,6 +125,7 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO}return out;} catch (err) {++ if (strict) throw err;console.warn('[Demand] hotness not read:', err.message);return ids ? new Map(ids.map((id) => [id, 0])) : new Map();}@@@ -200,5 +205,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, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,++ useValkey, available, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,};@@@ -15,6 +15,7 @@* media_router_decisions_total{provider,purpose,reason} counter* media_presign_cache_total{result} counter* media_storage_alerts_total{kind} counter++ * media_sweep_demand_source_total{source} counter (valkey | pg: the object sweep's demand signal)*/'use strict';@@@ -28,6 +29,7 @@ let breakerGauge = null; // gauge (collect)let decisions = null; // counterlet presign = null; // counterlet alerts = null; // counter++let demandSource = null; // counterfunction bind(registry) {if (latency) return; // idempotent@@@ -69,6 +71,13 @@ function bind(registry) {labelNames: ['kind'],});++ demandSource = registry.counter({++ name: 'media_sweep_demand_source_total',++ help: 'Object tiering sweeps by the demand signal their eligibility read: valkey (hourly reads) or pg (daily view counts, the fallback)',++ labelNames: ['source'],++ });++ for (const source of ['valkey', 'pg']) demandSource.inc({ source }, 0);++// Bind every hook the placement modules use.router.bindMetric((name, labels) => inc(name, labels));presignCache.bindMetric((name, labels) => inc(name, labels));@@@ -80,6 +89,7 @@ function inc(name, labels) {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);++ else if (name === 'media_sweep_demand_source_total') demandSource && demandSource.inc(labels || {}, 1);}/** Hook the signals module's recordSuccess/recordFailure to fill the histogram/counter. */@@@ -87,4 +97,4 @@ function observeLatency(provider, op, ms) { if (latency && ms > 0) latency.obserfunction 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++module.exports = { bind, inc, observeLatency, observeError, alert, _reset: () => { latency = errors = breakerGauge = decisions = presign = alerts = demandSource = null; } };\\ No newline at end of file@@@ -10,8 +10,16 @@ const db = require('../db/database');// The defaults match media.object_tier's per-sweep limits (server/objects/tier-policy.js).const BASE = { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0 };const CLASSES = ['video', 'image', 'download', 'game-asset', 'attachment', 'backup'];++// The sweep's hysteresis band (F2.5): reads in the last hour for the deployment's region (placement/demand.js++// hotness()) at or above promoteReadsPerHour promote, below demoteReadsPerHour demote, in between nothing moves.++// Images are small and cheap to serve locally, so they need more reads; downloads are large, so fewer.++const HYSTERESIS = { promoteReadsPerHour: 60, demoteReadsPerHour: 6 };++const CLASS_HYSTERESIS = {++ image: { promoteReadsPerHour: 300, demoteReadsPerHour: 30 },++ download: { promoteReadsPerHour: 30, demoteReadsPerHour: 3 },++};const DEFAULTS = {-- classes: Object.fromEntries(CLASSES.map((name) => [name, { ...BASE }])),++ classes: Object.fromEntries(CLASSES.map((name) => [name, { ...BASE, ...HYSTERESIS, ...CLASS_HYSTERESIS[name] }])),};const CLASS_SCHEMA = {@@@ -21,6 +29,8 @@ const CLASS_SCHEMA = {maxPromotionsPerSweep: { type: 'integer', minimum: 0, maximum: 100 },maxDemotionsPerSweep: { type: 'integer', minimum: 0, maximum: 500 },minResidencyMs: { type: 'integer', minimum: 0 },++ promoteReadsPerHour: { type: 'integer', minimum: 1, maximum: 1e9 },++ demoteReadsPerHour: { type: 'integer', minimum: 0, maximum: 1e9 },},};const SCHEMA = {@@@ -35,13 +45,23 @@ const SCHEMA = {},};++/** Each class's band is a band: demoteReadsPerHour below promoteReadsPerHour (an object never both promotes and demotes). */++function validate(v) {++ const errors = [];++ for (const name of CLASSES) {++ const c = { ...DEFAULTS.classes[name], ...v.classes?.[name] };++ if (c.demoteReadsPerHour >= c.promoteReadsPerHour) errors.push(`classes.${name}: demoteReadsPerHour must be below promoteReadsPerHour`);++ }++ return errors.length ? errors : true;++}++let store = null;async function init() {if (store) return store;store = await config.createConfigStore({db: db.getDb(), service: 'media', namespace: 'media.storage_policy',-- defaults: DEFAULTS, schema: SCHEMA,++ defaults: DEFAULTS, schema: SCHEMA, validate,legacy: () => ({}),onActivate: async () => {},log: { info: (m) => console.log(`[StoragePolicy] ${m}`), warn: (m) => console.warn(`[StoragePolicy] ${m}`), error: (m) => console.error(`[StoragePolicy] ${m}`) },@@@ -72,7 +92,7 @@ function settings() {let configured;try { configured = get().get(); } catch { configured = {}; }const classes = {};-- for (const name of CLASSES) classes[name] = { ...BASE, ...configured.classes?.[name] };++ for (const name of CLASSES) classes[name] = { ...DEFAULTS.classes[name], ...configured.classes?.[name] };return { classes };}@@@ -97,6 +117,12 @@ function budgetFor(name) {return { maxPromotionsPerSweep, maxDemotionsPerSweep };}++/** The class's hourly promote and demote thresholds (hysteresis: demote < promote). */++function bandFor(name) {++ const { promoteReadsPerHour, demoteReadsPerHour } = classPolicy(name);++ return { promoteReadsPerHour, demoteReadsPerHour };++}++function timestamp(value) {if (value instanceof Date) return value.getTime();if (typeof value === 'number') return value;@@@ -113,4 +139,4 @@ function mayMove({ class: name, lastMovedAt, now = Date.now() }) {return Number.isFinite(last) && Number.isFinite(current) && current - last >= minResidencyMs;}--module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, mayMove, _reset: () => { store = null; } };++module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, bandFor, mayMove, _reset: () => { store = null; } };@@@ -541,6 +541,105 @@ const crypto = require('crypto');await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);console.log('✅ gate off: residency- or hold-blocked moves are logged as refused and spend no slot');++ // ── 9e. Demand from Valkey (F2.5): hourly reads against each class's band; the daily counts when Valkey fails ──++ const demand = require('../server/placement/demand');++ const sweepWarns = [];++ const warn0 = console.warn;++ const vkReads = new Map();++ const mgets = [];++ let vkMode = 'up';++ demand.useValkey({ key: (...p) => 'ov:media:' + p.join(':'), client: {++ async mget(...keys) {++ mgets.push(keys.length);++ if (vkMode === 'down') throw new Error('ECONNREFUSED');++ if (vkMode === 'hung') return await new Promise(() => {});++ return keys.map((k) => (vkReads.has(k) ? String(vkReads.get(k)) : null));++ },++ async zrange() { return []; },++ async set() { return 'OK'; },++ async del() { return 1; },++ } });++ const setReads = (id, n) => vkReads.set(demand.counterKey(demand.region(), demand.bucketOf(Date.now()), id), n);++ const inR2 = async (o, days) => {++ buckets['r2-bucket'].set(`objects/live/${o.id}`, o.bytes);++ await model.upsertLocation(o.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${o.id}`, storage_class: 'cache', state: 'present', size_bytes: o.bytes.length, checksum: o.sum, verified: true });++ await ageR2(o.id, days);++ };++ const candViews = await db.all('SELECT object_id, day, unique_viewers, last_viewed_at FROM media_object_views_daily WHERE object_id = ?', [cand.id]);++ await db.run('DELETE FROM media_object_views_daily'); // no daily counts: only Valkey can move anything++ const hotVid = await addObject({ mime: 'video/mp4' }); setReads(hotVid.id, 75); // >= video promoteReadsPerHour 60++ const warmVid = await addObject({ mime: 'video/mp4' }); setReads(warmVid.id, 30); // in the band: stays local++ const warmR2 = await addObject({ mime: 'video/mp4' }); setReads(warmR2.id, 30); await inR2(warmR2, 30); // in the band: keeps R2++ const coldVid = await addObject({ mime: 'video/mp4' }); setReads(coldVid.id, 2); await inR2(coldVid, 30); // < video demoteReadsPerHour 6++ const sz = policy.settings();++ const scanned = Number((await db.get(`SELECT COUNT(*) AS n FROM media_objects o WHERE o.legacy_ref IS NULL AND o.lifecycle_status = 'ready' AND o.size_bytes >= ? AND o.size_bytes <= ?++ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.provider = 'r2' AND l.state = 'present')++ AND NOT EXISTS (SELECT 1 FROM apps a WHERE a.app_id = o.app_id AND a.env = 'sandbox')`, [Math.ceil(sz.promoteMinSizeMb * 1048576), Math.floor(sz.promoteMaxSizeMb * 1048576)])).n);++ const copies = Number((await db.get("SELECT COUNT(*) AS n FROM media_locations l JOIN media_objects o ON o.id = l.object_id WHERE l.provider = 'r2' AND o.legacy_ref IS NULL")).n);++ console.warn = (...a) => { sweepWarns.push(a.join(' ')); };++ s = await tiering.runSweep();++ console.warn = warn0;++ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['valkey', undefined], JSON.stringify(s));++ assert.ok(scanned > 1000, `the scan spans several pages (${scanned} objects)`);++ assert.strictEqual(mgets.length, Math.ceil(copies / 500) + Math.ceil(scanned / 500), `one MGET per page, never one per object: ${mgets}`);++ assert.ok(mgets.every((n) => n <= 500 * demand.WINDOW_BUCKETS), 'a page reads its objects\' hour of buckets in one MGET');++ assert.ok(!sweepWarns.some((w) => /PostgreSQL counts/.test(w)), sweepWarns.join('\n'));++ assert.ok(await r2Row(hotVid.id), 'hot in Valkey: promoted');++ d = await last(hotVid.id);++ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'done']);++ assert.ok(/^75 reads in the last hour \(demand valkey, region local\) >= promoteReadsPerHour 60 \(class video\); .*1 of budget 3 \(maxPromotionsPerSweep/.test(d.reason), d.reason);++ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'valkey', metric: 'reads_1h', value: 75, threshold: 60, region: 'local', window_s: 3600 });++ assert.ok(!await r2Row(coldVid.id), 'cold in Valkey: demoted');++ d = await last(coldVid.id);++ assert.deepStrictEqual([d.action, d.outcome], ['demote', 'done']);++ assert.ok(/^2 reads in the last hour \(demand valkey, region local\) < demoteReadsPerHour 6 \(class video\)/.test(d.reason), d.reason);++ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'valkey', metric: 'reads_1h', value: 2, threshold: 6, region: 'local', window_s: 3600 });++ assert.ok(!await r2Row(warmVid.id) && await r2Row(warmR2.id), 'between the thresholds nothing moves (hysteresis)');++ assert.deepStrictEqual([(await decisions(warmVid.id)).length, (await decisions(warmR2.id)).length], [0, 0]);++ let mtext = await registry.metricsAsync();++ assert.ok(/media_sweep_demand_source_total\{source="valkey"\} 1\b/.test(mtext), mtext.split('\n').filter((l) => /demand_source/.test(l)).join('\n'));++ const pgSweeps = () => Number((/media_sweep_demand_source_total\{source="pg"\} (\d+)/.exec(mtext) || [])[1]);++ const pgBefore = pgSweeps();++++ // Valkey refuses: the whole sweep reads the daily counts, exactly today's candidates and reasons, logged once.++ const pgHot = await addObject(); await setViews(pgHot.id, [[0, 900]]);++ const pgPromote = (await tiering.promotionCandidates(policy.settings(), { limit: 500, excludeSandbox: true })).map((c) => c.id);++ const pgDemote = (await tiering.demotionCandidates(policy.settings())).map((c) => c.row.id);++ assert.ok(pgPromote.includes(pgHot.id) && pgDemote.includes(warmR2.id), 'by the daily counts pgHot is popular and warmR2 idle');++ vkMode = 'down';++ mgets.length = 0;++ console.warn = (...a) => { sweepWarns.push(a.join(' ')); };++ s = await tiering.runSweep();++ console.warn = warn0;++ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey read failed: ECONNREFUSED'], JSON.stringify(s));++ assert.strictEqual(mgets.length, 1, 'the first failed read ends the Valkey scan');++ assert.deepStrictEqual([s.candidates.promote, s.candidates.demote], [pgPromote.length, pgDemote.length], 'the same candidates as the daily counts give');++ assert.strictEqual(sweepWarns.filter((w) => /Sweep eligibility reads the daily PostgreSQL counts: Valkey read failed: ECONNREFUSED/.test(w)).length, 1, 'logged once per sweep');++ assert.ok(await r2Row(pgHot.id) && !await r2Row(warmR2.id), 'the daily counts decide: pgHot promoted, idle warmR2 demoted');++ d = await last(pgHot.id);++ assert.ok(/^900 unique viewers in 7 days >= promoteMinUniqueViewers7d 500/.test(d.reason), d.reason);++ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'pg', metric: 'unique_viewers_7d', value: 900, threshold: 500 });++ d = await last(warmR2.id);++ assert.ok(/^no viewer since the kept counts began \(idle for demoteIdleDays 14\)/.test(d.reason), d.reason);++ assert.deepStrictEqual([JSON.parse(d.inputs).demand.source, JSON.parse(d.inputs).demand.metric], ['pg', 'last_viewed_day']);++ mtext = await registry.metricsAsync();++ assert.strictEqual(pgSweeps(), pgBefore + 1, 'counted as a pg sweep');++++ // Valkey hangs: the read times out and the sweep still finishes on the daily counts.++ vkMode = 'hung';++ console.warn = () => {};++ const t0 = Date.now();++ s = await tiering.runSweep();++ console.warn = warn0;++ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey read failed: valkey read timeout'], JSON.stringify(s));++ assert.ok(Date.now() - t0 < 10000, 'a stuck Valkey never stalls the sweep');++ demand.useValkey(null);++ s = await tiering.runSweep();++ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey is not configured']);++ await db.run('DELETE FROM media_object_views_daily');++ for (const v of candViews) await db.run('INSERT INTO media_object_views_daily (object_id, day, unique_viewers, last_viewed_at) VALUES (?, ?, ?, ?)', [v.object_id, v.day, v.unique_viewers, v.last_viewed_at]);++ console.log('✅ demand from Valkey: hot promotes, cold demotes, the band holds; one MGET per page; a failed or stuck Valkey falls back to the daily counts for the whole sweep');++// ── 10. The storage sweep runs it (step 4); a restore drill never does ──const vodSweep = await storage.runSweep();assert.ok(vodSweep.objects && vodSweep.objects.gate === true && typeof vodSweep.objects.promoted === 'number', JSON.stringify(vodSweep));@@@ -121,13 +121,18 @@ const path = require('path');assert.strictEqual((await demand.hotness()).size, 0); // the hung read times outassert.ok(Date.now() - before < 5000);assert.strictEqual((await demand.rollup()).announced, 0);++ // strict (the tiering sweep): a failure throws instead of answering zeros, so the caller can fall back.++ assert.strictEqual(demand.available(), true);++ await assert.rejects(demand.hotness({ objectIds: ['obj-a'], strict: true }), /ECONNREFUSED/);await new Promise((resolve) => setImmediate(resolve));assert.deepStrictEqual(unhandled, []);-- console.log('✅ a Valkey failure never throws into a read; reads answer no demand');++ console.log('✅ a Valkey failure never throws into a read; reads answer no demand; a strict read throws');// 6. rollup(): media.object.hot through the placement outbox, at most once per object per hour.demand.useValkey(null);demand._reset();++ assert.strictEqual(demand.available(), false);++ await assert.rejects(demand.hotness({ objectIds: ['obj-a'], strict: true }), /Valkey is not configured/);const stub = http.createServer((req, res) => {req.resume();req.on('end', () => {@@@ -16,11 +16,15 @@ const path = require('path');await policy.init();assert.strictEqual(policy.get().revision(), 1);++ const bands = { image: [300, 30], download: [30, 3] };for (const name of policy.CLASSES) {++ const [promoteReadsPerHour, demoteReadsPerHour] = bands[name] || [60, 6];assert.deepStrictEqual(policy.DEFAULTS.classes[name], {-- maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0,++ maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0, promoteReadsPerHour, demoteReadsPerHour,});assert.deepStrictEqual(policy.budgetFor(name), { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10 });++ assert.deepStrictEqual(policy.bandFor(name), { promoteReadsPerHour, demoteReadsPerHour });++ assert.ok(demoteReadsPerHour < promoteReadsPerHour, `${name}: hysteresis`);}assert.strictEqual(policy.mayMove({ class: 'video', lastMovedAt: 1000, now: 1000 }), true, 'zero residency is off');assert.throws(() => policy.budgetFor('unknown'), RangeError);@@@ -46,6 +50,13 @@ const path = require('path');assert.deepStrictEqual(policy.budgetFor('image'), { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4 }, 'second revision preserves image settings');assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 });++ // The hysteresis band: demote below promote, refused as a whole revision otherwise.++ await assert.rejects(policy.set({ classes: { video: { demoteReadsPerHour: 60 } } }, { reason: 'test: an empty band' }), /demoteReadsPerHour must be below promoteReadsPerHour/);++ assert.strictEqual(policy.get().revision(), 3, 'a refused band changes nothing');++ await policy.set({ classes: { video: { promoteReadsPerHour: 120, demoteReadsPerHour: 20 } } }, { reason: 'test: a wider video band' });++ assert.deepStrictEqual(policy.bandFor('video'), { promoteReadsPerHour: 120, demoteReadsPerHour: 20 });++ assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 }, 'the band keeps the budget');++policy._reset();console.log('placement-storage-policy: all checks passed');process.exit(0);
@@ -34,7 +34,8 @@
@@ -34,7 +34,8 @@
"media.* events through the SDK outbox, webhooks during the transition",
"media.* events through the SDK outbox, webhooks during the transition",
"restore-drill mode and N-1 compatibility tests",
"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"
+ "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",
+ "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"
],
],
"updated": "2026-10-03"
"updated": "2026-10-03"
}
}
@@ -133,7 +133,19 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
@@ -133,7 +133,19 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
(`MEDIA_DEMAND_REGION`, default `local`); per-viewer regions wait for an edge-provided header. `hotness()` sums the last
(`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
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.
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.
+- **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
+ other classes 60 / 6. At or above P an object may promote, below D its R2 copy may go (still in R2 at least
+ `demoteIdleDays`), in between nothing moves. Size bounds, per-class budgets, minimum residency, holds and back-off are
+ unchanged; value per dollar ranks at the hour's rate (reads × 24 × 30 a month). The reads are batched: one `MGET` per
+ page of 500 objects (the promotion scan pages by id; the R2 copies are read in pages of 500), never a round trip per
+ object. Valkey absent, erroring or timing out (2 s) anywhere in the scan sends the **whole** sweep back to the daily
+ PostgreSQL view counts (`media.object_tier` thresholds, today's rules), with one warning per sweep and
+ `media_sweep_demand_source_total{source="valkey"|"pg"}`; both candidate lists are read before anything moves. Every
+ decision's `inputs.demand` (`{ source, metric, value, threshold[, region, window_s] }`) and its reason say which source
+ and value drove it, dry runs included; a sweep's `media.replica.requested` and `media.replica.draining` carry the same
+ `demand`, and the sweep result carries `demand_source` (and `demand_fallback`).
## 7. Delivery: sticky, measured, canaried
## 7. Delivery: sticky, measured, canaried
@@ -173,6 +185,9 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines
@@ -173,6 +185,9 @@ request/playback → telemetry → Events → {Hotness, Health, Cost} engines
since }`) through the placement outbox for objects with at least 100 reads in the last hour, at most once per object per
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,
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`.
HyperLogLog, the PostgreSQL 5-minute rollup and `media.region.hot`.
+- **Shipped (F2.5):** the placement loop reads those counters: the tiering sweep promotes and demotes on the hour's reads
+ per object with per-class hysteresis (§6), and falls back to the daily PostgreSQL counts for a whole sweep when Valkey
+ does not answer.
- **Events emitted:** `media.object.hot`, `media.replica.requested|ready|draining|evicted`, `media.variant.requested|ready`,
- **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.region.hot`, `media.delivery.surge|degraded`, `media.provider.health_degraded`, `media.provider.capacity_warning`,
`media.provider.cost_threshold`.
`media.provider.cost_threshold`.
@@ -191,7 +206,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
@@ -191,7 +206,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
moves with a cleanup job; per-class hysteresis bands; `media_storage_alerts_total`, provider latency histograms; the
moves with a cleanup job; per-class hysteresis bands; `media_storage_alerts_total`, provider latency histograms; the
corrected price table; the nginx shield (slice + cache lock) on the Media host.
corrected price table; the nginx shield (slice + cache lock) on the Media host.
- **F2 one placement engine:** every class through one sweep, value-per-dollar under budgets, dry-run + simulator,
- **F2 one placement engine:** every class through one sweep, value-per-dollar under budgets, dry-run + simulator,
- decision log with class and reason, the events above, Valkey rollups of demand.
+ decision log with class and reason, the events above, Valkey rollups of demand (F2.4, shipped) and the sweep's
+ eligibility on them with per-class hysteresis and a PostgreSQL fallback (F2.5, shipped).
- **F3 segment-native video:** CMAF/HLS recording with the growing DVR playlist, the timeline index, packing after
- **F3 segment-native video:** CMAF/HLS recording with the growing DVR playlist, the timeline index, packing after
finalize, virtual clips, sprites and captions on the timeline, MP4 fallback; Live's player moves to HLS.
finalize, virtual clips, sprites and captions on the timeline, MP4 fallback; Live's player moves to HLS.
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
- **F4 reactive derivatives:** on-demand renditions and image variants, keep-vs-regenerate economics, AV1 for viral
@@ -10,6 +10,12 @@
@@ -10,6 +10,12 @@
* The thresholds say who may move; value per dollar (../placement/value.js, R2 Standard prices from
* The thresholds say who may move; value per dollar (../placement/value.js, R2 Standard prices from
* ../placement/cost-tiers.js) says who moves first: each class spends its per-sweep budget
* ../placement/cost-tiers.js) says who moves first: each class spends its per-sweep budget
* (../placement/storage-policy.js) on its highest-scoring promotions and its lowest-scoring demotions.
* (../placement/storage-policy.js) on its highest-scoring promotions and its lowest-scoring demotions.
+ * Demand (F2.5): the sweep's eligibility reads each object's reads in the last hour for this deployment's region
+ * (../placement/demand.js hotness(), one MGET per page of objects) against its class's hysteresis band
+ * (promoteReadsPerHour / demoteReadsPerHour, media.storage_policy): at or above the first it may promote, below the
+ * second it may demote, in between nothing moves. Without Valkey, or when a read fails or times out, the whole sweep
+ * reads the daily PostgreSQL counts instead (the thresholds above), logged once and counted in
+ * media_sweep_demand_source_total{source}. Each decision's inputs.demand says which source and value drove it.
*
*
* runSweep({ trigger }) one pass: rotate the popularity counts, then promote and demote by the policy.
* runSweep({ trigger }) one pass: rotate the popularity counts, then promote and demote by the policy.
* Step 4 of the storage sweep (server/vod/vod-storage.js runSweep), which a restore
* Step 4 of the storage sweep (server/vod/vod-storage.js runSweep), which a restore
@@ -187,6 +193,7 @@ async function frame(objectId, action, ctx) {
@@ -187,6 +193,7 @@ async function frame(objectId, action, ctx) {
const held = await model().isHeld(obj.id);
const held = await model().isHeld(obj.id);
const inputs = await inputsOf(obj, { locs, pop, held, settings });
const inputs = await inputsOf(obj, { locs, pop, held, settings });
if (ctx.valuePerDollar != null) inputs.value_per_dollar = ctx.valuePerDollar;
if (ctx.valuePerDollar != null) inputs.value_per_dollar = ctx.valuePerDollar;
+ if (ctx.demand) inputs.demand = ctx.demand;
const canon = canonicalLocation(obj, locs);
const canon = canonicalLocation(obj, locs);
const trigger = ctx.trigger || 'manual';
const trigger = ctx.trigger || 'manual';
const why = ctx.reason || 'requested';
const why = ctx.reason || 'requested';
@@ -232,7 +239,7 @@ async function promote(objectId, ctx = {}) {
@@ -232,7 +239,7 @@ async function promote(objectId, ctx = {}) {
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
await events().recordPlacement('media.replica.requested', obj.app_id, subjectOf(obj), {
await events().recordPlacement('media.replica.requested', obj.app_id, subjectOf(obj), {
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'promote',
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'promote',
- from: canon ? canon.provider : null, to: 'r2', provider: 'r2',
+ from: canon ? canon.provider : null, to: 'r2', provider: 'r2', ...(ctx.demand ? { demand: ctx.demand } : {}),
});
});
});
});
events().kick();
events().kick();
@@ -299,7 +306,7 @@ async function demote(objectId, ctx = {}) {
@@ -299,7 +306,7 @@ async function demote(objectId, ctx = {}) {
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {
await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
- provider: 'r2', key: r2.key,
+ provider: 'r2', key: r2.key, ...(ctx.demand ? { demand: ctx.demand } : {}),
});
});
});
});
events().kick();
events().kick();
@@ -326,8 +333,8 @@ async function demote(objectId, ctx = {}) {
@@ -326,8 +333,8 @@ async function demote(objectId, ctx = {}) {
// ── Candidates ───────────────────────────────────────────────
// ── Candidates ───────────────────────────────────────────────
/**
/**
- * Value per dollar of an R2 copy of this row (../placement/value.js): its 7-day unique viewers scaled to a month
- * of requests, against R2 Standard's storage, read and retrieval prices (media.cost_tiers) and, for a promotion,
+ * Value per dollar of an R2 copy of this row (../placement/value.js): its reads in the last hour (Valkey demand)
+ * or else its 7-day unique viewers, scaled to a month of requests, against R2 Standard's storage, read and retrieval prices (media.cost_tiers) and, for a promotion,
* the class's minimum residency beyond that month (media.storage_policy), which the copy commits to. Free
* the class's minimum residency beyond that month (media.storage_policy), which the copy commits to. Free
* allowances are not counted: a promotion is priced at the margin.
* allowances are not counted: a promotion is priced at the margin.
*/
*/
@@ -338,7 +345,8 @@ function scoreOf(row, cls, { committing = false } = {}) {
@@ -338,7 +345,8 @@ function scoreOf(row, cls, { committing = false } = {}) {
return value.score({
return value.score({
// A row that carries a month of counts (the demotion scan) is read as is: an idle copy has no viewer in the
// A row that carries a month of counts (the demotion scan) is read as is: an idle copy has no viewer in the
// 7-day window, so only the longer count tells a copy last viewed 20 days ago from one never viewed.
// 7-day window, so only the longer count tells a copy last viewed 20 days ago from one never viewed.
- requests: row.unique_viewers_30d != null ? Number(row.unique_viewers_30d) || 0
+ requests: row.reads_1h != null ? (Number(row.reads_1h) || 0) * 24 * MONTH_DAYS
+ : row.unique_viewers_30d != null ? Number(row.unique_viewers_30d) || 0
: (Number(row.unique_viewers_7d) || 0) * MONTH_DAYS / popularity.WINDOW_DAYS,
: (Number(row.unique_viewers_7d) || 0) * MONTH_DAYS / popularity.WINDOW_DAYS,
bytes: Number(row.size_bytes) || 0,
bytes: Number(row.size_bytes) || 0,
storageCost: storage,
storageCost: storage,
@@ -360,8 +368,10 @@ const SWEEP_KEEP = 500; // top-ranked promotion candidates a sweep
@@ -360,8 +368,10 @@ const SWEEP_KEEP = 500; // top-ranked promotion candidates a sweep
* before the limit applies (a most-viewed prefix would drop smaller objects that are worth more per dollar), but
* before the limit applies (a most-viewed prefix would drop smaller objects that are worth more per dollar), but
* the scan reads SCAN_PAGE rows at a time and holds only the best `limit`, so memory stays bounded.
* the scan reads SCAN_PAGE rows at a time and holds only the best `limit`, so memory stays bounded.
* excludeSandbox drops developer-sandbox tenants in the query (their objects are never tiered).
* excludeSandbox drops developer-sandbox tenants in the query (their objects are never tiered).
+ * With `demand` (the sweep's Valkey reader), the hourly reads replace the daily counts: see hotPromotionCandidates.
*/
*/
-async function promotionCandidates(settings, { now = Date.now(), appId = null, limit = 200, excludeSandbox = false } = {}) {
+async function promotionCandidates(settings, { now = Date.now(), appId = null, limit = 200, excludeSandbox = false, demand = null } = {}) {
+ if (demand) return await hotPromotionCandidates(settings, demand, { appId, limit, excludeSandbox });
const today = popularity.dayOf(now);
const today = popularity.dayOf(now);
const args = [popularity.windowStart(today, popularity.WINDOW_DAYS)];
const args = [popularity.windowStart(today, popularity.WINDOW_DAYS)];
const rank = (a, b) => b.value_per_dollar - a.value_per_dollar || Number(b.unique_viewers_7d) - Number(a.unique_viewers_7d) || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0);
const rank = (a, b) => b.value_per_dollar - a.value_per_dollar || Number(b.unique_viewers_7d) - Number(a.unique_viewers_7d) || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0);
@@ -381,6 +391,7 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, l
@@ -381,6 +391,7 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, l
for (const r of rows) {
for (const r of rows) {
r.class = storagePolicy.classOf(r);
r.class = storagePolicy.classOf(r);
r.value_per_dollar = scoreOf(r, r.class, { committing: true });
r.value_per_dollar = scoreOf(r, r.class, { committing: true });
+ r.demand = { source: 'pg', metric: 'unique_viewers_7d', value: Number(r.unique_viewers_7d) || 0, threshold: settings.promoteMinUniqueViewers7d };
}
}
best = best.concat(rows).sort(rank).slice(0, limit);
best = best.concat(rows).sort(rank).slice(0, limit);
if (rows.length < SCAN_PAGE) return best;
if (rows.length < SCAN_PAGE) return best;
@@ -388,12 +399,51 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, l
@@ -388,12 +399,51 @@ async function promotionCandidates(settings, { now = Date.now(), appId = null, l
}
}
}
}
+/** The demand fields a decision records for a Valkey reading: source, the hourly reads and the class's threshold. */
+function hotDemand(demand, reads, threshold) {
+ return { source: 'valkey', metric: 'reads_1h', value: reads, threshold, region: demand.region, window_s: demand.window_s };
+}
+
+/**
+ * The Valkey form of promotionCandidates: ready native objects in the size bounds with no R2 copy, paged by id
+ * (SCAN_PAGE rows, one hotness MGET per page), kept when their reads in the last hour reach their class's
+ * promoteReadsPerHour; ranked by value per dollar at that rate, then reads. A failed read throws (the sweep falls back).
+ */
+async function hotPromotionCandidates(settings, demand, { appId = null, limit = 200, excludeSandbox = false } = {}) {
+ const rank = (a, b) => b.value_per_dollar - a.value_per_dollar || b.reads_1h - a.reads_1h || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0);
+ let best = [];
+ for (let after = ''; ;) {
+ const rows = await db.all(`SELECT o.* FROM media_objects o
+ WHERE o.id > ? AND o.legacy_ref IS NULL AND o.lifecycle_status = 'ready' AND o.size_bytes >= ? AND o.size_bytes <= ?
+ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.provider = 'r2' AND l.state = 'present')${appId ? ' AND o.app_id = ?' : ''}${excludeSandbox
+ ? " AND NOT EXISTS (SELECT 1 FROM apps a WHERE a.app_id = o.app_id AND a.env = 'sandbox')" : ''}
+ ORDER BY o.id LIMIT ?`,
+ [after, Math.ceil(settings.promoteMinSizeMb * MB), Math.floor(settings.promoteMaxSizeMb * MB), ...(appId ? [appId] : []), SCAN_PAGE]);
+ const heat = rows.length ? await demand.read(rows.map((r) => r.id)) : new Map();
+ const hot = [];
+ for (const r of rows) {
+ r.class = storagePolicy.classOf(r);
+ r.reads_1h = heat.get(String(r.id)) || 0;
+ const { promoteReadsPerHour } = storagePolicy.bandFor(r.class);
+ if (r.reads_1h < promoteReadsPerHour) continue;
+ r.value_per_dollar = scoreOf(r, r.class, { committing: true });
+ r.demand = hotDemand(demand, r.reads_1h, promoteReadsPerHour);
+ hot.push(r);
+ }
+ best = best.concat(hot).sort(rank).slice(0, limit);
+ if (rows.length < SCAN_PAGE) return best;
+ after = rows[rows.length - 1].id;
+ }
+}
+
/**
/**
* Native objects with an R2 copy that should lose it: not ready any more (first, whatever their score), or idle,
* Native objects with an R2 copy that should lose it: not ready any more (first, whatever their score), or idle,
* lowest value per dollar of keeping the copy first (its viewers over the kept month, since an idle copy has none
* lowest value per dollar of keeping the copy first (its viewers over the kept month, since an idle copy has none
- * in the last 7 days), then the largest. → [{ row, reason, class, value_per_dollar }]
+ * in the last 7 days), then the largest. With `demand` (the sweep's Valkey reader, one MGET per SCAN_PAGE copies), a
+ * ready object is cold when its reads in the last hour fall below its class's demoteReadsPerHour instead of idle by
+ * the daily counts (still in R2 at least demoteIdleDays either way). → [{ row, reason, class, value_per_dollar, demand }]
*/
*/
-async function demotionCandidates(settings, { now = Date.now(), appId = null } = {}) {
+async function demotionCandidates(settings, { now = Date.now(), appId = null, demand = null } = {}) {
const idleFrom = popularity.windowStart(popularity.dayOf(now), settings.demoteIdleDays);
const idleFrom = popularity.windowStart(popularity.dayOf(now), settings.demoteIdleDays);
const monthFrom = popularity.windowStart(popularity.dayOf(now), popularity.KEEP_DAYS);
const monthFrom = popularity.windowStart(popularity.dayOf(now), popularity.KEEP_DAYS);
const rows = await db.all(`SELECT o.*, l.created_at AS r2_since,
const rows = await db.all(`SELECT o.*, l.created_at AS r2_since,
@@ -402,15 +452,29 @@ async function demotionCandidates(settings, { now = Date.now(), appId = null } =
@@ -402,15 +452,29 @@ async function demotionCandidates(settings, { now = Date.now(), appId = null } =
FROM media_locations l JOIN media_objects o ON o.id = l.object_id
FROM media_locations l JOIN media_objects o ON o.id = l.object_id
WHERE l.provider = 'r2' AND o.legacy_ref IS NULL${appId ? ' AND o.app_id = ?' : ''}
WHERE l.provider = 'r2' AND o.legacy_ref IS NULL${appId ? ' AND o.app_id = ?' : ''}
ORDER BY (o.lifecycle_status = 'ready'), o.id`, appId ? [monthFrom, appId] : [monthFrom]);
ORDER BY (o.lifecycle_status = 'ready'), o.id`, appId ? [monthFrom, appId] : [monthFrom]);
+ if (demand) {
+ for (let i = 0; i < rows.length; i += SCAN_PAGE) {
+ const page = rows.slice(i, i + SCAN_PAGE);
+ const heat = await demand.read(page.map((r) => r.id));
+ for (const r of page) r.reads_1h = heat.get(String(r.id)) || 0;
+ }
+ }
const out = [];
const out = [];
for (const r of rows) {
for (const r of rows) {
const cls = storagePolicy.classOf(r);
const cls = storagePolicy.classOf(r);
- const scored = { row: r, class: cls, value_per_dollar: scoreOf(r, cls) };
+ const { demoteReadsPerHour } = storagePolicy.bandFor(cls);
+ const scored = { row: r, class: cls, value_per_dollar: scoreOf(r, cls),
+ demand: demand ? hotDemand(demand, r.reads_1h, demoteReadsPerHour) : { source: 'pg', metric: 'last_viewed_day', value: r.last_viewed_day || null, threshold: idleFrom } };
if (r.lifecycle_status !== 'ready') { out.push({ ...scored, reason: `the object is ${r.lifecycle_status}` }); continue; }
if (r.lifecycle_status !== 'ready') { out.push({ ...scored, reason: `the object is ${r.lifecycle_status}` }); continue; }
- const idle = !r.last_viewed_day || r.last_viewed_day < idleFrom;
const since = msOf(r.r2_since);
const since = msOf(r.r2_since);
const settled = !Number.isFinite(since) || now - since >= settings.demoteIdleDays * DAY_MS;
const settled = !Number.isFinite(since) || now - since >= settings.demoteIdleDays * DAY_MS;
- if (idle && settled) {
+ if (!settled) continue;
+ if (demand) {
+ if (r.reads_1h < demoteReadsPerHour) {
+ out.push({ ...scored, reason: `${r.reads_1h} reads in the last hour (demand valkey, region ${demand.region}) < demoteReadsPerHour ${demoteReadsPerHour} (class ${cls}); `
+ + `in R2 since ${r.r2_since || 'unknown'} (at least demoteIdleDays ${settings.demoteIdleDays})` });
+ }
+ } else if (!r.last_viewed_day || r.last_viewed_day < idleFrom) {
out.push({ ...scored, reason: `no viewer since ${r.last_viewed_day || 'the kept counts began'} (idle for demoteIdleDays ${settings.demoteIdleDays}); in R2 since ${r.r2_since || 'unknown'}` });
out.push({ ...scored, reason: `no viewer since ${r.last_viewed_day || 'the kept counts began'} (idle for demoteIdleDays ${settings.demoteIdleDays}); in R2 since ${r.r2_since || 'unknown'}` });
}
}
}
}
@@ -426,6 +490,11 @@ function rankNote(cls, score, slot, budgetKey) {
@@ -426,6 +490,11 @@ function rankNote(cls, score, slot, budgetKey) {
}
}
function promoteReason(c, s, slot) {
function promoteReason(c, s, slot) {
+ if (c.demand && c.demand.source === 'valkey') {
+ return `${c.reads_1h} reads in the last hour (demand valkey, region ${c.demand.region}) >= promoteReadsPerHour ${c.demand.threshold} (class ${c.class}); `
+ + `${(Number(c.size_bytes) / MB).toFixed(1)} MB (promoteMinSizeMb ${s.promoteMinSizeMb} to promoteMaxSizeMb ${s.promoteMaxSizeMb}); `
+ + rankNote(c.class, c.value_per_dollar, slot, 'maxPromotionsPerSweep');
+ }
return `${c.unique_viewers_7d} unique viewers in 7 days >= promoteMinUniqueViewers7d ${s.promoteMinUniqueViewers7d}; last viewed ${c.last_viewed_day} `
return `${c.unique_viewers_7d} unique viewers in 7 days >= promoteMinUniqueViewers7d ${s.promoteMinUniqueViewers7d}; last viewed ${c.last_viewed_day} `
+ `(within promoteRecentAccessDays ${s.promoteRecentAccessDays}); ${(Number(c.size_bytes) / MB).toFixed(1)} MB (promoteMinSizeMb ${s.promoteMinSizeMb} to promoteMaxSizeMb ${s.promoteMaxSizeMb}); `
+ `(within promoteRecentAccessDays ${s.promoteRecentAccessDays}); ${(Number(c.size_bytes) / MB).toFixed(1)} MB (promoteMinSizeMb ${s.promoteMinSizeMb} to promoteMaxSizeMb ${s.promoteMaxSizeMb}); `
+ rankNote(c.class, c.value_per_dollar, slot, 'maxPromotionsPerSweep');
+ rankNote(c.class, c.value_per_dollar, slot, 'maxPromotionsPerSweep');
@@ -465,8 +534,36 @@ async function residencyRefusal(obj, cls, action, now) {
@@ -465,8 +534,36 @@ async function residencyRefusal(obj, cls, action, now) {
return `minimum residency of ${minResidencyMs} ms for class ${cls} (media.storage_policy) not reached since the last move at ${lastAt}; ${action} suppressed`;
return `minimum residency of ${minResidencyMs} ms for class ${cls} (media.storage_policy) not reached since the last move at ${lastAt}; ${action} suppressed`;
}
}
+/** The sweep's Valkey demand reader, or null when Valkey is not configured. read(ids) throws on any failure. */
+function valkeyDemand(now) {
+ const demand = require('../placement/demand');
+ if (!demand.available()) return null;
+ return { region: demand.region(), window_s: (demand.WINDOW_BUCKETS * demand.BUCKET_MS) / 1000,
+ read: (ids) => demand.hotness({ objectIds: ids, now, strict: true }) };
+}
+
/**
/**
- * One pass over native objects. → { gate, promoted, demoted, would_promote, would_demote, already, refused, failed,
+ * Both candidate lists, from one demand source for the whole sweep: Valkey's hourly reads, or (Valkey absent, or a
+ * read failed or timed out anywhere in the scan) the daily PostgreSQL counts. Read before anything moves, so a
+ * failure part-way never leaves a sweep acting on two signals. → { source, fallback, demote, promote }
+ */
+async function sweepCandidates(settings, now) {
+ const heat = valkeyDemand(now);
+ let fallback = heat ? null : 'Valkey is not configured';
+ if (heat) {
+ try {
+ return { source: 'valkey', fallback, demote: await demotionCandidates(settings, { now, demand: heat }),
+ promote: await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true, demand: heat }) };
+ } catch (err) {
+ fallback = `Valkey read failed: ${err.message}`;
+ }
+ }
+ return { source: 'pg', fallback, demote: await demotionCandidates(settings, { now }),
+ promote: await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true }) };
+}
+
+/**
+ * One pass over native objects. → { gate, demand_source, demand_fallback?, promoted, demoted, would_promote, would_demote, already, refused, failed,
* repeats, skipped_backoff, candidates: { promote, demote }, rotated, errors }
* repeats, skipped_backoff, candidates: { promote, demote }, rotated, errors }
*/
*/
async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -478,27 +575,35 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -478,27 +575,35 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
const out = { gate: !!settings.active, promoted: 0, demoted: 0, would_promote: 0, would_demote: 0, already: 0, refused: 0, failed: 0,
const out = { gate: !!settings.active, promoted: 0, demoted: 0, would_promote: 0, would_demote: 0, already: 0, refused: 0, failed: 0,
repeats: 0, skipped_backoff: 0, candidates: { promote: 0, demote: 0 }, rotated: await popularity.rotate({ now }), errors: [] };
repeats: 0, skipped_backoff: 0, candidates: { promote: 0, demote: 0 }, rotated: await popularity.rotate({ now }), errors: [] };
+ const picked = await sweepCandidates(settings, now);
+ out.demand_source = picked.source;
+ if (picked.fallback) {
+ out.demand_fallback = picked.fallback;
+ console.warn(`[ObjectTiers] Sweep eligibility reads the daily PostgreSQL counts: ${picked.fallback}`);
+ }
+ require('../placement/metrics-binding').inc('media_sweep_demand_source_total', { source: picked.source });
+
// Demotions first: they free the R2 cache, and a deleted object's copy should not wait behind promotions.
// Demotions first: they free the R2 cache, and a deleted object's copy should not wait behind promotions.
- const demote_ = await demotionCandidates(settings, { now });
+ const demote_ = picked.demote;
out.candidates.demote = demote_.length;
out.candidates.demote = demote_.length;
// Budgets and residency are per class (media.storage_policy); a class spends only its own budget, in the
// Budgets and residency are per class (media.storage_policy); a class spends only its own budget, in the
// candidates' value-per-dollar order. A refusal (residency, hold) or a back-off spends nothing, so the next
// candidates' value-per-dollar order. A refusal (residency, hold) or a back-off spends nothing, so the next
// in line takes the slot.
// in line takes the slot.
const spent = { promote: {}, demote: {} };
const spent = { promote: {}, demote: {} };
const room = (action, cls) => (spent[action][cls] || 0) < storagePolicy.budgetFor(cls)[action === 'promote' ? 'maxPromotionsPerSweep' : 'maxDemotionsPerSweep'];
const room = (action, cls) => (spent[action][cls] || 0) < storagePolicy.budgetFor(cls)[action === 'promote' ? 'maxPromotionsPerSweep' : 'maxDemotionsPerSweep'];
- for (const { row, reason, class: cls, value_per_dollar: score } of demote_) {
+ for (const { row, reason, class: cls, value_per_dollar: score, demand } of demote_) {
if (!room('demote', cls)) continue;
if (!room('demote', cls)) continue;
if (await inBackoff(row.id, 'demote')) { out.skipped_backoff++; continue; }
if (await inBackoff(row.id, 'demote')) { out.skipped_backoff++; continue; }
// A copy of an object that is no longer ready is removed regardless of residency.
// A copy of an object that is no longer ready is removed regardless of residency.
const residency = row.lifecycle_status === 'ready' ? await residencyRefusal(row, cls, 'demote', now) : null;
const residency = row.lifecycle_status === 'ready' ? await residencyRefusal(row, cls, 'demote', now) : null;
- const r = await demote(row.id, { trigger, settings, now, residency, valuePerDollar: score,
+ const r = await demote(row.id, { trigger, settings, now, residency, valuePerDollar: score, demand,
reason: `${reason}; ${rankNote(cls, score, (spent.demote[cls] || 0) + 1, 'maxDemotionsPerSweep')}` });
reason: `${reason}; ${rankNote(cls, score, (spent.demote[cls] || 0) + 1, 'maxDemotionsPerSweep')}` });
tally(out, { ...r, action: 'demote' });
tally(out, { ...r, action: 'demote' });
if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.demote[cls] = (spent.demote[cls] || 0) + 1;
if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.demote[cls] = (spent.demote[cls] || 0) + 1;
}
}
// Every eligible candidate is ranked (not a most-viewed prefix); the best SWEEP_KEEP are kept, so no class is crowded out of its budget.
// Every eligible candidate is ranked (not a most-viewed prefix); the best SWEEP_KEEP are kept, so no class is crowded out of its budget.
- const promote_ = await promotionCandidates(settings, { now, limit: SWEEP_KEEP, excludeSandbox: true });
+ const promote_ = picked.promote;
out.candidates.promote = promote_.length;
out.candidates.promote = promote_.length;
for (const c of promote_) {
for (const c of promote_) {
const cls = c.class;
const cls = c.class;
@@ -506,13 +611,12 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -506,13 +611,12 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }
if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }
const residency = await residencyRefusal(c, cls, 'promote', now);
const residency = await residencyRefusal(c, cls, 'promote', now);
const r = await promote(c.id, { trigger, reason: promoteReason(c, settings, (spent.promote[cls] || 0) + 1), settings, now, residency, valuePerDollar: c.value_per_dollar,
const r = await promote(c.id, { trigger, reason: promoteReason(c, settings, (spent.promote[cls] || 0) + 1), settings, now, residency, valuePerDollar: c.value_per_dollar,
- pop: { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } });
+ demand: c.demand, pop: picked.source === 'pg' ? { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } : undefined });
tally(out, { ...r, action: 'promote' });
tally(out, { ...r, action: 'promote' });
if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.promote[cls] = (spent.promote[cls] || 0) + 1;
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.
+ // Demand (F2.4): stage media.object.hot for objects over the hourly read threshold.
out.demand = await require('../placement/demand').rollup({ now });
out.demand = await require('../placement/demand').rollup({ now });
if (out.promoted || out.demoted || out.failed) {
if (out.promoted || out.demoted || out.failed) {
@@ -12,8 +12,8 @@
@@ -12,8 +12,8 @@
* <bucket> = floor(epoch_ms / BUCKET_MS). The region is one per deployment (MEDIA_DEMAND_REGION, default 'local');
* <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.
*
*
- * 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.
+ * 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.
*/
*/
const config = require('../config');
const config = require('../config');
@@ -32,6 +32,8 @@ const sets = new Map(); // fallback: hot key -> { members: Map(objec
@@ -32,6 +32,8 @@ const sets = new Map(); // fallback: hot key -> { members: Map(objec
const announced = new Map(); // fallback: announce key -> expiresAt
const announced = new Map(); // fallback: announce key -> expiresAt
function useValkey(handle) { vk = handle && handle.client ? handle : null; }
function useValkey(handle) { vk = handle && handle.client ? handle : null; }
+/** Valkey is configured: counts are shared across processes (the in-process fallback only sees this one). */
+function available() { return !!vk; }
function region() { return String((config.demand && config.demand.region) || 'local').replace(/[^A-Za-z0-9_.-]/g, '_') || 'local'; }
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); }
function bucketOf(now = Date.now()) { return Math.floor(now / BUCKET_MS); }
const name = (...parts) => (vk ? vk.key(...parts) : parts.join(':'));
const name = (...parts) => (vk ? vk.key(...parts) : parts.join(':'));
@@ -80,9 +82,11 @@ function record({ objectId, region: r = region(), now = Date.now() } = {}) {
@@ -80,9 +82,11 @@ function record({ objectId, region: r = region(), now = Date.now() } = {}) {
/**
/**
* Reads per object over the last `buckets` buckets (summed), newest bucket included. With `objectIds`, the counters
* 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.
* 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)
+ * 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 } = {}) {
+async function hotness({ region: r = region(), now = Date.now(), buckets = WINDOW_BUCKETS, objectIds = null, strict = false } = {}) {
+ if (strict && !vk) throw new Error('Valkey is not configured');
const out = new Map();
const out = new Map();
const window = windowOf(now, buckets);
const window = windowOf(now, buckets);
const ids = objectIds ? [...new Set(objectIds.filter((x) => x != null).map(String))] : null;
const ids = objectIds ? [...new Set(objectIds.filter((x) => x != null).map(String))] : null;
@@ -121,6 +125,7 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO
@@ -121,6 +125,7 @@ async function hotness({ region: r = region(), now = Date.now(), buckets = WINDO
}
}
return out;
return out;
} catch (err) {
} catch (err) {
+ if (strict) throw err;
console.warn('[Demand] hotness not read:', err.message);
console.warn('[Demand] hotness not read:', err.message);
return ids ? new Map(ids.map((id) => [id, 0])) : new Map();
return ids ? new Map(ids.map((id) => [id, 0])) : new Map();
}
}
@@ -200,5 +205,5 @@ function _reset() { counters.clear(); sets.clear(); announced.clear(); }
@@ -200,5 +205,5 @@ function _reset() { counters.clear(); sets.clear(); announced.clear(); }
module.exports = {
module.exports = {
BUCKET_MS, WINDOW_BUCKETS, KEY_TTL_S, TOP_N, HOT_READS, ANNOUNCE_S,
BUCKET_MS, WINDOW_BUCKETS, KEY_TTL_S, TOP_N, HOT_READS, ANNOUNCE_S,
- useValkey, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,
+ useValkey, available, region, bucketOf, counterKey, hotKey, announceKey, record, hotness, top, rollup, _reset,
};
};
@@ -15,6 +15,7 @@
@@ -15,6 +15,7 @@
* media_router_decisions_total{provider,purpose,reason} counter
* media_router_decisions_total{provider,purpose,reason} counter
* media_presign_cache_total{result} counter
* media_presign_cache_total{result} counter
* media_storage_alerts_total{kind} counter
* media_storage_alerts_total{kind} counter
+ * media_sweep_demand_source_total{source} counter (valkey | pg: the object sweep's demand signal)
*/
*/
'use strict';
'use strict';
@@ -28,6 +29,7 @@ let breakerGauge = null; // gauge (collect)
@@ -28,6 +29,7 @@ let breakerGauge = null; // gauge (collect)
let decisions = null; // counter
let decisions = null; // counter
let presign = null; // counter
let presign = null; // counter
let alerts = null; // counter
let alerts = null; // counter
+let demandSource = null; // counter
function bind(registry) {
function bind(registry) {
if (latency) return; // idempotent
if (latency) return; // idempotent
@@ -69,6 +71,13 @@ function bind(registry) {
@@ -69,6 +71,13 @@ function bind(registry) {
labelNames: ['kind'],
labelNames: ['kind'],
});
});
+ demandSource = registry.counter({
+ name: 'media_sweep_demand_source_total',
+ help: 'Object tiering sweeps by the demand signal their eligibility read: valkey (hourly reads) or pg (daily view counts, the fallback)',
+ labelNames: ['source'],
+ });
+ for (const source of ['valkey', 'pg']) demandSource.inc({ source }, 0);
+
// Bind every hook the placement modules use.
// Bind every hook the placement modules use.
router.bindMetric((name, labels) => inc(name, labels));
router.bindMetric((name, labels) => inc(name, labels));
presignCache.bindMetric((name, labels) => inc(name, labels));
presignCache.bindMetric((name, labels) => inc(name, labels));
@@ -80,6 +89,7 @@ function inc(name, labels) {
@@ -80,6 +89,7 @@ function inc(name, labels) {
else if (name === 'media_presign_cache_total') presign && presign.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_storage_alerts_total') alerts && alerts.inc(labels || {}, 1);
else if (name === 'media_provider_errors_total') errors && errors.inc(labels || {}, 1);
else if (name === 'media_provider_errors_total') errors && errors.inc(labels || {}, 1);
+ else if (name === 'media_sweep_demand_source_total') demandSource && demandSource.inc(labels || {}, 1);
}
}
/** Hook the signals module's recordSuccess/recordFailure to fill the histogram/counter. */
/** Hook the signals module's recordSuccess/recordFailure to fill the histogram/counter. */
@@ -87,4 +97,4 @@ function observeLatency(provider, op, ms) { if (latency && ms > 0) latency.obser
@@ -87,4 +97,4 @@ function observeLatency(provider, op, ms) { if (latency && ms > 0) latency.obser
function observeError(provider, op) { errors && errors.inc({ provider, op }, 1); }
function observeError(provider, op) { errors && errors.inc({ provider, op }, 1); }
function alert(kind) { alerts && alerts.inc({ kind }, 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
\ No newline at end of file
+module.exports = { bind, inc, observeLatency, observeError, alert, _reset: () => { latency = errors = breakerGauge = decisions = presign = alerts = demandSource = null; } };
\ No newline at end of file
\ No newline at end of file
@@ -10,8 +10,16 @@ const db = require('../db/database');
@@ -10,8 +10,16 @@ const db = require('../db/database');
// The defaults match media.object_tier's per-sweep limits (server/objects/tier-policy.js).
// The defaults match media.object_tier's per-sweep limits (server/objects/tier-policy.js).
const BASE = { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0 };
const BASE = { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0 };
const CLASSES = ['video', 'image', 'download', 'game-asset', 'attachment', 'backup'];
const CLASSES = ['video', 'image', 'download', 'game-asset', 'attachment', 'backup'];
+// The sweep's hysteresis band (F2.5): reads in the last hour for the deployment's region (placement/demand.js
+// hotness()) at or above promoteReadsPerHour promote, below demoteReadsPerHour demote, in between nothing moves.
+// Images are small and cheap to serve locally, so they need more reads; downloads are large, so fewer.
+const HYSTERESIS = { promoteReadsPerHour: 60, demoteReadsPerHour: 6 };
+const CLASS_HYSTERESIS = {
+ image: { promoteReadsPerHour: 300, demoteReadsPerHour: 30 },
+ download: { promoteReadsPerHour: 30, demoteReadsPerHour: 3 },
+};
const DEFAULTS = {
const DEFAULTS = {
- classes: Object.fromEntries(CLASSES.map((name) => [name, { ...BASE }])),
+ classes: Object.fromEntries(CLASSES.map((name) => [name, { ...BASE, ...HYSTERESIS, ...CLASS_HYSTERESIS[name] }])),
};
};
const CLASS_SCHEMA = {
const CLASS_SCHEMA = {
@@ -21,6 +29,8 @@ const CLASS_SCHEMA = {
@@ -21,6 +29,8 @@ const CLASS_SCHEMA = {
maxPromotionsPerSweep: { type: 'integer', minimum: 0, maximum: 100 },
maxPromotionsPerSweep: { type: 'integer', minimum: 0, maximum: 100 },
maxDemotionsPerSweep: { type: 'integer', minimum: 0, maximum: 500 },
maxDemotionsPerSweep: { type: 'integer', minimum: 0, maximum: 500 },
minResidencyMs: { type: 'integer', minimum: 0 },
minResidencyMs: { type: 'integer', minimum: 0 },
+ promoteReadsPerHour: { type: 'integer', minimum: 1, maximum: 1e9 },
+ demoteReadsPerHour: { type: 'integer', minimum: 0, maximum: 1e9 },
},
},
};
};
const SCHEMA = {
const SCHEMA = {
@@ -35,13 +45,23 @@ const SCHEMA = {
@@ -35,13 +45,23 @@ const SCHEMA = {
},
},
};
};
+/** Each class's band is a band: demoteReadsPerHour below promoteReadsPerHour (an object never both promotes and demotes). */
+function validate(v) {
+ const errors = [];
+ for (const name of CLASSES) {
+ const c = { ...DEFAULTS.classes[name], ...v.classes?.[name] };
+ if (c.demoteReadsPerHour >= c.promoteReadsPerHour) errors.push(`classes.${name}: demoteReadsPerHour must be below promoteReadsPerHour`);
+ }
+ return errors.length ? errors : true;
+}
+
let store = null;
let store = null;
async function init() {
async function init() {
if (store) return store;
if (store) return store;
store = await config.createConfigStore({
store = await config.createConfigStore({
db: db.getDb(), service: 'media', namespace: 'media.storage_policy',
db: db.getDb(), service: 'media', namespace: 'media.storage_policy',
- defaults: DEFAULTS, schema: SCHEMA,
+ defaults: DEFAULTS, schema: SCHEMA, validate,
legacy: () => ({}),
legacy: () => ({}),
onActivate: async () => {},
onActivate: async () => {},
log: { info: (m) => console.log(`[StoragePolicy] ${m}`), warn: (m) => console.warn(`[StoragePolicy] ${m}`), error: (m) => console.error(`[StoragePolicy] ${m}`) },
log: { info: (m) => console.log(`[StoragePolicy] ${m}`), warn: (m) => console.warn(`[StoragePolicy] ${m}`), error: (m) => console.error(`[StoragePolicy] ${m}`) },
@@ -72,7 +92,7 @@ function settings() {
@@ -72,7 +92,7 @@ function settings() {
let configured;
let configured;
try { configured = get().get(); } catch { configured = {}; }
try { configured = get().get(); } catch { configured = {}; }
const classes = {};
const classes = {};
- for (const name of CLASSES) classes[name] = { ...BASE, ...configured.classes?.[name] };
+ for (const name of CLASSES) classes[name] = { ...DEFAULTS.classes[name], ...configured.classes?.[name] };
return { classes };
return { classes };
}
}
@@ -97,6 +117,12 @@ function budgetFor(name) {
@@ -97,6 +117,12 @@ function budgetFor(name) {
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
}
}
+/** The class's hourly promote and demote thresholds (hysteresis: demote < promote). */
+function bandFor(name) {
+ const { promoteReadsPerHour, demoteReadsPerHour } = classPolicy(name);
+ return { promoteReadsPerHour, demoteReadsPerHour };
+}
+
function timestamp(value) {
function timestamp(value) {
if (value instanceof Date) return value.getTime();
if (value instanceof Date) return value.getTime();
if (typeof value === 'number') return value;
if (typeof value === 'number') return value;
@@ -113,4 +139,4 @@ function mayMove({ class: name, lastMovedAt, now = Date.now() }) {
@@ -113,4 +139,4 @@ function mayMove({ class: name, lastMovedAt, now = Date.now() }) {
return Number.isFinite(last) && Number.isFinite(current) && current - last >= minResidencyMs;
return Number.isFinite(last) && Number.isFinite(current) && current - last >= minResidencyMs;
}
}
-module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, mayMove, _reset: () => { store = null; } };
+module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, bandFor, mayMove, _reset: () => { store = null; } };
@@ -541,6 +541,105 @@ const crypto = require('crypto');
@@ -541,6 +541,105 @@ const crypto = require('crypto');
await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);
await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);
console.log('✅ gate off: residency- or hold-blocked moves are logged as refused and spend no slot');
console.log('✅ gate off: residency- or hold-blocked moves are logged as refused and spend no slot');
+ // ── 9e. Demand from Valkey (F2.5): hourly reads against each class's band; the daily counts when Valkey fails ──
+ const demand = require('../server/placement/demand');
+ const sweepWarns = [];
+ const warn0 = console.warn;
+ const vkReads = new Map();
+ const mgets = [];
+ let vkMode = 'up';
+ demand.useValkey({ key: (...p) => 'ov:media:' + p.join(':'), client: {
+ async mget(...keys) {
+ mgets.push(keys.length);
+ if (vkMode === 'down') throw new Error('ECONNREFUSED');
+ if (vkMode === 'hung') return await new Promise(() => {});
+ return keys.map((k) => (vkReads.has(k) ? String(vkReads.get(k)) : null));
+ },
+ async zrange() { return []; },
+ async set() { return 'OK'; },
+ async del() { return 1; },
+ } });
+ const setReads = (id, n) => vkReads.set(demand.counterKey(demand.region(), demand.bucketOf(Date.now()), id), n);
+ const inR2 = async (o, days) => {
+ buckets['r2-bucket'].set(`objects/live/${o.id}`, o.bytes);
+ await model.upsertLocation(o.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${o.id}`, storage_class: 'cache', state: 'present', size_bytes: o.bytes.length, checksum: o.sum, verified: true });
+ await ageR2(o.id, days);
+ };
+ const candViews = await db.all('SELECT object_id, day, unique_viewers, last_viewed_at FROM media_object_views_daily WHERE object_id = ?', [cand.id]);
+ await db.run('DELETE FROM media_object_views_daily'); // no daily counts: only Valkey can move anything
+ const hotVid = await addObject({ mime: 'video/mp4' }); setReads(hotVid.id, 75); // >= video promoteReadsPerHour 60
+ const warmVid = await addObject({ mime: 'video/mp4' }); setReads(warmVid.id, 30); // in the band: stays local
+ const warmR2 = await addObject({ mime: 'video/mp4' }); setReads(warmR2.id, 30); await inR2(warmR2, 30); // in the band: keeps R2
+ const coldVid = await addObject({ mime: 'video/mp4' }); setReads(coldVid.id, 2); await inR2(coldVid, 30); // < video demoteReadsPerHour 6
+ const sz = policy.settings();
+ const scanned = Number((await db.get(`SELECT COUNT(*) AS n FROM media_objects o WHERE o.legacy_ref IS NULL AND o.lifecycle_status = 'ready' AND o.size_bytes >= ? AND o.size_bytes <= ?
+ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.provider = 'r2' AND l.state = 'present')
+ AND NOT EXISTS (SELECT 1 FROM apps a WHERE a.app_id = o.app_id AND a.env = 'sandbox')`, [Math.ceil(sz.promoteMinSizeMb * 1048576), Math.floor(sz.promoteMaxSizeMb * 1048576)])).n);
+ const copies = Number((await db.get("SELECT COUNT(*) AS n FROM media_locations l JOIN media_objects o ON o.id = l.object_id WHERE l.provider = 'r2' AND o.legacy_ref IS NULL")).n);
+ console.warn = (...a) => { sweepWarns.push(a.join(' ')); };
+ s = await tiering.runSweep();
+ console.warn = warn0;
+ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['valkey', undefined], JSON.stringify(s));
+ assert.ok(scanned > 1000, `the scan spans several pages (${scanned} objects)`);
+ assert.strictEqual(mgets.length, Math.ceil(copies / 500) + Math.ceil(scanned / 500), `one MGET per page, never one per object: ${mgets}`);
+ assert.ok(mgets.every((n) => n <= 500 * demand.WINDOW_BUCKETS), 'a page reads its objects\' hour of buckets in one MGET');
+ assert.ok(!sweepWarns.some((w) => /PostgreSQL counts/.test(w)), sweepWarns.join('\n'));
+ assert.ok(await r2Row(hotVid.id), 'hot in Valkey: promoted');
+ d = await last(hotVid.id);
+ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'done']);
+ assert.ok(/^75 reads in the last hour \(demand valkey, region local\) >= promoteReadsPerHour 60 \(class video\); .*1 of budget 3 \(maxPromotionsPerSweep/.test(d.reason), d.reason);
+ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'valkey', metric: 'reads_1h', value: 75, threshold: 60, region: 'local', window_s: 3600 });
+ assert.ok(!await r2Row(coldVid.id), 'cold in Valkey: demoted');
+ d = await last(coldVid.id);
+ assert.deepStrictEqual([d.action, d.outcome], ['demote', 'done']);
+ assert.ok(/^2 reads in the last hour \(demand valkey, region local\) < demoteReadsPerHour 6 \(class video\)/.test(d.reason), d.reason);
+ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'valkey', metric: 'reads_1h', value: 2, threshold: 6, region: 'local', window_s: 3600 });
+ assert.ok(!await r2Row(warmVid.id) && await r2Row(warmR2.id), 'between the thresholds nothing moves (hysteresis)');
+ assert.deepStrictEqual([(await decisions(warmVid.id)).length, (await decisions(warmR2.id)).length], [0, 0]);
+ let mtext = await registry.metricsAsync();
+ assert.ok(/media_sweep_demand_source_total\{source="valkey"\} 1\b/.test(mtext), mtext.split('\n').filter((l) => /demand_source/.test(l)).join('\n'));
+ const pgSweeps = () => Number((/media_sweep_demand_source_total\{source="pg"\} (\d+)/.exec(mtext) || [])[1]);
+ const pgBefore = pgSweeps();
+
+ // Valkey refuses: the whole sweep reads the daily counts, exactly today's candidates and reasons, logged once.
+ const pgHot = await addObject(); await setViews(pgHot.id, [[0, 900]]);
+ const pgPromote = (await tiering.promotionCandidates(policy.settings(), { limit: 500, excludeSandbox: true })).map((c) => c.id);
+ const pgDemote = (await tiering.demotionCandidates(policy.settings())).map((c) => c.row.id);
+ assert.ok(pgPromote.includes(pgHot.id) && pgDemote.includes(warmR2.id), 'by the daily counts pgHot is popular and warmR2 idle');
+ vkMode = 'down';
+ mgets.length = 0;
+ console.warn = (...a) => { sweepWarns.push(a.join(' ')); };
+ s = await tiering.runSweep();
+ console.warn = warn0;
+ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey read failed: ECONNREFUSED'], JSON.stringify(s));
+ assert.strictEqual(mgets.length, 1, 'the first failed read ends the Valkey scan');
+ assert.deepStrictEqual([s.candidates.promote, s.candidates.demote], [pgPromote.length, pgDemote.length], 'the same candidates as the daily counts give');
+ assert.strictEqual(sweepWarns.filter((w) => /Sweep eligibility reads the daily PostgreSQL counts: Valkey read failed: ECONNREFUSED/.test(w)).length, 1, 'logged once per sweep');
+ assert.ok(await r2Row(pgHot.id) && !await r2Row(warmR2.id), 'the daily counts decide: pgHot promoted, idle warmR2 demoted');
+ d = await last(pgHot.id);
+ assert.ok(/^900 unique viewers in 7 days >= promoteMinUniqueViewers7d 500/.test(d.reason), d.reason);
+ assert.deepStrictEqual(JSON.parse(d.inputs).demand, { source: 'pg', metric: 'unique_viewers_7d', value: 900, threshold: 500 });
+ d = await last(warmR2.id);
+ assert.ok(/^no viewer since the kept counts began \(idle for demoteIdleDays 14\)/.test(d.reason), d.reason);
+ assert.deepStrictEqual([JSON.parse(d.inputs).demand.source, JSON.parse(d.inputs).demand.metric], ['pg', 'last_viewed_day']);
+ mtext = await registry.metricsAsync();
+ assert.strictEqual(pgSweeps(), pgBefore + 1, 'counted as a pg sweep');
+
+ // Valkey hangs: the read times out and the sweep still finishes on the daily counts.
+ vkMode = 'hung';
+ console.warn = () => {};
+ const t0 = Date.now();
+ s = await tiering.runSweep();
+ console.warn = warn0;
+ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey read failed: valkey read timeout'], JSON.stringify(s));
+ assert.ok(Date.now() - t0 < 10000, 'a stuck Valkey never stalls the sweep');
+ demand.useValkey(null);
+ s = await tiering.runSweep();
+ assert.deepStrictEqual([s.demand_source, s.demand_fallback], ['pg', 'Valkey is not configured']);
+ await db.run('DELETE FROM media_object_views_daily');
+ for (const v of candViews) await db.run('INSERT INTO media_object_views_daily (object_id, day, unique_viewers, last_viewed_at) VALUES (?, ?, ?, ?)', [v.object_id, v.day, v.unique_viewers, v.last_viewed_at]);
+ console.log('✅ demand from Valkey: hot promotes, cold demotes, the band holds; one MGET per page; a failed or stuck Valkey falls back to the daily counts for the whole sweep');
+
// ── 10. The storage sweep runs it (step 4); a restore drill never does ──
// ── 10. The storage sweep runs it (step 4); a restore drill never does ──
const vodSweep = await storage.runSweep();
const vodSweep = await storage.runSweep();
assert.ok(vodSweep.objects && vodSweep.objects.gate === true && typeof vodSweep.objects.promoted === 'number', JSON.stringify(vodSweep));
assert.ok(vodSweep.objects && vodSweep.objects.gate === true && typeof vodSweep.objects.promoted === 'number', JSON.stringify(vodSweep));
@@ -121,13 +121,18 @@ const path = require('path');
@@ -121,13 +121,18 @@ const path = require('path');
assert.strictEqual((await demand.hotness()).size, 0); // the hung read times out
assert.strictEqual((await demand.hotness()).size, 0); // the hung read times out
assert.ok(Date.now() - before < 5000);
assert.ok(Date.now() - before < 5000);
assert.strictEqual((await demand.rollup()).announced, 0);
assert.strictEqual((await demand.rollup()).announced, 0);
+ // strict (the tiering sweep): a failure throws instead of answering zeros, so the caller can fall back.
+ assert.strictEqual(demand.available(), true);
+ await assert.rejects(demand.hotness({ objectIds: ['obj-a'], strict: true }), /ECONNREFUSED/);
await new Promise((resolve) => setImmediate(resolve));
await new Promise((resolve) => setImmediate(resolve));
assert.deepStrictEqual(unhandled, []);
assert.deepStrictEqual(unhandled, []);
- console.log('✅ a Valkey failure never throws into a read; reads answer no demand');
+ console.log('✅ a Valkey failure never throws into a read; reads answer no demand; a strict read throws');
// 6. rollup(): media.object.hot through the placement outbox, at most once per object per hour.
// 6. rollup(): media.object.hot through the placement outbox, at most once per object per hour.
demand.useValkey(null);
demand.useValkey(null);
demand._reset();
demand._reset();
+ assert.strictEqual(demand.available(), false);
+ await assert.rejects(demand.hotness({ objectIds: ['obj-a'], strict: true }), /Valkey is not configured/);
const stub = http.createServer((req, res) => {
const stub = http.createServer((req, res) => {
req.resume();
req.resume();
req.on('end', () => {
req.on('end', () => {
@@ -16,11 +16,15 @@ const path = require('path');
@@ -16,11 +16,15 @@ const path = require('path');
await policy.init();
await policy.init();
assert.strictEqual(policy.get().revision(), 1);
assert.strictEqual(policy.get().revision(), 1);
+ const bands = { image: [300, 30], download: [30, 3] };
for (const name of policy.CLASSES) {
for (const name of policy.CLASSES) {
+ const [promoteReadsPerHour, demoteReadsPerHour] = bands[name] || [60, 6];
assert.deepStrictEqual(policy.DEFAULTS.classes[name], {
assert.deepStrictEqual(policy.DEFAULTS.classes[name], {
- maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0,
+ maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10, minResidencyMs: 0, promoteReadsPerHour, demoteReadsPerHour,
});
});
assert.deepStrictEqual(policy.budgetFor(name), { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10 });
assert.deepStrictEqual(policy.budgetFor(name), { maxPromotionsPerSweep: 3, maxDemotionsPerSweep: 10 });
+ assert.deepStrictEqual(policy.bandFor(name), { promoteReadsPerHour, demoteReadsPerHour });
+ assert.ok(demoteReadsPerHour < promoteReadsPerHour, `${name}: hysteresis`);
}
}
assert.strictEqual(policy.mayMove({ class: 'video', lastMovedAt: 1000, now: 1000 }), true, 'zero residency is off');
assert.strictEqual(policy.mayMove({ class: 'video', lastMovedAt: 1000, now: 1000 }), true, 'zero residency is off');
assert.throws(() => policy.budgetFor('unknown'), RangeError);
assert.throws(() => policy.budgetFor('unknown'), RangeError);
@@ -46,6 +50,13 @@ const path = require('path');
@@ -46,6 +50,13 @@ const path = require('path');
assert.deepStrictEqual(policy.budgetFor('image'), { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4 }, 'second revision preserves image settings');
assert.deepStrictEqual(policy.budgetFor('image'), { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4 }, 'second revision preserves image settings');
assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 });
assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 });
+ // The hysteresis band: demote below promote, refused as a whole revision otherwise.
+ await assert.rejects(policy.set({ classes: { video: { demoteReadsPerHour: 60 } } }, { reason: 'test: an empty band' }), /demoteReadsPerHour must be below promoteReadsPerHour/);
+ assert.strictEqual(policy.get().revision(), 3, 'a refused band changes nothing');
+ await policy.set({ classes: { video: { promoteReadsPerHour: 120, demoteReadsPerHour: 20 } } }, { reason: 'test: a wider video band' });
+ assert.deepStrictEqual(policy.bandFor('video'), { promoteReadsPerHour: 120, demoteReadsPerHour: 20 });
+ assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 }, 'the band keeps the budget');
+
policy._reset();
policy._reset();
console.log('placement-storage-policy: all checks passed');
console.log('placement-storage-policy: all checks passed');
process.exit(0);
process.exit(0);