liteworks 0.1.1

SQLite-backed task queue (stdlib only, no dependencies)

liteworks
=========

SQLite-backed task queue for Python 3.12+. Zero dependencies; uses Python's
stdlib HTTP server. Each queue is one SQLite file. Workers on multiple machines
connect over HTTP. No external service required.


ARCHITECTURE
------------

Uses a visibility timeout. The atomic claim is one SQL statement:

    UPDATE jobs SET claimed_by = ?, claimed_at = ?
    WHERE id = (
        SELECT id FROM jobs
        WHERE (claimed_by IS NULL OR claimed_at < ?) AND scheduled_at <= ?
        ORDER BY scheduled_at LIMIT 1
    ) RETURNING *

Ack uses a fencing token so a timed-out worker cannot clobber a fresh claim:

    DELETE FROM jobs WHERE id = ? AND claimed_by = ? RETURNING *

Ack + DLQ write are atomic (one SQL transaction via POST /{queue}/fail).
Snooze clears the claim and advances scheduled_at.
Retry schedules a future re-delivery by advancing scheduled_at and
incrementing delivery_count.

One server, multiple queues: each queue maps to {data_dir}/{queue}.db and
is created lazily on first use.

One handler per message type; fanout belongs in the handler.


INSTALL
-------

    pip install liteworks


QUICKSTART
----------

Start the server:

    liteworks serve --data-dir /var/liteworks --port 8000

In your application:

    from dataclasses import dataclass
    from liteworks import Liteworks, RetryPolicy, Snooze, Discard

    lw = Liteworks(base_url='http://localhost:8000', queue='emails')

    @dataclass
    class SendEmail:
        to: str
        subject: str

    @lw.handler
    def send_email(msg: SendEmail):
        # raise Snooze(60) to defer and retry in 60s (delivery_count unchanged)
        # raise Discard('bad_address') to move straight to DLQ
        pass

    lw.publish(SendEmail(to='[e-mail]', subject='Hello'))

    w = lw.worker()
    w.install_signal_handlers()
    w.run()


MULTIPLE QUEUES
---------------

One server hosts all queues. Each queue name maps to its own SQLite file.

    emails  = Liteworks(base_url='http://localhost:8000', queue='emails')
    reports = Liteworks(base_url='http://localhost:8000', queue='reports')

    emails.publish(SendEmail(...))
    reports.publish(GenerateReport(...))

Queue names: [a-zA-Z0-9_-], 1-64 characters.


DEMO
----

See demo.py for a runnable two-queue example with workers, stats, and a
browser dashboard. Run:

    python demo.py


STATS AND UI
------------

GET /{queue}/stats returns JSON:

    {
        "queue": "emails",
        "ready": 3,
        "claimed": 1,
        "scheduled": 0,
        "failed": 2,
        "completed": 47,
        "recent_failures": [
            {"id": "...", "reason": "SMTP timeout", "failed_at": 1720000000.0}
        ],
        "recent_runs": [
            {"id": "...", "worker_id": "w1", "completed_at": 1720000001.0}
        ]
    }

    lw.stats()   # via Transport; returns same dict

GET / returns a top-level dashboard (self-contained HTML, auto-refresh) showing
all queues and an enqueue form. Open http://localhost:8000/ in a browser.

GET /queues returns a JSON list of queue names.
GET /metrics returns Prometheus-format text metrics for all queues.


AUTH
----

Protect with a shared secret:

    liteworks serve --data-dir /var/liteworks --secret $LITEWORKS_SECRET

    lw = Liteworks(base_url='...', queue='emails', secret='...')

GET / is always accessible (browser loads the page before it can send auth).
GET /queues, GET /metrics, GET /{queue}/stats require Basic auth when secret set.
POST, DELETE, PATCH accept either HMAC-SHA256 x-signature (API clients) or
Basic auth (browser). Basic auth password is the secret; username is ignored.

Completed job retention (default: keep last 1000 per queue; 0 = disabled):

    liteworks serve --data-dir /var/liteworks --completed-retention 500

    lw = Liteworks(..., completed_retention=500)


RETRY POLICY
------------

Default: 3 retries, exponential backoff, 5s base, 300s cap, jitter on.

    from liteworks import RetryPolicy

    @lw.handler(policy=RetryPolicy(max_retries=5, base_delay=2.0, jitter=False))
    def handle(msg: MyMsg): ...


PUBLISH OPTIONS
---------------

    lw.publish(msg, msg_id='my-uuid')               # idempotency key
    lw.publish(msg, parent_id='parent-job-id')      # lineage
    lw.publish(msg, scheduled_at=time.time()+3600)  # deferred delivery


SCHEMA
------

Created automatically on first use of each queue:

    CREATE TABLE IF NOT EXISTS jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        delivery_count INTEGER DEFAULT 0,
        scheduled_at REAL NOT NULL,
        claimed_by TEXT,
        claimed_at REAL,
        created_at REAL NOT NULL
    );

    CREATE TABLE IF NOT EXISTS failed_jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        reason TEXT,
        failed_at REAL NOT NULL
    );

    CREATE TABLE IF NOT EXISTS completed_jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        delivery_count INTEGER DEFAULT 0,
        worker_id TEXT,
        completed_at REAL NOT NULL
    );

msg_id has a partial unique index (WHERE msg_id IS NOT NULL) enforcing
idempotent publish when the caller passes an explicit msg_id.
completed_jobs retains the last N entries per queue (trimmed on each ack),
where N is the completed_retention setting (default 1000; 0 disables tracking).


OPERATIONAL NOTES
-----------------

Visibility timeout and duplicate execution:
  liteworks delivers jobs at-least-once. If a handler runs longer than
  visibility_timeout, the job is reclaimed by another worker and executed
  again. The original worker's ack is silently dropped (fencing token), so
  there is no data corruption, but the work runs twice. To avoid this:

  - Set visibility_timeout to at least 2x the p99 execution time of your
    slowest handler.
  - Design handlers to be idempotent (safe to run more than once).

  There is no heartbeat or claim-renewal mechanism. If you need it, publish
  progress updates as separate jobs or use a longer timeout.

HTTPS:
  liteworks speaks plain HTTP. For TLS, terminate at a reverse proxy
  (Caddy, nginx) in front of the liteworks server. The HMAC signature on
  write requests provides integrity without TLS for internal networks, but
  do not expose the server directly to the internet without TLS.


DURABILITY
----------

Use Litestream to replicate each SQLite file to S3:

    litestream replicate /var/liteworks/emails.db s3://bucket/emails.db


PUBLIC API
----------

    serve(data_dir, host='', port=8000, secret=None,
          completed_retention=1000)                     module-level, blocking

    Liteworks(data_dir=None, queue='default',
              base_url='http://localhost:8000',
              visibility_timeout=30.0, secret=None,
              completed_retention=1000)
    Liteworks.handler(fn=None, *, policy=None)     decorator / decorator factory
    Liteworks.publish(msg, *, msg_id=None, parent_id=None, scheduled_at=None)
    Liteworks.stats()                              -> dict
    Liteworks.worker(*, poll_interval=0.5, backoff_interval=2.0, max_empty=4)
    Liteworks.serve(port=8000, host='')            blocking; requires data_dir

    Worker.run()
    Worker.stop()
    Worker.install_signal_handlers()

    Snooze(seconds)    raise from handler to defer without consuming a retry
    Discard(reason)    raise from handler to skip retries and go to DLQ

    RetryPolicy(max_retries=3, backoff='exponential', base_delay=5.0,
                max_delay=300.0, jitter=True)