openvibers/openvibe.media Commits
Alert when a placement move fails three times; hold dry-run ceilings (#24)
Changes
@@@ -134,3 +134,4 @@ MEDIA_CAPABILITY_GATE=report#MEDIA_JOBS_HEAVY_CONCURRENCY=1#MEDIA_JOBS_HEAVY_WHILE_RECORDING=0#MEDIA_INVARIANT_SCAN_HOURS=24 # the size-invariant validator (proposes split/remux; never runs them)++#MEDIA_MOVE_CLEANUP_MINUTES=60 # storage.move.cleanup: alert on objects whose placement move failed 3 times (0 = never)@@@ -123,7 +123,12 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:`backup` (archive fabric only).- **Every move is two-phase:** verify the new copy (size + sha256), commit the location in one transaction, thendelete old bytes after commit, with a retrying cleanup job and an alert after three failures. The row never points-- at bytes that do not exist.++ at bytes that do not exist. The sweep retries a failed move after 6 h and logs every failure as a `failed` row in++ `media_object_tier_decisions`; the job `storage.move.cleanup` (`server/jobs/move-cleanup.js`, light lane, system app++ only, report only) reads those rows every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never; not in a restore drill): an++ object whose move failed three times in the last 7 days (a later `done` of that move resets the count) raises one++ `storage.alert` of kind `move_cleanup_failed`, listing the objects (`media_storage_alerts_total{kind}`, the Events++ outbox and the per-app webhook, once per 6 h cooldown like every storage alert).- **Dry-run and simulator:** every policy can run in dry-run mode (logs what it would do and what it would cost), and asimulator replays recorded telemetry against a candidate policy before it is enabled.- **Demand rollups (F2.4, shipped):** every served viewer read (`router.route()` for playback or download, not a derive@@@ -164,7 +169,9 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`: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` followsonly 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)++ against it, so dry runs (gate off) are refused as a live sweep would be, without changing `spend` (a dry run++ repeated the same day is logged once but still counts, and a dry-run demotion checks its canonical copy like a live++ one before it frees any projected cost). 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@@@ -228,8 +235,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr- **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'seligibility 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.++ and per-class monthly R2 storage ceilings on its moves (F2.6, shipped), the move cleanup job and its alert after++ three failures (§6, shipped). Still open in F2: 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@@@ -758,6 +758,7 @@ job over only with the token it read and, past boot, only while the lease is sti| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) || `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) || `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |++| `storage.move.cleanup` | light | Placement moves whose delete keeps failing: an object with three `failed` rows in `media_object_tier_decisions` in the last 7 days (a later `done` resets them) raises one `storage.alert` of kind `move_cleanup_failed`. Service-wide under the system app `_media`; report only. `params {}`. Result `{ scanned, alerted, objects }`. Scheduled every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never); tenants cannot enqueue it (403) || `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and toonnew file mode 100644@@@ -0,0 +1,82 @@++/**++ * OpenVibe.Media — job type storage.move.cleanup (light lane): the loud half of a two-phase move's second phase++ * (docs/media-fabric.md §6). Once a move's new location is committed the old bytes are deleted; the sweep++ * (server/objects/tiering.js) retries a failed delete after FAILED_BACKOFF_H, and every failure is a `failed` row++ * in media_object_tier_decisions. This job reads those rows: an object whose move failed FAIL_THRESHOLD times++ * within the lookback (a later `done` of the same move starts the count over) raises one storage.alert of kind++ * move_cleanup_failed, which vod-storage's emitStorageEvent counts in media_storage_alerts_total{kind} and++ * delivers as the Events outbox row and per-app webhook. Report only: it never moves, deletes or changes a row.++ *++ * params {} service-wide; runs only under queue.SYSTEM_APP (no tenant can enqueue it)++ * result { scanned, alerted, objects } objects = the first 50 (object_id, action, fails, first_at, last_at, last_error)++ *++ * Scheduled every MEDIA_MOVE_CLEANUP_MINUTES (60; 0 = never) by the worker, which does not start in a restore drill.++ */++'use strict';++++const db = require('../db/database');++const queue = require('./queue');++++const { JobError } = queue;++const TYPE = 'storage.move.cleanup';++const FAIL_THRESHOLD = 3;++const LOOKBACK_H = 24 * 7;++const LISTED = 50;++++function envInt(name, fallback) {++ const v = parseInt(process.env[name] || '', 10);++ return Number.isFinite(v) ? v : fallback;++}++++/** Moves (object + action) with at least `threshold` failures in the window since their last `done`. */++async function failedMoves({ threshold = FAIL_THRESHOLD, lookbackH = LOOKBACK_H } = {}) {++ return (await db.all(`++ SELECT f.object_id, f.action, MAX(f.app_id) AS app_id, COUNT(*) AS fails,++ MIN(f.decided_at) AS first_at, MAX(f.decided_at) AS last_at,++ (ARRAY_AGG(f.error ORDER BY f.decided_at DESC, f.id DESC))[1] AS last_error++ FROM media_object_tier_decisions f++ WHERE f.outcome = 'failed' AND f.decided_at >= ov_now_iso(?)++ AND NOT EXISTS (SELECT 1 FROM media_object_tier_decisions d++ WHERE d.object_id = f.object_id AND d.action = f.action AND d.outcome = 'done' AND d.decided_at > f.decided_at)++ GROUP BY f.object_id, f.action++ HAVING COUNT(*) >= ?++ ORDER BY MAX(f.decided_at) DESC, f.object_id`, [`-${lookbackH} hours`, threshold])).map((r) => ({ ...r, fails: Number(r.fails) }));++}++++function validate({ appId, obj }) {++ if (appId !== queue.SYSTEM_APP) {++ throw new JobError('media.job.forbidden', `${TYPE} covers the whole service's placement: it runs on its schedule`, { status: 403, permanent: true });++ }++ if (obj) throw new JobError('media.job.invalid', `${TYPE} covers every object: leave object_id out`, { permanent: true });++ return {};++}++++async function run() {++ const rows = await failedMoves();++ if (!rows.length) return { scanned: 0, alerted: 0, objects: [] };++ const objects = rows.slice(0, LISTED);++ // emitStorageEvent counts media_storage_alerts_total{kind="move_cleanup_failed"} itself (one alert per cooldown, so the++ // payload carries the list and the kind stays a fixed label, never per object).++ const sent = await require('../vod/vod-storage').emitStorageEvent('storage.alert', 'move_cleanup_failed', { count: rows.length, objects });++ return { scanned: rows.length, alerted: sent ? rows.length : 0, objects };++}++++/** Queue one run every MEDIA_MOVE_CLEANUP_MINUTES (0 = never). Returns how many were queued (0 or 1). */++async function schedule(nowMs = Date.now()) {++ const minutes = envInt('MEDIA_MOVE_CLEANUP_MINUTES', 60);++ if (!(minutes > 0)) return 0;++ const period = Math.floor(nowMs / (minutes * 60 * 1000));++ try {++ const r = await queue.enqueue({ appId: queue.SYSTEM_APP, type: TYPE, params: {}, idempotencyKey: `${TYPE}:${minutes}m:${period}`, createdBy: 'system:schedule' });++ return r.created ? 1 : 0;++ } catch (err) {++ console.warn(`[Jobs] could not schedule ${TYPE}: ${err.message}`);++ return 0;++ }++}++++module.exports = {++ TYPE, FAIL_THRESHOLD, failedMoves,++ spec: { lane: 'light', maxAttempts: 2, timeoutMs: 5 * 60 * 1000, needsObject: false, validate, run },++ schedule,++};@@@ -12,6 +12,7 @@* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)++ * storage.move.cleanup light the alert for placement moves whose delete keeps failing (3+ failed decisions), service-wide*/'use strict';@@@ -28,5 +29,6 @@ queue.register('clip.cut', require('../vod/clip-jobs').spec);queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);queue.register('object.hash', require('./content-hash').spec);queue.register('storage.orphans.scan', require('./storage-orphans').spec);++queue.register('storage.move.cleanup', require('./move-cleanup').spec);module.exports = { names: () => queue.typeNames() };@@@ -21,7 +21,8 @@* Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every* MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them). Orphaned recordings* get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js). The storage orphan-- * report (storage.orphans.scan, report only) runs every MEDIA_ORPHAN_SCAN_DAYS under queue.SYSTEM_APP.++ * report (storage.orphans.scan, report only) runs every MEDIA_ORPHAN_SCAN_DAYS under queue.SYSTEM_APP;++ * storage.move.cleanup (alerts on placement moves that failed three times) every MEDIA_MOVE_CLEANUP_MINUTES.*/'use strict';@@@ -158,6 +159,7 @@ async function tick() {try { await require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }try { await require('./content-hash').schedule(now); } catch (err) { console.warn('[Jobs] content hash schedule:', err.message); }try { await require('./storage-orphans').schedule(now); } catch (err) { console.warn('[Jobs] storage orphan scan schedule:', err.message); }++ try { await require('./move-cleanup').schedule(now); } catch (err) { console.warn('[Jobs] move cleanup schedule:', err.message); }}try { await require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }if (now - lastPruneAt > 6 * 3600 * 1000) {@@@ -337,13 +337,14 @@ async function demote(objectId, ctx = {}) {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' });if (!canon || canon.provider === 'r2') return await log('refused', `${why}; refused: no canonical copy besides R2 (the R2 copy is kept)`, { error: 'no canonical copy' });-- // The last good copy is never the one removed: the canonical copy must check out now.++ // The last good copy is never the one removed: the canonical copy must check out now. A dry run checks it too,++ // so it proposes (and projects the freed R2 cost of) only the demotions a live sweep could make.const check = await confirmCanonical(obj, canon);if (!check.ok) return await log('refused', `${why}; refused: ${check.why}; the R2 copy is kept (it may be the last good copy)`, { error: check.why });++ if (!settings.active) return await log('dry_run', `gate off (active = false): would demote: ${why}`);await db.getDb().tx(async () => {await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',@@@ -632,11 +633,12 @@ 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++ // A slot is spent by a move, a failed move, a dry run (a repeat of one logged earlier today included: it is only++ // not logged again, and a live sweep would make that move), 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 spends = (r, action) => (action === 'promote' && r.blocked === 'provider') || ['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.@@@ -656,7 +658,7 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {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') {++ if (['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);@@@ -687,8 +689,8 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {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;++ if (['done', 'dry_run'].includes(r.outcome)) projected[cls] += cost;++ if (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) };@@@ -1550,6 +1550,7 @@ async function getStatus() {module.exports = {cleanupStaleDownloads,++ emitStorageEvent,DEFAULTS,REMOTE_PROVIDERS,OFFLOADABLE_WHERE,nnew file mode 100644@@@ -0,0 +1,114 @@++'use strict';++// storage.move.cleanup (server/jobs/move-cleanup.js): the alert half of a two-phase move's cleanup. An object whose++// move failed three times in the last 7 days (a later `done` resets the count; refused, already and dry_run rows are++// not failures) raises one storage.alert of kind move_cleanup_failed: media_storage_alerts_total{kind} once, one++// Events outbox row carrying the list. The job is service-wide (queue.SYSTEM_APP only), scheduled every++// MEDIA_MOVE_CLEANUP_MINUTES (0 = never), and changes nothing.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-movecleanup-'));++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++ Object.assign(process.env, {++ MEDIA_DATA_DIR: path.join(tmp, 'data'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),++ });++ delete process.env.MEDIA_MOVE_CLEANUP_MINUTES;++ process.env.EVENTS_URL = '';++ const db = require('../server/db/database');++ const queue = require('../server/jobs/queue');++ require('../server/jobs/types');++ const job = require('../server/jobs/move-cleanup');++ const worker = require('../server/jobs/worker');++ const metrics = require('openvibe-shared/metrics');++ const registry = metrics.createRegistry();++ require('../server/observability').domainMetrics(registry, { db, recorder: { activeCount: () => 0 }, events: { status: async () => ({ enabled: false }) } });++ const alerts = async () => {++ const m = /media_storage_alerts_total\{kind="move_cleanup_failed"\} (\d+)/.exec(await registry.metricsAsync());++ return m ? Number(m[1]) : 0;++ };++ // The Events outbox is off in a test: count what vod-storage hands events.record (the outbox row's source).++ const recorded = [];++ const events = require('../server/events');++ const realRecord = events.record;++ events.record = async (event, appId, data) => { if (event === 'storage.alert' && data.kind === 'move_cleanup_failed') recorded.push(data); return realRecord(event, appId, data); };++ const outbox = async () => recorded.length;++ const row = (objectId, action, outcome, ago = '0 hours', error = null) => db.run(++ `INSERT INTO media_object_tier_decisions (decided_at, object_id, app_id, action, outcome, trigger, reason, error)++ VALUES (ov_now_iso(?), ?, 'live', ?, ?, 'sweep', 'test', ?)`, [`-${ago}`, objectId, action, outcome, error]);++ const many = async (n, id, action, startH, error = null) => { for (let i = 0; i < n; i++) await row(id, action, 'failed', `${startH + i} hours`, error); };++++ // ── The threshold, the lookback and what counts as a failure ──++ await many(3, 'obj-three', 'demote', 1, 'the R2 copy is still there after the delete');++ await many(2, 'obj-two', 'demote', 1, 'boom');++ for (const outcome of ['refused', 'dry_run', 'already', 'refused', 'dry_run']) await row('obj-refused', 'demote', outcome);++ await many(3, 'obj-old', 'demote', 24 * 7 + 5); // all three failures before the lookback++ await many(2, 'obj-split', 'demote', 1); await row('obj-split', 'promote', 'failed', '1 hours'); // 2 + 1 of another action++ await many(3, 'obj-reset', 'demote', 10); await row('obj-reset', 'demote', 'done', '5 hours'); // a done after them resets the count++ await many(3, 'obj-after', 'promote', 1); await row('obj-after', 'promote', 'done', '20 hours'); // the done is older: the failures count++ const found = await job.failedMoves();++ assert.deepStrictEqual(found.map((r) => `${r.object_id}:${r.action}:${r.fails}`).sort(), ['obj-after:promote:3', 'obj-three:demote:3'], JSON.stringify(found));++ assert.strictEqual(found.find((r) => r.object_id === 'obj-three').last_error, 'the R2 copy is still there after the delete');++ assert.strictEqual(job.FAIL_THRESHOLD, 3);++ console.log('✅ failedMoves: three failures within the lookback, per object and action; a later done, refusals and dry runs do not count');++++ // ── SYSTEM_APP only ──++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-movecleanup' });++ assert.throws(() => job.spec.validate({ appId: 'live', params: {} }), (e) => e.status === 403 && e.code === 'media.job.forbidden');++ assert.deepStrictEqual(job.spec.validate({ appId: queue.SYSTEM_APP, params: {} }), {});++ const express = require('express');++ const http = require('http');++ const app = express();++ app.use(express.json());++ app.use('/api/v2/:app/jobs', require('../server/jobs/routes'));++ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });++ const data = JSON.stringify({ type: job.TYPE });++ const res = await new Promise((resolve, reject) => {++ const rq = http.request({ host: '127.0.0.1', port: server.address().port, path: '/api/v2/live/jobs', method: 'POST',++ headers: { Authorization: 'Bearer live-key-movecleanup', 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },++ (r) => { let b = ''; r.on('data', (c) => { b += c; }); r.on('end', () => resolve({ status: r.statusCode, body: JSON.parse(b || '{}') })); });++ rq.on('error', reject);++ rq.end(data);++ });++ server.close();++ assert.deepStrictEqual([res.status, res.body.code], [403, 'media.job.forbidden']);++ console.log('✅ a tenant cannot enqueue storage.move.cleanup; the system app can');++++ // ── Scheduling ──++ assert.strictEqual(await job.schedule(Date.now()), 1, 'queued');++ assert.strictEqual(await job.schedule(Date.now()), 0, 'once per period');++ process.env.MEDIA_MOVE_CLEANUP_MINUTES = '0';++ assert.strictEqual(await job.schedule(Date.now() + 3 * 3600 * 1000), 0, 'MEDIA_MOVE_CLEANUP_MINUTES=0 schedules nothing');++ delete process.env.MEDIA_MOVE_CLEANUP_MINUTES;++ const queued = await db.get("SELECT * FROM media_jobs WHERE job_type = 'storage.move.cleanup'");++ assert.deepStrictEqual([queued.app_id, queued.status, queued.created_by], [queue.SYSTEM_APP, 'queued', 'system:schedule']);++ console.log('✅ scheduled once per period under the system app; 0 disables');++++ // ── Run: one metric increment, one event listing the objects; a second run inside the cooldown adds none ──++ const before = { rows: Number((await db.get('SELECT COUNT(*) AS n FROM media_object_tier_decisions')).n), alerts: await alerts(), outbox: await outbox() };++ const done = await worker.runNow(queued.id);++ assert.strictEqual(done.status, 'succeeded', done.error || '');++ const result = JSON.parse(done.result);++ assert.deepStrictEqual([result.scanned, result.alerted, result.objects.map((o) => o.object_id).sort()], [2, 2, ['obj-after', 'obj-three']]);++ assert.deepStrictEqual([await alerts(), await outbox()], [before.alerts + 1, before.outbox + 1], 'one metric increment and one event');++ const payload = recorded[recorded.length - 1];++ assert.strictEqual(payload.count, 2, JSON.stringify(payload));++ assert.strictEqual(Number((await db.get('SELECT COUNT(*) AS n FROM media_object_tier_decisions')).n), before.rows, 'report only: no row changed');++ const again = JSON.parse((await worker.runNow((await queue.enqueue({ appId: queue.SYSTEM_APP, type: job.TYPE, params: {}, idempotencyKey: 'again' })).job.id)).result);++ assert.deepStrictEqual([again.scanned, again.alerted, await alerts(), await outbox()], [2, 0, before.alerts + 1, before.outbox + 1], 'the 6 h cooldown holds the second alert');++ console.log('✅ run: one media_storage_alerts_total{kind="move_cleanup_failed"} and one storage.alert event listing the objects');++++ // ── Nothing failing: nothing sent ──++ await db.run('DELETE FROM media_object_tier_decisions');++ const quiet = JSON.parse((await worker.runNow((await queue.enqueue({ appId: queue.SYSTEM_APP, type: job.TYPE, params: {}, idempotencyKey: 'quiet' })).job.id)).result);++ assert.deepStrictEqual([quiet.scanned, quiet.alerted, await outbox()], [0, 0, before.outbox + 1]);++ console.log('✅ no object over the threshold: no alert');++++ await db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('move cleanup: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -727,6 +727,31 @@ const crypto = require('crypto');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 second dry-run sweep the same day logs nothing new but still counts the repeated dry run, so the ceiling holds as before.++ s = await tiering.runSweep();++ assert.deepStrictEqual([(await decisions(gE[0].id)).length, (await decisions(gE[1].id)).length, (await decisions(gE[2].id)).length], [1, 1, 0], JSON.stringify(s));++ assert.strictEqual((await last(gE[1].id)).outcome, 'refused', 'a repeated dry run still closes the class: the next candidate is refused again, not proposed');++ const hv2 = s.constraints.hot_usd_per_month.video;++ assert.ok(Math.abs(hv2.projected - (spend1 + one)) < 1e-12 && Math.abs(hv2.spend - spend1) < 1e-12, `the repeat counts in projected: ${JSON.stringify(hv2)}`);++ // A dry-run demotion frees projected cost only when a live one could run: here the canonical file is gone, so it is refused.++ const lostCanon = await addObject({ mime: 'video/mp4' });++ buckets['r2-bucket'].set(`objects/live/${lostCanon.id}`, lostCanon.bytes);++ await model.upsertLocation(lostCanon.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${lostCanon.id}`, storage_class: 'cache', state: 'present', size_bytes: 4096, checksum: lostCanon.sum, verified: true });++ await ageR2(lostCanon.id, 30);++ fs.unlinkSync(lostCanon.file);++ const goodCanon = await addObject({ mime: 'video/mp4' });++ buckets['r2-bucket'].set(`objects/live/${goodCanon.id}`, goodCanon.bytes);++ await model.upsertLocation(goodCanon.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${goodCanon.id}`, storage_class: 'cache', state: 'present', size_bytes: 4096, checksum: goodCanon.sum, verified: true });++ await ageR2(goodCanon.id, 30);++ s = await tiering.runSweep();++ assert.strictEqual((await last(lostCanon.id)).outcome, 'refused');++ assert.ok(/the canonical local file is missing; the R2 copy is kept/.test((await last(lostCanon.id)).reason), (await last(lostCanon.id)).reason);++ assert.strictEqual((await last(goodCanon.id)).outcome, 'dry_run');++ const hv3 = s.constraints.hot_usd_per_month.video, freed = tiering.hotUsdPerMonth(4096);++ assert.ok(Math.abs(hv3.spend - (spend1 + 2 * freed)) < 1e-12, JSON.stringify(hv3));++ // (the two extra R2 copies now fill the ceiling, so the held-over promotion is refused and adds nothing)++ assert.ok(Math.abs(hv3.projected - (hv3.spend - freed)) < 1e-12, `only the demotion a live sweep could make frees projected cost: ${JSON.stringify(hv3)}`);++ await db.run("DELETE FROM media_object_tier_decisions WHERE object_id = ? AND outcome = 'refused'", [gE[0].id]); // the zero-ceiling refusal below is logged afresh// 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');
@@ -134,3 +134,4 @@ MEDIA_CAPABILITY_GATE=report
@@ -134,3 +134,4 @@ MEDIA_CAPABILITY_GATE=report
#MEDIA_JOBS_HEAVY_CONCURRENCY=1
#MEDIA_JOBS_HEAVY_CONCURRENCY=1
#MEDIA_JOBS_HEAVY_WHILE_RECORDING=0
#MEDIA_JOBS_HEAVY_WHILE_RECORDING=0
#MEDIA_INVARIANT_SCAN_HOURS=24 # the size-invariant validator (proposes split/remux; never runs them)
#MEDIA_INVARIANT_SCAN_HOURS=24 # the size-invariant validator (proposes split/remux; never runs them)
+#MEDIA_MOVE_CLEANUP_MINUTES=60 # storage.move.cleanup: alert on objects whose placement move failed 3 times (0 = never)
@@ -123,7 +123,12 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
@@ -123,7 +123,12 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
`backup` (archive fabric only).
`backup` (archive fabric only).
- **Every move is two-phase:** verify the new copy (size + sha256), commit the location in one transaction, then
- **Every move is two-phase:** verify the new copy (size + sha256), commit the location in one transaction, then
delete old bytes after commit, with a retrying cleanup job and an alert after three failures. The row never points
delete old bytes after commit, with a retrying cleanup job and an alert after three failures. The row never points
- at bytes that do not exist.
+ at bytes that do not exist. The sweep retries a failed move after 6 h and logs every failure as a `failed` row in
+ `media_object_tier_decisions`; the job `storage.move.cleanup` (`server/jobs/move-cleanup.js`, light lane, system app
+ only, report only) reads those rows every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never; not in a restore drill): an
+ object whose move failed three times in the last 7 days (a later `done` of that move resets the count) raises one
+ `storage.alert` of kind `move_cleanup_failed`, listing the objects (`media_storage_alerts_total{kind}`, the Events
+ outbox and the per-app webhook, once per 6 h cooldown like every storage alert).
- **Dry-run and simulator:** every policy can run in dry-run mode (logs what it would do and what it would cost), and a
- **Dry-run and simulator:** every policy can run in dry-run mode (logs what it would do and what it would cost), and a
simulator replays recorded telemetry against a candidate policy before it is enabled.
simulator replays recorded telemetry against a candidate policy before it is enabled.
- **Demand rollups (F2.4, shipped):** every served viewer read (`router.route()` for playback or download, not a derive
- **Demand rollups (F2.4, shipped):** every served viewer read (`router.route()` for playback or download, not a derive
@@ -164,7 +169,9 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
@@ -164,7 +169,9 @@ formats (AVIF/WebP), preview clips and AI thumbnails are `rebuildable = true`:
copy_usd_per_month, ceiling }`); the sweep result carries `constraints` (`promote_gate`, `refused.{provider,budget}`,
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
`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
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)
+ against it, so dry runs (gate off) are refused as a live sweep would be, without changing `spend` (a dry run
+ repeated the same day is logged once but still counts, and a dry-run demotion checks its canonical copy like a live
+ one before it frees any projected cost). 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.
set on `game-asset`, `attachment` or `backup` decides nothing yet.
## 7. Delivery: sticky, measured, canaried
## 7. Delivery: sticky, measured, canaried
@@ -228,8 +235,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
@@ -228,8 +235,8 @@ metric. Budgets per class and provider with a forecast; `media.provider.cost_thr
- **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), the provider-class gate
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.
+ and per-class monthly R2 storage ceilings on its moves (F2.6, shipped), the move cleanup job and its alert after
+ three failures (§6, shipped). Still open in F2: 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
@@ -758,6 +758,7 @@ job over only with the token it read and, past boot, only while the lease is sti
@@ -758,6 +758,7 @@ job over only with the token it read and, past boot, only while the lease is sti
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
+| `storage.move.cleanup` | light | Placement moves whose delete keeps failing: an object with three `failed` rows in `media_object_tier_decisions` in the last 7 days (a later `done` resets them) raises one `storage.alert` of kind `move_cleanup_failed`. Service-wide under the system app `_media`; report only. `params {}`. Result `{ scanned, alerted, objects }`. Scheduled every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never); tenants cannot enqueue it (403) |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+/**
+ * OpenVibe.Media — job type storage.move.cleanup (light lane): the loud half of a two-phase move's second phase
+ * (docs/media-fabric.md §6). Once a move's new location is committed the old bytes are deleted; the sweep
+ * (server/objects/tiering.js) retries a failed delete after FAILED_BACKOFF_H, and every failure is a `failed` row
+ * in media_object_tier_decisions. This job reads those rows: an object whose move failed FAIL_THRESHOLD times
+ * within the lookback (a later `done` of the same move starts the count over) raises one storage.alert of kind
+ * move_cleanup_failed, which vod-storage's emitStorageEvent counts in media_storage_alerts_total{kind} and
+ * delivers as the Events outbox row and per-app webhook. Report only: it never moves, deletes or changes a row.
+ *
+ * params {} service-wide; runs only under queue.SYSTEM_APP (no tenant can enqueue it)
+ * result { scanned, alerted, objects } objects = the first 50 (object_id, action, fails, first_at, last_at, last_error)
+ *
+ * Scheduled every MEDIA_MOVE_CLEANUP_MINUTES (60; 0 = never) by the worker, which does not start in a restore drill.
+ */
+'use strict';
+
+const db = require('../db/database');
+const queue = require('./queue');
+
+const { JobError } = queue;
+const TYPE = 'storage.move.cleanup';
+const FAIL_THRESHOLD = 3;
+const LOOKBACK_H = 24 * 7;
+const LISTED = 50;
+
+function envInt(name, fallback) {
+ const v = parseInt(process.env[name] || '', 10);
+ return Number.isFinite(v) ? v : fallback;
+}
+
+/** Moves (object + action) with at least `threshold` failures in the window since their last `done`. */
+async function failedMoves({ threshold = FAIL_THRESHOLD, lookbackH = LOOKBACK_H } = {}) {
+ return (await db.all(`
+ SELECT f.object_id, f.action, MAX(f.app_id) AS app_id, COUNT(*) AS fails,
+ MIN(f.decided_at) AS first_at, MAX(f.decided_at) AS last_at,
+ (ARRAY_AGG(f.error ORDER BY f.decided_at DESC, f.id DESC))[1] AS last_error
+ FROM media_object_tier_decisions f
+ WHERE f.outcome = 'failed' AND f.decided_at >= ov_now_iso(?)
+ AND NOT EXISTS (SELECT 1 FROM media_object_tier_decisions d
+ WHERE d.object_id = f.object_id AND d.action = f.action AND d.outcome = 'done' AND d.decided_at > f.decided_at)
+ GROUP BY f.object_id, f.action
+ HAVING COUNT(*) >= ?
+ ORDER BY MAX(f.decided_at) DESC, f.object_id`, [`-${lookbackH} hours`, threshold])).map((r) => ({ ...r, fails: Number(r.fails) }));
+}
+
+function validate({ appId, obj }) {
+ if (appId !== queue.SYSTEM_APP) {
+ throw new JobError('media.job.forbidden', `${TYPE} covers the whole service's placement: it runs on its schedule`, { status: 403, permanent: true });
+ }
+ if (obj) throw new JobError('media.job.invalid', `${TYPE} covers every object: leave object_id out`, { permanent: true });
+ return {};
+}
+
+async function run() {
+ const rows = await failedMoves();
+ if (!rows.length) return { scanned: 0, alerted: 0, objects: [] };
+ const objects = rows.slice(0, LISTED);
+ // emitStorageEvent counts media_storage_alerts_total{kind="move_cleanup_failed"} itself (one alert per cooldown, so the
+ // payload carries the list and the kind stays a fixed label, never per object).
+ const sent = await require('../vod/vod-storage').emitStorageEvent('storage.alert', 'move_cleanup_failed', { count: rows.length, objects });
+ return { scanned: rows.length, alerted: sent ? rows.length : 0, objects };
+}
+
+/** Queue one run every MEDIA_MOVE_CLEANUP_MINUTES (0 = never). Returns how many were queued (0 or 1). */
+async function schedule(nowMs = Date.now()) {
+ const minutes = envInt('MEDIA_MOVE_CLEANUP_MINUTES', 60);
+ if (!(minutes > 0)) return 0;
+ const period = Math.floor(nowMs / (minutes * 60 * 1000));
+ try {
+ const r = await queue.enqueue({ appId: queue.SYSTEM_APP, type: TYPE, params: {}, idempotencyKey: `${TYPE}:${minutes}m:${period}`, createdBy: 'system:schedule' });
+ return r.created ? 1 : 0;
+ } catch (err) {
+ console.warn(`[Jobs] could not schedule ${TYPE}: ${err.message}`);
+ return 0;
+ }
+}
+
+module.exports = {
+ TYPE, FAIL_THRESHOLD, failedMoves,
+ spec: { lane: 'light', maxAttempts: 2, timeoutMs: 5 * 60 * 1000, needsObject: false, validate, run },
+ schedule,
+};
@@ -12,6 +12,7 @@
@@ -12,6 +12,7 @@
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)
* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)
* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)
* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)
+ * storage.move.cleanup light the alert for placement moves whose delete keeps failing (3+ failed decisions), service-wide
*/
*/
'use strict';
'use strict';
@@ -28,5 +29,6 @@ queue.register('clip.cut', require('../vod/clip-jobs').spec);
@@ -28,5 +29,6 @@ queue.register('clip.cut', require('../vod/clip-jobs').spec);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
queue.register('object.hash', require('./content-hash').spec);
queue.register('object.hash', require('./content-hash').spec);
queue.register('storage.orphans.scan', require('./storage-orphans').spec);
queue.register('storage.orphans.scan', require('./storage-orphans').spec);
+queue.register('storage.move.cleanup', require('./move-cleanup').spec);
module.exports = { names: () => queue.typeNames() };
module.exports = { names: () => queue.typeNames() };
@@ -21,7 +21,8 @@
@@ -21,7 +21,8 @@
* Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every
* Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every
* MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them). Orphaned recordings
* MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them). Orphaned recordings
* get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js). The storage orphan
* get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js). The storage orphan
- * report (storage.orphans.scan, report only) runs every MEDIA_ORPHAN_SCAN_DAYS under queue.SYSTEM_APP.
+ * report (storage.orphans.scan, report only) runs every MEDIA_ORPHAN_SCAN_DAYS under queue.SYSTEM_APP;
+ * storage.move.cleanup (alerts on placement moves that failed three times) every MEDIA_MOVE_CLEANUP_MINUTES.
*/
*/
'use strict';
'use strict';
@@ -158,6 +159,7 @@ async function tick() {
@@ -158,6 +159,7 @@ async function tick() {
try { await require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }
try { await require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }
try { await require('./content-hash').schedule(now); } catch (err) { console.warn('[Jobs] content hash schedule:', err.message); }
try { await require('./content-hash').schedule(now); } catch (err) { console.warn('[Jobs] content hash schedule:', err.message); }
try { await require('./storage-orphans').schedule(now); } catch (err) { console.warn('[Jobs] storage orphan scan schedule:', err.message); }
try { await require('./storage-orphans').schedule(now); } catch (err) { console.warn('[Jobs] storage orphan scan schedule:', err.message); }
+ try { await require('./move-cleanup').schedule(now); } catch (err) { console.warn('[Jobs] move cleanup schedule:', err.message); }
}
}
try { await require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
try { await require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
if (now - lastPruneAt > 6 * 3600 * 1000) {
if (now - lastPruneAt > 6 * 3600 * 1000) {
@@ -337,13 +337,14 @@ async function demote(objectId, ctx = {}) {
@@ -337,13 +337,14 @@ async function demote(objectId, ctx = {}) {
return await log('refused', `${why}; refused: ${gate.reason} (provider class gate); the R2 copy is kept`, { error: gate.reason, blocked: 'provider' });
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 (!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' });
if (!canon || canon.provider === 'r2') return await log('refused', `${why}; refused: no canonical copy besides R2 (the R2 copy is kept)`, { error: 'no canonical copy' });
if (!canon || canon.provider === 'r2') return await log('refused', `${why}; refused: no canonical copy besides R2 (the R2 copy is kept)`, { error: 'no canonical copy' });
- // The last good copy is never the one removed: the canonical copy must check out now.
+ // The last good copy is never the one removed: the canonical copy must check out now. A dry run checks it too,
+ // so it proposes (and projects the freed R2 cost of) only the demotions a live sweep could make.
const check = await confirmCanonical(obj, canon);
const check = await confirmCanonical(obj, canon);
if (!check.ok) return await log('refused', `${why}; refused: ${check.why}; the R2 copy is kept (it may be the last good copy)`, { error: check.why });
if (!check.ok) return await log('refused', `${why}; refused: ${check.why}; the R2 copy is kept (it may be the last good copy)`, { error: check.why });
+ if (!settings.active) return await log('dry_run', `gate off (active = false): would demote: ${why}`);
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {
await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
@@ -632,11 +633,12 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -632,11 +633,12 @@ 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
+ // A slot is spent by a move, a failed move, a dry run (a repeat of one logged earlier today included: it is only
+ // not logged again, and a live sweep would make that move), 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
// (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
// 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.
// 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 spends = (r, action) => (action === 'promote' && r.blocked === 'provider') || ['done', 'failed', 'dry_run'].includes(r.outcome);
const blocked = (r) => { if (r.blocked) out.constraints.refused[r.blocked]++; };
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`
// 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.
// also counts dry runs (the moves the sweep would make), and the ceiling is held against it.
@@ -656,7 +658,7 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -656,7 +658,7 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
blocked(r);
blocked(r);
if (spends(r, 'demote')) spent.demote[cls] = (spent.demote[cls] || 0) + 1;
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.
// 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') {
+ if (['done', 'dry_run'].includes(r.outcome) && row.r2_state === 'present') {
const freed = hotUsdPerMonth(row.r2_size_bytes ?? row.size_bytes);
const freed = hotUsdPerMonth(row.r2_size_bytes ?? row.size_bytes);
projected[cls] = Math.max(0, projected[cls] - freed);
projected[cls] = Math.max(0, projected[cls] - freed);
if (r.outcome === 'done') spend[cls] = Math.max(0, spend[cls] - freed);
if (r.outcome === 'done') spend[cls] = Math.max(0, spend[cls] - freed);
@@ -687,8 +689,8 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
@@ -687,8 +689,8 @@ async function runSweep({ trigger = 'sweep', now = Date.now() } = {}) {
blocked(r);
blocked(r);
if (r.blocked === 'budget') closed.add(cls);
if (r.blocked === 'budget') closed.add(cls);
if (spends(r, 'promote')) spent.promote[cls] = (spent.promote[cls] || 0) + 1;
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;
+ if (['done', 'dry_run'].includes(r.outcome)) projected[cls] += cost;
+ if (r.outcome === 'done') spend[cls] += cost;
}
}
for (const cls of storagePolicy.CLASSES) {
for (const cls of storagePolicy.CLASSES) {
out.constraints.hot_usd_per_month[cls] = { spend: spend[cls], projected: projected[cls], ceiling: storagePolicy.hotCeilingFor(cls) };
out.constraints.hot_usd_per_month[cls] = { spend: spend[cls], projected: projected[cls], ceiling: storagePolicy.hotCeilingFor(cls) };
@@ -1550,6 +1550,7 @@ async function getStatus() {
@@ -1550,6 +1550,7 @@ async function getStatus() {
module.exports = {
module.exports = {
cleanupStaleDownloads,
cleanupStaleDownloads,
+ emitStorageEvent,
DEFAULTS,
DEFAULTS,
REMOTE_PROVIDERS,
REMOTE_PROVIDERS,
OFFLOADABLE_WHERE,
OFFLOADABLE_WHERE,
new file mode 100644
new file mode 100644
@@ -0,0 +1,114 @@
@@ -0,0 +1,114 @@
+'use strict';
+// storage.move.cleanup (server/jobs/move-cleanup.js): the alert half of a two-phase move's cleanup. An object whose
+// move failed three times in the last 7 days (a later `done` resets the count; refused, already and dry_run rows are
+// not failures) raises one storage.alert of kind move_cleanup_failed: media_storage_alerts_total{kind} once, one
+// Events outbox row carrying the list. The job is service-wide (queue.SYSTEM_APP only), scheduled every
+// MEDIA_MOVE_CLEANUP_MINUTES (0 = never), and changes nothing.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-movecleanup-'));
+ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+ Object.assign(process.env, {
+ MEDIA_DATA_DIR: path.join(tmp, 'data'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),
+ });
+ delete process.env.MEDIA_MOVE_CLEANUP_MINUTES;
+ process.env.EVENTS_URL = '';
+ const db = require('../server/db/database');
+ const queue = require('../server/jobs/queue');
+ require('../server/jobs/types');
+ const job = require('../server/jobs/move-cleanup');
+ const worker = require('../server/jobs/worker');
+ const metrics = require('openvibe-shared/metrics');
+ const registry = metrics.createRegistry();
+ require('../server/observability').domainMetrics(registry, { db, recorder: { activeCount: () => 0 }, events: { status: async () => ({ enabled: false }) } });
+ const alerts = async () => {
+ const m = /media_storage_alerts_total\{kind="move_cleanup_failed"\} (\d+)/.exec(await registry.metricsAsync());
+ return m ? Number(m[1]) : 0;
+ };
+ // The Events outbox is off in a test: count what vod-storage hands events.record (the outbox row's source).
+ const recorded = [];
+ const events = require('../server/events');
+ const realRecord = events.record;
+ events.record = async (event, appId, data) => { if (event === 'storage.alert' && data.kind === 'move_cleanup_failed') recorded.push(data); return realRecord(event, appId, data); };
+ const outbox = async () => recorded.length;
+ const row = (objectId, action, outcome, ago = '0 hours', error = null) => db.run(
+ `INSERT INTO media_object_tier_decisions (decided_at, object_id, app_id, action, outcome, trigger, reason, error)
+ VALUES (ov_now_iso(?), ?, 'live', ?, ?, 'sweep', 'test', ?)`, [`-${ago}`, objectId, action, outcome, error]);
+ const many = async (n, id, action, startH, error = null) => { for (let i = 0; i < n; i++) await row(id, action, 'failed', `${startH + i} hours`, error); };
+
+ // ── The threshold, the lookback and what counts as a failure ──
+ await many(3, 'obj-three', 'demote', 1, 'the R2 copy is still there after the delete');
+ await many(2, 'obj-two', 'demote', 1, 'boom');
+ for (const outcome of ['refused', 'dry_run', 'already', 'refused', 'dry_run']) await row('obj-refused', 'demote', outcome);
+ await many(3, 'obj-old', 'demote', 24 * 7 + 5); // all three failures before the lookback
+ await many(2, 'obj-split', 'demote', 1); await row('obj-split', 'promote', 'failed', '1 hours'); // 2 + 1 of another action
+ await many(3, 'obj-reset', 'demote', 10); await row('obj-reset', 'demote', 'done', '5 hours'); // a done after them resets the count
+ await many(3, 'obj-after', 'promote', 1); await row('obj-after', 'promote', 'done', '20 hours'); // the done is older: the failures count
+ const found = await job.failedMoves();
+ assert.deepStrictEqual(found.map((r) => `${r.object_id}:${r.action}:${r.fails}`).sort(), ['obj-after:promote:3', 'obj-three:demote:3'], JSON.stringify(found));
+ assert.strictEqual(found.find((r) => r.object_id === 'obj-three').last_error, 'the R2 copy is still there after the delete');
+ assert.strictEqual(job.FAIL_THRESHOLD, 3);
+ console.log('✅ failedMoves: three failures within the lookback, per object and action; a later done, refusals and dry runs do not count');
+
+ // ── SYSTEM_APP only ──
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key-movecleanup' });
+ assert.throws(() => job.spec.validate({ appId: 'live', params: {} }), (e) => e.status === 403 && e.code === 'media.job.forbidden');
+ assert.deepStrictEqual(job.spec.validate({ appId: queue.SYSTEM_APP, params: {} }), {});
+ const express = require('express');
+ const http = require('http');
+ const app = express();
+ app.use(express.json());
+ app.use('/api/v2/:app/jobs', require('../server/jobs/routes'));
+ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
+ const data = JSON.stringify({ type: job.TYPE });
+ const res = await new Promise((resolve, reject) => {
+ const rq = http.request({ host: '127.0.0.1', port: server.address().port, path: '/api/v2/live/jobs', method: 'POST',
+ headers: { Authorization: 'Bearer live-key-movecleanup', 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } },
+ (r) => { let b = ''; r.on('data', (c) => { b += c; }); r.on('end', () => resolve({ status: r.statusCode, body: JSON.parse(b || '{}') })); });
+ rq.on('error', reject);
+ rq.end(data);
+ });
+ server.close();
+ assert.deepStrictEqual([res.status, res.body.code], [403, 'media.job.forbidden']);
+ console.log('✅ a tenant cannot enqueue storage.move.cleanup; the system app can');
+
+ // ── Scheduling ──
+ assert.strictEqual(await job.schedule(Date.now()), 1, 'queued');
+ assert.strictEqual(await job.schedule(Date.now()), 0, 'once per period');
+ process.env.MEDIA_MOVE_CLEANUP_MINUTES = '0';
+ assert.strictEqual(await job.schedule(Date.now() + 3 * 3600 * 1000), 0, 'MEDIA_MOVE_CLEANUP_MINUTES=0 schedules nothing');
+ delete process.env.MEDIA_MOVE_CLEANUP_MINUTES;
+ const queued = await db.get("SELECT * FROM media_jobs WHERE job_type = 'storage.move.cleanup'");
+ assert.deepStrictEqual([queued.app_id, queued.status, queued.created_by], [queue.SYSTEM_APP, 'queued', 'system:schedule']);
+ console.log('✅ scheduled once per period under the system app; 0 disables');
+
+ // ── Run: one metric increment, one event listing the objects; a second run inside the cooldown adds none ──
+ const before = { rows: Number((await db.get('SELECT COUNT(*) AS n FROM media_object_tier_decisions')).n), alerts: await alerts(), outbox: await outbox() };
+ const done = await worker.runNow(queued.id);
+ assert.strictEqual(done.status, 'succeeded', done.error || '');
+ const result = JSON.parse(done.result);
+ assert.deepStrictEqual([result.scanned, result.alerted, result.objects.map((o) => o.object_id).sort()], [2, 2, ['obj-after', 'obj-three']]);
+ assert.deepStrictEqual([await alerts(), await outbox()], [before.alerts + 1, before.outbox + 1], 'one metric increment and one event');
+ const payload = recorded[recorded.length - 1];
+ assert.strictEqual(payload.count, 2, JSON.stringify(payload));
+ assert.strictEqual(Number((await db.get('SELECT COUNT(*) AS n FROM media_object_tier_decisions')).n), before.rows, 'report only: no row changed');
+ const again = JSON.parse((await worker.runNow((await queue.enqueue({ appId: queue.SYSTEM_APP, type: job.TYPE, params: {}, idempotencyKey: 'again' })).job.id)).result);
+ assert.deepStrictEqual([again.scanned, again.alerted, await alerts(), await outbox()], [2, 0, before.alerts + 1, before.outbox + 1], 'the 6 h cooldown holds the second alert');
+ console.log('✅ run: one media_storage_alerts_total{kind="move_cleanup_failed"} and one storage.alert event listing the objects');
+
+ // ── Nothing failing: nothing sent ──
+ await db.run('DELETE FROM media_object_tier_decisions');
+ const quiet = JSON.parse((await worker.runNow((await queue.enqueue({ appId: queue.SYSTEM_APP, type: job.TYPE, params: {}, idempotencyKey: 'quiet' })).job.id)).result);
+ assert.deepStrictEqual([quiet.scanned, quiet.alerted, await outbox()], [0, 0, before.outbox + 1]);
+ console.log('✅ no object over the threshold: no alert');
+
+ await db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('move cleanup: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -727,6 +727,31 @@ const crypto = require('crypto');
@@ -727,6 +727,31 @@ const crypto = require('crypto');
const hv = s.constraints.hot_usd_per_month.video;
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.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)}`);
assert.ok(Math.abs(hv.projected - (spend1 + one)) < 1e-12, `the projected spend counts the dry run: ${JSON.stringify(hv)}`);
+ // A second dry-run sweep the same day logs nothing new but still counts the repeated dry run, so the ceiling holds as before.
+ s = await tiering.runSweep();
+ assert.deepStrictEqual([(await decisions(gE[0].id)).length, (await decisions(gE[1].id)).length, (await decisions(gE[2].id)).length], [1, 1, 0], JSON.stringify(s));
+ assert.strictEqual((await last(gE[1].id)).outcome, 'refused', 'a repeated dry run still closes the class: the next candidate is refused again, not proposed');
+ const hv2 = s.constraints.hot_usd_per_month.video;
+ assert.ok(Math.abs(hv2.projected - (spend1 + one)) < 1e-12 && Math.abs(hv2.spend - spend1) < 1e-12, `the repeat counts in projected: ${JSON.stringify(hv2)}`);
+ // A dry-run demotion frees projected cost only when a live one could run: here the canonical file is gone, so it is refused.
+ const lostCanon = await addObject({ mime: 'video/mp4' });
+ buckets['r2-bucket'].set(`objects/live/${lostCanon.id}`, lostCanon.bytes);
+ await model.upsertLocation(lostCanon.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${lostCanon.id}`, storage_class: 'cache', state: 'present', size_bytes: 4096, checksum: lostCanon.sum, verified: true });
+ await ageR2(lostCanon.id, 30);
+ fs.unlinkSync(lostCanon.file);
+ const goodCanon = await addObject({ mime: 'video/mp4' });
+ buckets['r2-bucket'].set(`objects/live/${goodCanon.id}`, goodCanon.bytes);
+ await model.upsertLocation(goodCanon.id, { provider: 'r2', bucket: 'r2-bucket', key: `objects/live/${goodCanon.id}`, storage_class: 'cache', state: 'present', size_bytes: 4096, checksum: goodCanon.sum, verified: true });
+ await ageR2(goodCanon.id, 30);
+ s = await tiering.runSweep();
+ assert.strictEqual((await last(lostCanon.id)).outcome, 'refused');
+ assert.ok(/the canonical local file is missing; the R2 copy is kept/.test((await last(lostCanon.id)).reason), (await last(lostCanon.id)).reason);
+ assert.strictEqual((await last(goodCanon.id)).outcome, 'dry_run');
+ const hv3 = s.constraints.hot_usd_per_month.video, freed = tiering.hotUsdPerMonth(4096);
+ assert.ok(Math.abs(hv3.spend - (spend1 + 2 * freed)) < 1e-12, JSON.stringify(hv3));
+ // (the two extra R2 copies now fill the ceiling, so the held-over promotion is refused and adds nothing)
+ assert.ok(Math.abs(hv3.projected - (hv3.spend - freed)) < 1e-12, `only the demotion a live sweep could make frees projected cost: ${JSON.stringify(hv3)}`);
+ await db.run("DELETE FROM media_object_tier_decisions WHERE object_id = ? AND outcome = 'refused'", [gE[0].id]); // the zero-ceiling refusal below is logged afresh
// A zero ceiling closes the class even at a zero R2 price (where every copy would cost $0 and fit any ceiling).
// 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' });
await policy.set({ active: true }, { actor: { type: 'service', id: 'live' }, reason: 'test: gate on' });
const ct = require('../server/placement/cost-tiers');
const ct = require('../server/placement/cost-tiers');