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.
• Adaptive Idle Backoff: Automatically backs off polling intervals when queues are empty to preserve database CPU.
• Instant Wake-Up: Bypasses idle polling backoffs immediately when new messages are enqueued 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).

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",
          poll_interval: 1000,
          max_poll_interval: 30_000,
          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

Instant Wake-Up on Enqueue

To eliminate any latency when messages are inserted into an idle queue:

BroadwayPgQue.wake_up(MyApp.NotificationPipeline)

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

Polling & Scheduling Options

Acknowledgement & Error Handling Options

Telemetry Events

BroadwayPgQue emits standard :telemetry events for visibility:

• [: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.