openvibers/openvibe.media Commits
Hold hot-storage ceilings against projected dry-run spend and zero caps (#23)
Changes
@@@ -35,7 +35,8 @@"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",-- "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"++ "tiering on demand (F2.5): the object sweep promotes and demotes on the hour's Valkey reads against per-class hysteresis bands, one MGET per page, falling back to the daily PostgreSQL counts for a whole sweep when Valkey fails",++ "placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs"],"updated": "2026-10-03"}@@@ -146,6 +146,26 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:decision's `inputs.demand` (`{ source, metric, value, threshold[, region, window_s] }`) and its reason say which sourceand 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`).++- **Constraints (F2.6, shipped):** a sweep move needs the provider class it relies on: a promotion needs R2 eligible++ for `online-hot`, the demotion of a ready object needs its canonical copy's provider eligible for its class++ (`online-canonical`; the Media host's disk is always the `regional-cache`). Eligible means the circuit breaker is++ not open and no completed capability probe found a capability the class needs missing++ (`providers.placementGate()`, whatever `MEDIA_CAPABILITY_GATE` says; a probe not run or not completed decides++ nothing). Otherwise the move is refused and every copy stays; a promotion's refusal spends the class's slot (every++ promotion relies on R2, so a failing R2 logs at most a class budget of refusals per sweep), a demotion's does not++ (its gate is that object's own canonical provider). Each class may carry a monthly ceiling on its R2++ storage, `maxHotUsdPerMonth` in `media.storage_policy` (absent = none; `0` = the class never promotes, whatever++ R2's configured price), priced at R2 Standard's list price in++ `media.cost_tiers` over every present R2 copy of the class, gross of R2's free tier (`freeGb` is account-wide, not++ per class, so the ceiling binds up to its value early), and kept current as the sweep moves: the first++ promotion that would pass it is refused and closes the class's promotions for that sweep (a cheaper, lower-ranked++ object never jumps the value-per-dollar queue). Every decision records `inputs.provider_gate` (`{ provider, class,++ eligible, breaker, probe[, missing, reason] }`) and a sweep promotion `inputs.budget` (`{ class, usd_per_month,++ copy_usd_per_month, ceiling }`); the sweep result carries `constraints` (`promote_gate`, `refused.{provider,budget}`,++ `hot_usd_per_month.<class>.{spend,projected,ceiling}` for every class, `ceiling` null when unset). `spend` follows++ only the moves made; `projected` also counts the dry runs, and the ceiling (and `inputs.budget.usd_per_month`) is held++ against it, so dry runs (gate off) are refused as a live sweep would be, without changing `spend`. Native objects classify only as `video`, `image` or `download` today, so a ceiling (or budget)++ set on `game-asset`, `attachment` or `backup` decides nothing yet.## 7. Delivery: sticky, measured, canaried@@@ -207,7 +227,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thrcorrected 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 (F2.4, shipped) and the sweep's-- eligibility on them with per-class hysteresis and a PostgreSQL fallback (F2.5, shipped).++ eligibility on them with per-class hysteresis and a PostgreSQL fallback (F2.5, shipped), the provider-class gate++ and per-class monthly R2 storage ceilings on its moves (F2.6, shipped). Still open in F2: the move cleanup job and++ its alert after three failures (§6), every class (not only native objects ↔ R2) through the one sweep.- **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@@@ -16,6 +16,13 @@* 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.++ * Constraints (F2.6): a promotion needs R2 eligible for the online-hot class, and the demotion of a ready object needs++ * its canonical copy's provider eligible for its class (../placement/providers.js placementGate(): breaker not open,++ * no failed capability), else it is refused and the copies stay where they are; each decision's inputs.provider_gate++ * records the gate it saw. A class with maxHotUsdPerMonth (media.storage_policy) promotes only while its R2 copies'++ * list-price storage (media.cost_tiers) stays within it: the first promotion that would pass it is refused++ * (inputs.budget) and closes the class's promotions for the sweep, so a cheaper, lower-ranked object never jumps++ * the value-per-dollar queue.** 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@@@ -47,6 +54,7 @@ const policy = require('./tier-policy');const storagePolicy = require('../placement/storage-policy');const costTiers = require('../placement/cost-tiers');const value = require('../placement/value');++const providers = require('../placement/providers');const popularity = require('./popularity');const MB = 1024 * 1024;@@@ -128,6 +136,28 @@ function r2KeyFor(obj, canon) {return canon && canon.provider === 'b2' ? canon.key : `objects/${obj.app_id}/${obj.id}`;}++// ── Provider classes and the storage ceiling (F2.6) ──────────++++/** The class a canonical copy serves: the Media host's disk is the regional cache, a remote provider the online canonical. */++const canonicalClass = (provider) => (provider === 'local' ? 'regional-cache' : 'online-canonical');++++/** R2 Standard's list price of keeping `bytes` for a month (media.cost_tiers). */++function hotUsdPerMonth(bytes) {++ return (Number(bytes) || 0) / value.GB * (Number(costTiers.settings().r2.standard.storagePerGbMonth) || 0);++}++++const fmtUsd = (v) => `$${Number(Number(v).toPrecision(4))}`;++++/** What each class's present R2 copies cost a month at list price, gross of R2's account-wide free tier: { video: 0.12, image: 0, ... }. */++async function hotSpend() {++ const out = Object.fromEntries(storagePolicy.CLASSES.map((c) => [c, 0]));++ const rows = await db.all(`SELECT o.kind, o.mime_type, SUM(COALESCE(l.size_bytes, o.size_bytes, 0))::bigint AS bytes++ FROM media_locations l JOIN media_objects o ON o.id = l.object_id++ WHERE l.provider = 'r2' AND l.state = 'present' GROUP BY o.kind, o.mime_type`);++ for (const r of rows) out[storagePolicy.classOf(r)] += hotUsdPerMonth(r.bytes);++ return out;++}++// ── Decisions ────────────────────────────────────────────────function thresholdSnapshot(settings) {@@@ -194,7 +224,12 @@ async function frame(objectId, action, ctx) {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;++ if (ctx.budget) inputs.budget = ctx.budget;const canon = canonicalLocation(obj, locs);++ // The provider class the move relies on: R2 as online-hot for a promotion, the canonical copy's provider for a demotion.++ const gate = action === 'promote' ? providers.placementGate('r2', 'online-hot')++ : canon && canon.provider !== 'r2' ? providers.placementGate(canon.provider, canonicalClass(canon.provider)) : null;++ if (gate) inputs.provider_gate = gate;const trigger = ctx.trigger || 'manual';const why = ctx.reason || 'requested';const from = action === 'promote' ? (canon ? canon.provider : null) : 'r2';@@@ -204,7 +239,7 @@ async function frame(objectId, action, ctx) {if (!repeat) await recordDecision({ obj, action, from, to, outcome, trigger, reason, inputs, settings, error: extra.error });return { ok: outcome === 'done' || outcome === 'already', outcome, object_id: obj.id, repeat, ...extra };};-- return { obj, locs, canon, held, settings, why, log };++ return { obj, locs, canon, held, settings, why, log, gate };}/** Why an object may not be promoted at all, from the record (no I/O), or null. */@@@ -225,13 +260,15 @@ async function promoteRefusal(obj, { locs, held }) {async function promote(objectId, ctx = {}) {const f = await frame(objectId, 'promote', ctx);if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };-- const { obj, locs, canon, held, settings, why, log } = f;++ const { obj, locs, canon, held, settings, why, log, gate } = f;if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });const r2Now = locs.find((l) => l.provider === 'r2');if (r2Now && r2Now.state === 'present' && r2Now.verified_at) return await log('already', `${why}; the R2 copy is already present`);const refusal = await promoteRefusal(obj, { locs, held });if (refusal) return await log('refused', `${why}; refused: ${refusal}`, { error: refusal });++ if (!gate.eligible) return await log('refused', `${why}; refused: ${gate.reason} (provider class gate)`, { error: gate.reason, blocked: 'provider' });++ if (ctx.overBudget) return await log('refused', `${why}; refused: ${ctx.overBudget}`, { error: ctx.overBudget, blocked: 'budget' });const vs = vodStorage();if (!settings.active) return await log('dry_run', `gate off (active = false): would promote: ${why}${vs.providerAvailable('r2') ? '' : ' (R2 is not available now)'}`);if (!vs.providerAvailable('r2')) return await log('refused', `${why}; refused: R2 is not available`, { error: 'R2 not available' });@@@ -290,11 +327,15 @@ async function promote(objectId, ctx = {}) {async function demote(objectId, ctx = {}) {const f = await frame(objectId, 'demote', ctx);if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };-- const { obj, locs, canon, held, settings, why, log } = f;++ const { obj, locs, canon, held, settings, why, log, gate } = f;if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });const r2 = locs.find((l) => l.provider === 'r2');if (!r2) return await log('already', `${why}; there is no R2 copy`);++ // Reads of a ready object fall back to its canonical copy: never shift them onto a provider that is failing.++ if (obj.lifecycle_status === 'ready' && gate && !gate.eligible) {++ return await log('refused', `${why}; refused: ${gate.reason} (provider class gate); the R2 copy is kept`, { error: gate.reason, blocked: 'provider' });++ }const vs = vodStorage();if (!settings.active) return await log('dry_run', `gate off (active = false): would demote: ${why}`);if (!vs.providerConfigured('r2')) return await log('refused', `${why}; refused: R2 is not configured`, { error: 'R2 not configured' });@@@ -446,7 +487,7 @@ async function hotPromotionCandidates(settings, demand, { appId = null, limit =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,++ const rows = await db.all(`SELECT o.*, l.created_at AS r2_since, l.state AS r2_state, l.size_bytes AS r2_size_bytes,(SELECT MAX(d.day) FROM media_object_views_daily d WHERE d.object_id = o.id AND d.unique_viewers > 0) AS last_viewed_day,(SELECT COALESCE(SUM(d.unique_viewers), 0)::bigint FROM media_object_views_daily d WHERE d.object_id = o.id AND d.day >= ?) AS unique_viewers_30dFROM media_locations l JOIN media_objects o ON o.id = l.object_id@@@ -563,7 +604,8 @@ async function sweepCandidates(settings, now) {}/**-- * One pass over native objects. → { gate, demand_source, demand_fallback?, promoted, demoted, would_promote, would_demote, already, refused, failed,++ * One pass over native objects. → { gate, demand_source, demand_fallback?, constraints: { promote_gate, refused: { provider, budget },++ * hot_usd_per_month: { <class>: { spend, projected, ceiling } } }, 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() } = {}) {@@@ -590,6 +632,18 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {// 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: {} };++ // A slot is spent by a move, a failed move, a dry run, or a promotion's provider-gate refusal, repeated or not++ // (every promotion relies on R2, so the next in line would be refused for the same reason and a closed R2 costs++ // the class at most its budget of rows). A demotion's gate is the candidate's own canonical provider, so its++ // refusal spends nothing and an object with a healthy canonical copy takes the slot.++ const spends = (r, action) => (action === 'promote' && r.blocked === 'provider') || (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome));++ const blocked = (r) => { if (r.blocked) out.constraints.refused[r.blocked]++; };++ // The storage ceiling (F2.6): each class's R2 copies at list price. `spend` follows the moves made; `projected`++ // also counts dry runs (the moves the sweep would make), and the ceiling is held against it.++ const spend = await hotSpend();++ const projected = { ...spend };++ const closed = new Set();++ out.constraints = { promote_gate: providers.placementGate('r2', 'online-hot'), refused: { provider: 0, budget: 0 }, hot_usd_per_month: {} };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, demand } of demote_) {if (!room('demote', cls)) continue;@@@ -599,7 +653,14 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {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;++ blocked(r);++ if (spends(r, 'demote')) spent.demote[cls] = (spent.demote[cls] || 0) + 1;++ // hotSpend() counted only present copies: a corrupt or pending one leaves the class's spend as it was.++ if (!r.repeat && ['done', 'dry_run'].includes(r.outcome) && row.r2_state === 'present') {++ const freed = hotUsdPerMonth(row.r2_size_bytes ?? row.size_bytes);++ projected[cls] = Math.max(0, projected[cls] - freed);++ if (r.outcome === 'done') spend[cls] = Math.max(0, spend[cls] - freed);++ }}// 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.@@@ -607,13 +668,30 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {out.candidates.promote = promote_.length;for (const c of promote_) {const cls = c.class;-- if (!room('promote', cls)) continue;++ if (!room('promote', cls) || closed.has(cls)) continue;if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }const residency = await residencyRefusal(c, cls, 'promote', now);++ const cost = hotUsdPerMonth(c.size_bytes);++ const ceiling = storagePolicy.hotCeilingFor(cls);++ const usd = projected[cls];++ const budget = { class: cls, usd_per_month: usd, copy_usd_per_month: cost, ceiling };++ // A zero ceiling closes the class even when R2's configured price is zero (every copy would then cost $0).++ const overBudget = ceiling === 0++ ? `class ${cls} has maxHotUsdPerMonth $0 (media.storage_policy): it never promotes; the class promotes nothing more this sweep`++ : ceiling != null && usd + cost > ceiling++ ? `class ${cls} R2 storage would reach ${fmtUsd(usd + cost)}/month (${fmtUsd(usd)} now + ${fmtUsd(cost)} for this copy) > maxHotUsdPerMonth ${fmtUsd(ceiling)} (media.storage_policy); the class promotes nothing more this sweep`++ : null;const r = await promote(c.id, { trigger, reason: promoteReason(c, settings, (spent.promote[cls] || 0) + 1), settings, now, residency, valuePerDollar: c.value_per_dollar,-- demand: c.demand, pop: picked.source === 'pg' ? { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } : undefined });++ demand: c.demand, budget, overBudget, 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;++ blocked(r);++ if (r.blocked === 'budget') closed.add(cls);++ if (spends(r, 'promote')) spent.promote[cls] = (spent.promote[cls] || 0) + 1;++ if (!r.repeat && ['done', 'dry_run'].includes(r.outcome)) projected[cls] += cost;++ if (!r.repeat && r.outcome === 'done') spend[cls] += cost;++ }++ for (const cls of storagePolicy.CLASSES) {++ out.constraints.hot_usd_per_month[cls] = { spend: spend[cls], projected: projected[cls], ceiling: storagePolicy.hotCeilingFor(cls) };}// Demand (F2.4): stage media.object.hot for objects over the hourly read threshold.@@@ -698,7 +776,7 @@ async function report({ appId = null, limit = 20 } = {}) {}module.exports = {-- FAILED_BACKOFF_H, OUTCOMES,++ FAILED_BACKOFF_H, OUTCOMES, hotSpend, hotUsdPerMonth,runSweep, promote, demote, candidates, promotionCandidates, demotionCandidates,canonicalUnverified, confirmCanonical, r2KeyFor,counts24h, listDecisions, decisionPublic, report,@@@ -15,6 +15,8 @@* probeAll() probe every configured remote provider* startHealthLoop() the cheap per-minute HeadBucket and the hourly capability probe* healthStatus() { b2, r2 } each as { configured, healthy, lastCheck, lastError }++ * placementGate(name, cls) may a placement move put bytes on (or shift reads to) it for the class:++ * breaker not open, no failed capability the class needs (F2.6)** A provider that fails a capability is removed from the classes that need it until it passes* again — `probeProvider(name)` returning a capability list with `passed=false` is the only@@@ -92,6 +94,28 @@ function eligibleClasses(name) {return measured;}++/**++ * Whether placement may put bytes on (or shift reads to) a provider for a class (docs/media-fabric.md §2, F2.6):++ * the provider declares the class, its circuit breaker is not open, and no completed capability probe found a++ * capability the class needs missing. A probe not run yet, or one that could not complete, decides nothing (as for++ * eligibleClasses(): live health is the breaker's job). This++ * holds whatever MEDIA_CAPABILITY_GATE says: the read gate protects playback, and a placement move refused here++ * leaves every copy where it is. Whether the provider is configured is the caller's check.++ * → { provider, class, eligible, breaker, probe: 'pending'|'incomplete'|'passed'|'failed', missing?, reason? }++ */++function placementGate(name, cls) {++ const out = { provider: name, class: cls, eligible: true, breaker: 'closed', probe: 'passed' };++ if (!providerClasses(name).includes(cls)) return { ...out, eligible: false, reason: `${name} does not serve ${cls}` };++ if (name === 'local') return out;++ out.breaker = require('./signals').breakerState(name);++ const caps = capabilities(name);++ const missing = caps.passed ? CLASS_REQUIREMENTS[cls].filter((c) => !(caps.capabilities || {})[c]) : [];++ out.probe = caps.passed === null ? 'pending' : !caps.passed ? 'incomplete' : missing.length ? 'failed' : 'passed';++ if (out.breaker === 'open') return { ...out, eligible: false, reason: `${name} is not eligible for ${cls}: its circuit breaker is open` };++ if (out.probe === 'failed') return { ...out, eligible: false, missing, reason: `${name} is not eligible for ${cls}: the capability probe found ${missing.join(', ')} missing` };++ return out;++}++/** What classes each provider says it can serve. These are the design's assignments (docs/media-fabric.md §2):* B2 is the online canonical, R2 Standard the online hot tier, the Media host's NVMe the regional cache. */function providerClasses(name) {@@@ -315,7 +339,7 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};module.exports = {CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,-- isConfigured, providerClasses, eligibleClasses, measuredClasses, gateMode, capabilities,++ isConfigured, providerClasses, eligibleClasses, measuredClasses, placementGate, gateMode, capabilities,probe, probeAll, liveHealth, healthStatus,startHealthLoop, stop, reset,// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).@@@ -31,6 +31,9 @@ const CLASS_SCHEMA = {minResidencyMs: { type: 'integer', minimum: 0 },promoteReadsPerHour: { type: 'integer', minimum: 1, maximum: 1e9 },demoteReadsPerHour: { type: 'integer', minimum: 0, maximum: 1e9 },++ // F2.6: the class's R2 Standard storage spend ceiling, in dollars a month at media.cost_tiers' list price++ // (every present R2 copy of the class counts). Absent = no ceiling; 0 = the class never promotes, even at a $0 price.++ maxHotUsdPerMonth: { type: 'number', minimum: 0, maximum: 1e9 },},};const SCHEMA = {@@@ -117,6 +120,12 @@ function budgetFor(name) {return { maxPromotionsPerSweep, maxDemotionsPerSweep };}++/** The class's monthly R2 storage ceiling in dollars (maxHotUsdPerMonth), or null when it has none. */++function hotCeilingFor(name) {++ const { maxHotUsdPerMonth } = classPolicy(name);++ return maxHotUsdPerMonth == null ? null : maxHotUsdPerMonth;++}++/** The class's hourly promote and demote thresholds (hysteresis: demote < promote). */function bandFor(name) {const { promoteReadsPerHour, demoteReadsPerHour } = classPolicy(name);@@@ -139,4 +148,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, bandFor, mayMove, _reset: () => { store = null; } };++module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, hotCeilingFor, bandFor, mayMove, _reset: () => { store = null; } };@@@ -640,6 +640,114 @@ const crypto = require('crypto');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');++ // ── 9f. Constraints (F2.6): the provider class gate and each class's monthly R2 storage ceiling ──++ const providersMod = require('../server/placement/providers');++ const signals = require('../server/placement/signals');++ await db.run('DELETE FROM media_object_views_daily');++ await sp.set({ classes: { video: { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 500 } } }, { reason: 'test: video budget 1' });++ // R2's breaker is open: the best video is refused (its slot spent, so the next waits) and nothing moves.++ const gA = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];++ await setViews(gA[0].id, [[0, 900]]); await setViews(gA[1].id, [[0, 800]]);++ signals.openBreaker('r2');++ s = await tiering.runSweep();++ signals.closeBreaker('r2');++ assert.ok(!await r2Row(gA[0].id) && !await r2Row(gA[1].id), 'no promotion while R2 is not eligible for online-hot');++ d = await last(gA[0].id);++ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);++ assert.ok(/; refused: r2 is not eligible for online-hot: its circuit breaker is open \(provider class gate\)$/.test(d.reason), d.reason);++ assert.deepStrictEqual((({ provider, class: c, eligible, breaker }) => [provider, c, eligible, breaker])(JSON.parse(d.inputs).provider_gate), ['r2', 'online-hot', false, 'open']);++ assert.strictEqual((await decisions(gA[1].id)).length, 0, 'the refusal spent the class slot');++ assert.deepStrictEqual([s.constraints.promote_gate.eligible, s.constraints.refused.provider >= 1], [false, true], JSON.stringify(s.constraints));++ // A completed probe found a capability online-hot needs missing: refused the same way, naming it.++ for (const o of gA) await db.run('DELETE FROM media_object_views_daily WHERE object_id = ?', [o.id]);++ const gB = await addObject({ mime: 'video/mp4' }); await setViews(gB.id, [[0, 900]]);++ providersMod._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: false }, lastChecked: Date.now(), lastError: null });++ s = await tiering.runSweep();++ providersMod._setCapabilities('r2', { provider: 'r2', passed: null, capabilities: {}, lastChecked: null, lastError: null });++ d = await last(gB.id);++ assert.ok(!await r2Row(gB.id) && d.outcome === 'refused' && /r2 is not eligible for online-hot: the capability probe found range_206 missing/.test(d.reason), d.reason);++ assert.deepStrictEqual([JSON.parse(d.inputs).provider_gate.probe, JSON.parse(d.inputs).provider_gate.missing], ['failed', ['range_206']]);++ await db.run('DELETE FROM media_object_views_daily WHERE object_id = ?', [gB.id]);++ // A ready object's R2 copy stays while its canonical copy's provider (B2, online-canonical) is failing. The++ // refusal spends no demotion slot (the gate is per object): the smaller idle copy over a local canonical leaves.++ await sp.set({ classes: { video: { maxDemotionsPerSweep: 1 } } }, { reason: 'test: video demotion budget 1' });++ const gC = await addObject({ mime: 'video/mp4', canonical: 'b2' }); await inR2(gC, 30);++ const gL = await addObject({ mime: 'video/mp4', size: 2048 }); await inR2(gL, 30);++ signals.openBreaker('b2');++ s = await tiering.runSweep();++ signals.closeBreaker('b2');++ await sp.set({ classes: { video: { maxDemotionsPerSweep: 500 } } }, { reason: 'test: video demotion budget back' });++ assert.ok(!await r2Row(gL.id), 'a demotion refused by its own canonical provider leaves the slot to the next in line');++ d = await last(gC.id);++ assert.ok(await r2Row(gC.id), 'the R2 copy is kept while B2 is not eligible');++ assert.deepStrictEqual([d.action, d.outcome], ['demote', 'refused']);++ assert.ok(/; refused: b2 is not eligible for online-canonical: its circuit breaker is open \(provider class gate\); the R2 copy is kept$/.test(d.reason), d.reason);++ assert.deepStrictEqual([JSON.parse(d.inputs).provider_gate.provider, JSON.parse(d.inputs).provider_gate.eligible], ['b2', false]);++ // A corrupt R2 copy was never counted in the class's spend, so demoting it subtracts nothing.++ const gX = await addObject({ mime: 'video/mp4' }); await inR2(gX, 30);++ await db.run("UPDATE media_locations SET state = 'corrupt' WHERE object_id = ? AND provider = 'r2'", [gX.id]);++ s = await tiering.runSweep();++ assert.ok(!await r2Row(gC.id), 'B2 eligible again: the idle copy leaves');++ assert.ok(!await r2Row(gX.id), 'the corrupt copy leaves');++ assert.ok(Math.abs(s.constraints.hot_usd_per_month.video.spend - (await tiering.hotSpend()).video) < 1e-12, JSON.stringify(s.constraints));++ assert.deepStrictEqual(Object.keys(s.constraints.hot_usd_per_month).sort(), [...sp.CLASSES].sort(), 'every class is reported');++ assert.deepStrictEqual(JSON.parse((await last(gC.id)).inputs).provider_gate, { provider: 'b2', class: 'online-canonical', eligible: true, breaker: 'closed', probe: 'pending' });++ // The ceiling: room for one more 4 KB video copy; the first in line takes it, the next is refused and closes the class.++ const spend0 = (await tiering.hotSpend()).video;++ const one = tiering.hotUsdPerMonth(4096);++ assert.ok(one > 0 && Math.abs(one - 4096 / 1073741824 * 0.015) < 1e-15, `R2 Standard list price of 4 KB a month: ${one}`);++ const ceiling = spend0 + one * 1.5;++ await sp.set({ classes: { video: { maxPromotionsPerSweep: 3, maxHotUsdPerMonth: ceiling } } }, { reason: 'test: a video storage ceiling' });++ const gD = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];++ await setViews(gD[0].id, [[0, 900]]); await setViews(gD[1].id, [[0, 800]]); await setViews(gD[2].id, [[0, 700]]);++ s = await tiering.runSweep();++ assert.deepStrictEqual([!!await r2Row(gD[0].id), !!await r2Row(gD[1].id), !!await r2Row(gD[2].id)], [true, false, false], 'one copy fits under the ceiling');++ assert.deepStrictEqual(JSON.parse((await last(gD[0].id)).inputs).budget, { class: 'video', usd_per_month: spend0, copy_usd_per_month: one, ceiling });++ d = await last(gD[1].id);++ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);++ assert.ok(/; refused: class video R2 storage would reach \$[0-9.e-]+\/month \(\$[0-9.e-]+ now \+ \$[0-9.e-]+ for this copy\) > maxHotUsdPerMonth \$[0-9.e-]+ \(media\.storage_policy\); the class promotes nothing more this sweep$/.test(d.reason), d.reason);++ assert.strictEqual((await decisions(gD[2].id)).length, 0, 'a cheaper lower-ranked object does not jump the queue: the class is closed');++ assert.strictEqual(s.constraints.refused.budget, 1);++ assert.ok(Math.abs(s.constraints.hot_usd_per_month.video.spend - (spend0 + one)) < 1e-12 && s.constraints.hot_usd_per_month.video.ceiling === ceiling, JSON.stringify(s.constraints));++ assert.strictEqual(s.constraints.hot_usd_per_month.video.projected, s.constraints.hot_usd_per_month.video.spend, 'with the gate on, projected is actual');++ // Gate off: a dry run holds the ceiling against the projected spend (what it would move) and leaves the actual spend alone.++ await policy.set({ active: false }, { actor: { type: 'service', id: 'live' }, reason: 'test: gate off under a ceiling' });++ const spend1 = (await tiering.hotSpend()).video;++ await sp.set({ classes: { video: { maxHotUsdPerMonth: spend1 + one * 1.5 } } }, { reason: 'test: a video storage ceiling, dry run' });++ const gE = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];++ await setViews(gE[0].id, [[0, 990]]); await setViews(gE[1].id, [[0, 980]]); await setViews(gE[2].id, [[0, 970]]);++ s = await tiering.runSweep();++ assert.strictEqual(s.would_demote, 0, JSON.stringify(s));++ assert.ok(!await r2Row(gE[0].id) && !await r2Row(gE[1].id), 'a dry run moves nothing');++ const e0 = await last(gE[0].id), e1 = await last(gE[1].id);++ assert.deepStrictEqual([e0.outcome, e1.outcome], ['dry_run', 'refused']);++ assert.ok(/maxHotUsdPerMonth/.test(e1.reason), e1.reason);++ assert.ok(Math.abs(JSON.parse(e1.inputs).budget.usd_per_month - (JSON.parse(e0.inputs).budget.usd_per_month + one)) < 1e-12, e1.inputs);++ assert.strictEqual((await decisions(gE[2].id)).length, 0, 'the projected spend closed the class');++ const hv = s.constraints.hot_usd_per_month.video;++ assert.ok(Math.abs(hv.spend - spend1) < 1e-12 && Math.abs(hv.spend - (await tiering.hotSpend()).video) < 1e-12, `a dry run spends nothing: ${JSON.stringify(hv)}`);++ assert.ok(Math.abs(hv.projected - (spend1 + one)) < 1e-12, `the projected spend counts the dry run: ${JSON.stringify(hv)}`);++ // A zero ceiling closes the class even at a zero R2 price (where every copy would cost $0 and fit any ceiling).++ await policy.set({ active: true }, { actor: { type: 'service', id: 'live' }, reason: 'test: gate on' });++ const ct = require('../server/placement/cost-tiers');++ await ct.init();++ const prices = ct.settings();++ await ct.get().apply({ ...prices, r2: { ...prices.r2, standard: { ...prices.r2.standard, storagePerGbMonth: 0 }, infrequentAccess: { ...prices.r2.infrequentAccess, storagePerGbMonth: 0 } } }, { reason: 'test: free R2 storage' });++ await sp.set({ classes: { video: { maxHotUsdPerMonth: 0 } } }, { reason: 'test: a zero video ceiling' });++ assert.strictEqual(tiering.hotUsdPerMonth(4096), 0);++ s = await tiering.runSweep();++ await ct.get().apply(prices, { reason: 'test: R2 price back' });++ assert.ok(!await r2Row(gE[0].id) && !await r2Row(gE[1].id), 'a zero ceiling promotes nothing');++ d = await last(gE[0].id);++ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);++ assert.ok(/; refused: class video has maxHotUsdPerMonth \$0 \(media\.storage_policy\): it never promotes; the class promotes nothing more this sweep$/.test(d.reason), d.reason);++ assert.deepStrictEqual([s.constraints.refused.budget, s.constraints.hot_usd_per_month.video.ceiling], [1, 0], JSON.stringify(s.constraints));++ console.log('✅ ceiling: a dry run projects its spend without spending; a zero ceiling closes the class at any price');++ await sp.set({ classes: { video: { maxHotUsdPerMonth: 1e9, maxDemotionsPerSweep: 10 } } }, { reason: 'test: no video ceiling' });++ 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('✅ constraints: a provider failing its class (breaker or probe) refuses the move and keeps every copy; the monthly R2 storage ceiling closes a class in value order');++// ── 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));@@@ -65,6 +65,22 @@ const path = require('path');assert.deepStrictEqual(providers.eligibleClasses('r2'), ['online-hot'], 'an incomplete probe removes nothing');delete process.env.MEDIA_CAPABILITY_GATE;++ // The placement gate (F2.6) holds whatever the read gate's mode: a completed probe that found a capability the++ // class needs missing, or an open breaker, refuses a move; a probe not run or not completed decides nothing.++ const signals = require('../server/placement/signals');++ const gate = (name, cls) => { const g = providers.placementGate(name, cls); return [g.eligible, g.probe, g.breaker]; };++ assert.deepStrictEqual(gate('r2', 'online-hot'), [true, 'incomplete', 'closed'], 'an incomplete probe does not block placement');++ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: false }, lastChecked: Date.now(), lastError: null });++ assert.deepStrictEqual(providers.placementGate('r2', 'online-hot'), { provider: 'r2', class: 'online-hot', eligible: false, breaker: 'closed', probe: 'failed',++ missing: ['range_206'], reason: 'r2 is not eligible for online-hot: the capability probe found range_206 missing' });++ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: true }, lastChecked: Date.now(), lastError: null });++ assert.deepStrictEqual(gate('r2', 'online-hot'), [true, 'passed', 'closed']);++ signals.openBreaker('r2');++ assert.deepStrictEqual([...gate('r2', 'online-hot'), providers.placementGate('r2', 'online-hot').reason], [false, 'passed', 'open', 'r2 is not eligible for online-hot: its circuit breaker is open']);++ signals.closeBreaker('r2');++ assert.strictEqual(providers.placementGate('r2', 'online-canonical').eligible, false, 'a class the provider does not declare');++ assert.deepStrictEqual(gate('local', 'regional-cache'), [true, 'passed', 'closed'], 'the local disk is always eligible as the regional cache');++// isConfigured: false when env is empty, true for local.assert.strictEqual(providers.isConfigured('local'), true);assert.strictEqual(providers.isConfigured('b2'), false);@@@ -57,6 +57,13 @@ const path = require('path');assert.deepStrictEqual(policy.bandFor('video'), { promoteReadsPerHour: 120, demoteReadsPerHour: 20 });assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 }, 'the band keeps the budget');++ // The monthly R2 storage ceiling (F2.6): absent = none; a revision sets it per class.++ assert.strictEqual(policy.hotCeilingFor('video'), null, 'no ceiling by default');++ await policy.set({ classes: { video: { maxHotUsdPerMonth: 12.5 } } }, { reason: 'test: a video storage ceiling' });++ assert.deepStrictEqual([policy.hotCeilingFor('video'), policy.hotCeilingFor('image')], [12.5, null]);++ await assert.rejects(policy.set({ classes: { video: { maxHotUsdPerMonth: -1 } } }, { reason: 'test: a negative ceiling' }));++ assert.strictEqual(policy.hotCeilingFor('video'), 12.5, 'a refused ceiling changes nothing');++policy._reset();console.log('placement-storage-policy: all checks passed');process.exit(0);
@@ -35,7 +35,8 @@
@@ -35,7 +35,8 @@
"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"
+ "tiering on demand (F2.5): the object sweep promotes and demotes on the hour's Valkey reads against per-class hysteresis bands, one MGET per page, falling back to the daily PostgreSQL counts for a whole sweep when Valkey fails",
+ "placement constraints (F2.6): sweep moves refused while their provider fails its class (breaker or probe), and a per-class monthly R2 storage ceiling (maxHotUsdPerMonth) applied in value order, each explained in the decision's inputs"
],
],
"updated": "2026-10-03"
"updated": "2026-10-03"
}
}
@@ -146,6 +146,26 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
@@ -146,6 +146,26 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
decision's `inputs.demand` (`{ source, metric, value, threshold[, region, window_s] }`) and its reason say which source
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
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`).
`demand`, and the sweep result carries `demand_source` (and `demand_fallback`).
+- **Constraints (F2.6, shipped):** a sweep move needs the provider class it relies on: a promotion needs R2 eligible
+ for `online-hot`, the demotion of a ready object needs its canonical copy's provider eligible for its class
+ (`online-canonical`; the Media host's disk is always the `regional-cache`). Eligible means the circuit breaker is
+ not open and no completed capability probe found a capability the class needs missing
+ (`providers.placementGate()`, whatever `MEDIA_CAPABILITY_GATE` says; a probe not run or not completed decides
+ nothing). Otherwise the move is refused and every copy stays; a promotion's refusal spends the class's slot (every
+ promotion relies on R2, so a failing R2 logs at most a class budget of refusals per sweep), a demotion's does not
+ (its gate is that object's own canonical provider). Each class may carry a monthly ceiling on its R2
+ storage, `maxHotUsdPerMonth` in `media.storage_policy` (absent = none; `0` = the class never promotes, whatever
+ R2's configured price), priced at R2 Standard's list price in
+ `media.cost_tiers` over every present R2 copy of the class, gross of R2's free tier (`freeGb` is account-wide, not
+ per class, so the ceiling binds up to its value early), and kept current as the sweep moves: the first
+ promotion that would pass it is refused and closes the class's promotions for that sweep (a cheaper, lower-ranked
+ object never jumps the value-per-dollar queue). Every decision records `inputs.provider_gate` (`{ provider, class,
+ eligible, breaker, probe[, missing, reason] }`) and a sweep promotion `inputs.budget` (`{ class, usd_per_month,
+ copy_usd_per_month, ceiling }`); the sweep result carries `constraints` (`promote_gate`, `refused.{provider,budget}`,
+ `hot_usd_per_month.<class>.{spend,projected,ceiling}` for every class, `ceiling` null when unset). `spend` follows
+ only the moves made; `projected` also counts the dry runs, and the ceiling (and `inputs.budget.usd_per_month`) is held
+ against it, so dry runs (gate off) are refused as a live sweep would be, without changing `spend`. Native objects classify only as `video`, `image` or `download` today, so a ceiling (or budget)
+ set on `game-asset`, `attachment` or `backup` decides nothing yet.
## 7. Delivery: sticky, measured, canaried
## 7. Delivery: sticky, measured, canaried
@@ -207,7 +227,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
@@ -207,7 +227,9 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
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 (F2.4, shipped) and the sweep's
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).
+ eligibility on them with per-class hysteresis and a PostgreSQL fallback (F2.5, shipped), the provider-class gate
+ and per-class monthly R2 storage ceilings on its moves (F2.6, shipped). Still open in F2: the move cleanup job and
+ its alert after three failures (§6), every class (not only native objects ↔ R2) through the one sweep.
- **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
@@ -16,6 +16,13 @@
@@ -16,6 +16,13 @@
* second it may demote, in between nothing moves. Without Valkey, or when a read fails or times out, the whole sweep
* 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
* 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.
* media_sweep_demand_source_total{source}. Each decision's inputs.demand says which source and value drove it.
+ * Constraints (F2.6): a promotion needs R2 eligible for the online-hot class, and the demotion of a ready object needs
+ * its canonical copy's provider eligible for its class (../placement/providers.js placementGate(): breaker not open,
+ * no failed capability), else it is refused and the copies stay where they are; each decision's inputs.provider_gate
+ * records the gate it saw. A class with maxHotUsdPerMonth (media.storage_policy) promotes only while its R2 copies'
+ * list-price storage (media.cost_tiers) stays within it: the first promotion that would pass it is refused
+ * (inputs.budget) and closes the class's promotions for the sweep, so a cheaper, lower-ranked object never jumps
+ * the value-per-dollar queue.
*
*
* 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
@@ -47,6 +54,7 @@ const policy = require('./tier-policy');
@@ -47,6 +54,7 @@ const policy = require('./tier-policy');
const storagePolicy = require('../placement/storage-policy');
const storagePolicy = require('../placement/storage-policy');
const costTiers = require('../placement/cost-tiers');
const costTiers = require('../placement/cost-tiers');
const value = require('../placement/value');
const value = require('../placement/value');
+const providers = require('../placement/providers');
const popularity = require('./popularity');
const popularity = require('./popularity');
const MB = 1024 * 1024;
const MB = 1024 * 1024;
@@ -128,6 +136,28 @@ function r2KeyFor(obj, canon) {
@@ -128,6 +136,28 @@ function r2KeyFor(obj, canon) {
return canon && canon.provider === 'b2' ? canon.key : `objects/${obj.app_id}/${obj.id}`;
return canon && canon.provider === 'b2' ? canon.key : `objects/${obj.app_id}/${obj.id}`;
}
}
+// ── Provider classes and the storage ceiling (F2.6) ──────────
+
+/** The class a canonical copy serves: the Media host's disk is the regional cache, a remote provider the online canonical. */
+const canonicalClass = (provider) => (provider === 'local' ? 'regional-cache' : 'online-canonical');
+
+/** R2 Standard's list price of keeping `bytes` for a month (media.cost_tiers). */
+function hotUsdPerMonth(bytes) {
+ return (Number(bytes) || 0) / value.GB * (Number(costTiers.settings().r2.standard.storagePerGbMonth) || 0);
+}
+
+const fmtUsd = (v) => `$${Number(Number(v).toPrecision(4))}`;
+
+/** What each class's present R2 copies cost a month at list price, gross of R2's account-wide free tier: { video: 0.12, image: 0, ... }. */
+async function hotSpend() {
+ const out = Object.fromEntries(storagePolicy.CLASSES.map((c) => [c, 0]));
+ const rows = await db.all(`SELECT o.kind, o.mime_type, SUM(COALESCE(l.size_bytes, o.size_bytes, 0))::bigint AS bytes
+ FROM media_locations l JOIN media_objects o ON o.id = l.object_id
+ WHERE l.provider = 'r2' AND l.state = 'present' GROUP BY o.kind, o.mime_type`);
+ for (const r of rows) out[storagePolicy.classOf(r)] += hotUsdPerMonth(r.bytes);
+ return out;
+}
+
// ── Decisions ────────────────────────────────────────────────
// ── Decisions ────────────────────────────────────────────────
function thresholdSnapshot(settings) {
function thresholdSnapshot(settings) {
@@ -194,7 +224,12 @@ async function frame(objectId, action, ctx) {
@@ -194,7 +224,12 @@ async function frame(objectId, action, ctx) {
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;
if (ctx.demand) inputs.demand = ctx.demand;
+ if (ctx.budget) inputs.budget = ctx.budget;
const canon = canonicalLocation(obj, locs);
const canon = canonicalLocation(obj, locs);
+ // The provider class the move relies on: R2 as online-hot for a promotion, the canonical copy's provider for a demotion.
+ const gate = action === 'promote' ? providers.placementGate('r2', 'online-hot')
+ : canon && canon.provider !== 'r2' ? providers.placementGate(canon.provider, canonicalClass(canon.provider)) : null;
+ if (gate) inputs.provider_gate = gate;
const trigger = ctx.trigger || 'manual';
const trigger = ctx.trigger || 'manual';
const why = ctx.reason || 'requested';
const why = ctx.reason || 'requested';
const from = action === 'promote' ? (canon ? canon.provider : null) : 'r2';
const from = action === 'promote' ? (canon ? canon.provider : null) : 'r2';
@@ -204,7 +239,7 @@ async function frame(objectId, action, ctx) {
@@ -204,7 +239,7 @@ async function frame(objectId, action, ctx) {
if (!repeat) await recordDecision({ obj, action, from, to, outcome, trigger, reason, inputs, settings, error: extra.error });
if (!repeat) await recordDecision({ obj, action, from, to, outcome, trigger, reason, inputs, settings, error: extra.error });
return { ok: outcome === 'done' || outcome === 'already', outcome, object_id: obj.id, repeat, ...extra };
return { ok: outcome === 'done' || outcome === 'already', outcome, object_id: obj.id, repeat, ...extra };
};
};
- return { obj, locs, canon, held, settings, why, log };
+ return { obj, locs, canon, held, settings, why, log, gate };
}
}
/** Why an object may not be promoted at all, from the record (no I/O), or null. */
/** Why an object may not be promoted at all, from the record (no I/O), or null. */
@@ -225,13 +260,15 @@ async function promoteRefusal(obj, { locs, held }) {
@@ -225,13 +260,15 @@ async function promoteRefusal(obj, { locs, held }) {
async function promote(objectId, ctx = {}) {
async function promote(objectId, ctx = {}) {
const f = await frame(objectId, 'promote', ctx);
const f = await frame(objectId, 'promote', ctx);
if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };
if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };
- const { obj, locs, canon, held, settings, why, log } = f;
+ const { obj, locs, canon, held, settings, why, log, gate } = f;
if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });
if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });
if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });
if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });
const r2Now = locs.find((l) => l.provider === 'r2');
const r2Now = locs.find((l) => l.provider === 'r2');
if (r2Now && r2Now.state === 'present' && r2Now.verified_at) return await log('already', `${why}; the R2 copy is already present`);
if (r2Now && r2Now.state === 'present' && r2Now.verified_at) return await log('already', `${why}; the R2 copy is already present`);
const refusal = await promoteRefusal(obj, { locs, held });
const refusal = await promoteRefusal(obj, { locs, held });
if (refusal) return await log('refused', `${why}; refused: ${refusal}`, { error: refusal });
if (refusal) return await log('refused', `${why}; refused: ${refusal}`, { error: refusal });
+ if (!gate.eligible) return await log('refused', `${why}; refused: ${gate.reason} (provider class gate)`, { error: gate.reason, blocked: 'provider' });
+ if (ctx.overBudget) return await log('refused', `${why}; refused: ${ctx.overBudget}`, { error: ctx.overBudget, blocked: 'budget' });
const vs = vodStorage();
const vs = vodStorage();
if (!settings.active) return await log('dry_run', `gate off (active = false): would promote: ${why}${vs.providerAvailable('r2') ? '' : ' (R2 is not available now)'}`);
if (!settings.active) return await log('dry_run', `gate off (active = false): would promote: ${why}${vs.providerAvailable('r2') ? '' : ' (R2 is not available now)'}`);
if (!vs.providerAvailable('r2')) return await log('refused', `${why}; refused: R2 is not available`, { error: 'R2 not available' });
if (!vs.providerAvailable('r2')) return await log('refused', `${why}; refused: R2 is not available`, { error: 'R2 not available' });
@@ -290,11 +327,15 @@ async function promote(objectId, ctx = {}) {
@@ -290,11 +327,15 @@ async function promote(objectId, ctx = {}) {
async function demote(objectId, ctx = {}) {
async function demote(objectId, ctx = {}) {
const f = await frame(objectId, 'demote', ctx);
const f = await frame(objectId, 'demote', ctx);
if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };
if (f.missing) return { ok: false, outcome: 'refused', error: 'object not found' };
- const { obj, locs, canon, held, settings, why, log } = f;
+ const { obj, locs, canon, held, settings, why, log, gate } = f;
if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });
if (ctx.residency) return await log('refused', `${why}; refused: ${ctx.residency}`, { error: ctx.residency });
if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });
if (held) return await log('refused', `${why}; refused: under a retention hold (a hold freezes placement)`, { error: 'held', held: true });
const r2 = locs.find((l) => l.provider === 'r2');
const r2 = locs.find((l) => l.provider === 'r2');
if (!r2) return await log('already', `${why}; there is no R2 copy`);
if (!r2) return await log('already', `${why}; there is no R2 copy`);
+ // Reads of a ready object fall back to its canonical copy: never shift them onto a provider that is failing.
+ if (obj.lifecycle_status === 'ready' && gate && !gate.eligible) {
+ return await log('refused', `${why}; refused: ${gate.reason} (provider class gate); the R2 copy is kept`, { error: gate.reason, blocked: 'provider' });
+ }
const vs = vodStorage();
const vs = vodStorage();
if (!settings.active) return await log('dry_run', `gate off (active = false): would demote: ${why}`);
if (!settings.active) return await log('dry_run', `gate off (active = false): would demote: ${why}`);
if (!vs.providerConfigured('r2')) return await log('refused', `${why}; refused: R2 is not configured`, { error: 'R2 not configured' });
if (!vs.providerConfigured('r2')) return await log('refused', `${why}; refused: R2 is not configured`, { error: 'R2 not configured' });
@@ -446,7 +487,7 @@ async function hotPromotionCandidates(settings, demand, { appId = null, limit =
@@ -446,7 +487,7 @@ async function hotPromotionCandidates(settings, demand, { appId = null, limit =
async function demotionCandidates(settings, { now = Date.now(), appId = null, demand = 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, l.state AS r2_state, l.size_bytes AS r2_size_bytes,
(SELECT MAX(d.day) FROM media_object_views_daily d WHERE d.object_id = o.id AND d.unique_viewers > 0) AS last_viewed_day,
(SELECT MAX(d.day) FROM media_object_views_daily d WHERE d.object_id = o.id AND d.unique_viewers > 0) AS last_viewed_day,
(SELECT COALESCE(SUM(d.unique_viewers), 0)::bigint FROM media_object_views_daily d WHERE d.object_id = o.id AND d.day >= ?) AS unique_viewers_30d
(SELECT COALESCE(SUM(d.unique_viewers), 0)::bigint FROM media_object_views_daily d WHERE d.object_id = o.id AND d.day >= ?) AS unique_viewers_30d
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
@@ -563,7 +604,8 @@ async function sweepCandidates(settings, now) {
@@ -563,7 +604,8 @@ async function sweepCandidates(settings, now) {
}
}
/**
/**
- * One pass over native objects. → { gate, demand_source, demand_fallback?, promoted, demoted, would_promote, would_demote, already, refused, failed,
+ * One pass over native objects. → { gate, demand_source, demand_fallback?, constraints: { promote_gate, refused: { provider, budget },
+ * hot_usd_per_month: { <class>: { spend, projected, ceiling } } }, 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() } = {}) {
@@ -590,6 +632,18 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -590,6 +632,18 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
// 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: {} };
+ // A slot is spent by a move, a failed move, a dry run, or a promotion's provider-gate refusal, repeated or not
+ // (every promotion relies on R2, so the next in line would be refused for the same reason and a closed R2 costs
+ // the class at most its budget of rows). A demotion's gate is the candidate's own canonical provider, so its
+ // refusal spends nothing and an object with a healthy canonical copy takes the slot.
+ const spends = (r, action) => (action === 'promote' && r.blocked === 'provider') || (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome));
+ const blocked = (r) => { if (r.blocked) out.constraints.refused[r.blocked]++; };
+ // The storage ceiling (F2.6): each class's R2 copies at list price. `spend` follows the moves made; `projected`
+ // also counts dry runs (the moves the sweep would make), and the ceiling is held against it.
+ const spend = await hotSpend();
+ const projected = { ...spend };
+ const closed = new Set();
+ out.constraints = { promote_gate: providers.placementGate('r2', 'online-hot'), refused: { provider: 0, budget: 0 }, hot_usd_per_month: {} };
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, demand } 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;
@@ -599,7 +653,14 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -599,7 +653,14 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
const r = await demote(row.id, { trigger, settings, now, residency, valuePerDollar: score, demand,
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;
+ blocked(r);
+ if (spends(r, 'demote')) spent.demote[cls] = (spent.demote[cls] || 0) + 1;
+ // hotSpend() counted only present copies: a corrupt or pending one leaves the class's spend as it was.
+ if (!r.repeat && ['done', 'dry_run'].includes(r.outcome) && row.r2_state === 'present') {
+ const freed = hotUsdPerMonth(row.r2_size_bytes ?? row.size_bytes);
+ projected[cls] = Math.max(0, projected[cls] - freed);
+ if (r.outcome === 'done') spend[cls] = Math.max(0, spend[cls] - freed);
+ }
}
}
// 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.
@@ -607,13 +668,30 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -607,13 +668,30 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
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;
- if (!room('promote', cls)) continue;
+ if (!room('promote', cls) || closed.has(cls)) continue;
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 cost = hotUsdPerMonth(c.size_bytes);
+ const ceiling = storagePolicy.hotCeilingFor(cls);
+ const usd = projected[cls];
+ const budget = { class: cls, usd_per_month: usd, copy_usd_per_month: cost, ceiling };
+ // A zero ceiling closes the class even when R2's configured price is zero (every copy would then cost $0).
+ const overBudget = ceiling === 0
+ ? `class ${cls} has maxHotUsdPerMonth $0 (media.storage_policy): it never promotes; the class promotes nothing more this sweep`
+ : ceiling != null && usd + cost > ceiling
+ ? `class ${cls} R2 storage would reach ${fmtUsd(usd + cost)}/month (${fmtUsd(usd)} now + ${fmtUsd(cost)} for this copy) > maxHotUsdPerMonth ${fmtUsd(ceiling)} (media.storage_policy); the class promotes nothing more this sweep`
+ : null;
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,
- demand: c.demand, pop: picked.source === 'pg' ? { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } : undefined });
+ demand: c.demand, budget, overBudget, 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;
+ blocked(r);
+ if (r.blocked === 'budget') closed.add(cls);
+ if (spends(r, 'promote')) spent.promote[cls] = (spent.promote[cls] || 0) + 1;
+ if (!r.repeat && ['done', 'dry_run'].includes(r.outcome)) projected[cls] += cost;
+ if (!r.repeat && r.outcome === 'done') spend[cls] += cost;
+ }
+ for (const cls of storagePolicy.CLASSES) {
+ out.constraints.hot_usd_per_month[cls] = { spend: spend[cls], projected: projected[cls], ceiling: storagePolicy.hotCeilingFor(cls) };
}
}
// Demand (F2.4): stage media.object.hot for objects over the hourly read threshold.
// Demand (F2.4): stage media.object.hot for objects over the hourly read threshold.
@@ -698,7 +776,7 @@ async function report({ appId = null, limit = 20 } = {}) {
@@ -698,7 +776,7 @@ async function report({ appId = null, limit = 20 } = {}) {
}
}
module.exports = {
module.exports = {
- FAILED_BACKOFF_H, OUTCOMES,
+ FAILED_BACKOFF_H, OUTCOMES, hotSpend, hotUsdPerMonth,
runSweep, promote, demote, candidates, promotionCandidates, demotionCandidates,
runSweep, promote, demote, candidates, promotionCandidates, demotionCandidates,
canonicalUnverified, confirmCanonical, r2KeyFor,
canonicalUnverified, confirmCanonical, r2KeyFor,
counts24h, listDecisions, decisionPublic, report,
counts24h, listDecisions, decisionPublic, report,
@@ -15,6 +15,8 @@
@@ -15,6 +15,8 @@
* probeAll() probe every configured remote provider
* probeAll() probe every configured remote provider
* startHealthLoop() the cheap per-minute HeadBucket and the hourly capability probe
* startHealthLoop() the cheap per-minute HeadBucket and the hourly capability probe
* healthStatus() { b2, r2 } each as { configured, healthy, lastCheck, lastError }
* healthStatus() { b2, r2 } each as { configured, healthy, lastCheck, lastError }
+ * placementGate(name, cls) may a placement move put bytes on (or shift reads to) it for the class:
+ * breaker not open, no failed capability the class needs (F2.6)
*
*
* A provider that fails a capability is removed from the classes that need it until it passes
* A provider that fails a capability is removed from the classes that need it until it passes
* again — `probeProvider(name)` returning a capability list with `passed=false` is the only
* again — `probeProvider(name)` returning a capability list with `passed=false` is the only
@@ -92,6 +94,28 @@ function eligibleClasses(name) {
@@ -92,6 +94,28 @@ function eligibleClasses(name) {
return measured;
return measured;
}
}
+/**
+ * Whether placement may put bytes on (or shift reads to) a provider for a class (docs/media-fabric.md §2, F2.6):
+ * the provider declares the class, its circuit breaker is not open, and no completed capability probe found a
+ * capability the class needs missing. A probe not run yet, or one that could not complete, decides nothing (as for
+ * eligibleClasses(): live health is the breaker's job). This
+ * holds whatever MEDIA_CAPABILITY_GATE says: the read gate protects playback, and a placement move refused here
+ * leaves every copy where it is. Whether the provider is configured is the caller's check.
+ * → { provider, class, eligible, breaker, probe: 'pending'|'incomplete'|'passed'|'failed', missing?, reason? }
+ */
+function placementGate(name, cls) {
+ const out = { provider: name, class: cls, eligible: true, breaker: 'closed', probe: 'passed' };
+ if (!providerClasses(name).includes(cls)) return { ...out, eligible: false, reason: `${name} does not serve ${cls}` };
+ if (name === 'local') return out;
+ out.breaker = require('./signals').breakerState(name);
+ const caps = capabilities(name);
+ const missing = caps.passed ? CLASS_REQUIREMENTS[cls].filter((c) => !(caps.capabilities || {})[c]) : [];
+ out.probe = caps.passed === null ? 'pending' : !caps.passed ? 'incomplete' : missing.length ? 'failed' : 'passed';
+ if (out.breaker === 'open') return { ...out, eligible: false, reason: `${name} is not eligible for ${cls}: its circuit breaker is open` };
+ if (out.probe === 'failed') return { ...out, eligible: false, missing, reason: `${name} is not eligible for ${cls}: the capability probe found ${missing.join(', ')} missing` };
+ return out;
+}
+
/** What classes each provider says it can serve. These are the design's assignments (docs/media-fabric.md §2):
/** What classes each provider says it can serve. These are the design's assignments (docs/media-fabric.md §2):
* B2 is the online canonical, R2 Standard the online hot tier, the Media host's NVMe the regional cache. */
* B2 is the online canonical, R2 Standard the online hot tier, the Media host's NVMe the regional cache. */
function providerClasses(name) {
function providerClasses(name) {
@@ -315,7 +339,7 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};
@@ -315,7 +339,7 @@ function reset() { stop(); state.capabilities = {}; state.lastHealthCheck = {};
module.exports = {
module.exports = {
CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,
CLASSES, CLASS_REQUIREMENTS, PROBE_PREFIX, PROBE_KEY,
- isConfigured, providerClasses, eligibleClasses, measuredClasses, gateMode, capabilities,
+ isConfigured, providerClasses, eligibleClasses, measuredClasses, placementGate, gateMode, capabilities,
probe, probeAll, liveHealth, healthStatus,
probe, probeAll, liveHealth, healthStatus,
startHealthLoop, stop, reset,
startHealthLoop, stop, reset,
// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
// Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
@@ -31,6 +31,9 @@ const CLASS_SCHEMA = {
@@ -31,6 +31,9 @@ const CLASS_SCHEMA = {
minResidencyMs: { type: 'integer', minimum: 0 },
minResidencyMs: { type: 'integer', minimum: 0 },
promoteReadsPerHour: { type: 'integer', minimum: 1, maximum: 1e9 },
promoteReadsPerHour: { type: 'integer', minimum: 1, maximum: 1e9 },
demoteReadsPerHour: { type: 'integer', minimum: 0, maximum: 1e9 },
demoteReadsPerHour: { type: 'integer', minimum: 0, maximum: 1e9 },
+ // F2.6: the class's R2 Standard storage spend ceiling, in dollars a month at media.cost_tiers' list price
+ // (every present R2 copy of the class counts). Absent = no ceiling; 0 = the class never promotes, even at a $0 price.
+ maxHotUsdPerMonth: { type: 'number', minimum: 0, maximum: 1e9 },
},
},
};
};
const SCHEMA = {
const SCHEMA = {
@@ -117,6 +120,12 @@ function budgetFor(name) {
@@ -117,6 +120,12 @@ function budgetFor(name) {
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
}
}
+/** The class's monthly R2 storage ceiling in dollars (maxHotUsdPerMonth), or null when it has none. */
+function hotCeilingFor(name) {
+ const { maxHotUsdPerMonth } = classPolicy(name);
+ return maxHotUsdPerMonth == null ? null : maxHotUsdPerMonth;
+}
+
/** The class's hourly promote and demote thresholds (hysteresis: demote < promote). */
/** The class's hourly promote and demote thresholds (hysteresis: demote < promote). */
function bandFor(name) {
function bandFor(name) {
const { promoteReadsPerHour, demoteReadsPerHour } = classPolicy(name);
const { promoteReadsPerHour, demoteReadsPerHour } = classPolicy(name);
@@ -139,4 +148,4 @@ function mayMove({ class: name, lastMovedAt, now = Date.now() }) {
@@ -139,4 +148,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, bandFor, mayMove, _reset: () => { store = null; } };
+module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, hotCeilingFor, bandFor, mayMove, _reset: () => { store = null; } };
@@ -640,6 +640,114 @@ const crypto = require('crypto');
@@ -640,6 +640,114 @@ const crypto = require('crypto');
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]);
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');
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');
+ // ── 9f. Constraints (F2.6): the provider class gate and each class's monthly R2 storage ceiling ──
+ const providersMod = require('../server/placement/providers');
+ const signals = require('../server/placement/signals');
+ await db.run('DELETE FROM media_object_views_daily');
+ await sp.set({ classes: { video: { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 500 } } }, { reason: 'test: video budget 1' });
+ // R2's breaker is open: the best video is refused (its slot spent, so the next waits) and nothing moves.
+ const gA = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];
+ await setViews(gA[0].id, [[0, 900]]); await setViews(gA[1].id, [[0, 800]]);
+ signals.openBreaker('r2');
+ s = await tiering.runSweep();
+ signals.closeBreaker('r2');
+ assert.ok(!await r2Row(gA[0].id) && !await r2Row(gA[1].id), 'no promotion while R2 is not eligible for online-hot');
+ d = await last(gA[0].id);
+ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);
+ assert.ok(/; refused: r2 is not eligible for online-hot: its circuit breaker is open \(provider class gate\)$/.test(d.reason), d.reason);
+ assert.deepStrictEqual((({ provider, class: c, eligible, breaker }) => [provider, c, eligible, breaker])(JSON.parse(d.inputs).provider_gate), ['r2', 'online-hot', false, 'open']);
+ assert.strictEqual((await decisions(gA[1].id)).length, 0, 'the refusal spent the class slot');
+ assert.deepStrictEqual([s.constraints.promote_gate.eligible, s.constraints.refused.provider >= 1], [false, true], JSON.stringify(s.constraints));
+ // A completed probe found a capability online-hot needs missing: refused the same way, naming it.
+ for (const o of gA) await db.run('DELETE FROM media_object_views_daily WHERE object_id = ?', [o.id]);
+ const gB = await addObject({ mime: 'video/mp4' }); await setViews(gB.id, [[0, 900]]);
+ providersMod._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: false }, lastChecked: Date.now(), lastError: null });
+ s = await tiering.runSweep();
+ providersMod._setCapabilities('r2', { provider: 'r2', passed: null, capabilities: {}, lastChecked: null, lastError: null });
+ d = await last(gB.id);
+ assert.ok(!await r2Row(gB.id) && d.outcome === 'refused' && /r2 is not eligible for online-hot: the capability probe found range_206 missing/.test(d.reason), d.reason);
+ assert.deepStrictEqual([JSON.parse(d.inputs).provider_gate.probe, JSON.parse(d.inputs).provider_gate.missing], ['failed', ['range_206']]);
+ await db.run('DELETE FROM media_object_views_daily WHERE object_id = ?', [gB.id]);
+ // A ready object's R2 copy stays while its canonical copy's provider (B2, online-canonical) is failing. The
+ // refusal spends no demotion slot (the gate is per object): the smaller idle copy over a local canonical leaves.
+ await sp.set({ classes: { video: { maxDemotionsPerSweep: 1 } } }, { reason: 'test: video demotion budget 1' });
+ const gC = await addObject({ mime: 'video/mp4', canonical: 'b2' }); await inR2(gC, 30);
+ const gL = await addObject({ mime: 'video/mp4', size: 2048 }); await inR2(gL, 30);
+ signals.openBreaker('b2');
+ s = await tiering.runSweep();
+ signals.closeBreaker('b2');
+ await sp.set({ classes: { video: { maxDemotionsPerSweep: 500 } } }, { reason: 'test: video demotion budget back' });
+ assert.ok(!await r2Row(gL.id), 'a demotion refused by its own canonical provider leaves the slot to the next in line');
+ d = await last(gC.id);
+ assert.ok(await r2Row(gC.id), 'the R2 copy is kept while B2 is not eligible');
+ assert.deepStrictEqual([d.action, d.outcome], ['demote', 'refused']);
+ assert.ok(/; refused: b2 is not eligible for online-canonical: its circuit breaker is open \(provider class gate\); the R2 copy is kept$/.test(d.reason), d.reason);
+ assert.deepStrictEqual([JSON.parse(d.inputs).provider_gate.provider, JSON.parse(d.inputs).provider_gate.eligible], ['b2', false]);
+ // A corrupt R2 copy was never counted in the class's spend, so demoting it subtracts nothing.
+ const gX = await addObject({ mime: 'video/mp4' }); await inR2(gX, 30);
+ await db.run("UPDATE media_locations SET state = 'corrupt' WHERE object_id = ? AND provider = 'r2'", [gX.id]);
+ s = await tiering.runSweep();
+ assert.ok(!await r2Row(gC.id), 'B2 eligible again: the idle copy leaves');
+ assert.ok(!await r2Row(gX.id), 'the corrupt copy leaves');
+ assert.ok(Math.abs(s.constraints.hot_usd_per_month.video.spend - (await tiering.hotSpend()).video) < 1e-12, JSON.stringify(s.constraints));
+ assert.deepStrictEqual(Object.keys(s.constraints.hot_usd_per_month).sort(), [...sp.CLASSES].sort(), 'every class is reported');
+ assert.deepStrictEqual(JSON.parse((await last(gC.id)).inputs).provider_gate, { provider: 'b2', class: 'online-canonical', eligible: true, breaker: 'closed', probe: 'pending' });
+ // The ceiling: room for one more 4 KB video copy; the first in line takes it, the next is refused and closes the class.
+ const spend0 = (await tiering.hotSpend()).video;
+ const one = tiering.hotUsdPerMonth(4096);
+ assert.ok(one > 0 && Math.abs(one - 4096 / 1073741824 * 0.015) < 1e-15, `R2 Standard list price of 4 KB a month: ${one}`);
+ const ceiling = spend0 + one * 1.5;
+ await sp.set({ classes: { video: { maxPromotionsPerSweep: 3, maxHotUsdPerMonth: ceiling } } }, { reason: 'test: a video storage ceiling' });
+ const gD = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];
+ await setViews(gD[0].id, [[0, 900]]); await setViews(gD[1].id, [[0, 800]]); await setViews(gD[2].id, [[0, 700]]);
+ s = await tiering.runSweep();
+ assert.deepStrictEqual([!!await r2Row(gD[0].id), !!await r2Row(gD[1].id), !!await r2Row(gD[2].id)], [true, false, false], 'one copy fits under the ceiling');
+ assert.deepStrictEqual(JSON.parse((await last(gD[0].id)).inputs).budget, { class: 'video', usd_per_month: spend0, copy_usd_per_month: one, ceiling });
+ d = await last(gD[1].id);
+ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);
+ assert.ok(/; refused: class video R2 storage would reach \$[0-9.e-]+\/month \(\$[0-9.e-]+ now \+ \$[0-9.e-]+ for this copy\) > maxHotUsdPerMonth \$[0-9.e-]+ \(media\.storage_policy\); the class promotes nothing more this sweep$/.test(d.reason), d.reason);
+ assert.strictEqual((await decisions(gD[2].id)).length, 0, 'a cheaper lower-ranked object does not jump the queue: the class is closed');
+ assert.strictEqual(s.constraints.refused.budget, 1);
+ assert.ok(Math.abs(s.constraints.hot_usd_per_month.video.spend - (spend0 + one)) < 1e-12 && s.constraints.hot_usd_per_month.video.ceiling === ceiling, JSON.stringify(s.constraints));
+ assert.strictEqual(s.constraints.hot_usd_per_month.video.projected, s.constraints.hot_usd_per_month.video.spend, 'with the gate on, projected is actual');
+ // Gate off: a dry run holds the ceiling against the projected spend (what it would move) and leaves the actual spend alone.
+ await policy.set({ active: false }, { actor: { type: 'service', id: 'live' }, reason: 'test: gate off under a ceiling' });
+ const spend1 = (await tiering.hotSpend()).video;
+ await sp.set({ classes: { video: { maxHotUsdPerMonth: spend1 + one * 1.5 } } }, { reason: 'test: a video storage ceiling, dry run' });
+ const gE = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];
+ await setViews(gE[0].id, [[0, 990]]); await setViews(gE[1].id, [[0, 980]]); await setViews(gE[2].id, [[0, 970]]);
+ s = await tiering.runSweep();
+ assert.strictEqual(s.would_demote, 0, JSON.stringify(s));
+ assert.ok(!await r2Row(gE[0].id) && !await r2Row(gE[1].id), 'a dry run moves nothing');
+ const e0 = await last(gE[0].id), e1 = await last(gE[1].id);
+ assert.deepStrictEqual([e0.outcome, e1.outcome], ['dry_run', 'refused']);
+ assert.ok(/maxHotUsdPerMonth/.test(e1.reason), e1.reason);
+ assert.ok(Math.abs(JSON.parse(e1.inputs).budget.usd_per_month - (JSON.parse(e0.inputs).budget.usd_per_month + one)) < 1e-12, e1.inputs);
+ assert.strictEqual((await decisions(gE[2].id)).length, 0, 'the projected spend closed the class');
+ const hv = s.constraints.hot_usd_per_month.video;
+ assert.ok(Math.abs(hv.spend - spend1) < 1e-12 && Math.abs(hv.spend - (await tiering.hotSpend()).video) < 1e-12, `a dry run spends nothing: ${JSON.stringify(hv)}`);
+ assert.ok(Math.abs(hv.projected - (spend1 + one)) < 1e-12, `the projected spend counts the dry run: ${JSON.stringify(hv)}`);
+ // A zero ceiling closes the class even at a zero R2 price (where every copy would cost $0 and fit any ceiling).
+ await policy.set({ active: true }, { actor: { type: 'service', id: 'live' }, reason: 'test: gate on' });
+ const ct = require('../server/placement/cost-tiers');
+ await ct.init();
+ const prices = ct.settings();
+ await ct.get().apply({ ...prices, r2: { ...prices.r2, standard: { ...prices.r2.standard, storagePerGbMonth: 0 }, infrequentAccess: { ...prices.r2.infrequentAccess, storagePerGbMonth: 0 } } }, { reason: 'test: free R2 storage' });
+ await sp.set({ classes: { video: { maxHotUsdPerMonth: 0 } } }, { reason: 'test: a zero video ceiling' });
+ assert.strictEqual(tiering.hotUsdPerMonth(4096), 0);
+ s = await tiering.runSweep();
+ await ct.get().apply(prices, { reason: 'test: R2 price back' });
+ assert.ok(!await r2Row(gE[0].id) && !await r2Row(gE[1].id), 'a zero ceiling promotes nothing');
+ d = await last(gE[0].id);
+ assert.deepStrictEqual([d.action, d.outcome], ['promote', 'refused']);
+ assert.ok(/; refused: class video has maxHotUsdPerMonth \$0 \(media\.storage_policy\): it never promotes; the class promotes nothing more this sweep$/.test(d.reason), d.reason);
+ assert.deepStrictEqual([s.constraints.refused.budget, s.constraints.hot_usd_per_month.video.ceiling], [1, 0], JSON.stringify(s.constraints));
+ console.log('✅ ceiling: a dry run projects its spend without spending; a zero ceiling closes the class at any price');
+ await sp.set({ classes: { video: { maxHotUsdPerMonth: 1e9, maxDemotionsPerSweep: 10 } } }, { reason: 'test: no video ceiling' });
+ 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('✅ constraints: a provider failing its class (breaker or probe) refuses the move and keeps every copy; the monthly R2 storage ceiling closes a class in value order');
+
// ── 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));
@@ -65,6 +65,22 @@ const path = require('path');
@@ -65,6 +65,22 @@ const path = require('path');
assert.deepStrictEqual(providers.eligibleClasses('r2'), ['online-hot'], 'an incomplete probe removes nothing');
assert.deepStrictEqual(providers.eligibleClasses('r2'), ['online-hot'], 'an incomplete probe removes nothing');
delete process.env.MEDIA_CAPABILITY_GATE;
delete process.env.MEDIA_CAPABILITY_GATE;
+ // The placement gate (F2.6) holds whatever the read gate's mode: a completed probe that found a capability the
+ // class needs missing, or an open breaker, refuses a move; a probe not run or not completed decides nothing.
+ const signals = require('../server/placement/signals');
+ const gate = (name, cls) => { const g = providers.placementGate(name, cls); return [g.eligible, g.probe, g.breaker]; };
+ assert.deepStrictEqual(gate('r2', 'online-hot'), [true, 'incomplete', 'closed'], 'an incomplete probe does not block placement');
+ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: false }, lastChecked: Date.now(), lastError: null });
+ assert.deepStrictEqual(providers.placementGate('r2', 'online-hot'), { provider: 'r2', class: 'online-hot', eligible: false, breaker: 'closed', probe: 'failed',
+ missing: ['range_206'], reason: 'r2 is not eligible for online-hot: the capability probe found range_206 missing' });
+ providers._setCapabilities('r2', { provider: 'r2', passed: true, capabilities: { range: true, range_206: true }, lastChecked: Date.now(), lastError: null });
+ assert.deepStrictEqual(gate('r2', 'online-hot'), [true, 'passed', 'closed']);
+ signals.openBreaker('r2');
+ assert.deepStrictEqual([...gate('r2', 'online-hot'), providers.placementGate('r2', 'online-hot').reason], [false, 'passed', 'open', 'r2 is not eligible for online-hot: its circuit breaker is open']);
+ signals.closeBreaker('r2');
+ assert.strictEqual(providers.placementGate('r2', 'online-canonical').eligible, false, 'a class the provider does not declare');
+ assert.deepStrictEqual(gate('local', 'regional-cache'), [true, 'passed', 'closed'], 'the local disk is always eligible as the regional cache');
+
// isConfigured: false when env is empty, true for local.
// isConfigured: false when env is empty, true for local.
assert.strictEqual(providers.isConfigured('local'), true);
assert.strictEqual(providers.isConfigured('local'), true);
assert.strictEqual(providers.isConfigured('b2'), false);
assert.strictEqual(providers.isConfigured('b2'), false);
@@ -57,6 +57,13 @@ const path = require('path');
@@ -57,6 +57,13 @@ const path = require('path');
assert.deepStrictEqual(policy.bandFor('video'), { promoteReadsPerHour: 120, demoteReadsPerHour: 20 });
assert.deepStrictEqual(policy.bandFor('video'), { promoteReadsPerHour: 120, demoteReadsPerHour: 20 });
assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 }, 'the band keeps the budget');
assert.deepStrictEqual(policy.budgetFor('video'), { maxPromotionsPerSweep: 1, maxDemotionsPerSweep: 10 }, 'the band keeps the budget');
+ // The monthly R2 storage ceiling (F2.6): absent = none; a revision sets it per class.
+ assert.strictEqual(policy.hotCeilingFor('video'), null, 'no ceiling by default');
+ await policy.set({ classes: { video: { maxHotUsdPerMonth: 12.5 } } }, { reason: 'test: a video storage ceiling' });
+ assert.deepStrictEqual([policy.hotCeilingFor('video'), policy.hotCeilingFor('image')], [12.5, null]);
+ await assert.rejects(policy.set({ classes: { video: { maxHotUsdPerMonth: -1 } } }, { reason: 'test: a negative ceiling' }));
+ assert.strictEqual(policy.hotCeilingFor('video'), 12.5, 'a refused ceiling changes nothing');
+
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);