openvibers/openvibe.media Commits
Take sweep budgets and residency per class from storage policy (#14)
Changes
@@@ -52,6 +52,7 @@ const ready = (async () => {await require('./vod/tier-config').init(require('./vod/vod-storage').DEFAULTS);await require('./objects/tier-policy').init();await require('./placement/cost-tiers').init();++ await require('./placement/storage-policy').init(); // per-class sweep budgets and residency; before the tiering sweep starts// Placement core (F1 part A): the provider registry, signals + router. Boot the cheap live// health loop after the DB is open (it reads memory only) and the router is wired in.if (!drill.enabled) {@@@ -35,6 +35,7 @@ const crypto = require('crypto');const db = require('../db/database');const drill = require('../drill');const policy = require('./tier-policy');++const storagePolicy = require('../placement/storage-policy');const popularity = require('./popularity');const MB = 1024 * 1024;@@@ -127,7 +128,7 @@ async function inputsOf(obj, { locs, pop, held, settings }) {unique_viewers_7d: pop ? pop.unique_viewers : 0,last_viewed_day: pop ? pop.last_viewed_day : await popularity.lastViewedDay(obj.id),size_bytes: Number(obj.size_bytes) || 0,-- kind: obj.kind, visibility: obj.visibility, lifecycle_status: obj.lifecycle_status,++ kind: obj.kind, class: storagePolicy.classOf(obj), visibility: obj.visibility, lifecycle_status: obj.lifecycle_status,content_hash: !!obj.content_hash,canonical: canon ? { provider: canon.provider, state: canon.state, verified_at: canon.verified_at || null,checksum_matches: !!(canon.checksum && obj.content_hash && canon.checksum === obj.content_hash) } : null,@@@ -207,6 +208,7 @@ 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;++ 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`);@@@ -259,6 +261,7 @@ 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;++ 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`);@@@ -346,6 +349,21 @@ function tally(out, r) {if (r.outcome === 'failed') out.errors.push({ object_id: r.object_id, action: r.action, error: r.error });}++/** When the object last moved: its latest completed promotion or demotion, else null. */++async function lastMovedAt(objectId) {++ const r = await db.get(`SELECT MAX(decided_at) AS at FROM media_object_tier_decisions WHERE object_id = ? AND outcome = 'done'`, [objectId]);++ return r && r.at ? r.at : null;++}++++/** Why the class's minimum residency (media.storage_policy) holds this object where it is, or null. */++async function residencyRefusal(obj, cls, action, now) {++ const { minResidencyMs } = storagePolicy.settings().classes[cls];++ if (!minResidencyMs) return null;++ const lastAt = await lastMovedAt(obj.id);++ if (storagePolicy.mayMove({ class: cls, lastMovedAt: lastAt, now })) return null;++ return `minimum residency of ${minResidencyMs} ms for class ${cls} (media.storage_policy) not reached since the last move at ${lastAt}; ${action} suppressed`;++}++/*** One pass over native objects. → { gate, promoted, demoted, would_promote, would_demote, already, refused, failed,* repeats, skipped_backoff, candidates: { promote, demote }, rotated, errors }@@@ -362,26 +380,33 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {// Demotions first: they free the R2 cache, and a deleted object's copy should not wait behind promotions.const demote_ = await demotionCandidates(settings, { now });out.candidates.demote = demote_.length;-- let budget = settings.maxDemotionsPerSweep;++ // Budgets and residency are per class (media.storage_policy); a class spends only its own budget.++ const spent = { promote: {}, demote: {} };++ const room = (action, cls) => (spent[action][cls] || 0) < storagePolicy.budgetFor(cls)[action === 'promote' ? 'maxPromotionsPerSweep' : 'maxDemotionsPerSweep'];for (const { row, reason } of demote_) {-- if (budget <= 0) break;++ const cls = storagePolicy.classOf(row);++ if (!room('demote', cls)) continue;if (await inBackoff(row.id, 'demote')) { out.skipped_backoff++; continue; }-- const r = await demote(row.id, { trigger, reason, settings, now });++ // 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, reason, settings, now, residency });tally(out, { ...r, action: 'demote' });-- if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) budget--;++ if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.demote[cls] = (spent.demote[cls] || 0) + 1;}const promote_ = [];-- for (const c of await promotionCandidates(settings, { now, limit: Math.max(50, settings.maxPromotionsPerSweep * 10) })) if (!await db.isSandboxTenant(c.app_id)) promote_.push(c);++ const promoteLimit = Math.max(...storagePolicy.CLASSES.map((c) => storagePolicy.budgetFor(c).maxPromotionsPerSweep));++ for (const c of await promotionCandidates(settings, { now, limit: Math.max(50, promoteLimit * 10) })) if (!await db.isSandboxTenant(c.app_id)) promote_.push(c);out.candidates.promote = promote_.length;-- budget = settings.maxPromotionsPerSweep;for (const c of promote_) {-- if (budget <= 0) break;++ const cls = storagePolicy.classOf(c);++ if (!room('promote', cls)) continue;if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }-- const r = await promote(c.id, { trigger, reason: promoteReason(c, settings), settings, now,++ const residency = await residencyRefusal(c, cls, 'promote', now);++ const r = await promote(c.id, { trigger, reason: promoteReason(c, settings), settings, now, residency,pop: { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } });tally(out, { ...r, action: 'promote' });-- if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) budget--;++ if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.promote[cls] = (spent.promote[cls] || 0) + 1;}if (out.promoted || out.demoted || out.failed) {@@@ -1,13 +1,13 @@/*** Revisioned placement budgets and residency policy (docs/media-fabric.md §6, §10).-- * This namespace is additive: existing tiering sweeps do not read it yet.++ * The object sweep (server/objects/tiering.js) takes its per-class budgets and residency from here.*/'use strict';const config = require('openvibe-shared/config');const db = require('../db/database');--// Match media.object_tier (server/objects/tier-policy.js) until placement uses this policy.++// 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'];const DEFAULTS = {@@@ -81,6 +81,17 @@ function classPolicy(name) {return settings().classes[name];}++/** The placement class of a native object: video/audio → video, image/thumbnail → image, everything else → download. */++function classOf(obj) {++ const kind = obj && obj.kind;++ if (kind === 'vod' || kind === 'clip') return 'video';++ if (kind === 'thumbnail' || kind === 'screenshot' || kind === 'avatar') return 'image';++ const major = String((obj && obj.mime_type) || '').toLowerCase().split('/')[0];++ if (major === 'video' || major === 'audio') return 'video';++ if (major === 'image') return 'image';++ return 'download';++}++function budgetFor(name) {const { maxPromotionsPerSweep, maxDemotionsPerSweep } = classPolicy(name);return { maxPromotionsPerSweep, maxDemotionsPerSweep };@@@ -102,4 +113,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, budgetFor, mayMove, _reset: () => { store = null; } };++module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, mayMove, _reset: () => { store = null; } };@@@ -81,6 +81,7 @@ const crypto = require('crypto');// The revisioned tier policies load once the database is open (server/index.js does this at boot).await storage.tierConfig.init(storage.DEFAULTS);await require('../server/objects/tier-policy').init();++ await require('../server/placement/storage-policy').init();const popularity = require('../server/objects/popularity');const policy = require('../server/objects/tier-policy');const tiering = require('../server/objects/tiering');@@@ -222,6 +223,7 @@ const crypto = require('crypto');assert.ok(/^gate off \(active = false\): would promote: 600 unique viewers in 7 days >= promoteMinUniqueViewers7d 500/.test(d.reason), d.reason);const din = JSON.parse(d.inputs);assert.deepStrictEqual([din.unique_viewers_7d, din.last_viewed_day, din.gate_active, din.held, din.canonical.provider, din.canonical.checksum_matches], [600, today, false, false, 'local', true]);++ assert.strictEqual(din.class, 'download', 'every decision row records the class that set its budget');assert.deepStrictEqual(JSON.parse(d.thresholds).active, { value: false, source: 'default' });assert.deepStrictEqual([(await last(cached.id)).action, (await last(cached.id)).outcome], ['demote', 'dry_run']);s = await tiering.runSweep();@@@ -362,7 +364,7 @@ const crypto = require('crypto');assert.ok(/still there/.test(m.error));// Seven hours ago: past the back-off, so the sweep below tries again (and still within the 24-hour counts).await db.run("UPDATE media_object_tier_decisions SET decided_at = ov_now_iso('-7 hours') WHERE object_id = ? AND outcome = 'failed'", [sticky.id]);-- await policy.set({ maxDemotionsPerSweep: 20 }, { reason: 'test: room for every demotion' });++ await require('../server/placement/storage-policy').set({ classes: { download: { maxDemotionsPerSweep: 20 } } }, { reason: 'test: room for every demotion' });s = await tiering.runSweep();assert.ok(!await r2Row(hot.id) && !buckets['r2-bucket'].has(`objects/live/${hot.id}`), 'an idle object leaves R2');assert.ok(fs.existsSync(hot.file) && (await db.get("SELECT state FROM media_locations WHERE object_id = ? AND provider = 'local'", [hot.id])).state === 'present');@@@ -421,7 +423,7 @@ const crypto = require('crypto');const ns = a.body.namespaces.find((n) => n.namespace === 'media.object_tier');assert.ok(ns && ns.values.active === true && ns.values.promoteMinUniqueViewers7d === 500, 'the policy is in the config routes');a = await call('/api/v1/live/admin/storage/config/media.object_tier/history');-- assert.deepStrictEqual(a.body.snapshots.map((x) => x.state), ['active', 'superseded', 'superseded', 'superseded']);++ assert.deepStrictEqual(a.body.snapshots.map((x) => x.state), ['active', 'superseded', 'superseded']);const post = (p, body) => new Promise((resolve, reject) => {const data = JSON.stringify(body);const rq = http.request(base + p, { method: 'POST', headers: { Authorization: 'Bearer live-key-objtier', 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },@@@ -456,6 +458,35 @@ const crypto = require('crypto');assert.ok(/Native objects/.test(html) && /Activation gate/.test(html) && html.includes(bad.id), 'the /me/ops page shows the gate and the recent decisions');console.log('✅ decisions: /tiers/objects/policy and /decisions (filters, pages, app-scoped), the config routes, media_tier_decisions_24h{target}, and the operator views');++ // ── 9b. Per-class budgets and residency come from media.storage_policy ──++ const sp = require('../server/placement/storage-policy');++ await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);++ const vids = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];++ const imgs = [await addObject({ mime: 'image/png' }), await addObject({ mime: 'image/png' })];++ const dl = await addObject();++ for (const o of [...vids, ...imgs, dl]) await setViews(o.id, [[0, 900]]);++ await sp.set({ classes: { video: { maxPromotionsPerSweep: 1 }, image: { maxPromotionsPerSweep: 2, minResidencyMs: 86400000 }, download: { maxPromotionsPerSweep: 0 } } },++ { reason: 'test: per-class budgets' });++ s = await tiering.runSweep();++ const promotedOf = async (list) => (await Promise.all(list.map(async (o) => !!await r2Row(o.id)))).filter(Boolean).length;++ assert.deepStrictEqual([await promotedOf(vids), await promotedOf(imgs), await promotedOf([dl])], [1, 2, 0], 'each class spends its own promotion budget');++ assert.strictEqual((await decisions(dl.id)).length, 0, 'a class with a zero budget is not tried');++ const r2Done = new Set((await Promise.all(vids.map(async (o) => (await r2Row(o.id)) ? o.id : null))).filter(Boolean));++ assert.strictEqual(JSON.parse((await last(imgs[0].id)).inputs).class, 'image');++ assert.strictEqual(JSON.parse((await last(vids.find((o) => r2Done.has(o.id)).id)).inputs).class, 'video');++ // Residency: the images were just promoted, so an idle demotion is suppressed and logged as refused.++ await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);++ for (const o of imgs) await ageR2(o.id, 30);++ s = await tiering.runSweep();++ assert.ok(await r2Row(imgs[0].id) && await r2Row(imgs[1].id), 'residency keeps the fresh copies');++ d = await last(imgs[0].id);++ assert.deepStrictEqual([d.action, d.outcome, JSON.parse(d.inputs).class], ['demote', 'refused', 'image']);++ assert.ok(/minimum residency of 86400000 ms for class image/.test(d.reason), d.reason);++ await sp.set({ classes: { image: { minResidencyMs: 0 }, download: { maxPromotionsPerSweep: 3 } } }, { reason: 'test: residency off, budgets back' });++ s = await tiering.runSweep();++ assert.ok(!await r2Row(imgs[0].id) && !await r2Row(imgs[1].id), 'with no residency the idle copies leave');++ console.log('✅ per-class budgets and residency from media.storage_policy; every decision row shows its class; a residency-suppressed demotion logs refused');++// ── 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));@@@ -24,6 +24,10 @@ const path = require('path');}assert.strictEqual(policy.mayMove({ class: 'video', lastMovedAt: 1000, now: 1000 }), true, 'zero residency is off');assert.throws(() => policy.budgetFor('unknown'), RangeError);++ const cls = (kind, mime_type) => policy.classOf({ kind, mime_type });++ assert.deepStrictEqual([cls('vod'), cls('clip'), cls('file', 'audio/mpeg'), cls('file', 'Video/MP4'), cls('asset', 'video/webm')], Array(5).fill('video'));++ assert.deepStrictEqual([cls('thumbnail'), cls('screenshot'), cls('avatar'), cls('file', 'image/png')], Array(4).fill('image'));++ assert.deepStrictEqual([cls('file', 'application/zip'), cls('file'), cls('asset', null), policy.classOf(null)], Array(4).fill('download'));await policy.set({ classes: { image: { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4, minResidencyMs: 60_000 } } },{ actor: { type: 'service', id: 'media-test' }, reason: 'test image placement policy' });
@@ -52,6 +52,7 @@ const ready = (async () => {
@@ -52,6 +52,7 @@ const ready = (async () => {
await require('./vod/tier-config').init(require('./vod/vod-storage').DEFAULTS);
await require('./vod/tier-config').init(require('./vod/vod-storage').DEFAULTS);
await require('./objects/tier-policy').init();
await require('./objects/tier-policy').init();
await require('./placement/cost-tiers').init();
await require('./placement/cost-tiers').init();
+ await require('./placement/storage-policy').init(); // per-class sweep budgets and residency; before the tiering sweep starts
// Placement core (F1 part A): the provider registry, signals + router. Boot the cheap live
// Placement core (F1 part A): the provider registry, signals + router. Boot the cheap live
// health loop after the DB is open (it reads memory only) and the router is wired in.
// health loop after the DB is open (it reads memory only) and the router is wired in.
if (!drill.enabled) {
if (!drill.enabled) {
@@ -35,6 +35,7 @@ const crypto = require('crypto');
@@ -35,6 +35,7 @@ const crypto = require('crypto');
const db = require('../db/database');
const db = require('../db/database');
const drill = require('../drill');
const drill = require('../drill');
const policy = require('./tier-policy');
const policy = require('./tier-policy');
+const storagePolicy = require('../placement/storage-policy');
const popularity = require('./popularity');
const popularity = require('./popularity');
const MB = 1024 * 1024;
const MB = 1024 * 1024;
@@ -127,7 +128,7 @@ async function inputsOf(obj, { locs, pop, held, settings }) {
@@ -127,7 +128,7 @@ async function inputsOf(obj, { locs, pop, held, settings }) {
unique_viewers_7d: pop ? pop.unique_viewers : 0,
unique_viewers_7d: pop ? pop.unique_viewers : 0,
last_viewed_day: pop ? pop.last_viewed_day : await popularity.lastViewedDay(obj.id),
last_viewed_day: pop ? pop.last_viewed_day : await popularity.lastViewedDay(obj.id),
size_bytes: Number(obj.size_bytes) || 0,
size_bytes: Number(obj.size_bytes) || 0,
- kind: obj.kind, visibility: obj.visibility, lifecycle_status: obj.lifecycle_status,
+ kind: obj.kind, class: storagePolicy.classOf(obj), visibility: obj.visibility, lifecycle_status: obj.lifecycle_status,
content_hash: !!obj.content_hash,
content_hash: !!obj.content_hash,
canonical: canon ? { provider: canon.provider, state: canon.state, verified_at: canon.verified_at || null,
canonical: canon ? { provider: canon.provider, state: canon.state, verified_at: canon.verified_at || null,
checksum_matches: !!(canon.checksum && obj.content_hash && canon.checksum === obj.content_hash) } : null,
checksum_matches: !!(canon.checksum && obj.content_hash && canon.checksum === obj.content_hash) } : null,
@@ -207,6 +208,7 @@ async function promote(objectId, ctx = {}) {
@@ -207,6 +208,7 @@ 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 } = 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 });
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`);
@@ -259,6 +261,7 @@ async function demote(objectId, ctx = {}) {
@@ -259,6 +261,7 @@ 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 } = 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 });
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`);
@@ -346,6 +349,21 @@ function tally(out, r) {
@@ -346,6 +349,21 @@ function tally(out, r) {
if (r.outcome === 'failed') out.errors.push({ object_id: r.object_id, action: r.action, error: r.error });
if (r.outcome === 'failed') out.errors.push({ object_id: r.object_id, action: r.action, error: r.error });
}
}
+/** When the object last moved: its latest completed promotion or demotion, else null. */
+async function lastMovedAt(objectId) {
+ const r = await db.get(`SELECT MAX(decided_at) AS at FROM media_object_tier_decisions WHERE object_id = ? AND outcome = 'done'`, [objectId]);
+ return r && r.at ? r.at : null;
+}
+
+/** Why the class's minimum residency (media.storage_policy) holds this object where it is, or null. */
+async function residencyRefusal(obj, cls, action, now) {
+ const { minResidencyMs } = storagePolicy.settings().classes[cls];
+ if (!minResidencyMs) return null;
+ const lastAt = await lastMovedAt(obj.id);
+ if (storagePolicy.mayMove({ class: cls, lastMovedAt: lastAt, now })) return null;
+ return `minimum residency of ${minResidencyMs} ms for class ${cls} (media.storage_policy) not reached since the last move at ${lastAt}; ${action} suppressed`;
+}
+
/**
/**
* One pass over native objects. → { gate, promoted, demoted, would_promote, would_demote, already, refused, failed,
* One pass over native objects. → { gate, 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 }
@@ -362,26 +380,33 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -362,26 +380,33 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
// 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_ = await demotionCandidates(settings, { now });
out.candidates.demote = demote_.length;
out.candidates.demote = demote_.length;
- let budget = settings.maxDemotionsPerSweep;
+ // Budgets and residency are per class (media.storage_policy); a class spends only its own budget.
+ const spent = { promote: {}, demote: {} };
+ const room = (action, cls) => (spent[action][cls] || 0) < storagePolicy.budgetFor(cls)[action === 'promote' ? 'maxPromotionsPerSweep' : 'maxDemotionsPerSweep'];
for (const { row, reason } of demote_) {
for (const { row, reason } of demote_) {
- if (budget <= 0) break;
+ const cls = storagePolicy.classOf(row);
+ 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; }
- const r = await demote(row.id, { trigger, reason, settings, now });
+ // 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, reason, settings, now, residency });
tally(out, { ...r, action: 'demote' });
tally(out, { ...r, action: 'demote' });
- if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) budget--;
+ if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.demote[cls] = (spent.demote[cls] || 0) + 1;
}
}
const promote_ = [];
const promote_ = [];
- for (const c of await promotionCandidates(settings, { now, limit: Math.max(50, settings.maxPromotionsPerSweep * 10) })) if (!await db.isSandboxTenant(c.app_id)) promote_.push(c);
+ const promoteLimit = Math.max(...storagePolicy.CLASSES.map((c) => storagePolicy.budgetFor(c).maxPromotionsPerSweep));
+ for (const c of await promotionCandidates(settings, { now, limit: Math.max(50, promoteLimit * 10) })) if (!await db.isSandboxTenant(c.app_id)) promote_.push(c);
out.candidates.promote = promote_.length;
out.candidates.promote = promote_.length;
- budget = settings.maxPromotionsPerSweep;
for (const c of promote_) {
for (const c of promote_) {
- if (budget <= 0) break;
+ const cls = storagePolicy.classOf(c);
+ if (!room('promote', cls)) continue;
if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }
if (await inBackoff(c.id, 'promote')) { out.skipped_backoff++; continue; }
- const r = await promote(c.id, { trigger, reason: promoteReason(c, settings), settings, now,
+ const residency = await residencyRefusal(c, cls, 'promote', now);
+ const r = await promote(c.id, { trigger, reason: promoteReason(c, settings), settings, now, residency,
pop: { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } });
pop: { unique_viewers: c.unique_viewers_7d, last_viewed_day: c.last_viewed_day } });
tally(out, { ...r, action: 'promote' });
tally(out, { ...r, action: 'promote' });
- if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) budget--;
+ if (!r.repeat && ['done', 'failed', 'dry_run'].includes(r.outcome)) spent.promote[cls] = (spent.promote[cls] || 0) + 1;
}
}
if (out.promoted || out.demoted || out.failed) {
if (out.promoted || out.demoted || out.failed) {
@@ -1,13 +1,13 @@
@@ -1,13 +1,13 @@
/**
/**
* Revisioned placement budgets and residency policy (docs/media-fabric.md §6, §10).
* Revisioned placement budgets and residency policy (docs/media-fabric.md §6, §10).
- * This namespace is additive: existing tiering sweeps do not read it yet.
+ * The object sweep (server/objects/tiering.js) takes its per-class budgets and residency from here.
*/
*/
'use strict';
'use strict';
const config = require('openvibe-shared/config');
const config = require('openvibe-shared/config');
const db = require('../db/database');
const db = require('../db/database');
-// Match media.object_tier (server/objects/tier-policy.js) until placement uses this policy.
+// 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'];
const DEFAULTS = {
const DEFAULTS = {
@@ -81,6 +81,17 @@ function classPolicy(name) {
@@ -81,6 +81,17 @@ function classPolicy(name) {
return settings().classes[name];
return settings().classes[name];
}
}
+/** The placement class of a native object: video/audio → video, image/thumbnail → image, everything else → download. */
+function classOf(obj) {
+ const kind = obj && obj.kind;
+ if (kind === 'vod' || kind === 'clip') return 'video';
+ if (kind === 'thumbnail' || kind === 'screenshot' || kind === 'avatar') return 'image';
+ const major = String((obj && obj.mime_type) || '').toLowerCase().split('/')[0];
+ if (major === 'video' || major === 'audio') return 'video';
+ if (major === 'image') return 'image';
+ return 'download';
+}
+
function budgetFor(name) {
function budgetFor(name) {
const { maxPromotionsPerSweep, maxDemotionsPerSweep } = classPolicy(name);
const { maxPromotionsPerSweep, maxDemotionsPerSweep } = classPolicy(name);
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
return { maxPromotionsPerSweep, maxDemotionsPerSweep };
@@ -102,4 +113,4 @@ function mayMove({ class: name, lastMovedAt, now = Date.now() }) {
@@ -102,4 +113,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, budgetFor, mayMove, _reset: () => { store = null; } };
+module.exports = { DEFAULTS, SCHEMA, CLASSES, init, get, set, settings, classOf, budgetFor, mayMove, _reset: () => { store = null; } };
@@ -81,6 +81,7 @@ const crypto = require('crypto');
@@ -81,6 +81,7 @@ const crypto = require('crypto');
// The revisioned tier policies load once the database is open (server/index.js does this at boot).
// The revisioned tier policies load once the database is open (server/index.js does this at boot).
await storage.tierConfig.init(storage.DEFAULTS);
await storage.tierConfig.init(storage.DEFAULTS);
await require('../server/objects/tier-policy').init();
await require('../server/objects/tier-policy').init();
+ await require('../server/placement/storage-policy').init();
const popularity = require('../server/objects/popularity');
const popularity = require('../server/objects/popularity');
const policy = require('../server/objects/tier-policy');
const policy = require('../server/objects/tier-policy');
const tiering = require('../server/objects/tiering');
const tiering = require('../server/objects/tiering');
@@ -222,6 +223,7 @@ const crypto = require('crypto');
@@ -222,6 +223,7 @@ const crypto = require('crypto');
assert.ok(/^gate off \(active = false\): would promote: 600 unique viewers in 7 days >= promoteMinUniqueViewers7d 500/.test(d.reason), d.reason);
assert.ok(/^gate off \(active = false\): would promote: 600 unique viewers in 7 days >= promoteMinUniqueViewers7d 500/.test(d.reason), d.reason);
const din = JSON.parse(d.inputs);
const din = JSON.parse(d.inputs);
assert.deepStrictEqual([din.unique_viewers_7d, din.last_viewed_day, din.gate_active, din.held, din.canonical.provider, din.canonical.checksum_matches], [600, today, false, false, 'local', true]);
assert.deepStrictEqual([din.unique_viewers_7d, din.last_viewed_day, din.gate_active, din.held, din.canonical.provider, din.canonical.checksum_matches], [600, today, false, false, 'local', true]);
+ assert.strictEqual(din.class, 'download', 'every decision row records the class that set its budget');
assert.deepStrictEqual(JSON.parse(d.thresholds).active, { value: false, source: 'default' });
assert.deepStrictEqual(JSON.parse(d.thresholds).active, { value: false, source: 'default' });
assert.deepStrictEqual([(await last(cached.id)).action, (await last(cached.id)).outcome], ['demote', 'dry_run']);
assert.deepStrictEqual([(await last(cached.id)).action, (await last(cached.id)).outcome], ['demote', 'dry_run']);
s = await tiering.runSweep();
s = await tiering.runSweep();
@@ -362,7 +364,7 @@ const crypto = require('crypto');
@@ -362,7 +364,7 @@ const crypto = require('crypto');
assert.ok(/still there/.test(m.error));
assert.ok(/still there/.test(m.error));
// Seven hours ago: past the back-off, so the sweep below tries again (and still within the 24-hour counts).
// Seven hours ago: past the back-off, so the sweep below tries again (and still within the 24-hour counts).
await db.run("UPDATE media_object_tier_decisions SET decided_at = ov_now_iso('-7 hours') WHERE object_id = ? AND outcome = 'failed'", [sticky.id]);
await db.run("UPDATE media_object_tier_decisions SET decided_at = ov_now_iso('-7 hours') WHERE object_id = ? AND outcome = 'failed'", [sticky.id]);
- await policy.set({ maxDemotionsPerSweep: 20 }, { reason: 'test: room for every demotion' });
+ await require('../server/placement/storage-policy').set({ classes: { download: { maxDemotionsPerSweep: 20 } } }, { reason: 'test: room for every demotion' });
s = await tiering.runSweep();
s = await tiering.runSweep();
assert.ok(!await r2Row(hot.id) && !buckets['r2-bucket'].has(`objects/live/${hot.id}`), 'an idle object leaves R2');
assert.ok(!await r2Row(hot.id) && !buckets['r2-bucket'].has(`objects/live/${hot.id}`), 'an idle object leaves R2');
assert.ok(fs.existsSync(hot.file) && (await db.get("SELECT state FROM media_locations WHERE object_id = ? AND provider = 'local'", [hot.id])).state === 'present');
assert.ok(fs.existsSync(hot.file) && (await db.get("SELECT state FROM media_locations WHERE object_id = ? AND provider = 'local'", [hot.id])).state === 'present');
@@ -421,7 +423,7 @@ const crypto = require('crypto');
@@ -421,7 +423,7 @@ const crypto = require('crypto');
const ns = a.body.namespaces.find((n) => n.namespace === 'media.object_tier');
const ns = a.body.namespaces.find((n) => n.namespace === 'media.object_tier');
assert.ok(ns && ns.values.active === true && ns.values.promoteMinUniqueViewers7d === 500, 'the policy is in the config routes');
assert.ok(ns && ns.values.active === true && ns.values.promoteMinUniqueViewers7d === 500, 'the policy is in the config routes');
a = await call('/api/v1/live/admin/storage/config/media.object_tier/history');
a = await call('/api/v1/live/admin/storage/config/media.object_tier/history');
- assert.deepStrictEqual(a.body.snapshots.map((x) => x.state), ['active', 'superseded', 'superseded', 'superseded']);
+ assert.deepStrictEqual(a.body.snapshots.map((x) => x.state), ['active', 'superseded', 'superseded']);
const post = (p, body) => new Promise((resolve, reject) => {
const post = (p, body) => new Promise((resolve, reject) => {
const data = JSON.stringify(body);
const data = JSON.stringify(body);
const rq = http.request(base + p, { method: 'POST', headers: { Authorization: 'Bearer live-key-objtier', 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },
const rq = http.request(base + p, { method: 'POST', headers: { Authorization: 'Bearer live-key-objtier', 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },
@@ -456,6 +458,35 @@ const crypto = require('crypto');
@@ -456,6 +458,35 @@ const crypto = require('crypto');
assert.ok(/Native objects/.test(html) && /Activation gate/.test(html) && html.includes(bad.id), 'the /me/ops page shows the gate and the recent decisions');
assert.ok(/Native objects/.test(html) && /Activation gate/.test(html) && html.includes(bad.id), 'the /me/ops page shows the gate and the recent decisions');
console.log('✅ decisions: /tiers/objects/policy and /decisions (filters, pages, app-scoped), the config routes, media_tier_decisions_24h{target}, and the operator views');
console.log('✅ decisions: /tiers/objects/policy and /decisions (filters, pages, app-scoped), the config routes, media_tier_decisions_24h{target}, and the operator views');
+ // ── 9b. Per-class budgets and residency come from media.storage_policy ──
+ const sp = require('../server/placement/storage-policy');
+ await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);
+ const vids = [await addObject({ mime: 'video/mp4' }), await addObject({ mime: 'video/mp4' })];
+ const imgs = [await addObject({ mime: 'image/png' }), await addObject({ mime: 'image/png' })];
+ const dl = await addObject();
+ for (const o of [...vids, ...imgs, dl]) await setViews(o.id, [[0, 900]]);
+ await sp.set({ classes: { video: { maxPromotionsPerSweep: 1 }, image: { maxPromotionsPerSweep: 2, minResidencyMs: 86400000 }, download: { maxPromotionsPerSweep: 0 } } },
+ { reason: 'test: per-class budgets' });
+ s = await tiering.runSweep();
+ const promotedOf = async (list) => (await Promise.all(list.map(async (o) => !!await r2Row(o.id)))).filter(Boolean).length;
+ assert.deepStrictEqual([await promotedOf(vids), await promotedOf(imgs), await promotedOf([dl])], [1, 2, 0], 'each class spends its own promotion budget');
+ assert.strictEqual((await decisions(dl.id)).length, 0, 'a class with a zero budget is not tried');
+ const r2Done = new Set((await Promise.all(vids.map(async (o) => (await r2Row(o.id)) ? o.id : null))).filter(Boolean));
+ assert.strictEqual(JSON.parse((await last(imgs[0].id)).inputs).class, 'image');
+ assert.strictEqual(JSON.parse((await last(vids.find((o) => r2Done.has(o.id)).id)).inputs).class, 'video');
+ // Residency: the images were just promoted, so an idle demotion is suppressed and logged as refused.
+ await db.run('DELETE FROM media_object_views_daily WHERE object_id != ?', [cand.id]);
+ for (const o of imgs) await ageR2(o.id, 30);
+ s = await tiering.runSweep();
+ assert.ok(await r2Row(imgs[0].id) && await r2Row(imgs[1].id), 'residency keeps the fresh copies');
+ d = await last(imgs[0].id);
+ assert.deepStrictEqual([d.action, d.outcome, JSON.parse(d.inputs).class], ['demote', 'refused', 'image']);
+ assert.ok(/minimum residency of 86400000 ms for class image/.test(d.reason), d.reason);
+ await sp.set({ classes: { image: { minResidencyMs: 0 }, download: { maxPromotionsPerSweep: 3 } } }, { reason: 'test: residency off, budgets back' });
+ s = await tiering.runSweep();
+ assert.ok(!await r2Row(imgs[0].id) && !await r2Row(imgs[1].id), 'with no residency the idle copies leave');
+ console.log('✅ per-class budgets and residency from media.storage_policy; every decision row shows its class; a residency-suppressed demotion logs refused');
+
// ── 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));
@@ -24,6 +24,10 @@ const path = require('path');
@@ -24,6 +24,10 @@ const path = require('path');
}
}
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);
+ const cls = (kind, mime_type) => policy.classOf({ kind, mime_type });
+ assert.deepStrictEqual([cls('vod'), cls('clip'), cls('file', 'audio/mpeg'), cls('file', 'Video/MP4'), cls('asset', 'video/webm')], Array(5).fill('video'));
+ assert.deepStrictEqual([cls('thumbnail'), cls('screenshot'), cls('avatar'), cls('file', 'image/png')], Array(4).fill('image'));
+ assert.deepStrictEqual([cls('file', 'application/zip'), cls('file'), cls('asset', null), policy.classOf(null)], Array(4).fill('download'));
await policy.set({ classes: { image: { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4, minResidencyMs: 60_000 } } },
await policy.set({ classes: { image: { maxPromotionsPerSweep: 2, maxDemotionsPerSweep: 4, minResidencyMs: 60_000 } } },
{ actor: { type: 'service', id: 'media-test' }, reason: 'test image placement policy' });
{ actor: { type: 'service', id: 'media-test' }, reason: 'test image placement policy' });