A durable job queue on SQLite, D1 or PostgreSQL

A process can die after accepting work but before reporting its result. Watchdog stores the job, claims ready work, and records one terminal outcome for each job. Your host supplies the executor and the wake that drives the next tick. You do not write the queued/running/settled state machine yourself.

Execution is once per claim, not a promise of exactly-once external effects. Recovery can run a job again. Use a stable effect identity with a provider that supports deduplication, and report outcome_unknown when the evidence cannot establish what happened.

Before you start

TERMINAL

npm install @fungi.computer/watchdog

Watchdog is ESM-only. Use Node 24 for the node:sqlite example. A Cloudflare Durable Object supplies its own synchronous SQLite storage. Shared D1 and PostgreSQL use separate adapters and require migrations before opening.

1. Run a job on node:sqlite

Save this as queue.ts in an ESM project:

TS

import { DatabaseSync } from "node:sqlite";
import { Watchdog, type WatchdogQueuedJob } from "@fungi.computer/watchdog";
import { fromNodeSqlite } from "@fungi.computer/watchdog/node-sqlite";

const database = new DatabaseSync("jobs.sqlite");
const active = new Set<string>();
const job: WatchdogQueuedJob = {
  jobId: "report-1",
  queue: "reports",
  lane: "default",
  priority: 0,
  payload: { reportId: "monthly-1" },
  recovery: { maxRecoveries: 1 },
  state: "queued",
};

try {
  const queue = await Watchdog.make({
    owner: fromNodeSqlite(database),
    execute: async (claimed, signal) => {
      active.add(claimed.jobId);
      try {
        if (signal.aborted) return { kind: "cancelled" };
        console.log(claimed.payload);
        return { kind: "completed" };
      } finally {
        active.delete(claimed.jobId);
      }
    },
    isAlive: async (running) => active.has(running.jobId),
    // This demo ticks manually; a service needs a real timer or durable alarm.
    wake: { recompute: async () => {} },
  });

  console.log(await queue.ingest(job));
  console.log(await queue.tick());
  console.log(await queue.read(job.jobId));
} finally {
  database.close();
}

Run node queue.ts. The first run ingests and settles report-1 as completed. Re-running with the same database and identical job reports a duplicate admission and no ready work. An identical replay does not create another job.

jobs.sqlite is persistent; do not delete it to recover from a crash. This example’s liveness set is valid only for one runtime executing in this process. After a restart it is empty, so a prior running claim is no longer alive. It is not a probe for a process running elsewhere. When finished experimenting, close the program before removing its disposable database.

2. Put the queue inside a Durable Object

The queue API does not change. Replace the Node adapter with the Durable Object’s ctx.storage.

Illustrative: this opener receives the object’s storage and host callbacks. The object must implement those callbacks and arrange for its alarm to call tick(); this is not a complete Worker deployment.

TS

import {
  Watchdog,
  type WatchdogOptions,
  type WatchdogSqliteOwner,
} from "@fungi.computer/watchdog";

export function openObjectQueue(
  storage: WatchdogSqliteOwner,
  ports: Omit<WatchdogOptions, "owner">,
) {
  return Watchdog.make({ ...ports, owner: storage });
}

// In your Durable Object: await openObjectQueue(ctx.storage, ports).
// Then use the same ingest(job), tick() and read(job.jobId) calls.

Open one SQLite runtime per owner. The host’s isAlive checks whether the claimed execution still exists, and wake.recompute() arms the next timer or alarm after durable changes. A no-op wake does not make a production queue run itself.

3. Use a shared D1 or PostgreSQL database

Apply watchdogD1Schema or watchdogPostgresSchema through your migration owner first. These are package-owned SQL statement arrays exported from the respective adapter. Opening a runtime never migrates the database.

Illustrative: these openers require an already migrated database and real executor/wake callbacks. They open the same job engine as the SQLite example.

TS

import {
  createD1Watchdog,
  type AsyncWatchdogOptions,
} from "@fungi.computer/watchdog/d1";

export function openD1Queue(
  database: Parameters<typeof createD1Watchdog>[0]["database"],
  ports: AsyncWatchdogOptions,
) {
  return createD1Watchdog({ ...ports, database });
}

For PostgreSQL:

TS

import {
  createPostgresWatchdog,
  type AsyncWatchdogOptions,
} from "@fungi.computer/watchdog/postgres";

export function openPostgresQueue(
  database: Parameters<typeof createPostgresWatchdog>[0]["database"],
  ports: AsyncWatchdogOptions,
) {
  return createPostgresWatchdog({ ...ports, database });
}

Both runtimes expose ingest, read, tick, cancel, failLane, purge and hasPendingWork. Use the same queued job shape from step 1. They determine liveness from private claim leases rather than a host-supplied isAlive. Leases last two minutes and renew every ten seconds. An expired or replaced claim cannot renew or settle; the executor must honour its AbortSignal.

A durable host wake must still drive ticks and recovery after a process dies. For atomic admission with a domain write, D1’s enqueue(job) prepares a statement for the producer’s native batch; PostgreSQL’s enqueue(client, job) uses its existing transaction client. The caller owns that transaction.

Recovery, payloads and lanes

For an owner-local SQLite queue, each tick first probes the oldest running claim. A false liveness result requeues it if recovery budget remains; otherwise it settles as interrupted. A throwing probe leaves the claim unchanged and can report liveness_unknown. Unknown is not evidence that a job has died.

Watchdog treats the JSON payload, queue name and lane as opaque domain values. Your executor validates what the payload means. Ordinary tick() is single-flight per runtime; another tick while it is running returns busy. Giving jobs different lanes does not itself create a worker pool.

The executor can report completed, failed, cancelled or outcome_unknown. Exceptions and interruptions remain failures, not fabricated success receipts. Provider effects remain at-least-once.

For cron cadences, occurrence identity and retry policy, use Almanac. The host still owns storage and timers. See Watchdog’s API and examples for cancellation, bounded lane refusal and transactional admission.