sdk-http-vertx: response writes ignore Vert.x backpressure

Bắt đầu với HttpResponseFlowAdapter.onSubscribe và onNext, sau đó so sánh contract backpressure của Vert.x WriteStream với helper Pump của nó. Giới hạn việc tạo response, tạm dừng các request khi writeQueueFull() là true và tiếp tục chúng thông qua drainHandler; xem xét HttpRequestFlowAdapter.handleIncomingBuffer để kiểm tra giới hạn queue tương ứng.

Problem

HttpResponseFlowAdapter in sdk-http-vertx ignores Vert.x's write-side backpressure on the HTTP/2 response stream:

• onSubscribe pulls unboundedly:

  this.outputSubscription = subscription;
  this.outputSubscription.request(Long.MAX_VALUE);
  

• onNext writes each slice unconditionally:

  this.httpServerResponse.write(
      Buffer.buffer(Unpooled.wrappedBuffer(slice.asReadOnlyByteBuffer())));
  

  No check on HttpServerResponse.writeQueueFull(), no drainHandler(...) to resume.

The Reactive-Streams contract and Vert.x's own WriteStream docs both expect callers to throttle production when writeQueueFull() returns true. The adapter throws away that signal.

Observed evidence

While investigating a separate e2e issue, instrumentation on the SDK side recorded writeQueueFull=true on the very first response writes under a 50-concurrent ctx.run × 10×64 KiB workload (Restate runtime as the peer). That confirms the writeQueue does cross Vert.x's high-watermark threshold in realistic Restate workloads — the adapter just keeps writing past it.

(We did not observe SDK-side heap growth or autoRead toggling in that test, so this issue is a code-correctness / future-proofing fix rather than the cause of the failure we were chasing.)

Proposal

Apply standard Reactive-Streams + Vert.x backpressure to HttpResponseFlowAdapter:

• Replace request(Long.MAX_VALUE) with a bounded initial request (e.g. request(N)).
• In onNext, after writing, check httpServerResponse.writeQueueFull():
  • If full, do not call subscription.request(...); instead install httpServerResponse.drainHandler(v -> subscription.request(M)) to resume.
  • If not full, request the next batch immediately.
• This is the same pattern Vert.x's own Pump helper implements.

Companion (lower priority, same class of bug)

The request side has an analogous unbounded enqueue: HttpRequestFlowAdapter.handleIncomingBuffer pushes incoming buffers into an ArrayDeque<ByteBuffer> without an upper bound. Worth bounding while we're in the same module.

Files

• sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpResponseFlowAdapter.java
• sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java (companion)