Background jobs

Gebweb ships a DB-backed background-job runner: producers enqueue work from request handlers; one or more gebweb worker processes poll a shared gebweb_jobs table and run the registered @Job("name") handler.

The queue lives in the same database as your application data (SQLite, Postgres, or MySQL); there is no separate broker to operate. Failed jobs retry with exponential backoff up to maxAttempts, then move to the failed status with last_error captured.

Wiring

import gebweb;
import db;

let conn = db.connect("sqlite", "./app.db");
let app = gebweb.app([HomeController()]);
gebweb.useJobs(app, conn);
gebweb.registerJobHandlers(app, [WelcomeMailer()]);

useJobs auto-creates the gebweb_jobs table if it doesn't exist. Options:

Pass overrides as a dict: gebweb.useJobs(app, conn, {"maxAttempts": 10}).

Enqueuing

Any code holding the app reference can enqueue a job:

class SignupController {
    @Post("/signup")
    func signup(SignupForm form): dict<string, any> {
        let user = userRepo.create(form);
        gebweb.enqueue(app, "welcome-email", {"to": user.email, "name": user.name});
        return {"id": user.id};
    }
}

gebweb.enqueue(app, name, payload, opts?) returns the new job id. payload is any JSON-stringifiable dict. Options:

• runAt: unix-seconds timestamp for delayed execution.
• priority: "high", "default", "low", or any string drain order you configure (see Priorities). Overrides the handler's @Job(priority: ...) default.
• unique: a dedupe key (see Unique jobs). Overrides the handler's @Job(unique: ...) default.

Handlers

Declare a class with one or more methods decorated with @Job("name"). Each method receives a gebweb.Job context:

import gebweb;

class WelcomeMailer {
    @Job("welcome-email")
    func send(gebweb.Job job): void {
        let to = job.payload["to"] as string;
        let name = job.payload["name"] as string;
        mailer.send(to, "Welcome, " + name + "!", "<p>Glad you're here.</p>");
    }
}

Job fields:

Throwing from a handler triggers a retry (or final failure once maxAttempts is reached); returning normally marks the job completed.

Job options

@Job takes named arguments that configure a handler's scheduling and failure behaviour:

class Mailer {
    @Job("welcome-email", priority: "high", retry: {"maxAttempts": 3, "backoff": "exponential", "baseMs": 1000}, timeoutMs: 30000)
    func send(gebweb.Job job): void { ... }
}

Any of these can be set per-enqueue too via the enqueue opts (priority, unique), which override the handler default.

Priorities

Each job carries a priority string ("default" if unset). The worker drains all jobs of the highest priority before moving to the next, so a flood of low-priority work never starves urgent jobs. Configure the order with queues (highest first):

gebweb.runWorker(app, {"queues": ["high", "default", "low"]});

The default order is ["high", "default", "low"]. A job whose priority is not in the list is treated as lowest. The CLI worker takes the same default; pass a custom order from runWorker in main.gb.

Unique jobs

A unique key dedupes a job while an equivalent one is still pending or running: enqueuing the same key returns the existing job id instead of inserting a duplicate. Once that job completed or failed, a fresh enqueue is allowed again.

The key is a template. A literal string is used verbatim; a $payload.<field> template resolves against the enqueued payload:

class Reindex {
    /* one reindex per user at a time */
    @Job("reindex", unique: "$payload.userId")
    func run(gebweb.Job job): void { ... }
}

gebweb.enqueue(app, "reindex", {"userId": "42"});
gebweb.enqueue(app, "reindex", {"userId": "42"});   /* same id: deduped */
gebweb.enqueue(app, "reindex", {"userId": "99"});   /* distinct */

The guard is enforced at enqueue time; a partial unique index backs it where the database supports one.

Per-handler retry

retry overrides the queue-wide maxAttempts / backoffMs for one handler. As a dict it takes maxAttempts plus a backoff curve ("fixed", "linear", or "exponential") over baseMs:

@Job("charge", retry: {"maxAttempts": 5, "backoff": "exponential", "baseMs": 1000})

For full control, pass a callable that maps the (1-based) attempt number to a delay in milliseconds:

@Job("charge", retry: func(attempt: int): int { return attempt * 2000; })

Per-job timeout

timeoutMs bounds how long a single execution may run. Past the deadline the worker stops waiting and releases the job's claim so it retries (or fails once attempts are exhausted). The timeout is cooperative: the handler's own in-flight work is not forcibly interrupted, it finishes in the background while the worker moves on.

Running a worker

The gebweb worker CLI subcommand re-runs your project's src/main.gb with GEBWEB_RUN=worker in the environment. Branch on it so the same binary serves HTTP and background work:

import gebweb;
import db;
import sys;

let conn = db.connect("sqlite", "./app.db");
let app = gebweb.app([HomeController()]);
gebweb.useJobs(app, conn);
gebweb.registerJobHandlers(app, [WelcomeMailer()]);

if (sys.getenv("GEBWEB_RUN") == "worker") {
    gebweb.runWorker(app);
} else {
    gebweb.serve(app, "0.0.0.0:3000");
}

Run multiple workers in parallel by starting gebweb worker in multiple processes - each row is claimed atomically via a conditional UPDATE, so the same job can't be picked up twice.

Graceful shutdown

On SIGTERM or SIGINT, a worker finishes the job it is currently running and then stops, rather than being killed mid-job. In the long-running mode gebweb.runWorker(app) installs the signal handlers for you (they call gebweb.shutdown(app), which flips the drain flag the loop checks between jobs). The gebweb worker CLI forwards termination signals to the worker process, so a container stop (Docker / Kubernetes SIGTERM) drains in-flight work before exit. A job interrupted before its worker could finish is recovered by crash recovery.

The signal handlers are installed only in the blocking production mode; the test helpers (drainOnce / maxJobs) leave the host process's signal handling untouched.

Concurrency within a worker

By default a worker runs one job at a time. Set maxConcurrency to run several at once in a single process:

gebweb.runWorker(app, {"maxConcurrency": 4});

The worker keeps up to that many jobs in flight via a bounded async pool, claiming more as each finishes. This raises throughput for IO-bound work (a job waiting on the network or database lets others run) without starting more processes.

Concurrent jobs share the same handler instance, so handlers run under maxConcurrency > 1 must be concurrency-safe: they may use job.payload, local variables, and the database freely, but must not mutate shared in-process state (a handler field holding a dict/list, a module-level collection). This is the same contract as request handlers under load. If you are unsure, leave maxConcurrency at 1 and scale out with multiple worker processes instead, which keeps each worker fully isolated.

maxConcurrency composes with maxJobs and drainOnce (both wait for all in-flight jobs before returning) and with graceful shutdown (in-flight jobs finish before the worker stops).

Per-queue concurrency

queueConcurrency caps how many jobs from one queue (priority) run at the same time, within the overall maxConcurrency. Use it to keep a CPU-heavy queue from taking every slot:

gebweb.runWorker(app, {
    "queues": ["high", "default", "reports"],
    "maxConcurrency": 4,
    "queueConcurrency": {"reports": 1},
});

A queue only drains when it is listed in queues, so add any custom priority there. Jobs reach a queue through @Job(priority: "reports") or gebweb.enqueue(app, name, payload, {"priority": "reports"}).

In-process worker

A single-binary deployment can run the worker inside the server process instead of as a separate gebweb worker. Pass inProcessWorker to useJobs: true, or a dict of the runWorker options above (queues, maxConcurrency, queueConcurrency, names, workerId):

gebweb.useJobs(app, conn, {
    "inProcessWorker": {"queues": ["default", "reports"], "maxConcurrency": 4, "queueConcurrency": {"reports": 1}},
});
gebweb.registerJobHandlers(app, [Reports(), Mailers()]);
gebweb.cli(app);

The worker starts the first time the app listens (gebweb.cli, gebweb.serve or gebweb.listen) and runs as a background task alongside request handling, so its handlers follow the same concurrency-safe contract as maxConcurrency > 1. It ticks a registered scheduler like a standalone worker. gebweb.shutdown(app) (called by gebweb.cli on SIGINT/SIGTERM) stops it claiming new jobs and waits up to the shutdown timeout for in-flight jobs; a job cut off there is recovered by crash recovery. If the worker loop itself fails (for example the database is briefly unavailable) it logs the error and restarts after pollIntervalMs.

With SQLite, the server's own writes and the worker's writes come from different connections; open the job connection with "txLock": "immediate" (or keep the queue in its own database file) so read-then-write transactions wait for each other instead of failing with database is locked.

Filtering by job name

By default a worker drains every job name. To pin a worker process to a subset of names (so one server handles email, another handles image-resize), pass --job <name> to the CLI:

gebweb worker --job email --job sms

The flag is repeatable; the CLI populates GEBWEB_WORKER_JOBS which gebweb.runWorker(app) reads automatically. Equivalently in code:

gebweb.runWorker(app, {"names": ["email", "sms"]});

A worker pinned to a name list only ever claims rows whose name column appears in the list, leaving other jobs in pending for a differently-scoped worker (or the same worker on a later run) to process.

Crash recovery

When a worker claims a job it sets the row to running and stamps a lock. If that worker crashes mid-job, the row would otherwise stay running forever. The worker periodically reclaims any job whose lock is older than reclaimAfterMs (default 15 minutes), returning it to pending so a healthy worker picks it up.

Set reclaimAfterMs longer than your slowest job (and any per-job timeoutMs); otherwise a job that is still legitimately running could be reclaimed and run a second time. Set it to 0 to disable reclaiming. A reclaimed job whose attempts have already reached maxAttempts is sent straight to the dead-letter queue rather than retried, so a job that keeps crashing its worker cannot loop forever.

Dead-letter queue

Jobs that exhaust their retries land in the failed status with last_error captured. The gebweb worker dlq subcommand inspects and recovers them. Like gebweb migrate, it connects directly via $DATABASE_URL rather than loading your app:

gebweb worker dlq list                 # show failed jobs
gebweb worker dlq retry <id> [<id>...] # re-queue specific jobs
gebweb worker dlq retry --all          # re-queue every failed job
gebweb worker dlq purge <id> [<id>...] # delete specific jobs
gebweb worker dlq purge --all          # delete every failed job

retry resets a job to pending with its attempt counter cleared and run_at in the past, so the next worker picks it up immediately. purge deletes the rows outright. Both report how many rows they touched; neither affects jobs in any other status.

Inspection

For ad-hoc dashboards or tests:

• gebweb.jobs.get(app.jobConfig, id): ?dict<string, any> - fetch a row by id.
• gebweb.jobs.stats(app.jobConfig): dict<string, int> - counts keyed by status (pending, running, completed, failed).

Test pattern

Use the in-memory SQLite driver and runWorker in drain mode:

let app = gebweb.app([]);
let conn = db.connect("sqlite", ":memory:");
gebweb.useJobs(app, conn);
gebweb.registerJobHandlers(app, [MyHandler()]);
gebweb.enqueue(app, "my-job", {"x": 1});
gebweb.runWorker(app, {"drainOnce": true});

drainOnce: true processes every pending job - including follow-ons enqueued by handlers during the drain - and returns when the queue is empty. maxJobs: N is the safety-capped variant: same drain semantics, but stop after N jobs total to guard against runaway re-enqueueing in a buggy handler.

Scheduled tasks

The same worker process can run cron-style scheduled tasks. Multiple worker processes coordinate via leader election on a gebweb_scheduler_leader row, so scaling out doesn't double-fire.

class Cleanup {
    @Scheduled("0 * * * *")             /* every hour on the hour */
    func sweepStaleSessions(): void {
        db.exec("DELETE FROM sessions WHERE expires_at < ?", datetime.nowUnix());
    }

    @Scheduled("*/15 * * * *")          /* every 15 minutes */
    func refreshLeaderboard(): void {
        leaderboardService.rebuild();
    }
}

gebweb.useJobs(app, conn);
gebweb.useScheduler(app);
gebweb.registerScheduledTasks(app, [Cleanup()]);

Cron grammar (POSIX five-field subset): minute (0-59), hour (0-23), day-of-month (1-31), month (1-12), day-of-week (0-6, Sun=0; 7 also accepted as Sunday).

Each field accepts: *, N, N,M,O, N-M, */N, M-N/S.

When both day-of-month and day-of-week are restricted they are AND-ed (so 0 0 1 * 1 means "midnight on the 1st only if that day is Monday"). This differs from Vixie cron's OR semantics; the AND behaviour matches what users expect more often.

gebweb.useScheduler(app, opts = {}) options:

Scheduler exceptions are logged to stderr and swallowed - one broken task does not stop the rest. For at-most-once delivery under leader-election semantics, prefer enqueueing a regular @Job from the scheduled task body (the job retry machinery takes over from there).

Reference