broadway_pgque

# BroadwayPgQue

Official [Broadway](https://github.com/dashbitco/broadway) producer for [PgQue](https://pgque.dev/) — a high-throughput, zero-bloat background queue engine for PostgreSQL.

---

## Features

- **Zero Table Bloat:** Integrates with PgQue's rotating partitioned tables and instant `TRUNCATE` cycle.
- **Demand-Driven Concurrency:** Native backpressure and memory safety backed by Broadway and GenStage.
- **Native `LISTEN / NOTIFY` Wake-Up:** Subscribes to `LISTEN pgque_<queue>` via `Postgrex.Notifications`—wakes up in `< 1 ms` when `pgque.ticker()` seals a `0.1s` batch, while sleeping for `30s` when idle so your database is never hammered.
- **Adaptive Idle Backoff & Instant Elixir Wake-Up:** Automatically backs off polling when `LISTEN` is disabled (e.g. behind PgBouncer) and supports instant wakeup via `BroadwayPgQue.wake_up/1`.
- **Custom Transformers:** Easily decode JSON or translate raw PgQue records into domain structs.
- **Flexible Acknowledgements:** Built-in support for PgQue batch acknowledgements, delayed retries, and dead-letter queue (DLQ) routing.
- **Cooperative Subconsumers:** Built-in support for PgQue consumer groups (`pgque.receive_coop`).

---

## Why PgQue? (And When to Use It)

Most PostgreSQL queues rely on `SELECT ... FOR UPDATE SKIP LOCKED` combined with per-row `UPDATE` and `DELETE`. Under sustained load—or when a long-running query pins the `xmin` horizon—this creates massive dead-tuple accumulation, `VACUUM` pressure, index bloat, and throughput degradation.

[PgQue](https://github.com/NikolayS/pgque) brings Skype's battle-tested **PgQ** architecture to modern, managed PostgreSQL (14+) in pure SQL/PL/pgSQL:
- **Snapshot-Based Batching & `TRUNCATE` Rotation:** Events are grouped into batches via transaction snapshots and stored across rotating partitioned tables that are periodically `TRUNCATE`d. Dead tuples in the hot path stay at `0`.
- **Shared-Log Fan-Out (Kafka-like Cursors):** Unlike competing-consumer queues where a row is locked and deleted by one worker, PgQue maintains independent cursors per consumer on a shared event log—multiple pipelines can consume the exact same stream without duplicating data.
- **Zero Extensions Required:** Pure SQL + `pg_cron` (or `pg_timetable` / external ticker). Works out of the box on RDS, Aurora, Supabase, Neon, Cloud SQL, and self-hosted Postgres.

### When to Choose `BroadwayPgQue`

- **High-throughput event streams & ingestion** inside Postgres where `SKIP LOCKED` queues suffer from table bloat and autovacuum starvation.
- **Event fan-out:** You want multiple independent Broadway pipelines (e.g., `AnalyticsPipeline`, `NotificationsPipeline`, `AuditLogPipeline`) to each receive every event from a single shared queue without inserting duplicate rows.
- **Sustained heavy load alongside OLAP/reporting queries:** `TRUNCATE`-based rotation is immune to `xmin` horizon pinning.
- **Polyglot architectures:** Producers written in Go, Python, TypeScript, Ruby, or SQL triggers publishing into Postgres, consumed with Elixir Broadway's backpressure and batching.

### When NOT to Choose It (Use a Job Queue like [Oban](https://github.com/oban-bg/oban) Instead)

- **Single-digit millisecond (< 3 ms) dispatch is your top priority:** PgQue groups events into snapshot ticks (default 100 ms tick = ~50 ms median delivery latency, tunable down to ~8 ms). If you need instant row-level locking dispatch at low volume, `SKIP LOCKED` is faster.
- **Per-job workflow features:** You need individual job priorities, unique/deduplicated jobs, cron scheduling, or DAG workflows (`Oban.Workflow`). PgQue is a **durable event/message log**, not a per-job state machine.

---

## Installation

Add `broadway_pgque` to your dependencies in `mix.exs`:

```elixir
def deps do
  [
    {:broadway_pgque, "~> 0.1.0"},
    {:postgrex, ">= 0.0.0"}
  ]
end
```

---

## Usage

Define a Broadway pipeline module:

```elixir
defmodule MyApp.NotificationPipeline do
  use Broadway

  alias BroadwayPgQue.Producer, as: PgQueProducer

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          PgQueProducer,
          repo: MyApp.Repo,
          queue: "notifications",
          concurrency: 1
        }
      ],
      processors: [
        default: [concurrency: 20]
      ],
      batchers: [
        default: [batch_size: 100, batch_timeout: 2000]
      ]
    )
  end

  @impl Broadway
  def handle_message(_processor, message, _context) do
    # message.data contains the raw payload
    # message.metadata contains :id, :batch_id, :type, :retry_count, :created_at
    MyApp.Notifications.send(message.data)
    message
  end

  @impl Broadway
  def handle_batch(:default, messages, _batch_info, _context) do
    # Batch processing with automatic chunk buffering
    payloads = Enum.map(messages, & &1.data)
    MyApp.Notifications.send_bulk(payloads)
    messages
  end
end
```

### How Wake-Up & Polling Work (`LISTEN / NOTIFY` + Adaptive Fallback)

Just like the official PgQue Python and Ruby clients, `BroadwayPgQue` combines **PostgreSQL `LISTEN / NOTIFY`** with a safety-net fallback timer so you get **full `0.1s` speed without hammering the database**:

1. **Instant `LISTEN pgque_<queue>` Wake-Up (Default `listen: :auto`):**
   Whenever `pgque.ticker()` seals a new batch in Postgres (every `100 ms` by default), it emits `PERFORM pg_notify('pgque_' || queue_name, tick_id::text)`. `BroadwayPgQue.Producer` listens on `pgque_<queue>` via `Postgrex.Notifications` and fetches the batch **immediately (`< 1 ms` after the tick)**.
2. **Zero-Delay Backlog Draining:**
   As long as batches are available and processor demand exists, the producer fetches the next batch **immediately (`0 ms` delay)** as soon as the active batch is acknowledged.
3. **Zero Idle Database Spam (`30s` Safety Net):**
   When the queue is empty and `LISTEN` is active, the producer does **not** poll Postgres every `0.1s`—it sleeps for `max_poll_interval` (`30_000 ms` = 30s) as a safety net and wakes up instantly on the next `NOTIFY` or `BroadwayPgQue.wake_up/1`:

```elixir
BroadwayPgQue.wake_up(MyApp.NotificationPipeline)
```

### Faster Tick Tuning & Cooperative Parallelism

#### 1. Speed Up PgQue's Ticker in PostgreSQL
By default, `pgque.start()` ticks **10 times per second (every 100 ms)** (~50 ms median delivery latency). You can increase the tick frequency at runtime (accepts exact divisors of `1000` in the `1..1000` ms range; picks up on the next 1s scheduler slot):

```sql
-- 20 ticks/sec (every 50 ms)
SELECT pgque.set_tick_period_ms(50);

-- 100 ticks/sec (every 10 ms, ~8 ms median end-to-end delivery)
SELECT pgque.set_tick_period_ms(10);

-- Inspect current ticker status
SELECT * FROM pgque.status();
```

*(Note: Idle queues are cheap—when there are no producer writes, `pgque.ticker()` sees no sequence change, returns `NULL` without writing to WAL or emitting `NOTIFY`, and only writes an idle heartbeat tick once per `ticker_idle_period` = 1 minute).*

In tests or interactive demos where you want an immediate batch without waiting for the background ticker:

```sql
SELECT pgque.force_next_tick('notifications');
SELECT pgque.ticker();
```

#### 2. Cooperative Subconsumers & Shared `Postgrex.Notifications`
If downstream I/O (e.g., HTTP APIs, emails, webhooks) is the bottleneck, enable **cooperative subconsumers** (`receive_mode: :coop`) with multiple producer stages so workers pull slices of the batch concurrently.

When `repo: MyApp.Repo` is configured, all producer stages automatically share **a single `Postgrex.Notifications` connection** (`MyApp.Repo.BroadwayPgQue.Notifications`) out of the box (you can also pass an explicit `:notifications_conn` if you run your own `Postgrex.Notifications` in your supervision tree):

```elixir
producer: [
  module: {
    BroadwayPgQue.Producer,
    repo: MyApp.Repo,                         # Automatically shares 1 LISTEN connection across all workers
    queue: "notifications",
    consumer: "notifications_group",
    receive_mode: :coop,                       # Parallelize across cooperative subconsumers
    max_batch_size: 500,                       # Pull larger batches per round-trip
    ticker_max_lag: "0.02 seconds",            # Match faster 20ms-50ms tick rate
    rotation_period: "30 minutes"              # Optional: faster table rotation for high tick rates
  },
  concurrency: 4                               # Spawns worker_0 .. worker_3 cooperative subconsumers
]
```

#### 3. Choosing Settings by Workload (including PgBouncer `listen: false`)

| Workload Profile | Broadway Producer Settings | Recommended PgQue Ticker Settings |
| :--- | :--- | :--- |
| **Default (`LISTEN/NOTIFY` active)** | `listen: :auto`, `poll_type: :adaptive`, `max_poll_interval: 30_000` (0 idle spam, instant wakeup on tick) | Default (`100 ms` tick, `ticker_max_lag: "0.1 seconds"`) |
| **Ultra-low latency (`< 20ms` e2e)** | `listen: :auto`, `ticker_max_lag: "0.02 seconds"` | `SELECT pgque.set_tick_period_ms(20);` (50 ticks/sec) |
| **PgBouncer Transaction Mode (`listen: false`)** | `listen: false`, `poll_type: :adaptive`, `poll_interval: 100..250`, `max_poll_interval: 5_000` | Default (`100 ms` tick) + call `BroadwayPgQue.wake_up/1` |
| **Slow Background Batching (1s–5s+)** | `listen: false`, `poll_type: :adaptive`, `poll_interval: 2_000`, `max_poll_interval: 30_000` | `set_tick_period_ms(1000)`, `ticker_max_lag: "1 second"` |

### Custom Message Transformers

Transform raw messages before Broadway processors handle them:

```elixir
defmodule MyApp.Transformer do
  def transform(raw_msg, ack_ref, ack_data, _opts) do
    decoded = Jason.decode!(raw_msg.payload)

    %Broadway.Message{
      data: decoded,
      metadata: %{id: raw_msg.msg_id, type: raw_msg.type},
      acknowledger: {ack_data.acknowledger, ack_ref, Map.put(ack_data, :msg_id, raw_msg.msg_id)}
    }
  end
end
```

And configure:

```elixir
producer: [
  module: {
    PgQueProducer,
    repo: MyApp.Repo,
    queue: "notifications",
    transformer: {MyApp.Transformer, :transform, []}
  }
]
```

---

## Configuration Reference

`BroadwayPgQue.Producer` uses [NimbleOptions](https://hexdocs.pm/nimble_options) to validate all options.

### Database & Connection Pooling Options

`BroadwayPgQue` can run **100% without Ecto**. You can choose your connection pooling module or let it auto-detect:

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:pool` | `module` | `nil` (auto) | Module implementing `BroadwayPgQue.Pool` (`BroadwayPgQue.Pool.Postgrex`, `BroadwayPgQue.Pool.Ecto`, or custom). |
| `:conn` | `pid` \| `atom` | *None* | Direct Postgrex connection process PID or registered name (**no Ecto required**). |
| `:connection_opts` | `keyword` | *None* | Options passed to `Postgrex.start_link/1` to spin up a dedicated connection/pool (**no Ecto required**). |
| `:repo` | `atom` | *None* | Shared `Ecto.Repo` module. |
| `:pool_opts` | `keyword` | `[]` | Options passed to custom `pool.init/1` callback. |
| `:query_timeout` | `timeout` | `15_000` | Query timeout in milliseconds for PgQue SQL operations. |

#### 1. Pure Postgrex (No Ecto Required!)

To run BroadwayPgQue without having Ecto in your project at all:

```elixir
# Using an existing Postgrex process or pool
producer: [
  module: {
    BroadwayPgQue.Producer,
    pool: BroadwayPgQue.Pool.Postgrex,
    conn: :my_postgrex_pool, # or conn pid
    queue: "orders"
  }
]

# Or letting BroadwayPgQue manage its own dedicated Postgrex connection:
producer: [
  module: {
    BroadwayPgQue.Producer,
    queue: "orders",
    connection_opts: [
      hostname: "localhost",
      database: "my_db",
      username: "postgres",
      pool_size: 2
    ]
  }
]
```

#### 2. Writing a Custom Pool Module

Implement the `BroadwayPgQue.Pool` behaviour to plug in your own connection pool, telemetry wrapper, or test mock:

```elixir
defmodule MyApp.CustomPool do
  @behaviour BroadwayPgQue.Pool

  @impl true
  def init(opts) do
    # initialize or resolve connection state
    {:ok, opts[:conn_name]}
  end

  @impl true
  def query(conn_name, sql, params, opts) do
    Postgrex.query(conn_name, sql, params, opts)
  end
end
```

### Wake-Up, Polling & Scheduling Options

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:listen` | `:auto` \| `boolean` | `:auto` | Subscribes to `LISTEN pgque_<queue>` via `Postgrex.Notifications` for instant wakeup when `pgque.ticker()` seals a batch. Set to `false` behind PgBouncer `transaction` pooling. |
| `:notifications_conn` | `pid` \| `atom` | `nil` | Optional PID or registered name of an existing `Postgrex.Notifications` process to share across producers. |
| `:consumer` | `string` \| `atom` | `"broadway_<queue>"` | Consumer cursor registration name in PgQue. Different consumer names receive an independent copy of the stream (fan-out). |
| `:subconsumer` | `string` \| `atom` | `nil` | Subconsumer base name for cooperative consumer groups (`pgque.receive_coop`). Auto-suffixed when producer `concurrency > 1`. |
| `:receive_mode` | `:auto` \| `:standard` \| `:coop` | `:auto` | Uses `:coop` (`pgque.receive_coop`) if `:subconsumer` is provided, `:standard` otherwise. |
| `:poll_type` | `:adaptive` \| `:fixed` | `:adaptive` | Polling strategy when queue is empty: `:adaptive` (30s safety net when `LISTEN` is active, or exponential backoff up to `max_poll_interval`) or `:fixed` (constant `poll_interval`). |
| `:poll_interval` | `pos_integer` | `1000` | Base fallback polling interval in milliseconds when `LISTEN` is disabled. |
| `:max_poll_interval` | `pos_integer` | `30_000` | Maximum backoff / safety-net interval in milliseconds when queue is empty (matches Python/Ruby `30s` fallback). |
| `:backoff_factor` | `float` | `1.5` | Multiplier applied when consecutive polls yield no messages. |
| `:max_batch_size` | `pos_integer` | `100` | Maximum messages pulled per batch query. |
| `:ticker_max_lag` | `string` | `"0.1 seconds"` | PostgreSQL setting for queue ticker responsiveness (`pgque.set_queue_config`). |
| `:rotation_period` | `string` | `nil` | Optional interval for ring-buffer table rotation (e.g. `"2 hours"`, `"30 minutes"`). |
| `:history_tables` | `pos_integer` | `nil` | Optional number of partition tables in the queue ring-buffer (alias for `:ntables`, minimum `3`). |

### Acknowledgement & Error Handling Options

| Option | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `:acknowledger` | `atom` | `BroadwayPgQue.Acknowledger` | Broadway acknowledger module. |
| `:on_failure` | `:dead` \| `:retry` \| `function` | `:dead` | Action for unhandled failed messages: move to DLQ (`:dead`) or retry (`:retry`). |
| `:default_retry_delay` | `pos_integer` | `60` | Delay in seconds when scheduling failed messages for retry. |
| `:transformer` | `mfa` \| `function` | `{BroadwayPgQue.Producer, :transform, []}` | Callback to transform raw database maps into `Broadway.Message`. |

### Telemetry Events

`BroadwayPgQue` emits standard `:telemetry` events for visibility:

- `[:broadway_pgque, :notify, :received]` — `%{system_time: integer()}`, `%{queue: String.t(), consumer: String.t(), payload: String.t() | nil}`
- `[:broadway_pgque, :fetch, :start]` — `%{system_time: integer()}`, `%{queue: String.t(), consumer: String.t(), demand: integer()}`
- `[:broadway_pgque, :fetch, :stop]` — `%{duration: integer(), count: integer()}`, `%{queue: String.t(), consumer: String.t(), batch_id: integer()}`
- `[:broadway_pgque, :ack, :start]` — `%{system_time: integer()}`, `%{batch_id: integer(), queue: String.t(), total_count: integer(), failed_count: integer()}`
- `[:broadway_pgque, :ack, :stop]` — `%{duration: integer()}`, `%{batch_id: integer(), queue: String.t()}`
- `[:broadway_pgque, :ack, :exception]` — `%{duration: integer()}`, `%{batch_id: integer(), queue: String.t(), error: term()}`

---

## License

Apache License 2.0.