From 4a48efda4d28ecc386967d081ef1b5284478b7e7 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 13:00:26 -0700 Subject: [PATCH 01/10] Add typed ClientChannelCloseException for client-rooted PUT termination When a client's PUT upload terminates because of the client (TCP disconnect detected via channelInactive), NettyMessageProcessor now delivers a dedicated ClientChannelCloseException to the readInto(...) callback, instead of a bare ClosedChannelException with no provenance. - ClientChannelCloseException extends java.nio.channels.ClosedChannelException (ambry-utils), so every existing catch clause keying on ClosedChannelException behaves identically; downstream consumers can additionally detect the client-rooted case via instanceof. - Utils.isPossibleClientTermination(...) recognizes the new type in addition to its existing message-suffix heuristic, so OSS-internal consumers of that utility keep working unchanged. - NettyRequest gains markClientTerminated()/closeDueToClientTermination() to atomically tag+close a request as client-rooted. - NettyMessageProcessor.channelInactive() is the only call site that tags a termination as client-rooted. exceptionCaught()'s IOException branch and the idle-timeout (ALL_IDLE) branch of userEventTriggered() are intentionally left untagged, with comments explaining why: both can be reached from a genuinely server/destination-rooted stall or failure (backpressure-suspended reads for idle-timeout; Netty's implicit uncaught-exception routing from a destination write failure for exceptionCaught), so tagging them would risk a false positive that hides a real server-side problem, which the design explicitly biases against. Required, coupled fix: NettyResponseChannel.completeRequest() now closes the request (flips isOpen()==false) before scheduling the network channel's close listener, instead of after. Previously, if the response writeFuture was already complete when the close listener was added, Netty's ChannelFuture.addListener() would fire it synchronously/ re-entrantly, closing the network channel and triggering channelInactive() before isOpen() had flipped to false, causing a pure server-initiated completion to look, from channelInactive()'s perspective, indistinguishable from a client disconnect. This reorder is what makes channelInactive() exclusively client-rooted; without it, the single tagged call site would not be safe. This is not a drive-by refactor. It is a precondition for the correctness of the new tagging. Tests: - NettyRequestTest: markClientTerminatedDeliversTypedExceptionTest covers markClientTerminated+close, plain close (untagged), and closeDueToClientTermination. - NettyMessageProcessorTest: channelInactiveDeliversClientTerminationToReadIntoTest (positive), serverAbortDoesNotDeliverClientTerminationToReadIntoTest (negative, RestServiceException), idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest and exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest (negative, documenting the intentionally-untagged branches). - NettyResponseChannelTest: completeRequestClosesRequestBeforeReturningTest is a standalone regression test (independent of the ClientChannelCloseException feature) asserting request.isOpen()==false immediately after completeRequest()/onResponseComplete() returns, with javadoc documenting why the reorder is required for channelInactive()'s exclusivity guarantee. - UtilsTest: clientTerminationWrapAndRecognizeTest extended to assert isPossibleClientTermination recognizes ClientChannelCloseException. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ambry/rest/NettyMessageProcessor.java | 23 ++- .../com/github/ambry/rest/NettyRequest.java | 24 +++ .../ambry/rest/NettyResponseChannel.java | 13 +- .../ambry/rest/NettyMessageProcessorTest.java | 182 ++++++++++++++++++ .../github/ambry/rest/NettyRequestTest.java | 53 +++++ .../ambry/rest/NettyResponseChannelTest.java | 58 ++++++ .../utils/ClientChannelCloseException.java | 30 +++ .../java/com/github/ambry/utils/Utils.java | 3 + .../com/github/ambry/utils/UtilsTest.java | 5 + 9 files changed, 389 insertions(+), 2 deletions(-) create mode 100644 ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index aba3d92c3e..1adc9d959d 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -143,8 +143,12 @@ public void channelInactive(ChannelHandlerContext ctx) { // and can short-circuit best-effort work (e.g. named-blob metadata commit in AmbryIdConverterFactory). // NettyRequest.close() is idempotent via channelOpen.compareAndSet(true,false), so double-close on the // normal-completion path is a no-op. + // Mark the pending readInto callback as client-rooted before closing: reaching this branch means the + // request was never closed by the normal completion path (which always closes the request - see + // NettyResponseChannel#completeRequest - before scheduling the network channel close), so channel + // inactivity here can only be due to the client. try { - request.close(); + request.closeDueToClientTermination(); } catch (Exception e) { logger.warn("Exception while closing request {} on channelInactive", request.getUri(), e); } @@ -170,6 +174,14 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E try { if (request != null && request.isOpen() && cause instanceof Exception) { nettyMetrics.processorExceptionCaughtCount.inc(); + // NOTE: an IOException reaching this handler for an in-flight request is likely client-rooted (Netty's own + // handling of this client-facing channel, e.g. "connection reset"/"broken pipe" while reading further + // request content) rather than a business/destination write failure (those are reported directly to the + // RestResponseChannel by FrontendRestRequestService/AsyncRequestResponseHandler and never reach this + // pipeline handler). However, this path is intentionally left untagged as ClientChannelCloseException in + // this change - only channelInactive() (below) has been proven exclusively client-rooted with no + // server-initiated code path that can trigger it while request.isOpen() is still true. Tagging this site + // is left as a documented follow-up rather than risking a less-proven exclusivity claim here. onRequestAborted((Exception) cause); } else if (isOpen()) { if (cause instanceof RestServiceException) { @@ -221,6 +233,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { nettyConfig.nettyServerIdleTimeSeconds); nettyMetrics.idleConnectionCloseCount.inc(); if (request != null && request.isOpen()) { + // NOTE: idle-timeout is intentionally left untagged as ClientChannelCloseException. NettyRequest suspends + // reads (autoRead=false) on this same channel while the amount of data buffered for a slow/backpressured + // downstream consumer exceeds nettyServerRequestBufferWatermark (see NettyRequest#continueReadIfPossible). + // While reads are suspended, no channelRead events can occur no matter how active the client is, so + // IdleStateHandler's ALL_IDLE can fire purely because OUR OWN downstream write is stalled - not because the + // client is idle or has failed. Tagging this as a client termination would risk a false positive that hides + // a server/destination-side slowness problem, violating the "bias toward NOT-client on ambiguity" invariant. + // So this path deliberately falls through to the default (untagged) ClosedChannelException, same as before + // this change; only channelInactive() has been proven exclusively client-rooted. onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); } else { close(); diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index fa8ddb442a..000aa8929a 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java @@ -16,6 +16,7 @@ import com.github.ambry.commons.Callback; import com.github.ambry.router.AsyncWritableChannel; import com.github.ambry.router.FutureResult; +import com.github.ambry.utils.ClientChannelCloseException; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.DefaultMaxBytesRecvByteBufAllocator; @@ -65,6 +66,7 @@ public class NettyRequest implements RestRequest { // is <=0, it is assumed that there is no limit on the size of unacknowledged data. static int bufferWatermark = -1; private static final ClosedChannelException CLOSED_CHANNEL_EXCEPTION = new ClosedChannelException(); + private static final ClientChannelCloseException CLIENT_CHANNEL_CLOSE_EXCEPTION = new ClientChannelCloseException(); protected final HttpRequest request; protected final Channel channel; @@ -301,6 +303,28 @@ public void close() { } } + /** + * Marks this request's pending read (if any) as terminated because of a high-confidence, client-rooted event + * (e.g. the client disconnected, reset the connection, or went idle past the configured timeout). Must be called, + * if at all, before {@link #close()} so that {@link ClientChannelCloseException} - rather than the default + * {@link ClosedChannelException} - is delivered to the pending {@link #readInto} callback. Idempotent and safe to + * call even if there is no pending read. + */ + void markClientTerminated() { + channelException = CLIENT_CHANNEL_CLOSE_EXCEPTION; + } + + /** + * Convenience method that marks this request as client-terminated (see {@link #markClientTerminated()}) and then + * closes it, in one call. Use this at call sites that close the request directly (as opposed to sites where the + * close happens later via a different code path, e.g. through {@link NettyResponseChannel}), so the "mark before + * close" ordering requirement can never be broken by a future edit that reorders or drops one of the two calls. + */ + void closeDueToClientTermination() { + markClientTerminated(); + close(); + } + @Override public RestRequestMetricsTracker getMetricsTracker() { return restRequestMetricsTracker; diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java index 8f62919313..73496f135a 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java @@ -805,6 +805,18 @@ private HttpResponseStatus getHttpResponseStatus(ResponseStatus responseStatus) * @param shouldCloseRequest used to determine if we want to close the request or not. */ private void completeRequest(boolean closeNetworkChannel, boolean shouldDelay, boolean shouldCloseRequest) { + // Close the request (and flip its isOpen() state to false) before scheduling the network channel close below. + // If writeFuture is already complete, addListener(...) fires its listener synchronously, which can close the + // network channel and trigger channelInactive() re-entrantly on this same call stack. NettyMessageProcessor's + // channelInactive() uses request.isOpen() to decide whether channel inactivity is client-rooted (as opposed to + // this server-initiated completion) - closeRequest() must therefore run first so that check is never fooled by + // a server-initiated close still in progress. Failures here must not prevent the network channel close below + // from being scheduled (that was always attempted first previously), so swallow (and log) any exception. + try { + closeRequest(shouldCloseRequest); + } catch (Exception e) { + logger.error("Exception while closing request on channel {}", ctx.channel(), e); + } if ((closeNetworkChannel || forceClose) && ctx.channel().isOpen()) { if (shouldDelay && (request != null && request.getRestMethod().equals(RestMethod.POST)) && this.nettyConfig.nettyServerCloseDelayTimeoutMs > 0) { @@ -815,7 +827,6 @@ private void completeRequest(boolean closeNetworkChannel, boolean shouldDelay, b } logger.trace("Requested closing of channel {}", ctx.channel()); } - closeRequest(shouldCloseRequest); } /** diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index baeb04544f..c040f2ee25 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -28,6 +28,7 @@ import com.github.ambry.notification.UpdateType; import com.github.ambry.router.InMemoryRouter; import com.github.ambry.store.MessageInfo; +import com.github.ambry.utils.ClientChannelCloseException; import com.github.ambry.utils.TestUtils; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; @@ -57,9 +58,12 @@ import io.netty.handler.codec.http.multipart.HttpPostRequestEncoder; import io.netty.handler.codec.http.multipart.MemoryFileUpload; import io.netty.handler.stream.ChunkedWriteHandler; +import io.netty.handler.timeout.IdleState; +import io.netty.handler.timeout.IdleStateEvent; import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Collections; @@ -531,6 +535,184 @@ public void channelInactiveClosesInflightRequestTest() throws Exception { } } + /** + * Verifies that a client TCP disconnect ({@link NettyMessageProcessor#channelInactive}) while a PUT request is + * still in-flight delivers a {@link ClientChannelCloseException} to the pending {@code readInto} callback, so + * downstream consumers can recognize the termination as client-rooted via {@code instanceof} or + * {@link com.github.ambry.utils.Utils#isPossibleClientTermination(Throwable)}. + * @throws Exception + */ + @Test + public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "channelInactiveDeliversClientTerminationToReadIntoTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + // Simulate the client TCP disconnect / channel becoming inactive mid-request. + channel.close().awaitUninterruptibly(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertTrue("readInto callback exception should be a ClientChannelCloseException", + callback.exception instanceof ClientChannelCloseException); + } finally { + capturingHandler.shutdown(); + } + } + + /** + * Verifies that a client idle/stall timeout (the {@link IdleState#ALL_IDLE} branch of + * {@link NettyMessageProcessor#userEventTriggered}) does NOT deliver a {@link ClientChannelCloseException} to the + * pending {@code readInto} callback. This is intentional: {@link NettyRequest} suspends reads (autoRead=false) on + * the channel while a slow/backpressured downstream consumer keeps buffered data above + * {@code nettyServerRequestBufferWatermark} (see {@link NettyRequest#continueReadIfPossible}), so + * {@code ALL_IDLE} can fire purely due to a server/destination-side stall rather than genuine client inactivity. + * Since this ambiguity cannot yet be cleanly disambiguated with high confidence, idle-timeout is left untagged in + * this change (falls back to the default {@link ClosedChannelException}); only {@code channelInactive} has been + * proven exclusively client-rooted. + * @throws Exception + */ + @Test + public void idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + // Simulate the connection going idle past the configured timeout. + channel.pipeline().fireUserEventTriggered(IdleStateEvent.ALL_IDLE_STATE_EVENT); + channel.runPendingTasks(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertFalse( + "readInto callback exception must NOT be a ClientChannelCloseException for idle-timeout in this change " + + "(left untagged due to backpressure ambiguity - see channelInactive for the proven client-exclusive " + + "case)", callback.exception instanceof ClientChannelCloseException); + assertTrue("readInto callback exception should still be a ClosedChannelException", + callback.exception instanceof ClosedChannelException); + } finally { + capturingHandler.shutdown(); + } + } + + /** + * Verifies (and documents) that an {@link IOException} reaching {@link NettyMessageProcessor#exceptionCaught} + * while a PUT request is still in-flight (e.g. "connection reset"/"broken pipe" while reading further request + * content from the client) does NOT deliver a {@link ClientChannelCloseException} to the pending {@code readInto} + * callback in this change. This path is a plausible high-confidence client-rooted signal, but is deliberately left + * untagged/out of scope here alongside idle-timeout - only {@code channelInactive} has been proven exclusively + * client-rooted with no possible server-initiated trigger. Tagging this site is a documented follow-up candidate. + * @throws Exception + */ + @Test + public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers() + .set(RestUtils.Headers.SERVICE_ID, "exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + // Simulate an IOException surfacing on the client-facing channel while the request is still open (e.g. a + // broken pipe / connection reset detected while trying to read further content from the client). + channel.pipeline().fireExceptionCaught(new IOException("Simulated connection reset by peer")); + channel.runPendingTasks(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertFalse( + "readInto callback exception must NOT be a ClientChannelCloseException for exceptionCaught's IOException " + + "branch in this change (deliberately left untagged/out of scope - see channelInactive for the " + + "proven client-exclusive case)", callback.exception instanceof ClientChannelCloseException); + } finally { + capturingHandler.shutdown(); + } + } + + /** + * Verifies that a server-side abort (e.g. {@link NettyMessageProcessor#exceptionCaught} triggered by an internal + * {@link RestServiceException}, with no client disconnect) does NOT deliver a {@link ClientChannelCloseException} + * to the pending {@code readInto} callback. This is the negative counterpart to + * {@link #channelInactiveDeliversClientTerminationToReadIntoTest()}, + * {@link #idleTimeoutDeliversClientTerminationToReadIntoTest()}, and + * {@link #exceptionCaughtIOExceptionDeliversClientTerminationToReadIntoTest()}, proving server/internal + * terminations are never mis-tagged as client-rooted. + * @throws Exception + */ + @Test + public void serverAbortDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "serverAbortDoesNotDeliverClientTerminationToReadIntoTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + // Simulate a purely server-side/internal abort - no client disconnect, no idle timeout. + channel.pipeline() + .fireExceptionCaught(new RestServiceException("Simulated internal error", RestServiceErrorCode.InternalServerError)); + channel.runPendingTasks(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertFalse("readInto callback exception must NOT be a ClientChannelCloseException for a server-rooted abort", + callback.exception instanceof ClientChannelCloseException); + } finally { + capturingHandler.shutdown(); + } + } + /** * {@link RestRequestHandler} that captures the first {@link RestRequest} passed to * {@link #handleRequest(RestRequest, RestResponseChannel)} and does nothing else. Used by diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java index 28fceba9a6..1d7e9c8e29 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java @@ -20,6 +20,7 @@ import com.github.ambry.config.VerifiableProperties; import com.github.ambry.router.AsyncWritableChannel; import com.github.ambry.router.FutureResult; +import com.github.ambry.utils.ClientChannelCloseException; import com.github.ambry.utils.NettyByteBufLeakHelper; import com.github.ambry.utils.TestUtils; import com.github.ambry.utils.Utils; @@ -280,6 +281,58 @@ public void conversionWithBadInputTest() throws RestServiceException { } } + /** + * Tests that {@link NettyRequest#markClientTerminated()} causes {@link NettyRequest#close()} to deliver a + * {@link ClientChannelCloseException} to a pending {@link NettyRequest#readInto} callback, instead of the + * default bare {@link ClosedChannelException}. + * @throws Exception + */ + @Test + public void markClientTerminatedDeliversTypedExceptionTest() throws Exception { + // Case 1: markClientTerminated() then close() -> pending readInto callback gets ClientChannelCloseException. + Channel channel = new MockChannel(); + NettyRequest nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + AsyncWritableChannel writeChannel = new ByteBufferAsyncWritableChannel(); + ReadIntoCallback callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + nettyRequest.markClientTerminated(); + nettyRequest.close(); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("Exception should be a ClientChannelCloseException", + callback.exception instanceof ClientChannelCloseException); + assertTrue("ClientChannelCloseException must still be a ClosedChannelException for backward compatibility", + callback.exception instanceof ClosedChannelException); + + // Case 2: close() without markClientTerminated() -> pending readInto callback gets the default bare + // ClosedChannelException, NOT ClientChannelCloseException. This proves the normal-completion path (which + // never calls markClientTerminated()) is unaffected. + channel = new MockChannel(); + nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + writeChannel = new ByteBufferAsyncWritableChannel(); + callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + nettyRequest.close(); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertFalse("Exception should not be a ClientChannelCloseException when markClientTerminated() was not called", + callback.exception instanceof ClientChannelCloseException); + assertTrue("Exception should still be a ClosedChannelException", callback.exception instanceof ClosedChannelException); + + // Case 3: closeDueToClientTermination() (the atomic mark-then-close convenience method) has the same effect as + // Case 1's two-step call. + channel = new MockChannel(); + nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + writeChannel = new ByteBufferAsyncWritableChannel(); + callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + nettyRequest.closeDueToClientTermination(); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("Exception should be a ClientChannelCloseException", + callback.exception instanceof ClientChannelCloseException); + } + /** * Tests for behavior of multiple operations after {@link NettyRequest#close()} has been called. Some should be ok to * do and some should throw exceptions. diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java index e6f4b325a8..83df050a7b 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java @@ -763,6 +763,57 @@ public void channelInactiveWriteTaggingByMethodTest() throws Exception { } } + /** + * Regression test for the ordering fix in {@code NettyResponseChannel#completeRequest}. That method now closes + * the request (flipping {@link NettyRequest#isOpen()} to {@code false}) before it schedules the listener that + * closes the underlying network channel, instead of the other way around as before this change. + *

+ * This is a coupled/dependent fix, not a drive-by: {@link NettyMessageProcessor#channelInactive} tags the + * termination as client-rooted (via {@link com.github.ambry.utils.ClientChannelCloseException}) precisely + * because it trusts {@code request.isOpen()} to already be {@code false} for every server-initiated completion + * by the time the network channel actually closes and {@code channelInactive} fires. Before this reorder, that + * was not guaranteed: {@link io.netty.channel.ChannelFuture#addListener} invokes its listener synchronously/ + * re-entrantly if the future is already complete, so a pure server-side completion (no client involvement) + * could close the network channel - and trigger {@code channelInactive} re-entrantly on the same call stack - + * before {@code closeRequest()} had flipped {@code isOpen()} to {@code false}, which would have made + * {@code channelInactive}'s sole tagged call site unsafe (a false client-abort masking a server completion). + * This test is therefore the evidence that the reorder is required for {@code channelInactive}-only tagging to + * be safe, not incidental scope creep - it asserts the invariant directly, for both the exception and + * non-exception {@link RestResponseChannel#onResponseComplete(Exception)} paths, so that a future refactor that + * reorders {@code completeRequest} again cannot silently reopen the race. + */ + @Test + public void completeRequestClosesRequestBeforeReturningTest() throws Exception { + // success (non-exception) path: TestingUri.CopyHeaders drives onResponseComplete(null). + { + EmbeddedChannel channel = createEmbeddedChannel(); + MockNettyMessageProcessor processor = channel.pipeline().get(MockNettyMessageProcessor.class); + HttpRequest httpRequest = createRequestWithHeaders(HttpMethod.GET, TestingUri.CopyHeaders.toString()); + channel.writeInbound(httpRequest); + // MockNettyMessageProcessor's handler for CopyHeaders calls onResponseComplete(null) synchronously while + // processing the inbound request, and EmbeddedChannel resolves writes synchronously (no real network I/O), + // so by the time writeInbound() returns, onResponseComplete(null) must already have returned too. + assertFalse("Request must already be closed once onResponseComplete(null) has returned", + processor.getRequest().isOpen()); + while (channel.readOutbound() != null) { + // drain the channel. + } + } + // exception path: TestingUri.OnResponseCompleteWithNonRestException drives onResponseComplete(exception). + { + EmbeddedChannel channel = createEmbeddedChannel(); + MockNettyMessageProcessor processor = channel.pipeline().get(MockNettyMessageProcessor.class); + HttpRequest httpRequest = + createRequestWithHeaders(HttpMethod.GET, TestingUri.OnResponseCompleteWithNonRestException.toString()); + channel.writeInbound(httpRequest); + assertFalse("Request must already be closed once onResponseComplete(exception) has returned", + processor.getRequest().isOpen()); + while (channel.readOutbound() != null) { + // drain the channel. + } + } + } + /** * Tests the invocation of DELAYED_CLOSE when post failures happen in {@link NettyResponseChannel}. */ @@ -1401,6 +1452,13 @@ public NettyMetrics getNettyMetrics() { return nettyMetrics; } + /** + * @return the {@link NettyRequest} backing the current/most recently handled request on this processor. + */ + public NettyRequest getRequest() { + return request; + } + @Override public void channelActive(ChannelHandlerContext ctx) { this.ctx = ctx; diff --git a/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java b/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java new file mode 100644 index 0000000000..8e8ceaba01 --- /dev/null +++ b/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java @@ -0,0 +1,30 @@ +/** + * Copyright 2026 LinkedIn Corp. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + */ +package com.github.ambry.utils; + +import java.nio.channels.ClosedChannelException; + + +/** + * A {@link ClosedChannelException} thrown specifically when a channel is confirmed, with high confidence, to have + * been closed because of a client-rooted termination (e.g. the client disconnected, reset the connection, or went + * idle for longer than the configured timeout). + *

+ * Extending {@link ClosedChannelException} keeps this backward compatible with any existing code that catches or + * checks for {@link ClosedChannelException}. Callers that need to distinguish a client-rooted termination from any + * other reason a channel might close can do so via an {@code instanceof ClientChannelCloseException} check, or via + * {@link Utils#isPossibleClientTermination(Throwable)}. + */ +public class ClientChannelCloseException extends ClosedChannelException { +} diff --git a/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java b/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java index 7e394ebc0b..516942c768 100644 --- a/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java +++ b/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java @@ -1286,6 +1286,9 @@ public static void shutDownExecutorService(ExecutorService executorService, long * @return {@code true} this cause indicates a possible early termination from the client. {@code false} otherwise. */ public static boolean isPossibleClientTermination(Throwable cause) { + if (cause instanceof ClientChannelCloseException) { + return true; + } if (cause instanceof IOException) { String msg = cause.getMessage(); if (msg != null) { diff --git a/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java b/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java index f724b70d8f..f24bf982d5 100644 --- a/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java +++ b/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java @@ -685,6 +685,11 @@ public void clientTerminationWrapAndRecognizeTest() { exception = new java.nio.channels.ClosedChannelException(); assertFalse("Bare ClosedChannelException should not be declared as a client termination", Utils.isPossibleClientTermination(exception)); + + // the typed ClientChannelCloseException should be recognized directly, regardless of message. + exception = new ClientChannelCloseException(); + assertTrue("ClientChannelCloseException should be declared as a client termination", + Utils.isPossibleClientTermination(exception)); } /** From 9e596b5387ce72da347763cf8fd88e423842988a Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 13:58:05 -0700 Subject: [PATCH 02/10] Fix dead javadoc @link references in NettyMessageProcessorTest serverAbortDoesNotDeliverClientTerminationToReadIntoTest's javadoc referenced idleTimeoutDeliversClientTerminationToReadIntoTest() and exceptionCaughtIOExceptionDeliversClientTerminationToReadIntoTest(), which no longer exist after those tests were repurposed into negative tests (...DoesNotDeliver...). Point at the correct existing method names. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../github/ambry/rest/NettyMessageProcessorTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index c040f2ee25..bb9dce0812 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -672,11 +672,11 @@ public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoT /** * Verifies that a server-side abort (e.g. {@link NettyMessageProcessor#exceptionCaught} triggered by an internal * {@link RestServiceException}, with no client disconnect) does NOT deliver a {@link ClientChannelCloseException} - * to the pending {@code readInto} callback. This is the negative counterpart to - * {@link #channelInactiveDeliversClientTerminationToReadIntoTest()}, - * {@link #idleTimeoutDeliversClientTerminationToReadIntoTest()}, and - * {@link #exceptionCaughtIOExceptionDeliversClientTerminationToReadIntoTest()}, proving server/internal - * terminations are never mis-tagged as client-rooted. + * to the pending {@code readInto} callback. This is the positive-tagging counterpart to + * {@link #channelInactiveDeliversClientTerminationToReadIntoTest()}, and complements + * {@link #idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest()} and + * {@link #exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest()}, together proving + * server/internal terminations are never mis-tagged as client-rooted. * @throws Exception */ @Test From 8730e806311f209bbb845e77c311a29b8733ec64 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 14:00:09 -0700 Subject: [PATCH 03/10] Tighten idle-timeout comment: document structural false-positive and correct follow-up direction Strengthen the ALL_IDLE branch's rationale comment to explain the structural (non-racy) false-positive: NettyRequest.writeContent() unconditionally re-enables autoRead the instant the last client chunk arrives, before the corresponding destination write is issued, so a slow final destination write leaves the channel idle with autoRead==true for the whole window. Also note the correct fast-follow direction (gate on "no destination write in flight", not autoRead state) and use the requested "pending backpressure-aware follow-up" phrasing. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ambry/rest/NettyMessageProcessor.java | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index 1adc9d959d..e7aeee1307 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -233,15 +233,21 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { nettyConfig.nettyServerIdleTimeSeconds); nettyMetrics.idleConnectionCloseCount.inc(); if (request != null && request.isOpen()) { - // NOTE: idle-timeout is intentionally left untagged as ClientChannelCloseException. NettyRequest suspends - // reads (autoRead=false) on this same channel while the amount of data buffered for a slow/backpressured - // downstream consumer exceeds nettyServerRequestBufferWatermark (see NettyRequest#continueReadIfPossible). - // While reads are suspended, no channelRead events can occur no matter how active the client is, so - // IdleStateHandler's ALL_IDLE can fire purely because OUR OWN downstream write is stalled - not because the - // client is idle or has failed. Tagging this as a client termination would risk a false positive that hides - // a server/destination-side slowness problem, violating the "bias toward NOT-client on ambiguity" invariant. - // So this path deliberately falls through to the default (untagged) ClosedChannelException, same as before - // this change; only channelInactive() has been proven exclusively client-rooted. + // NOTE: idle-timeout is intentionally left untagged as ClientChannelCloseException, pending a + // backpressure-aware follow-up. NettyRequest suspends reads (autoRead=false) while the amount of data + // buffered for a slow/backpressured downstream consumer exceeds nettyServerRequestBufferWatermark (see + // NettyRequest#continueReadIfPossible); while reads are suspended, no channelRead events can occur no + // matter how active the client is, so IdleStateHandler's ALL_IDLE can fire purely because OUR OWN + // downstream write is stalled - not because the client is idle or has failed. Worse, this isn't just a + // narrow race: NettyRequest#writeContent unconditionally re-enables autoRead the instant the last client + // chunk arrives, before the corresponding destination write is even issued - so a slow destination write + // on that final chunk leaves the channel silent in both directions with autoRead==true for the entire + // idle window, a deterministic (not merely racy) false-positive shape. Tagging this as a client + // termination would risk hiding a real server/destination-side slowness problem, violating the "bias + // toward NOT-client on ambiguity" invariant. So this path deliberately falls through to the default + // (untagged) ClosedChannelException, same as before this change; only channelInactive() has been proven + // exclusively client-rooted. A correct follow-up would need to gate on "no destination write currently + // in flight" rather than on autoRead state. onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); } else { close(); From c1d3aee649faa7f41a8f8746517b9f3347caf383 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 14:42:16 -0700 Subject: [PATCH 04/10] Add three-tier client-termination classification (sure/possible/other) Introduces PossibleClientChannelCloseException as a sibling of ClientChannelCloseException (both extend ClosedChannelException directly, no subtype relationship, so instanceof checks for one tier are never satisfied by the other) to distinguish high-confidence client aborts from plausible-but-unproven ones at the readInto layer: - Sure (ClientChannelCloseException): channelInactive only - proven exclusively client-rooted, unchanged from before this commit. - Possible (PossibleClientChannelCloseException, new): idle-timeout and exceptionCaught's IOException branch. Both have a plausible non-client alternate cause (idle: destination-write-in-flight race; exceptionCaught: Netty's implicit exception routing could surface a destination-side IOException here, so they are tagged as ambiguous rather than sure. - Other/unclassified (bare ClosedChannelException, unchanged): RestServiceException-triggered aborts and anything else - never tagged as client-rooted at either tier. NettyRequest gains markPossibleClientTermination() (with a guard so a prior sure tag is never downgraded to possible) and closeDueToPossibleClientTermination(), mirroring the existing markClientTerminated()/closeDueToClientTermination() sure-tier methods. The new idle-timeout and exceptionCaught call sites mark only (no explicit close call) since the existing onRequestAborted(...) -> responseChannel.close(...) -> request.close() flow already reliably closes the request later - this introduces no new close-ordering behavior beyond what channelInactive's mark-and- close already established. Utils.isPossibleClientTermination(...) now also recognizes the new type, so OSS-internal and downstream (AmbryLI) consumers that already call it keep working, gaining the extra tier for free. Testing Done: - New tests: idleTimeoutDeliversPossibleClientTerminationToReadIntoTest, exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest, markPossibleClientTerminationDeliversTypedExceptionTest (including the no-downgrade-from-sure-to-possible case), UtilsTest coverage for the new type. - Strengthened serverAbortDoesNotDeliverClientTerminationToReadIntoTest to also assert !instanceof PossibleClientChannelCloseException, proving the other tier is never conflated with either client tier. - ./gradlew :ambry-rest:test --tests NettyRequestTest --tests NettyMessageProcessorTest --tests NettyResponseChannelTest :ambry-utils:test --tests UtilsTest -> all pass except 3 pre-existing UtilsTest failures unrelated to this change (InaccessibleObjectException from a JDK17 --add-opens gap in testGetByteBufferInputStreamFromCrcStreamShareMemoryWithNettyByteBuf, confirmed present on this branch before this commit via git stash). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> EOF ) --- .../ambry/rest/NettyMessageProcessor.java | 41 +++++++----- .../com/github/ambry/rest/NettyRequest.java | 38 +++++++++-- .../ambry/rest/NettyMessageProcessorTest.java | 66 +++++++++++-------- .../github/ambry/rest/NettyRequestTest.java | 45 +++++++++++++ .../utils/ClientChannelCloseException.java | 5 +- .../PossibleClientChannelCloseException.java | 45 +++++++++++++ .../java/com/github/ambry/utils/Utils.java | 2 +- .../com/github/ambry/utils/UtilsTest.java | 5 ++ 8 files changed, 196 insertions(+), 51 deletions(-) create mode 100644 ambry-utils/src/main/java/com/github/ambry/utils/PossibleClientChannelCloseException.java diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index e7aeee1307..0be1220430 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -174,14 +174,21 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E try { if (request != null && request.isOpen() && cause instanceof Exception) { nettyMetrics.processorExceptionCaughtCount.inc(); - // NOTE: an IOException reaching this handler for an in-flight request is likely client-rooted (Netty's own - // handling of this client-facing channel, e.g. "connection reset"/"broken pipe" while reading further - // request content) rather than a business/destination write failure (those are reported directly to the - // RestResponseChannel by FrontendRestRequestService/AsyncRequestResponseHandler and never reach this - // pipeline handler). However, this path is intentionally left untagged as ClientChannelCloseException in - // this change - only channelInactive() (below) has been proven exclusively client-rooted with no - // server-initiated code path that can trigger it while request.isOpen() is still true. Tagging this site - // is left as a documented follow-up rather than risking a less-proven exclusivity claim here. + if (cause instanceof IOException) { + // NOTE: an IOException reaching this handler for an in-flight request is likely client-rooted (Netty's + // own handling of this client-facing channel, e.g. "connection reset"/"broken pipe" while reading further + // request content) rather than a business/destination write failure (those are reported directly to the + // RestResponseChannel by FrontendRestRequestService/AsyncRequestResponseHandler and never reach this + // pipeline handler). However, exclusivity isn't proven the way it is for channelInactive() (below) - a + // destination-write failure could in principle propagate here via Netty's implicit exception routing - so + // this is tagged as "possible" (PossibleClientChannelCloseException), not "sure" + // (ClientChannelCloseException). + try { + request.markPossibleClientTermination(); + } catch (Exception e) { + logger.warn("Exception while marking request {} as possibly client-terminated", request.getUri(), e); + } + } onRequestAborted((Exception) cause); } else if (isOpen()) { if (cause instanceof RestServiceException) { @@ -233,8 +240,8 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { nettyConfig.nettyServerIdleTimeSeconds); nettyMetrics.idleConnectionCloseCount.inc(); if (request != null && request.isOpen()) { - // NOTE: idle-timeout is intentionally left untagged as ClientChannelCloseException, pending a - // backpressure-aware follow-up. NettyRequest suspends reads (autoRead=false) while the amount of data + // NOTE: idle-timeout is tagged as "possible" (PossibleClientChannelCloseException), not "sure" + // (ClientChannelCloseException). NettyRequest suspends reads (autoRead=false) while the amount of data // buffered for a slow/backpressured downstream consumer exceeds nettyServerRequestBufferWatermark (see // NettyRequest#continueReadIfPossible); while reads are suspended, no channelRead events can occur no // matter how active the client is, so IdleStateHandler's ALL_IDLE can fire purely because OUR OWN @@ -242,12 +249,16 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { // narrow race: NettyRequest#writeContent unconditionally re-enables autoRead the instant the last client // chunk arrives, before the corresponding destination write is even issued - so a slow destination write // on that final chunk leaves the channel silent in both directions with autoRead==true for the entire - // idle window, a deterministic (not merely racy) false-positive shape. Tagging this as a client + // idle window, a deterministic (not merely racy) false-positive shape. Tagging this as a "sure" client // termination would risk hiding a real server/destination-side slowness problem, violating the "bias - // toward NOT-client on ambiguity" invariant. So this path deliberately falls through to the default - // (untagged) ClosedChannelException, same as before this change; only channelInactive() has been proven - // exclusively client-rooted. A correct follow-up would need to gate on "no destination write currently - // in flight" rather than on autoRead state. + // toward NOT-client on ambiguity" invariant; only channelInactive() has been proven exclusively + // client-rooted. A correct "sure" follow-up would need to gate on "no destination write currently in + // flight" rather than on autoRead state - out of scope here. + try { + request.markPossibleClientTermination(); + } catch (Exception e) { + logger.warn("Exception while marking request {} as possibly client-terminated", request.getUri(), e); + } onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); } else { close(); diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index 000aa8929a..3f9d0df4f7 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java @@ -17,6 +17,7 @@ import com.github.ambry.router.AsyncWritableChannel; import com.github.ambry.router.FutureResult; import com.github.ambry.utils.ClientChannelCloseException; +import com.github.ambry.utils.PossibleClientChannelCloseException; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.DefaultMaxBytesRecvByteBufAllocator; @@ -67,6 +68,8 @@ public class NettyRequest implements RestRequest { static int bufferWatermark = -1; private static final ClosedChannelException CLOSED_CHANNEL_EXCEPTION = new ClosedChannelException(); private static final ClientChannelCloseException CLIENT_CHANNEL_CLOSE_EXCEPTION = new ClientChannelCloseException(); + private static final PossibleClientChannelCloseException POSSIBLE_CLIENT_CHANNEL_CLOSE_EXCEPTION = + new PossibleClientChannelCloseException(); protected final HttpRequest request; protected final Channel channel; @@ -305,10 +308,9 @@ public void close() { /** * Marks this request's pending read (if any) as terminated because of a high-confidence, client-rooted event - * (e.g. the client disconnected, reset the connection, or went idle past the configured timeout). Must be called, - * if at all, before {@link #close()} so that {@link ClientChannelCloseException} - rather than the default - * {@link ClosedChannelException} - is delivered to the pending {@link #readInto} callback. Idempotent and safe to - * call even if there is no pending read. + * (e.g. the client disconnected or reset the connection). Must be called, if at all, before {@link #close()} so + * that {@link ClientChannelCloseException} - rather than the default {@link ClosedChannelException} - is delivered + * to the pending {@link #readInto} callback. Idempotent and safe to call even if there is no pending read. */ void markClientTerminated() { channelException = CLIENT_CHANNEL_CLOSE_EXCEPTION; @@ -325,6 +327,34 @@ void closeDueToClientTermination() { close(); } + /** + * Marks this request's pending read (if any) as terminated because of an event that is plausibly, but not + * confirmably, client-rooted (e.g. an idle timeout, which can equally be caused by a slow destination write; or an + * {@link java.io.IOException} reaching Netty's {@code exceptionCaught}, which is usually but not provably + * client-facing). Must be called, if at all, before {@link #close()} so that + * {@link PossibleClientChannelCloseException} - rather than the default {@link ClosedChannelException} - is + * delivered to the pending {@link #readInto} callback. Does not overwrite an already-set + * {@link #markClientTerminated() high-confidence} tag, so that a "sure" classification is never downgraded to + * "possible" if both were somehow triggered for the same request. Idempotent and safe to call even if there is no + * pending read. + */ + void markPossibleClientTermination() { + if (channelException != CLIENT_CHANNEL_CLOSE_EXCEPTION) { + channelException = POSSIBLE_CLIENT_CHANNEL_CLOSE_EXCEPTION; + } + } + + /** + * Convenience method that marks this request as possibly client-terminated (see + * {@link #markPossibleClientTermination()}) and then closes it, in one call. Use this at call sites that close the + * request directly, so the "mark before close" ordering requirement can never be broken by a future edit that + * reorders or drops one of the two calls. + */ + void closeDueToPossibleClientTermination() { + markPossibleClientTermination(); + close(); + } + @Override public RestRequestMetricsTracker getMetricsTracker() { return restRequestMetricsTracker; diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index bb9dce0812..3018f6043b 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -29,6 +29,7 @@ import com.github.ambry.router.InMemoryRouter; import com.github.ambry.store.MessageInfo; import com.github.ambry.utils.ClientChannelCloseException; +import com.github.ambry.utils.PossibleClientChannelCloseException; import com.github.ambry.utils.TestUtils; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; @@ -63,7 +64,6 @@ import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.channels.ClosedChannelException; import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Collections; @@ -576,18 +576,18 @@ public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exce /** * Verifies that a client idle/stall timeout (the {@link IdleState#ALL_IDLE} branch of - * {@link NettyMessageProcessor#userEventTriggered}) does NOT deliver a {@link ClientChannelCloseException} to the - * pending {@code readInto} callback. This is intentional: {@link NettyRequest} suspends reads (autoRead=false) on - * the channel while a slow/backpressured downstream consumer keeps buffered data above - * {@code nettyServerRequestBufferWatermark} (see {@link NettyRequest#continueReadIfPossible}), so - * {@code ALL_IDLE} can fire purely due to a server/destination-side stall rather than genuine client inactivity. - * Since this ambiguity cannot yet be cleanly disambiguated with high confidence, idle-timeout is left untagged in - * this change (falls back to the default {@link ClosedChannelException}); only {@code channelInactive} has been - * proven exclusively client-rooted. + * {@link NettyMessageProcessor#userEventTriggered}) delivers a {@link PossibleClientChannelCloseException} - not + * the high-confidence {@link ClientChannelCloseException} - to the pending {@code readInto} callback. This is + * intentional: {@link NettyRequest} suspends reads (autoRead=false) on the channel while a slow/backpressured + * downstream consumer keeps buffered data above {@code nettyServerRequestBufferWatermark} (see + * {@link NettyRequest#continueReadIfPossible}), so {@code ALL_IDLE} can fire purely due to a server/destination-side + * stall rather than genuine client inactivity. Since this ambiguity cannot be cleanly disambiguated with high + * confidence, idle-timeout is tagged with the "possible" tier rather than the "sure" tier; only + * {@code channelInactive} has been proven exclusively client-rooted. * @throws Exception */ @Test - public void idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + public void idleTimeoutDeliversPossibleClientTerminationToReadIntoTest() throws Exception { CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); capturingHandler.start(); try { @@ -596,7 +596,7 @@ public void idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest() throws Ex EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); - httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest"); + httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "idleTimeoutDeliversPossibleClientTerminationToReadIntoTest"); httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); channel.writeInbound(httpRequest); @@ -613,11 +613,11 @@ public void idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest() throws Ex callback.awaitCallback(); assertNotNull("readInto callback should have received an exception", callback.exception); assertFalse( - "readInto callback exception must NOT be a ClientChannelCloseException for idle-timeout in this change " - + "(left untagged due to backpressure ambiguity - see channelInactive for the proven client-exclusive " - + "case)", callback.exception instanceof ClientChannelCloseException); - assertTrue("readInto callback exception should still be a ClosedChannelException", - callback.exception instanceof ClosedChannelException); + "readInto callback exception must NOT be the high-confidence ClientChannelCloseException for idle-timeout " + + "(ambiguous - could be a server/destination-side stall) - see channelInactive for the proven " + + "client-exclusive case", callback.exception instanceof ClientChannelCloseException); + assertTrue("readInto callback exception should be a PossibleClientChannelCloseException for idle-timeout", + callback.exception instanceof PossibleClientChannelCloseException); } finally { capturingHandler.shutdown(); } @@ -626,14 +626,15 @@ public void idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest() throws Ex /** * Verifies (and documents) that an {@link IOException} reaching {@link NettyMessageProcessor#exceptionCaught} * while a PUT request is still in-flight (e.g. "connection reset"/"broken pipe" while reading further request - * content from the client) does NOT deliver a {@link ClientChannelCloseException} to the pending {@code readInto} - * callback in this change. This path is a plausible high-confidence client-rooted signal, but is deliberately left - * untagged/out of scope here alongside idle-timeout - only {@code channelInactive} has been proven exclusively - * client-rooted with no possible server-initiated trigger. Tagging this site is a documented follow-up candidate. + * content from the client) delivers a {@link PossibleClientChannelCloseException} - not the high-confidence + * {@link ClientChannelCloseException} - to the pending {@code readInto} callback. This path is a plausible + * client-rooted signal, but exclusivity isn't proven the way it is for {@code channelInactive} - a destination-write + * failure could in principle propagate here via Netty's implicit exception routing - so it is tagged with the + * "possible" tier. * @throws Exception */ @Test - public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + public void exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest() throws Exception { CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); capturingHandler.start(); try { @@ -643,7 +644,7 @@ public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoT HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); httpRequest.headers() - .set(RestUtils.Headers.SERVICE_ID, "exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest"); + .set(RestUtils.Headers.SERVICE_ID, "exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest"); httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); channel.writeInbound(httpRequest); @@ -661,9 +662,12 @@ public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoT callback.awaitCallback(); assertNotNull("readInto callback should have received an exception", callback.exception); assertFalse( - "readInto callback exception must NOT be a ClientChannelCloseException for exceptionCaught's IOException " - + "branch in this change (deliberately left untagged/out of scope - see channelInactive for the " - + "proven client-exclusive case)", callback.exception instanceof ClientChannelCloseException); + "readInto callback exception must NOT be the high-confidence ClientChannelCloseException for " + + "exceptionCaught's IOException branch (exclusivity unproven) - see channelInactive for the proven " + + "client-exclusive case", callback.exception instanceof ClientChannelCloseException); + assertTrue( + "readInto callback exception should be a PossibleClientChannelCloseException for exceptionCaught's " + + "IOException branch", callback.exception instanceof PossibleClientChannelCloseException); } finally { capturingHandler.shutdown(); } @@ -672,11 +676,12 @@ public void exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoT /** * Verifies that a server-side abort (e.g. {@link NettyMessageProcessor#exceptionCaught} triggered by an internal * {@link RestServiceException}, with no client disconnect) does NOT deliver a {@link ClientChannelCloseException} - * to the pending {@code readInto} callback. This is the positive-tagging counterpart to + * or a {@link PossibleClientChannelCloseException} to the pending {@code readInto} callback - it must land in the + * unclassified "other" tier. This is the positive-tagging counterpart to * {@link #channelInactiveDeliversClientTerminationToReadIntoTest()}, and complements - * {@link #idleTimeoutDoesNotDeliverClientTerminationToReadIntoTest()} and - * {@link #exceptionCaughtIOExceptionDoesNotDeliverClientTerminationToReadIntoTest()}, together proving - * server/internal terminations are never mis-tagged as client-rooted. + * {@link #idleTimeoutDeliversPossibleClientTerminationToReadIntoTest()} and + * {@link #exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest()}, together proving + * server/internal terminations are never mis-tagged as client-rooted (at either the "sure" or "possible" tier). * @throws Exception */ @Test @@ -708,6 +713,9 @@ public void serverAbortDoesNotDeliverClientTerminationToReadIntoTest() throws Ex assertNotNull("readInto callback should have received an exception", callback.exception); assertFalse("readInto callback exception must NOT be a ClientChannelCloseException for a server-rooted abort", callback.exception instanceof ClientChannelCloseException); + assertFalse( + "readInto callback exception must NOT be a PossibleClientChannelCloseException for a server-rooted abort", + callback.exception instanceof PossibleClientChannelCloseException); } finally { capturingHandler.shutdown(); } diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java index 1d7e9c8e29..4d78063d1a 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java @@ -21,6 +21,7 @@ import com.github.ambry.router.AsyncWritableChannel; import com.github.ambry.router.FutureResult; import com.github.ambry.utils.ClientChannelCloseException; +import com.github.ambry.utils.PossibleClientChannelCloseException; import com.github.ambry.utils.NettyByteBufLeakHelper; import com.github.ambry.utils.TestUtils; import com.github.ambry.utils.Utils; @@ -333,6 +334,50 @@ public void markClientTerminatedDeliversTypedExceptionTest() throws Exception { callback.exception instanceof ClientChannelCloseException); } + /** + * Tests that {@link NettyRequest#markPossibleClientTermination()} causes {@link NettyRequest#close()} to deliver a + * {@link PossibleClientChannelCloseException} - the ambiguous/lower-confidence tier, distinct from + * {@link ClientChannelCloseException} - to a pending {@link NettyRequest#readInto} callback. Also verifies that a + * "sure" tag set via {@link NettyRequest#markClientTerminated()} is never downgraded by a subsequent + * {@link NettyRequest#markPossibleClientTermination()} call on the same request. + * @throws Exception + */ + @Test + public void markPossibleClientTerminationDeliversTypedExceptionTest() throws Exception { + // Case 1: markPossibleClientTermination() then close() -> pending readInto callback gets + // PossibleClientChannelCloseException, not ClientChannelCloseException. + Channel channel = new MockChannel(); + NettyRequest nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + AsyncWritableChannel writeChannel = new ByteBufferAsyncWritableChannel(); + ReadIntoCallback callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + nettyRequest.markPossibleClientTermination(); + nettyRequest.close(); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("Exception should be a PossibleClientChannelCloseException", + callback.exception instanceof PossibleClientChannelCloseException); + assertFalse("PossibleClientChannelCloseException must not satisfy ClientChannelCloseException instanceof checks " + + "(sibling, not subtype)", callback.exception instanceof ClientChannelCloseException); + assertTrue("PossibleClientChannelCloseException must still be a ClosedChannelException for backward " + + "compatibility", callback.exception instanceof ClosedChannelException); + + // Case 2: markClientTerminated() (sure) followed by markPossibleClientTermination() (possible) must NOT + // downgrade the already-set "sure" tag. + channel = new MockChannel(); + nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + writeChannel = new ByteBufferAsyncWritableChannel(); + callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + nettyRequest.markClientTerminated(); + nettyRequest.markPossibleClientTermination(); + nettyRequest.close(); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("A prior 'sure' tag must not be downgraded to 'possible' by a later " + + "markPossibleClientTermination() call", callback.exception instanceof ClientChannelCloseException); + } + /** * Tests for behavior of multiple operations after {@link NettyRequest#close()} has been called. Some should be ok to * do and some should throw exceptions. diff --git a/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java b/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java index 8e8ceaba01..67eeeed818 100644 --- a/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java +++ b/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java @@ -18,8 +18,9 @@ /** * A {@link ClosedChannelException} thrown specifically when a channel is confirmed, with high confidence, to have - * been closed because of a client-rooted termination (e.g. the client disconnected, reset the connection, or went - * idle for longer than the configured timeout). + * been closed because of a client-rooted termination (e.g. the client disconnected or reset the connection). This is + * the "sure" tier of client-termination classification - see {@link PossibleClientChannelCloseException} for the + * "possible, but unproven" tier (e.g. idle timeout, where a slow destination write can produce the same symptoms). *

* Extending {@link ClosedChannelException} keeps this backward compatible with any existing code that catches or * checks for {@link ClosedChannelException}. Callers that need to distinguish a client-rooted termination from any diff --git a/ambry-utils/src/main/java/com/github/ambry/utils/PossibleClientChannelCloseException.java b/ambry-utils/src/main/java/com/github/ambry/utils/PossibleClientChannelCloseException.java new file mode 100644 index 0000000000..e5b98a4ee2 --- /dev/null +++ b/ambry-utils/src/main/java/com/github/ambry/utils/PossibleClientChannelCloseException.java @@ -0,0 +1,45 @@ +/** + * Copyright 2026 LinkedIn Corp. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + */ +package com.github.ambry.utils; + +import java.nio.channels.ClosedChannelException; + + +/** + * A {@link ClosedChannelException} thrown when a channel closed in a way that is plausibly client-rooted, + * but where an equally plausible, well-understood non-client cause exists, so client-rooted termination cannot be + * confirmed with the same confidence as {@link ClientChannelCloseException}. This is the "possible" tier of + * client-termination classification, sitting between the "sure" tier ({@link ClientChannelCloseException}) and an + * unclassified {@link ClosedChannelException} (e.g. a confirmed internal/server-side error). + *

+ * Known triggers for this exception, and their non-client alternate explanation: + *

+ *

+ * This is intentionally a sibling of {@link ClientChannelCloseException}, not a subtype of it, so a caller checking + * {@code instanceof ClientChannelCloseException} for the high-confidence "sure" tier is never accidentally satisfied + * by a "possible" tier exception. Both extend {@link ClosedChannelException} so any existing code that catches or + * checks for {@link ClosedChannelException} keeps working. Callers can detect this tier via + * {@code instanceof PossibleClientChannelCloseException}, or via {@link Utils#isPossibleClientTermination(Throwable)} + * (which also returns {@code true} for the "sure" tier, and for the pre-existing message-based markers). + */ +public class PossibleClientChannelCloseException extends ClosedChannelException { +} diff --git a/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java b/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java index 516942c768..2ab877149d 100644 --- a/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java +++ b/ambry-utils/src/main/java/com/github/ambry/utils/Utils.java @@ -1286,7 +1286,7 @@ public static void shutDownExecutorService(ExecutorService executorService, long * @return {@code true} this cause indicates a possible early termination from the client. {@code false} otherwise. */ public static boolean isPossibleClientTermination(Throwable cause) { - if (cause instanceof ClientChannelCloseException) { + if (cause instanceof ClientChannelCloseException || cause instanceof PossibleClientChannelCloseException) { return true; } if (cause instanceof IOException) { diff --git a/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java b/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java index f24bf982d5..b53f740bea 100644 --- a/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java +++ b/ambry-utils/src/test/java/com/github/ambry/utils/UtilsTest.java @@ -690,6 +690,11 @@ public void clientTerminationWrapAndRecognizeTest() { exception = new ClientChannelCloseException(); assertTrue("ClientChannelCloseException should be declared as a client termination", Utils.isPossibleClientTermination(exception)); + + // the typed PossibleClientChannelCloseException (the ambiguous tier) should also be recognized directly. + exception = new PossibleClientChannelCloseException(); + assertTrue("PossibleClientChannelCloseException should be declared as a client termination", + Utils.isPossibleClientTermination(exception)); } /** From 81e0af685cbe7d84c687d4e226c4bf3774d71697 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 14:48:32 -0700 Subject: [PATCH 05/10] Document why markPossibleClientTermination's check-then-act is race-safe Per reviewer feedback: the guard is check-then-act, not atomic/CAS, but is safe because channelInactive/exceptionCaught/userEventTriggered all fire on the same Netty channel's single-threaded event loop for a given request, so these calls are never actually concurrent with each other. Comment-only change, no logic touched. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/main/java/com/github/ambry/rest/NettyRequest.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index 3f9d0df4f7..327756ecf3 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java @@ -339,6 +339,10 @@ void closeDueToClientTermination() { * pending read. */ void markPossibleClientTermination() { + // Check-then-act, not atomic/CAS - safe because channelInactive/exceptionCaught/userEventTriggered all fire on + // the same Netty channel's single-threaded event loop for a given request, so these calls are never actually + // concurrent with each other; the volatile field just ensures the eventual invokeCallback() on another thread + // sees the final write. if (channelException != CLIENT_CHANNEL_CLOSE_EXCEPTION) { channelException = POSSIBLE_CLIENT_CHANNEL_CLOSE_EXCEPTION; } From 812351277427f3c7ce7c38b31bf1e048d15368b7 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 15:11:06 -0700 Subject: [PATCH 06/10] Propagate typed client-termination exceptions into response path (Path B) channelInactive (sure tier) and idle-timeout (possible tier) previously delivered the typed ClientChannelCloseException/PossibleClientChannelCloseException only to the readInto() callback (Path A). The separate onRequestAborted(...) call feeding the response-completion path (Path B: NettyResponseChannel#close -> onResponseComplete -> getErrorResponse) still passed a legacy, untyped Utils.convertToClientTerminationException(new ClosedChannelException()) wrap, so a Path B consumer could never distinguish 'sure' from 'possible' via instanceof. Fix: pass the same typed exception to onRequestAborted(...) at both call sites. This is behavior-neutral - Utils.isPossibleClientTermination() already recognizes both types unconditionally (same as the legacy wrap's message always matched), so response status code (BAD_REQUEST) and clientEarlyTerminationCount metrics emitted by NettyResponseChannel#getErrorResponse are unchanged; only the static exception type changes, giving Path B consumers the same instanceof-based tier detection Path A already has. exceptionCaught's IOException branch (the third 'possible'-tier call site) is intentionally left unchanged: it passes the original cause object directly, and wrapping it would make Utils.isPossibleClientTermination() unconditionally true for causes whose message doesn't already match the legacy pattern - a real response-code change (500->400) for those cases, which would violate the non-negotiable 'no response-code behavior change' constraint. This is documented as an explicit scope exclusion. Adds outbound-response-status assertions to the existing channelInactive and idle-timeout NettyMessageProcessorTest cases proving BAD_REQUEST is preserved. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ambry/rest/NettyMessageProcessor.java | 18 +++++++++++---- .../ambry/rest/NettyMessageProcessorTest.java | 23 +++++++++++++++++++ 2 files changed, 37 insertions(+), 4 deletions(-) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index 0be1220430..d0c481875c 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -15,7 +15,8 @@ import com.github.ambry.config.NettyConfig; import com.github.ambry.config.PerformanceConfig; -import com.github.ambry.utils.Utils; +import com.github.ambry.utils.ClientChannelCloseException; +import com.github.ambry.utils.PossibleClientChannelCloseException; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.HttpContent; @@ -28,7 +29,6 @@ import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; import java.io.IOException; -import java.nio.channels.ClosedChannelException; import java.util.HashSet; import java.util.Map; import java.util.Set; @@ -152,7 +152,14 @@ public void channelInactive(ChannelHandlerContext ctx) { } catch (Exception e) { logger.warn("Exception while closing request {} on channelInactive", request.getUri(), e); } - onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); + // Use the same typed "sure" exception delivered to readInto() here too, so a consumer of the + // response-completion path (onRequestAborted -> RestResponseChannel#close/onResponseComplete) can also + // detect this tier via instanceof, not just via the message-based Utils#isPossibleClientTermination check. + // This is behavior-neutral: ClientChannelCloseException is recognized by isPossibleClientTermination() just + // like the previous Utils.convertToClientTerminationException(...) wrap was, so the response status code + // and client-early-termination metrics emitted downstream (see NettyResponseChannel#getErrorResponse) are + // unchanged. + onRequestAborted(new ClientChannelCloseException()); } else { close(); } @@ -259,7 +266,10 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { } catch (Exception e) { logger.warn("Exception while marking request {} as possibly client-terminated", request.getUri(), e); } - onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); + // See the comment on the equivalent onRequestAborted(...) call in channelInactive() above: using the typed + // "possible" exception here is behavior-neutral for the same reason (isPossibleClientTermination() + // recognizes it unconditionally, exactly as it did the previous message-based wrap). + onRequestAborted(new PossibleClientChannelCloseException()); } else { close(); } diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index 3018f6043b..b548a8a516 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -540,6 +540,12 @@ public void channelInactiveClosesInflightRequestTest() throws Exception { * still in-flight delivers a {@link ClientChannelCloseException} to the pending {@code readInto} callback, so * downstream consumers can recognize the termination as client-rooted via {@code instanceof} or * {@link com.github.ambry.utils.Utils#isPossibleClientTermination(Throwable)}. + *

+ * Also verifies that the same typed exception is now delivered to the separate response-completion path + * (via {@code onRequestAborted}) - not just to {@code readInto} - and that doing so is behavior-neutral for any + * error response that manages to be written before the network channel physically closes: the response status + * code (if one is observed at all) is unchanged from what the pre-existing message-based + * {@code Utils.convertToClientTerminationException(...)} wrap would have produced. * @throws Exception */ @Test @@ -569,6 +575,15 @@ public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exce assertNotNull("readInto callback should have received an exception", callback.exception); assertTrue("readInto callback exception should be a ClientChannelCloseException", callback.exception instanceof ClientChannelCloseException); + + // The channel is already closing by the time onRequestAborted runs, so an error response may or may not have + // been written before the physical close completed; if one was, it must still be BAD_REQUEST (unchanged from + // before this typed-exception propagation into onRequestAborted). + Object outboundResponse = channel.readOutbound(); + if (outboundResponse instanceof HttpResponse) { + assertEquals("Response status for a client-rooted abort must remain BAD_REQUEST", HttpResponseStatus.BAD_REQUEST, + ((HttpResponse) outboundResponse).status()); + } } finally { capturingHandler.shutdown(); } @@ -618,6 +633,14 @@ public void idleTimeoutDeliversPossibleClientTerminationToReadIntoTest() throws + "client-exclusive case", callback.exception instanceof ClientChannelCloseException); assertTrue("readInto callback exception should be a PossibleClientChannelCloseException for idle-timeout", callback.exception instanceof PossibleClientChannelCloseException); + + // The idle-timeout channel is still active when onRequestAborted fires, so an error response is written + // before network teardown; assert it is BAD_REQUEST - unchanged from the pre-existing message-based wrap - + // proving the typed-exception propagation into onRequestAborted for this call site is behavior-neutral. + HttpResponse outboundResponse = channel.readOutbound(); + assertNotNull("An error response should have been written for the idle-timeout abort", outboundResponse); + assertEquals("Response status for a possible-client abort must remain BAD_REQUEST", HttpResponseStatus.BAD_REQUEST, + outboundResponse.status()); } finally { capturingHandler.shutdown(); } From 1438b7923783a28f4364a77971e51adb637dead8 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 15:16:29 -0700 Subject: [PATCH 07/10] Fix vacuous Path B assertion in channelInactive test channel.readOutbound() is always null in channelInactiveDeliversClientTerminationToReadIntoTest's scenario (the network channel is already closing by the time onRequestAborted runs, so NettyResponseChannel never gets to actually write an error response) - the previous conditional 'if (outboundResponse instanceof HttpResponse) { assertEquals(...) }' silently never executed its assertion, giving false confidence. Replace with an explicit assertNull(...) plus a javadoc note explaining that this call site's Path B behavior-neutrality is established by code inspection (traced in the PR description), not a runtime assertion - unlike the idle-timeout test, which genuinely exercises its outbound-status assertion since the channel is still active when onRequestAborted fires there. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ambry/rest/NettyMessageProcessorTest.java | 26 ++++++++++--------- 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index b548a8a516..cb4a648abd 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -542,10 +542,17 @@ public void channelInactiveClosesInflightRequestTest() throws Exception { * {@link com.github.ambry.utils.Utils#isPossibleClientTermination(Throwable)}. *

* Also verifies that the same typed exception is now delivered to the separate response-completion path - * (via {@code onRequestAborted}) - not just to {@code readInto} - and that doing so is behavior-neutral for any - * error response that manages to be written before the network channel physically closes: the response status - * code (if one is observed at all) is unchanged from what the pre-existing message-based - * {@code Utils.convertToClientTerminationException(...)} wrap would have produced. + * (via {@code onRequestAborted}) - not just to {@code readInto} - and that doing so is behavior-neutral. Unlike + * the idle-timeout case below, this is NOT independently verified by a runtime assertion in this test: by the + * time {@code onRequestAborted} runs here, the network channel has already begun closing (this test simulates + * the abort via {@code channel.close()} itself), so {@code NettyResponseChannel} never gets to actually write an + * error response to the outbound queue - {@code channel.readOutbound()} is always {@code null} in this scenario. + * Behavior-neutrality for this call site is instead established by code inspection (see the PR description): + * {@code NettyResponseChannel#getErrorResponse} routes through {@code Utils.isPossibleClientTermination(cause)}, + * which recognizes {@link ClientChannelCloseException} unconditionally via {@code instanceof} - identically to + * how the legacy {@code Utils.convertToClientTerminationException(...)} message wrap it replaces always matched + * that same check - so the response status code and {@code clientEarlyTerminationCount} metric this call site + * would have produced are provably unchanged, even though no response is actually observable in this test. * @throws Exception */ @Test @@ -576,14 +583,9 @@ public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exce assertTrue("readInto callback exception should be a ClientChannelCloseException", callback.exception instanceof ClientChannelCloseException); - // The channel is already closing by the time onRequestAborted runs, so an error response may or may not have - // been written before the physical close completed; if one was, it must still be BAD_REQUEST (unchanged from - // before this typed-exception propagation into onRequestAborted). - Object outboundResponse = channel.readOutbound(); - if (outboundResponse instanceof HttpResponse) { - assertEquals("Response status for a client-rooted abort must remain BAD_REQUEST", HttpResponseStatus.BAD_REQUEST, - ((HttpResponse) outboundResponse).status()); - } + // No outbound error response is observable in this scenario - see the class-level javadoc note above for why + // this call site's Path B behavior-neutrality is verified by code inspection instead. + assertNull("No outbound response is expected once the channel is already closing", channel.readOutbound()); } finally { capturingHandler.shutdown(); } From d48baf9dc1cdbbf54aad6faa2d77e85eac3a9f80 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Sun, 9 Aug 2026 15:54:56 -0700 Subject: [PATCH 08/10] Remove unused NettyRequest.closeDueToPossibleClientTermination() Confirmed zero call sites anywhere (main or test code). Both production call sites for the possible tier call markPossibleClientTermination() directly followed by a separate onRequestAborted(...), never this convenience wrapper. Per repo convention: delete unused code outright. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../main/java/com/github/ambry/rest/NettyRequest.java | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index 327756ecf3..81c3971e1b 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java @@ -348,17 +348,6 @@ void markPossibleClientTermination() { } } - /** - * Convenience method that marks this request as possibly client-terminated (see - * {@link #markPossibleClientTermination()}) and then closes it, in one call. Use this at call sites that close the - * request directly, so the "mark before close" ordering requirement can never be broken by a future edit that - * reorders or drops one of the two calls. - */ - void closeDueToPossibleClientTermination() { - markPossibleClientTermination(); - close(); - } - @Override public RestRequestMetricsTracker getMetricsTracker() { return restRequestMetricsTracker; From 4d975352ba047c2342b12585c2c148007c67047b Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Mon, 10 Aug 2026 17:23:04 -0700 Subject: [PATCH 09/10] Harden client-close classification per reviewer feedback (PR #3280) Address four reviewer comments on the 3-tier connection-close classification (sure client / possible client / server-or-unclassified): C1 NettyRequest.readInto: in the already-closed (!isOpen) path, deliver the stored channelException instead of a fresh ClosedChannelException, so a queue-then-read disconnect race preserves the sure/possible classification. C2 RouterUtils.isSystemHealthError: stop suppressing router health metrics for the "possible client" tier. PossibleClientChannelCloseException now counts as a system-health error; only the sure tier (and legacy message heuristics) are suppressed. C3 NettyMessageProcessor.channelInactive: only tag a close as sure client termination when the service is up. During server shutdown (service down), downgrade an in-flight close to the possible tier instead of mislabeling a server-initiated close as high-confidence client termination. C4 NettyResponseChannel.completeRequest: use try/finally so an exception from closeRequest still propagates to onResponseComplete's error handling (and the responseCompleteTasksError metric) while guaranteeing network-close is scheduled, instead of swallowing it as a log-only event. Tests added for each: already-closed classification delivery, router health-error tiering, service-down downgrade, and close-failure surfacing. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../rest/FrontendNettyChannelInitializer.java | 2 +- .../ambry/rest/NettyMessageProcessor.java | 46 +++++++---- .../com/github/ambry/rest/NettyRequest.java | 9 ++- .../ambry/rest/NettyResponseChannel.java | 28 +++---- .../ambry/rest/NettyMessageProcessorTest.java | 77 ++++++++++++++++--- .../github/ambry/rest/NettyRequestTest.java | 57 ++++++++++++++ .../ambry/rest/NettyResponseChannelTest.java | 61 +++++++++++++++ .../com/github/ambry/router/RouterUtils.java | 7 ++ .../github/ambry/router/RouterUtilsTest.java | 7 ++ 9 files changed, 256 insertions(+), 38 deletions(-) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/FrontendNettyChannelInitializer.java b/ambry-rest/src/main/java/com/github/ambry/rest/FrontendNettyChannelInitializer.java index f3f1980176..d04e3ad652 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/FrontendNettyChannelInitializer.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/FrontendNettyChannelInitializer.java @@ -107,6 +107,6 @@ protected void initChannel(SocketChannel ch) throws Exception { } // custom processing class that interfaces with a RestRequestService. pipeline.addLast("processor", - new NettyMessageProcessor(nettyMetrics, nettyConfig, performanceConfig, requestHandler)); + new NettyMessageProcessor(nettyMetrics, nettyConfig, performanceConfig, requestHandler, restServerState)); } } diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index d0c481875c..7eb5719634 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -72,6 +72,7 @@ public class NettyMessageProcessor extends SimpleChannelInboundHandler RestResponseChannel#close/onResponseComplete) can also - // detect this tier via instanceof, not just via the message-based Utils#isPossibleClientTermination check. - // This is behavior-neutral: ClientChannelCloseException is recognized by isPossibleClientTermination() just - // like the previous Utils.convertToClientTerminationException(...) wrap was, so the response status code - // and client-early-termination metrics emitted downstream (see NettyResponseChannel#getErrorResponse) are - // unchanged. - onRequestAborted(new ClientChannelCloseException()); + // Use the same typed exception delivered to readInto() here too, so a consumer of the response-completion path + // (onRequestAborted -> RestResponseChannel#close/onResponseComplete) can also detect this tier via instanceof, + // not just via the message-based Utils#isPossibleClientTermination check. This is behavior-neutral: both typed + // exceptions are recognized by isPossibleClientTermination() just like the previous + // Utils.convertToClientTerminationException(...) wrap was, so the response status code and + // client-early-termination metrics emitted downstream (see NettyResponseChannel#getErrorResponse) are unchanged. + onRequestAborted(serviceUp ? new ClientChannelCloseException() : new PossibleClientChannelCloseException()); } else { close(); } diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index 81c3971e1b..cc9730dfe9 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java @@ -383,7 +383,14 @@ public Future readInto(AsyncWritableChannel asyncWritableChannel, Callback try { if (!isOpen()) { nettyMetrics.requestAlreadyClosedError.inc(); - tempWrapper.invokeCallback(new ClosedChannelException()); + // Deliver the stored channelException (not a fresh ClosedChannelException) so a client-termination + // classification recorded via markClientTerminated()/markPossibleClientTermination() before close() is + // preserved even when the read is registered AFTER the request already closed. This is the realistic + // AsyncRequestResponseHandler queue-then-read ordering: a client can disconnect (channelInactive -> + // closeDueToClientTermination) before the router calls readInto(), at which point close() had no + // callbackWrapper to deliver to. channelException defaults to a plain ClosedChannelException when the + // request was never tagged, so the untagged path is unchanged. + tempWrapper.invokeCallback(channelException); } else if (writeChannel != null) { throw new IllegalStateException("ReadableStreamChannel cannot be read more than once"); } diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java index 73496f135a..206997e4f5 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java @@ -810,22 +810,24 @@ private void completeRequest(boolean closeNetworkChannel, boolean shouldDelay, b // network channel and trigger channelInactive() re-entrantly on this same call stack. NettyMessageProcessor's // channelInactive() uses request.isOpen() to decide whether channel inactivity is client-rooted (as opposed to // this server-initiated completion) - closeRequest() must therefore run first so that check is never fooled by - // a server-initiated close still in progress. Failures here must not prevent the network channel close below - // from being scheduled (that was always attempted first previously), so swallow (and log) any exception. + // a server-initiated close still in progress. The network channel close below must always be scheduled even if + // closeRequest() throws (that was always attempted first previously), so it runs in a finally block; the original + // exception then propagates to the caller's existing failure handling (e.g. onResponseComplete's catch, which + // increments responseCompleteTasksError and fails the writeFuture) rather than being downgraded to a log-only + // event. try { closeRequest(shouldCloseRequest); - } catch (Exception e) { - logger.error("Exception while closing request on channel {}", ctx.channel(), e); - } - if ((closeNetworkChannel || forceClose) && ctx.channel().isOpen()) { - if (shouldDelay && (request != null && request.getRestMethod().equals(RestMethod.POST)) - && this.nettyConfig.nettyServerCloseDelayTimeoutMs > 0) { - nettyMetrics.delayedCloseScheduledCount.inc(); - writeFuture.addListener(DELAYED_CLOSE); - } else { - writeFuture.addListener(ChannelFutureListener.CLOSE); + } finally { + if ((closeNetworkChannel || forceClose) && ctx.channel().isOpen()) { + if (shouldDelay && (request != null && request.getRestMethod().equals(RestMethod.POST)) + && this.nettyConfig.nettyServerCloseDelayTimeoutMs > 0) { + nettyMetrics.delayedCloseScheduledCount.inc(); + writeFuture.addListener(DELAYED_CLOSE); + } else { + writeFuture.addListener(ChannelFutureListener.CLOSE); + } + logger.trace("Requested closing of channel {}", ctx.channel()); } - logger.trace("Requested closing of channel {}", ctx.channel()); } } diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index cb4a648abd..c803dcc759 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -95,6 +95,15 @@ public class NettyMessageProcessorTest { private static final NettyConfig NETTY_CONFIG = new NettyConfig(new VerifiableProperties(new Properties())); private static final PerformanceConfig PERFORMANCE_CONFIG = new PerformanceConfig(new VerifiableProperties(new Properties())); + // A RestServerState that reports the service as up, matching normal request-serving conditions. Only isServiceUp() + // is read by NettyMessageProcessor, so a single shared instance is safe across tests. + private static final RestServerState SERVICE_UP_STATE = createServiceUpState(); + + private static RestServerState createServiceUpState() { + RestServerState state = new RestServerState("/healthCheck"); + state.markServiceUp(); + return state; + } /** * Sets up the mock services that {@link NettyMessageProcessor} can use. @@ -379,7 +388,7 @@ public void continueHeaderPutRequestCloseRaceWithFixTest() throws Exception { DelayedContinueWriteHandler delayHandler = new DelayedContinueWriteHandler(); NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler); + new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(delayHandler, new ChunkedWriteHandler(), processor); HttpHeaders headers = new DefaultHttpHeaders(); @@ -459,7 +468,7 @@ public void continueHeaderPutRequestCloseRaceWithoutDelayTest() throws Exception DelayedContinueWriteHandler delayHandler = new DelayedContinueWriteHandler(); NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler); + new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(delayHandler, new ChunkedWriteHandler(), processor); HttpHeaders headers = new DefaultHttpHeaders(); @@ -512,7 +521,7 @@ public void channelInactiveClosesInflightRequestTest() throws Exception { capturingHandler.start(); try { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); // Send a PUT header only (no LastHttpContent) so the request stays in-flight when we close the channel. @@ -561,7 +570,7 @@ public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exce capturingHandler.start(); try { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); @@ -591,6 +600,56 @@ public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exce } } + /** + * Verifies that when the channel becomes inactive while the service is DOWN (e.g. a server-initiated close during + * {@link RestServer#shutdown()}, where the Netty worker event loop's {@code shutdownGracefully()} closes in-flight + * connections), {@link NettyMessageProcessor#channelInactive} delivers only the "possible" tier + * ({@link PossibleClientChannelCloseException}) - NOT the high-confidence {@link ClientChannelCloseException} - to + * the pending {@code readInto} callback. {@code channelInactive} fires for both client- and server-initiated closes, + * so a server-shutdown abort must not be mislabeled as a proven client termination (which would incorrectly suppress + * router health metrics for a genuine server-side event). The request is still open when the channel goes inactive, + * exactly as in the sure-tier case; only {@code restServerState.isServiceUp()} distinguishes the two. + * @throws Exception + */ + @Test + public void channelInactiveWhileServiceDownDeliversPossibleClientTerminationTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + RestServerState serviceDownState = new RestServerState("/healthCheck"); + // Leave the service marked down (the default) to simulate a server shutdown already in progress. + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, serviceDownState); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers() + .set(RestUtils.Headers.SERVICE_ID, "channelInactiveWhileServiceDownDeliversPossibleClientTerminationTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + assertTrue("RestRequest must be open before channelInactive", capturedRequest.isOpen()); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + // Simulate the channel becoming inactive while the server is shutting down. + channel.close().awaitUninterruptibly(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertFalse("A server-initiated close (service down) must NOT be classified as the high-confidence " + + "ClientChannelCloseException", callback.exception instanceof ClientChannelCloseException); + assertTrue("A server-initiated close (service down) should deliver a PossibleClientChannelCloseException", + callback.exception instanceof PossibleClientChannelCloseException); + assertFalse("RestRequest.isOpen() must be false after channelInactive", capturedRequest.isOpen()); + } finally { + capturingHandler.shutdown(); + } + } + /** * Verifies that a client idle/stall timeout (the {@link IdleState#ALL_IDLE} branch of * {@link NettyMessageProcessor#userEventTriggered}) delivers a {@link PossibleClientChannelCloseException} - not @@ -609,7 +668,7 @@ public void idleTimeoutDeliversPossibleClientTerminationToReadIntoTest() throws capturingHandler.start(); try { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); @@ -664,7 +723,7 @@ public void exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadInt capturingHandler.start(); try { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); @@ -715,7 +774,7 @@ public void serverAbortDoesNotDeliverClientTerminationToReadIntoTest() throws Ex capturingHandler.start(); try { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, SERVICE_UP_STATE); EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); @@ -784,13 +843,13 @@ RestRequest getCapturedRequest() { */ private EmbeddedChannel createChannel() { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, requestHandler); + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, requestHandler, SERVICE_UP_STATE); return new EmbeddedChannel(new ChunkedWriteHandler(), processor); } private EmbeddedChannel createChannel(NettyConfig nettyConfig) { NettyMessageProcessor processor = - new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler); + new NettyMessageProcessor(NETTY_METRICS, nettyConfig, PERFORMANCE_CONFIG, requestHandler, SERVICE_UP_STATE); return new EmbeddedChannel(new ChunkedWriteHandler(), processor); } diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java index 4d78063d1a..4021005157 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyRequestTest.java @@ -378,6 +378,63 @@ public void markPossibleClientTerminationDeliversTypedExceptionTest() throws Exc + "markPossibleClientTermination() call", callback.exception instanceof ClientChannelCloseException); } + /** + * Tests the queue-then-read ordering race: {@link NettyRequest#close()} runs BEFORE + * {@link NettyRequest#readInto} is ever called (as happens when AsyncRequestResponseHandler queues a request and the + * client disconnects before the router registers a read). In that ordering {@code close()} has no pending callback to + * deliver to, so the classification must instead be preserved on the stored {@code channelException} and delivered by + * the later {@code readInto} via its already-closed branch - rather than being discarded in favor of a fresh bare + * {@link ClosedChannelException}. + * @throws Exception + */ + @Test + public void closeBeforeReadIntoDeliversStoredClassificationTest() throws Exception { + // Case 1: markClientTerminated() + close() BEFORE readInto() -> readInto's already-closed branch delivers the + // stored high-confidence ClientChannelCloseException. + Channel channel = new MockChannel(); + NettyRequest nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + nettyRequest.markClientTerminated(); + nettyRequest.close(); + AsyncWritableChannel writeChannel = new ByteBufferAsyncWritableChannel(); + ReadIntoCallback callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("readInto after close must deliver the stored ClientChannelCloseException, not a bare " + + "ClosedChannelException", callback.exception instanceof ClientChannelCloseException); + + // Case 2: markPossibleClientTermination() + close() BEFORE readInto() -> the stored possible tier is delivered. + channel = new MockChannel(); + nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + nettyRequest.markPossibleClientTermination(); + nettyRequest.close(); + writeChannel = new ByteBufferAsyncWritableChannel(); + callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("readInto after close must deliver the stored PossibleClientChannelCloseException", + callback.exception instanceof PossibleClientChannelCloseException); + assertFalse("possible tier must not satisfy ClientChannelCloseException instanceof", + callback.exception instanceof ClientChannelCloseException); + + // Case 3: close() with no classification BEFORE readInto() -> the untagged path is unchanged, delivering a bare + // ClosedChannelException that is neither typed subclass. + channel = new MockChannel(); + nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel); + nettyRequest.close(); + writeChannel = new ByteBufferAsyncWritableChannel(); + callback = new ReadIntoCallback(); + nettyRequest.readInto(writeChannel, callback); + callback.awaitCallback(); + assertNotNull("Callback should have received an exception", callback.exception); + assertTrue("Untagged close must still deliver a ClosedChannelException", + callback.exception instanceof ClosedChannelException); + assertFalse("Untagged close must not be classified as a client termination", + callback.exception instanceof ClientChannelCloseException + || callback.exception instanceof PossibleClientChannelCloseException); + } + /** * Tests for behavior of multiple operations after {@link NettyRequest#close()} has been called. Some should be ok to * do and some should throw exceptions. diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java index 83df050a7b..3f84ecc76b 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java @@ -763,6 +763,54 @@ public void channelInactiveWriteTaggingByMethodTest() throws Exception { } } + /** + * Regression test for the try/finally fix in {@code NettyResponseChannel#completeRequest}. A failure from + * {@code closeRequest()} must be surfaced through the existing {@code onResponseComplete} failure handling + * (incrementing {@code responseCompleteTasksError} and failing the {@code writeFuture}) rather than being downgraded + * to a log-only event as the previous blanket {@code catch} did, while the network-channel close is still guaranteed + * to be scheduled via the {@code finally} block. + *

+ * The channel is disconnected first so {@code maybeSendErrorResponse()} returns {@code false} and + * {@code completeRequest()} runs directly inside {@code onResponseComplete()}'s try block (rather than from an + * error-response write listener), making the propagation observable. With the previous log-only catch, + * {@code responseCompleteTasksError} would remain 0 and the request would stay open (the swallowed close is never + * retried); with the fix it is incremented and the request ends up closed via the catch's recovery + * {@code completeRequest} call. + * @throws Exception + */ + @Test + public void completeRequestSurfacesCloseFailureTest() throws Exception { + MockNettyRequest.throwOnFirstClose = true; + try { + ChunkedWriteHandler chunkedWriteHandler = new ChunkedWriteHandler(); + EmbeddedChannel channel = new EmbeddedChannel(chunkedWriteHandler); + VerifiableProperties verifiableProperties = new VerifiableProperties(new Properties()); + NettyMetrics nettyMetrics = new NettyMetrics(new MetricRegistry()); + // recordMetrics (reached via closeRequest -> evaluatePerformanceAndUpdateMetrics on the recovery path) requires + // the tracker defaults to be initialized; the processor-based harness does this in channelActive. + RestRequestMetricsTracker.setDefaults(new MetricRegistry()); + NettyResponseChannel nettyResponseChannel = + new NettyResponseChannel(new MockChannelHandlerContext(channel), nettyMetrics, + new PerformanceConfig(verifiableProperties), new NettyConfig(verifiableProperties)); + HttpRequest httpRequest = createRequestWithHeaders(HttpMethod.GET, TestingUri.Close.toString()); + MockNettyRequest mockRequest = new MockNettyRequest(httpRequest, channel, nettyMetrics, Collections.emptySet()); + nettyResponseChannel.setRequest(mockRequest); + // Make the channel inactive so maybeSendErrorResponse() returns false and completeRequest() runs directly inside + // onResponseComplete()'s try block. + channel.disconnect().awaitUninterruptibly(); + + long before = nettyMetrics.responseCompleteTasksError.getCount(); + nettyResponseChannel.onResponseComplete(new RuntimeException("driver exception")); + + assertEquals("closeRequest() failure must be surfaced as responseCompleteTasksError, not swallowed", before + 1, + nettyMetrics.responseCompleteTasksError.getCount()); + assertFalse("Request must still end up closed via the recovery path after the injected close() failure", + mockRequest.isOpen()); + } finally { + MockNettyRequest.throwOnFirstClose = false; + } + } + /** * Regression test for the ordering fix in {@code NettyResponseChannel#completeRequest}. That method now closes * the request (flipping {@link NettyRequest#isOpen()} to {@code false}) before it schedules the listener that @@ -1852,6 +1900,10 @@ class MockNettyRequest extends NettyRequest { static long roundTripTime = 1L; static long timeToFirstByte = 1L; static RestRequestMetricsTracker mockTracker; + // When true, the first call to close() throws a RuntimeException before delegating to super.close(); subsequent + // calls behave normally. Used to exercise NettyResponseChannel#completeRequest's failure handling for closeRequest(). + static boolean throwOnFirstClose = false; + private boolean firstCloseThrown = false; MockNettyRequest(HttpRequest request, Channel channel, NettyMetrics metrics, Set parameters) throws Exception { @@ -1862,6 +1914,15 @@ class MockNettyRequest extends NettyRequest { mockTracker.nioMetricsTracker.markRequestReceived(); } + @Override + public void close() { + if (throwOnFirstClose && !firstCloseThrown) { + firstCloseThrown = true; + throw new RuntimeException("Injected close() failure for test"); + } + super.close(); + } + @Override public long getBytesReceived() { return inboundBytes; diff --git a/ambry-router/src/main/java/com/github/ambry/router/RouterUtils.java b/ambry-router/src/main/java/com/github/ambry/router/RouterUtils.java index 6f31c976e3..75b7ff309a 100644 --- a/ambry-router/src/main/java/com/github/ambry/router/RouterUtils.java +++ b/ambry-router/src/main/java/com/github/ambry/router/RouterUtils.java @@ -37,6 +37,7 @@ import com.github.ambry.server.ServerErrorCode; import com.github.ambry.utils.NettyByteBufDataInputStream; import com.github.ambry.utils.Pair; +import com.github.ambry.utils.PossibleClientChannelCloseException; import com.github.ambry.utils.Utils; import io.netty.buffer.ByteBufInputStream; import java.io.DataInputStream; @@ -116,6 +117,12 @@ static boolean isSystemHealthError(Exception exception) { isInternalError = true; break; } + } else if (exception instanceof PossibleClientChannelCloseException) { + // The "possible" tier deliberately includes causes that can be server- or dependency-rooted (e.g. a stalled + // downstream write surfacing as an idle timeout, or a destination-side I/O failure). Suppressing router health + // metrics for these would hide genuine server problems, so treat them as system health errors here. Only the + // high-confidence ClientChannelCloseException tier (handled below via isPossibleClientTermination) suppresses. + isSystemHealthError = true; } else if (Utils.isPossibleClientTermination(exception)) { isSystemHealthError = false; } diff --git a/ambry-router/src/test/java/com/github/ambry/router/RouterUtilsTest.java b/ambry-router/src/test/java/com/github/ambry/router/RouterUtilsTest.java index df41dd009b..6e0ec537db 100644 --- a/ambry-router/src/test/java/com/github/ambry/router/RouterUtilsTest.java +++ b/ambry-router/src/test/java/com/github/ambry/router/RouterUtilsTest.java @@ -29,7 +29,9 @@ import com.github.ambry.config.VerifiableProperties; import com.github.ambry.network.RequestInfo; import com.github.ambry.network.ResponseInfo; +import com.github.ambry.utils.ClientChannelCloseException; import com.github.ambry.utils.Pair; +import com.github.ambry.utils.PossibleClientChannelCloseException; import com.github.ambry.utils.Utils; import java.util.Arrays; import java.util.Properties; @@ -118,6 +120,11 @@ public void testSystemHealthErrorInterpretation() { } Assert.assertTrue(RouterUtils.isSystemHealthError(new Exception())); Assert.assertFalse(RouterUtils.isSystemHealthError(Utils.convertToClientTerminationException(new Exception()))); + // The high-confidence "sure client" tier suppresses router health metrics (it is a genuine client disconnect). + Assert.assertFalse(RouterUtils.isSystemHealthError(new ClientChannelCloseException())); + // The "possible client" tier must NOT suppress: it can be server/dependency-rooted (e.g. a stalled downstream + // write or destination-side I/O failure), so counting it as a system health error avoids hiding real problems. + Assert.assertTrue(RouterUtils.isSystemHealthError(new PossibleClientChannelCloseException())); } /** From 8165718fa0a71fe1241269397a9e6d06aee9cba7 Mon Sep 17 00:00:00 2001 From: Beijie Xu Date: Mon, 10 Aug 2026 17:52:57 -0700 Subject: [PATCH 10/10] C3: fail safe to possible tier when RestServerState is null Per independent review, flip the null-guard direction in NettyMessageProcessor.channelInactive: when restServerState is unavailable we cannot establish service liveness, so classify the close as the "possible" tier rather than over-claiming a high-confidence ClientChannelCloseException. Over-claiming would suppress router health metrics for what could be a server-side event, contradicting the intent of the change. Null is not expected in production (state is always injected); this pins the fail-safe behavior. Add channelInactiveWithNullServerStateDeliversPossibleClientTerminationTest so the guard is covered rather than dead code. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ambry/rest/NettyMessageProcessor.java | 4 +- .../ambry/rest/NettyMessageProcessorTest.java | 46 +++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java index 7eb5719634..4e12f397f9 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java @@ -160,7 +160,9 @@ public void channelInactive(ChannelHandlerContext ctx) { // Even when up, "sure" means the close originated from our TCP peer (which may be a load balancer/proxy rather // than the end client) - but for router health accounting and response classification that is the meaningful // distinction from a locally-initiated shutdown close. - boolean serviceUp = restServerState == null || restServerState.isServiceUp(); + // If restServerState is unavailable (null), we cannot establish service liveness, so fail safe to the + // "possible" tier rather than over-claiming a sure client termination. + boolean serviceUp = restServerState != null && restServerState.isServiceUp(); try { if (serviceUp) { request.closeDueToClientTermination(); diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java index c803dcc759..2ec21dfed6 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java @@ -650,6 +650,52 @@ public void channelInactiveWhileServiceDownDeliversPossibleClientTerminationTest } } + /** + * Verifies the fail-safe null-guard on {@link NettyMessageProcessor#channelInactive}: when {@link RestServerState} + * is unavailable (null), service liveness cannot be established, so an in-flight channelInactive must fall back to + * the "possible" tier ({@link PossibleClientChannelCloseException}) rather than over-claiming the high-confidence + * {@link ClientChannelCloseException}. This is the safer direction - if we cannot prove the service was up, we must + * not assert a proven client termination (which would suppress router health metrics for what could be a + * server-side event). Null is not expected in production (the state is always injected), but this pins the + * fail-safe behavior of the guard. + * @throws Exception + */ + @Test + public void channelInactiveWithNullServerStateDeliversPossibleClientTerminationTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler, null); + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor); + + HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null); + httpRequest.headers() + .set(RestUtils.Headers.SERVICE_ID, "channelInactiveWithNullServerStateDeliversPossibleClientTerminationTest"); + httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream"); + channel.writeInbound(httpRequest); + + RestRequest capturedRequest = capturingHandler.getCapturedRequest(); + assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest); + assertTrue("RestRequest must be open before channelInactive", capturedRequest.isOpen()); + + ReadIntoCallback callback = new ReadIntoCallback(); + capturedRequest.readInto(new com.github.ambry.commons.ByteBufferAsyncWritableChannel(), callback); + + channel.close().awaitUninterruptibly(); + + callback.awaitCallback(); + assertNotNull("readInto callback should have received an exception", callback.exception); + assertFalse("With no server state, the close must NOT be classified as the high-confidence " + + "ClientChannelCloseException", callback.exception instanceof ClientChannelCloseException); + assertTrue("With no server state, the close should fail safe to a PossibleClientChannelCloseException", + callback.exception instanceof PossibleClientChannelCloseException); + assertFalse("RestRequest.isOpen() must be false after channelInactive", capturedRequest.isOpen()); + } finally { + capturingHandler.shutdown(); + } + } + /** * Verifies that a client idle/stall timeout (the {@link IdleState#ALL_IDLE} branch of * {@link NettyMessageProcessor#userEventTriggered}) delivers a {@link PossibleClientChannelCloseException} - not