schema.sql

CREATE TABLE IF NOT EXISTS jobs (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    queue           text        NOT NULL DEFAULT 'default',
    tenant          text        NOT NULL DEFAULT 'default',
    task            text        NOT NULL,
    payload         jsonb       NOT NULL DEFAULT '{}'::jsonb,
    -- scheduled: waiting for run_at (delayed, or backing off after a failure)
    -- queued:    runnable now, ordered by vtime
    status          text        NOT NULL DEFAULT 'queued'
                    CHECK (status IN ('scheduled', 'queued', 'running', 'done', 'dead')),
    -- Virtual time for fair ordering across tenants; set when the job becomes
    -- runnable. Workers always take the lowest vtime first.
    vtime           double precision,
    attempts        integer     NOT NULL DEFAULT 0,
    max_attempts    integer     NOT NULL DEFAULT 5 CHECK (max_attempts >= 1),
    run_at          timestamptz NOT NULL DEFAULT now(),
    locked_by       text,
    locked_until    timestamptz,
    last_error      text,
    idempotency_key text UNIQUE,
    created_at      timestamptz NOT NULL DEFAULT now(),
    started_at      timestamptz,
    finished_at     timestamptz,
    updated_at      timestamptz NOT NULL DEFAULT now(),
    CHECK (status <> 'queued' OR vtime IS NOT NULL)
);

-- One row per (queue, tenant): where that tenant has got to in virtual time,
-- and its share of the workers (weight 3 = three times the share of weight 1).
CREATE TABLE IF NOT EXISTS tenants (
    queue      text             NOT NULL,
    tenant     text             NOT NULL,
    weight     double precision NOT NULL DEFAULT 1 CHECK (weight > 0),
    last_vtime double precision NOT NULL DEFAULT 0,
    PRIMARY KEY (queue, tenant)
);

-- The claim query: lowest vtime among runnable jobs in one queue. A partial
-- index keeps it small however many finished jobs pile up in the table.
CREATE INDEX IF NOT EXISTS jobs_runnable_idx
    ON jobs (queue, vtime, id) WHERE status = 'queued';

-- The promoter: scheduled jobs whose time has come.
CREATE INDEX IF NOT EXISTS jobs_scheduled_idx
    ON jobs (run_at, id) WHERE status = 'scheduled';

-- The reaper: running jobs whose lease has expired.
CREATE INDEX IF NOT EXISTS jobs_running_lease_idx
    ON jobs (locked_until) WHERE status = 'running';