From d3348658214c5265e04fdedbd1d1d1ec40019c60 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:12:37 -0400 Subject: [PATCH 1/9] Add R1 protocol-neutral transport and lifecycle foundations Provide bounded borrowed-stream and subprocess transport, explicit STDIO helper selection, idle HTTP disconnect signals and remembered pre-registration cancellation. Preserve HTTP/1 pipelining and upgrade handoff with finite lookahead, and document independent SDK ownership and connection limits. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 16 + .../src/main/java/org/restlet/Request.java | 44 ++ .../main/java/org/restlet/engine/Engine.java | 30 ++ .../restlet/engine/netty/NettyClientCall.java | 7 + .../restlet/engine/netty/NettyServerCall.java | 33 +- .../engine/netty/NettyServerHandler.java | 95 +++- .../engine/stdio/StreamTransportHelper.java | 381 ++++++++++++++++ .../org/restlet/util/StreamTransport.java | 142 ++++++ .../java/org/restlet/util/WrapperRequest.java | 10 + .../org/restlet/RequestLifecycleTestCase.java | 69 +++ .../connector/HttpDisconnectTestCase.java | 212 +++++++++ .../connector/RpcSdkBoundaryTestCase.java | 119 ++++- .../StdioHelperSelectionTestCase.java | 109 +++++ .../engine/netty/Http1LookaheadTestCase.java | 101 +++++ .../restlet/util/StreamTransportTestCase.java | 414 ++++++++++++++++++ readme.md | 124 ++++++ 16 files changed, 1898 insertions(+), 8 deletions(-) create mode 100644 org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java create mode 100644 org.restlet/src/main/java/org/restlet/util/StreamTransport.java create mode 100644 org.restlet/src/test/java/org/restlet/RequestLifecycleTestCase.java create mode 100644 org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java create mode 100644 org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java create mode 100644 org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java create mode 100644 org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java diff --git a/changes.md b/changes.md index fec9b57522..eabd5778c6 100644 --- a/changes.md +++ b/changes.md @@ -77,7 +77,23 @@ Changes log explicitly. Without a valid configuration, `CorsFilter` adds no CORS headers. Issue #1611. - CORS: `CorsFilter` now applies configuration changes made after the first request (the internal `CorsResponseHelper` is rebuilt by the setters). Issue #1611. + - STDIO connector construction fails fast without an explicit or uniquely registered + supporting helper class. Applications with multiple STDIO protocol helpers must + select one explicitly. - Enhancements + - Added the supported `org.restlet.util.StreamTransport` seam for protocol-neutral, + bounded synchronous byte I/O, serialized writes, borrowed-stream cancellation and + owned subprocess lifecycle with separate bounded stderr capture. No protocol + framing, default STDIO helper or SDK dependency is added to core. + - STDIO connector construction now requires an explicit or uniquely registered + supporting helper class, rather than implicitly selecting the first protocol helper. + - Added `Request.getDisconnectSignal()` and connector notification of abnormal + termination independently of entity I/O. Abort requests made before connector + registration are remembered; wrappers delegate lifecycle and copies remain independent. + - HTTP/1.1 observes idle disconnects after request-body consumption while preserving + ordered pipelining with bounded decoded lookahead (1024 messages / 1 MiB accounted + content/header storage). Overflow closes the connection; protocol upgrades retain + their existing handoff. HTTP/2 and HTTP/3 keep stream-level disconnect notification. - HTTP/2 and HTTP/3 clients deliver response headers without waiting for the first response body byte. This allows idle duplex and SSE responses to expose their headers before the peer sends a message. diff --git a/org.restlet/src/main/java/org/restlet/Request.java b/org.restlet/src/main/java/org/restlet/Request.java index cb5eeee2fe..b21621ba1b 100644 --- a/org.restlet/src/main/java/org/restlet/Request.java +++ b/org.restlet/src/main/java/org/restlet/Request.java @@ -10,9 +10,13 @@ import java.util.List; import java.util.Map.Entry; +import java.util.Objects; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BooleanSupplier; import org.restlet.data.CacheDirective; import org.restlet.data.ChallengeResponse; @@ -85,6 +89,12 @@ public static Request getCurrent() { /** Action used by an active client connector to abort this request. */ private volatile BooleanSupplier abortAction; + private final AtomicBoolean abortRequested = new AtomicBoolean(); + + private final CompletableFuture disconnectSignal = new CompletableFuture<>(); + + private final AtomicBoolean disconnectReported = new AtomicBoolean(); + /** The trailing headers received with the request entity. */ private volatile Series
trailers; @@ -314,8 +324,12 @@ private void copyClientInfo(final Request request) { * closing the socket. * * @return True if the request was aborted. + * @implNote An attempt made before connector registration is remembered and applied when an + * active action is installed. It still returns false until an active action accepts it. Use + * a new Request for a new exchange after cancellation. */ public boolean abort() { + abortRequested.set(true); BooleanSupplier action = abortAction; return action != null && action.getAsBoolean(); } @@ -325,9 +339,39 @@ public boolean abort() { * this while a request is in flight and clear it when the exchange ends. * * @param action The abort action, or null when no exchange is active. + * @implNote The action must be idempotent: concurrent registration and abort attempts can + * invoke it more than once. */ public void setAbortAction(BooleanSupplier action) { this.abortAction = action; + if (action != null && abortRequested.get()) { + action.getAsBoolean(); + } + } + + /** + * Returns a read-only completion stage for abnormal transport disconnection of this exchange. + * It completes once with the cause, even if subscribed after the disconnect, and does not + * complete on normal response completion. Connectors report failure independently of entity + * reads/writes, so an idle application can observe cancellation. Completion runs off the + * connector event loop; consumers should still use their own executor for expensive work. Not + * all connectors support this signal. Copies have independent lifecycle state. + */ + public CompletionStage getDisconnectSignal() { + return disconnectSignal.minimalCompletionStage(); + } + + /** + * Reports abnormal disconnection from a connector. The first cause wins; normal completion must + * not call this method. + * + * @param cause The non-null transport failure. + */ + public void notifyDisconnect(Throwable cause) { + Objects.requireNonNull(cause, "cause"); + if (disconnectReported.compareAndSet(false, true)) { + Thread.startVirtualThread(() -> disconnectSignal.complete(cause)); + } } /** diff --git a/org.restlet/src/main/java/org/restlet/engine/Engine.java b/org.restlet/src/main/java/org/restlet/engine/Engine.java index 3e636c2b3f..262aaf74fa 100644 --- a/org.restlet/src/main/java/org/restlet/engine/Engine.java +++ b/org.restlet/src/main/java/org/restlet/engine/Engine.java @@ -476,6 +476,7 @@ public org.restlet.engine.connector.ConnectorHelper createHelper( if (client.getProtocols().isEmpty()) { return result; } + requireUniqueStdioHelper(client.getProtocols(), helperClass, getRegisteredClients()); org.restlet.engine.connector.ConnectorHelper connector = null; for (final Iterator> iter = @@ -533,6 +534,7 @@ public ConnectorHelper createHelper(Server server, String helperClass) { if (server.getProtocols().isEmpty()) { return result; } + requireUniqueStdioHelper(server.getProtocols(), helperClass, getRegisteredServers()); ConnectorHelper connector = null; for (final Iterator> iter = getRegisteredServers().iterator(); @@ -575,6 +577,34 @@ public ConnectorHelper createHelper(Server server, String helperClass) { return result; } + private static void requireUniqueStdioHelper( + List protocols, + String helperClass, + List> helpers) { + if (!protocols.contains(Protocol.STDIO)) { + return; + } + long matches = + helpers.stream() + .filter( + helper -> + new HashSet<>(helper.getProtocols()).containsAll(protocols)) + .filter( + helper -> + helperClass == null + || helper.getClass() + .getCanonicalName() + .equals(helperClass)) + .map(Object::getClass) + .distinct() + .count(); + if (matches != 1) { + throw new IllegalStateException( + "STDIO requires one explicitly selected or uniquely registered connector helper; found " + + matches); + } + } + /** * Discovers the authenticator helpers and register the default helpers. * diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java index e5b81846d5..be992f7977 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java @@ -387,6 +387,7 @@ private void finishRequestBody( } private Status handleConnectFailure(ConnectException cause) { + currentRequest.notifyDisconnect(cause); completeEntityRelease(); helper.logFailure("Unable to connect to the remote HTTP server", cause); abort(); @@ -394,6 +395,7 @@ private Status handleConnectFailure(ConnectException cause) { } private Status handleCommunicationFailure(IOException cause) { + currentRequest.notifyDisconnect(cause); completeEntityRelease(); helper.logFailure( "An error occurred during the communication with the remote HTTP server", cause); @@ -402,6 +404,7 @@ private Status handleCommunicationFailure(IOException cause) { } private Status handleInternalFailure(RuntimeException cause) { + currentRequest.notifyDisconnect(cause); completeEntityRelease(); helper.logFailure("An unexpected error occurred while sending the request", cause); abort(); @@ -868,6 +871,10 @@ void onFailure(Throwable cause) { lock.unlock(); } ChannelInputStream stream = this.body; + org.restlet.Request request = currentRequest; + if (request != null) { + request.notifyDisconnect(cause); + } if (stream != null) { stream.fail(cause); } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index aa27fe27f6..09bca09ca7 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -131,6 +131,10 @@ class NettyServerCall { /** Indicates whether the exchange was completed. */ private volatile boolean completed; + private NettyRequest restletRequest; + + private Throwable disconnectCause; + /** The response body stream, if any. */ private volatile ChannelOutputStream responseBody; @@ -215,8 +219,12 @@ private Logger getLogger() { * @param context The context of the server connector. * @return The Restlet request. */ - NettyRequest toRequest(Context context) { + synchronized NettyRequest toRequest(Context context) { NettyRequest result = new NettyRequest(context, this); + restletRequest = result; + if (disconnectCause != null) { + result.notifyDisconnect(disconnectCause); + } result.getAttributes().put(ATTRIBUTE_HEADERS, requestHeaders); result.getAttributes().put(ATTRIBUTE_VERSION, kind.version()); @@ -244,6 +252,15 @@ NettyRequest toRequest(Context context) { return result; } + synchronized void onDisconnect(Throwable cause) { + if (!completed && disconnectCause == null) { + disconnectCause = cause; + if (restletRequest != null) { + restletRequest.notifyDisconnect(cause); + } + } + } + /** * Creates the Restlet response of this exchange. * @@ -720,7 +737,19 @@ void complete() { if (!kind.isStream()) { if (keepAlive) { // Ready for the next request on this connection. - future.addListener(f -> channel.read()); + future.addListener( + f -> { + if (f.isSuccess()) { + var context = + channel.pipeline().context(NettyServerHandler.class); + if (context != null) { + ((NettyServerHandler) context.handler()) + .responseComplete(context, this); + } + } else { + channel.close(); + } + }); } else { future.addListener(ChannelFutureListener.CLOSE); } diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index e6c7223b69..8356a52a1a 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java @@ -15,6 +15,7 @@ import io.netty.handler.codec.http.HttpContent; import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpHeaderValues; +import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpUtil; @@ -25,7 +26,10 @@ import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.nio.channels.ClosedChannelException; +import java.util.ArrayDeque; +import java.util.Deque; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicReference; import java.util.logging.Level; import org.restlet.engine.Engine; @@ -33,9 +37,9 @@ * Inbound handler that converts Netty HTTP messages into {@link NettyServerCall}s and runs them on * the call executor. One instance per HTTP/1.1 connection, or per HTTP/2 or HTTP/3 stream. * - *

HTTP/1.1 requests on a connection are processed one at a time: reading is suspended while a - * call is in progress and resumed when its response is complete. This keeps pipelined requests in - * order without buffering them. + *

HTTP/1.1 requests are processed one at a time. After a request body ends, one read observes + * peer disconnects even while the application is idle. Pipelined messages from that read are + * retained within finite limits until the preceding response is flushed. */ class NettyServerHandler extends ChannelInboundHandlerAdapter { @@ -55,9 +59,18 @@ class NettyServerHandler extends ChannelInboundHandlerAdapter { /** The call receiving the current request body. */ private NettyServerCall currentCall; + private final AtomicReference activeCall = new AtomicReference<>(); + + private final Deque pending = new ArrayDeque<>(); + private long pendingBytes; + private static final int MAX_PENDING_MESSAGES = 1024; + private static final int MAX_PENDING_BYTES = 1024 * 1024; + /** Whether to read the rest of the current HTTP/1.1 request once the channel read completes. */ private boolean readRequestBody; + private boolean readAfterBody; + NettyServerHandler(NettyServerHelper helper, boolean confidential, StreamKind kind) { this.helper = helper; this.confidential = confidential; @@ -85,6 +98,18 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (kind == StreamKind.HTTP1 && currentBody == null && activeCall.get() != null) { + long bytes = pendingSize(msg); + if (pending.size() >= MAX_PENDING_MESSAGES + || pendingBytes + bytes > MAX_PENDING_BYTES) { + ReferenceCountUtil.release(msg); + exceptionCaught(ctx, new IOException("HTTP/1.1 pipelined request limit exceeded")); + } else { + pending.addLast(msg); + pendingBytes += bytes; + } + return; + } try { if (msg instanceof HttpRequest request) { onRequest(ctx, request); @@ -99,12 +124,45 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { } } + private static long pendingSize(Object msg) { + long bytes = msg instanceof HttpContent content ? content.content().readableBytes() : 0; + if (msg instanceof HttpRequest request) { + bytes += 2L * request.uri().length(); + for (var header : request.headers()) { + bytes += 2L * (header.getKey().length() + header.getValue().length()); + } + } + if (msg instanceof LastHttpContent content) { + for (var header : content.trailingHeaders()) { + bytes += 2L * (header.getKey().length() + header.getValue().length()); + } + } + return bytes; + } + + /** Invoked on the event loop after the previous response has been flushed. */ + void responseComplete(ChannelHandlerContext ctx, NettyServerCall call) { + activeCall.compareAndSet(call, null); + while (!pending.isEmpty() && (activeCall.get() == null || currentBody != null)) { + Object msg = pending.removeFirst(); + pendingBytes -= pendingSize(msg); + channelRead(ctx, msg); + } + if (pending.isEmpty()) { + ctx.read(); + } + } + private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { DecoderResult result = request.decoderResult(); if (result.isFailure()) { sendError(ctx, HttpResponseStatus.BAD_REQUEST); return; } + // Upgrade/tunnel bytes belong to the replacement protocol, not the HTTP decoder. + readAfterBody = + !request.headers().contains(HttpHeaderNames.UPGRADE) + && !request.method().equals(HttpMethod.CONNECT); if (HttpUtil.is100ContinueExpected(request)) { ctx.writeAndFlush( @@ -124,6 +182,7 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { NettyServerCall call = new NettyServerCall(helper, ctx.channel(), kind, request, body, confidential); currentCall = call; + activeCall.set(call); if (kind != StreamKind.HTTP3) { call.setAltSvc(helper.getAltSvc()); } @@ -131,6 +190,8 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { if (!helper.tryAcquireCall()) { body.close(); currentBody = null; + currentCall = null; + activeCall.compareAndSet(call, null); sendError(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE); return; } @@ -141,6 +202,8 @@ private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { helper.releaseCall(); body.close(); currentBody = null; + currentCall = null; + activeCall.compareAndSet(call, null); sendError(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE); } @@ -187,8 +250,9 @@ private void onContent(ChannelHandlerContext ctx, HttpContent content) { if (last) { body.end(); currentBody = null; - // Do not read the next HTTP/1.1 request until the current response is complete; the - // call requests the next read when it completes. + if (kind == StreamKind.HTTP1 && readAfterBody && pending.isEmpty()) { + ctx.read(); + } } else if (readMore) { ctx.read(); } @@ -214,6 +278,9 @@ private void runCall(NettyServerCall call) { try { call.complete(); } finally { + if (kind.isStream()) { + activeCall.compareAndSet(call, null); + } helper.endCall(); Engine.clearThreadLocalVariables(); } @@ -223,9 +290,23 @@ private void runCall(NettyServerCall call) { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { failBody(new ClosedChannelException()); + releasePending(); super.channelInactive(ctx); } + @Override + public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { + releasePending(); + super.handlerRemoved(ctx); + } + + private void releasePending() { + while (!pending.isEmpty()) { + ReferenceCountUtil.release(pending.removeFirst()); + } + pendingBytes = 0; + } + @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { @@ -249,6 +330,10 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { } private void failBody(Throwable cause) { + NettyServerCall call = activeCall.get(); + if (call != null) { + call.onDisconnect(cause); + } ChannelInputStream body = currentBody; if (body != null) { body.fail(cause); diff --git a/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java new file mode 100644 index 0000000000..55543451c2 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java @@ -0,0 +1,381 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.engine.stdio; + +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.OutputStream; +import java.time.Duration; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantLock; + +/** Internal implementation of the supported {@link org.restlet.util.StreamTransport} seam. */ +public final class StreamTransportHelper { + + private final List command; + private final Runnable cancelIo; + private final ReentrantLock readLock = new ReentrantLock(); + private final ReentrantLock writeLock = new ReentrantLock(); + private final Set operations = new HashSet<>(); + private InputStream input; + private OutputStream output; + private Process process; + private Thread stderrReader; + private IOException stderrFailure; + private byte[] stderrTail; + private int stderrSize; + private int stderrPosition; + private int maxOperationBytes = 65536; + private int maxStderrBytes = 8192; + private Duration shutdownTimeout = Duration.ofSeconds(5); + private Duration processExitGrace = Duration.ofMillis(250); + private boolean started; + private boolean stopped; + private CompletableFuture stopping; + + public StreamTransportHelper(InputStream input, OutputStream output, Runnable cancelIo) { + this.input = Objects.requireNonNull(input, "input"); + this.output = Objects.requireNonNull(output, "output"); + this.cancelIo = Objects.requireNonNull(cancelIo, "cancelIo"); + command = null; + } + + public StreamTransportHelper(List command) { + this.command = List.copyOf(command); + if (this.command.isEmpty() || this.command.getFirst().isBlank()) { + throw new IllegalArgumentException("An explicit executable is required"); + } + cancelIo = null; + } + + public synchronized int getMaxOperationBytes() { + return maxOperationBytes; + } + + public synchronized void setMaxOperationBytes(int bytes) { + requireConfigurable(); + if (bytes <= 0) { + throw new IllegalArgumentException("maxOperationBytes must be positive"); + } + maxOperationBytes = bytes; + } + + public synchronized int getMaxStderrBytes() { + return maxStderrBytes; + } + + public synchronized void setMaxStderrBytes(int bytes) { + requireConfigurable(); + if (bytes <= 0) { + throw new IllegalArgumentException("maxStderrBytes must be positive"); + } + maxStderrBytes = bytes; + } + + public synchronized Duration getShutdownTimeout() { + return shutdownTimeout; + } + + public synchronized void setShutdownTimeout(Duration timeout) { + requireConfigurable(); + if (timeout.isNegative() || timeout.isZero() || timeout.compareTo(processExitGrace) <= 0) { + throw new IllegalArgumentException("shutdownTimeout must exceed processExitGrace"); + } + timeout.toNanos(); + shutdownTimeout = timeout; + } + + public synchronized Duration getProcessExitGrace() { + return processExitGrace; + } + + public synchronized void setProcessExitGrace(Duration grace) { + requireConfigurable(); + if (grace.isNegative() || grace.compareTo(shutdownTimeout) >= 0) { + throw new IllegalArgumentException( + "processExitGrace must be nonnegative and below shutdownTimeout"); + } + grace.toNanos(); + processExitGrace = grace; + } + + private void requireConfigurable() { + if (started || stopped) { + throw new IllegalStateException("Transport configuration is fixed after start or stop"); + } + } + + public synchronized void start() throws IOException { + if (stopped) { + throw new IllegalStateException("A stopped transport cannot be restarted"); + } + if (started) { + return; + } + if (command != null) { + stderrTail = new byte[maxStderrBytes]; + stderrReader = + Thread.ofVirtual().name("restlet-stdio-stderr").unstarted(this::drainStderr); + process = new ProcessBuilder(command).start(); + input = process.getInputStream(); + output = process.getOutputStream(); + try { + stderrReader.start(); + } catch (RuntimeException | Error e) { + stopped = true; + process.destroyForcibly(); + IOException closeFailure = closePipes(null); + if (closeFailure != null) { + e.addSuppressed(closeFailure); + } + try { + process.getErrorStream().close(); + } catch (IOException closeError) { + e.addSuppressed(closeError); + } + throw e; + } + } + started = true; + } + + public synchronized boolean isStarted() { + return started && !stopped; + } + + private synchronized void enter() throws IOException { + if (!isStarted()) { + throw new IOException("Stream transport is not accepting I/O"); + } + if (stderrFailure != null) { + throw new IOException("Subprocess stderr drain failed", stderrFailure); + } + operations.add(Thread.currentThread()); + } + + private synchronized void leave() { + operations.remove(Thread.currentThread()); + notifyAll(); + } + + public int read(byte[] buffer, int offset, int length) throws IOException { + Objects.checkFromIndexSize(offset, length, buffer.length); + lock(readLock); + try { + enter(); + try { + int count = input.read(buffer, offset, Math.min(length, maxOperationBytes)); + if (count == -1 && process != null) { + if (!process.waitFor(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS)) { + throw new IOException( + "Subprocess stdout ended without process termination"); + } + if (process.exitValue() != 0 && !isStopped()) { + throw new IOException("Subprocess exited with code " + process.exitValue()); + } + } + return count; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted awaiting subprocess exit"); + } finally { + leave(); + } + } finally { + readLock.unlock(); + } + } + + public void write(byte[] bytes) throws IOException { + Objects.requireNonNull(bytes, "bytes"); + if (bytes.length > getMaxOperationBytes()) { + throw new IllegalArgumentException("Write exceeds maxOperationBytes"); + } + lock(writeLock); + try { + enter(); + try { + output.write(bytes); + output.flush(); + } finally { + leave(); + } + } finally { + writeLock.unlock(); + } + } + + private static void lock(ReentrantLock lock) throws InterruptedIOException { + try { + lock.lockInterruptibly(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted awaiting stream I/O"); + } + } + + private synchronized boolean isStopped() { + return stopped; + } + + private void drainStderr() { + try (InputStream errors = process.getErrorStream()) { + byte[] chunk = new byte[Math.min(4096, maxStderrBytes)]; + int count; + while ((count = errors.read(chunk)) != -1) { + synchronized (this) { + for (int i = 0; i < count; i++) { + stderrTail[stderrPosition] = chunk[i]; + stderrPosition = (stderrPosition + 1) % stderrTail.length; + stderrSize = Math.min(stderrSize + 1, stderrTail.length); + } + } + } + } catch (IOException e) { + synchronized (this) { + stderrFailure = e; + } + } + } + + public synchronized byte[] getStderrTail() { + byte[] result = new byte[stderrSize]; + for (int i = 0; i < result.length; i++) { + result[i] = + stderrTail[ + (stderrPosition - stderrSize + i + stderrTail.length) + % stderrTail.length]; + } + return result; + } + + public synchronized long getProcessId() { + if (process == null) { + throw new IllegalStateException("No owned subprocess has been started"); + } + return process.pid(); + } + + public void stop() throws IOException { + CompletableFuture completion; + synchronized (this) { + if (stopping == null) { + stopped = true; + stopping = new CompletableFuture<>(); + if (!started) { + stopping.complete(null); + } else { + Thread.startVirtualThread(this::cleanup); + } + } + completion = stopping; + } + try { + completion.get(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted stopping stream transport"); + } catch (TimeoutException e) { + throw new IOException( + "Stream transport shutdown timed out; borrowed I/O must support cancellation", + e); + } catch (ExecutionException e) { + throw new IOException("Stream transport shutdown failed", e.getCause()); + } + } + + private void cleanup() { + try { + IOException failure = null; + synchronized (this) { + operations.forEach(Thread::interrupt); + } + if (process != null) { + boolean exited = process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS); + if (exited && process.exitValue() != 0) { + failure = new IOException("Subprocess exited with code " + process.exitValue()); + } + if (!exited) { + process.destroy(); + if (!process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS)) { + process.destroyForcibly(); + } + } + process.waitFor(); + failure = closePipes(failure); + stderrReader.join(); + } else { + try { + cancelIo.run(); + } catch (RuntimeException e) { + failure = new IOException("Borrowed I/O cancellation failed", e); + } + } + synchronized (this) { + while (!operations.isEmpty()) { + wait(); + } + if (stderrFailure != null) { + failure = + append( + failure, + new IOException( + "Subprocess stderr drain failed", stderrFailure)); + } + } + if (process == null) { + try { + output.flush(); + } catch (IOException e) { + failure = append(failure, e); + } + } + if (failure != null) { + throw failure; + } + stopping.complete(null); + } catch (IOException | RuntimeException e) { + stopping.completeExceptionally(e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + stopping.completeExceptionally(e); + } finally { + if (process != null && process.isAlive()) { + process.destroyForcibly(); + } + } + } + + private IOException closePipes(IOException failure) { + for (java.io.Closeable pipe : List.of(input, output)) { + try { + pipe.close(); + } catch (IOException e) { + failure = append(failure, e); + } + } + return failure; + } + + private static IOException append(IOException failure, IOException next) { + if (failure == null) { + return next; + } + failure.addSuppressed(next); + return failure; + } +} diff --git a/org.restlet/src/main/java/org/restlet/util/StreamTransport.java b/org.restlet/src/main/java/org/restlet/util/StreamTransport.java new file mode 100644 index 0000000000..df287e4117 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/util/StreamTransport.java @@ -0,0 +1,142 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.util; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.time.Duration; +import java.util.List; +import org.restlet.engine.stdio.StreamTransportHelper; + +/** + * Protocol-neutral, blocking byte transport for borrowed streams or an owned subprocess. + * + *

Reads are on demand, with no background reader or input prefetch. Writes are serialized and + * flushed individually. A write is an opaque byte sequence, not a message or a framing convention. + * Configure limits before {@link #start()}; a stopped transport cannot be restarted. + * + *

Borrowed streams are never closed. Their owner must provide a prompt cancellation action that + * unblocks active I/O without closing borrowed standard streams. Thread interruption is also + * attempted. Arbitrary {@link InputStream}s need not be interruptible: if cancellation does not + * release I/O, {@link #stop()} fails within its timeout rather than claiming successful cleanup. No + * JVM-global standard stream is redirected. + */ +public final class StreamTransport implements AutoCloseable { + + private final StreamTransportHelper helper; + + /** + * Creates a transport borrowing both streams. + * + * @param input The protocol input. + * @param output The protocol output. + * @param cancelIo Prompt action to release blocked borrowed I/O on stop. + */ + public StreamTransport(InputStream input, OutputStream output, Runnable cancelIo) { + helper = new StreamTransportHelper(input, output, cancelIo); + } + + /** + * Creates a transport owning a subprocess and its pipes. No shell is used and no command is + * inferred from a URI. Stderr is drained separately into a bounded tail. + * + * @param command Executable followed by individual arguments. + */ + public StreamTransport(List command) { + helper = new StreamTransportHelper(command); + } + + /** Returns the maximum bytes read or written by one operation (default 65536). */ + public int getMaxOperationBytes() { + return helper.getMaxOperationBytes(); + } + + /** Sets the positive per-operation byte limit before start. */ + public void setMaxOperationBytes(int bytes) { + helper.setMaxOperationBytes(bytes); + } + + /** Returns the maximum retained subprocess stderr bytes (default 8192). */ + public int getMaxStderrBytes() { + return helper.getMaxStderrBytes(); + } + + /** Sets the positive stderr tail capacity before start. */ + public void setMaxStderrBytes(int bytes) { + helper.setMaxStderrBytes(bytes); + } + + /** Returns the maximum shutdown duration (default five seconds). */ + public Duration getShutdownTimeout() { + return helper.getShutdownTimeout(); + } + + /** Sets the positive shutdown timeout before start. */ + public void setShutdownTimeout(Duration timeout) { + helper.setShutdownTimeout(timeout); + } + + /** Returns the subprocess exit grace period (default 250 milliseconds). */ + public Duration getProcessExitGrace() { + return helper.getProcessExitGrace(); + } + + /** Sets a nonnegative process exit grace period shorter than the shutdown timeout. */ + public void setProcessExitGrace(Duration grace) { + helper.setProcessExitGrace(grace); + } + + /** Starts admission and, when configured, the owned subprocess. */ + public void start() throws IOException { + helper.start(); + } + + /** Indicates whether new I/O operations are accepted. */ + public boolean isStarted() { + return helper.isStarted(); + } + + /** + * Reads at most the configured operation limit, returning -1 on clean EOF. Only one reader + * executes at a time; a chunk has no application-level framing significance. + */ + public int read(byte[] buffer, int offset, int length) throws IOException { + return helper.read(buffer, offset, length); + } + + /** Writes and flushes one opaque sequence, rejecting values over the operation limit. */ + public void write(byte[] bytes) throws IOException { + helper.write(bytes); + } + + /** Returns a snapshot of the bounded stderr tail; borrowed streams have no stderr pipe. */ + public byte[] getStderrTail() { + return helper.getStderrTail(); + } + + /** Returns the owned subprocess PID, or throws when no subprocess is running. */ + public long getProcessId() { + return helper.getProcessId(); + } + + /** + * Stops admission, cancels active I/O and waits within the shutdown timeout. Flushes, but never + * closes, borrowed output. Closes owned pipes and terminates only the owned process. Reports + * I/O, stderr-drain, process-exit and timeout failures. + */ + public void stop() throws IOException { + helper.stop(); + } + + @Override + public void close() throws IOException { + stop(); + } +} diff --git a/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java b/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java index d5ad4f25f7..beafd73343 100644 --- a/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java +++ b/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java @@ -57,6 +57,16 @@ public void setAbortAction(BooleanSupplier action) { wrappedRequest.setAbortAction(action); } + @Override + public java.util.concurrent.CompletionStage getDisconnectSignal() { + return wrappedRequest.getDisconnectSignal(); + } + + @Override + public void notifyDisconnect(Throwable cause) { + wrappedRequest.notifyDisconnect(cause); + } + @Override public void commit(Response response) { wrappedRequest.commit(response); diff --git a/org.restlet/src/test/java/org/restlet/RequestLifecycleTestCase.java b/org.restlet/src/test/java/org/restlet/RequestLifecycleTestCase.java new file mode 100644 index 0000000000..e57c9ebdf0 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/RequestLifecycleTestCase.java @@ -0,0 +1,69 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; + +import java.io.IOException; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.restlet.data.Method; +import org.restlet.util.WrapperRequest; + +class RequestLifecycleTestCase { + + @Test + void reportsOnceWithLateSubscriptionAndWrapperDelegation() throws Exception { + Request request = new Request(Method.GET, "http://localhost/"); + WrapperRequest wrapper = new WrapperRequest(request); + IOException failure = new IOException("first"); + wrapper.notifyDisconnect(failure); + request.notifyDisconnect(new IOException("second")); + AtomicInteger callbacks = new AtomicInteger(); + var signal = + wrapper.getDisconnectSignal() + .thenAccept( + cause -> { + assertSame(failure, cause); + callbacks.incrementAndGet(); + }); + signal.toCompletableFuture().get(5, TimeUnit.SECONDS); + assertEquals(1, callbacks.get()); + assertSame( + failure, + request.getDisconnectSignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertFalse(new Request(request).getDisconnectSignal().toCompletableFuture().isDone()); + } + + @Test + void racesAbortWithConnectorRegistrationWithoutLosingCancellation() throws Exception { + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + for (int i = 0; i < 100; i++) { + Request request = new Request(); + AtomicInteger accepted = new AtomicInteger(); + var abort = executor.submit(request::abort); + var register = + executor.submit( + () -> + request.setAbortAction( + () -> { + accepted.incrementAndGet(); + return true; + })); + abort.get(5, TimeUnit.SECONDS); + register.get(5, TimeUnit.SECONDS); + org.junit.jupiter.api.Assertions.assertTrue(accepted.get() >= 1); + } + } + } +} diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java new file mode 100644 index 0000000000..8594eedd4c --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java @@ -0,0 +1,212 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.engine.connector; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.Server; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Protocol; +import org.restlet.engine.Engine; +import org.restlet.representation.StringRepresentation; + +class HttpDisconnectTestCase { + + private Client client; + private Server server; + + @BeforeEach + void setUp() { + Engine.clearThreadLocalVariables(); + Engine.register(false); + Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null)); + Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null)); + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.stop(); + } + if (server != null) { + server.stop(); + } + Engine.clearThreadLocalVariables(); + } + + private void start(String mode, Restlet handler) throws Exception { + Context serverContext = new Context(); + serverContext.getParameters().add("http.transport.protocol", mode); + server = new Server(serverContext, Protocol.HTTP, 0, handler); + server.start(); + Context clientContext = new Context(); + clientContext.getParameters().add("httpClientTransportMode", mode); + client = new Client(clientContext, Protocol.HTTP); + client.start(); + } + + @ParameterizedTest + @ValueSource(strings = {"HTTP1_1", "HTTP2"}) + void notifiesIdleServerAfterEntireRequestBodyWasConsumed(String mode) throws Exception { + CountDownLatch idle = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(1); + AtomicReference cause = new AtomicReference<>(); + AtomicReference handlerFailure = new AtomicReference<>(); + AtomicInteger notifications = new AtomicInteger(); + start( + mode, + new Restlet() { + @Override + public void handle(Request request, Response response) { + if (request.getResourceRef().getPath().equals("/healthy")) { + response.setEntity("healthy", MediaType.TEXT_PLAIN); + return; + } + try { + assertEquals("body-consumed", request.getEntityAsText()); + var disconnected = + request.getDisconnectSignal() + .thenAccept( + error -> { + notifications.incrementAndGet(); + cause.set(error); + }); + idle.countDown(); + disconnected.toCompletableFuture().get(5, TimeUnit.SECONDS); + } catch (Exception | AssertionError e) { + handlerFailure.set(e); + } finally { + finished.countDown(); + } + } + }); + Request request = + new Request( + Method.POST, + "http://localhost:" + server.getActualPort() + "/idle", + new StringRepresentation("body-consumed", MediaType.TEXT_PLAIN)); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var response = executor.submit(() -> client.handle(request)); + assertTrue(idle.await(5, TimeUnit.SECONDS)); + assertTrue(request.abort()); + assertNotNull( + request.getDisconnectSignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertTrue(finished.await(5, TimeUnit.SECONDS)); + if (handlerFailure.get() != null) { + throw new AssertionError( + "Idle server did not observe disconnect", handlerFailure.get()); + } + assertNotNull(cause.get()); + assertEquals(1, notifications.get()); + assertTrue(response.get(5, TimeUnit.SECONDS).getStatus().isConnectorError()); + } + Response healthy = + client.handle( + new Request( + Method.GET, + "http://localhost:" + server.getActualPort() + "/healthy")); + assertEquals("healthy", healthy.getEntityAsText()); + assertFalse(healthy.getRequest().getDisconnectSignal().toCompletableFuture().isDone()); + } + + @ParameterizedTest + @ValueSource(strings = {"HTTP1_1", "HTTP2"}) + void remembersCancellationBeforeConnectorRegistration(String mode) throws Exception { + AtomicInteger received = new AtomicInteger(); + start( + mode, + new Restlet() { + @Override + public void handle(Request request, Response response) { + received.incrementAndGet(); + response.setEntity("unexpected", MediaType.TEXT_PLAIN); + } + }); + Request request = new Request(Method.GET, "http://localhost:" + server.getActualPort()); + assertFalse(request.abort(), "No active connector has accepted cancellation yet"); + Response response = client.handle(request); + assertTrue(response.getStatus().isConnectorError()); + assertNotNull(request.getDisconnectSignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertEquals(0, received.get()); + } + + @Test + void preservesHttp1PipeliningOrderWhileWatchingForIdleDisconnect() throws Exception { + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + AtomicInteger received = new AtomicInteger(); + AtomicReference first = new AtomicReference<>(); + start( + "HTTP1_1", + new Restlet() { + @Override + public void handle(Request request, Response response) { + int index = received.incrementAndGet(); + if (index == 1) { + first.set(request); + firstStarted.countDown(); + try { + if (!releaseFirst.await(5, TimeUnit.SECONDS)) { + throw new AssertionError( + "First pipelined request was not released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + response.setEntity("response-" + index, MediaType.TEXT_PLAIN); + } + }); + try (var socket = new java.net.Socket("localhost", server.getActualPort())) { + socket.setSoTimeout(5000); + var output = socket.getOutputStream(); + output.write( + ("GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n") + .getBytes(java.nio.charset.StandardCharsets.US_ASCII)); + output.flush(); + assertTrue(firstStarted.await(5, TimeUnit.SECONDS)); + output.write( + ("GET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .getBytes(java.nio.charset.StandardCharsets.US_ASCII)); + output.flush(); + releaseFirst.countDown(); + String responses = + new String( + socket.getInputStream().readAllBytes(), + java.nio.charset.StandardCharsets.US_ASCII); + assertEquals(2, received.get()); + assertTrue(responses.contains("response-1"), responses); + assertTrue( + responses.indexOf("response-2") > responses.indexOf("response-1"), responses); + assertFalse(first.get().getDisconnectSignal().toCompletableFuture().isDone()); + } finally { + releaseFirst.countDown(); + } + } +} diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/RpcSdkBoundaryTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/RpcSdkBoundaryTestCase.java index 94d03585d9..33aedffb91 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/RpcSdkBoundaryTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/RpcSdkBoundaryTestCase.java @@ -22,12 +22,14 @@ import java.util.Map; import java.util.ServiceLoader; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.restlet.Client; class RpcSdkBoundaryTestCase { @Test - void resolvesPinnedSdkModulesWithoutOfficialNetworkTransports() throws Exception { + void resolvesPinnedSdkModulesWithoutOfficialNetworkTransports(@TempDir Path temporary) + throws Exception { Map> modules = Map.of( "io.grpc", @@ -72,6 +74,99 @@ void resolvesPinnedSdkModulesWithoutOfficialNetworkTransports() throws Exception var paths = new java.util.ArrayList<>(dependencies); paths.add( Path.of(Client.class.getProtectionDomain().getCodeSource().getLocation().toURI())); + Path descriptor = temporary.resolve("module-info.java"); + Path example = temporary.resolve("PublicTransportExample.java"); + Files.writeString(descriptor, "module r1.consumer { requires org.restlet; }"); + Files.writeString( + example, + """ + package example; + import java.io.IOException; + import java.io.InputStream; + import java.io.OutputStream; + import org.restlet.util.StreamTransport; + public final class PublicTransportExample { + public static void exchange(InputStream input, OutputStream output, Runnable cancel) + throws IOException { + try (var transport = new StreamTransport(input, output, cancel)) { + transport.start(); + transport.write(new byte[] {0, 1, 2}); + transport.read(new byte[8], 0, 8); + } + } + } + """); + var diagnostics = new java.io.ByteArrayOutputStream(); + String modulePath = + paths.stream() + .map(Path::toString) + .collect(java.util.stream.Collectors.joining(File.pathSeparator)); + assertEquals( + 0, + javax.tools.ToolProvider.getSystemJavaCompiler() + .run( + null, + diagnostics, + diagnostics, + "--module-path", + modulePath, + "-d", + temporary.resolve("classes").toString(), + descriptor.toString(), + example.toString()), + diagnostics.toString()); + Path grpcExample = temporary.resolve("GrpcBoundaryExample.java"); + Files.writeString( + grpcExample, + """ + package example; + import io.grpc.CallOptions; + import io.grpc.ClientCall; + import io.grpc.ManagedChannel; + import io.grpc.MethodDescriptor; + public final class GrpcBoundaryExample { + public static ClientCall call(ManagedChannel channel, + MethodDescriptor method, CallOptions options) { + return channel.newCall(method, options); + } + } + """); + Path mcpExample = temporary.resolve("McpBoundaryExample.java"); + Files.writeString( + mcpExample, + """ + package example; + import java.io.IOException; + import io.modelcontextprotocol.json.McpJsonMapper; + import io.modelcontextprotocol.spec.McpSchema; + public final class McpBoundaryExample { + public static McpSchema.JSONRPCMessage decode(McpJsonMapper mapper, String json) + throws IOException { + return McpSchema.deserializeJsonRpcMessage(mapper, json); + } + } + """); + for (var entry : + Map.of("io.grpc", grpcExample, "io.modelcontextprotocol.sdk.mcp.core", mcpExample) + .entrySet()) { + Files.writeString( + descriptor, "module r1.consumer { requires " + entry.getKey() + "; }"); + diagnostics.reset(); + assertEquals( + 0, + javax.tools.ToolProvider.getSystemJavaCompiler() + .run( + null, + diagnostics, + diagnostics, + "--module-path", + modulePath, + "-d", + temporary.resolve(entry.getKey()).toString(), + descriptor.toString(), + entry.getValue().toString()), + diagnostics.toString()); + } ModuleFinder finder = ModuleFinder.of(paths.toArray(Path[]::new)); var roots = new java.util.HashSet<>(modules.keySet()); roots.add("org.restlet"); @@ -82,6 +177,28 @@ void resolvesPinnedSdkModulesWithoutOfficialNetworkTransports() throws Exception .defineModulesWithOneLoader( configuration, ClassLoader.getPlatformClassLoader()); assertTrue(layer.findModule("org.restlet").isPresent()); + assertFalse( + layer.findModule("org.restlet") + .orElseThrow() + .isExported("org.restlet.engine.stdio")); + Class transportApi = + layer.findLoader("org.restlet").loadClass("org.restlet.util.StreamTransport"); + var bytes = new java.io.ByteArrayOutputStream(); + Object transport = + transportApi + .getConstructor( + java.io.InputStream.class, + java.io.OutputStream.class, + Runnable.class) + .newInstance( + new java.io.ByteArrayInputStream(new byte[0]), + bytes, + (Runnable) () -> {}); + transportApi.getMethod("start").invoke(transport); + transportApi.getMethod("write", byte[].class).invoke(transport, new byte[] {0, 1, 2}); + transportApi.getMethod("stop").invoke(transport); + org.junit.jupiter.api.Assertions.assertArrayEquals( + new byte[] {0, 1, 2}, bytes.toByteArray()); assertTrue( layer.findModule("org.restlet").orElseThrow().getDescriptor().requires().stream() .noneMatch( diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java new file mode 100644 index 0000000000..7a57598463 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java @@ -0,0 +1,109 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.engine.connector; + +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.List; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Server; +import org.restlet.data.Protocol; +import org.restlet.engine.Engine; + +class StdioHelperSelectionTestCase { + + @BeforeEach + void setUp() { + Engine.clearThreadLocalVariables(); + Engine.register(false); + } + + @AfterEach + void tearDown() { + Engine.clearThreadLocalVariables(); + } + + @Test + void rejectsMissingAndAmbiguousHelpersRatherThanSelectingAProtocol() { + assertThrows(IllegalStateException.class, () -> new Client(Protocol.STDIO)); + assertThrows(IllegalStateException.class, () -> new Server(Protocol.STDIO)); + Engine engine = Engine.getInstance(); + engine.getRegisteredClients().add(new BytesClient(null)); + engine.getRegisteredClients().add(new OtherClient(null)); + engine.getRegisteredServers().add(new BytesServer(null)); + engine.getRegisteredServers().add(new OtherServer(null)); + assertThrows(IllegalStateException.class, () -> new Client(Protocol.STDIO)); + assertThrows(IllegalStateException.class, () -> new Server(Protocol.STDIO)); + assertThrows( + IllegalStateException.class, + () -> new Client(new Context(), List.of(Protocol.STDIO), "missing.Helper")); + } + + @Test + void selectsExplicitHelperOrTheUniqueRegisteredProtocol() { + Engine engine = Engine.getInstance(); + engine.getRegisteredClients().add(new BytesClient(null)); + engine.getRegisteredServers().add(new BytesServer(null)); + Context uniqueClient = new Context(); + new Client(uniqueClient, Protocol.STDIO); + assertInstanceOf( + BytesClient.class, uniqueClient.getAttributes().get("org.restlet.engine.helper")); + Context uniqueServer = new Context(); + new Server(uniqueServer, Protocol.STDIO); + assertInstanceOf( + BytesServer.class, uniqueServer.getAttributes().get("org.restlet.engine.helper")); + engine.getRegisteredClients().add(new OtherClient(null)); + engine.getRegisteredServers().add(new OtherServer(null)); + Context explicitClient = new Context(); + new Client(explicitClient, List.of(Protocol.STDIO), OtherClient.class.getCanonicalName()); + assertInstanceOf( + OtherClient.class, explicitClient.getAttributes().get("org.restlet.engine.helper")); + Context explicitServer = new Context(); + new Server( + explicitServer, + List.of(Protocol.STDIO), + null, + -1, + null, + OtherServer.class.getCanonicalName()); + assertInstanceOf( + OtherServer.class, explicitServer.getAttributes().get("org.restlet.engine.helper")); + } + + public static class BytesClient extends ClientHelper { + public BytesClient(Client client) { + super(client); + getProtocols().add(Protocol.STDIO); + } + } + + public static class OtherClient extends BytesClient { + public OtherClient(Client client) { + super(client); + } + } + + public static class BytesServer extends ServerHelper { + public BytesServer(Server server) { + super(server); + getProtocols().add(Protocol.STDIO); + } + } + + public static class OtherServer extends BytesServer { + public OtherServer(Server server) { + super(server); + } + } +} diff --git a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java new file mode 100644 index 0000000000..3ad2ac7740 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java @@ -0,0 +1,101 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.engine.netty; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.DefaultHttpContent; +import io.netty.handler.codec.http.DefaultHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpVersion; +import java.util.ArrayList; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.restlet.engine.connector.HttpServerHelper; + +class Http1LookaheadTestCase { + + @Test + void acceptsExactlyOneMiBThenRejectsOverflowAndReleasesContent() throws Exception { + EmbeddedChannel channel = waitingForResponse(); + var full = new DefaultHttpContent(Unpooled.buffer(1024 * 1024).writeZero(1024 * 1024)); + var overflow = new DefaultHttpContent(Unpooled.buffer(1).writeByte(1)); + try { + channel.writeInbound(full); + assertTrue(channel.isActive()); + assertEquals(1, full.refCnt()); + channel.writeInbound(overflow); + assertFalse(channel.isActive()); + assertEquals(0, full.refCnt()); + assertEquals(0, overflow.refCnt()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void acceptsExactly1024MessagesThenRejectsOverflowAndReleasesContent() throws Exception { + EmbeddedChannel channel = waitingForResponse(); + var content = new ArrayList(); + try { + for (int i = 0; i < 1024; i++) { + var message = new DefaultHttpContent(Unpooled.buffer(1).writeByte(1)); + content.add(message); + channel.writeInbound(message); + } + assertTrue(channel.isActive()); + var overflow = new DefaultHttpContent(Unpooled.buffer(1).writeByte(1)); + content.add(overflow); + channel.writeInbound(overflow); + assertFalse(channel.isActive()); + for (var message : content) { + assertEquals(0, message.refCnt()); + } + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void releasesQueuedContentWhenTheHandlerIsReplacedForAnUpgrade() throws Exception { + EmbeddedChannel channel = waitingForResponse(); + var content = new DefaultHttpContent(Unpooled.buffer(1).writeByte(1)); + try { + channel.writeInbound(content); + channel.pipeline().remove(NettyServerHandler.class); + assertTrue(channel.isActive()); + assertEquals(0, content.refCnt()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @SuppressWarnings("unchecked") + private static EmbeddedChannel waitingForResponse() throws Exception { + var helper = new HttpServerHelper(null); + var handler = new NettyServerHandler(helper, false, StreamKind.HTTP1); + var channel = new EmbeddedChannel(handler); + var call = + new NettyServerCall( + helper, + channel, + StreamKind.HTTP1, + new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/"), + new ChannelInputStream(channel, 0), + false); + var field = NettyServerHandler.class.getDeclaredField("activeCall"); + field.setAccessible(true); + ((AtomicReference) field.get(handler)).set(call); + return channel; + } +} diff --git a/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java new file mode 100644 index 0000000000..7db46e2f9f --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java @@ -0,0 +1,414 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.util; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.OutputStream; +import java.nio.file.Path; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class StreamTransportTestCase { + + @Test + void boundsOpaqueReadsAndWritesWithoutClosingBorrowedStreams() throws Exception { + var input = new TrackedInput(new byte[] {0, 10, (byte) 255, 0, 9}); + var output = new TrackedOutput(); + try (var transport = new StreamTransport(input, output, () -> {})) { + transport.setMaxOperationBytes(4); + assertThrows(IOException.class, () -> transport.read(new byte[8], 0, 8)); + transport.start(); + assertEquals(0, input.reads.get(), "Start must not read or prefetch protocol bytes"); + assertThrows(IllegalStateException.class, () -> transport.setMaxOperationBytes(5)); + byte[] bytes = new byte[8]; + assertEquals(4, transport.read(bytes, 0, bytes.length)); + assertArrayEquals(new byte[] {0, 10, (byte) 255, 0}, Arrays.copyOf(bytes, 4)); + assertEquals(1, transport.read(bytes, 0, bytes.length)); + assertEquals(-1, transport.read(bytes, 0, bytes.length)); + transport.write(new byte[] {0, 1, 10, (byte) 255}); + assertThrows(IllegalArgumentException.class, () -> transport.write(new byte[5])); + assertArrayEquals(new byte[] {0, 1, 10, (byte) 255}, output.toByteArray()); + assertEquals(0, transport.getStderrTail().length); + } + assertFalse(input.closed); + assertFalse(output.closed); + assertTrue(output.flushes.get() >= 2); + } + + @Test + void serializesCompleteWritesFromConcurrentCallers() throws Exception { + var output = + new ByteArrayOutputStream() { + @Override + public void write(byte[] bytes) { + for (byte value : bytes) { + write(value); + Thread.yield(); + } + } + }; + try (var transport = + new StreamTransport( + new ByteArrayInputStream(new byte[0]), output, () -> {}); + var executor = Executors.newVirtualThreadPerTaskExecutor()) { + transport.start(); + var futures = new java.util.ArrayList>(); + for (int i = 0; i < 16; i++) { + byte value = (byte) i; + futures.add( + executor.submit( + () -> { + byte[] sequence = new byte[1024]; + Arrays.fill(sequence, value); + transport.write(sequence); + return null; + })); + } + for (var future : futures) { + future.get(5, TimeUnit.SECONDS); + } + byte[] bytes = output.toByteArray(); + assertEquals(16 * 1024, bytes.length); + for (int offset = 0; offset < bytes.length; offset += 1024) { + byte value = bytes[offset]; + for (int i = offset; i < offset + 1024; i++) { + assertEquals(value, bytes[i]); + } + } + } + } + + @Test + void stopsBlockedBorrowedInputWithoutClosingIt() throws Exception { + CountDownLatch reading = new CountDownLatch(1); + AtomicInteger cancelled = new AtomicInteger(); + InputStream input = + new InputStream() { + @Override + public int read() throws IOException { + reading.countDown(); + try { + new CountDownLatch(1).await(); + throw new AssertionError("Unreachable"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Cancelled read"); + } + } + + @Override + public void close() { + fail("Borrowed input must not be closed"); + } + }; + var output = new TrackedOutput(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor(); + var transport = new StreamTransport(input, output, cancelled::incrementAndGet)) { + transport.start(); + var read = + executor.submit( + () -> + assertThrows( + IOException.class, + () -> transport.read(new byte[1], 0, 1))); + assertTrue(reading.await(5, TimeUnit.SECONDS)); + transport.stop(); + read.get(5, TimeUnit.SECONDS); + assertEquals(1, cancelled.get()); + assertFalse(transport.isStarted()); + assertThrows(IOException.class, () -> transport.write(new byte[1])); + assertThrows(IllegalStateException.class, transport::start); + } + assertFalse(output.closed); + } + + @Test + void stopsBlockedAndQueuedWritesWithoutClosingBorrowedOutput() throws Exception { + CountDownLatch writing = new CountDownLatch(1); + OutputStream output = + new OutputStream() { + @Override + public void write(int value) throws IOException { + writing.countDown(); + try { + new CountDownLatch(1).await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Cancelled write"); + } + } + + @Override + public void close() { + fail("Borrowed output must not be closed"); + } + }; + try (var executor = Executors.newVirtualThreadPerTaskExecutor(); + var transport = + new StreamTransport( + new ByteArrayInputStream(new byte[0]), output, () -> {})) { + transport.start(); + var active = + executor.submit( + () -> + assertThrows( + IOException.class, + () -> transport.write(new byte[] {1}))); + assertTrue(writing.await(5, TimeUnit.SECONDS)); + var queued = + executor.submit( + () -> + assertThrows( + IOException.class, + () -> transport.write(new byte[] {2}))); + transport.stop(); + active.get(5, TimeUnit.SECONDS); + queued.get(5, TimeUnit.SECONDS); + } + } + + @Test + void reportsUncancellableBorrowedIoInsteadOfClaimingSuccessfulStop() throws Exception { + CountDownLatch reading = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + InputStream input = + new InputStream() { + @Override + public int read() { + reading.countDown(); + boolean done = false; + while (!done) { + try { + release.await(); + done = true; + } catch (InterruptedException e) { + // Deliberately models an input that does not support interruption. + } + } + return -1; + } + }; + var transport = new StreamTransport(input, new ByteArrayOutputStream(), () -> {}); + transport.setProcessExitGrace(Duration.ZERO); + transport.setShutdownTimeout(Duration.ofMillis(100)); + transport.start(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var read = executor.submit(() -> transport.read(new byte[1], 0, 1)); + try { + assertTrue(reading.await(5, TimeUnit.SECONDS)); + assertTimeoutPreemptively( + Duration.ofSeconds(2), + () -> assertThrows(IOException.class, transport::stop)); + assertFalse(transport.isStarted()); + } finally { + release.countDown(); + } + assertEquals(-1, read.get(5, TimeUnit.SECONDS)); + transport.stop(); + } + } + + @Test + void propagatesIoAndCancellationFailures() throws Exception { + var transport = + new StreamTransport( + new InputStream() { + @Override + public int read() throws IOException { + throw new IOException("input-failure"); + } + }, + new OutputStream() { + @Override + public void write(int value) throws IOException { + throw new IOException("output-failure"); + } + }, + () -> { + throw new IllegalStateException("cancel-failure"); + }); + transport.start(); + assertEquals( + "input-failure", + assertThrows(IOException.class, () -> transport.read(new byte[1], 0, 1)) + .getMessage()); + assertEquals( + "output-failure", + assertThrows(IOException.class, () -> transport.write(new byte[1])).getMessage()); + assertInstanceOf( + IllegalStateException.class, + assertThrows(IOException.class, transport::stop).getCause().getCause()); + } + + @Test + void launchesOwnedProcessAndDrainsBoundedStderrSeparately() throws Exception { + var transport = new StreamTransport(command("echo")); + transport.setMaxStderrBytes(257); + transport.start(); + long pid = transport.getProcessId(); + try { + byte[] bytes = new byte[] {0, 10, 13, (byte) 255, 42}; + transport.write(bytes); + byte[] received = new byte[bytes.length]; + int offset = 0; + while (offset < received.length) { + int count = transport.read(received, offset, received.length - offset); + assertTrue(count > 0); + offset += count; + } + assertArrayEquals(bytes, received); + assertEquals(257, transport.getStderrTail().length); + assertTrue(Arrays.equals(new byte[257], transport.getStderrTail())); + } finally { + transport.stop(); + } + assertFalse(ProcessHandle.of(pid).map(ProcessHandle::isAlive).orElse(false)); + } + + @Test + void terminatesOwnedProcessWhileStdoutIsBlocked() throws Exception { + var transport = new StreamTransport(command("idle")); + transport.start(); + long pid = transport.getProcessId(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var read = + executor.submit( + () -> { + try { + return transport.read(new byte[1], 0, 1); + } catch (IOException e) { + assertFalse(transport.isStarted()); + return -1; + } + }); + transport.stop(); + assertEquals(-1, read.get(5, TimeUnit.SECONDS)); + } + assertFalse(ProcessHandle.of(pid).map(ProcessHandle::isAlive).orElse(false)); + } + + @Test + void surfacesProcessExitAndStartupFailure() throws Exception { + var failed = new StreamTransport(command("fail")); + failed.start(); + try { + IOException error = + assertThrows(IOException.class, () -> failed.read(new byte[1], 0, 1)); + assertTrue(error.getMessage().contains("23")); + } finally { + assertThrows(IOException.class, failed::stop); + } + var unread = new StreamTransport(command("fail")); + unread.start(); + ProcessHandle.of(unread.getProcessId()).orElseThrow().onExit().get(5, TimeUnit.SECONDS); + assertTrue( + assertThrows(IOException.class, unread::stop) + .getCause() + .getMessage() + .contains("23")); + try (var transport = + new StreamTransport( + List.of(Path.of("missing-r1-executable").toAbsolutePath().toString()))) { + assertThrows(IOException.class, transport::start); + } + assertThrows(IllegalArgumentException.class, () -> new StreamTransport(List.of())); + } + + private static List command(String mode) { + String executable = + Path.of( + System.getProperty("java.home"), + "bin", + System.getProperty("os.name").toLowerCase().contains("win") + ? "java.exe" + : "java") + .toString(); + return List.of( + executable, + "-cp", + System.getProperty( + "surefire.test.class.path", System.getProperty("java.class.path")), + Child.class.getName(), + mode); + } + + public static class Child { + public static void main(String[] args) throws Exception { + if ("fail".equals(args[0])) { + System.exit(23); + } + if ("idle".equals(args[0])) { + Thread.sleep(60000); + return; + } + System.err.write(new byte[200000]); + System.err.flush(); + byte[] bytes = new byte[1024]; + int count; + while ((count = System.in.read(bytes)) != -1) { + System.out.write(bytes, 0, count); + System.out.flush(); + } + } + } + + private static final class TrackedInput extends ByteArrayInputStream { + private boolean closed; + private final AtomicInteger reads = new AtomicInteger(); + + private TrackedInput(byte[] bytes) { + super(bytes); + } + + @Override + public synchronized int read(byte[] buffer, int offset, int length) { + reads.incrementAndGet(); + return super.read(buffer, offset, length); + } + + @Override + public void close() { + closed = true; + } + } + + private static final class TrackedOutput extends ByteArrayOutputStream { + private boolean closed; + private final AtomicInteger flushes = new AtomicInteger(); + + @Override + public void close() { + closed = true; + } + + @Override + public void flush() { + flushes.incrementAndGet(); + } + } +} diff --git a/readme.md b/readme.md index 1d4555f414..5ddb7d92be 100644 --- a/readme.md +++ b/readme.md @@ -12,6 +12,130 @@ Restlet applications run directly in JVMs with minimal dependencies, resulting i ![](https://avatars.githubusercontent.com/u/374452?s=400&u=191f2eada671b2a07f61782cf638bdda0ccb6d3f&v=4) +## R1 transport foundations + +R1 supplies protocol-neutral mechanics, not production MCP, gRPC or Smithy adapters. +The extensions remain independent and retain their official SDK contracts. No RPC superclass, +cross-protocol message model, dispatcher or status mapping is introduced. + +### Supported byte transport and ownership + +`org.restlet.util.StreamTransport` is the supported core seam. Its implementation in +`org.restlet.engine.stdio` is not exported. It needs no protocol SDK or Netty types: + +```java +try (var transport = new StreamTransport(input, output, cancelIo)) { + transport.setMaxOperationBytes(65536); + transport.start(); + transport.write(new byte[] {0, 1, 2}); + int count = transport.read(buffer, 0, buffer.length); +} +``` + +The streams and cancellation action are supplied explicitly. Borrowed streams, including +standard streams, are **never closed**; output is flushed on writes and stop. The cancellation +action must promptly release blocked borrowed I/O without closing global standard streams. +Interruption is also attempted, but an arbitrary `InputStream` may ignore it. If the owner's +cancellation does not release I/O, stop reports a timeout; cleanup remains pending until the +owner releases it. This is not a guarantee of safely interrupting arbitrary `System.in` reads. +No global stream is redirected. + +`new StreamTransport(List.of(executable, argument))` instead owns the subprocess and its pipes. +No shell or URI-derived command is used. Stop closes owned pipes and terminates only that +process, not a process tree. Stderr is drained independently into a bounded tail; protocol +output is never mixed with it. Startup, I/O, unexpected nonzero exit, stderr-drain and shutdown +failures are surfaced. The stderr tail remains available after stop. + +| Configuration | Default | Contract | +|---|---|---| +| `maxOperationBytes` | 65536 | Positive; reads are capped, oversized writes rejected before I/O | +| `maxStderrBytes` | 8192 | Positive bounded tail for owned subprocess stderr | +| `shutdownTimeout` | 5 seconds | Positive caller wait bound, greater than exit grace | +| `processExitGrace` | 250 milliseconds | Nonnegative and below shutdown timeout; wait before destroy and again before forced termination | + +Configuration is fixed after start/stop. Reads are synchronous, demand-driven and serialized: +there is no protocol reader, prefetch or input queue. Writes are serialized and flushed as +opaque sequences, with no internal task queue. Extensions must bound their own callers, +message assembly, SDK demand and pending calls; a read chunk is not a protocol message. +Stopped transports reject new work and cannot restart. + +`Protocol.STDIO` identifies bytes/pipes, **not MCP**. Core installs no default STDIO helper. +Connector construction requires one uniquely registered supporting helper class or an +explicit supporting helper class; missing/ambiguous choices fail rather than taking the +first protocol implementation. MCP framing, SDK correlation, routing and helpers belong to R2. + +### HTTP lifecycle and connection decisions + +`Request.getDisconnectSignal()` returns a read-only stage completed once with the first +abnormal transport cause, including explicit client cancellation. Normal completion does not +complete it. Netty reports it independently of application entity I/O, including an idle +server whose entire request body has been consumed. Completion runs off the connector event +loop; expensive continuations should still use an application executor. Wrappers delegate +the signal; request copies have independent lifecycle state. + +An abort before connector registration is remembered, although `abort()` still returns false +until an active action accepts it. Connector abort actions must be idempotent. Use a new +request after cancellation. Connection/send failures also complete the signal. + +HTTP/1.1 performs a read after the body to observe peer closure. Pipelined requests from that +read wait until the preceding response is flushed, preserving response order. This decoded +lookahead is limited to 1024 messages and 1 MiB of content/header storage accounting; overflow +reports a transport failure and closes the connection. Reads pause when lookahead is queued, +so notification remains subject to transport back-pressure. Upgrade/CONNECT requests retain +their existing read handoff to the replacement protocol. HTTP/2 and experimental HTTP/3 retain +their stream lifecycle and flow control; this is not a new SDK flow-control implementation. + +R1 deliberately adds **no physical-connection control API**. The destination URI remains the +source of HTTP/TLS authority, DNS resolution and pool identity in existing connectors. +Extensions must not pretend that replacing `Host` implements separate SNI/resolved-address +selection. gRPC resolver-address/authority divergence, physical connection pinning, per-connection +admission, GOAWAY, PING and SDK keepalive must be explicitly rejected if requested by a future +adapter until implemented and tested. Start with URI-origin routing and adapter-wide limits; +SDK streaming/readiness conformance remains R3/R4/R6 work. + +### Independent SDK integration contracts + +| Extension delivery | SDK-owned contract | Restlet integration / ownership / finite limits | +|---|---|---| +| MCP (R2) | SDK 2.1.0 records, codecs, dispatch, correlation and supported revisions | Borrow injected HTTP connectors or `StreamTransport`; close adapter sessions/tasks, not borrowed connectors. Bound request/frame bytes, pending calls, sessions, replay history and SDK queues before admission. Profile compatibility remains explicit; R0 did not authorize Ikanos migration. | +| gRPC (R3/R4/R6) | grpc-java 1.85.1 `ManagedChannel`, `ClientCall`, service definitions, SDK framing/status/metadata | Borrow configured HTTP/2 connectors; own logical SDK runtimes/call workers. Pin internal SPI use. Bound inbound/outbound message bytes, active calls and queued bytes; map receive demand/readiness before claiming streaming support. URI-origin routing only initially. | +| Smithy (R5/R6) | Candidate smithy-java 1.7.0 native client transport, modeled operations/errors, signing | API remains unvalidated: prototype before adding adapter classes. Borrow Restlet HTTP connectors, preserve SDK authority/signing, bound bodies/concurrency and release owned work. Experimental server remains deferred. | + +These are delivery contracts, **not available adapter constructors or validated Smithy APIs**. +Each delivery must choose and test finite protocol limits rather than inheriting an unlimited +queue. Core transport chunks do not substitute for SDK message limits. Borrowed connector +instances remain running after adapter shutdown; only explicitly owned resources may be stopped. + +Initial extension configuration is defined independently (positive limits fixed before start): +MCP uses 1 MiB request/frame bytes, 128 pending calls, 64 sessions, and 1 MiB queued bytes per +session, with replay disabled unless a finite history is explicitly configured. gRPC uses +4 MiB inbound and outbound messages, 128 adapter-wide active calls, and at most 4 MiB queued +bytes per call, including any frame awaiting SDK demand. Smithy's provisional prototype uses +1 MiB buffered bodies and 128 active calls, subject to R5 verification rather than a supported +SDK claim. All over-limit admissions fail through the selected protocol's own error contract; +they are never truncated, silently dropped or mapped through a common RPC error. Deadlines +remain native SDK settings. These are R1 design defaults to enforce and test when each extension +lands, not limits already enforced by the R0 fixtures or by core `StreamTransport`. + +The regression suite compiles a separate named-module consumer of the public byte seam and +checks that its implementation remains unexported and core has no MCP/gRPC SDK requirements. +The inherited R0 fixtures compile and exercise the independent native MCP/gRPC contracts; +they remain test-only. Run the foundations plus relevant connector/SDK regressions: + +```powershell +.\mvnw.cmd -pl org.restlet '-Dtest=StreamTransportTestCase,RequestLifecycleTestCase,WrapperRequestTestCase,StdioHelperSelectionTestCase,RpcSdkBoundaryTestCase,HttpDisconnectTestCase,NettyConnectionUpgradeTestCase,GrpcSdkTransportSpikeTestCase,GrpcSdkServerHostSpikeTestCase,McpSdkTransportSpikeTestCase,McpStreamableSdkTransportSpikeTestCase,*Http*TestCase,*EventStream*TestCase' test +``` + +Use the opt-in Ikanos properties below to include the live differential rather than skip it. + +R1 closing validation on Java 25: the live-enabled selector above passed **134 tests, zero +failures/errors/skips**, including HTTP/3, TLS, SSE, protocol upgrades, exact lookahead limits, +STDIO ownership/shutdown and independent named-module compilation. The full core run executed +2257 tests with one shutdown wall-clock tolerance failure (1.715 s versus 1 s +/- 200 ms). +The same test failed on the unchanged R0 baseline (1.635 s); the five-case shutdown class then +passed in isolation on R1. No timing assertion was relaxed. This is a recorded baseline timing +limitation, not a clean full-suite claim. + ## R0 SDK interoperability probes The MCP and gRPC interoperability spikes are **test-only**, not production RPC extensions. From 6d97689bf1d5ec5d5ee00f6d9f0c16b6b11300ac Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:23:05 -0400 Subject: [PATCH 2/9] Refine R1 cleanup and pass Javadoc and static checks Separate cleanup phases, preserve interruption causes, retain duration overflow validation, and avoid catching fatal errors. Use supported Javadoc markup and deterministic child blocking. Revalidated the 134-case live suite and strict Javadoc generation without relaxing safeguards. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../src/main/java/org/restlet/Request.java | 12 +- .../restlet/engine/netty/NettyServerCall.java | 38 +++--- .../engine/stdio/StreamTransportHelper.java | 112 ++++++++++-------- .../StdioHelperSelectionTestCase.java | 4 +- .../restlet/util/StreamTransportTestCase.java | 12 +- readme.md | 2 + 6 files changed, 101 insertions(+), 79 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/Request.java b/org.restlet/src/main/java/org/restlet/Request.java index b21621ba1b..99f8022594 100644 --- a/org.restlet/src/main/java/org/restlet/Request.java +++ b/org.restlet/src/main/java/org/restlet/Request.java @@ -323,10 +323,11 @@ private void copyClientInfo(final Request request) { * Ask the connector to attempt to abort the related network connection, for example immediately * closing the socket. * + *

An attempt made before connector registration is remembered and applied when an active + * action is installed. It still returns false until an active action accepts it. Use a new + * Request for a new exchange after cancellation. + * * @return True if the request was aborted. - * @implNote An attempt made before connector registration is remembered and applied when an - * active action is installed. It still returns false until an active action accepts it. Use - * a new Request for a new exchange after cancellation. */ public boolean abort() { abortRequested.set(true); @@ -338,9 +339,10 @@ public boolean abort() { * Sets the action used to abort an active client request. Connector implementations configure * this while a request is in flight and clear it when the exchange ends. * + *

The action must be idempotent: concurrent registration and abort attempts can invoke it + * more than once. + * * @param action The abort action, or null when no exchange is active. - * @implNote The action must be idempotent: concurrent registration and abort attempts can - * invoke it more than once. */ public void setAbortAction(BooleanSupplier action) { this.abortAction = action; diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index 09bca09ca7..f76e2ad637 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -735,24 +735,7 @@ void complete() { requestBody.close(); if (!kind.isStream()) { - if (keepAlive) { - // Ready for the next request on this connection. - future.addListener( - f -> { - if (f.isSuccess()) { - var context = - channel.pipeline().context(NettyServerHandler.class); - if (context != null) { - ((NettyServerHandler) context.handler()) - .responseComplete(context, this); - } - } else { - channel.close(); - } - }); - } else { - future.addListener(ChannelFutureListener.CLOSE); - } + completeHttp1(future, keepAlive); } } finally { if (eventStreamPermit) { @@ -763,6 +746,25 @@ void complete() { } } + private void completeHttp1(ChannelFuture future, boolean keepAlive) { + if (!keepAlive) { + future.addListener(ChannelFutureListener.CLOSE); + return; + } + future.addListener( + f -> { + if (!f.isSuccess()) { + channel.close(); + return; + } + var context = channel.pipeline().context(NettyServerHandler.class); + if (context != null + && context.handler() instanceof NettyServerHandler handler) { + handler.responseComplete(context, this); + } + }); + } + static void setTrailerNames( io.netty.handler.codec.http.HttpHeaders headers, Series

trailers) { String names = diff --git a/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java index 55543451c2..584303c485 100644 --- a/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java +++ b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java @@ -95,8 +95,7 @@ public synchronized void setShutdownTimeout(Duration timeout) { if (timeout.isNegative() || timeout.isZero() || timeout.compareTo(processExitGrace) <= 0) { throw new IllegalArgumentException("shutdownTimeout must exceed processExitGrace"); } - timeout.toNanos(); - shutdownTimeout = timeout; + shutdownTimeout = Duration.ofNanos(timeout.toNanos()); } public synchronized Duration getProcessExitGrace() { @@ -109,8 +108,7 @@ public synchronized void setProcessExitGrace(Duration grace) { throw new IllegalArgumentException( "processExitGrace must be nonnegative and below shutdownTimeout"); } - grace.toNanos(); - processExitGrace = grace; + processExitGrace = Duration.ofNanos(grace.toNanos()); } private void requireConfigurable() { @@ -135,7 +133,7 @@ public synchronized void start() throws IOException { output = process.getOutputStream(); try { stderrReader.start(); - } catch (RuntimeException | Error e) { + } catch (RuntimeException e) { stopped = true; process.destroyForcibly(); IOException closeFailure = closePipes(null); @@ -190,8 +188,7 @@ public int read(byte[] buffer, int offset, int length) throws IOException { } return count; } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new InterruptedIOException("Interrupted awaiting subprocess exit"); + throw interrupted("Interrupted awaiting subprocess exit", e); } finally { leave(); } @@ -223,11 +220,17 @@ private static void lock(ReentrantLock lock) throws InterruptedIOException { try { lock.lockInterruptibly(); } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new InterruptedIOException("Interrupted awaiting stream I/O"); + throw interrupted("Interrupted awaiting stream I/O", e); } } + private static InterruptedIOException interrupted(String message, InterruptedException cause) { + Thread.currentThread().interrupt(); + var failure = new InterruptedIOException(message); + failure.initCause(cause); + return failure; + } + private synchronized boolean isStopped() { return stopped; } @@ -287,8 +290,7 @@ public void stop() throws IOException { try { completion.get(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS); } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new InterruptedIOException("Interrupted stopping stream transport"); + throw interrupted("Interrupted stopping stream transport", e); } catch (TimeoutException e) { throw new IOException( "Stream transport shutdown timed out; borrowed I/O must support cancellation", @@ -300,49 +302,13 @@ public void stop() throws IOException { private void cleanup() { try { - IOException failure = null; synchronized (this) { operations.forEach(Thread::interrupt); } - if (process != null) { - boolean exited = process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS); - if (exited && process.exitValue() != 0) { - failure = new IOException("Subprocess exited with code " + process.exitValue()); - } - if (!exited) { - process.destroy(); - if (!process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS)) { - process.destroyForcibly(); - } - } - process.waitFor(); - failure = closePipes(failure); - stderrReader.join(); - } else { - try { - cancelIo.run(); - } catch (RuntimeException e) { - failure = new IOException("Borrowed I/O cancellation failed", e); - } - } - synchronized (this) { - while (!operations.isEmpty()) { - wait(); - } - if (stderrFailure != null) { - failure = - append( - failure, - new IOException( - "Subprocess stderr drain failed", stderrFailure)); - } - } + IOException failure = process != null ? stopProcess() : cancelBorrowedIo(); + failure = awaitOperations(failure); if (process == null) { - try { - output.flush(); - } catch (IOException e) { - failure = append(failure, e); - } + failure = flushBorrowedOutput(failure); } if (failure != null) { throw failure; @@ -360,6 +326,52 @@ private void cleanup() { } } + private IOException stopProcess() throws InterruptedException { + boolean exited = process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS); + IOException failure = + exited && process.exitValue() != 0 + ? new IOException("Subprocess exited with code " + process.exitValue()) + : null; + if (!exited) { + process.destroy(); + if (!process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS)) { + process.destroyForcibly(); + } + } + process.waitFor(); + failure = closePipes(failure); + stderrReader.join(); + return failure; + } + + private IOException cancelBorrowedIo() { + try { + cancelIo.run(); + return null; + } catch (RuntimeException e) { + return new IOException("Borrowed I/O cancellation failed", e); + } + } + + private synchronized IOException awaitOperations(IOException failure) + throws InterruptedException { + while (!operations.isEmpty()) { + wait(); + } + return stderrFailure == null + ? failure + : append(failure, new IOException("Subprocess stderr drain failed", stderrFailure)); + } + + private IOException flushBorrowedOutput(IOException failure) { + try { + output.flush(); + return failure; + } catch (IOException e) { + return append(failure, e); + } + } + private IOException closePipes(IOException failure) { for (java.io.Closeable pipe : List.of(input, output)) { try { diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java index 7a57598463..f6043afb19 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java @@ -45,9 +45,11 @@ void rejectsMissingAndAmbiguousHelpersRatherThanSelectingAProtocol() { engine.getRegisteredServers().add(new OtherServer(null)); assertThrows(IllegalStateException.class, () -> new Client(Protocol.STDIO)); assertThrows(IllegalStateException.class, () -> new Server(Protocol.STDIO)); + Context missing = new Context(); + List protocols = List.of(Protocol.STDIO); assertThrows( IllegalStateException.class, - () -> new Client(new Context(), List.of(Protocol.STDIO), "missing.Helper")); + () -> new Client(missing, protocols, "missing.Helper")); } @Test diff --git a/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java index 7db46e2f9f..cb3e073268 100644 --- a/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java +++ b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java @@ -115,7 +115,7 @@ public int read() throws IOException { try { new CountDownLatch(1).await(); throw new AssertionError("Unreachable"); - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); throw new InterruptedIOException("Cancelled read"); } @@ -157,7 +157,7 @@ public void write(int value) throws IOException { writing.countDown(); try { new CountDownLatch(1).await(); - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); throw new InterruptedIOException("Cancelled write"); } @@ -206,7 +206,7 @@ public int read() { try { release.await(); done = true; - } catch (InterruptedException e) { + } catch (InterruptedException _) { // Deliberately models an input that does not support interruption. } } @@ -301,7 +301,7 @@ void terminatesOwnedProcessWhileStdoutIsBlocked() throws Exception { () -> { try { return transport.read(new byte[1], 0, 1); - } catch (IOException e) { + } catch (IOException _) { assertFalse(transport.isStarted()); return -1; } @@ -363,7 +363,9 @@ public static void main(String[] args) throws Exception { System.exit(23); } if ("idle".equals(args[0])) { - Thread.sleep(60000); + if (!new CountDownLatch(1).await(1, TimeUnit.MINUTES)) { + throw new IllegalStateException("Idle test child was not stopped"); + } return; } System.err.write(new byte[200000]); diff --git a/readme.md b/readme.md index 5ddb7d92be..382c414a98 100644 --- a/readme.md +++ b/readme.md @@ -135,6 +135,8 @@ STDIO ownership/shutdown and independent named-module compilation. The full core The same test failed on the unchanged R0 baseline (1.635 s); the five-case shutdown class then passed in isolation on R1. No timing assertion was relaxed. This is a recorded baseline timing limitation, not a clean full-suite claim. +The final selector was rerun together with strict `javadoc:javadoc` generation after the +cleanup/quality-check refinements; both passed. No quality gate or workflow was weakened. ## R0 SDK interoperability probes From 75aba2385459fefa31254f003ae4789533fa94de Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Wed, 21 Oct 2026 21:40:22 -0400 Subject: [PATCH 3/9] Publish HTTP completion before releasing body consumers Reproduce the EOF-before-terminal-publication gap with a held exchange lock, then publish completion before waking readers. Extend lifecycle coverage with immediate closes on empty 401/403 responses and the existing authorization cases. The expanded live regression selector passes 145 cases together with strict Javadoc generation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 2 + .../restlet/engine/netty/NettyClientCall.java | 8 ++-- .../connector/HttpDisconnectTestCase.java | 36 +++++++++++++++ .../engine/netty/Http1LookaheadTestCase.java | 46 +++++++++++++++++++ readme.md | 12 +++-- 5 files changed, 96 insertions(+), 8 deletions(-) diff --git a/changes.md b/changes.md index eabd5778c6..c4484e2572 100644 --- a/changes.md +++ b/changes.md @@ -90,6 +90,8 @@ Changes log - Added `Request.getDisconnectSignal()` and connector notification of abnormal termination independently of entity I/O. Abort requests made before connector registration are remembered; wrappers delegate lifecycle and copies remain independent. + - HTTP clients publish response completion before exposing body EOF to consumers, + so an empty terminal response is not exposed while its exchange still appears active. - HTTP/1.1 observes idle disconnects after request-body consumption while preserving ordered pipelining with bounded decoded lookahead (1024 messages / 1 MiB accounted content/header storage). Overflow closes the connection; protocol upgrades retain diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java index be992f7977..99b0188d8d 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java @@ -822,10 +822,6 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) { currentResponse.getTrailers().add(name, trailer.getValue()); } } - ChannelInputStream stream = this.body; - if (stream != null) { - stream.end(); - } Representation emptyEntity; lock.lock(); try { @@ -835,6 +831,10 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) { } finally { lock.unlock(); } + ChannelInputStream stream = this.body; + if (stream != null) { + stream.end(); + } if (emptyEntity != null) { emptyEntity.release(); } diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java index 8594eedd4c..b2022c2515 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java @@ -32,6 +32,7 @@ import org.restlet.data.MediaType; import org.restlet.data.Method; import org.restlet.data.Protocol; +import org.restlet.data.Status; import org.restlet.engine.Engine; import org.restlet.representation.StringRepresentation; @@ -209,4 +210,39 @@ public void handle(Request request, Response response) { releaseFirst.countDown(); } } + + @ParameterizedTest + @ValueSource(strings = {"HTTP1_1", "HTTP2"}) + void preservesEmptyErrorStatusWhenTheClientClosesImmediately(String mode) throws Exception { + start( + mode, + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setStatus( + request.getResourceRef().getPath().equals("/401") + ? Status.CLIENT_ERROR_UNAUTHORIZED + : Status.CLIENT_ERROR_FORBIDDEN); + } + }); + for (int i = 0; i < 64; i++) { + Status expected = + i % 2 == 0 ? Status.CLIENT_ERROR_UNAUTHORIZED : Status.CLIENT_ERROR_FORBIDDEN; + Request request = + new Request( + Method.GET, + "http://localhost:" + + server.getActualPort() + + "/" + + expected.getCode()); + Response response = client.handle(request); + client.stop(); + assertEquals( + expected, + response.getStatus(), + () -> "Terminal response changed: " + response.getStatus().getThrowable()); + assertFalse(request.getDisconnectSignal().toCompletableFuture().isDone()); + client.start(); + } + } } diff --git a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java index 3ad2ac7740..1cf6448c6b 100644 --- a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java @@ -10,6 +10,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import io.netty.buffer.Unpooled; @@ -19,8 +20,14 @@ import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpVersion; import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; import org.junit.jupiter.api.Test; +import org.restlet.data.Reference; import org.restlet.engine.connector.HttpServerHelper; class Http1LookaheadTestCase { @@ -80,6 +87,45 @@ void releasesQueuedContentWhenTheHandlerIsReplacedForAnUpgrade() throws Exceptio } } + @Test + void publishesClientCompletionBeforeReleasingAnEmptyBodyConsumer() throws Exception { + var trailers = new io.netty.handler.codec.http.DefaultHttpHeaders(); + new NettyClientCall(null, "GET", new Reference("http://localhost/"), "/", false) + .onResponseComplete(trailers); + var call = new NettyClientCall(null, "GET", new Reference("http://localhost/"), "/", false); + var body = new ChannelInputStream(null, 0); + var bodyField = NettyClientCall.class.getDeclaredField("body"); + bodyField.setAccessible(true); + bodyField.set(call, body); + var lockField = NettyClientCall.class.getDeclaredField("lock"); + lockField.setAccessible(true); + ReentrantLock lock = (ReentrantLock) lockField.get(call); + CountDownLatch completing = new CountDownLatch(1); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + java.util.concurrent.Callable reader = body::read; + var read = executor.submit(reader); + lock.lock(); + java.util.concurrent.Future completion; + try { + completion = + executor.submit( + () -> { + completing.countDown(); + call.onResponseComplete(trailers); + }); + assertTrue(completing.await(5, TimeUnit.SECONDS)); + assertThrows( + TimeoutException.class, + () -> read.get(1, TimeUnit.SECONDS), + "EOF must not be observable while terminal state publication is blocked"); + } finally { + lock.unlock(); + } + completion.get(5, TimeUnit.SECONDS); + assertEquals(-1, read.get(5, TimeUnit.SECONDS)); + } + } + @SuppressWarnings("unchecked") private static EmbeddedChannel waitingForResponse() throws Exception { var helper = new HttpServerHelper(null); diff --git a/readme.md b/readme.md index 382c414a98..adf7ba64d4 100644 --- a/readme.md +++ b/readme.md @@ -76,6 +76,8 @@ the signal; request copies have independent lifecycle state. An abort before connector registration is remembered, although `abort()` still returns false until an active action accepts it. Connector abort actions must be idempotent. Use a new request after cancellation. Connection/send failures also complete the signal. +Client completion state is published before waking response-body consumers at EOF, so an +empty completed response is not exposed while its exchange still appears active. HTTP/1.1 performs a read after the body to observe peer closure. Pipelined requests from that read wait until the preceding response is flushed, preserving response order. This decoded @@ -123,20 +125,22 @@ The inherited R0 fixtures compile and exercise the independent native MCP/gRPC c they remain test-only. Run the foundations plus relevant connector/SDK regressions: ```powershell -.\mvnw.cmd -pl org.restlet '-Dtest=StreamTransportTestCase,RequestLifecycleTestCase,WrapperRequestTestCase,StdioHelperSelectionTestCase,RpcSdkBoundaryTestCase,HttpDisconnectTestCase,NettyConnectionUpgradeTestCase,GrpcSdkTransportSpikeTestCase,GrpcSdkServerHostSpikeTestCase,McpSdkTransportSpikeTestCase,McpStreamableSdkTransportSpikeTestCase,*Http*TestCase,*EventStream*TestCase' test +.\mvnw.cmd -pl org.restlet '-Dtest=StreamTransportTestCase,RequestLifecycleTestCase,WrapperRequestTestCase,StdioHelperSelectionTestCase,RpcSdkBoundaryTestCase,HttpDisconnectTestCase,NettyConnectionUpgradeTestCase,SecurityTestCase,GrpcSdkTransportSpikeTestCase,GrpcSdkServerHostSpikeTestCase,McpSdkTransportSpikeTestCase,McpStreamableSdkTransportSpikeTestCase,*Http*TestCase,*EventStream*TestCase' test ``` Use the opt-in Ikanos properties below to include the live differential rather than skip it. -R1 closing validation on Java 25: the live-enabled selector above passed **134 tests, zero +R1 closing validation on Java 25: the live-enabled selector above passed **145 tests, zero failures/errors/skips**, including HTTP/3, TLS, SSE, protocol upgrades, exact lookahead limits, STDIO ownership/shutdown and independent named-module compilation. The full core run executed 2257 tests with one shutdown wall-clock tolerance failure (1.715 s versus 1 s +/- 200 ms). The same test failed on the unchanged R0 baseline (1.635 s); the five-case shutdown class then passed in isolation on R1. No timing assertion was relaxed. This is a recorded baseline timing limitation, not a clean full-suite claim. -The final selector was rerun together with strict `javadoc:javadoc` generation after the -cleanup/quality-check refinements; both passed. No quality gate or workflow was weakened. +The expanded selector adds a deterministic terminal-publication regression (confirmed failing +before its fix), repeated immediate client closes on empty 401/403 responses over HTTP/1.1 and +HTTP/2, and the eight existing authorization tests after a CI investigation. It was rerun together +with strict `javadoc:javadoc` generation; both passed. No quality gate or workflow was weakened. ## R0 SDK interoperability probes From 105456760e39648be9431bde323aaa23f89cc9b9 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:07:24 -0400 Subject: [PATCH 4/9] fix: bound STDIO transport shutdown and release lock waiters Close the owned child's stdin before waiting for a graceful exit, cap the exit waits against the shutdown deadline, return EOF when stdout closes before the child exits, and reject I/O waiting on a lock once the transport stops. Co-Authored-By: Claude Opus 5.5 --- .../engine/stdio/StreamTransportHelper.java | 84 ++++++++++---- .../restlet/util/StreamTransportTestCase.java | 103 ++++++++++++++++++ 2 files changed, 164 insertions(+), 23 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java index 584303c485..d9a545f1f9 100644 --- a/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java +++ b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java @@ -31,6 +31,7 @@ public final class StreamTransportHelper { private final ReentrantLock readLock = new ReentrantLock(); private final ReentrantLock writeLock = new ReentrantLock(); private final Set operations = new HashSet<>(); + private final Set lockWaiters = new HashSet<>(); private InputStream input; private OutputStream output; private Process process; @@ -177,14 +178,13 @@ public int read(byte[] buffer, int offset, int length) throws IOException { enter(); try { int count = input.read(buffer, offset, Math.min(length, maxOperationBytes)); - if (count == -1 && process != null) { - if (!process.waitFor(shutdownTimeout.toNanos(), TimeUnit.NANOSECONDS)) { - throw new IOException( - "Subprocess stdout ended without process termination"); - } - if (process.exitValue() != 0 && !isStopped()) { - throw new IOException("Subprocess exited with code " + process.exitValue()); - } + // EOF does not imply termination: only report an exit failure seen within grace. + if (count == -1 + && process != null + && process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS) + && process.exitValue() != 0 + && !isStopped()) { + throw new IOException("Subprocess exited with code " + process.exitValue()); } return count; } catch (InterruptedException e) { @@ -216,11 +216,25 @@ public void write(byte[] bytes) throws IOException { } } - private static void lock(ReentrantLock lock) throws InterruptedIOException { + /** Acquires an I/O lock, unless the transport stops while waiting for it. */ + private void lock(ReentrantLock lock) throws IOException { + synchronized (this) { + if (!isStarted()) { + throw new IOException("Stream transport is not accepting I/O"); + } + lockWaiters.add(Thread.currentThread()); + } try { lock.lockInterruptibly(); } catch (InterruptedException e) { + if (isStopped()) { + throw new IOException("Stream transport is not accepting I/O"); + } throw interrupted("Interrupted awaiting stream I/O", e); + } finally { + synchronized (this) { + lockWaiters.remove(Thread.currentThread()); + } } } @@ -302,10 +316,12 @@ public void stop() throws IOException { private void cleanup() { try { + long deadline = System.nanoTime() + shutdownTimeout.toNanos(); synchronized (this) { operations.forEach(Thread::interrupt); + lockWaiters.forEach(Thread::interrupt); } - IOException failure = process != null ? stopProcess() : cancelBorrowedIo(); + IOException failure = process != null ? stopProcess(deadline) : cancelBorrowedIo(); failure = awaitOperations(failure); if (process == null) { failure = flushBorrowedOutput(failure); @@ -326,24 +342,43 @@ private void cleanup() { } } - private IOException stopProcess() throws InterruptedException { - boolean exited = process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS); + /** + * Delivers EOF on the child's stdin, then terminates it if it does not exit within the grace + * period. The forcible kill happens halfway between the grace period and the shutdown deadline, + * leaving the rest of the shutdown timeout for pipe cleanup. + * + * @param deadline The {@link System#nanoTime()} shutdown deadline. + */ + private IOException stopProcess(long deadline) throws InterruptedException { + long grace = processExitGrace.toNanos(); + long killDeadline = deadline - (shutdownTimeout.toNanos() - grace) / 2; + // Closing may block behind a stuck write, which the process termination releases. + CompletableFuture stdinClosed = + CompletableFuture.supplyAsync(() -> close(output), Thread.ofVirtual()::start); + boolean exited = waitForExit(grace, killDeadline); IOException failure = exited && process.exitValue() != 0 ? new IOException("Subprocess exited with code " + process.exitValue()) : null; if (!exited) { process.destroy(); - if (!process.waitFor(processExitGrace.toNanos(), TimeUnit.NANOSECONDS)) { + if (!waitForExit(grace, killDeadline)) { process.destroyForcibly(); } } process.waitFor(); - failure = closePipes(failure); + failure = append(failure, stdinClosed.join()); + failure = append(failure, close(input)); stderrReader.join(); return failure; } + /** Waits for the process exit for at most the given duration and before the deadline. */ + private boolean waitForExit(long nanos, long deadline) throws InterruptedException { + long remaining = Math.min(nanos, deadline - System.nanoTime()); + return process.waitFor(Math.max(0, remaining), TimeUnit.NANOSECONDS); + } + private IOException cancelBorrowedIo() { try { cancelIo.run(); @@ -373,19 +408,22 @@ private IOException flushBorrowedOutput(IOException failure) { } private IOException closePipes(IOException failure) { - for (java.io.Closeable pipe : List.of(input, output)) { - try { - pipe.close(); - } catch (IOException e) { - failure = append(failure, e); - } + failure = append(failure, close(input)); + return append(failure, close(output)); + } + + private static IOException close(java.io.Closeable pipe) { + try { + pipe.close(); + return null; + } catch (IOException e) { + return e; } - return failure; } private static IOException append(IOException failure, IOException next) { - if (failure == null) { - return next; + if (failure == null || next == null) { + return failure == null ? next : failure; } failure.addSuppressed(next); return failure; diff --git a/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java index cb3e073268..d8ead03962 100644 --- a/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java +++ b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java @@ -23,11 +23,13 @@ import java.io.InputStream; import java.io.InterruptedIOException; import java.io.OutputStream; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.time.Duration; import java.util.Arrays; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -233,6 +235,47 @@ public int read() { } } + @Test + void rejectsQueuedReadsWhenStopTimesOut() throws Exception { + CountDownLatch reading = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + InputStream input = + new InputStream() { + @Override + public int read() { + reading.countDown(); + boolean done = false; + while (!done) { + try { + release.await(); + done = true; + } catch (InterruptedException _) { + // Deliberately models an input that does not support interruption. + } + } + return -1; + } + }; + var transport = new StreamTransport(input, new ByteArrayOutputStream(), () -> {}); + transport.setProcessExitGrace(Duration.ZERO); + transport.setShutdownTimeout(Duration.ofMillis(100)); + transport.start(); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + try { + executor.submit(() -> transport.read(new byte[1], 0, 1)); + assertTrue(reading.await(5, TimeUnit.SECONDS)); + var queued = executor.submit(() -> transport.read(new byte[1], 0, 1)); + assertThrows(IOException.class, transport::stop); + ExecutionException failure = + assertThrows( + ExecutionException.class, () -> queued.get(5, TimeUnit.SECONDS)); + assertInstanceOf(IOException.class, failure.getCause()); + } finally { + release.countDown(); + } + } + } + @Test void propagatesIoAndCancellationFailures() throws Exception { var transport = @@ -290,6 +333,41 @@ void launchesOwnedProcessAndDrainsBoundedStderrSeparately() throws Exception { assertFalse(ProcessHandle.of(pid).map(ProcessHandle::isAlive).orElse(false)); } + @Test + void returnsEofWhenStdoutClosesBeforeProcessExit() throws Exception { + var transport = new StreamTransport(command("close-stdout")); + transport.start(); + try { + assertEquals( + -1, + assertTimeoutPreemptively( + Duration.ofSeconds(3), () -> transport.read(new byte[1], 0, 1))); + } finally { + transport.stop(); + } + } + + @Test + void deliversStdinEofBeforeTerminatingOwnedProcess() throws Exception { + var transport = new StreamTransport(command("graceful")); + transport.setProcessExitGrace(Duration.ofSeconds(4)); + transport.start(); + transport.stop(); + assertEquals( + "clean-exit", new String(transport.getStderrTail(), StandardCharsets.US_ASCII)); + } + + @Test + void killsOwnedProcessWithinTheShutdownTimeout() throws Exception { + var transport = new StreamTransport(command("stubborn")); + transport.setShutdownTimeout(Duration.ofSeconds(3)); + transport.setProcessExitGrace(Duration.ofSeconds(2)); + transport.start(); + long pid = transport.getProcessId(); + assertTimeoutPreemptively(Duration.ofSeconds(5), transport::stop); + assertFalse(ProcessHandle.of(pid).map(ProcessHandle::isAlive).orElse(false)); + } + @Test void terminatesOwnedProcessWhileStdoutIsBlocked() throws Exception { var transport = new StreamTransport(command("idle")); @@ -362,6 +440,31 @@ public static void main(String[] args) throws Exception { if ("fail".equals(args[0])) { System.exit(23); } + if ("close-stdout".equals(args[0])) { + System.out.close(); + System.in.transferTo(OutputStream.nullOutputStream()); + return; + } + if ("graceful".equals(args[0])) { + System.in.transferTo(OutputStream.nullOutputStream()); + System.err.print("clean-exit"); + System.err.flush(); + return; + } + if ("stubborn".equals(args[0])) { + // Ignores stdin EOF and delays its exit on normal termination signals. + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + try { + Thread.sleep(60000); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } + })); + args = new String[] {"idle"}; + } if ("idle".equals(args[0])) { if (!new CountDownLatch(1).await(1, TimeUnit.MINUTES)) { throw new IllegalStateException("Idle test child was not stopped"); From 1398f76c7c793679ccc37addb1bce1bced5d4a5d Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:11:04 -0400 Subject: [PATCH 5/9] fix: release connections of cancelled HTTP client calls A call cancelled before or while acquiring its channel no longer acquires a connection, or releases the one acquired concurrently, so pre-cancelled requests do not hold pool permits. Co-Authored-By: Claude Opus 5.5 --- .../restlet/engine/netty/NettyClientCall.java | 28 ++++++++++++++-- .../connector/HttpDisconnectTestCase.java | 33 +++++++++++++++++++ 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java index 99b0188d8d..065679a078 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java @@ -320,10 +320,10 @@ private Status sendRequest(org.restlet.Request request) { boolean backgroundWriterStarted = false; try { - this.channel = helper.acquire(this); if (cancelled.get()) { throw new IOException("Request cancelled by client"); } + attachChannel(helper.acquire(this)); backgroundWriterStarted = sendRequestBody(request, entity, connectorService); awaitHead(); result = new Status(statusCode, reasonPhrase); @@ -343,6 +343,28 @@ private Status sendRequest(org.restlet.Request request) { return result; } + /** + * Attaches the acquired channel, unless the exchange was cancelled during the acquisition, in + * which case the channel is released at once. + * + * @param ch The acquired channel. + * @throws IOException If the exchange was cancelled. + */ + private void attachChannel(Channel ch) throws IOException { + boolean alreadyReleased; + lock.lock(); + try { + this.channel = ch; + alreadyReleased = released; + } finally { + lock.unlock(); + } + if (alreadyReleased) { + helper.release(ch, false); + throw new IOException("Request cancelled by client"); + } + } + private boolean sendRequestBody( org.restlet.Request request, Representation entity, ConnectorService connectorService) throws IOException { @@ -938,6 +960,7 @@ private void maybeRelease() { /** Releases the channel without reusing it. */ private void abort() { + Channel ch; lock.lock(); try { if (released) { @@ -945,11 +968,12 @@ private void abort() { } released = true; reusable = false; + // Read under the lock: a channel attached later is released by attachChannel(). + ch = this.channel; } finally { lock.unlock(); } clearAbortAction(); - Channel ch = this.channel; if (ch != null) { helper.release(ch, false); } diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java index b2022c2515..44f1c8cb3a 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java @@ -245,4 +245,37 @@ public void handle(Request request, Response response) { client.start(); } } + + @Test + void preCancelledRequestDoesNotRetainItsConnection() throws Exception { + Context serverContext = new Context(); + serverContext.getParameters().add("http.transport.protocol", "HTTP1_1"); + server = + new Server( + serverContext, + Protocol.HTTP, + 0, + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity("healthy", MediaType.TEXT_PLAIN); + } + }); + server.start(); + Context clientContext = new Context(); + clientContext.getParameters().add("httpClientTransportMode", "HTTP1_1"); + clientContext.getParameters().add("maxConnectionsPerDestination", "1"); + clientContext.getParameters().add("connectTimeout", "1000"); + client = new Client(clientContext, Protocol.HTTP); + client.start(); + String uri = "http://localhost:" + server.getActualPort() + "/"; + for (int i = 0; i < 2; i++) { + Request cancelled = new Request(Method.GET, uri); + assertFalse(cancelled.abort()); + assertTrue(client.handle(cancelled).getStatus().isConnectorError()); + } + Response response = client.handle(new Request(Method.GET, uri)); + assertEquals(Status.SUCCESS_OK, response.getStatus()); + assertEquals("healthy", response.getEntityAsText()); + } } From ebe4ed198cda715fe7ae547a144547787cd253b8 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:13:56 -0400 Subject: [PATCH 6/9] fix: report redirected exchange disconnects on the original request With followRedirects, a failure of a redirected exchange now completes the disconnect signal of the caller's request, not only the internal copy. Co-Authored-By: Claude Opus 5.5 --- .../engine/netty/NettyClientHelper.java | 2 ++ .../connector/HttpDisconnectTestCase.java | 21 +++++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java index 862361fb40..632e3bd099 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java @@ -311,6 +311,8 @@ public void handle(Request request, Response response) { handleOnce(current, response); next = (redirects < getMaxRedirects()) ? nextRedirect(current, response) : null; if (next != null) { + // The redirected exchange still belongs to the caller's logical request. + next.getDisconnectSignal().thenAccept(request::notifyDisconnect); if (response.getEntity() != null) { response.getEntity().release(); } diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java index 44f1c8cb3a..7324dbbe2c 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java @@ -278,4 +278,25 @@ public void handle(Request request, Response response) { assertEquals(Status.SUCCESS_OK, response.getStatus()); assertEquals("healthy", response.getEntityAsText()); } + + @Test + void notifiesOriginalRequestWhenRedirectedExchangeFails() throws Exception { + int closedPort; + try (var socket = new java.net.ServerSocket(0)) { + closedPort = socket.getLocalPort(); + } + start( + "HTTP1_1", + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.redirectTemporary("http://localhost:" + closedPort + "/gone"); + } + }); + Request request = new Request(Method.GET, "http://localhost:" + server.getActualPort()); + Request copy = new Request(request); + assertTrue(client.handle(request).getStatus().isConnectorError()); + assertNotNull(request.getDisconnectSignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + assertFalse(copy.getDisconnectSignal().toCompletableFuture().isDone()); + } } From 62750f551113ece09e0345093c65f17562db657f Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:20:42 -0400 Subject: [PATCH 7/9] fix: keep HTTP/1.1 upgrades behind active exchanges intact Do not resume reading after replaying a queued Upgrade or CONNECT request, and serve an h2c upgrade pipelined behind an active HTTP/1.1 exchange over HTTP/1.1 instead of switching protocols before the preceding response. Co-Authored-By: Claude Opus 5.5 --- .../engine/netty/NettyServerHandler.java | 14 +- .../engine/netty/NettyServerHelper.java | 10 +- .../engine/netty/Http1LookaheadTestCase.java | 162 ++++++++++++++++++ 3 files changed, 184 insertions(+), 2 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index 8356a52a1a..92f95dfb93 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java @@ -148,11 +148,23 @@ void responseComplete(ChannelHandlerContext ctx, NettyServerCall call) { pendingBytes -= pendingSize(msg); channelRead(ctx, msg); } - if (pending.isEmpty()) { + // Bytes following an upgrade or tunnel request belong to the replacement protocol. + if (pending.isEmpty() + && (activeCall.get() == null || currentBody != null || readAfterBody)) { ctx.read(); } } + /** + * Indicates whether an HTTP/1.1 exchange is in progress or queued on this connection. Must be + * called on the event loop. + * + * @return True if an exchange is in progress or queued. + */ + boolean isExchangeActive() { + return activeCall.get() != null || !pending.isEmpty(); + } + private void onRequest(ChannelHandlerContext ctx, HttpRequest request) { DecoderResult result = request.decoderResult(); if (result.isFailure()) { diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java index cbeb48b8a1..05602a916f 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHelper.java @@ -23,6 +23,7 @@ import io.netty.channel.nio.NioIoHandler; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http.HttpRequest; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.HttpServerKeepAliveHandler; import io.netty.handler.codec.http.HttpServerUpgradeHandler; @@ -405,7 +406,14 @@ private void addCleartextHttp2Handlers(ChannelPipeline pipeline) { return null; }; HttpServerUpgradeHandler upgradeHandler = - new HttpServerUpgradeHandler(sourceCodec, upgradeCodecFactory); + new HttpServerUpgradeHandler(sourceCodec, upgradeCodecFactory) { + @Override + protected boolean shouldHandleUpgradeRequest(HttpRequest request) { + // An upgrade pipelined behind an HTTP/1.1 exchange is served over HTTP/1.1. + NettyServerHandler handler = pipeline.get(NettyServerHandler.class); + return handler == null || !handler.isExchangeActive(); + } + }; // Prior knowledge: the connection starts directly with the HTTP/2 preface. ChannelInitializer priorKnowledge = diff --git a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java index 1cf6448c6b..27eb990b0d 100644 --- a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java @@ -14,19 +14,37 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.http.DefaultHttpContent; import io.netty.handler.codec.http.DefaultHttpRequest; +import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.LastHttpContent; +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.Socket; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import org.junit.jupiter.api.Test; +import org.restlet.Context; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.Restlet; +import org.restlet.Server; +import org.restlet.data.MediaType; +import org.restlet.data.Protocol; import org.restlet.data.Reference; import org.restlet.engine.connector.HttpServerHelper; @@ -87,6 +105,150 @@ void releasesQueuedContentWhenTheHandlerIsReplacedForAnUpgrade() throws Exceptio } } + @Test + void doesNotReadAfterReplayingAnUpgradeRequest() throws Exception { + assertEquals(0, readsAfterReplaying(HttpHeaderNames.UPGRADE.toString())); + assertTrue(readsAfterReplaying(null) > 0); + } + + @SuppressWarnings("unchecked") + private static int readsAfterReplaying(String header) throws Exception { + var helper = new HttpServerHelper(null); + var executor = NettyServerHelper.class.getDeclaredField("callExecutor"); + executor.setAccessible(true); + executor.set(helper, new IdleExecutor()); + AtomicInteger reads = new AtomicInteger(); + var handler = new NettyServerHandler(helper, false, StreamKind.HTTP1); + var channel = + new EmbeddedChannel( + new ChannelOutboundHandlerAdapter() { + @Override + public void read(ChannelHandlerContext ctx) { + reads.incrementAndGet(); + ctx.read(); + } + }, + handler); + try { + var active = + new NettyServerCall( + helper, + channel, + StreamKind.HTTP1, + new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/"), + new ChannelInputStream(channel, 0), + false); + var field = NettyServerHandler.class.getDeclaredField("activeCall"); + field.setAccessible(true); + ((AtomicReference) field.get(handler)).set(active); + var queued = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/next"); + if (header != null) { + queued.headers().set(header, "websocket"); + } + channel.writeInbound(queued, LastHttpContent.EMPTY_LAST_CONTENT); + reads.set(0); + handler.responseComplete(channel.pipeline().context(handler), active); + return reads.get(); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void servesH2cUpgradePipelinedBehindAnActiveExchangeOverHttp11() throws Exception { + org.restlet.engine.Engine.register(false); + org.restlet.engine.Engine.getInstance() + .getRegisteredServers() + .add(new HttpServerHelper(null)); + CountDownLatch slowStarted = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Context context = new Context(); + context.getParameters().add("http.transport.protocol", "HTTP2"); + var server = + new Server( + context, + Protocol.HTTP, + 0, + new Restlet() { + @Override + public void handle(Request request, Response response) { + String path = request.getResourceRef().getPath(); + if ("/slow".equals(path)) { + slowStarted.countDown(); + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } + } + response.setEntity(path, MediaType.TEXT_PLAIN); + } + }); + server.start(); + try (var socket = new Socket("localhost", server.getActualPort())) { + socket.setSoTimeout(5000); + OutputStream out = socket.getOutputStream(); + out.write( + ("GET /slow HTTP/1.1\r\nHost: localhost\r\n\r\n") + .getBytes(StandardCharsets.US_ASCII)); + out.flush(); + assertTrue(slowStarted.await(5, TimeUnit.SECONDS)); + out.write( + ("GET /upgrade HTTP/1.1\r\nHost: localhost\r\n" + + "Connection: Upgrade, HTTP2-Settings\r\n" + + "Upgrade: h2c\r\nHTTP2-Settings: AAMAAABkAAQAAP__\r\n\r\n") + .getBytes(StandardCharsets.US_ASCII)); + out.flush(); + Thread.sleep(200); + release.countDown(); + var in = + new BufferedReader( + new InputStreamReader( + socket.getInputStream(), StandardCharsets.US_ASCII)); + List statusLines = new ArrayList<>(); + String line; + while (statusLines.size() < 2 && (line = in.readLine()) != null) { + int status = line.indexOf("HTTP/1.1 "); + if (status >= 0) { + statusLines.add(line.substring(status)); + } + } + assertEquals(List.of("HTTP/1.1 200 OK", "HTTP/1.1 200 OK"), statusLines); + } finally { + release.countDown(); + server.stop(); + } + } + + /** An executor that accepts calls without running them. */ + private static final class IdleExecutor extends java.util.concurrent.AbstractExecutorService { + @Override + public void execute(Runnable command) {} + + @Override + public void shutdown() {} + + @Override + public List shutdownNow() { + return List.of(); + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return true; + } + } + @Test void publishesClientCompletionBeforeReleasingAnEmptyBodyConsumer() throws Exception { var trailers = new io.netty.handler.codec.http.DefaultHttpHeaders(); From 45ffe88bc7a2ed7de55f9b319c646ae1c4936c3b Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 02:27:44 -0400 Subject: [PATCH 8/9] fix: report disconnects until the terminal response is flushed Server disconnect signals are now suppressed only once the terminal response write succeeds, and a failed terminal write is reported as a disconnect. Co-Authored-By: Claude Opus 5.5 --- .../restlet/engine/netty/NettyServerCall.java | 40 +++++++++++++++++-- .../engine/netty/Http1LookaheadTestCase.java | 37 +++++++++++++++++ 2 files changed, 74 insertions(+), 3 deletions(-) diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index f76e2ad637..c8834fbb5c 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java +++ b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java @@ -135,6 +135,12 @@ class NettyServerCall { private Throwable disconnectCause; + /** Indicates whether the terminal response write was flushed, or the exchange handed over. */ + private boolean responseDelivered; + + /** The write of a response head that also ends the response, if any. */ + private ChannelFuture responseEndFuture; + /** The response body stream, if any. */ private volatile ChannelOutputStream responseBody; @@ -253,7 +259,7 @@ synchronized NettyRequest toRequest(Context context) { } synchronized void onDisconnect(Throwable cause) { - if (!completed && disconnectCause == null) { + if (!responseDelivered && disconnectCause == null) { disconnectCause = cause; if (restletRequest != null) { restletRequest.notifyDisconnect(cause); @@ -426,6 +432,7 @@ void commit(Response response) { */ private void upgrade(Response response, NettyConnectionUpgrade upgrade) { completed = true; + markResponseDelivered(); responseHeadWritten = true; requestBody.close(); try { @@ -646,7 +653,10 @@ private void writeHead(boolean emptyBody) { setHttp1Framing(head, emptyBody, hasTrailers, bodyAllowed); } - channel.write(head); + ChannelFuture future = channel.write(head); + if (responseEndedWithHead) { + responseEndFuture = future; + } } /** Creates the response head with the status and the response headers. */ @@ -718,6 +728,7 @@ void complete() { return; } if (responseEndedWithHead) { + trackDelivery(responseEndFuture); channel.flush(); requestBody.close(); return; @@ -731,7 +742,7 @@ void complete() { copyHeaders(responseTrailers, withTrailers.trailingHeaders(), kind.isStream()); lastContent = withTrailers; } - ChannelFuture future = channel.writeAndFlush(lastContent); + ChannelFuture future = trackDelivery(channel.writeAndFlush(lastContent)); requestBody.close(); if (!kind.isStream()) { @@ -746,6 +757,29 @@ void complete() { } } + /** + * Reports a failed terminal response write as a disconnect, and stops reporting disconnects + * once it is flushed. + * + * @param future The terminal write. + * @return The terminal write. + */ + private ChannelFuture trackDelivery(ChannelFuture future) { + future.addListener( + f -> { + if (f.isSuccess()) { + markResponseDelivered(); + } else { + onDisconnect(f.cause()); + } + }); + return future; + } + + private synchronized void markResponseDelivered() { + responseDelivered = true; + } + private void completeHttp1(ChannelFuture future, boolean keepAlive) { if (!keepAlive) { future.addListener(ChannelFutureListener.CLOSE); diff --git a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java index 27eb990b0d..84d3153a44 100644 --- a/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java @@ -24,6 +24,7 @@ import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.LastHttpContent; import java.io.BufferedReader; +import java.io.IOException; import java.io.InputStreamReader; import java.io.OutputStream; import java.net.Socket; @@ -220,6 +221,42 @@ public void handle(Request request, Response response) { } } + @Test + void reportsFailedTerminalResponseWriteAsDisconnect() throws Exception { + var helper = new HttpServerHelper(null); + IOException failure = new IOException("reset while flushing"); + var channel = + new EmbeddedChannel( + new io.netty.channel.ChannelOutboundHandlerAdapter() { + @Override + public void write( + ChannelHandlerContext ctx, + Object msg, + io.netty.channel.ChannelPromise promise) { + io.netty.util.ReferenceCountUtil.release(msg); + promise.setFailure(failure); + } + }); + try { + var call = + new NettyServerCall( + helper, + channel, + StreamKind.HTTP1, + new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/"), + new ChannelInputStream(channel, 0), + false); + Request request = call.toRequest(new Context()); + call.complete(); + channel.runPendingTasks(); + assertEquals( + failure, + request.getDisconnectSignal().toCompletableFuture().get(5, TimeUnit.SECONDS)); + } finally { + channel.finishAndReleaseAll(); + } + } + /** An executor that accepts calls without running them. */ private static final class IdleExecutor extends java.util.concurrent.AbstractExecutorService { @Override From 3d9090a1d679cd3814767d00cbedf9298ff0b296 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:28:47 -0400 Subject: [PATCH 9/9] test: only expect held gRPC headers when the SDK hosts the server The Restlet-hosted spike server sends the HTTP/2 response headers as soon as its handle method returns, which races with the first request message. The "no headers before they are sent" check therefore only holds for the SDK-hosted server and failed intermittently on CI for the Restlet-hosted case. Co-Authored-By: Claude Opus 5.5 --- .../GrpcSdkTransportSpikeTestCase.java | 20 +++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/GrpcSdkTransportSpikeTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/GrpcSdkTransportSpikeTestCase.java index 442e734e96..9052b33588 100644 --- a/org.restlet/src/test/java/org/restlet/engine/connector/GrpcSdkTransportSpikeTestCase.java +++ b/org.restlet/src/test/java/org/restlet/engine/connector/GrpcSdkTransportSpikeTestCase.java @@ -335,8 +335,7 @@ void cancelsActiveCallBeforeHeaders(boolean restletHosted) throws Exception { openInteropChannel(holding.definition(), restletHosted); HeldCalls held = startHeldCalls(1); assertTrue(holding.received.await(5, TimeUnit.SECONDS)); - assertEquals( - 0, held.headerCount.get(), "The SDK must not report headers before they are sent"); + assertNoHeadersReported(held, restletHosted); held.calls.getFirst().cancel("pre-header cancellation", null); assertTrue(held.closed.await(5, TimeUnit.SECONDS)); assertTrue(holding.cancelled.await(5, TimeUnit.SECONDS)); @@ -353,8 +352,7 @@ void shutsDownAllActiveCalls(boolean restletHosted) throws Exception { openInteropChannel(holding.definition(), restletHosted); HeldCalls held = startHeldCalls(2); assertTrue(holding.received.await(5, TimeUnit.SECONDS)); - assertEquals( - 0, held.headerCount.get(), "The SDK must not report headers before they are sent"); + assertNoHeadersReported(held, restletHosted); channel.shutdownNow(); assertTrue(held.closed.await(5, TimeUnit.SECONDS)); assertTrue(holding.cancelled.await(5, TimeUnit.SECONDS)); @@ -404,6 +402,20 @@ public void onClose(Status status, Metadata trailers) { return held; } + /** + * Verifies that the client SDK reported no response headers while the service holds them. The + * Restlet-hosted spike sends the HTTP/2 response headers as soon as its handle method returns, + * which races with the first message, so only the SDK-hosted server reliably holds them. + */ + private static void assertNoHeadersReported(HeldCalls held, boolean restletHosted) { + if (!restletHosted) { + assertEquals( + 0, + held.headerCount.get(), + "The SDK must not report headers before they are sent"); + } + } + /** Verifies that the I/O workers of every bidirectional call of the interop transport ended. */ private void assertWorkersFinished(int count) throws InterruptedException { if (interopFactory != null) {