Build a resumable video migration worker: ledger, bounded concurrency, 429 backoff

TL;DR

We're building a migration worker that moves a video library into a hosted video API without
tripping rate limits or double-creating assets. SQLite ledger for resumability, a semaphore with
adaptive concurrency, Retry-After aware backoff, idempotent creates, and a webhook receiver so we
never poll for status.

The naive version of this script is a for loop over a file list. It works until the API starts
returning 429, the process dies at file 3,400, and you have no idea what completed. Cloudflare's
documented global API limit is 1,200 requests per five minutes per user, and blowing through it
blocks all your API calls for the next five minutes, including the ones your production app makes.
Mux's own migration guide says the same thing in gentler words: use a background job queue, control
concurrency, and listen for webhooks instead of polling.

Let's build the thing that does that. Node 22.x, better-sqlite3, no framework.

1. The ledger 📒

Resumability is a database, not a clever loop. One row per asset, one state machine.

mkdir video-migration && cd video-migration
npm init -y
npm pkg set type=module          # everything below is ESM
npm i better-sqlite3 undici

// db.js
import Database from 'better-sqlite3';

export const db = new Database('migration.db');

db.exec(`
  CREATE TABLE IF NOT EXISTS assets (
    id            TEXT PRIMARY KEY,   -- our stable ID, also the idempotency key
    source_url    TEXT NOT NULL,
    state         TEXT NOT NULL DEFAULT 'pending',
      -- pending | inventoried | uploading | processing | ready | failed | rejected
    remote_id     TEXT,
    duration_s    REAL,
    attempts      INTEGER NOT NULL DEFAULT 0,
    retry_after   INTEGER,            -- unix ts; do not re-select before this
    last_error    TEXT,
    updated_at    INTEGER NOT NULL DEFAULT (unixepoch())
  );
  CREATE INDEX IF NOT EXISTS idx_state ON assets(state);
`);

// A real claim: select, then flip to 'uploading' in the same transaction so a second
// worker (or a restart that overlaps a running process) cannot pick up the same rows.
export const claimBatch = db.transaction((limit = 200) => {
  const rows = db.prepare(`
    SELECT * FROM assets
    WHERE state IN ('inventoried','failed')
      AND attempts < 5
      AND (retry_after IS NULL OR retry_after <= unixepoch())
    ORDER BY attempts ASC, rowid ASC
    LIMIT ?
  `).all(limit);

  const claim = db.prepare(
    `UPDATE assets SET state='uploading', updated_at=unixepoch() WHERE id = ?`
  );
  for (const r of rows) claim.run(r.id);
  return rows;
});

// Anything left 'uploading' from a previous run was interrupted, not finished.
export function releaseStale(olderThanSeconds = 900) {
  db.prepare(`
    UPDATE assets SET state='inventoried'
    WHERE state='uploading' AND updated_at < unixepoch() - ?
  `).run(olderThanSeconds);
}

export function setState(id, state, patch = {}) {
  const cols = Object.keys(patch);
  const sets = ['state = ?', 'updated_at = unixepoch()', ...cols.map(c => `${c} = ?`)];
  db.prepare(`UPDATE assets SET ${sets.join(', ')} WHERE id = ?`)
    .run(state, ...cols.map(c => patch[c]), id);
}

Note rejected as a state distinct from failed. A file that FFmpeg cannot parse is never going to
succeed, and it does not belong in a retry queue. That distinction is the difference between a
migration that finishes and one that grinds forever against twelve corrupt files.

2. Inventory before you migrate 🔍

Run a read-only pass with ffprobe first. It is cheap, it touches nothing remote, and it tells you
what kind of migration you are actually about to run.

// inventory.js
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { db, setState } from './db.js';

const run = promisify(execFile);

async function probe(url) {
  const { stdout } = await run('ffprobe', [
    '-v', 'error',
    '-print_format', 'json',
    '-show_format', '-show_streams',
    url
  ], { maxBuffer: 8 * 1024 * 1024 });
  return JSON.parse(stdout);
}

for (const row of db.prepare(`SELECT * FROM assets WHERE state = 'pending'`).all()) {
  try {
    const info = await probe(row.source_url);
    const v = info.streams.find(s => s.codec_type === 'video');
    if (!v) throw new Error('no video stream');
    setState(row.id, 'inventoried', { duration_s: Number(info.format.duration) });
  } catch (err) {
    setState(row.id, 'rejected', { last_error: String(err.message).slice(0, 500) });
  }
}

$ node inventory.js
$ sqlite3 migration.db "SELECT state, count(*) FROM assets GROUP BY state"
inventoried|9612
rejected|41

💡 Tip: those 41 rejects are a list a human reads once. They are not a bug in your worker.

3. A rate limiter that reacts

Fixed concurrency is a guess. What we want is a dial that goes down when the API pushes back and
creeps up when it does not.

// governor.js
export class Governor {
  constructor({ min = 1, max = 16, start = 4 } = {}) {
    this.min = min; this.max = max; this.limit = start;
    this.inFlight = 0; this.queue = [];
    this.pausedUntil = 0;
    this.okStreak = 0;
  }

  async acquire() {
    while (Date.now() < this.pausedUntil || this.inFlight >= this.limit) {
      await new Promise(r => setTimeout(r, 50));
    }
    this.inFlight++;
  }

  release() { this.inFlight--; }

  // Called on every response.
  observe(status, retryAfterHeader) {
    if (status === 429 || status === 503) {
      const secs = parseRetryAfter(retryAfterHeader) ?? 5;
      this.pausedUntil = Date.now() + secs * 1000;
      this.limit = Math.max(this.min, Math.floor(this.limit / 2));
      this.okStreak = 0;
      console.warn(`[governor] backpressure: pausing ${secs}s, concurrency -> ${this.limit}`);
    } else if (status < 400) {
      if (++this.okStreak >= 50 && this.limit < this.max) {
        this.limit++; this.okStreak = 0;
        console.info(`[governor] concurrency -> ${this.limit}`);
      }
    }
  }
}

function parseRetryAfter(h) {
  if (!h) return null;
  const n = Number(h);
  if (Number.isFinite(n)) return n;               // delta-seconds
  const t = Date.parse(h);                        // HTTP-date
  return Number.isFinite(t) ? Math.max(0, (t - Date.now()) / 1000) : null;
}

Halve on pushback, add one on sustained success. It is TCP congestion control with the serial numbers
filed off, and it works for the same reason.

4. Idempotent creates

Retrying an upload must not create a second asset. Most video APIs give you at least one of: an
idempotency key header, a client reference ID, or free-form passthrough metadata you can query later.
Use your own stable asset ID as the key so it survives a restart.

// migrate.js
import { request } from 'undici';
import { db, claimBatch, setState, releaseStale } from './db.js';
import { Governor } from './governor.js';

releaseStale();   // recover rows a previous crashed run left mid-flight

const API = process.env.VIDEO_API_BASE;      // e.g. https://api.example.com/v1
const TOKEN = process.env.VIDEO_API_TOKEN;
const gov = new Governor({ start: 4, max: 16 });

async function createAsset(row) {
  const res = await request(`${API}/videos`, {
    method: 'POST',
    headers: {
      'authorization': `Bearer ${TOKEN}`,
      'content-type': 'application/json',
      'idempotency-key': row.id            // <- the whole trick
    },
    body: JSON.stringify({
      input: row.source_url,
      metadata: { migration_ref: row.id }  // fallback lookup key
    })
  });
  const body = await res.body.json().catch(() => ({}));
  gov.observe(res.statusCode, res.headers['retry-after']);
  return { status: res.statusCode, body };
}

async function findExisting(row) {
  const res = await request(`${API}/videos?metadata[migration_ref]=${row.id}`, {
    headers: { authorization: `Bearer ${TOKEN}` }
  });
  const body = await res.body.json().catch(() => ({}));
  gov.observe(res.statusCode, res.headers['retry-after']);
  return body?.data?.[0]?.id ?? null;
}

async function migrateOne(row) {
  await gov.acquire();
  try {
    let { status, body } = await createAsset(row);

    if (status === 429 || status >= 500) {
      // back off this row specifically, so the next claimBatch does not grab it immediately
      const delay = Math.min(300, 2 ** row.attempts) + Math.random() * 5;   // jitter
      setState(row.id, 'failed', {
        attempts: row.attempts + 1,
        retry_after: Math.floor(Date.now() / 1000 + delay),
        last_error: `transient ${status}`
      });
      return;
    }
    if (status === 409) {                       // already created, provider told us
      const existing = await findExisting(row);
      if (existing) return setState(row.id, 'processing', { remote_id: existing });
    }
    if (status >= 400) {
      setState(row.id, 'rejected', {
        attempts: row.attempts + 1,
        last_error: JSON.stringify(body).slice(0, 500)
      });
      return;
    }
    setState(row.id, 'processing', { remote_id: body.id ?? body.data?.id });
  } finally {
    gov.release();
  }
}

for (;;) {
  const batch = claimBatch(200);
  if (batch.length === 0) break;
  await Promise.all(batch.map(migrateOne));
}
console.log('upload pass complete');

Realistic output on a library that hits the ceiling:

$ node migrate.js
[governor] concurrency -> 5
[governor] concurrency -> 6
[governor] backpressure: pausing 12s, concurrency -> 3
[governor] concurrency -> 4
upload pass complete

Kill it with Ctrl-C at any point and rerun. It picks up from the ledger.

5. Webhooks, not polling 🔔

processing is not ready. Polling ten thousand assets for status is how you spend your entire rate
limit budget asking questions instead of doing work. Take the event instead.

// webhook.js
import http from 'node:http';
import crypto from 'node:crypto';
import { setState } from './db.js';

const SECRET = process.env.WEBHOOK_SECRET;

function verify(rawBuffer, sig) {
  const expected = crypto.createHmac('sha256', SECRET).update(rawBuffer).digest('hex');
  // timingSafeEqual throws on length mismatch, so guard first
  const a = Buffer.from(expected), b = Buffer.from(sig ?? '', 'utf8');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

http.createServer((req, res) => {
  const chunks = [];
  req.on('data', c => chunks.push(c));
  req.on('end', () => {
    const raw = Buffer.concat(chunks);   // HMAC the bytes, never a decoded string
    if (!verify(raw, req.headers['x-signature'])) {
      res.writeHead(401).end();
      return;
    }
    const evt = JSON.parse(raw.toString('utf8'));
    const ref = evt.data?.metadata?.migration_ref;
    if (ref && evt.type === 'video.ready') setState(ref, 'ready');
    if (ref && evt.type === 'video.errored') {
      setState(ref, 'rejected', { last_error: evt.data.error ?? 'ingest failed' });
    }
    res.writeHead(204).end();   // ack fast, work later
  });
}).listen(8787);

⚠️ Note: ack the webhook before you do any slow work. Providers retry on timeout, and a slow
handler turns one event into five.

6. Reconcile, then cut over ✅

The migration is not done when the uploads finish. It is done when the destination matches the
inventory.

Pull the destination's view of each asset back into a second table, then diff it. A reconcile.js
that walks state='ready' rows, fetches GET /videos/{remote_id}, and writes
(migration_ref, remote_duration_s) into a remote_assets table is about thirty lines using the same
Governor. Then:

-- create it once, populate it from the reconcile pass
CREATE TABLE IF NOT EXISTS remote_assets (
  migration_ref     TEXT PRIMARY KEY,
  remote_duration_s REAL
);

-- durations that drifted by more than half a second are the canary
SELECT a.id, a.duration_s, r.remote_duration_s
FROM assets a JOIN remote_assets r ON r.migration_ref = a.id
WHERE abs(a.duration_s - r.remote_duration_s) > 0.5;

A duration mismatch almost always means the ingest made a decision you did not ask for: a variable
frame rate source got normalized, a broken trailing segment got dropped, an audio-only tail got cut.
Worth looking at every single one.

Then flip playback behind a per-asset or per-tenant flag so a bad mapping is a revert, not an
incident. Keep the source library readable for a few months. Delete it when the reconciliation report
has been boring for a while.

What's next

• Add a dry_run mode that migrates a deliberately weird slice (oldest, largest, unusual track layouts) instead of a random sample. Fifty strange files teach you more than a thousand normal ones.
• If your provider supports resumable uploads over tus, swap the create call for a tus client so a dropped connection mid-file does not restart the transfer.
• Export the ledger states to your metrics backend and alert on failed growth rate. A migration that silently stops making progress at 2am is the failure mode this whole design exists to prevent.

The pieces here (a queue, a ledger, an adaptive limiter, a webhook receiver) are ones you have all
written before in other contexts. The only real mistake is looking at a directory listing and a POST
endpoint and concluding the shortest path between them is a loop.