Distributed background jobs

Some work is too heavy or too slow to do inline: transcoding a video, rendering a PDF, running a batch export, calling a model over a big input. The reusable answer on Muhkoo is a worker pool over a Space — a durable job queue where the work survives a reload and any device the user is signed in on can pick up the next item. No servers to run: the queue is a client.db table, coordination rides a client.space, and the payload lives as offline-first shards so a job is portable between machines.

This page is the architecture. It composes surfaces you already have — nothing here is a dedicated “jobs” API; it’s a pattern you assemble. It’s the exact design behind Muhkoo Theater, where your browser and an optional native worker drain the same transcode queue together.

Note

The whole system is owner-scoped — every worker is a device signed in as the same user, so contention is low and trust is implicit. This is a pattern for spreading one user’s work across their devices, not a public compute marketplace.

The shape

Four pieces, each a surface you already know:

• A jobs table (client.db) — the durable source of truth. Survives reloads; visible to every device.
• A claim + heartbeat lease — a worker marks a job as its own, then proves it’s alive; a job whose worker goes silent is reclaimed. This is what makes the queue crash-resilient.
• A Space for real-time coordination (client.space) — the DB poll is the durable floor; ephemeral Space signals make it instant (wake idle workers, mirror progress across devices, remote-cancel).
• Offline-first payloads (client.storage) — store the heavy input once as shards; the job row carries only the manifest (a read capability), so any worker can fetch the bytes (peer-to-peer within a network, origin as fallback) and do the work.

1. The jobs table

The row is the durable state machine. Beyond your own fields, the columns that make it a queue are the status, the lease (worker_id / heartbeat_at), and the payload pointer (input_manifest):

// in your app spec's tables[]
{
  "table": "jobs",
  "columns": [
    { "name": "type",           "type": "text",      "nullable": false }, // what kind of work
    { "name": "title",          "type": "text",      "nullable": false },
    { "name": "status",         "type": "text",      "nullable": false }, // queued|processing|done|error|paused
    { "name": "progress",       "type": "real",      "nullable": true  }, // 0..1
    { "name": "phase",          "type": "text",      "nullable": true  }, // human label e.g. "Encoding 720p…"
    { "name": "input_manifest", "type": "json",      "nullable": true  }, // read capability for the payload
    { "name": "input_size",     "type": "integer",   "nullable": true  },
    { "name": "worker_id",      "type": "text",      "nullable": true  }, // the lease holder
    { "name": "worker_label",   "type": "text",      "nullable": true  }, // for UI attribution
    { "name": "claimed_at",     "type": "timestamp", "nullable": true  },
    { "name": "heartbeat_at",   "type": "timestamp", "nullable": true  }, // liveness — drives reclaim
    { "name": "result_id",      "type": "integer",   "nullable": true  }, // link to the produced record
    { "name": "error",          "type": "text",      "nullable": true  },
    { "name": "owner",          "type": "text",      "nullable": false }, // the user's commitment
    { "name": "created_at",     "type": "timestamp", "nullable": false },
    { "name": "updated_at",     "type": "timestamp", "nullable": false }
  ],
  "indexes": [
    { "name": "idx_jobs_owner",  "columns": ["owner"] },
    { "name": "idx_jobs_status", "columns": ["status"] }
  ]
}

Everything reads and writes it filtered by owner — the owner-scoping rule — so a device only ever sees its own user’s jobs. Enqueue is just an insert:

const jobs = () => client.db.table("jobs");

await jobs().insert({
  type: "render", title, status: "queued", progress: 0, phase: "Queued",
  input_manifest: null, input_size: file.size,
  owner: myCommitment, created_at: Date.now(), updated_at: Date.now(),
});

2. Store the payload once, carry a manifest

Don’t put bytes in the row, and don’t make workers re-upload. Store the heavy input once as encrypted shards and let the job carry the resulting manifest — an opaque read capability. Any worker that has the manifest can reconstruct the bytes; within a network the shards transfer peer-to-peer, with origin as the fallback.

// after enqueue, in the background (so encoding can start on the in-memory copy):
const { manifest } = await space.putFile(file, { name: file.name, type: file.type });
await jobs().update(id, { input_manifest: manifest, updated_at: Date.now() });

Tip

Set input_manifest after the store completes, and retry it a few times. A job with a null manifest can only be worked by the device still holding the original in memory — persisting the manifest is what makes it claimable by any worker and resumable after a reload.

A worker fetches the payload by manifest. The json column can hand the manifest back as a string — parse it first:

const manifest = typeof job.input_manifest === "string"
  ? JSON.parse(job.input_manifest) : job.input_manifest;
const { data } = await client.storage.readByManifest(manifest);

3. Claim with a heartbeat lease

A worker claims the next workable job by writing its id onto the row, then re-reads to confirm it won — a cheap optimistic lock that’s plenty for a single user’s low-contention devices (a rare double-process is wasteful, not wrong). A job is workable if it’s queued, or processing but its worker has gone silent past a staleness window — that second case is the crash recovery.

const STALE_MS = 45_000;

async function claimNext(workerId, workerLabel) {
  const { rows } = await jobs().query({
    where: [{ column: "owner", op: "eq", value: myCommitment }],
    orderBy: { column: "created_at", dir: "asc" }, limit: 100,
  });
  const now = Date.now();
  const cand = rows.find((j) =>
    j.input_manifest && (                         // only claim what we can fetch
      j.status === "queued" ||
      (j.status === "processing" && (!j.heartbeat_at || now - j.heartbeat_at > STALE_MS))
    ));
  if (!cand) return null;

  await jobs().update(cand._id, {
    status: "processing", worker_id: workerId, worker_label: workerLabel,
    claimed_at: now, heartbeat_at: now, updated_at: now,
  });
  const fresh = await jobs().get(cand._id);          // re-read…
  const row = fresh?.row ?? fresh;
  if (row?.worker_id !== workerId) return claimNext(workerId, workerLabel); // …lost the race
  return row;
}

While processing, beat the heartbeat (throttled — every few seconds, not every progress tick) so other workers know the job is alive. If this worker dies, the heartbeat stops and the job falls back to reclaimable after STALE_MS:

async function process(job) {
  const bytes = await fetchPayload(job);
  const result = await doTheWork(bytes, (fraction, label) => {
    // throttle DB writes; the UI updates from a faster in-memory cache
    void jobs().update(job._id, { progress: fraction, phase: label, heartbeat_at: Date.now() });
  });
  await jobs().update(job._id, { status: "done", progress: 1, phase: "Done", result_id: result._id });
}

Only claim what you can actually do

Filter candidates on input_manifest (or a local in-memory copy). If a worker can’t fetch a job’s payload, it must leave it queued for a worker that can — never claim-and-fail. If you claim and then discover the payload isn’t reachable, release the lease (status: "queued", worker_id: null) instead of erroring.

4. Make it instant with a Space

Polling the table every few seconds is the durable floor, but it’s laggy. A per-user Space makes coordination real-time by carrying ephemeral signals alongside the durable table. Reserve one message shape and switch on kind:

type JobSignal = {
  _t: "jobsig"; src: string;                 // sender's worker id
  kind: "new" | "progress" | "done" | "control" | "hello";
  jobId?: number; progress?: number; phase?: string;
  worker?: string; score?: number; native?: boolean;
};

• new — enqueued a job → idle workers wake and try to claim.
• progress — the encoding worker streams its fraction so other devices mirror live progress for a job they aren’t running.
• done — refresh + look for the next job.
• control — “pause/cancel job X”; whichever device is running it aborts.
• hello — presence + capability (see grading below).

const space = await client.space.joinChannel(mediaSpaceName);
space.onMessage((e) => {
  const b = e.message?.body;
  if (b?._t !== "jobsig" || b.src === myWorkerId) return; // ignore our own
  if (b.kind === "new" || b.kind === "done") kickWorker();
  else if (b.kind === "control") maybeAbort(b.jobId);
  // …mirror progress, track presence…
});
const post = (sig) => space.sendMessage({ _t: "jobsig", src: myWorkerId, ...sig });

Tip

The Space is coordination only — the jobs table remains the source of truth. If a signal is missed (a device was offline), the next DB poll still converges. Build the durable path first, then layer the Space on top for latency.

5. Many workers, graded by power

Every device signed in as the user is a worker, and they’re not equal — a laptop’s browser is slower than a many-core box running a native worker. Each worker announces a rough capability score in its hello, and a weaker worker defers a claimable job to a stronger one for a grace window, stepping in only if the job looks stuck. That keeps the fast machine busy without the slow one racing it — while the slow one still contributes by storing/serving payloads.

const MY_SCORE = navigator.hardwareConcurrency || 4;   // native worker: cores × ~30
const peers = new Map();                               // src -> { score, ts }
const strongerOnline = () => [...peers.values()]
  .some((w) => Date.now() - w.ts < 30_000 && w.score > MY_SCORE);

// in claimNext, before taking a queued job:
if (strongerOnline() && Date.now() - job.created_at < BACKSTOP_MS) return null; // defer

Two timing guards keep it smooth:

• A startup grace — right after joining, hold off claiming until you’ve heard peers’ presence, so a stronger worker gets first dibs.
• A backstop — take a deferred job anyway if it’s sat unclaimed too long (the strong worker is overwhelmed or gone).

This is also where worker types matter for the UI: workers tag their hello with native: true/false, so you can show “1 server · 2 browsers” and nudge the user to add power.

6. A native worker (scaling past the browser)

The browser is a capable worker, but for real throughput you can run the same loop as a Node process — it logs in as the user (same ZK auth), joins the same Space, and claims from the same table. Because the payload is a manifest, it just readByManifests the bytes and does the work with native tools (Theater uses native ffmpeg). Package it as a CLI so a user can point a spare machine at their queue:

npx @yourapp/worker --username you

Nothing in the queue changes — a native worker is simply a peer with a high score. Add machines, add throughput.

Note

Offline + P2P are browser features (IndexedDB / WebRTC). In Node the worker reads payloads over origin instead — same manifests, same results. The browser peers shards over WebRTC when devices are co-present; origin is always the fallback.

7. Large payloads: stream, don’t buffer

A payload that fits in memory is easy. A multi-GB one — a 4K movie, a big dataset — is where naive code breaks: reading a whole File/Blob into one array throws past the browser’s ~2 GB ArrayBuffer cap (and Node’s ~4 GB), surfacing as the infamous “the file could not be read.” The queue handles this in two places.

Storing streams by design. space.putFile slices the source per chunk as it encrypts and uploads, so the peak footprint is ~one chunk no matter how big the file — enqueuing a 30 GB input never materializes it in memory.

Reading it back, pick the surface by size. Small payloads: getFile / readByManifest return the whole Uint8Array. Multi-GB payloads: use space.getFileStream(manifest) — an async iterator of decrypted chunks — and pipe straight to disk or a socket, never holding the whole thing:

// Native worker: stream the raw to a temp file, then run a native tool on it.
import { createWriteStream } from "node:fs";
import { once } from "node:events";

const out = createWriteStream(tmpPath);
for await (const chunk of space.getFileStream(manifest)) {
  if (!out.write(chunk)) await once(out, "drain");   // respect backpressure
}
await new Promise((res, rej) => out.end((e) => (e ? rej(e) : res())));
await runNativeTool(tmpPath);                          // ffmpeg reads the file directly

Capability includes size

Grading by power has a second axis: can this worker even hold the payload? An in-browser transcoder (ffmpeg.wasm, ~2 GB of WASM memory) cannot process a 4K movie — so it must not claim one. Add a size ceiling to your “only claim what you can do” filter and let the payload fall to a native worker that streams to disk:

const MAX_LOCAL_BYTES = 1_000_000_000; // ~1 GB — the in-browser tool's ceiling
const canDo = (j) =>
  (haveLocalCopy(j) || j.input_manifest) && (j.input_size ?? 0) <= MAX_LOCAL_BYTES;

Surface it in the UI (“too large for this browser — waiting for the native worker”) so an oversized job reads as routed, not stuck.

Gotchas worth stealing

These are the sharp edges the Theater build hit — cheaper to read than to rediscover:

• Persist a stable worker id (e.g. in localStorage). If it changes on every reload, a device sees its own in-flight jobs as belonging to a stranger and the attribution/coordination breaks.
• Parse the manifest — the json column may return it as a string; feed a parsed object to readByManifest or you’ll get manifest.chunks is not iterable.
• Throttle writes. Drive the UI from an in-memory cache + pub/sub for instant feedback; write the DB (progress, heartbeat) every few seconds, not on every tick.
• Claim is best-effort. Write-then-re-read verification is enough for one user’s devices. Don’t reach for heavyweight distributed locking.
• Release, don’t fail. A worker that can’t fetch a payload returns the job to queued; only a genuine processing error sets status: "error".

Where to go next

• Choosing where data lives — why the queue is client.db, the payload is client.storage, and coordination is client.space.
• client.db reference — queries, update, the owner-scoping rule.
• client.space reference — channels, sendMessage / onMessage, presence.
• Offline & sync — content-addressed shards, the manifest as a read capability, and peer transfer.