PgQueuer v1.0.0

From here on: strict semver. Patch for bug fixes,
minor for backward-compat features, major only for breaks. Public API frozen.

This release is a hard reset: cleaner core, fewer footguns, smaller surface,
new building blocks for retries, failure handling, and AI introspection.

Why upgrade

Database-level retry, finally done right. Raise RetryRequested from any
handler — the job stays in the queue, gets a bumped attempts counter, and
re-runs after your chosen delay. Survives worker crashes, restarts, OOM kills.

from pgqueuer import RetryRequested
from datetime import timedelta

@pgq.entrypoint("call_api")
async def call_api(job: Job) -> None:
    if response.status == 429:
        raise RetryRequested(delay=timedelta(seconds=30), reason="rate limited")

Exponential backoff in one line. DatabaseRetryEntrypointExecutor wraps
any handler with retries + backoff, computed from job.attempts. No
async-timeout, no in-process state.

@pgq.entrypoint(
    "flaky_api",
    executor_factory=lambda p: DatabaseRetryEntrypointExecutor(
        parameters=p, max_attempts=5, max_delay=timedelta(minutes=5),
    ),
)
async def flaky_api(job: Job) -> None: ...

Hold failed jobs for inspection. on_failure="hold" parks terminally
failed jobs in the queue table with status='failed' instead of deleting
them. Inspect with pgq failed, requeue with pgq requeue 42 43 44.

Concurrency limits are now global. concurrency_limit=5 means 5 across
your entire fleet, enforced at the database via FOR UPDATE SKIP LOCKED.
No more per-process semaphores that lied to you under horizontal scale.

MCP server for AI agents. pip install pgqueuer[mcp] ships an MCP server
with 11 read-only tools (queue size, throughput, stale jobs, schema info,
failed jobs, …). Plug it into Claude Code, Cursor, or any MCP client to let
your AI assistant inspect queue state.

Scheduled tasks get shared resources too. ScheduleContext mirrors the
queue handler Context — pass HTTP clients, DB pools, ML models in once via
PgQueuer(driver, resources={...}).

Bug fixes worth calling out

• Deferred jobs wake within ~100ms of their eligible time. Previously
  they could wait the full dequeue_timeout (default 30s).
• TOCTOU race on deferred jobs fixed — no more unnecessary full-timeout
  sleeps when a job becomes eligible mid-dequeue.
• Dequeue CTE split to restore partial-index usage on the hot path —
  meaningful throughput recovery on busy queues.
• Heartbeat waits its interval before the first send — no redundant DB
  write on short jobs.
• Psycopg NOTIFY watcher survives bad callbacks instead of dying silent.
• In-memory adapter correctly releases dedupe_key on held/failed jobs.

Breaking changes — the short version

Full list and migration steps in
RELEASE.md (all
23 numbered) and the upgrade guide.

The themes you'll hit:

1. Async only. Sync entrypoints (def handler) raise TypeError at
   decoration time. Wrap blocking calls with asyncio.to_thread().

2. Factories are async context managers. Use @asynccontextmanager +
   yield. Plain async def factory() -> PgQueuer is rejected.

3. Schema migration required. Run pgq upgrade (or pgq install on a
   fresh DB). Adds the attempts column and 'failed' enum value.

4. Concurrency knobs collapsed. serialized_dispatch=True →
   concurrency_limit=1. Per-entrypoint retry_timer= → single
   heartbeat_timeout= on pgq.run() (default 30s). requests_per_second
   removed — use concurrency_limit for deterministic backpressure.

5. CLI connection flags gone. --pg-host / --pg-user / etc. removed.
   Use standard libpq env vars (PGHOST, PGUSER, …) or --pg-dsn.

6. Retry executor renamed. RetryWithBackoffEntrypointExecutor →
   DatabaseRetryEntrypointExecutor. Drops the async-timeout runtime dep.

7. Manager constructors take a Queries. QueueManager(driver) →
   QueueManager(Queries(driver)). Same for SchedulerManager and
   CompletionWatcher. PgQueuer(driver) is unchanged.

8. 14 internal shim modules removed from package root (pgqueuer.buffers,
   pgqueuer.qb, pgqueuer.helpers, …). Public API shims (pgqueuer.db,
   .queries, .qm, .sm, …) are untouched.

9. Custom async Driver implementations must add notify(channel, payload).
   pg_notify moved from the queries layer into the driver protocol.
   SyncDriver is unaffected.

Upgrade in 60 seconds

pip install --upgrade pgqueuer
pgq upgrade

Then:

# 1. async handlers
@pgq.entrypoint("my_job")
async def my_job(job: Job) -> None: ...

# 2. async-context-manager factory
@asynccontextmanager
async def factory():
    conn = await asyncpg.connect()
    yield PgQueuer(AsyncpgDriver(conn))
    await conn.close()

# 3. global heartbeat instead of per-entrypoint retry_timer
await pgq.run(heartbeat_timeout=timedelta(seconds=60))

Run your tests. Most breakages surface at decoration or startup time, so
problems are immediately visible.

What's next

v1.x is for additions and fixes. No more API churn without a major bump.

Thanks to everyone who filed issues, sent PRs, and stress-tested early
versions. PgQueuer is production-ready.

Full Changelog: https://github.com/janbjorge/pgqueuer/compare/v0.26.3...v1.0.0