google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494

# google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494

## What's going on

PR #1494 added google-application-default credentials and StreamingPull auto-reconnect to the connector. After running the resulting subscriber under realistic load we hit three structural correctness gaps:

1. **(introduced in #1494)** `autoExtendAckDeadlines` doesn't actually cover messages buffered behind a slow downstream stage, so Pub/Sub redelivers them.
2. **(pre-existing since 2019, newly observable)** `GooglePubSub.subscribe(...)` propagates `maxOutstandingMessages`, `maxOutstandingBytes`, and `clientId` from the initial `StreamingPullRequest` into every keepalive request, which the server rejects with `INVALID_ARGUMENT`. The leak has been in `subscribe` since commit `3f1a4f7f28` (Jan 2019, akka.stream.alpakka era), but no caller had reason to set those fields until #1494's `RestartSettings` overload made long-lived streaming-pull subscribers viable. It's essentially a latent bug surfaced by the new use cases #1494 enables.
3. **(introduced in #1494)** `flowControlGate` is structurally identical to bug 1: it counts permits only when it pushes downstream, so its limit can't bound what the server delivers.

All three are reproducible with stub `SubscriberClient`s, no live Pub/Sub needed. See the tests at the bottom.

---

## Bug 1: `autoExtendAckDeadlines` doesn't cover messages waiting in the buffer

### What we expected

When you wrap a Pub/Sub source with `autoExtendAckDeadlines`, you'd expect every message that has reached your client to get its deadline extended in the background. That way Pub/Sub won't redeliver while you're still working on it.

### What actually happens

Tracking only starts when a message reaches the inside of the operator. The current code uses a plain `.map`:

```scala
Flow[ReceivedMessage]
  .via(killSwitch.flow)
  .map { msg =>
    tracked.put(msg.ackId, System.nanoTime())
    msg
  }
```

A `.map` only runs its function when downstream asks for the next element. So in a normal pipeline like this:

```scala
GooglePubSub.subscribe(request, 1.second)
  .via(GooglePubSub.autoExtendAckDeadlines(sub, 8.seconds, 30))
  .mapAsync(parallelism = 10)(processMessage)   // ~5 to 7 seconds per message
```

if all 10 `mapAsync` slots are busy, nothing flows through `.map`. Meanwhile Pub/Sub has already pushed dozens more messages over the gRPC stream. Those messages sit in the gRPC adapter's buffer with their server-side deadline ticking, but our code never tracks them. The background ticker fires every 8 seconds, looks at `tracked`, sees only the 10 messages currently held by `mapAsync`, and extends just those. The rest hit 60 seconds, expire, get redelivered. We end up processing duplicates.

### Why "just put a `.buffer` in front of it" doesn't help

A `.buffer` upstream of `autoExtendAckDeadlines` lets messages queue up, but `.map` still won't fire on them until downstream pulls. A `.buffer` downstream pushes the bottleneck back by N elements, but once that buffer fills, the bug returns.

### Why eager-pull is the architecturally right answer

#### What goes wrong

Pub/Sub starts a per-message ack deadline timer at server dispatch. The client must declare possession of each message (track its ackId) before that deadline expires, or the server redelivers. The current `.map`-based tracker only declares possession when downstream pulls, and downstream pulls are paced by the slowest stage in the pipeline (typically `mapAsync(parallelism)`).

When that pace is slower than the deadline, messages get redelivered. The threshold:

```
(buffered_count × per_message_seconds / parallelism) > deadline_seconds
```

Concretely: 60 messages buffered, 6s per message, `mapAsync(10)`, 60s deadline → messages at the back of the queue wait ~36s for tracking, some slip past 60s, get redelivered.

Where that buffering lives matters for the fix:

- **`mapConcat(_.receivedMessages.toVector)`** holds each `StreamingPullResponse`'s batch (typically 50 to 100 messages) and emits them one per downstream pull.
- **gRPC-Java + Netty** hold messages below the Pekko stream layer entirely: HTTP/2 receive window (~1 MB default), `MessageDeframer`'s assembled-message queue. Pekko gRPC's own `GraphStage` is demand-driven (it asks gRPC for one message per `onPull`), so there's no Pekko-layer buffer to shrink. The buffer that lets the server outpace the consumer sits *underneath* Pekko Streams, where user code can't reach it.

#### Why this is the deep cause

The bug exists because two distinct timescales get conflated:

- **Tracking** ("I have this message"): must complete in microseconds, because it has to beat the ack deadline (tens of seconds, fixed by the server).
- **Processing** ("I am done with this message"): can take whatever it takes; the deadline is extendable via `ModifyAckDeadline` for as long as the user wants.

A demand-driven `.map` puts both on the same clock: tracking only fires when downstream pulls, which is paced by processing. *That conflation is the bug.* The framework's default pull semantics happen to be incompatible with a protocol that has a server-controlled, deadlined unacked window. Same shape recurs in any at-least-once protocol with these properties (Kafka long-poll with manual commit, SQS visibility timeout, RabbitMQ prefetch). gRPC is incidental.

#### Why eager-pull is uniquely the right shape

Eager-pull splits the two clocks. A `GraphStage` that pulls from upstream into a bounded internal buffer whenever it has room runs the tracking work in `onPush` *on receipt*, then enqueues for downstream. Tracking now happens at adapter-delivery speed (microseconds); processing still flows through the buffer at its own pace.

What this buys that nothing else does:

- **Tracking decouples from processing.** Slow downstream no longer starves fast-required tracking.
- **Backpressure reaches the right buffer.** When the eager-pull buffer fills, the stage stops pulling. That signal back-propagates through Pekko gRPC's GraphStage (stops calling `call.request(1)`), through gRPC-Java (no demand), into Netty (HTTP/2 receive window stops refreshing), and finally to the server (stops dispatching). Eager-pull is the only Pekko-layer construct that can tell the server "stop" without also breaking tracking. `.buffer(N)`, `.async`, larger `parallelism`, or a custom `Sink` for tracking either fail to apply backpressure all the way down or break tracking, downstream flow, or the abstraction.
- **Bug 2 complements this at the protocol layer.** Server-side `maxOutstandingMessages` bounds the lower buffers via the protocol; eager-pull bounds the stream-layer buffer via backpressure. Together: the server has at most N messages in flight to the client, and the client tracks every one of them on receipt.

Google's Java client uses the same shape (`MessageDispatcher.processReceivedMessages` registers in `pendingMessages` before the user callback) for the same reason. Independent corroboration of the constraint; the constraint itself comes from the protocol.

### Suggested fix

Replace the `.map` with a small `GraphStage` that pulls from upstream eagerly into a bounded buffer, calling the tracking callback the moment it grabs each message. Google's official Java client (`google-cloud-pubsub`) does the same thing in `MessageDispatcher.processReceivedMessages`: it registers messages in `pendingMessages` before handing them to user code.

Sketch:

```scala
private[grpc] final class EagerPullTrackingStage[T](maxBuffer: Int, onTrack: T => Unit)
    extends GraphStage[FlowShape[T, T]] {
  // preStart: pull(in)
  // onPush:   onTrack(msg); buffer.offer(msg); pushIfPossible(); pullIfPossible()
  // onPull:   pushIfPossible(); pullIfPossible()
  // pullIfPossible: only pulls if buffer has room
}
```

The buffer needs an upper bound or it will OOM. With a sensible default (1000, matching Google's client) it's fine for almost everyone. New overloads can expose `maxBuffer` for tuning. The cleanest pairing is with server-side flow control via `maxOutstandingMessages`, which is what bug 2 is about.

---

## Bug 2: `subscribe()` makes `maxOutstandingMessages` unusable

### What we expected

Pub/Sub's streaming pull protocol lets you cap delivery server-side via `maxOutstandingMessages` and `maxOutstandingBytes` on the initial `StreamingPullRequest`. We expected to set those on the request we pass to `GooglePubSub.subscribe()` and have everything just work.

### What actually happens

`subscribe()` builds the keepalive request like this:

```scala
val subsequentRequest = request
  .withSubscription("")
  .withStreamAckDeadlineSeconds(0)
```

It clears two fields, but the Pub/Sub API has five fields that must only appear on the initial request:

| Field                       | Cleared currently? |
|-----------------------------|--------------------|
| `subscription`              | yes                |
| `streamAckDeadlineSeconds`  | yes                |
| `clientId`                  | no                 |
| `maxOutstandingMessages`    | no                 |
| `maxOutstandingBytes`       | no                 |

The keepalive tick re-sends our original request with three of those fields still set. About a second after the stream starts, the server replies:

```
INVALID_ARGUMENT: You must only set max_outstanding_messages and
max_outstanding_bytes on the initial request.
```

The whole stream fails. There's no way to recover from inside the public API. The only workaround is to bypass `subscribe()` entirely and call `subscriberClient.streamingPull` directly.

### Suggested fix

Build the subsequent request from the proto's default instance:

```scala
val subsequentRequest = StreamingPullRequest.defaultInstance
```

That clears every initial-only field at once. It's also safe against future proto revisions: if Google adds another initial-only field next year, this code won't accidentally leak it.

The keepalive tick is purely a heartbeat in our implementation. It doesn't carry anything that needs to come from the original request.

---

## Bug 3: `flowControlGate` doesn't actually flow-control

### What we expected

`flowControlGate(FlowControl(maxOutstandingMessages = 100))` looks like the protocol-level cap that Google's `FlowController` provides. We expected it to bound how many unacked messages the system holds at any given time, including ones still in flight from the server.

### What actually happens

It only counts messages that have already been pushed past the gate to downstream. Looking at `impl/FlowControlGateStage.scala`:

```scala
override def onPull(): Unit = {
  if (flowControl.outstanding.get() < flowControl.maxOutstandingMessages) {
    pull(in)
  } else {
    downstreamWaiting = true
  }
}

override def onPush(): Unit = {
  val msg = grab(in)
  flowControl.acquire()
  push(out, msg)
}
```

Same shape as the old `.map` tracker from bug 1: a 1-in-1-out passthrough whose `onPull` only pulls upstream when downstream asks. So when `mapAsync` saturates, the gate stops pulling, but messages keep arriving from the server into the gRPC adapter buffer with no permit cost. The `outstanding` counter stays low even when the system is holding hundreds of unacked messages.

In practice this means three things go wrong:

1. **It doesn't mitigate bug 1.** Wrapping the pipeline with `flowControlGate(FlowControl(100))` doesn't reduce the redelivery rate, because the gate isn't actually bounding what the server sends.
2. **`outstandingCount` underreports.** Telemetry built on this counter shows you a tiny fraction of what's really in flight.
3. **The name is misleading.** Users reach for "flow control" expecting Google's semantics and get downstream credit accounting instead.

### Suggested fix

Rewrite `FlowControlGateStage` with the same eager-pull pattern as bug 1: acquire the permit on receipt (in `onPush`), buffer the message internally, push to downstream when demand arrives. The buffer is naturally bounded by the FlowControl limit, so no separate `maxBuffer` parameter is needed.

```scala
override def onPush(): Unit = {
  val msg = grab(in)
  flowControl.acquire()       // count on receipt, not on push downstream
  buffer.offer(msg)
  pushIfPossible()
  pullIfPossible()
}

override def onPull(): Unit = {
  pushIfPossible()
  pullIfPossible()
}

private def pullIfPossible(): Unit =
  if (flowControl.outstanding.get() < flowControl.maxOutstandingMessages
      && !hasBeenPulled(in) && !isClosed(in)) pull(in)
  else if (buffer.isEmpty && isClosed(in)) completeStage()
```

After this change, `outstandingCount` reflects messages that have been received from the server (whether or not they have flowed past the gate yet), the gate stops pulling once the limit is hit, and combined with gRPC backpressure the server eventually stops sending too. That's what most users reaching for "flow control" actually want.

### Note on the relationship with bug 2

After bug 2 lands, `StreamingPullRequest.maxOutstandingMessages` gives you protocol-level flow control on the wire, with the server respecting the cap directly. That is strictly better than `flowControlGate` for the common case of "don't let the server send me more than N at a time." `flowControlGate` then survives as an application-level primitive for cases that genuinely need it, like per-tenant fairness across multiple subscribers sharing one Pub/Sub subscription, where server-side flow control can't tell callers apart.

---

## Why these three bugs go together

Bug 1 and bug 3 are the same defect in two different operators: a 1-in-1-out passthrough whose `onPull` only pulls upstream when downstream demands. Both need the same eager-pull fix. Bug 2 is what unlocks the natural way to bound the eager-pull buffer for either: server-side `StreamingPullRequest.maxOutstandingMessages`, which makes the gRPC adapter stop pulling at the protocol level once the cap is reached.

Without bug 2 fixed, you can't use the server-side cap, so the eager-pull buffers in bugs 1 and 3 have to be sized by hand. Without bug 1 or bug 3 fixed, even a correctly capped server-side flow doesn't help, because messages still pile up past the demand-bound stages the moment any downstream stage stalls.

Fixing all three gives you: the server stops sending past the cap (bug 2), the eager-pull tracker covers every in-flight message for deadline extension (bug 1), the eager-pull gate enforces the in-flight limit on receipt (bug 3), and the keepalive tick stays valid forever (bug 2). Each fix on its own helps a little. All three together make the subscriber actually work under load.

---

## Reproduction

All three tests use stub `SubscriberClient`s, no live Pub/Sub. They live in `src/test/scala/org/apache/pekko/stream/connectors/googlecloud/pubsub/grpc/AutoExtendAckDeadlinesSpec.scala` (despite the file name, it covers all three bugs since each is a small focused test that fits naturally beside the existing autoExtend tests).

The stubs are defined at the bottom of that same file. Each subclasses a `TestSubscriberClientBase` trait that overrides every `SubscriberClient` method with `???`, so the concrete stubs only override the one or two methods the test actually exercises:

- `SucceedingClient`: `modifyAckDeadline` returns `Future.successful(Empty())`. Used by bug 3.
- `CapturingClient`: `modifyAckDeadline` records the incoming request into a `ConcurrentLinkedQueue` then returns success. Used by bug 1 to inspect which ackIds the ticker extended.
- `CapturingStreamingPullClient`: overrides `streamingPull` to drain the input request `Source` into a queue and return a never-completing `Source.maybe[StreamingPullResponse]`, so the response stream stays open while the test inspects what was sent on the client → server side. Used by bug 2.

### Bug 1

```scala
"track all in-flight messages eagerly even when downstream is backpressured" in {
  val captured = new ConcurrentLinkedQueue[ModifyAckDeadlineRequest]()
  val testSubscriber = new GrpcSubscriber(new CapturingClient(captured))
  val gate = Promise[ReceivedMessage]()  // mapAsync hangs forever
  val killSwitch = KillSwitches.shared("test")

  Source(1 to 10).map(i => makeMsg(i.toString))
    .via(GooglePubSub.autoExtendAckDeadlines(subscription, 200.millis, 30))
    .via(killSwitch.flow)
    .mapAsync(1)(_ => gate.future)
    .withAttributes(PubSubAttributes.subscriber(testSubscriber))
    .runWith(Sink.ignore)

  Thread.sleep(800)
  killSwitch.shutdown()

  val firstReq = captured.poll()
  // Today: firstReq.ackIds.size == 1     (only the message held by mapAsync)
  // Want:  firstReq.ackIds.size == 10    (everything we received)
  firstReq.ackIds.toSet shouldBe (1 to 10).map(_.toString).toSet
}
```

### Bug 2

```scala
"send only allowed fields on subsequent StreamingPullRequest messages" in {
  val captured = new ConcurrentLinkedQueue[StreamingPullRequest]()
  val testSubscriber = new GrpcSubscriber(new CapturingStreamingPullClient(captured)(system))

  val initial = StreamingPullRequest()
    .withSubscription(subscription)
    .withStreamAckDeadlineSeconds(60)
    .withClientId("test-client")
    .withMaxOutstandingMessages(100L)
    .withMaxOutstandingBytes(10485760L)

  val cancellable = GooglePubSub.subscribe(initial, 100.millis)
    .withAttributes(PubSubAttributes.subscriber(testSubscriber))
    .toMat(Sink.ignore)(Keep.left).run()

  Thread.sleep(500); cancellable.futureValue.cancel()

  val first = captured.poll()
  first.maxOutstandingMessages shouldBe 100L  // initial request kept verbatim

  val subsequent = Iterator.continually(Option(captured.poll())).takeWhile(_.isDefined).flatten.toList
  subsequent should not be empty
  // Today this fails: every subsequent request still carries clientId,
  // maxOutstandingMessages, maxOutstandingBytes from the initial request.
  subsequent.foreach { req =>
    req.subscription shouldBe ""
    req.streamAckDeadlineSeconds shouldBe 0
    req.clientId shouldBe ""
    req.maxOutstandingMessages shouldBe 0L
    req.maxOutstandingBytes shouldBe 0L
  }
}
```

### Bug 3

```scala
"acquire permits on receipt, not on push to downstream (eager-pull)" in {
  val testSubscriber = new GrpcSubscriber(new SucceedingClient())
  val limit = 5
  val flowControl = FlowControl(maxOutstandingMessages = limit.toLong)
  val gate = Promise[ReceivedMessage]() // mapAsync(1) hangs forever
  val killSwitch = KillSwitches.shared("flowControlGateTest")

  Source(1 to 100).map(i => makeMsg(i.toString))
    .via(GooglePubSub.flowControlGate(flowControl))
    .via(killSwitch.flow)
    .mapAsync(1)(_ => gate.future)
    .withAttributes(PubSubAttributes.subscriber(testSubscriber))
    .runWith(Sink.ignore)

  Thread.sleep(300)

  // Today: outstandingCount == 1 (only the message held by mapAsync got a permit acquired).
  // Want:  outstandingCount == 5 (the gate eagerly pulls and counts up to the limit).
  flowControl.outstandingCount shouldBe limit.toLong

  killSwitch.shutdown()
}
```

---

## Recommended high-level API: `Subscriber`

Once all three bugs are fixed at the operator level, the connector still has an ergonomic problem. To get a correct subscriber today the user has to compose four operators by hand:

```scala
GooglePubSub.subscribe(request, 1.second, restartSettings)
  .via(GooglePubSub.autoExtendAckDeadlines(...))
  .via(GooglePubSub.flowControlGate(flowControl))
  .mapAsync(10)(processMessage)
  .map(...)
  .runWith(GooglePubSub.acknowledge(parallelism = 1, flowControl))
```

There are several ways to compose these wrong: putting `autoExtendAckDeadlines` inside `RestartSource` instead of the 3-arg `subscribe` (loses tracking on each reconnect), forgetting to pair `acknowledge(_, flowControl)` with `flowControlGate(flowControl)` (permits never released), passing the wrong `subscription` string to `autoExtendAckDeadlines` and the request (extension goes to a different subscription), etc. The traps are silent.

A `GooglePubSub.subscriber(...)` resource bundles the configuration once and exposes a clean `Source` and `Sink` that compose correctly by construction:

```scala
val subscriber = GooglePubSub.subscriber(
  request          = StreamingPullRequest()
                       .withSubscription(subscriptionFqrs)
                       .withStreamAckDeadlineSeconds(60)
                       .withMaxOutstandingMessages(1000),
  pollInterval     = 1.second,
  ackDeadline      = AckDeadline.Fixed(extensionInterval = 8.seconds, deadlineSeconds = 30),
  restartSettings  = Some(RestartSettings(100.millis, 10.seconds, 0.2)),
  flowControl      = Some(FlowControl(maxOutstandingMessages = 1000)))

subscriber.source
  .mapAsync(10)(processMessage)
  .map(msg => AcknowledgeRequest(subscriptionFqrs, Seq(msg.ackId)))
  .runWith(subscriber.acknowledge(parallelism = 1))
```

Or even simpler, with the `run` convenience:

```scala
subscriber.run(parallelism = 10)(processMessage)
```

What `Subscriber` guarantees:

- Restart wraps only the inner `subscribe`, not the deadline tracker, so tracking survives reconnect by construction. Users can't accidentally put `autoExtendAckDeadlines` inside their own `RestartSource` and lose state on every reconnect.
- **In-flight messages are preserved across reconnect.** `RestartSource` re-materializes only `subscribe`; the eager-pull tracker, flow-control gate, and downstream operators stay alive. Messages buffered inside those operators at the moment a gRPC stream dies stay in their buffers, keep getting their deadlines extended by the long-lived ticker, and continue flowing downstream when the new stream comes up. This is actually better than Google's `MessageDispatcher`, which tracks ackIds for extension but doesn't buffer the message payload itself; here the payload survives in the stream and processing resumes on the original instance.
- `subscriber.acknowledge(...)` releases flow-control permits if `flowControl` is set, and records completion latencies if `ackDeadline` is `Adaptive`. No risk of mis-pairing the operators.
- The deadline tracker's ticker starts at construction and stops at `close()`. The source's `watchTermination` triggers `close()` automatically on stream completion.
- Two strategies for ack deadlines via `AckDeadline.Fixed` or `AckDeadline.Adaptive`. Other config knobs stay optional.

The low-level operators stay available for users who genuinely need different composition, like sharing one extender across multiple subscriptions or splitting the source for fan-out. For the 95% case, `Subscriber` is the entry point.

`AckDeadlineExtender` survives as the underlying primitive that `Subscriber` is built on; users who want fine control over its lifecycle can keep using it directly.

A Java DSL equivalent of `Subscriber` is straightforward to add but has been left for a follow-up issue. The Java DSL splits its `GrpcSubscriber` from the Scala one (each wraps a different generated client class), so a clean Java `Subscriber` either needs a `JavaSubscriberClient`-based extender or has to bridge to the Scala client. That's a small design call worth its own discussion. In the meantime, Java users get the bug 1, 2, and 3 fixes automatically through the existing Java DSL operators, and they can use the Java factories on `AckDeadlineExtender.create(...)` for restart-safe deadline tracking.

# Visual summary: what was wrong and what this PR introduces

## The structural pattern shared by bugs 1 and 3

```
═══════════════════════════════════ BEFORE: demand-bound 1-in-1-out ═════════════════════════════════════

  upstream                             this stage                          downstream
  (gRPC adapter)                       (.map tracker / flowControlGate)    (mapAsync, etc.)
  ────────────                         ──────────────                      ──────────────────
                                                                           
                                        onPull():                          
                                          if (room) pull(in)               ◄── pull
                                                                           
                                        onPush():                          
                                          grab(in)                         
                                          do work (track / acquire)        
                                          push(out, msg)                   ──► msg
                                                                           
                                        (work runs ONLY when downstream    
                                         pulls; nothing eager.)            

  When downstream stalls (e.g. mapAsync(parallelism=10) saturated):
    1. downstream stops pulling
    2. this stage's onPull() never fires        ───►  no upstream pull
    3. messages pile up in upstream buffers     ───►  but work never runs
    4. server-side deadline ticks anyway        ───►  REDELIVERY (bug 1)
                                                ───►  outstanding undercounted (bug 3)


═══════════════════════════════════ AFTER: eager-pull with bounded buffer ═══════════════════════════════

  upstream                             this stage                          downstream
  (gRPC adapter)                       (EagerPullTrackingStage /           (mapAsync, etc.)
                                        rewritten FlowControlGateStage)
  ────────────                         ──────────────                      ──────────────────
                                                                           
                                        preStart():                        
                                          pull(in)              ◄────── starts pulling on its own
                                                                           
  push msg ─────────►  drain  ─────►   onPush():                          
                                          grab(in)                         
                                          do work IMMEDIATELY              ◄── tracked / permit acquired
                                            on receipt                         on receipt
                                          buffer.offer(msg)                
                                          pushIfPossible()      ──────►   ──► msg (if downstream demand)
                                          pullIfPossible()                 
                                                                           
                                        onPull():                          
                                          pushIfPossible()                 ──► msg (drain buffer)
                                          pullIfPossible()                 
                                                                           
                                        pullIfPossible():                  
                                          if (buffer < cap                  
                                              && upstream open) pull(in)   

  When downstream stalls:
    1. downstream stops pulling
    2. buffer fills up to its cap        ───►  backpressure flows upstream
    3. but every message we ever          ───►  tracked / permit acquired
       received has already had              correctly (no redelivery,
       its work done on receipt              outstanding reflects reality)


═══════════════════════════════════ Where the buffering actually lives ═══════════════════════════════════

  The "upstream buffers" in the diagram above aren't a single thing, and most of
  them are not in Pekko Streams at all. Concretely:

      Server
        │ HTTP/2
        ▼
   ┌──────────────────────────────────────────────────────────┐
   │ Netty + gRPC-Java                                        │  ◄── BELOW Pekko.
   │   HTTP/2 receive window (~1 MB by default)               │      Holds bytes /
   │   MessageDeframer (assembled msgs, not yet delivered)    │      assembled msgs.
   │   AsyncCallback queue (gRPC thread → GraphStage thread)  │      Not reachable
   └──────────────────────────────────────────────────────────┘      from user code.
        │   ▲
        │   │ call.request(1) per onPull
        ▼   │
   ┌──────────────────────────────────────────────────────────┐
   │ PekkoNettyGrpcClientGraphStage                           │  ◄── DEMAND-DRIVEN.
   │   onPull → call.request(1)                               │      At most 1
   │   onMessage callback → emit(out, msg); requested -= 1    │      element held.
   └──────────────────────────────────────────────────────────┘
        │  one StreamingPullResponse (a BATCH of N msgs)
        ▼
   ┌──────────────────────────────────────────────────────────┐
   │ mapConcat(_.receivedMessages.toVector)                   │  ◄── DOMINANT
   │   holds the batch (typically 50 to 100 msgs)             │      Pekko-visible
   │   emits one per downstream pull                          │      buffer.
   └──────────────────────────────────────────────────────────┘
        │  one ReceivedMessage per pull
        ▼
   ┌──────────────────────────────────────────────────────────┐
   │ .map { tracked.put(ackId, ...); msg }     ◄── BUG 1      │  ◄── TRACKING.
   │   demand-driven 1-in-1-out                               │      Fires only
   │   onPush only fires when downstream pulls                │      on downstream
   └──────────────────────────────────────────────────────────┘      demand.
        │
        ▼
   ┌──────────────────────────────────────────────────────────┐
   │ mapAsync(parallelism = 10)                               │  ◄── PACES THE
   │   slow processing (5 to 7s per message)                  │      ENTIRE STACK
   │   pulls only when a slot frees                           │      ABOVE.
   └──────────────────────────────────────────────────────────┘

  Why this matters for the fix:

  1. There is no Pekko-layer buffer to "make smaller" between Pekko gRPC and
     mapConcat. Tweaking .buffer(N) or async boundary defaults won't help.

  2. The buffer responsible for letting the server outpace the consumer lives
     BELOW Pekko entirely (Netty + gRPC-Java). The only way to bound it from
     stream code is by withholding upstream demand: when the eager-pull stage
     stops calling pull(in), Pekko gRPC stops calling call.request(1), gRPC-Java
     has no demand to satisfy, Netty's HTTP/2 window stops refreshing, and the
     server stops dispatching.

  3. Eager-pull is the only Pekko-layer construct that reaches all the way
     down to HTTP/2 without breaking tracking. That's why it's the right shape;
     other patterns (.buffer, .async, custom Sink) either fail to backpressure
     the lower buffers or break some other property.

  4. Bug 2's fix is the protocol-layer complement. Server-side
     maxOutstandingMessages caps the gRPC-Java + Netty buffers via the protocol
     itself, so backpressure has less work to do.


═══════════════════════════════════ Bug 2 (the keepalive request leak) ══════════════════════════════════

  Initial request  ───►   server  ──► OK, stream opens
                                                              
  Pre-fix subsequentRequest = request.withSubscription("").withStreamAckDeadlineSeconds(0)
                                          │
                                          │ still carries:
                                          ├─ clientId = "client-A"
                                          ├─ maxOutstandingMessages = 100   ◄── server forbids these
                                          └─ maxOutstandingBytes    = 10MB     on subsequent requests
                                          
  Tick #2 (1s later) ───►  server  ──► ❌ INVALID_ARGUMENT
                                          stream fails


  Post-fix subsequentRequest = StreamingPullRequest.defaultInstance
                                          │
                                          ├─ subscription          = ""
                                          ├─ streamAckDeadlineSecs = 0
                                          ├─ clientId              = ""
                                          ├─ maxOutstandingMessages= 0L
                                          └─ maxOutstandingBytes   = 0L
                                          (every initial-only field cleared,
                                           forward-safe against future protos)
                                          
  Every tick ─────────►   server  ──► ✓ accepted, stream stays open
```

---

## Composition before vs after

```
═══════════════════════════════════ BEFORE: hand-composed, easy to get wrong ═════════════════════════════

  GooglePubSub.subscribe(req, 1.s, restartSettings)   ◄─── req carries maxOutstandingMessages
                                                            (bug 2 → INVALID_ARGUMENT)
    .via(GooglePubSub.autoExtendAckDeadlines(sub, ...)) ◄─── .map tracker (bug 1)
    .via(GooglePubSub.flowControlGate(fc))              ◄─── push-time permit (bug 3)
    .mapAsync(10)(processMessage)
    .map(_ => AcknowledgeRequest(...))
    .runWith(GooglePubSub.acknowledge(1, fc))         ◄─── must remember to pair fc here too

  Silent traps:
    - autoExtend inside RestartSource → tracking lost on every reconnect
    - acknowledge without fc → permits never released → gate locks forever
    - subscription mismatch between req and autoExtend → extends a different sub


═══════════════════════════════════ AFTER: one resource, correct by construction ═════════════════════════

  val subscriber = GooglePubSub.subscriber(
    request          = req w/ maxOutstandingMessages = N,    ◄── bug 2 fix automatic
    pollInterval     = 1.s,
    ackDeadline      = AckDeadline.Fixed(8.s, 30),           ◄── extender owns ticker
    restartSettings  = Some(...),                            ◄── wraps INNER subscribe only
    flowControl      = Some(FlowControl(N)))                 ◄── eager-pull gate

  subscriber.source                                          ◄── composes:
    .mapAsync(10)(processMessage)                                subscribe → autoExtend(extender)
    .map(_ => AcknowledgeRequest(...))                           → flowControlGate → watchTermination
    .runWith(subscriber.acknowledge(1))                       ◄── releases fc permits + records
                                                                   completion latencies automatically

  By construction:
    + restart only re-materializes inner subscribe
    + extender + flowControl + buffers stay alive across reconnects
    + messages buffered at moment of disconnect KEEP BEING EXTENDED
    + no permit leaks possible (acknowledge knows about fc)


═══════════════════════════════════ Restart-safety detail ═════════════════════════════════════════════════

  Inner gRPC stream fails:
                              stays alive
                          ────────────────────►
  RestartSource ── re-mat ── EagerPullTracker ── EagerPullFlowGate ── mapAsync ── ack
       │                          ▲                    ▲
       │                          │                    │
       └─ subscribe restarts      buffer holds         outstanding still
          new gRPC stream         pre-disconnect       reflects in-flight
          comes up                messages             across the gap

  Extender (caller-owned):
       ────────────────────────────────────────────────►
       map<ackId,t>     ticker fires every extensionInterval
       persists         continues extending pre-disconnect ackIds
       across           during the backoff window
       reconnects

  This matches Google's MessageDispatcher lifecycle (built once per
  StreamingSubscriberConnection, reused across every gRPC stream restart).
  And it goes further: Pekko Streams' buffering means the message PAYLOAD
  also survives, not just the ackId, so processing resumes on the original
  message instance.
```



Navigation Menu

Uh oh!

google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494 #1620

google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494

What's going on

Bug 1: `autoExtendAckDeadlines` doesn't cover messages waiting in the buffer

What we expected

What actually happens

Why "just put a `.buffer` in front of it" doesn't help

Why eager-pull is the architecturally right answer

What goes wrong

Why this is the deep cause

Why eager-pull is uniquely the right shape

Suggested fix

Bug 2: `subscribe()` makes `maxOutstandingMessages` unusable

What we expected

What actually happens

Suggested fix

Bug 3: `flowControlGate` doesn't actually flow-control

What we expected

What actually happens

Suggested fix

Note on the relationship with bug 2

Why these three bugs go together

Reproduction

Bug 1

Bug 2

Bug 3

Recommended high-level API: `Subscriber`

Visual summary: what was wrong and what this PR introduces

The structural pattern shared by bugs 1 and 3

Composition before vs after

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Field	Cleared currently?
`subscription`	yes
`streamAckDeadlineSeconds`	yes
`clientId`	no
`maxOutstandingMessages`	no
`maxOutstandingBytes`	no

Uh oh!

google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494 #1620

Description

google-cloud-pub-sub-grpc: subscriber correctness follow-up to #1494

What's going on

Bug 1: autoExtendAckDeadlines doesn't cover messages waiting in the buffer

What we expected

What actually happens

Why "just put a .buffer in front of it" doesn't help

Why eager-pull is the architecturally right answer

What goes wrong

Why this is the deep cause

Why eager-pull is uniquely the right shape

Suggested fix

Bug 2: subscribe() makes maxOutstandingMessages unusable

What we expected

What actually happens

Suggested fix

Bug 3: flowControlGate doesn't actually flow-control

What we expected

What actually happens

Suggested fix

Note on the relationship with bug 2

Why these three bugs go together

Reproduction

Bug 1

Bug 2

Bug 3

Recommended high-level API: Subscriber

Visual summary: what was wrong and what this PR introduces

The structural pattern shared by bugs 1 and 3

Composition before vs after

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Bug 1: `autoExtendAckDeadlines` doesn't cover messages waiting in the buffer

Why "just put a `.buffer` in front of it" doesn't help

Bug 2: `subscribe()` makes `maxOutstandingMessages` unusable

Bug 3: `flowControlGate` doesn't actually flow-control

Recommended high-level API: `Subscriber`