From ea6c853b41af0e4d5f38124bdc158c5f09d0d611 Mon Sep 17 00:00:00 2001 From: Nikhil Bharadwaj Ramashasthri Date: Tue, 14 Jul 2026 07:01:09 -0700 Subject: [PATCH 1/2] Bound the HTTP request buffer queue and emit static responses exactly once Two demand-handling fixes surfaced by restatedev/sdk-java#614: 1. HttpRequestFlowAdapter pushed every incoming buffer into an unbounded ArrayDeque regardless of subscriber demand. Pause the request stream on construction and mirror subscriber demand into Vert.x pull-mode fetch(), so buffers are only delivered (and queued) as requested. 2. StaticResponseRequestProcessor (discovery/health responses) re-emitted the whole response body and re-signalled onComplete on every request() call, violating Reactive Streams rules 1.7/2.12. This is currently masked by subscribers requesting Long.MAX_VALUE exactly once: any subscriber using bounded per-item requests duplicates the discovery response until the runtime rejects it as undecodable JSON. Emit exactly once. Verified with the sdk-testing testcontainers suite against a real runtime, plus new unit tests for both classes. --- .../core/StaticResponseRequestProcessor.java | 10 ++ .../StaticResponseRequestProcessorTest.java | 98 +++++++++++++ sdk-http-vertx/build.gradle.kts | 4 + .../http/vertx/HttpRequestFlowAdapter.java | 10 ++ .../vertx/HttpRequestFlowAdapterTest.java | 134 ++++++++++++++++++ 5 files changed, 256 insertions(+) create mode 100644 sdk-core/src/test/java/dev/restate/sdk/core/StaticResponseRequestProcessorTest.java create mode 100644 sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java diff --git a/sdk-core/src/main/java/dev/restate/sdk/core/StaticResponseRequestProcessor.java b/sdk-core/src/main/java/dev/restate/sdk/core/StaticResponseRequestProcessor.java index afa07db5e..b84e1783a 100644 --- a/sdk-core/src/main/java/dev/restate/sdk/core/StaticResponseRequestProcessor.java +++ b/sdk-core/src/main/java/dev/restate/sdk/core/StaticResponseRequestProcessor.java @@ -10,6 +10,7 @@ import dev.restate.common.Slice; import java.util.concurrent.Flow; +import java.util.concurrent.atomic.AtomicBoolean; class StaticResponseRequestProcessor implements RequestProcessor { @@ -37,6 +38,12 @@ public String responseContentType() { public void subscribe(Flow.Subscriber subscriber) { subscriber.onSubscribe( new Flow.Subscription() { + // Reactive Streams rules 1.7/2.12: the single response body must be emitted exactly + // once, no matter how many times request() is invoked. Re-emitting on every request() + // used to be masked by subscribers requesting Long.MAX_VALUE exactly once, and + // duplicated the whole response once subscribers switched to bounded re-requests. + private final AtomicBoolean emitted = new AtomicBoolean(false); + @Override public void request(long l) { if (l <= 0) { @@ -44,6 +51,9 @@ public void request(long l) { new IllegalStateException("subscription request is negative: " + l)); return; } + if (this.emitted.getAndSet(true)) { + return; + } subscriber.onNext(responseBody); subscriber.onComplete(); } diff --git a/sdk-core/src/test/java/dev/restate/sdk/core/StaticResponseRequestProcessorTest.java b/sdk-core/src/test/java/dev/restate/sdk/core/StaticResponseRequestProcessorTest.java new file mode 100644 index 000000000..f2de3454e --- /dev/null +++ b/sdk-core/src/test/java/dev/restate/sdk/core/StaticResponseRequestProcessorTest.java @@ -0,0 +1,98 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.core; + +import static org.assertj.core.api.Assertions.assertThat; + +import dev.restate.common.Slice; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Flow; +import org.junit.jupiter.api.Test; + +/** + * The static response (discovery, health) must be emitted exactly once, no matter how the + * subscriber shapes its demand (Reactive Streams rules 1.7/2.12). Before this test, every {@code + * request()} call re-emitted the whole body and re-signalled completion, which duplicated the + * discovery response as soon as a subscriber used bounded per-item requests instead of a single + * {@code request(Long.MAX_VALUE)}. + */ +class StaticResponseRequestProcessorTest { + + @Test + void shouldEmitTheResponseBodyExactlyOnceForBoundedDemand() { + StaticResponseRequestProcessor processor = + new StaticResponseRequestProcessor(200, "application/json", Slice.wrap("{\"a\":1}")); + RecordingSubscriber subscriber = new RecordingSubscriber(); + + processor.subscribe(subscriber); + subscriber.subscription.request(1); + // A well-behaved publisher must ignore further demand after completion + subscriber.subscription.request(1); + subscriber.subscription.request(Long.MAX_VALUE); + + assertThat(subscriber.received).hasSize(1); + assertThat(subscriber.completions).isEqualTo(1); + assertThat(subscriber.errors).isEmpty(); + } + + @Test + void shouldEmitTheResponseBodyExactlyOnceForUnboundedDemand() { + StaticResponseRequestProcessor processor = + new StaticResponseRequestProcessor(200, "application/json", Slice.wrap("{\"a\":1}")); + RecordingSubscriber subscriber = new RecordingSubscriber(); + + processor.subscribe(subscriber); + subscriber.subscription.request(Long.MAX_VALUE); + + assertThat(subscriber.received).hasSize(1); + assertThat(subscriber.completions).isEqualTo(1); + assertThat(subscriber.errors).isEmpty(); + } + + @Test + void shouldSignalAnErrorForNonPositiveDemand() { + StaticResponseRequestProcessor processor = + new StaticResponseRequestProcessor(200, "application/json", Slice.wrap("{\"a\":1}")); + RecordingSubscriber subscriber = new RecordingSubscriber(); + + processor.subscribe(subscriber); + subscriber.subscription.request(0); + + assertThat(subscriber.received).isEmpty(); + assertThat(subscriber.errors).hasSize(1); + } + + private static final class RecordingSubscriber implements Flow.Subscriber { + private Flow.Subscription subscription; + private final List received = new ArrayList<>(); + private final List errors = new ArrayList<>(); + private int completions = 0; + + @Override + public void onSubscribe(Flow.Subscription subscription) { + this.subscription = subscription; + } + + @Override + public void onNext(Slice slice) { + received.add(slice); + } + + @Override + public void onError(Throwable throwable) { + errors.add(throwable); + } + + @Override + public void onComplete() { + completions++; + } + } +} diff --git a/sdk-http-vertx/build.gradle.kts b/sdk-http-vertx/build.gradle.kts index 9aa5a8ca2..4c202e371 100644 --- a/sdk-http-vertx/build.gradle.kts +++ b/sdk-http-vertx/build.gradle.kts @@ -23,4 +23,8 @@ dependencies { // Version of vertx is pinned above exclude(group = "io.vertx") } + + testImplementation(libs.junit.jupiter) + testImplementation(libs.assertj) + testRuntimeOnly(libs.junit.platform.launcher) } diff --git a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java index bed7f9613..d77efee43 100644 --- a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java +++ b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java @@ -31,6 +31,10 @@ class HttpRequestFlowAdapter implements Flow.Publisher { HttpRequestFlowAdapter(HttpServerRequest httpServerRequest) { this.httpServerRequest = httpServerRequest; this.buffers = new ArrayDeque<>(); + // Companion fix for https://github.com/restatedev/sdk-java/issues/614: switch the request + // stream to pull mode so Vert.x only delivers buffers as the subscriber signals demand (see + // fetch below), instead of pushing the whole body into the queue above without bound. + this.httpServerRequest.pause(); } @Override @@ -73,6 +77,12 @@ private void handleSubscriptionRequest(long l) { } tryProgress(); + + // Mirror the subscriber demand into Vert.x pull-mode demand: at most this many buffers will + // be delivered to handleIncomingBuffer, so the queue is bounded by what was requested. + if (!this.httpServerRequest.isEnded()) { + this.httpServerRequest.fetch(l); + } } private void handleIncomingBuffer(Buffer buffer) { diff --git a/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java b/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java new file mode 100644 index 000000000..866a2692b --- /dev/null +++ b/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java @@ -0,0 +1,134 @@ +// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH +// +// This file is part of the Restate Java SDK, +// which is released under the MIT license. +// +// You can find a copy of the license in file LICENSE in the root +// directory of this repository or package, or at +// https://github.com/restatedev/sdk-java/blob/main/LICENSE +package dev.restate.sdk.http.vertx; + +import static org.assertj.core.api.Assertions.assertThat; + +import dev.restate.common.Slice; +import io.vertx.core.Handler; +import io.vertx.core.buffer.Buffer; +import io.vertx.core.http.HttpServerRequest; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Flow; +import org.junit.jupiter.api.Test; + +/** + * Tests for the read-side demand signaling of {@link HttpRequestFlowAdapter} (companion fix of + * issue #614): the request stream is paused on construction and buffers are pulled via {@code + * fetch} as the subscriber signals demand, so the internal queue stays bounded. + */ +class HttpRequestFlowAdapterTest { + + @Test + void shouldPauseTheRequestStreamOnConstruction() { + FakeHttpServerRequest request = new FakeHttpServerRequest(); + + new HttpRequestFlowAdapter(request.proxy()); + + assertThat(request.paused).isTrue(); + } + + @Test + void shouldMirrorSubscriberDemandIntoFetch() { + FakeHttpServerRequest request = new FakeHttpServerRequest(); + HttpRequestFlowAdapter adapter = new HttpRequestFlowAdapter(request.proxy()); + RecordingSubscriber subscriber = new RecordingSubscriber(); + + adapter.subscribe(subscriber); + subscriber.subscription.request(5); + + assertThat(request.fetches).containsExactly(5L); + } + + @Test + void shouldDeliverFetchedBuffersToTheSubscriber() { + FakeHttpServerRequest request = new FakeHttpServerRequest(); + HttpRequestFlowAdapter adapter = new HttpRequestFlowAdapter(request.proxy()); + RecordingSubscriber subscriber = new RecordingSubscriber(); + + adapter.subscribe(subscriber); + subscriber.subscription.request(2); + request.bufferHandler.handle(Buffer.buffer("hello")); + request.bufferHandler.handle(Buffer.buffer("world")); + + assertThat(subscriber.received).hasSize(2); + assertThat(request.fetches).containsExactly(2L); + } + + private static final class RecordingSubscriber implements Flow.Subscriber { + private Flow.Subscription subscription; + private final List received = new ArrayList<>(); + + @Override + public void onSubscribe(Flow.Subscription subscription) { + this.subscription = subscription; + } + + @Override + public void onNext(Slice slice) { + received.add(slice); + } + + @Override + public void onError(Throwable throwable) {} + + @Override + public void onComplete() {} + } + + /** + * Hand-rolled {@link HttpServerRequest} fake backed by a dynamic proxy, covering only the methods + * the adapter touches. Avoids pulling a mocking library into the project for one test. + */ + private static final class FakeHttpServerRequest implements InvocationHandler { + private boolean paused = false; + private boolean ended = false; + private final List fetches = new ArrayList<>(); + private Handler bufferHandler; + + private HttpServerRequest proxy() { + return (HttpServerRequest) + Proxy.newProxyInstance( + HttpServerRequest.class.getClassLoader(), + new Class[] {HttpServerRequest.class}, + this); + } + + @SuppressWarnings("unchecked") + @Override + public Object invoke(Object proxy, Method method, Object[] args) { + return switch (method.getName()) { + case "pause" -> { + this.paused = true; + yield proxy; + } + case "fetch" -> { + this.fetches.add((Long) args[0]); + yield proxy; + } + case "handler" -> { + this.bufferHandler = (Handler) args[0]; + yield proxy; + } + case "exceptionHandler", "endHandler" -> proxy; + case "isEnded" -> this.ended; + case "toString" -> "FakeHttpServerRequest"; + case "hashCode" -> System.identityHashCode(proxy); + case "equals" -> proxy == args[0]; + default -> + throw new UnsupportedOperationException( + "Unexpected call on the request fake: " + method.getName()); + }; + } + } +} From c4124750682e54aece3c14a96f07fa82ade4a490 Mon Sep 17 00:00:00 2001 From: Nikhil Bharadwaj Ramashasthri Date: Tue, 14 Jul 2026 07:01:09 -0700 Subject: [PATCH 2/2] Revert the HttpRequestFlowAdapter pause/fetch change per review Keep only the StaticResponseRequestProcessor exactly-once fix and its test. --- sdk-http-vertx/build.gradle.kts | 4 - .../http/vertx/HttpRequestFlowAdapter.java | 10 -- .../vertx/HttpRequestFlowAdapterTest.java | 134 ------------------ 3 files changed, 148 deletions(-) delete mode 100644 sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java diff --git a/sdk-http-vertx/build.gradle.kts b/sdk-http-vertx/build.gradle.kts index 4c202e371..9aa5a8ca2 100644 --- a/sdk-http-vertx/build.gradle.kts +++ b/sdk-http-vertx/build.gradle.kts @@ -23,8 +23,4 @@ dependencies { // Version of vertx is pinned above exclude(group = "io.vertx") } - - testImplementation(libs.junit.jupiter) - testImplementation(libs.assertj) - testRuntimeOnly(libs.junit.platform.launcher) } diff --git a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java index d77efee43..bed7f9613 100644 --- a/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java +++ b/sdk-http-vertx/src/main/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapter.java @@ -31,10 +31,6 @@ class HttpRequestFlowAdapter implements Flow.Publisher { HttpRequestFlowAdapter(HttpServerRequest httpServerRequest) { this.httpServerRequest = httpServerRequest; this.buffers = new ArrayDeque<>(); - // Companion fix for https://github.com/restatedev/sdk-java/issues/614: switch the request - // stream to pull mode so Vert.x only delivers buffers as the subscriber signals demand (see - // fetch below), instead of pushing the whole body into the queue above without bound. - this.httpServerRequest.pause(); } @Override @@ -77,12 +73,6 @@ private void handleSubscriptionRequest(long l) { } tryProgress(); - - // Mirror the subscriber demand into Vert.x pull-mode demand: at most this many buffers will - // be delivered to handleIncomingBuffer, so the queue is bounded by what was requested. - if (!this.httpServerRequest.isEnded()) { - this.httpServerRequest.fetch(l); - } } private void handleIncomingBuffer(Buffer buffer) { diff --git a/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java b/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java deleted file mode 100644 index 866a2692b..000000000 --- a/sdk-http-vertx/src/test/java/dev/restate/sdk/http/vertx/HttpRequestFlowAdapterTest.java +++ /dev/null @@ -1,134 +0,0 @@ -// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH -// -// This file is part of the Restate Java SDK, -// which is released under the MIT license. -// -// You can find a copy of the license in file LICENSE in the root -// directory of this repository or package, or at -// https://github.com/restatedev/sdk-java/blob/main/LICENSE -package dev.restate.sdk.http.vertx; - -import static org.assertj.core.api.Assertions.assertThat; - -import dev.restate.common.Slice; -import io.vertx.core.Handler; -import io.vertx.core.buffer.Buffer; -import io.vertx.core.http.HttpServerRequest; -import java.lang.reflect.InvocationHandler; -import java.lang.reflect.Method; -import java.lang.reflect.Proxy; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.Flow; -import org.junit.jupiter.api.Test; - -/** - * Tests for the read-side demand signaling of {@link HttpRequestFlowAdapter} (companion fix of - * issue #614): the request stream is paused on construction and buffers are pulled via {@code - * fetch} as the subscriber signals demand, so the internal queue stays bounded. - */ -class HttpRequestFlowAdapterTest { - - @Test - void shouldPauseTheRequestStreamOnConstruction() { - FakeHttpServerRequest request = new FakeHttpServerRequest(); - - new HttpRequestFlowAdapter(request.proxy()); - - assertThat(request.paused).isTrue(); - } - - @Test - void shouldMirrorSubscriberDemandIntoFetch() { - FakeHttpServerRequest request = new FakeHttpServerRequest(); - HttpRequestFlowAdapter adapter = new HttpRequestFlowAdapter(request.proxy()); - RecordingSubscriber subscriber = new RecordingSubscriber(); - - adapter.subscribe(subscriber); - subscriber.subscription.request(5); - - assertThat(request.fetches).containsExactly(5L); - } - - @Test - void shouldDeliverFetchedBuffersToTheSubscriber() { - FakeHttpServerRequest request = new FakeHttpServerRequest(); - HttpRequestFlowAdapter adapter = new HttpRequestFlowAdapter(request.proxy()); - RecordingSubscriber subscriber = new RecordingSubscriber(); - - adapter.subscribe(subscriber); - subscriber.subscription.request(2); - request.bufferHandler.handle(Buffer.buffer("hello")); - request.bufferHandler.handle(Buffer.buffer("world")); - - assertThat(subscriber.received).hasSize(2); - assertThat(request.fetches).containsExactly(2L); - } - - private static final class RecordingSubscriber implements Flow.Subscriber { - private Flow.Subscription subscription; - private final List received = new ArrayList<>(); - - @Override - public void onSubscribe(Flow.Subscription subscription) { - this.subscription = subscription; - } - - @Override - public void onNext(Slice slice) { - received.add(slice); - } - - @Override - public void onError(Throwable throwable) {} - - @Override - public void onComplete() {} - } - - /** - * Hand-rolled {@link HttpServerRequest} fake backed by a dynamic proxy, covering only the methods - * the adapter touches. Avoids pulling a mocking library into the project for one test. - */ - private static final class FakeHttpServerRequest implements InvocationHandler { - private boolean paused = false; - private boolean ended = false; - private final List fetches = new ArrayList<>(); - private Handler bufferHandler; - - private HttpServerRequest proxy() { - return (HttpServerRequest) - Proxy.newProxyInstance( - HttpServerRequest.class.getClassLoader(), - new Class[] {HttpServerRequest.class}, - this); - } - - @SuppressWarnings("unchecked") - @Override - public Object invoke(Object proxy, Method method, Object[] args) { - return switch (method.getName()) { - case "pause" -> { - this.paused = true; - yield proxy; - } - case "fetch" -> { - this.fetches.add((Long) args[0]); - yield proxy; - } - case "handler" -> { - this.bufferHandler = (Handler) args[0]; - yield proxy; - } - case "exceptionHandler", "endHandler" -> proxy; - case "isEnded" -> this.ended; - case "toString" -> "FakeHttpServerRequest"; - case "hashCode" -> System.identityHashCode(proxy); - case "equals" -> proxy == args[0]; - default -> - throw new UnsupportedOperationException( - "Unexpected call on the request fake: " + method.getName()); - }; - } - } -}