BroadwayPgQue

Official Broadway producer for PgQue — 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 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 TRUNCATEd. 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 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:

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

Usage

Define a Broadway pipeline module:

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:

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):

-- 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:

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):

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)

Custom Message Transformers

Transform raw messages before Broadway processors handle them:

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:

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

Configuration Reference

BroadwayPgQue.Producer uses NimbleOptions 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:

1. Pure Postgrex (No Ecto Required!)

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

# 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:

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

Acknowledgement & Error Handling Options

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.