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 aba3d92c3e..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 @@ -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; @@ -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: 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(); } @@ -170,6 +201,21 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E try { if (request != null && request.isOpen() && cause instanceof Exception) { nettyMetrics.processorExceptionCaughtCount.inc(); + 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) { @@ -221,7 +267,29 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) { nettyConfig.nettyServerIdleTimeSeconds); nettyMetrics.idleConnectionCloseCount.inc(); if (request != null && request.isOpen()) { - onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException())); + // 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 + // 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 "sure" client + // termination would risk hiding a real server/destination-side slowness problem, violating the "bias + // 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); + } + // 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/main/java/com/github/ambry/rest/NettyRequest.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyRequest.java index fa8ddb442a..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 @@ -16,6 +16,8 @@ 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 com.github.ambry.utils.PossibleClientChannelCloseException; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.DefaultMaxBytesRecvByteBufAllocator; @@ -65,6 +67,9 @@ 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(); + private static final PossibleClientChannelCloseException POSSIBLE_CLIENT_CHANNEL_CLOSE_EXCEPTION = + new PossibleClientChannelCloseException(); protected final HttpRequest request; protected final Channel channel; @@ -301,6 +306,48 @@ 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 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; + } + + /** + * 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(); + } + + /** + * 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() { + // 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; + } + } + @Override public RestRequestMetricsTracker getMetricsTracker() { return restRequestMetricsTracker; @@ -336,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 8f62919313..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 @@ -805,17 +805,30 @@ 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) { - 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); + // 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. 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); + } 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()); } - 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..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 @@ -28,6 +28,8 @@ 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.PossibleClientChannelCloseException; import com.github.ambry.utils.TestUtils; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; @@ -57,6 +59,8 @@ 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; @@ -91,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. @@ -375,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(); @@ -455,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(); @@ -508,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. @@ -531,6 +544,313 @@ 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)}. + *

+ * 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. 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 + public void channelInactiveDeliversClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + 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); + 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); + + // 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(); + } + } + + /** + * 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 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 + * 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 idleTimeoutDeliversPossibleClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + 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); + httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "idleTimeoutDeliversPossibleClientTerminationToReadIntoTest"); + 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 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); + + // 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(); + } + } + + /** + * 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) 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 exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + 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); + httpRequest.headers() + .set(RestUtils.Headers.SERVICE_ID, "exceptionCaughtIOExceptionDeliversPossibleClientTerminationToReadIntoTest"); + 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 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(); + } + } + + /** + * 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} + * 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 #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 + public void serverAbortDoesNotDeliverClientTerminationToReadIntoTest() throws Exception { + CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler(); + capturingHandler.start(); + try { + NettyMessageProcessor processor = + 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); + 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); + assertFalse( + "readInto callback exception must NOT be a PossibleClientChannelCloseException for a server-rooted abort", + callback.exception instanceof PossibleClientChannelCloseException); + } finally { + capturingHandler.shutdown(); + } + } + /** * {@link RestRequestHandler} that captures the first {@link RestRequest} passed to * {@link #handleRequest(RestRequest, RestResponseChannel)} and does nothing else. Used by @@ -569,13 +889,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 28fceba9a6..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 @@ -20,6 +20,8 @@ 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.PossibleClientChannelCloseException; import com.github.ambry.utils.NettyByteBufLeakHelper; import com.github.ambry.utils.TestUtils; import com.github.ambry.utils.Utils; @@ -280,6 +282,159 @@ 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 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 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 e6f4b325a8..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,105 @@ 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 + * 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 +1500,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; @@ -1794,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 { @@ -1804,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())); } /** 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..67eeeed818 --- /dev/null +++ b/ambry-utils/src/main/java/com/github/ambry/utils/ClientChannelCloseException.java @@ -0,0 +1,31 @@ +/** + * 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 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 + * 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/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 7e394ebc0b..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,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 || cause instanceof PossibleClientChannelCloseException) { + 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..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 @@ -685,6 +685,16 @@ 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)); + + // 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)); } /**