diff --git a/changes.md b/changes.md index fec9b57522..c4484e2572 100644 --- a/changes.md +++ b/changes.md @@ -77,7 +77,25 @@ 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 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 + 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..99f8022594 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; @@ -313,9 +323,14 @@ 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. */ public boolean abort() { + abortRequested.set(true); BooleanSupplier action = abortAction; return action != null && action.getAsBoolean(); } @@ -324,10 +339,41 @@ 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. */ 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..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 { @@ -387,6 +409,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 +417,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 +426,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(); @@ -819,10 +844,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 { @@ -832,6 +853,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(); } @@ -868,6 +893,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); } @@ -931,6 +960,7 @@ private void maybeRelease() { /** Releases the channel without reusing it. */ private void abort() { + Channel ch; lock.lock(); try { if (released) { @@ -938,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/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/main/java/org/restlet/engine/netty/NettyServerCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerCall.java index aa27fe27f6..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 @@ -131,6 +131,16 @@ class NettyServerCall { /** Indicates whether the exchange was completed. */ private volatile boolean completed; + private NettyRequest restletRequest; + + 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; @@ -215,8 +225,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 +258,15 @@ NettyRequest toRequest(Context context) { return result; } + synchronized void onDisconnect(Throwable cause) { + if (!responseDelivered && disconnectCause == null) { + disconnectCause = cause; + if (restletRequest != null) { + restletRequest.notifyDisconnect(cause); + } + } + } + /** * Creates the Restlet response of this exchange. * @@ -409,6 +432,7 @@ void commit(Response response) { */ private void upgrade(Response response, NettyConnectionUpgrade upgrade) { completed = true; + markResponseDelivered(); responseHeadWritten = true; requestBody.close(); try { @@ -629,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. */ @@ -701,6 +728,7 @@ void complete() { return; } if (responseEndedWithHead) { + trackDelivery(responseEndFuture); channel.flush(); requestBody.close(); return; @@ -714,16 +742,11 @@ 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()) { - if (keepAlive) { - // Ready for the next request on this connection. - future.addListener(f -> channel.read()); - } else { - future.addListener(ChannelFutureListener.CLOSE); - } + completeHttp1(future, keepAlive); } } finally { if (eventStreamPermit) { @@ -734,6 +757,48 @@ 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); + 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/netty/NettyServerHandler.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyServerHandler.java index e6c7223b69..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 @@ -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,57 @@ 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); + } + // 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()) { 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 +194,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 +202,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 +214,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 +262,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 +290,9 @@ private void runCall(NettyServerCall call) { try { call.complete(); } finally { + if (kind.isStream()) { + activeCall.compareAndSet(call, null); + } helper.endCall(); Engine.clearThreadLocalVariables(); } @@ -223,9 +302,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 +342,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/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/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..d9a545f1f9 --- /dev/null +++ b/org.restlet/src/main/java/org/restlet/engine/stdio/StreamTransportHelper.java @@ -0,0 +1,431 @@ +/** + * 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 final Set lockWaiters = 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"); + } + shutdownTimeout = Duration.ofNanos(timeout.toNanos()); + } + + 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"); + } + processExitGrace = Duration.ofNanos(grace.toNanos()); + } + + 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 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)); + // 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) { + throw interrupted("Interrupted awaiting subprocess exit", e); + } 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(); + } + } + + /** 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()); + } + } + } + + 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; + } + + 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) { + throw interrupted("Interrupted stopping stream transport", e); + } 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 { + long deadline = System.nanoTime() + shutdownTimeout.toNanos(); + synchronized (this) { + operations.forEach(Thread::interrupt); + lockWaiters.forEach(Thread::interrupt); + } + IOException failure = process != null ? stopProcess(deadline) : cancelBorrowedIo(); + failure = awaitOperations(failure); + if (process == null) { + failure = flushBorrowedOutput(failure); + } + 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(); + } + } + } + + /** + * 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 (!waitForExit(grace, killDeadline)) { + process.destroyForcibly(); + } + } + process.waitFor(); + 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(); + 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) { + 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; + } + } + + private static IOException append(IOException failure, IOException next) { + if (failure == null || next == null) { + return failure == null ? next : failure; + } + 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/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) { 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..7324dbbe2c --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/HttpDisconnectTestCase.java @@ -0,0 +1,302 @@ +/** + * 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.data.Status; +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(); + } + } + + @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(); + } + } + + @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()); + } + + @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()); + } +} 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..f6043afb19 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/StdioHelperSelectionTestCase.java @@ -0,0 +1,111 @@ +/** + * 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)); + Context missing = new Context(); + List protocols = List.of(Protocol.STDIO); + assertThrows( + IllegalStateException.class, + () -> new Client(missing, protocols, "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..84d3153a44 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/netty/Http1LookaheadTestCase.java @@ -0,0 +1,346 @@ +/** + * 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.assertThrows; +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.IOException; +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; + +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(); + } + } + + @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(); + } + } + + @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 + 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(); + 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); + 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..d8ead03962 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/util/StreamTransportTestCase.java @@ -0,0 +1,519 @@ +/** + * 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.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; +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 _) { + 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 _) { + 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 _) { + // 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 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 = + 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 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")); + 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 _) { + 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 ("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"); + } + 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..adf7ba64d4 100644 --- a/readme.md +++ b/readme.md @@ -12,6 +12,136 @@ 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. +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 +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,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 **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 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 The MCP and gRPC interoperability spikes are **test-only**, not production RPC extensions.