Metadata-Version: 2.4
Name: postgresforge
Version: 1.0.1
Summary: A PostgreSQL-native distributed task queue for Python providing durable jobs, at-least-once execution, leases, and operational tooling without requiring Redis or Kafka.
Project-URL: Homepage, https://github.com/google/postforge
Project-URL: Repository, https://github.com/google/postforge
Author-email: Antigravity <antigravity@google.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.9
Requires-Dist: asyncpg>=0.29.0
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: mypy>=1.10.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Description-Content-Type: text/markdown

# PostForge

PostForge is a rigorous, transactional, at-least-once Python task queue and background worker system built **natively on PostgreSQL**.

**No Redis. No RabbitMQ. No Kafka.** 

PostgreSQL is your durable source of truth and distributed coordination layer. PostForge leverages native PostgreSQL features like `FOR UPDATE SKIP LOCKED` and row-level `NOWAIT` semantics to provide high-throughput, conflict-free concurrency.

## Philosophy

PostForge is designed for applications where consistency and correctness matter more than ultra-low-latency message fanout. 

By keeping your tasks in the same database as your application data, you gain **Atomic Enqueueing**: you can enqueue a background job in the *exact same transaction* as your business logic. If the transaction rolls back, the job is never enqueued. If it commits, the job is guaranteed to run.

## Features

- **PostgreSQL Native**: Relies solely on PostgreSQL `SKIP LOCKED`. No external message brokers required.
- **Transactional Enqueue**: Enqueue jobs atomically inside your existing application database transactions.
- **Idempotency Contracts**: Built-in idempotency keys ensure tasks are never double-enqueued in race conditions.
- **Lease Fencing**: Safe concurrency. When a worker claims a job, it holds a cryptographic lease token. If the worker stalls and loses its lease, any subsequent operations by the stalled worker are strictly fenced out.
- **Worker Recovery & Heartbeats**: Workers automatically heartbeat their jobs. If a worker crashes, its leases expire, and other workers safely recover and retry the abandoned jobs.
- **At-Least-Once Execution**: Guaranteed delivery.
- **Automated Cruft Cleanup**: Terminal jobs (completed/failed) are continuously swept to prevent table bloat.
- **Observability**: Built-in operational CLI and SDK primitives to introspect the queue.
- **Asyncio Native**: Built from the ground up for Python `asyncio` and `asyncpg`.

---

## Installation

```bash
pip install postgresforge
```

*Requires Python 3.9+ and PostgreSQL 12+.*

## Quickstart

### 1. Database Setup

Apply the PostForge schema to your PostgreSQL database. The schema creates the `postforge_jobs`, `postforge_idempotency`, and `postforge_workers` tables.

```bash
# You can find the schema in schema.sql
psql -d mydatabase -f schema.sql
```

### 2. Enqueueing Jobs

You can use the PostForge SDK to enqueue jobs from your web application (e.g. FastAPI, Starlette).

```python
import asyncpg
from postforge.queue import enqueue

async def process_payment(conn: asyncpg.Connection, user_id: str, amount: float):
    # Enqueue a job transactionally!
    await enqueue(
        conn=conn,
        queue="payments",
        task="charge_card",
        payload={"user_id": user_id, "amount": amount},
        priority=10
    )
```

### 3. Worker Runtime

Define your tasks and start a worker to process them.

```python
import asyncio
import logging
from postforge.worker import Worker
from postforge.models import Job

logging.basicConfig(level=logging.INFO)

async def charge_card(job: Job) -> None:
    # Business logic here
    user_id = job.payload["user_id"]
    amount = job.payload["amount"]
    print(f"Charging {user_id} ${amount}...")

async def main():
    dsn = "postgres://user:password@localhost:5432/mydatabase"
    
    # Initialize the worker
    worker = Worker(dsn=dsn, queue="payments", concurrency=10)
    
    # Register tasks
    worker.register("charge_card", charge_card)
    
    # Run the worker loop
    await worker.start()

if __name__ == "__main__":
    asyncio.run(main())
```

---

## Execution Semantics & Architecture

### At-Least-Once Delivery
PostForge provides **at-least-once** execution semantics. If a worker process abruptly dies while executing a task, the job's lease will eventually expire. Another worker's background recovery loop will detect the expired lease and transition the job back to the `available` state for re-execution.

### Lease Fencing
To prevent the "Split-Brain Worker" problem, PostForge issues a unique cryptographic `lease_token` (a UUID) every time a job is claimed. If a worker stalls (e.g. due to CPU starvation or network partition), its lease expires and another worker may claim the job (getting a *new* lease token). If the original worker wakes up and attempts to `complete()` the job using its old lease token, PostgreSQL will explicitly reject it with a `LeaseLostError`.

### Transactional Ownership
When you enqueue a job using an existing `asyncpg.Connection` that is inside a transaction, the job is bound to that transaction's visibility. It will not be visible to workers until the transaction commits.

---

## CLI & Observability

PostForge provides a minimal, secure CLI to inspect jobs and workers.

### Configuration
Set the `POSTFORGE_DSN` environment variable:
```bash
export POSTFORGE_DSN="postgres://user:password@localhost:5432/db"
```

### Queue Statistics
Show aggregate queue metrics without loading jobs into memory.
```bash
postforge queue stats
postforge queue stats --queue payments
```

### Job Inspection
Inspect a specific job. For security, `lease_token` and `locked_until` are carefully managed, and the `lease_token` is never exposed by the CLI.
```bash
postforge job get <job-id>
```

### Job Listing
List jobs, filtered by queue or status.
```bash
postforge job list
postforge job list --status failed
postforge job list --queue payments --limit 50
```

### Administrative Retry
Atomically retry a permanently failed job. This preserves `attempts` and `last_error` so that the historical failure is not silenced.
```bash
postforge job retry <job-id>
```

### Worker Registry
List active and stale workers. Workers automatically register on startup and send periodic heartbeats. 
```bash
postforge worker list
```

---

## Benchmarks & Limitations

PostForge uses native Postgres locking. Benchmarks on standard hardware demonstrate robust performance for background task workloads.

* **Enqueueing**: ~100k jobs/sec via batch inserts.
* **Claim Contention**: ~1.7k claims/sec under extreme lock contention.
* **End-to-End Throughput**: ~400 full job lifecycles (enqueue -> claim -> execute -> complete) per second per Python worker process.

**Important Note**: These are *local benchmark observations*, not universal performance guarantees. Your actual throughput will depend entirely on your PostgreSQL server capacity, network latency, and the duration/complexity of your task execution logic. The primary bottleneck for pure end-to-end throughput is typically the Python `asyncio` event loop managing the volume of distinct transaction round-trips per job, not the database itself.

If your system requires processing millions of events per second, you need a stream processing engine like Kafka, not a database-backed task queue. PostForge is designed for high-value transactional background tasks (e.g. processing payments, sending emails, generating reports) where database consistency is paramount.
