Media: sweep the work directories of finished or vanished jobs (12.8 GB were left on 10-15) (#131)

ddiff --git a/server/jobs/derive.js b/server/jobs/derive.js
iindex 4edf94f..4455c6c 100644
---- a/server/jobs/derive.js
++++ b/server/jobs/derive.js
@@@ -207,6 +207,41 @@ function cleanupWork(jobId) {
      try { fs.rmSync(path.join(config.objects.path, '.jobs', jobId), { recursive: true, force: true }); } catch { /* best effort */ }
  }
  
++/**
++ * Remove the work directories (OBJECTS_PATH/.jobs/<job id>) that no job will use again: their job finished (succeeded,
++ * failed, cancelled) or no longer exists, and nothing has written to them for `minAgeMs`. A handler cleans up in its
++ * own `finally`, but a process killed mid-job (a restart, a deploy, the out-of-memory killer) never gets there: on
++ * 2026-10-15 four such directories held 12.8 GB of a 96 GB disk. A queued or running job's directory is kept (a retry
++ * may resume in it). → { removed: [ids], bytes }
++ */
++async function sweepWork({ minAgeMs = 3600 * 1000, now = Date.now() } = {}) {
++    const root = path.join(config.objects.path, '.jobs');
++    let names = [];
++    try { names = fs.readdirSync(root); } catch { return { removed: [], bytes: 0 }; }
++    const sizeOf = (p) => {
++        let total = 0;
++        const walk = (d) => { for (const e of fs.readdirSync(d, { withFileTypes: true })) { const f = path.join(d, e.name); if (e.isDirectory()) walk(f); else { try { total += fs.statSync(f).size; } catch { /* gone */ } } } };
++        try { walk(p); } catch { /* gone */ }
++        return total;
++    };
++    const removed = [];
++    let bytes = 0;
++    for (const name of names) {
++        if (!/^mjob_[0-9A-HJKMNP-TV-Z]{26}$/.test(name)) continue;
++        const dir = path.join(root, name);
++        let st;
++        try { st = fs.statSync(dir); } catch { continue; }
++        if (!st.isDirectory() || now - st.mtimeMs < minAgeMs) continue;
++        const job = await db.get('SELECT status FROM media_jobs WHERE id = ?', [name]);
++        if (job && !['succeeded', 'failed', 'cancelled'].includes(job.status)) continue;
++        const size = sizeOf(dir);
++        try { fs.rmSync(dir, { recursive: true, force: true }); } catch { continue; }
++        removed.push(name);
++        bytes += size;
++    }
++    return { removed, bytes };
++}
++
  /**
   * Register one finished output file as a private native object derived from `src`: create the row,
   * move the bytes under OBJECTS_PATH/<app>/<id>, verified local location, relationship; `alsoInTx`
@@@ -353,7 +388,7 @@ module.exports = {
      remux: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateRemux, run: runRemux },
      resolveSource, isMediaObject, extFor,
      // for the preview jobs (previews.js)
--    loadSource, adopt, ffmpeg, inputArgs, probeDuration, workDir, cleanupWork, existing, checkRoom,
++    loadSource, adopt, ffmpeg, inputArgs, probeDuration, workDir, cleanupWork, sweepWork, existing, checkRoom,
      // and object.cmaf (cmaf.js)
      budgetMs, probeStreams, sha256File,
  };
ddiff --git a/server/jobs/worker.js b/server/jobs/worker.js
iindex 9d33107..8e9d9ae 100644
---- a/server/jobs/worker.js
++++ b/server/jobs/worker.js
@@@ -36,6 +36,15 @@ let timer = null;
  let started = false;
  let ticking = false;
  let lastScheduleAt = 0;
++let lastSweepAt = 0;
++
++/** The work directories of finished or vanished jobs (derive.sweepWork), at start and hourly. */
++async function sweepWorkDirs() {
++    try {
++        const r = await require('./derive').sweepWork();
++        if (r.removed.length) console.log(`[Jobs] removed ${r.removed.length} leftover work director${r.removed.length === 1 ? 'y' : 'ies'} (${(r.bytes / 1e9).toFixed(2)} GB) of finished or vanished jobs`);
++    } catch (err) { console.warn('[Jobs] work directory sweep:', err.message); }
++}
  let lastPruneAt = 0;
  let kicked = false;
  
@@@ -162,6 +171,10 @@ async function tick() {
                  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 - lastSweepAt > 3600 * 1000) {
++                lastSweepAt = now;
++                await sweepWorkDirs();
++            }
              if (now - lastPruneAt > 6 * 3600 * 1000) {
                  lastPruneAt = now;
                  try { const n = await queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail/hash job(s)`); } catch { /* next time */ }
@@@ -202,6 +215,9 @@ async function start() {
      // or a job claimed before start()): failing it here would refuse its own result as stale (jobs.test.js flaked on it).
      const n = await queue.recoverInterrupted({ all: true, except: new Set(running.keys()) });
      if (n) console.log(`[Jobs] ${n} job(s) interrupted by the restart were requeued or failed`);
++    // A job the restart failed for good left its work directory behind: sweep it now (and hourly from tick()).
++    lastSweepAt = Date.now();
++    await sweepWorkDirs();
      lastScheduleAt = Date.now();           // the first scheduled scan waits a few minutes after boot
      timer = setInterval(() => { tick().catch(() => {}); }, cfg().pollMs);
      if (timer.unref) timer.unref();
ddiff --git a/test/jobs-work-sweep.test.js b/test/jobs-work-sweep.test.js
nnew file mode 100644
iindex 0000000..44be3e9
---- /dev/null
++++ b/test/jobs-work-sweep.test.js
@@@ -0,0 +1,56 @@
++'use strict';
++// Leftover job work directories (server/jobs/derive.js sweepWork, run at worker start and hourly): a process killed
++// mid-job never reaches the handler's own cleanup, and on 2026-10-15 four such directories held 12.8 GB. The sweep
++// removes OBJECTS_PATH/.jobs/<id> when the job finished (succeeded, failed, cancelled) or no longer exists and nothing
++// wrote to it for an hour; a queued or running job's directory, a fresh one and anything not named like a job stay.
++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-sweep-'));
++    const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++    for (const [k, n] of [['VOD_PATH', 'vods'], ['CLIPS_PATH', 'clips'], ['FILES_PATH', 'files'], ['THUMBNAILS_PATH', 'thumbnails'], ['PASTES_PATH', 'pastes'], ['OBJECTS_PATH', 'objects']]) process.env[k] = dir(n);
++    try {
++        const db = require('../server/db/database');
++        await db.upsertApp({ app_id: 'live', api_key: 'live-key' });
++        const derive = require('../server/jobs/derive');
++        const jobs = path.join(process.env.OBJECTS_PATH, '.jobs');
++        const id = (n) => `mjob_01M4RDBWXB2ANQJP3W6N1TF7${String(n).padStart(2, '0').replace(/[ILOU]/g, 'A')}`;
++        const make = (name, { ageMs = 2 * 3600 * 1000, bytes = 1024 } = {}) => {
++            const d = path.join(jobs, name);
++            fs.mkdirSync(path.join(d, 'out'), { recursive: true });
++            fs.writeFileSync(path.join(d, 'out', '000001.m4s'), Buffer.alloc(bytes));
++            const t = (Date.now() - ageMs) / 1000;
++            fs.utimesSync(d, t, t);
++            return d;
++        };
++        const job = async (n, status) => db.run(`INSERT INTO media_jobs (id, app_id, job_type, status, params, created_at, updated_at) VALUES (?, 'live', 'object.cmaf', ?, '{}', ov_now(), ov_now())`, [id(n), status]);
++        await job(1, 'failed');
++        await job(2, 'succeeded');
++        await job(3, 'cancelled');
++        await job(4, 'queued');
++        await job(5, 'running');
++        await job(6, 'failed');
++        for (const n of [1, 2, 3, 4, 5]) make(id(n), { bytes: 4096 });
++        make(id(6), { ageMs: 60 * 1000 });          // failed a minute ago: not yet
++        make(id(7));                                  // no job row at all: vanished
++        make('not-a-job');                            // not named like a job: never touched
++
++        const r = await derive.sweepWork();
++        assert.deepStrictEqual(r.removed.sort(), [id(1), id(2), id(3), id(7)].sort());
++        assert.strictEqual(r.bytes, 4096 * 3 + 1024);
++        const left = fs.readdirSync(jobs).sort();
++        assert.deepStrictEqual(left, [id(4), id(5), id(6), 'not-a-job'].sort(), 'queued, running, fresh and foreign stay');
++        assert.deepStrictEqual((await derive.sweepWork()).removed, [], 'a second sweep finds nothing');
++
++        const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'jobs', 'worker.js'), 'utf8');
++        assert.match(src, /recoverInterrupted\(\{ all: true[\s\S]{0,400}await sweepWorkDirs\(\);/, 'the worker sweeps at start');
++        assert.match(src, /lastSweepAt > 3600 \* 1000\) \{\s*lastSweepAt = now;\s*await sweepWorkDirs\(\);/, 'and hourly');
++        console.log('jobs-work-sweep: ok');
++        await db.close();
++    } finally {
++        fs.rmSync(tmp, { recursive: true, force: true });
++    }
++})().catch((e) => { console.error(e); process.exit(1); });