Add GRPC Streaming producer

[rockneurotiko]

Hey!

Sorry for the big PR, I talked with @josevalim and @wojtekmach at the Elixir Conf EU about opening this PR or creating an off_broadway_cloud_pub_sub library, and they agreed that it's worth opening the PR and see where it leads.

In my team we use google pubsub a lot, to the point that the pull method is not enough, specifically for these reasons:

• A lot of open connections to google, it makes the observability hard to know what's happening and it generate a lot of traffic (HTTP and SSL data transfer overhead)

• There is no way to extend the lease while processing messages, this makes that if you have a slow (or sometimes slow) subscriber, you have to set a high deadline for topic

• On shutdown, the http call is just killed (because it can't wait the long-poll to finish in 30 seconds), and google don't handle it properly (not in HTTP 1.1 nor HTTP 2), and if there were messages that were going to be sent to that http call, pubsub thinks it was sent, making it go to the deadline and fail (and retry or not, depends on the setting). This together with the high deadline in the point before makes that on deployments/restarts, some messages get delayed by the full long deadline.

Since this is the official broadway library for pubsub, I think it should support grpc streaming. All official google's pubsub libraries (I think all except PHP) use GRPC streaming as default, they usually provide http client code to do pull requests manually, but the high level abstractions don't have support for it (except Go). In the docs they recommend using it if possible and that http don't guarantee low latency or high throughput).

This PR adds GRPC Streaming producer, it's totally optional to use it, and the existing pull producer is not changed, so the library users don't have breaking changes and they have to opt-in for the streaming producer instead of the pull one.

This is all the features that the streaming producer has, to have feature-parity with the official libraries:

• Dual GRPC connections, one for reading the messages (the bidi StreamingPull), and another for sending acks, modAcks and lease extensions (it's called unary connection). Both reconnect independently, so acks can keep flowing while the stream is reconnecting, and stream traffic doesn't head-of-line-block the ack traffic.

• Server-side flow control via max_outstanding_messages and max_outstanding_bytes (the latter defaults to 100 MiB). The server stops pushing when either limit is hit, so memory pressure is bounded regardless of how fast the downstream processors are.

• Backpressure integrated with Broadway/GenStage demand. Internally the producer tracks "outstanding" messages (dispatched, waiting for ack) and "buffered" messages (received from the stream, waiting for downstream demand). Buffered messages survive reconnects, and pending_demand is preserved across reconnects too: GenStage only emits new demand when capacity opens up, so resetting it on reconnect would deadlock the pipeline.

• Automatic lease extension while messages are being processed, with adaptive p99-based ack deadlines (so a slow subscriber doesn't keep getting redeliveries, but a fast one doesn't waste modAck calls either). Capped by max_extension_ms (default 60 minutes) so a stuck consumer can't hold messages forever and eventually lets the server redeliver them to another subscriber.

• Reconnection handling with configurable backoff (:rand_exp, :exp, :rand, or :stop), with backoff_min/backoff_max defaulting to 100ms / 60s. Randomised exponential backoff is the default to avoid thundering herd after a mass disconnect (e.g. a load balancer drain). Outstanding and buffered messages are preserved across reconnects; the producer only shuts down after repeated non recoverable errors (tracked by a RetryTracker).

• Batched acks and modAcks via the unary connection, with configurable ack_batch_interval_ms and ack_batch_max_size (capped at the Google API limit of 2,500 ack_ids per RPC). ModAcks are grouped by deadline value because a ModifyAckDeadline RPC carries a single deadline for all of its ack_ids, so grouping minimises the number of RPCs per flush cycle.

• Exactly-once delivery support, auto-detected from SubscriptionProperties at runtime. When enabled:

  • The retry deadline for ack/modAck RPCs is automatically bumped from 60s to 600s, because with exactly-once enabled the server can return transient errors for extended periods while it guarantees exactly-once semantics, and giving up early would cause spurious redeliveries.
  • Messages are gated on a receipt modAck: they aren't dispatched to processors until the server confirms receipt, so if the receipt fails we can nack and let the server redeliver rather than risk a duplicate.
  • If the property flips back to disabled at runtime, the configured retry deadline is restored.

• Message ordering support via enable_message_ordering. Uses Broadway's built-in partition_by so messages with the same ordering_key go to the same processor and are processed sequentially. A stable client_id is propagated to all stream managers so the server can use sticky assignment, which is required for ordered subscriptions to keep delivering the same ordering key to the same subscriber across reconnects.

• Graceful shutdown with a configurable drain_timeout_ms (default 30s): stop pulling new messages, wait for outstanding to be acked, then nack the remainder per the on_shutdown setting (:noop, :nack, or {:nack, seconds}, defaulting to {:nack, 5} to avoid thundering herd on rolling deploys), then close the stream cleanly with CloseSend. This is what makes restarts not lose latency on in-flight messages, the third bullet point at the top of this PR.

• Configurable ack behaviour per message outcome via on_success / on_failure (:ack, :noop, or {:nack, seconds}), keeping the same shape as the existing pull producer's API so users switching producers don't have to relearn it.

• N:N producer-manager topology: each Broadway producer has its own StreamManager (owning one StreamingPull stream) and shares a single UnaryAckSupervisor per producer, supervising AckBatcher and UnaryRpcClient. This scales horizontally: more producers means more streams to Google, each with their own flow control budget, which is the only way to get past the per-stream throughput ceiling.

• Pluggable HTTP/2 adapter (:gun default, :mint, or any module implementing GRPC.Client.Adapter). Both built-in adapters are provided by the grpc library. HTTP/2 PING keepalives are sent every 30s on the :gun adapter to keep Google's load balancer from closing idle connections, which it does after roughly 20s by default.

• Pluggable gRPC client behaviour (BroadwayCloudPubSub.Streaming.Client): the default GrpcClient wraps the grpc library, but it can be swapped out for tests or alternative transports without touching the producer.

• gRPC client-side interceptors (interceptors option) attached to both channels, useful for logging, tracing, custom auth, etc. Standard GRPC.Client.Interceptor behaviour.

• Pluggable auth via goth / token_generator (same MFArgs API as the existing pull producer), with the token re-fetched on every (re)connection so short-lived tokens don't expire on a long-lived stream.

• Pub/Sub emulator support via grpc_endpoint and use_ssl (set to "localhost:8085" and false respectively to point at the local emulator, same setup we use for the integration tests).

• Telemetry events that allow for great observability: stream lifecycle (connect, disconnect, reconnect, drain), ack batch flush timings and sizes, unary RPC spans, lease extension cycles, outstanding/buffered message gauges, pending demand and drain status. The telemetry_metadata option lets users attach static or dynamic (MFA) extra metadata to every event, so the events plug straight into Datadog / Prometheus / etc. without per-call boilerplate.

• Minimal generated protobuf surface: the proto generation is filtered down to the three RPCs we actually use (StreamingPull, Acknowledge, ModifyAckDeadline), going from 124 generated modules / 2647 lines down to 12 modules / 265 lines. The full Subscriber service has 17 RPCs plus Publisher and SchemaService, and pulling all of them in would bloat compile times and conflict with the googleapis Hex dep's google.api.* annotation modules. Reproducible via buf generate.

I know that it's a big PR, but I would love to start the conversation and the review. If agreed to have it in the library, I would wait until grpc 1.0 is released, but the PR can be reviewed, it uses the release candidate at the moment.

I've been working on this for more a couple of months and we've been running it in production for some weeks now, I can share in private datadog screenshots showing how it performs and the potential of the telemetry events. I also have integration and stress tests that require the emulator (or real pubsub), I haven't included them on the PR but I can share them.

On top of the PR itself, I have a few internal documents that didn't seem worth pushing to the branch but I'm happy to share if useful, they capture the non obvious architectural rationale (Mint stream cancellation quirks, reconnect deduplication, exactly once receipt gating, why we don't dedupe ack_ids intra stream, etc.), a comparison of the official Go / Java / Python / Node.js clients that informed a lot of the defaults, and an architecture write up of how the supervision tree, stream manager and ack pipeline fit together. Just let me know and I'll send them over, or inline whatever bits would help the review.

[josevalim]

Hi @rockneurotiko! Thank you for the pull request!

Would you like to become a maintainer of this project, given you are already maintaining a fork anyway? One option is to release the current version as v1.0 and then you can start the GRPC as v2.0, maintained by you/your team.

[rockneurotiko]

@josevalim Sure! I'm more than happy to become a mantainer of the library.

The versioning idea sounds great, even the 1.0 could be with both since the existing producer is not changed at all, but the double version is a safe approach that makes sense

[josevalim]

@rockneurotiko I have invited you to the new organization. I will branch and ship v1.0 and then you can proceed with v2.0. It is up to you if you want to keep maintaining the current interface!

[rockneurotiko]

@rockneurotiko I have invited you to the new organization. I will branch and ship v1.0 and then you can proceed with v2.0. It is up to you if you want to keep maintaining the current interface!

@josevalim Great, I'll pick it from there. It seems that the CI didn't work, today github had issues with the actions, maybe it's worth retrying them.