# `BroadwayCloudPubSub.Producer`
[🔗](https://github.com/dashbitco/broadway_cloud_pub_sub/blob/v2.0.0-rc.0/lib/broadway_cloud_pub_sub/producer.ex#L1)

A Broadway producer that uses the gRPC StreamingPull API to receive
messages from a Google Cloud Pub/Sub subscription.

## Overview

This producer opens a persistent bidirectional gRPC stream to the Pub/Sub
service and receives messages as the server pushes them. This is more
efficient than the HTTP pull approach (`BroadwayCloudPubSub.Pull.Producer`) for
workloads that require low latency or high throughput.

Each producer process (N = `producer: [concurrency: N]`) starts and links
its own **StreamManager** (GenServer), giving N independent gRPC streams
sharing a single `clientID`.

Key components:

  * **StreamManager** - GenServer that owns the gRPC bidirectional stream,
    manages connection lifecycle (connect/reconnect/backoff), extends message
    leases, and dispatches messages to the linked Producer when demand is
    available. Started via `start_link` from `Producer.init/1`.

  * **Producer** - GenStage process that bridges StreamManager to Broadway.
    Tracks downstream demand and forwards messages to processors.

  * **UnaryAckSupervisor** - shared across all producers. Supervises
    AckBatcher and UnaryRpcClient, which batch and send ack/nack/modifyAckDeadline
    requests via separate unary RPCs (not on the streaming connection).

## Usage

    Broadway.start_link(MyPipeline,
      name: MyPipeline,
      producer: [
        module:
          {BroadwayCloudPubSub.Producer,
           goth: MyApp.Goth,
           subscription: "projects/my-project/subscriptions/my-subscription",
           max_outstanding_messages: 1000}
      ],
      processors: [default: [concurrency: 10]]
    )

## Options

* `:subscription` - Required. The name of the subscription, including the project.
  For example, if your project is `"my-project"` and your
  subscription is `"my-subscription"`, the full name is
  `"projects/my-project/subscriptions/my-subscription"`.

* `:goth` (`t:atom/0`) - The `Goth` module to use for authentication. Note that this option only
  applies to the default token generator.

* `:token_generator` - An MFArgs tuple that will be called before each gRPC connection to fetch
  an authentication token. Should return `{:ok, String.t()} | {:error, any()}`.
  By default this will invoke `Goth.fetch/1` with the `:goth` option.

* `:max_outstanding_messages` (`t:pos_integer/0`) - The maximum number of outstanding messages (delivered but not yet
  acknowledged) that the server will push. Acts as the primary flow
  control mechanism. Analogous to AMQP `prefetch_count`. The default value is `1000`.

* `:max_outstanding_bytes` (`t:pos_integer/0`) - The maximum total size in bytes of outstanding messages. The server
  will not push more messages if the total byte size of outstanding
  messages exceeds this limit. The default value is `104857600`.

* `:on_success` - Configures the acknowledgement behaviour for successfully processed
  messages. The default value is `:ack`.

* `:on_failure` - Configures the acknowledgement behaviour for failed messages. The
  default makes failed messages immediately available for redelivery. The default value is `{:nack, 0}`.

* `:on_shutdown` - Configures what happens to messages received but not yet processed
  when the producer is shut down.

      - `{:nack, seconds}` - Sends a `modifyAckDeadline` request with the
      given `seconds` for all outstanding messages, making them available
      for redelivery after that delay.
      - `:nack` - Equivalent to `{:nack, 0}`. Immediately makes unprocessed
      messages available for redelivery.
      - `:noop` - Does nothing. Messages become available after their ack
      deadline expires naturally.

  The default value is `{:nack, 5}`.

* `:stream_ack_deadline_seconds` - The number of seconds the server will wait before re-delivering an
  unacknowledged message. Must be between 10 and 600.
  The producer will extend leases automatically before this deadline. The default value is `60`.

* `:max_extension_ms` (`t:pos_integer/0`) - The maximum total time in milliseconds that a message's ack deadline will
  be extended from the moment of initial receipt. After this duration, the
  message is dropped from lease management and the server will redeliver it.
  This prevents a stuck consumer from holding messages indefinitely. The default value is `3600000`.

* `:enable_message_ordering` (`t:boolean/0`) - When `true`, messages with the same `ordering_key` are routed to the
  same Broadway processor and processed sequentially. This guarantees
  in-order delivery for ordered subscriptions.

      Ordering is enforced via Broadway's built-in `:partition_by` option,
      which assigns messages with the same `orderingKey` metadata to the
      same processor partition. The subscription itself must also have
      message ordering enabled in Google Cloud Pub/Sub.

      When `false` (default), messages are distributed across processors
      without regard to ordering key, matching the unordered behaviour of a
      standard Pub/Sub subscription.

      Note: the server will also report whether the subscription has ordering
      enabled in each `StreamingPullResponse.subscription_properties`. This
      client-side option controls whether to enforce it in the Broadway
      processing topology.

  The default value is `false`.

* `:client_id` (`t:String.t/0`) - An identifier shared across all streaming connections opened by this
  pipeline. When a stream disconnects and reconnects with the same
  `client_id`, the server transfers any guarantees (e.g. ordered delivery
  assignment) from the old stream to the new one. If not provided, a
  random ID is generated once at pipeline startup. This is sufficient for
  most use cases: the ID is stable across gRPC reconnections within the
  same process lifetime, which is all the server requires. You only need
  to set this explicitly if you want a human-readable value for debugging.

* `:ack_batch_interval_ms` - Interval in milliseconds at which batched ack and modifyAckDeadline
  requests are flushed to the Pub/Sub service via unary RPCs.
  Lower values reduce end-to-end ack latency; higher values improve
  batching efficiency. The default value is `100`.

* `:ack_batch_max_size` - Maximum number of ack_ids to accumulate before triggering an
  immediate flush, regardless of the timer. Each unary RPC carries
  at most 2,500 ack_ids (the Google API limit), so values above 2,500
  result in multiple RPCs per flush. The default value is `2500`.

* `:retry_deadline_ms` (`t:pos_integer/0`) - Maximum total time in milliseconds to keep retrying a failed acknowledge or
  modifyAckDeadline request before giving up and dropping the ack_ids.

      The default of 60,000ms (60 seconds) applies to standard delivery subscriptions.
      When exactly-once delivery is detected from subscription properties, the library
      automatically switches to 600,000ms (600 seconds) for exactly-once acks. The
      configured value is restored if exactly-once delivery is later disabled on the
      subscription.

  The default value is `60000`.

* `:backoff_type` - The backoff strategy used when reconnecting after a stream failure.

      - `:rand_exp` - Randomized exponential backoff. Adds jitter to
      prevent thundering herd after mass disconnects.
      - `:exp` - Pure exponential backoff.
      - `:rand` - Random value between `backoff_min` and `backoff_max`.
      - `:stop` - Do not reconnect. The producer will crash after one
      failure.

  The default value is `:rand_exp`.

* `:backoff_min` (`t:pos_integer/0`) - Minimum reconnection backoff in milliseconds. The default value is `100`.

* `:backoff_max` (`t:pos_integer/0`) - Maximum reconnection backoff in milliseconds. The default value is `60000`.

* `:drain_timeout_ms` (`t:pos_integer/0`) - Maximum time in milliseconds to wait for in-flight messages to be
  processed and acknowledged during graceful shutdown. After this timeout,
  any remaining outstanding messages are nacked (per the `on_shutdown`
  setting) and the connection is force-closed.

      This drain phase waits for all outstanding messages to be acked before
      calling `CloseSend` on the stream.

  The default value is `30000`.

* `:adapter` - The gRPC HTTP/2 adapter to use for the streaming connection.

      - `:gun` - Uses the Gun HTTP/2 client. Well-tested and the
      traditional adapter for the Elixir gRPC library.
      - `:mint` - Uses the Mint HTTP/2 client. May be preferable where
      Gun is not available.
      - Any module implementing the `GRPC.Client.Adapter` behaviour.

      Both built-in adapters are provided by the `grpc` dependency. The
      adapter choice does not affect the public API or message semantics.

  The default value is `:gun`.

* `:grpc_endpoint` - The gRPC endpoint for the Cloud Pub/Sub service. Useful for testing
  with the Pub/Sub emulator (e.g., `"localhost:8085"`). The default value is `"pubsub.googleapis.com:443"`.

* `:use_ssl` (`t:boolean/0`) - Whether to use TLS when connecting to the gRPC endpoint. Set to `false`
  when connecting to the Pub/Sub emulator, which does not use TLS. The default value is `true`.

* `:keepalive_interval_ms` (`t:pos_integer/0`) - Interval in milliseconds at which HTTP/2 PING frames are sent on the gRPC
  connection to keep it alive. This prevents Google Cloud's load balancer
  from closing idle connections (which it does after roughly 20 seconds by
  default). Only applies to the `:gun` adapter. The default value is `30000`.

* `:interceptors` - A list of client-side gRPC interceptors attached to every channel opened
  by the producer (both the StreamingPull channel and the unary ack/modack channel).

      Each entry is either a bare module (e.g. `MyInterceptor`, which calls
      `MyInterceptor.init([])`) or a `{module, opts}` tuple (e.g.
      `{MyInterceptor, level: :debug}`, which calls
      `MyInterceptor.init(level: :debug)`).

      Modules must implement the `GRPC.Client.Interceptor` behaviour (`init/1` and `call/4`).

      ## Example

          interceptors: [GRPC.Client.Interceptors.Logger]
          interceptors: [{GRPC.Client.Interceptors.Logger, level: :warning}]

  The default value is `[]`.

* `:grpc_client` - The module implementing the `BroadwayCloudPubSub.Streaming.Client` behaviour.
  The built-in `BroadwayCloudPubSub.Streaming.GrpcClient` uses the `grpc`
  library to communicate with Google Cloud Pub/Sub.

      Accepts either a bare module or a `{module, opts}` tuple. When a tuple is
      given, `opts` are merged into the producer options and passed to
      `c:BroadwayCloudPubSub.Streaming.Client.init/1`:

          grpc_client: {MyGrpcClient, channel_opts: [transport_opts: []]}

      Swap this for testing or custom gRPC transports.

  The default value is `BroadwayCloudPubSub.Streaming.GrpcClient`.

* `:telemetry_metadata` - Extra data to attach to every telemetry event emitted by the streaming
  producer. The value is included in the event metadata under the `:extra`
  key.

      Accepts either a static term (e.g. a map or keyword list), which is
      stored once and included verbatim in every event, or an
      `{module, function, args}` tuple that is called on every event emission
      and whose return value is used as the `:extra` value (useful for
      attaching dynamic data such as node names or runtime counters).

      When not set, no `:extra` key is added to event metadata.

## Acknowledgements

Use `:on_success` and `:on_failure` to control how messages are acknowledged
with Pub/Sub. Both can also be changed per-message via
`Broadway.Message.configure_ack/2`.

Supported values:

  * `:ack` - acknowledge the message; Pub/Sub removes it from the subscription.
  * `:noop` - do nothing; the message is redelivered after the subscription's
    `ackDeadlineSeconds` expires.
  * `:nack` - equivalent to `{:nack, 0}`; makes the message immediately
    available for redelivery.
  * `{:nack, seconds}` - sets `ackDeadlineSeconds` to `seconds` for the
    message, controlling when it becomes available for redelivery (0-600).

Acks and deadline modifications are batched by **AckBatcher** and flushed to
Pub/Sub via unary RPCs at a configurable interval (`:ack_batch_interval_ms`,
default 100ms) or when the batch reaches `:ack_batch_max_size` (default 2500).
Batching is done on a separate unary connection, independently of the
streaming connection. See [Telemetry](#module-telemetry) for ack-related events.

## Flow control

Flow control is managed at the gRPC stream level via `:max_outstanding_messages`
and `:max_outstanding_bytes`. The Pub/Sub server will not push more messages
than these limits allow. This is the primary backpressure mechanism.

StreamManager also tracks GenStage demand from the Producer and buffers
messages internally when demand is zero, preventing unbounded mailbox growth.

See also [Lease management](#module-lease-management) for how message deadlines
are extended while flow control holds messages in the buffer.

## Lease management

The producer automatically extends message acknowledgement deadlines before
they expire. Leases are extended by sending `modifyAckDeadline` requests via
the AckBatcher. The extension interval is derived from the effective ack
deadline with randomized jitter to spread out RPC calls.

Messages are tracked until they are acknowledged, nacked, or until
`:max_extension_ms` elapses (default 60 minutes), after which the server
redelivers them. This prevents a stuck consumer from holding messages
indefinitely. The `extend_leases` and `lease_expired` telemetry events
(see [Telemetry](#module-telemetry)) provide visibility into lease activity.

## Exactly-once delivery

When the subscription has exactly-once delivery enabled, the server signals
this via `StreamingPullResponse.subscription_properties`. The producer
detects this automatically and enforces a minimum lease extension interval
(the server requires at least 60 seconds between extensions for exactly-once
subscriptions).

For exactly-once subscriptions, increase `:retry_deadline_ms` to 600,000ms
(10 minutes) to allow the unary RPC client enough time to retry transient
ack failures - the server requires successful ack receipt before guaranteeing
exactly-once semantics. The library automatically adjusts `:retry_deadline_ms`
when the subscription's exactly-once status changes at runtime.

## Message ordering

Set `enable_message_ordering: true` to route messages with the same
`ordering_key` to the same Broadway processor, ensuring sequential processing
per key. The subscription must also have message ordering enabled in Pub/Sub.

Ordering is enforced via Broadway's `:partition_by` option, which is
automatically injected into all processor groups when this option is set.

## Graceful shutdown

On shutdown, the producer:

  1. Nacks all buffered messages (received but not yet dispatched to
     processors) per the `:on_shutdown` option (default `{:nack, 5}`).
  2. Stops the gRPC stream to prevent new messages from arriving.
  3. Waits up to `:drain_timeout_ms` (default 30s) for in-flight messages
     (dispatched to processors but not yet acked/nacked) to be processed.
  4. Force-closes the stream after the drain timeout.

The drain lifecycle is tracked via the `drain` telemetry span
(see [Telemetry](#module-telemetry)).

## Error handling

gRPC stream errors are classified as retryable or terminal:

  * **Retryable** (e.g. `DEADLINE_EXCEEDED`, `UNAVAILABLE`, `UNAUTHENTICATED`) -
    the stream is closed and reconnected after a backoff delay. A new OAuth2
    token is fetched on each reconnect.
  * **Terminal** (e.g. `NOT_FOUND`, `PERMISSION_DENIED`, `INVALID_ARGUMENT`) -
    the StreamManager stops and Broadway's supervision restarts the pipeline.

Reconnect backoff is configurable via `:backoff_type`, `:backoff_min`, and
`:backoff_max`. The default is randomized exponential (`:rand_exp`) starting
at 100ms and capped at 60s.

## Pub/Sub Emulator

To use with the local Pub/Sub emulator:

    {BroadwayCloudPubSub.Producer,
     subscription: "projects/my-project/subscriptions/my-subscription",
     grpc_endpoint: "localhost:8085",
     use_ssl: false,
     token_generator: {MyApp, :emulator_token, []}}

## Differences from `BroadwayCloudPubSub.Pull.Producer`

  * **Push-based**: Messages arrive via a persistent gRPC stream rather than
    being fetched on demand via HTTP pull requests.
  * **Flow control**: Controlled at the gRPC stream level via
    `:max_outstanding_messages` and `:max_outstanding_bytes` rather than
    per-request polling. See [Flow control](#module-flow-control).
  * **Graceful shutdown**: The stream is closed immediately on shutdown to
    stop new messages arriving; the unary channel stays up so in-flight
    messages can still be acked or nacked during the drain window. The pull
    producer has no drain phase. See [Graceful shutdown](#module-graceful-shutdown).
  * **Lease extension**: Message deadlines are extended automatically to
    prevent redelivery while processing. The pull producer relies on the
    subscription-level ack deadline only. See
    [Lease management](#module-lease-management).
  * **Enhanced telemetry**: Emits a richer set of events covering connection
    lifecycle, lease activity, ack/modack RPC spans, drain lifecycle, and
    per-cycle pressure snapshots. See [Telemetry](#module-telemetry).

## Telemetry

This producer emits the following [Telemetry](https://github.com/beam-telemetry/telemetry)
events. All events share the top-level prefix `[:broadway_cloud_pub_sub, :streaming]`,
followed by a layer sub-prefix.

All event metadata maps include an `:extra` key when the `:telemetry_metadata` option
is configured. Its value is the static term provided, or the return value of the MFA
called at emission time.

### Stream events - `[:broadway_cloud_pub_sub, :streaming, :stream, ...]`

Emitted by `StreamManager`. Metadata: `%{name: broadway_name, subscription: subscription}`
(plus `:extra` when `:telemetry_metadata` is set).

#### Backpressure

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :pressure_snapshot]` -
    point-in-time snapshot of pipeline backpressure, emitted on every lease
    extension cycle. Useful for diagnosing throughput bottlenecks without
    enabling tracing.

    Measurements: `%{outstanding_count: non_neg_integer(), buffered_count: non_neg_integer(), pending_demand: non_neg_integer()}`

    * `outstanding_count` - messages received but not yet acked or nacked.
    * `buffered_count` - messages waiting in the internal buffer for producer demand.
    * `pending_demand` - units of GenStage demand currently unfulfilled.

#### Connection lifecycle

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :connect]` - gRPC
    StreamingPull stream successfully established.

    Measurements: `%{}`

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :disconnect]` - gRPC
    stream closed or errored.

    Measurements: `%{}`

    Metadata includes: `reason: term()` - the error or close reason
    (e.g. a `GRPC.RPCError`, `:stream_closed`, `:connection_down`).

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :connection_failure]` -
    connection attempt failed before the stream was established.

    Measurements: `%{}`

    Metadata includes: `reason: term()` - the connection error.

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :reconnect]` - reconnect
    scheduled after a disconnect or connection failure. The backoff delay
    indicates how long the StreamManager will wait before the next connection
    attempt.

    Measurements: `%{delay: pos_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :terminal_error]` -
    non-retryable gRPC error received. StreamManager stops after this event.

    Measurements: `%{}`

    Metadata includes: `reason: term()` - the terminal gRPC error.

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :keepalive]` - keep-alive
    ping sent on the gRPC connection.

    Measurements: `%{deadline: pos_integer()}`

#### Messages

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :receive_messages]` -
    messages received from the stream and forwarded to the producer.

    Measurements: `%{count: pos_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :ack]` - acknowledge
    request dispatched to AckBatcher.

    Measurements: `%{count: pos_integer()}`

#### Lease management

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :extend_leases]` - lease
    extension cycle ran; modack requests dispatched for outstanding messages.

    Measurements: `%{count: non_neg_integer(), deadline: pos_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :lease_expired]` -
    outstanding messages dropped because they exceeded `:max_extension_ms`.

    Measurements: `%{count: pos_integer()}`

#### Exactly-once delivery

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :receipt_modack_stale]` -
    pending receipt modack entries that exceeded the 60-second staleness
    threshold were nacked for fast redelivery. Emitted during the lease
    extension cycle.

    Measurements: `%{count: pos_integer()}`

#### Graceful shutdown

  * `[:broadway_cloud_pub_sub, :streaming, :stream, :drain, :start | :stop | :exception]` -
    span tracking the full graceful drain lifecycle, from
    `prepare_for_draining/1` through completion, timeout, or unexpected
    termination. Uses the same convention as `:telemetry.span/3`.

    * `[:broadway_cloud_pub_sub, :streaming, :stream, :drain, :start]` - drain
      initiated. Emitted before the stream is closed or any messages are nacked.

      Measurements: `%{system_time: integer(), monotonic_time: integer(),
      buffered_count: non_neg_integer(), outstanding_count: non_neg_integer(),
      pending_receipt_modack_count: non_neg_integer()}`

    * `[:broadway_cloud_pub_sub, :streaming, :stream, :drain, :stop]` - all
      in-flight messages were processed and stream closed cleanly.

      Measurements: `%{duration: non_neg_integer(), monotonic_time: integer()}`

    * `[:broadway_cloud_pub_sub, :streaming, :stream, :drain, :exception]` -
      drain ended abnormally.

      Measurements: `%{duration: non_neg_integer(), monotonic_time: integer()}`
      (plus `remaining_count: non_neg_integer()` for `:timeout` and `:terminate` kinds)

      Metadata includes `kind` and `reason` identifying the cause:

      * `kind: :timeout, reason: :drain_timeout` - `drain_timeout_ms` elapsed
        before all messages were acked. Remaining messages are nacked immediately.
      * `kind: :terminate, reason: term()` - the GenServer was terminated while
        a drain was in progress.
      * `kind: :error, reason: binary()` - an exception was raised inside
        `prepare_for_draining/1` itself.

### AckBatcher events - `[:broadway_cloud_pub_sub, :streaming, :ack_batcher, ...]`

Emitted by `AckBatcher`. Metadata: `%{name: broadway_name, subscription: subscription}`
(plus `:extra` when `:telemetry_metadata` is set).

  * `[:broadway_cloud_pub_sub, :streaming, :ack_batcher, :flush_deferred]` -
    flush deferred because UnaryRpcClient was not yet available (e.g.
    restarting after a crash).

    Measurements: `%{ack_count: non_neg_integer(), modack_groups: non_neg_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :ack_batcher, :modack_retry_exhausted]` -
    modack ack_ids dropped after reaching the maximum retry attempt count.

    Measurements: `%{count: pos_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :ack_batcher, :ack_retry_expired]` -
    ack ack_ids dropped because they exceeded the exactly-once retry deadline.

    Measurements: `%{count: pos_integer()}`

  * `[:broadway_cloud_pub_sub, :streaming, :ack_batcher, :modack_retry_expired]` -
    modack ack_ids dropped because they exceeded the exactly-once retry deadline.

    Measurements: `%{count: pos_integer()}`

### Unary RPC client events - `[:broadway_cloud_pub_sub, :streaming, :unary, ...]`

Emitted by `UnaryRpcClient`. Metadata: `%{name: broadway_name, subscription: subscription}`
(plus `:extra` when `:telemetry_metadata` is set).

  * `[:broadway_cloud_pub_sub, :streaming, :unary, :connect]` - unary RPC
    channel reconnected after a failure.

    Measurements: `%{}`

  * `[:broadway_cloud_pub_sub, :streaming, :unary, :connection_failure]` -
    unary RPC channel connect attempt failed.

    Measurements: `%{}`

    Metadata includes: `reason: term()` - the connection error.

  * `[:broadway_cloud_pub_sub, :streaming, :unary, :permanent_failure]` -
    one or more ack_ids were permanently rejected by the server (e.g. ack_id
    expired). These are dropped and not retried.

    Measurements: `%{count: pos_integer()}`

### gRPC client spans - `[:broadway_cloud_pub_sub, :streaming, :grpc_client, ...]`

Emitted by `GrpcClient` (the default `BroadwayCloudPubSub.Streaming.Client`
implementation) as `:telemetry.span/3` spans.
Metadata: `%{name: broadway_name, subscription: subscription, count: ack_count}`
(plus `:extra` when `:telemetry_metadata` is set).

  * `[:broadway_cloud_pub_sub, :streaming, :grpc_client, :ack, :start | :stop | :exception]` -
    wraps each `Acknowledge` unary RPC call.

  * `[:broadway_cloud_pub_sub, :streaming, :grpc_client, :modack, :start | :stop | :exception]` -
    wraps each `ModifyAckDeadline` unary RPC call.

# `partition_by`

---

*Consult [api-reference.md](api-reference.md) for complete listing*