openvibers/openvibe.media Commits

R2 tier decisions are recorded and explainable: every promotion to and demotion from the R2 popularity cache gets a media_tier_decisions row with the inputs it saw (view_count, last_accessed_at, provider, recording, hold, size, created_at), the R2 thresholds in force each with its source (default or a media_settings override), the reason in words, who asked (sweep, admin, clip, drill, manual) and the outcome (done, already, refused, failed, with the error). Before, a tier move only flipped media_locations and wrote a console line. The sweep states its rule per VOD (view_count 25 >= r2MinViews 20 and last accessed within r2RecentAccessDays 3; idle longer than r2MaxIdleDays 14), admin moves name the app, a clip cut that fetches an R2 VOD home (which drops the R2 copy) is logged as a demotion to local, and the eviction drill's evict and re-warm are logged as such. Staff read the policy through the app's key: GET /api/v1/:app/admin/storage/tiers/policy (each threshold with value, default and source, the promote and demote rules, provider status, the app's 24-hour counts and 20 latest decisions) and GET /tiers/decisions (filters vod_id, action, outcome; paged by before_id), both app-scoped and read-only; /metrics gains media_tier_decisions_24h{action,outcome}. The thresholds are unchanged (the test pins them). test/tier-decisions.test.js runs the sweep, admin moves, a hot-fetch and a drill step against a fake S3 through the real SDK.

OP OpenVibers committed 24 Sept 2026 · 7114433029 · 8 files, +403 −18


Changes

ddiff --git a/README.md b/README.md
iindex 23c1f0c..9fc3450 100644
---- a/README.md
++++ b/README.md
@@@ -265,6 +265,8 @@ directories are shared across apps) and responses carry a `note` saying so.
| POST | `/admin/storage/tiers/sweep` | run the tiering sweep now → sweep summary |
| POST | `/admin/storage/tiers/move` | `{ vod_id, target: local\|hot\|b2\|cold\|r2 }` — reuses the storage engine's verified move logic; VOD must belong to the app |
| POST | `/admin/storage/tiers/bulk-move` | `{ ids: [...], target }` (max 50) → `{ moved, bytes, errors? }` |
++| GET | `/admin/storage/tiers/policy` | **read-only R2 policy**: each threshold (`r2Enabled`, `r2MinViews`, `r2RecentAccessDays`, `r2MaxIdleDays`, `r2MaxPerSweep`) as `{ value, default, source: default\|setting }`, the promote and demote rules in words, the R2/B2 provider status, and this app's decisions: `last_24h` counts per action and outcome and the 20 most recent |
++| GET | `/admin/storage/tiers/decisions?vod_id&action&outcome&limit&before_id` | this app's R2 promote/demote decisions, newest first: `{ id, decided_at, vod_id, object_id, action, from_provider, to_provider, outcome: done\|already\|refused\|failed, trigger: sweep\|admin\|clip\|drill\|manual, reason, inputs, thresholds, error }` → `{ decisions, next_before_id }` |
| GET | `/admin/storage/buckets` | sanitized bucket status per provider: `{ configured, endpoint, bucket, region, healthy, reachable }` via a live HeadBucket probe — **credentials are never returned** |
### Webhooks (outbound)
@@@ -409,7 +411,11 @@ model current. Full reference: **[docs/object-model.md](docs/object-model.md)**.
R2 popularity cache; verified uploads before any deletion; periodic sweep
(offload by age/views/last-access, aggressive drain under disk pressure, R2
promote/demote); presigned-302 playback. Knobs live in `media_settings` under
--`storage_tier.*` (JSON values). CLI:
++`storage_tier.*` (JSON values). Every promotion to and demotion from R2 (the sweep, an admin move, a
++clip cut fetching an R2 VOD home, the eviction drill) is logged in `media_tier_decisions` with the
++inputs it saw (views, last access, provider, recording, hold, size), the thresholds in force and their
++source, the reason and the outcome; `/admin/storage/tiers/policy` shows the policy and the log, and
++`media_tier_decisions_24h{action,outcome}` counts them on `/metrics` (`test/tier-decisions.test.js`). CLI:
```
node server/vod/vod-storage.js check|migrate-legacy|drain [targetPct]
ddiff --git a/scripts/r2-eviction-drill.js b/scripts/r2-eviction-drill.js
iindex 7085378..0a71909 100644
---- a/scripts/r2-eviction-drill.js
++++ b/scripts/r2-eviction-drill.js
@@@ -171,7 +171,7 @@ async function runDrill(opts, deps) {
const before = await served();
if (!step('before: served from R2', before.provider === 'r2' && before.first_mib_sha256 === expectSha, before)) return fail('playback is not coming from R2 before the eviction');
// 2. evict
-- const ev = await storage.demoteFromR2(vod.id);
++ const ev = await storage.demoteFromR2(vod.id, { trigger: 'drill', reason: 'R2 eviction drill: evict the R2 copy' });
const rowAfterEvict = db.get('SELECT storage_provider FROM vods WHERE id = ?', [vod.id]);
const r2Gone = await storage.headObject('r2', art.key);
art.locations.after_evict = locations();
@@@ -185,7 +185,7 @@ async function runDrill(opts, deps) {
return fail('after the eviction the VOD is not served from B2');
}
// 4. re-warm
-- const pr = await storage.promoteToR2(vod.id);
++ const pr = await storage.promoteToR2(vod.id, { trigger: 'drill', reason: 'R2 eviction drill: re-warm R2 from B2' });
const r2Back = await storage.headObject('r2', art.key);
const rowAfterWarm = db.get('SELECT storage_provider FROM vods WHERE id = ?', [vod.id]);
if (!step('re-warm: R2 copy restored from B2', !!(pr && pr.ok) && !!r2Back && r2Back.size === b2.size && rowAfterWarm.storage_provider === 'r2', { result: pr, r2_head: r2Back, storage_provider: rowAfterWarm.storage_provider })) {
ddiff --git a/server/admin/routes.js b/server/admin/routes.js
iindex 45f5f07..6b21ef6 100644
---- a/server/admin/routes.js
++++ b/server/admin/routes.js
@@@ -16,6 +16,11 @@
* POST /tiers/move { vod_id, target: local|hot|b2|cold|r2 }
* POST /tiers/bulk-move { ids: [...], target } (max 50)
* GET /buckets sanitized B2/R2 bucket config + live reachability probe
++ * GET /tiers/policy read-only R2 popularity policy: each threshold with its source (default |
++ * setting), the promote/demote rules, providers, and this app's recent
++ * decisions (media_tier_decisions) with 24-hour counts
++ * GET /tiers/decisions this app's R2 promote/demote decisions, newest first
++ * (?vod_id&action=promote|demote&outcome&limit≤200&before_id)
*/
'use strict';
@@@ -328,6 +333,67 @@ router.put('/tiers/settings', (req, res) => {
}
});
++// ── R2 policy and decision log (read-only) ──────────────────
++function _decisionPublic(r) {
++ const parse = (v) => { try { return JSON.parse(v || '{}'); } catch { return {}; } };
++ return {
++ id: r.id, decided_at: r.decided_at, vod_id: r.vod_id, object_id: r.object_id || null, action: r.action,
++ from_provider: r.from_provider, to_provider: r.to_provider, outcome: r.outcome, trigger: r.trigger, reason: r.reason,
++ inputs: parse(r.inputs), thresholds: parse(r.thresholds), error: r.error || null,
++ };
++}
++
++router.get('/tiers/policy', (req, res) => {
++ try {
++ const status = vodStorage.getStatus();
++ const counts = { promote: { done: 0, already: 0, refused: 0, failed: 0 }, demote: { done: 0, already: 0, refused: 0, failed: 0 } };
++ for (const r of db.all(`SELECT action, outcome, COUNT(*) AS n FROM media_tier_decisions
++ WHERE app_id = ? AND decided_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 day') GROUP BY action, outcome`, [req.appId])) {
++ if (counts[r.action]) counts[r.action][r.outcome] = r.n;
++ }
++ res.json({
++ r2: {
++ ...vodStorage.r2Policy(status.settings),
++ enabled: !!status.settings.r2Enabled,
++ provider: status.providers.r2,
++ canonical: status.providers.b2,
++ },
++ decisions: {
++ last_24h: counts,
++ recent: db.all('SELECT * FROM media_tier_decisions WHERE app_id = ? ORDER BY id DESC LIMIT 20', [req.appId]).map(_decisionPublic),
++ },
++ note: 'Read-only. Thresholds change through PUT /tiers/settings (media_settings storage_tier.*); decisions are this app\'s only.',
++ });
++ } catch (err) {
++ console.error('[Admin] Tier policy error:', err.message);
++ res.status(500).json({ error: 'Failed to read the tier policy' });
++ }
++});
++
++router.get('/tiers/decisions', (req, res) => {
++ try {
++ const q = req.query;
++ const conds = ['app_id = ?'], params = [req.appId];
++ if (q.vod_id != null) { conds.push('vod_id = ?'); params.push(parseInt(q.vod_id, 10) || 0); }
++ if (q.action) {
++ if (!['promote', 'demote'].includes(String(q.action))) return res.status(400).json({ error: 'action must be promote or demote' });
++ conds.push('action = ?'); params.push(String(q.action));
++ }
++ if (q.outcome) {
++ if (!['done', 'already', 'refused', 'failed'].includes(String(q.outcome))) return res.status(400).json({ error: 'outcome must be done, already, refused or failed' });
++ conds.push('outcome = ?'); params.push(String(q.outcome));
++ }
++ if (q.before_id != null) { conds.push('id < ?'); params.push(parseInt(q.before_id, 10) || 0); }
++ const limit = Math.min(Math.max(parseInt(q.limit, 10) || 50, 1), 200);
++ const rows = db.all(`SELECT * FROM media_tier_decisions WHERE ${conds.join(' AND ')} ORDER BY id DESC LIMIT ?`, [...params, limit + 1]);
++ const page = rows.slice(0, limit);
++ res.json({ decisions: page.map(_decisionPublic), next_before_id: rows.length > limit ? page[page.length - 1].id : null, limit });
++ } catch (err) {
++ console.error('[Admin] Tier decisions error:', err.message);
++ res.status(500).json({ error: 'Failed to list tier decisions' });
++ }
++});
++
// ── POST /tiers/sweep — trigger a sweep now ──────────────────
router.post('/tiers/sweep', async (req, res) => {
try {
@@@ -341,18 +407,19 @@ router.post('/tiers/sweep', async (req, res) => {
});
// target → mover (predecessor's hot/cold vocabulary kept as aliases)
++// ctx { trigger, reason } goes into the R2 decision log (media_tier_decisions) for moves into or out of R2.
const MOVERS = {
-- local: (id) => vodStorage.moveToHot(id),
-- hot: (id) => vodStorage.moveToHot(id),
++ local: (id, ctx) => vodStorage.moveToHot(id, ctx),
++ hot: (id, ctx) => vodStorage.moveToHot(id, ctx),
b2: (id) => vodStorage.moveToCold(id),
cold: (id) => vodStorage.moveToCold(id),
-- r2: (id) => vodStorage.promoteToR2(id),
++ r2: (id, ctx) => vodStorage.promoteToR2(id, ctx),
};
async function _moveScoped(appId, rawId, target) {
const id = parseInt(rawId, 10);
if (!Number.isFinite(id) || !db.getVodById(id, appId)) return { ok: false, error: 'VOD not found' };
-- return MOVERS[target](id);
++ return MOVERS[target](id, { trigger: 'admin', reason: `admin move to ${target} (app ${appId})` });
}
// ── POST /tiers/move — move one VOD between tiers ────────────
ddiff --git a/server/db/schema.sql b/server/db/schema.sql
iindex 8fa0d7c..c72d808 100644
---- a/server/db/schema.sql
++++ b/server/db/schema.sql
@@@ -360,6 +360,29 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (
resolved_at DATETIME
);
++-- R2 tier decisions (server/vod/vod-storage.js recordTierDecision): one row for every promotion to and
++-- demotion from the R2 popularity cache, whoever asked (the sweep, an admin move, the eviction drill),
++-- with the inputs it saw, the policy thresholds in force and the reason. Read by the staff policy
++-- endpoint (GET /api/v1/:app/admin/storage/tiers/policy and /tiers/decisions).
++CREATE TABLE IF NOT EXISTS media_tier_decisions (
++ id INTEGER PRIMARY KEY AUTOINCREMENT,
++ decided_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
++ vod_id INTEGER NOT NULL,
++ app_id TEXT,
++ object_id TEXT,
++ action TEXT NOT NULL CHECK(action IN ('promote', 'demote')),
++ from_provider TEXT,
++ to_provider TEXT,
++ outcome TEXT NOT NULL CHECK(outcome IN ('done', 'already', 'refused', 'failed')),
++ trigger TEXT NOT NULL, -- sweep | admin | drill | clip | manual
++ reason TEXT NOT NULL,
++ inputs TEXT NOT NULL DEFAULT '{}', -- JSON: view_count, last_accessed_at, storage_provider, is_recording, held, file_size, created_at
++ thresholds TEXT NOT NULL DEFAULT '{}',-- JSON: the R2 policy in force, each value with its source (default | setting)
++ error TEXT
++);
++CREATE INDEX IF NOT EXISTS idx_media_tier_decisions_vod ON media_tier_decisions(vod_id, id);
++CREATE INDEX IF NOT EXISTS idx_media_tier_decisions_app ON media_tier_decisions(app_id, id);
++
-- Object changes waiting to become events (server/events.js recordObjectChanges): media_objects
-- triggers (database.js ensureObjectTriggers) write one row per visibility change and per deletion,
-- in the transaction that makes the change, whatever path made it; the rows become
ddiff --git a/server/observability.js b/server/observability.js
iindex 367f391..e35b385 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -56,6 +56,11 @@ function storageDirs(config) {
function domainMetrics(registry, { db, recorder, events }) {
registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });
++ registry.gauge({
++ name: 'media_tier_decisions_24h', help: 'R2 promotions and demotions decided in the last 24 hours, by action and outcome (media_tier_decisions)', labelNames: ['action', 'outcome'],
++ collect: () => db.all(`SELECT action, outcome, COUNT(*) AS n FROM media_tier_decisions WHERE decided_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 day')
++ GROUP BY action, outcome`).map((r) => ({ labels: { action: r.action, outcome: r.outcome }, value: r.n })),
++ });
registry.gauge({
name: 'media_object_locations', help: 'Stored copies of media objects by provider and verification state', labelNames: ['provider', 'state'],
collect: () => db.all('SELECT provider, state, COUNT(*) AS n FROM media_locations GROUP BY provider, state').map((r) => ({ labels: { provider: r.provider, state: r.state }, value: r.n })),
ddiff --git a/server/vod/clip-jobs.js b/server/vod/clip-jobs.js
iindex d77aae4..3b56b85 100644
---- a/server/vod/clip-jobs.js
++++ b/server/vod/clip-jobs.js
@@@ -62,7 +62,7 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
const need = (Number(vod.file_size) || 0) * HOT_FETCH_FREE_MULTIPLE;
if (need && freeBytes() > need) {
console.log(`[Clips] clip ${clipId}: pulling vod ${vod.id} (${(vod.file_size / 1048576).toFixed(0)} MB) back to local disk for a reliable cut`);
-- const r = await vodStorage.moveToHot(vod.id).catch(e => ({ ok: false, error: e.message }));
++ const r = await vodStorage.moveToHot(vod.id, { trigger: 'clip', reason: `fetched to local disk to cut clip ${clipId}` }).catch(e => ({ ok: false, error: e.message }));
if (r && r.ok) source = await vodStorage.resolveMediaSource(db.get('SELECT * FROM vods WHERE id = ?', [vod.id]));
else console.warn(`[Clips] clip ${clipId}: hot fetch failed: ${r && r.error}`);
} else {
ddiff --git a/server/vod/vod-storage.js b/server/vod/vod-storage.js
iindex d5d6fca..f72b9ea 100644
---- a/server/vod/vod-storage.js
++++ b/server/vod/vod-storage.js
@@@ -138,6 +138,68 @@ function setSetting(key, value) {
else db.run('INSERT INTO media_settings (key, value) VALUES (?, ?)', [dbKey, JSON.stringify(value)]);
}
++/** Is this storage_tier.* setting overridden in media_settings? ('setting') or the built-in default ('default'). */
++function settingSource(key) {
++ try { return db.get('SELECT 1 AS x FROM media_settings WHERE key = ?', [`storage_tier.${key}`]) ? 'setting' : 'default'; } catch { return 'default'; }
++}
++
++// r2Enabled, r2MinViews, r2RecentAccessDays, r2MaxIdleDays, r2MaxPerSweep
++const R2_THRESHOLDS = Object.keys(DEFAULTS).filter(k => /^r2[A-Z]/.test(k));
++const R2_DEMOTE_PER_SWEEP = 10;
++
++/**
++ * The R2 popularity policy in force (read-only; PUT /admin/storage/tiers/settings changes it):
++ * each threshold as { value, default, source }, plus the rules the sweep applies.
++ */
++function r2Policy(settings = getSettings()) {
++ const t = {};
++ for (const k of R2_THRESHOLDS) t[k] = { value: settings[k], default: DEFAULTS[k], source: settingSource(k) };
++ return {
++ thresholds: t,
++ promote: `a local or B2 VOD that is not recording, with view_count >= r2MinViews (${settings.r2MinViews}) and last accessed within r2RecentAccessDays (${settings.r2RecentAccessDays} days); at most r2MaxPerSweep (${settings.r2MaxPerSweep}) per sweep`,
++ demote: `an R2 VOD not accessed for r2MaxIdleDays (${settings.r2MaxIdleDays} days), or never; at most ${R2_DEMOTE_PER_SWEEP} per sweep; the B2 canonical copy is checked (or copied back) first, and a held VOD is never moved`,
++ demote_per_sweep: R2_DEMOTE_PER_SWEEP,
++ };
++}
++
++function _thresholdSnapshot(settings = getSettings()) {
++ const out = {};
++ for (const k of R2_THRESHOLDS) out[k] = { value: settings[k], source: settingSource(k) };
++ return out;
++}
++
++function _tierInputs(vod) {
++ if (!vod) return {};
++ return {
++ view_count: Number(vod.view_count) || 0, last_accessed_at: vod.last_accessed_at || null, storage_provider: providerOf(vod),
++ is_recording: !!vod.is_recording, held: _held(vod), file_size: Number(vod.file_size) || 0, created_at: vod.created_at || null,
++ };
++}
++
++function _outcomeOf(result, action, from) {
++ if (!result) return 'failed';
++ // Promoting a VOD R2 already serves (the copy is re-checked) changes nothing: 'already'.
++ if (result.ok) return result.already || (action === 'promote' && from === 'r2') ? 'already' : 'done';
++ if (result.held || /recording|not configured|not available|not found|No source/i.test(String(result.error || ''))) return 'refused';
++ return 'failed';
++}
++
++/**
++ * Log one R2 promotion or demotion (media_tier_decisions): what the VOD looked like, the thresholds
++ * in force, why, and how it ended. Never throws: the move already happened (or did not).
++ */
++function recordTierDecision({ vod, vodId, action, from, to, result, trigger = 'manual', reason = 'requested' }) {
++ try {
++ db.run(`INSERT INTO media_tier_decisions (vod_id, app_id, object_id, action, from_provider, to_provider, outcome, trigger, reason, inputs, thresholds, error)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
++ [Number(vodId), vod ? vod.app_id : null, vod ? vod.object_id || null : null, action, from || null, to || null, _outcomeOf(result, action, from),
++ String(trigger).slice(0, 40), String(reason).slice(0, 500), JSON.stringify(_tierInputs(vod)), JSON.stringify(_thresholdSnapshot()),
++ result && !result.ok ? String(result.error || 'failed').slice(0, 500) : null]);
++ } catch (err) {
++ console.warn(`[VodStorage] tier decision for VOD ${vodId} not logged: ${err.message}`);
++ }
++}
++
// ── Provider clients ─────────────────────────────────────────
function providerConfigured(name) {
@@@ -457,7 +519,17 @@ function cleanupStaleDownloads() {
} catch { /* */ }
}
--async function moveToHot(vodId) {
++/** Restore an offloaded VOD to local disk. Restoring one served from R2 drops its R2 copy: logged as a demotion. */
++async function moveToHot(vodId, ctx = {}) {
++ const before = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
++ const result = await _moveToHot(vodId);
++ if (before && providerOf(before) === 'r2') {
++ recordTierDecision({ vod: before, vodId, action: 'demote', from: 'r2', to: 'local', result, trigger: ctx.trigger || 'manual', reason: ctx.reason || 'restored to local disk' });
++ }
++ return result;
++}
++
++async function _moveToHot(vodId) {
const vod = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
if (!vod || !vod.file_path) return { ok: false, error: 'VOD not found' };
@@@ -525,8 +597,18 @@ async function moveToHot(vodId) {
}
}
--/** Promote a popular VOD to R2 (free egress). Keeps the B2 canonical copy. */
--async function promoteToR2(vodId) {
++/**
++ * Promote a VOD to R2 (free egress), keeping the B2 canonical copy, and log the decision.
++ * ctx { trigger: sweep | admin | drill | manual, reason } says who asked and why.
++ */
++async function promoteToR2(vodId, ctx = {}) {
++ const before = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
++ const result = await _promoteToR2(vodId);
++ recordTierDecision({ vod: before, vodId, action: 'promote', from: before ? providerOf(before) : null, to: 'r2', result, ...ctx });
++ return result;
++}
++
++async function _promoteToR2(vodId) {
const vod = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
if (!vod || !vod.file_path) return { ok: false, error: 'VOD not found' };
if (vod.is_recording) return { ok: false, error: 'VOD is currently recording' };
@@@ -571,8 +653,15 @@ async function promoteToR2(vodId) {
}
}
--/** Demote a stale R2 VOD back to B2-only. */
--async function demoteFromR2(vodId) {
++/** Demote an R2 VOD back to B2-only, and log the decision (ctx as for promoteToR2). */
++async function demoteFromR2(vodId, ctx = {}) {
++ const before = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
++ const result = await _demoteFromR2(vodId);
++ recordTierDecision({ vod: before, vodId, action: 'demote', from: before ? providerOf(before) : null, to: 'b2', result, ...ctx });
++ return result;
++}
++
++async function _demoteFromR2(vodId) {
const vod = db.get('SELECT * FROM vods WHERE id = ?', [vodId]);
if (!vod) return { ok: false, error: 'VOD not found' };
if (providerOf(vod) !== 'r2') return { ok: true, already: true };
@@@ -919,7 +1008,7 @@ async function runSweep() {
// 2) R2 promotion for popular VODs
if (settings.r2Enabled && providerConfigured('r2') && providerHealthy.r2 !== false) {
const popular = db.all(`
-- SELECT id FROM vods
++ SELECT id, view_count, last_accessed_at FROM vods
WHERE COALESCE(storage_provider, 'local') IN ('local', 'b2')
AND COALESCE(is_recording, 0) = 0
AND COALESCE(view_count, 0) >= ?
@@@ -928,7 +1017,8 @@ async function runSweep() {
LIMIT ?
`, [settings.r2MinViews, `-${settings.r2RecentAccessDays} days`, settings.r2MaxPerSweep]);
for (const vod of popular) {
-- const result = await promoteToR2(vod.id);
++ const result = await promoteToR2(vod.id, { trigger: 'sweep',
++ reason: `view_count ${vod.view_count || 0} >= r2MinViews ${settings.r2MinViews} and last accessed ${vod.last_accessed_at} (within r2RecentAccessDays ${settings.r2RecentAccessDays})` });
if (result.ok && !result.already) { promoted++; bytesFreed += result.bytes || 0; }
else if (result.held) skippedHeld++;
else if (!result.ok) errors.push({ id: vod.id, error: result.error });
@@@ -936,13 +1026,14 @@ async function runSweep() {
// 3) R2 demotion for stale VODs
const stale = db.all(`
-- SELECT id FROM vods
++ SELECT id, last_accessed_at FROM vods
WHERE storage_provider = 'r2'
AND (last_accessed_at IS NULL OR last_accessed_at <= datetime('now', ?))
-- LIMIT 10
++ LIMIT ${R2_DEMOTE_PER_SWEEP}
`, [`-${settings.r2MaxIdleDays} days`]);
for (const vod of stale) {
-- const result = await demoteFromR2(vod.id);
++ const result = await demoteFromR2(vod.id, { trigger: 'sweep',
++ reason: `last accessed ${vod.last_accessed_at || 'never'} (idle longer than r2MaxIdleDays ${settings.r2MaxIdleDays})` });
if (result.ok && !result.already) demoted++;
}
}
@@@ -1295,6 +1386,8 @@ module.exports = {
moveToHot,
promoteToR2,
demoteFromR2,
++ r2Policy,
++ recordTierDecision,
deleteVodObjects,
deleteObject,
deleteLegacyPasteScreenshot,
ddiff --git a/test/tier-decisions.test.js b/test/tier-decisions.test.js
nnew file mode 100644
iindex 0000000..e533dd2
---- /dev/null
++++ b/test/tier-decisions.test.js
@@@ -0,0 +1,191 @@
++'use strict';
++// R2 tier decisions (media_tier_decisions; server/vod/vod-storage.js) and the staff policy endpoint
++// (GET /api/v1/:app/admin/storage/tiers/policy, /tiers/decisions): every promotion to and demotion
++// from R2, whoever asked (the sweep, an admin move, a clip hot-fetch, the eviction drill), is logged
++// with the inputs it saw, the thresholds in force (each with its source) and the reason, whatever the
++// outcome (done, already, refused, failed). The thresholds themselves are unchanged. A fake S3 (B2 and
++// R2 buckets) is spoken to by the real AWS SDK through the storage engine.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-tierlog-'));
++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++
++const buckets = { 'b2-bucket': new Map(), 'r2-bucket': new Map() };
++let refusePutTo = null;
++function decodeAwsChunked(buf) {
++ const out = [];
++ let i = 0;
++ for (;;) {
++ const nl = buf.indexOf('\r\n', i);
++ const size = parseInt(buf.subarray(i, nl).toString().split(';')[0], 16);
++ if (!size) break;
++ out.push(buf.subarray(nl + 2, nl + 2 + size));
++ i = nl + 2 + size + 2;
++ }
++ return Buffer.concat(out);
++}
++const s3 = http.createServer((req, res) => {
++ const chunks = [];
++ req.on('data', (c) => chunks.push(c));
++ req.on('end', () => {
++ const u = new URL(req.url, 'http://x');
++ const [, bucket, ...rest] = decodeURIComponent(u.pathname).split('/');
++ const key = rest.join('/');
++ const store = buckets[bucket];
++ if (!store) { res.writeHead(404); return res.end(); }
++ if (!key && req.method === 'HEAD') { res.writeHead(200); return res.end(); }
++ const obj = store.get(key);
++ if (req.method === 'HEAD' || req.method === 'GET') {
++ if (!obj) { res.writeHead(404, { 'Content-Type': 'application/xml' }); return res.end(req.method === 'GET' ? '<Error><Code>NoSuchKey</Code></Error>' : undefined); }
++ res.writeHead(200, { 'Content-Length': obj.length, ETag: '"e"', 'Content-Type': 'video/webm' });
++ return res.end(req.method === 'GET' ? obj : undefined);
++ }
++ if (req.method === 'PUT') {
++ if (refusePutTo === bucket) { res.writeHead(500, { 'Content-Type': 'application/xml' }); return res.end('<Error><Code>InternalError</Code></Error>'); }
++ let body = Buffer.concat(chunks);
++ if (req.headers['x-amz-decoded-content-length'] || /aws-chunked/.test(req.headers['content-encoding'] || '')) body = decodeAwsChunked(body);
++ store.set(key, body);
++ res.writeHead(200, { ETag: '"e"' });
++ return res.end();
++ }
++ if (req.method === 'DELETE') { store.delete(key); res.writeHead(204); return res.end(); }
++ res.writeHead(405); res.end();
++ });
++});
++
++(async () => {
++ await new Promise((r) => s3.listen(0, '127.0.0.1', r));
++ const endpoint = `http://127.0.0.1:${s3.address().port}`;
++ Object.assign(process.env, {
++ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
++ MEDIA_B2_ENDPOINT: endpoint, MEDIA_B2_BUCKET: 'b2-bucket', MEDIA_B2_KEY_ID: 'k', MEDIA_B2_APP_KEY: 's', MEDIA_B2_REGION: 'us-west-004',
++ MEDIA_R2_ENDPOINT: endpoint, MEDIA_R2_BUCKET: 'r2-bucket', MEDIA_R2_ACCESS_KEY_ID: 'k', MEDIA_R2_SECRET_ACCESS_KEY: 's',
++ AWS_REQUEST_CHECKSUM_CALCULATION: 'WHEN_REQUIRED', AWS_RESPONSE_CHECKSUM_VALIDATION: 'WHEN_REQUIRED',
++ });
++ const db = require('../server/db/database');
++ const model = require('../server/objects/model');
++ const storage = require('../server/vod/vod-storage');
++ db.upsertApp({ app_id: 'live', api_key: 'live-key-tierlog' });
++ db.upsertApp({ app_id: 'games', api_key: 'games-key-tierlog' });
++
++ // The thresholds are not changed by this work.
++ assert.deepStrictEqual([storage.DEFAULTS.r2Enabled, storage.DEFAULTS.r2MinViews, storage.DEFAULTS.r2RecentAccessDays, storage.DEFAULTS.r2MaxIdleDays, storage.DEFAULTS.r2MaxPerSweep],
++ [true, 20, 3, 14, 5]);
++ // Keep the sweep's disk-pressure drain out of this test (whatever this machine's disk looks like).
++ storage.setSetting('hotDiskPressurePct', 101);
++ storage.setSetting('criticalDiskPct', 101);
++ storage.setSetting('minFreeGb', 0);
++
++ const bytes = crypto.randomBytes(200000);
++ const addVod = (id, { app = 'live', provider = 'local', views = 0, accessed = null } = {}) => {
++ const name = `vod-${app}-${id}-1726900000000.webm`;
++ const key = `vods/${name}`;
++ if (provider === 'local') fs.writeFileSync(path.join(process.env.VOD_PATH, name), bytes);
++ else { buckets['b2-bucket'].set(key, bytes); if (provider === 'r2') buckets['r2-bucket'].set(key, bytes); }
++ db.run(`INSERT INTO vods (id, app_id, title, file_path, file_size, is_public, visibility, duration_seconds, storage_provider, storage_key, view_count, last_accessed_at, created_at)
++ VALUES (?, ?, 't', ?, ?, 1, 'public', 60, ?, ?, ?, ${accessed ? `datetime('now', '${accessed}')` : 'NULL'}, datetime('now'))`,
++ [id, app, path.join(process.env.VOD_PATH, name), bytes.length, provider, provider === 'local' ? null : key, views]);
++ model.sync('vod', id);
++ return key;
++ };
++ const decisions = (vodId) => db.all('SELECT * FROM media_tier_decisions WHERE vod_id = ? ORDER BY id', [vodId]);
++
++ // ── 1. The sweep: a popular VOD is promoted, an idle R2 VOD demoted, each with its reason ──
++ addVod(1, { views: 25, accessed: '-1 hours' });
++ addVod(2, { provider: 'r2', views: 3, accessed: '-30 days' });
++ addVod(3, { views: 5, accessed: '-1 hours' }); // below the threshold: no decision
++ const sweep = await storage.runSweep();
++ assert.deepStrictEqual([sweep.promoted, sweep.demoted], [1, 1], JSON.stringify(sweep));
++ const [p] = decisions(1);
++ assert.deepStrictEqual([p.action, p.from_provider, p.to_provider, p.outcome, p.trigger], ['promote', 'local', 'r2', 'done', 'sweep']);
++ assert.ok(/view_count 25 >= r2MinViews 20/.test(p.reason) && /r2RecentAccessDays 3/.test(p.reason), p.reason);
++ const pin = JSON.parse(p.inputs);
++ assert.deepStrictEqual([pin.view_count, pin.storage_provider, pin.is_recording, pin.held], [25, 'local', false, false]);
++ assert.deepStrictEqual(JSON.parse(p.thresholds).r2MinViews, { value: 20, source: 'default' });
++ const [d] = decisions(2);
++ assert.deepStrictEqual([d.action, d.from_provider, d.to_provider, d.outcome, d.trigger], ['demote', 'r2', 'b2', 'done', 'sweep']);
++ assert.ok(/idle longer than r2MaxIdleDays 14/.test(d.reason), d.reason);
++ assert.strictEqual(decisions(3).length, 0);
++ console.log('✅ sweep promotions and demotions are logged with inputs, thresholds (and their source) and the reason');
++
++ // ── 2. Every other path and outcome: admin move (refused: held; failed: R2 down), clip hot-fetch, drill ──
++ const express = require('express');
++ const app = express();
++ app.use(express.json());
++ app.use('/api/v1/:app/admin/storage', require('../server/admin/routes'));
++ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++ const call = (method, p, body, key = 'live-key-tierlog') => new Promise((resolve, reject) => {
++ const data = body ? JSON.stringify(body) : '';
++ const req = http.request({ host: '127.0.0.1', port: server.address().port, path: p, method,
++ headers: { Authorization: `Bearer ${key}`, 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },
++ (res) => { let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, body: JSON.parse(b || '{}') })); });
++ req.on('error', reject);
++ req.end(data);
++ });
++ addVod(4, { views: 1 });
++ const hold = model.placeHold({ object_id: db.get('SELECT object_id FROM vods WHERE id = 4').object_id, kind: 'creator_pin', reason: 'test' });
++ let r = await call('POST', '/api/v1/live/admin/storage/tiers/move', { vod_id: 4, target: 'r2' });
++ assert.strictEqual(r.body.held, true);
++ assert.deepStrictEqual([decisions(4)[0].outcome, decisions(4)[0].trigger], ['refused', 'admin']);
++ assert.ok(/admin move to r2/.test(decisions(4)[0].reason));
++ assert.strictEqual(JSON.parse(decisions(4)[0].inputs).held, true);
++ model.releaseHold(hold.id, 'test');
++ refusePutTo = 'r2-bucket';
++ r = await call('POST', '/api/v1/live/admin/storage/tiers/move', { vod_id: 4, target: 'r2' });
++ refusePutTo = null;
++ assert.strictEqual(r.body.ok, false);
++ const failed = decisions(4)[1];
++ assert.deepStrictEqual([failed.outcome, failed.action], ['failed', 'promote']);
++ assert.ok(failed.error);
++ r = await call('POST', '/api/v1/live/admin/storage/tiers/move', { vod_id: 1, target: 'r2' });
++ assert.strictEqual(decisions(1)[1].outcome, 'already', 'a promotion that finds it already there is logged too');
++ // Restoring an R2 VOD to local drops its R2 copy: a demotion.
++ await storage.moveToHot(1, { trigger: 'clip', reason: 'fetched to local disk to cut clip 9' });
++ const hot = decisions(1)[2];
++ assert.deepStrictEqual([hot.action, hot.from_provider, hot.to_provider, hot.outcome, hot.trigger], ['demote', 'r2', 'local', 'done', 'clip']);
++ await storage.demoteFromR2(2, { trigger: 'drill', reason: 'R2 eviction drill: evict the R2 copy' });
++ assert.deepStrictEqual([decisions(2)[1].outcome, decisions(2)[1].trigger], ['already', 'drill']);
++ addVod(50, { app: 'games', views: 30, accessed: '-1 hours' });
++ await storage.promoteToR2(50, { trigger: 'admin', reason: 'games' });
++ console.log('✅ admin moves, clip hot-fetches and the drill are logged; refused (held), failed and already outcomes too');
++
++ // ── 3. The read-only staff policy endpoint and the decision list (app-scoped) ──
++ r = await call('GET', '/api/v1/live/admin/storage/tiers/policy');
++ assert.strictEqual(r.status, 200);
++ const pol = r.body;
++ assert.deepStrictEqual(pol.r2.thresholds.r2MinViews, { value: 20, default: 20, source: 'default' });
++ assert.deepStrictEqual(pol.r2.thresholds.r2MaxIdleDays, { value: 14, default: 14, source: 'default' });
++ assert.strictEqual(pol.r2.enabled, true);
++ assert.ok(/view_count >= r2MinViews \(20\)/.test(pol.r2.promote) && /r2MaxIdleDays \(14 days\)/.test(pol.r2.demote));
++ assert.strictEqual(pol.r2.provider.configured, true);
++ assert.deepStrictEqual(pol.decisions.last_24h.promote, { done: 1, already: 1, refused: 1, failed: 1 });
++ assert.deepStrictEqual(pol.decisions.last_24h.demote, { done: 2, already: 1, refused: 0, failed: 0 });
++ assert.ok(pol.decisions.recent.every(x => x.vod_id !== 50), 'another app\'s decisions are not listed');
++ assert.ok(pol.decisions.recent[0].inputs && pol.decisions.recent[0].thresholds, 'each decision carries its inputs and thresholds');
++ await call('PUT', '/api/v1/live/admin/storage/tiers/settings', { r2MinViews: 20 });
++ r = await call('GET', '/api/v1/live/admin/storage/tiers/policy');
++ assert.deepStrictEqual(r.body.r2.thresholds.r2MinViews, { value: 20, default: 20, source: 'setting' }, 'an override shows as a setting');
++ r = await call('GET', '/api/v1/live/admin/storage/tiers/decisions?action=demote&limit=2');
++ assert.deepStrictEqual([r.body.decisions.length, r.body.decisions.every(x => x.action === 'demote'), r.body.next_before_id != null], [2, true, true]);
++ const next = await call('GET', `/api/v1/live/admin/storage/tiers/decisions?action=demote&limit=2&before_id=${r.body.next_before_id}`);
++ assert.strictEqual(next.body.decisions.length, 1);
++ r = await call('GET', '/api/v1/live/admin/storage/tiers/decisions?vod_id=4');
++ assert.deepStrictEqual(r.body.decisions.map(x => x.outcome), ['failed', 'refused']);
++ assert.strictEqual((await call('GET', '/api/v1/live/admin/storage/tiers/decisions?action=move')).status, 400);
++ assert.strictEqual((await call('GET', '/api/v1/games/admin/storage/tiers/decisions', null, 'games-key-tierlog')).body.decisions.length, 1);
++ assert.strictEqual((await call('GET', '/api/v1/live/admin/storage/tiers/policy', null, 'wrong-key')).status >= 401, true, 'staff only (the app key)');
++ console.log('✅ GET /tiers/policy: thresholds with their source, the rules, 24 h counts and recent decisions; /tiers/decisions filters and pages');
++
++ server.close();
++ s3.close();
++ db.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('\n✅ All tier decision tests passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });