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 (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 (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 (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 (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 (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 (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 (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 (pos_integer/0) - Minimum reconnection backoff in milliseconds. The default value is 100.

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

• :drain_timeout_ms (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 (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 (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 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 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) 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).

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

Telemetry

This producer emits the following 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.