openvibers/openvibe.media Commits
Media reports the recordings and uploads a restart would cut (#320)
Changes
@@@ -539,7 +539,7 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.| path | behavior ||---|---|| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |--| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress` (ffmpeg ingest) and `chunked_recordings_in_progress` (browser MediaRecorder sessions that received a chunk in the last five minutes; additive, so a reader that predates it keeps using the first). |++| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress` (ffmpeg ingest), `chunked_recordings_in_progress` (browser MediaRecorder sessions that received a chunk in the last five minutes), `recordings_untracked_in_progress` (a recording the database still holds — `vods.is_recording = 1` — that no session in memory knows and whose file is still growing: a chunked upload that resumed after a restart) and `uploads_in_progress` (an object upload whose bytes are moving: a single-part PUT, a multipart part, the assembly of a multipart upload). The three later fields are additive, so a reader that predates one keeps using the others; a count that cannot be answered is `null` (unknown), never zero. || `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_timeline_local_bytes`, `media_timeline_local_evictions`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). || `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. || `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |@@@ -741,7 +741,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md). The rules the code keeps:Production deploys with `sudo ovhost deploy media` on the host (strategy `git-checkout`: fetch,fast-forward `/opt/openvibe.media`, install on a lockfile change, restart, wait for `/api/ready`, 60 s).--ovhost refuses the restart while a VOD is recording (`recordings_in_progress` plus `chunked_recordings_in_progress` on `/api/ready`); `--wait-idle`++ovhost refuses the restart while a VOD is recording or an object upload is in flight (`recordings_in_progress`, `chunked_recordings_in_progress`, `recordings_untracked_in_progress` and `uploads_in_progress` on `/api/ready`); `--wait-idle`holds it until none is. The unit is `openvibe-media.service` ([deploy/systemd/](deploy/systemd/openvibe-media.service),runs as `ubuntu`, media under `/opt/openvibe.media/data`) on `127.0.0.1:4100`, the env file`/etc/openvibe/media.env`. The listener belongs to `openvibe-media.socket` (with the@@@ -23,6 +23,7 @@ const { once } = require('events');const { ids } = require('openvibe-contracts');const db = require('../db/database');const config = require('../config');++const uploadSessions = require('./upload-sessions');const MB = 1024 * 1024;const MAX_PARTS = 10000;@@@ -101,12 +102,16 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {const want = partSize(session, n);const tmp = `${partFile(session.id, n)}.${crypto.randomBytes(4).toString('hex')}.tmp`;fs.mkdirSync(sessionDir(session.id), { recursive: true });++ // The bytes are moving: a restart of Media would cut this part (server/protected-sessions.js).++ const key = `${session.id}:${n}`;++ uploadSessions.begin(key, 'part');++ const release = () => uploadSessions.end(key);return new Promise((resolve) => {const hash = crypto.createHash('sha256');const out = fs.createWriteStream(tmp);let got = 0, done = false;const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };-- const finish = (r) => { if (done) return; done = true; resolve(r); };++ const finish = (r) => { if (done) return; done = true; release(); resolve(r); };const fail = async (r) => {if (done) return;const end = async () => { drop(); await finish(r); };@@@ -117,6 +122,7 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {got += c.length;if (got > want) { req.unpipe(out); req.resume(); return await fail({ status: 413, code: 'media.upload.part_too_large', error: `Part ${n} is ${want} bytes` }); }hash.update(c);++ uploadSessions.touch(key);});req.on('aborted', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));req.on('error', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));@@@ -56,6 +56,7 @@ const hls = require('./hls');const ctypes = require('./content-type');const multipart = require('./multipart');const namespaces = require('./namespaces');++const uploadSessions = require('./upload-sessions');const { contentDisposition } = require('./disposition');const { tenantAuth, tenantCors, tenantPath, namespaceGrant } = require('../auth');const { announce } = require('../announce');@@@ -214,7 +215,7 @@ async function reserveAgain(req, res, obj, bytes) {// ── Content upload (mounted ahead of the JSON body parser — see index.js) ──/** Stream the request body to `tmp`, hashing as it goes; stop past `cap` bytes. */--function receive(req, tmp, cap) {++function receive(req, tmp, cap, onProgress) {return new Promise((resolve) => {const hash = crypto.createHash('sha256');const out = fs.createWriteStream(tmp);@@@ -227,6 +228,7 @@ function receive(req, tmp, cap) {n += chunk.length;if (n > cap) { req.unpipe(out); req.resume(); return fail({ error: 'too_large' }); }hash.update(chunk);++ if (onProgress) onProgress(); // the bytes are moving: the upload is a protected session});req.on('aborted', () => fail({ error: 'aborted' }));req.on('error', () => fail({ error: 'aborted' }));@@@ -266,7 +268,14 @@ async function putContent(req, res) {fs.mkdirSync(tmpDir, { recursive: true });const tmp = path.join(tmpDir, `${obj.id}-${crypto.randomBytes(4).toString('hex')}`);await namespaces.touch(obj.id); // the upload is active: its reservation does not expire under it-- const got = await receive(req, tmp, cap);++ // The bytes are moving: a restart of Media would cut this upload (server/protected-sessions.js).++ uploadSessions.begin(obj.id, 'object');++ let got;++ try {++ got = await receive(req, tmp, cap, () => uploadSessions.touch(obj.id));++ } finally {++ uploadSessions.end(obj.id);++ }const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };if (got.error) {drop();@@@ -547,7 +556,14 @@ router.post('/:id/multipart/:uploadId/complete', mpWrite, tenantCors, limits('meif (!await multipart.beginComplete(session)) return problem(res, 409, 'media.upload.not_active', 'The upload is being completed already');const b = req.body || {};const dest = model.objectFilePath(obj);-- const got = await multipart.assemble(session, dest, Array.isArray(b.parts) ? b.parts : null);++ // The assembly is a long copy: a restart of Media would cut it (server/protected-sessions.js).++ uploadSessions.begin(`${session.id}:complete`, 'complete', { idleMs: uploadSessions.ASSEMBLY_IDLE_MS });++ let got;++ try {++ got = await multipart.assemble(session, dest, Array.isArray(b.parts) ? b.parts : null);++ } finally {++ uploadSessions.end(`${session.id}:complete`);++ }if (got.error) {await multipart.reopen(session);return problem(res, got.status, got.code, got.error, got.missing ? { missing: got.missing } : undefined);nnew file mode 100644@@@ -0,0 +1,50 @@++'use strict';++/**++ * OpenVibe.Media — object uploads whose bytes are moving.++ *++ * An upload in flight is cut by a restart of Media: its bytes sit in a temporary file that the++ * restart loses, and the client has to send them again. The parts of a multipart upload are on++ * disk and its session is in the database, so a restart between two parts costs the client++ * nothing (it resumes), but a restart while a part streams, or while the parts are assembled into++ * the object, is lost like any other upload.++ *++ * So this registry counts the upload steps whose bytes are moving: a single-part PUT++ * (objects/routes.js putContent), a multipart part (objects/multipart.js receivePart) and the++ * assembly of a multipart upload (objects/routes.js complete).++ *++ * A step that stops making progress (a stalled client, a dropped connection) is forgotten after++ * UPLOAD_IDLE_MS, like a browser chunk session (vod/chunk-sessions.js): a stalled upload must not++ * pin ovhost's idle wait forever, and a step that never ends (the process killed) leaves the++ * registry empty on the next boot anyway.++ */++const UPLOAD_IDLE_MS = 2 * 60 * 1000;++/** The assembly of a multipart upload copies every part: it takes minutes, and it is still an upload. */++const ASSEMBLY_IDLE_MS = 30 * 60 * 1000;++++/** key → { kind, lastActivityAt, idleMs } */++const sessions = new Map();++++/** An upload step starts. */++function begin(key, kind = 'upload', { idleMs = UPLOAD_IDLE_MS } = {}) {++ sessions.set(String(key), { kind, lastActivityAt: Date.now(), idleMs });++}++++/** A byte arrived: the step is alive. */++function touch(key) {++ const s = sessions.get(String(key));++ if (s) s.lastActivityAt = Date.now();++}++++/** The step ended (done, refused or dropped). */++function end(key) {++ sessions.delete(String(key));++}++++/** Upload steps whose bytes moved within their idle window. */++function countActive(now = Date.now(), idleMs = UPLOAD_IDLE_MS) {++ let n = 0;++ for (const s of sessions.values()) if (now - (s.lastActivityAt || 0) <= (s.idleMs || idleMs)) n++;++ return n;++}++++module.exports = { UPLOAD_IDLE_MS, ASSEMBLY_IDLE_MS, sessions, begin, touch, end, countActive };@@@ -24,9 +24,32 @@ const { createReadiness, skip } = require('openvibe-shared/ready');const copyReport = require('./objects/copy-report');const chunkSessions = require('./vod/chunk-sessions');++const uploadSessions = require('./objects/upload-sessions');++const protectedSessions = require('./protected-sessions');const OUTBOX_BACKLOG_LIMIT = 1000;++/**++ * A count that cannot be answered is unknown (null), never zero: ovhost reads null as unknown and++ * holds the restart. The count is bounded, so a slow database cannot make the whole /api/ready++ * answer late (its own timeout is 2 s) and lose every field.++ */++async function countOrUnknown(fn, timeoutMs = 1500) {++ let timer;++ try {++ const n = Number(await Promise.race([++ Promise.resolve().then(() => fn()),++ new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`no answer in ${timeoutMs} ms`)), timeoutMs); }),++ ]));++ return Number.isFinite(n) ? n : null;++ } catch (err) {++ console.warn('[Ready] protected-session count failed:', err.message);++ return null;++ } finally {++ clearTimeout(timer);++ }++}++function instrument(app, { release }) {return metrics.instrument(app, {service: 'media',@@@ -177,14 +200,20 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, remservice: 'media',release,checks,-- // ovhost's protected-session probe sums both: an ffmpeg recording (RTP/RTMP ingest) and an open-- // browser chunk session are protected sessions alike. The second field is additive, so a Host-- // that predates it keeps reading recordings_in_progress; a Media that predates it omits it, and-- // the Host falls back to the first count.-- details: () => ({++ // ovhost's protected-session probe sums these: an ffmpeg recording (RTP/RTMP ingest), an open++ // browser chunk session, a recording the database still holds (a chunked upload that resumed++ // after a restart) and an object upload whose bytes are moving are protected sessions alike.++ // Every field but the first is additive, so a Host that predates one keeps reading the others;++ // a Media that predates one omits it, and the Host counts what it does report. A count that++ // cannot be answered is null (unknown), never zero: the Host treats unknown as active.++ details: async () => ({...(drill ? { mode: 'drill' } : {}),recordings_in_progress: recorder.activeCount(),chunked_recordings_in_progress: chunkSessions.countActive(),++ // A drill looks at no stored file (that is the drill's own rule: the copy is read, the bytes++ // are not), so it cannot tell which recordings the database still holds: unknown, not zero.++ recordings_untracked_in_progress: drill ? null : await countOrUnknown(() => protectedSessions.untrackedRecordings({ db, recorder, chunkSessions })),++ uploads_in_progress: uploadSessions.countActive(),}),});}nnew file mode 100644@@@ -0,0 +1,74 @@++'use strict';++/**++ * OpenVibe.Media — the protected sessions a restart of Media would cut.++ *++ * ovhost's Media probe (OpenVibe.Host lib/probes.js) asks /api/ready how many protected sessions++ * there are and holds the restart while the answer is not zero. Four kinds count:++ *++ * recordings an ffmpeg recording (RTMP pull, RTP ingest) — the recorder's own map.++ * chunked a browser MediaRecorder session (vod/chunk-sessions.js): a chunk arrived in++ * the last five minutes.++ * untracked a recording the database still holds (vods.is_recording = 1) that no session++ * in memory knows, and whose file is still growing. It is the recording a++ * chunked upload resumed after a restart of Media: the session is rebuilt on++ * the next chunk, but until then the recording is open, and the orphan sweep++ * (jobs/vod-finalize.js) leaves it alone for exactly as long as its file grows.++ * uploads an object upload whose bytes are moving right now++ * (objects/upload-sessions.js): a single-part PUT, a multipart part, the++ * assembly of a multipart upload.++ *++ * The four counts do not overlap: the untracked ones are the recordings no in-memory session++ * holds. A count that cannot be answered is null, never zero: ovhost reads null as unknown and++ * treats it as active, so a database that cannot answer holds the restart instead of cutting it.++ */++const fs = require('fs');++const db = require('./db/database');++const recorder = require('./vod/recorder');++const chunkSessions = require('./vod/chunk-sessions');++const uploadSessions = require('./objects/upload-sessions');++++/** The window a recording file counts as still growing (the orphan sweep's own grace). */++function graceMs() {++ const s = Number(process.env.MEDIA_FINALIZE_ORPHAN_GRACE_S);++ return (Number.isFinite(s) && s > 0 ? s : 180) * 1000;++}++++/** The ids Media is writing right now: an ffmpeg recording or a browser chunk session. */++function knownIds(rec, sessions) {++ const ids = new Set();++ const recs = (rec && rec.activeRecordings && typeof rec.activeRecordings.keys === 'function') ? rec.activeRecordings.keys() : [];++ for (const id of recs) ids.add(Number(id));++ for (const id of (sessions && sessions.sessions ? sessions.sessions.keys() : [])) ids.add(Number(id));++ return ids;++}++++/**++ * Recordings the database still holds that Media is not writing in memory, and whose file is++ * still growing. A file that has not changed for the grace window is an orphan, and the sweep++ * finalizes it: it is not a protected session.++ */++async function untrackedRecordings({ db: dbi = db, recorder: rec = recorder, chunkSessions: sessions = chunkSessions, now = Date.now() } = {}) {++ const known = knownIds(rec, sessions);++ const rows = await dbi.all('SELECT id, file_path FROM vods WHERE is_recording = 1');++ let n = 0;++ for (const row of rows) {++ if (known.has(Number(row.id))) continue;++ let mtime = 0;++ try { mtime = row.file_path ? fs.statSync(row.file_path).mtimeMs : 0; } catch { mtime = 0; }++ if (!mtime || now - mtime >= graceMs()) continue;++ n++;++ }++ return n;++}++++/** Every protected session Media can see: { recordings, chunked, untracked, uploads }. */++async function counts({ db: dbi = db, recorder: rec = recorder, chunkSessions: sessions = chunkSessions, uploadSessions: uploads = uploadSessions, now = Date.now() } = {}) {++ return {++ recordings: rec.activeCount(),++ chunked: sessions.countActive(now),++ untracked: await untrackedRecordings({ db: dbi, recorder: rec, chunkSessions: sessions, now }),++ uploads: uploads.countActive(now),++ };++}++++module.exports = { counts, untrackedRecordings, graceMs, knownIds };@@@ -71,6 +71,16 @@ const { spawn } = require('child_process');body = JSON.parse(r.body);assert.deepStrictEqual([body.recordings_in_progress, body.chunked_recordings_in_progress], [1, 1]);chunkSessions.sessions.clear();++ // The other two protected counts: a recording the database still holds (no session in memory++ // knows it, its file still grows) and an upload whose bytes are moving.++ assert.strictEqual(body.recordings_untracked_in_progress, 0, 'no recording is left open without a session');++ assert.strictEqual(body.uploads_in_progress, 0, 'no upload is in flight: the additive field is present at 0');++ const uploadSessions = require('../server/objects/upload-sessions');++ uploadSessions.begin('obj-ready', 'object');++ r = await request(base, '/api/ready');++ body = JSON.parse(r.body);++ assert.strictEqual(body.uploads_in_progress, 1, 'an upload whose bytes are moving is a protected session');++ uploadSessions.end('obj-ready');const names = Object.keys(body.checks).sort();assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'timeline_cache', 'valkey']);// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optionalnnew file mode 100644@@@ -0,0 +1,82 @@++'use strict';++// The protected sessions a restart of Media would cut (server/protected-sessions.js): a recording++// the database still holds that no session in memory knows, and whose file is still growing (a++// chunked upload that resumed after a restart), and the uploads whose bytes are moving right now++// (server/objects/upload-sessions.js). ovhost's Media probe reads these counts and holds the++// restart while they are not zero.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-protected-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++Object.assign(process.env, {++ 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'),++});++++(async () => {++ const db = require('../server/db/database');++ const protectedSessions = require('../server/protected-sessions');++ const uploadSessions = require('../server/objects/upload-sessions');++ const chunkSessions = require('../server/vod/chunk-sessions');++ await db.upsertApp({ app_id: 'live', api_key: 'live-key-protected' });++++ // ── The upload registry: the bytes are moving ──++ assert.strictEqual(uploadSessions.countActive(), 0, 'nothing is uploading');++ uploadSessions.begin('obj-1', 'object');++ assert.strictEqual(uploadSessions.countActive(), 1, 'a single-part upload in flight counts');++ uploadSessions.begin('mup-1:2', 'part');++ assert.strictEqual(uploadSessions.countActive(), 2, 'a multipart part in flight counts too');++ uploadSessions.end('mup-1:2');++ assert.strictEqual(uploadSessions.countActive(), 1);++ // A step that stopped making progress is forgotten: a stalled client must not pin ovhost's wait.++ const stale = Date.now() - uploadSessions.UPLOAD_IDLE_MS - 1000;++ uploadSessions.sessions.get('obj-1').lastActivityAt = stale;++ assert.strictEqual(uploadSessions.countActive(), 0, 'a step silent past the idle window does not count');++ uploadSessions.touch('obj-1');++ assert.strictEqual(uploadSessions.countActive(), 1, 'a byte arrived: the step is alive again');++ uploadSessions.end('obj-1');++ assert.strictEqual(uploadSessions.countActive(), 0, 'the step ended');++ console.log('✅ an upload counts while its bytes move, and is forgotten when it stalls');++++ // ── A recording the database holds that no session in memory knows ──++ const file = path.join(tmp, 'vods', 'vod-live-1.webm');++ fs.writeFileSync(file, 'growing');++ const vod = await db.createVod({ app_id: 'live', title: 'protected' });++ const id = Number(vod.lastInsertRowid);++ await db.run('UPDATE vods SET is_recording = 1, file_path = ? WHERE id = ?', [file, id]);++++ const emptyRecorder = { activeCount: () => 0, activeRecordings: new Map() };++ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 1,++ 'a recording whose file is still growing is protected, even when no session in memory knows it');++++ // The orphan sweep leaves a recording alone for as long as its file grows (the same grace): a++ // file that has not changed for longer is an orphan, not a protected session.++ const old = new Date(Date.now() - (protectedSessions.graceMs() + 60_000));++ fs.utimesSync(file, old, old);++ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 0,++ 'a file that stopped growing is an orphan, not a protected session');++ fs.utimesSync(file, new Date(), new Date());++++ // A recording a session in memory holds is counted there, not twice.++ const withRecorder = { activeCount: () => 1, activeRecordings: new Map([[id, {}]]) };++ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: withRecorder, chunkSessions }), 0,++ 'an ffmpeg recording is counted by the recorder, not again as untracked');++ chunkSessions.sessions.set(id, { lastChunkAt: Date.now() });++ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 0,++ 'a browser chunk session is counted by its own session, not again as untracked');++ chunkSessions.sessions.clear();++++ // ── Every count together ──++ const counts = await protectedSessions.counts({ db, recorder: emptyRecorder, chunkSessions, uploadSessions });++ assert.deepStrictEqual(counts, { recordings: 0, chunked: 0, untracked: 1, uploads: 0 },++ 'the four counts do not overlap');++ console.log('✅ the protected counts do not overlap, and a growing recording counts once');++++ await db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [id]);++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ protected-session tests passed');++})().catch((err) => { console.error('❌', err); process.exit(1); });nnew file mode 100644@@@ -0,0 +1,67 @@++'use strict';++// An object upload whose bytes are moving is a protected session: a restart of Media would cut it++// (the bytes are in a temporary file), so ovhost's Media probe must hold the restart while it runs++// (server/objects/upload-sessions.js, server/protected-sessions.js).++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const crypto = require('crypto');++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-uploadsess-'));++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++ process.env.VOD_PATH = dir('vods');++ process.env.FILES_PATH = dir('files');++ process.env.OBJECTS_PATH = dir('objects');++ process.env.THUMBNAILS_PATH = dir('thumbnails');++ process.env.OV_NETWORK_URL = 'https://openvibe.network';++ process.env.MEDIA_PUBLIC_URL = 'https://media.test';++ process.env.MEDIA_SIGNING_SECRET = 'test-signing-secret';++++ const db = require('../server/db/database');++ const uploadSessions = require('../server/objects/upload-sessions');++ const objectRoutes = require('../server/objects/routes');++ const express = require('express');++++ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });++++ const app = express();++ app.put('/api/v2/:app/objects/:id/content', ...objectRoutes.contentHandlers);++ app.use(express.json());++ app.use('/api/v2/:app/objects', objectRoutes);++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${server.address().port}`;++ const sha = (b) => crypto.createHash('sha256').update(b).digest('hex');++++ const body = Buffer.from('hello world, this upload is watched');++ const init = await fetch(`${base}/api/v2/live/objects`, {++ method: 'POST',++ headers: { authorization: 'Bearer live-key', 'content-type': 'application/json' },++ body: JSON.stringify({ kind: 'file', mime_type: 'text/plain', size_bytes: body.length, filename: 'watched.txt', visibility: 'private', content_hash: sha(body) }),++ });++ const initText = await init.text();++ assert.strictEqual(init.status, 201, initText);++ const id = JSON.parse(initText).id;++++ // The upload streams in two parts: the first half is on the wire while the second has not moved.++ const put = http.request(`${base}/api/v2/live/objects/${id}/content`, {++ method: 'PUT',++ headers: { authorization: 'Bearer live-key', 'content-type': 'application/octet-stream', 'content-length': String(body.length) },++ });++ const done = new Promise((resolve) => { put.on('response', (res) => { res.resume(); resolve(res.statusCode); }); });++ put.write(body.subarray(0, 10));++ await new Promise((r) => setTimeout(r, 100));++ assert.strictEqual(uploadSessions.countActive(), 1, 'the upload counts while its bytes are moving');++ put.end(body.subarray(10));++ const status = await done;++ assert.strictEqual(status, 200, 'the upload finished');++ assert.strictEqual(uploadSessions.countActive(), 0, 'the upload ended: nothing counts any more');++ console.log('✅ an object upload counts as a protected session while its bytes move');++++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ upload-session tests passed');++})().catch((err) => { console.error('❌', err); process.exit(1); });
@@ -539,7 +539,7 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.
@@ -539,7 +539,7 @@ aggregated and stay queued. The runbook is `docs/cutover-media-billing-readings.
| path | behavior |
| path | behavior |
|---|---|
|---|---|
| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
-| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress` (ffmpeg ingest) and `chunked_recordings_in_progress` (browser MediaRecorder sessions that received a chunk in the last five minutes; additive, so a reader that predates it keeps using the first). |
+| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a real round trip that names the store, postgresql or pglite, and counts the tenants), `valkey` optional (skipped without `VALKEY_URL`), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks`, `remote_b2` / `remote_r2`, `events_outbox`, and `timeline_cache` (bytes vs budget; free disk vs floor). Also `recordings_in_progress` (ffmpeg ingest), `chunked_recordings_in_progress` (browser MediaRecorder sessions that received a chunk in the last five minutes), `recordings_untracked_in_progress` (a recording the database still holds — `vods.is_recording = 1` — that no session in memory knows and whose file is still growing: a chunked upload that resumed after a restart) and `uploads_in_progress` (an object upload whose bytes are moving: a single-part PUT, a multipart part, the assembly of a multipart upload). The three later fields are additive, so a reader that predates one keeps using the others; a count that cannot be answered is `null` (unknown), never zero. |
| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_timeline_local_bytes`, `media_timeline_local_evictions`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |
| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_timeline_local_bytes`, `media_timeline_local_evictions`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |
| `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. |
| `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. |
| `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |
| `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |
@@ -741,7 +741,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md). The rules the code keeps:
@@ -741,7 +741,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md). The rules the code keeps:
Production deploys with `sudo ovhost deploy media` on the host (strategy `git-checkout`: fetch,
Production deploys with `sudo ovhost deploy media` on the host (strategy `git-checkout`: fetch,
fast-forward `/opt/openvibe.media`, install on a lockfile change, restart, wait for `/api/ready`, 60 s).
fast-forward `/opt/openvibe.media`, install on a lockfile change, restart, wait for `/api/ready`, 60 s).
-ovhost refuses the restart while a VOD is recording (`recordings_in_progress` plus `chunked_recordings_in_progress` on `/api/ready`); `--wait-idle`
+ovhost refuses the restart while a VOD is recording or an object upload is in flight (`recordings_in_progress`, `chunked_recordings_in_progress`, `recordings_untracked_in_progress` and `uploads_in_progress` on `/api/ready`); `--wait-idle`
holds it until none is. The unit is `openvibe-media.service` ([deploy/systemd/](deploy/systemd/openvibe-media.service),
holds it until none is. The unit is `openvibe-media.service` ([deploy/systemd/](deploy/systemd/openvibe-media.service),
runs as `ubuntu`, media under `/opt/openvibe.media/data`) on `127.0.0.1:4100`, the env file
runs as `ubuntu`, media under `/opt/openvibe.media/data`) on `127.0.0.1:4100`, the env file
`/etc/openvibe/media.env`. The listener belongs to `openvibe-media.socket` (with the
`/etc/openvibe/media.env`. The listener belongs to `openvibe-media.socket` (with the
@@ -23,6 +23,7 @@ const { once } = require('events');
@@ -23,6 +23,7 @@ const { once } = require('events');
const { ids } = require('openvibe-contracts');
const { ids } = require('openvibe-contracts');
const db = require('../db/database');
const db = require('../db/database');
const config = require('../config');
const config = require('../config');
+const uploadSessions = require('./upload-sessions');
const MB = 1024 * 1024;
const MB = 1024 * 1024;
const MAX_PARTS = 10000;
const MAX_PARTS = 10000;
@@ -101,12 +102,16 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
@@ -101,12 +102,16 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
const want = partSize(session, n);
const want = partSize(session, n);
const tmp = `${partFile(session.id, n)}.${crypto.randomBytes(4).toString('hex')}.tmp`;
const tmp = `${partFile(session.id, n)}.${crypto.randomBytes(4).toString('hex')}.tmp`;
fs.mkdirSync(sessionDir(session.id), { recursive: true });
fs.mkdirSync(sessionDir(session.id), { recursive: true });
+ // The bytes are moving: a restart of Media would cut this part (server/protected-sessions.js).
+ const key = `${session.id}:${n}`;
+ uploadSessions.begin(key, 'part');
+ const release = () => uploadSessions.end(key);
return new Promise((resolve) => {
return new Promise((resolve) => {
const hash = crypto.createHash('sha256');
const hash = crypto.createHash('sha256');
const out = fs.createWriteStream(tmp);
const out = fs.createWriteStream(tmp);
let got = 0, done = false;
let got = 0, done = false;
const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
- const finish = (r) => { if (done) return; done = true; resolve(r); };
+ const finish = (r) => { if (done) return; done = true; release(); resolve(r); };
const fail = async (r) => {
const fail = async (r) => {
if (done) return;
if (done) return;
const end = async () => { drop(); await finish(r); };
const end = async () => { drop(); await finish(r); };
@@ -117,6 +122,7 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
@@ -117,6 +122,7 @@ function receivePart(req, session, n, { expectSha256 = null } = {}) {
got += c.length;
got += c.length;
if (got > want) { req.unpipe(out); req.resume(); return await fail({ status: 413, code: 'media.upload.part_too_large', error: `Part ${n} is ${want} bytes` }); }
if (got > want) { req.unpipe(out); req.resume(); return await fail({ status: 413, code: 'media.upload.part_too_large', error: `Part ${n} is ${want} bytes` }); }
hash.update(c);
hash.update(c);
+ uploadSessions.touch(key);
});
});
req.on('aborted', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));
req.on('aborted', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));
req.on('error', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));
req.on('error', async () => await fail({ status: 400, code: 'media.object.upload_interrupted', error: 'Upload did not complete' }));
@@ -56,6 +56,7 @@ const hls = require('./hls');
@@ -56,6 +56,7 @@ const hls = require('./hls');
const ctypes = require('./content-type');
const ctypes = require('./content-type');
const multipart = require('./multipart');
const multipart = require('./multipart');
const namespaces = require('./namespaces');
const namespaces = require('./namespaces');
+const uploadSessions = require('./upload-sessions');
const { contentDisposition } = require('./disposition');
const { contentDisposition } = require('./disposition');
const { tenantAuth, tenantCors, tenantPath, namespaceGrant } = require('../auth');
const { tenantAuth, tenantCors, tenantPath, namespaceGrant } = require('../auth');
const { announce } = require('../announce');
const { announce } = require('../announce');
@@ -214,7 +215,7 @@ async function reserveAgain(req, res, obj, bytes) {
@@ -214,7 +215,7 @@ async function reserveAgain(req, res, obj, bytes) {
// ── Content upload (mounted ahead of the JSON body parser — see index.js) ──
// ── Content upload (mounted ahead of the JSON body parser — see index.js) ──
/** Stream the request body to `tmp`, hashing as it goes; stop past `cap` bytes. */
/** Stream the request body to `tmp`, hashing as it goes; stop past `cap` bytes. */
-function receive(req, tmp, cap) {
+function receive(req, tmp, cap, onProgress) {
return new Promise((resolve) => {
return new Promise((resolve) => {
const hash = crypto.createHash('sha256');
const hash = crypto.createHash('sha256');
const out = fs.createWriteStream(tmp);
const out = fs.createWriteStream(tmp);
@@ -227,6 +228,7 @@ function receive(req, tmp, cap) {
@@ -227,6 +228,7 @@ function receive(req, tmp, cap) {
n += chunk.length;
n += chunk.length;
if (n > cap) { req.unpipe(out); req.resume(); return fail({ error: 'too_large' }); }
if (n > cap) { req.unpipe(out); req.resume(); return fail({ error: 'too_large' }); }
hash.update(chunk);
hash.update(chunk);
+ if (onProgress) onProgress(); // the bytes are moving: the upload is a protected session
});
});
req.on('aborted', () => fail({ error: 'aborted' }));
req.on('aborted', () => fail({ error: 'aborted' }));
req.on('error', () => fail({ error: 'aborted' }));
req.on('error', () => fail({ error: 'aborted' }));
@@ -266,7 +268,14 @@ async function putContent(req, res) {
@@ -266,7 +268,14 @@ async function putContent(req, res) {
fs.mkdirSync(tmpDir, { recursive: true });
fs.mkdirSync(tmpDir, { recursive: true });
const tmp = path.join(tmpDir, `${obj.id}-${crypto.randomBytes(4).toString('hex')}`);
const tmp = path.join(tmpDir, `${obj.id}-${crypto.randomBytes(4).toString('hex')}`);
await namespaces.touch(obj.id); // the upload is active: its reservation does not expire under it
await namespaces.touch(obj.id); // the upload is active: its reservation does not expire under it
- const got = await receive(req, tmp, cap);
+ // The bytes are moving: a restart of Media would cut this upload (server/protected-sessions.js).
+ uploadSessions.begin(obj.id, 'object');
+ let got;
+ try {
+ got = await receive(req, tmp, cap, () => uploadSessions.touch(obj.id));
+ } finally {
+ uploadSessions.end(obj.id);
+ }
const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
if (got.error) {
if (got.error) {
drop();
drop();
@@ -547,7 +556,14 @@ router.post('/:id/multipart/:uploadId/complete', mpWrite, tenantCors, limits('me
@@ -547,7 +556,14 @@ router.post('/:id/multipart/:uploadId/complete', mpWrite, tenantCors, limits('me
if (!await multipart.beginComplete(session)) return problem(res, 409, 'media.upload.not_active', 'The upload is being completed already');
if (!await multipart.beginComplete(session)) return problem(res, 409, 'media.upload.not_active', 'The upload is being completed already');
const b = req.body || {};
const b = req.body || {};
const dest = model.objectFilePath(obj);
const dest = model.objectFilePath(obj);
- const got = await multipart.assemble(session, dest, Array.isArray(b.parts) ? b.parts : null);
+ // The assembly is a long copy: a restart of Media would cut it (server/protected-sessions.js).
+ uploadSessions.begin(`${session.id}:complete`, 'complete', { idleMs: uploadSessions.ASSEMBLY_IDLE_MS });
+ let got;
+ try {
+ got = await multipart.assemble(session, dest, Array.isArray(b.parts) ? b.parts : null);
+ } finally {
+ uploadSessions.end(`${session.id}:complete`);
+ }
if (got.error) {
if (got.error) {
await multipart.reopen(session);
await multipart.reopen(session);
return problem(res, got.status, got.code, got.error, got.missing ? { missing: got.missing } : undefined);
return problem(res, got.status, got.code, got.error, got.missing ? { missing: got.missing } : undefined);
new file mode 100644
new file mode 100644
@@ -0,0 +1,50 @@
@@ -0,0 +1,50 @@
+'use strict';
+/**
+ * OpenVibe.Media — object uploads whose bytes are moving.
+ *
+ * An upload in flight is cut by a restart of Media: its bytes sit in a temporary file that the
+ * restart loses, and the client has to send them again. The parts of a multipart upload are on
+ * disk and its session is in the database, so a restart between two parts costs the client
+ * nothing (it resumes), but a restart while a part streams, or while the parts are assembled into
+ * the object, is lost like any other upload.
+ *
+ * So this registry counts the upload steps whose bytes are moving: a single-part PUT
+ * (objects/routes.js putContent), a multipart part (objects/multipart.js receivePart) and the
+ * assembly of a multipart upload (objects/routes.js complete).
+ *
+ * A step that stops making progress (a stalled client, a dropped connection) is forgotten after
+ * UPLOAD_IDLE_MS, like a browser chunk session (vod/chunk-sessions.js): a stalled upload must not
+ * pin ovhost's idle wait forever, and a step that never ends (the process killed) leaves the
+ * registry empty on the next boot anyway.
+ */
+const UPLOAD_IDLE_MS = 2 * 60 * 1000;
+/** The assembly of a multipart upload copies every part: it takes minutes, and it is still an upload. */
+const ASSEMBLY_IDLE_MS = 30 * 60 * 1000;
+
+/** key → { kind, lastActivityAt, idleMs } */
+const sessions = new Map();
+
+/** An upload step starts. */
+function begin(key, kind = 'upload', { idleMs = UPLOAD_IDLE_MS } = {}) {
+ sessions.set(String(key), { kind, lastActivityAt: Date.now(), idleMs });
+}
+
+/** A byte arrived: the step is alive. */
+function touch(key) {
+ const s = sessions.get(String(key));
+ if (s) s.lastActivityAt = Date.now();
+}
+
+/** The step ended (done, refused or dropped). */
+function end(key) {
+ sessions.delete(String(key));
+}
+
+/** Upload steps whose bytes moved within their idle window. */
+function countActive(now = Date.now(), idleMs = UPLOAD_IDLE_MS) {
+ let n = 0;
+ for (const s of sessions.values()) if (now - (s.lastActivityAt || 0) <= (s.idleMs || idleMs)) n++;
+ return n;
+}
+
+module.exports = { UPLOAD_IDLE_MS, ASSEMBLY_IDLE_MS, sessions, begin, touch, end, countActive };
@@ -24,9 +24,32 @@ const { createReadiness, skip } = require('openvibe-shared/ready');
@@ -24,9 +24,32 @@ const { createReadiness, skip } = require('openvibe-shared/ready');
const copyReport = require('./objects/copy-report');
const copyReport = require('./objects/copy-report');
const chunkSessions = require('./vod/chunk-sessions');
const chunkSessions = require('./vod/chunk-sessions');
+const uploadSessions = require('./objects/upload-sessions');
+const protectedSessions = require('./protected-sessions');
const OUTBOX_BACKLOG_LIMIT = 1000;
const OUTBOX_BACKLOG_LIMIT = 1000;
+/**
+ * A count that cannot be answered is unknown (null), never zero: ovhost reads null as unknown and
+ * holds the restart. The count is bounded, so a slow database cannot make the whole /api/ready
+ * answer late (its own timeout is 2 s) and lose every field.
+ */
+async function countOrUnknown(fn, timeoutMs = 1500) {
+ let timer;
+ try {
+ const n = Number(await Promise.race([
+ Promise.resolve().then(() => fn()),
+ new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`no answer in ${timeoutMs} ms`)), timeoutMs); }),
+ ]));
+ return Number.isFinite(n) ? n : null;
+ } catch (err) {
+ console.warn('[Ready] protected-session count failed:', err.message);
+ return null;
+ } finally {
+ clearTimeout(timer);
+ }
+}
+
function instrument(app, { release }) {
function instrument(app, { release }) {
return metrics.instrument(app, {
return metrics.instrument(app, {
service: 'media',
service: 'media',
@@ -177,14 +200,20 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
@@ -177,14 +200,20 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
service: 'media',
service: 'media',
release,
release,
checks,
checks,
- // ovhost's protected-session probe sums both: an ffmpeg recording (RTP/RTMP ingest) and an open
- // browser chunk session are protected sessions alike. The second field is additive, so a Host
- // that predates it keeps reading recordings_in_progress; a Media that predates it omits it, and
- // the Host falls back to the first count.
- details: () => ({
+ // ovhost's protected-session probe sums these: an ffmpeg recording (RTP/RTMP ingest), an open
+ // browser chunk session, a recording the database still holds (a chunked upload that resumed
+ // after a restart) and an object upload whose bytes are moving are protected sessions alike.
+ // Every field but the first is additive, so a Host that predates one keeps reading the others;
+ // a Media that predates one omits it, and the Host counts what it does report. A count that
+ // cannot be answered is null (unknown), never zero: the Host treats unknown as active.
+ details: async () => ({
...(drill ? { mode: 'drill' } : {}),
...(drill ? { mode: 'drill' } : {}),
recordings_in_progress: recorder.activeCount(),
recordings_in_progress: recorder.activeCount(),
chunked_recordings_in_progress: chunkSessions.countActive(),
chunked_recordings_in_progress: chunkSessions.countActive(),
+ // A drill looks at no stored file (that is the drill's own rule: the copy is read, the bytes
+ // are not), so it cannot tell which recordings the database still holds: unknown, not zero.
+ recordings_untracked_in_progress: drill ? null : await countOrUnknown(() => protectedSessions.untrackedRecordings({ db, recorder, chunkSessions })),
+ uploads_in_progress: uploadSessions.countActive(),
}),
}),
});
});
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,74 @@
@@ -0,0 +1,74 @@
+'use strict';
+/**
+ * OpenVibe.Media — the protected sessions a restart of Media would cut.
+ *
+ * ovhost's Media probe (OpenVibe.Host lib/probes.js) asks /api/ready how many protected sessions
+ * there are and holds the restart while the answer is not zero. Four kinds count:
+ *
+ * recordings an ffmpeg recording (RTMP pull, RTP ingest) — the recorder's own map.
+ * chunked a browser MediaRecorder session (vod/chunk-sessions.js): a chunk arrived in
+ * the last five minutes.
+ * untracked a recording the database still holds (vods.is_recording = 1) that no session
+ * in memory knows, and whose file is still growing. It is the recording a
+ * chunked upload resumed after a restart of Media: the session is rebuilt on
+ * the next chunk, but until then the recording is open, and the orphan sweep
+ * (jobs/vod-finalize.js) leaves it alone for exactly as long as its file grows.
+ * uploads an object upload whose bytes are moving right now
+ * (objects/upload-sessions.js): a single-part PUT, a multipart part, the
+ * assembly of a multipart upload.
+ *
+ * The four counts do not overlap: the untracked ones are the recordings no in-memory session
+ * holds. A count that cannot be answered is null, never zero: ovhost reads null as unknown and
+ * treats it as active, so a database that cannot answer holds the restart instead of cutting it.
+ */
+const fs = require('fs');
+const db = require('./db/database');
+const recorder = require('./vod/recorder');
+const chunkSessions = require('./vod/chunk-sessions');
+const uploadSessions = require('./objects/upload-sessions');
+
+/** The window a recording file counts as still growing (the orphan sweep's own grace). */
+function graceMs() {
+ const s = Number(process.env.MEDIA_FINALIZE_ORPHAN_GRACE_S);
+ return (Number.isFinite(s) && s > 0 ? s : 180) * 1000;
+}
+
+/** The ids Media is writing right now: an ffmpeg recording or a browser chunk session. */
+function knownIds(rec, sessions) {
+ const ids = new Set();
+ const recs = (rec && rec.activeRecordings && typeof rec.activeRecordings.keys === 'function') ? rec.activeRecordings.keys() : [];
+ for (const id of recs) ids.add(Number(id));
+ for (const id of (sessions && sessions.sessions ? sessions.sessions.keys() : [])) ids.add(Number(id));
+ return ids;
+}
+
+/**
+ * Recordings the database still holds that Media is not writing in memory, and whose file is
+ * still growing. A file that has not changed for the grace window is an orphan, and the sweep
+ * finalizes it: it is not a protected session.
+ */
+async function untrackedRecordings({ db: dbi = db, recorder: rec = recorder, chunkSessions: sessions = chunkSessions, now = Date.now() } = {}) {
+ const known = knownIds(rec, sessions);
+ const rows = await dbi.all('SELECT id, file_path FROM vods WHERE is_recording = 1');
+ let n = 0;
+ for (const row of rows) {
+ if (known.has(Number(row.id))) continue;
+ let mtime = 0;
+ try { mtime = row.file_path ? fs.statSync(row.file_path).mtimeMs : 0; } catch { mtime = 0; }
+ if (!mtime || now - mtime >= graceMs()) continue;
+ n++;
+ }
+ return n;
+}
+
+/** Every protected session Media can see: { recordings, chunked, untracked, uploads }. */
+async function counts({ db: dbi = db, recorder: rec = recorder, chunkSessions: sessions = chunkSessions, uploadSessions: uploads = uploadSessions, now = Date.now() } = {}) {
+ return {
+ recordings: rec.activeCount(),
+ chunked: sessions.countActive(now),
+ untracked: await untrackedRecordings({ db: dbi, recorder: rec, chunkSessions: sessions, now }),
+ uploads: uploads.countActive(now),
+ };
+}
+
+module.exports = { counts, untrackedRecordings, graceMs, knownIds };
@@ -71,6 +71,16 @@ const { spawn } = require('child_process');
@@ -71,6 +71,16 @@ const { spawn } = require('child_process');
body = JSON.parse(r.body);
body = JSON.parse(r.body);
assert.deepStrictEqual([body.recordings_in_progress, body.chunked_recordings_in_progress], [1, 1]);
assert.deepStrictEqual([body.recordings_in_progress, body.chunked_recordings_in_progress], [1, 1]);
chunkSessions.sessions.clear();
chunkSessions.sessions.clear();
+ // The other two protected counts: a recording the database still holds (no session in memory
+ // knows it, its file still grows) and an upload whose bytes are moving.
+ assert.strictEqual(body.recordings_untracked_in_progress, 0, 'no recording is left open without a session');
+ assert.strictEqual(body.uploads_in_progress, 0, 'no upload is in flight: the additive field is present at 0');
+ const uploadSessions = require('../server/objects/upload-sessions');
+ uploadSessions.begin('obj-ready', 'object');
+ r = await request(base, '/api/ready');
+ body = JSON.parse(r.body);
+ assert.strictEqual(body.uploads_in_progress, 1, 'an upload whose bytes are moving is a protected session');
+ uploadSessions.end('obj-ready');
const names = Object.keys(body.checks).sort();
const names = Object.keys(body.checks).sort();
assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'timeline_cache', 'valkey']);
assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'remote_r2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods', 'timeline_cache', 'valkey']);
// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optional
// r2 is not configured: listed as skipped with the reason, never a pretend pass (WS-Q task 7), and an optional
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+'use strict';
+// The protected sessions a restart of Media would cut (server/protected-sessions.js): a recording
+// the database still holds that no session in memory knows, and whose file is still growing (a
+// chunked upload that resumed after a restart), and the uploads whose bytes are moving right now
+// (server/objects/upload-sessions.js). ovhost's Media probe reads these counts and holds the
+// restart while they are not zero.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-protected-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+Object.assign(process.env, {
+ 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'),
+});
+
+(async () => {
+ const db = require('../server/db/database');
+ const protectedSessions = require('../server/protected-sessions');
+ const uploadSessions = require('../server/objects/upload-sessions');
+ const chunkSessions = require('../server/vod/chunk-sessions');
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key-protected' });
+
+ // ── The upload registry: the bytes are moving ──
+ assert.strictEqual(uploadSessions.countActive(), 0, 'nothing is uploading');
+ uploadSessions.begin('obj-1', 'object');
+ assert.strictEqual(uploadSessions.countActive(), 1, 'a single-part upload in flight counts');
+ uploadSessions.begin('mup-1:2', 'part');
+ assert.strictEqual(uploadSessions.countActive(), 2, 'a multipart part in flight counts too');
+ uploadSessions.end('mup-1:2');
+ assert.strictEqual(uploadSessions.countActive(), 1);
+ // A step that stopped making progress is forgotten: a stalled client must not pin ovhost's wait.
+ const stale = Date.now() - uploadSessions.UPLOAD_IDLE_MS - 1000;
+ uploadSessions.sessions.get('obj-1').lastActivityAt = stale;
+ assert.strictEqual(uploadSessions.countActive(), 0, 'a step silent past the idle window does not count');
+ uploadSessions.touch('obj-1');
+ assert.strictEqual(uploadSessions.countActive(), 1, 'a byte arrived: the step is alive again');
+ uploadSessions.end('obj-1');
+ assert.strictEqual(uploadSessions.countActive(), 0, 'the step ended');
+ console.log('✅ an upload counts while its bytes move, and is forgotten when it stalls');
+
+ // ── A recording the database holds that no session in memory knows ──
+ const file = path.join(tmp, 'vods', 'vod-live-1.webm');
+ fs.writeFileSync(file, 'growing');
+ const vod = await db.createVod({ app_id: 'live', title: 'protected' });
+ const id = Number(vod.lastInsertRowid);
+ await db.run('UPDATE vods SET is_recording = 1, file_path = ? WHERE id = ?', [file, id]);
+
+ const emptyRecorder = { activeCount: () => 0, activeRecordings: new Map() };
+ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 1,
+ 'a recording whose file is still growing is protected, even when no session in memory knows it');
+
+ // The orphan sweep leaves a recording alone for as long as its file grows (the same grace): a
+ // file that has not changed for longer is an orphan, not a protected session.
+ const old = new Date(Date.now() - (protectedSessions.graceMs() + 60_000));
+ fs.utimesSync(file, old, old);
+ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 0,
+ 'a file that stopped growing is an orphan, not a protected session');
+ fs.utimesSync(file, new Date(), new Date());
+
+ // A recording a session in memory holds is counted there, not twice.
+ const withRecorder = { activeCount: () => 1, activeRecordings: new Map([[id, {}]]) };
+ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: withRecorder, chunkSessions }), 0,
+ 'an ffmpeg recording is counted by the recorder, not again as untracked');
+ chunkSessions.sessions.set(id, { lastChunkAt: Date.now() });
+ assert.strictEqual(await protectedSessions.untrackedRecordings({ db, recorder: emptyRecorder, chunkSessions }), 0,
+ 'a browser chunk session is counted by its own session, not again as untracked');
+ chunkSessions.sessions.clear();
+
+ // ── Every count together ──
+ const counts = await protectedSessions.counts({ db, recorder: emptyRecorder, chunkSessions, uploadSessions });
+ assert.deepStrictEqual(counts, { recordings: 0, chunked: 0, untracked: 1, uploads: 0 },
+ 'the four counts do not overlap');
+ console.log('✅ the protected counts do not overlap, and a growing recording counts once');
+
+ await db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [id]);
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ protected-session tests passed');
+})().catch((err) => { console.error('❌', err); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,67 @@
@@ -0,0 +1,67 @@
+'use strict';
+// An object upload whose bytes are moving is a protected session: a restart of Media would cut it
+// (the bytes are in a temporary file), so ovhost's Media probe must hold the restart while it runs
+// (server/objects/upload-sessions.js, server/protected-sessions.js).
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const crypto = require('crypto');
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-uploadsess-'));
+ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+ process.env.VOD_PATH = dir('vods');
+ process.env.FILES_PATH = dir('files');
+ process.env.OBJECTS_PATH = dir('objects');
+ process.env.THUMBNAILS_PATH = dir('thumbnails');
+ process.env.OV_NETWORK_URL = 'https://openvibe.network';
+ process.env.MEDIA_PUBLIC_URL = 'https://media.test';
+ process.env.MEDIA_SIGNING_SECRET = 'test-signing-secret';
+
+ const db = require('../server/db/database');
+ const uploadSessions = require('../server/objects/upload-sessions');
+ const objectRoutes = require('../server/objects/routes');
+ const express = require('express');
+
+ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+
+ const app = express();
+ app.put('/api/v2/:app/objects/:id/content', ...objectRoutes.contentHandlers);
+ app.use(express.json());
+ app.use('/api/v2/:app/objects', objectRoutes);
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const sha = (b) => crypto.createHash('sha256').update(b).digest('hex');
+
+ const body = Buffer.from('hello world, this upload is watched');
+ const init = await fetch(`${base}/api/v2/live/objects`, {
+ method: 'POST',
+ headers: { authorization: 'Bearer live-key', 'content-type': 'application/json' },
+ body: JSON.stringify({ kind: 'file', mime_type: 'text/plain', size_bytes: body.length, filename: 'watched.txt', visibility: 'private', content_hash: sha(body) }),
+ });
+ const initText = await init.text();
+ assert.strictEqual(init.status, 201, initText);
+ const id = JSON.parse(initText).id;
+
+ // The upload streams in two parts: the first half is on the wire while the second has not moved.
+ const put = http.request(`${base}/api/v2/live/objects/${id}/content`, {
+ method: 'PUT',
+ headers: { authorization: 'Bearer live-key', 'content-type': 'application/octet-stream', 'content-length': String(body.length) },
+ });
+ const done = new Promise((resolve) => { put.on('response', (res) => { res.resume(); resolve(res.statusCode); }); });
+ put.write(body.subarray(0, 10));
+ await new Promise((r) => setTimeout(r, 100));
+ assert.strictEqual(uploadSessions.countActive(), 1, 'the upload counts while its bytes are moving');
+ put.end(body.subarray(10));
+ const status = await done;
+ assert.strictEqual(status, 200, 'the upload finished');
+ assert.strictEqual(uploadSessions.countActive(), 0, 'the upload ended: nothing counts any more');
+ console.log('✅ an object upload counts as a protected session while its bytes move');
+
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ upload-session tests passed');
+})().catch((err) => { console.error('❌', err); process.exit(1); });