Improve Http2 ping timeout logic (#4027)

* 1/2 Improve HTTP2/PING timeout logic

Improve PING timeout logic to avoid premature closing of the connection that
can occur due to scheduling or flushing delays.

* 2/2 Improve HTTP2/PING timeout logic

Improve PING timeout logic, by allowing data frame packets to reset the timer.
Since the goal is to identify and tear down a dead connection, receiving
data packets serve as activity as well. This will help in case of scheduling
and network delays if the PING ACK is not received in time.

* Add changelog

* Fix tests based on new ping timeout logic

* Address review comments

Author: akidambisrinivasan

.changes/next-release/bugfix-AWSSDKforJavav2-6454991.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "akidambisrinivasan",
+    "description": "Improve HTTP2/PING timeout logic. Improvements are to reset the ack timers on incoming data packets because it supports liveness of the connection. Also avoiding premature timeouts due to delays in flushing or scheduling."
+}

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/timers/HttpClientApiCallTimeoutTest.java

@@ -56,6 +56,8 @@ public class HttpClientApiCallTimeoutTest {
 
     private AmazonSyncHttpClient httpClient;
 
+    private final static int EPSILON_MILLIS = 10;
+
     @Before
     public void setup() {
         httpClient = testClientBuilder()
@@ -70,7 +72,7 @@ public void successfulResponse_SlowResponseHandler_ThrowsApiCallTimeoutException
                     .willReturn(aResponse().withStatus(200).withBody("{}")));
 
         assertThatThrownBy(() -> requestBuilder().execute(combinedSyncResponseHandler(
-            superSlowResponseHandler(API_CALL_TIMEOUT.toMillis()), null)))
+            superSlowResponseHandler(API_CALL_TIMEOUT.toMillis() + EPSILON_MILLIS), null)))
             .isInstanceOf(ApiCallTimeoutException.class);
     }
 

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/Http2PingHandler.java

@@ -19,8 +19,8 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelPipeline;
-import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http2.DefaultHttp2PingFrame;
 import io.netty.handler.codec.http2.Http2PingFrame;
 import io.netty.util.concurrent.ScheduledFuture;
@@ -37,18 +37,24 @@
  * If a channel is found to be unhealthy, this will invoke {@link ChannelPipeline#fireExceptionCaught(Throwable)}.
  */
 @SdkInternalApi
-public class Http2PingHandler extends SimpleChannelInboundHandler<Http2PingFrame> {
+public class Http2PingHandler extends ChannelInboundHandlerAdapter {
     private static final NettyClientLogger log = NettyClientLogger.getLogger(Http2PingHandler.class);
     private static final Http2PingFrame DEFAULT_PING_FRAME = new DefaultHttp2PingFrame(0);
 
+    /**
+     * Time limit in ms for delays that results in a warning message being printed.
+     */
+    private final long delayWarningTimeLimitMs;
+
     private final long pingTimeoutMillis;
 
     private ScheduledFuture<?> periodicPing;
     private long lastPingSendTime = 0;
-    private long lastPingAckTime = 0;
+    private long lastMsgRcvdTime = 0;
 
     public Http2PingHandler(int pingTimeoutMillis) {
         this.pingTimeoutMillis = pingTimeoutMillis;
+        delayWarningTimeLimitMs = Math.min(100, pingTimeoutMillis / 10);
     }
 
     @Override
@@ -62,7 +68,7 @@ private void start(Protocol protocol, ChannelHandlerContext ctx) {
         if (protocol == Protocol.HTTP2 && periodicPing == null) {
             periodicPing = ctx.channel()
                               .eventLoop()
-                              .scheduleAtFixedRate(() -> doPeriodicPing(ctx.channel()), 0, pingTimeoutMillis, MILLISECONDS);
+                              .schedule(() -> doPeriodicPing(ctx.channel()), 0, MILLISECONDS);
         }
     }
 
@@ -78,33 +84,68 @@ public void channelInactive(ChannelHandlerContext ctx) {
     }
 
     @Override
-    protected void channelRead0(ChannelHandlerContext ctx, Http2PingFrame frame) {
-        if (frame.ack()) {
-            log.debug(ctx.channel(), () -> "Received PING ACK from channel " + ctx.channel());
-            lastPingAckTime = System.currentTimeMillis();
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        // if there is any message read on this connection, it also suggests the channel is
+        // active, extend the time of connection inactiveness for ping timeout logic.
+        lastMsgRcvdTime = System.currentTimeMillis();
+        if (msg instanceof Http2PingFrame) {
+            doChannelRead(ctx, (Http2PingFrame) msg);
         } else {
+            ctx.fireChannelRead(msg);
+        }
+    }
+
+    private void doChannelRead(ChannelHandlerContext ctx, Http2PingFrame frame) {
+        log.debug(ctx.channel(), () -> "Received PING from channel, ack=" + frame.ack());
+        if (!frame.ack()) {
             ctx.fireChannelRead(frame);
         }
     }
 
     private void doPeriodicPing(Channel channel) {
-        if (lastPingAckTime <= lastPingSendTime - pingTimeoutMillis) {
+        if (lastMsgRcvdTime <= lastPingSendTime - pingTimeoutMillis) {
+            log.warn(channel, () -> "PING timeout occurred");
             long timeSinceLastPingSend = System.currentTimeMillis() - lastPingSendTime;
             channelIsUnhealthy(channel, new PingFailedException("Server did not respond to PING after " +
                                                                 timeSinceLastPingSend + "ms (limit: " +
                                                                 pingTimeoutMillis + "ms)"));
         } else {
+            log.debug(channel, () -> "Sending HTTP2/PING frame");
+            long scheduleTime = lastPingSendTime == 0 ? 0 : System.currentTimeMillis() - lastPingSendTime;
+            if (scheduleTime - pingTimeoutMillis > delayWarningTimeLimitMs) {
+                log.warn(channel, () -> "PING timer scheduled after " + scheduleTime + "ms");
+            }
             sendPing(channel);
         }
     }
 
     private void sendPing(Channel channel) {
+        long writeMs = System.currentTimeMillis();
         channel.writeAndFlush(DEFAULT_PING_FRAME).addListener(res -> {
             if (!res.isSuccess()) {
                 log.debug(channel, () -> "Failed to write and flush PING frame to connection", res.cause());
                 channelIsUnhealthy(channel, new PingFailedException("Failed to send PING to the service", res.cause()));
             } else {
+                log.debug(channel, () -> "Successfully flushed PING frame to connection");
                 lastPingSendTime = System.currentTimeMillis();
+                long flushTime = lastPingSendTime - writeMs;
+                if (flushTime > delayWarningTimeLimitMs) {
+                    log.warn(channel, () -> "Flushing PING frame took " + flushTime + "ms");
+                }
+                // ping frame was flushed to the socket, schedule to send the next ping now to avoid premature timeout.
+                //
+                // Scenario we want to avoid is shown below (NOTE: ptm - pingTimeoutMillis, Wx - Write Ping x,
+                // Fx - Flush Ping x, Rx - Receive ack x, T -Timeout)
+                // When timer is scheduled periodically, even though ack2 comes < ptm after being flushed to the socket, we
+                // will still timeout at 2ptm.
+                // 0          1ptm       2ptm       3ptm
+                // |----------|----------|----------|-------> time
+                // W1F1 R1    W2      F2 T  R2
+                // When timer is scheduled after flushing, we allow time for ack to come back and don't prematurely timeout.
+                // 0            ptm1               ptm2
+                // |-|----------|------|----------|-----------> time
+                // W1F1 R1      W2     F2   R2    W3
+                periodicPing = channel.eventLoop().schedule(() -> doPeriodicPing(channel), pingTimeoutMillis, MILLISECONDS);
             }
         });
     }

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/fault/PingTimeoutTest.java

@@ -99,23 +99,37 @@ public void methodTeardown() throws InterruptedException {
         netty = null;
     }
 
+    /**
+     * Default ping timeout value is 5000ms, so when the channel has no activity including
+     * no data packets received or ping acks received for the timeout value, the connection
+     * will be closed. During connection setup for testing, there will be Setting frame and
+     * the SettingsAck frame exchanged which counts as activity on the channel, so the channel
+     * should timeout during the second interval which is 10 seconds.
+     */
     @Test
-    public void pingHealthCheck_null_shouldThrowExceptionAfter5Sec() {
+    public void pingHealthCheck_null_shouldThrowExceptionAfter10Sec() {
         Instant a = Instant.now();
         assertThatThrownBy(() -> makeRequest(null).join())
             .hasMessageContaining("An error occurred on the connection")
             .hasCauseInstanceOf(IOException.class)
             .hasRootCauseInstanceOf(PingFailedException.class);
-        assertThat(Duration.between(a, Instant.now())).isBetween(Duration.ofSeconds(5), Duration.ofSeconds(7));
+        assertThat(Duration.between(a, Instant.now())).isBetween(Duration.ofSeconds(10), Duration.ofSeconds(12));
     }
 
+    /**
+     * Test when ping timeout value is 3000ms. When the channel has no activity including
+     * no data packets received or ping acks received for the timeout value, the connection
+     * will be closed. During connection setup for testing, there will be Settings frame and
+     * the SettingsAck frame exchanged which counts as activity on the channel, so the channel
+     * should timeout during the second interval which is 6 seconds.
+     */
     @Test
-    public void pingHealthCheck_10sec_shouldThrowExceptionAfter10Secs() {
+    public void pingHealthCheck_3sec_shouldThrowExceptionAfter6Secs() {
         Instant a = Instant.now();
-        assertThatThrownBy(() -> makeRequest(Duration.ofSeconds(10)).join()).hasCauseInstanceOf(IOException.class)
+        assertThatThrownBy(() -> makeRequest(Duration.ofSeconds(3)).join()).hasCauseInstanceOf(IOException.class)
                                                                             .hasMessageContaining("An error occurred on the connection")
                                                                             .hasRootCauseInstanceOf(PingFailedException.class);
-        assertThat(Duration.between(a, Instant.now())).isBetween(Duration.ofSeconds(10), Duration.ofSeconds(12));
+        assertThat(Duration.between(a, Instant.now())).isBetween(Duration.ofSeconds(6), Duration.ofSeconds(8));
     }
 
     @Test

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/fault/ServerNotRespondingTest.java

@@ -54,6 +54,7 @@
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -114,8 +115,9 @@ public void connectionNotAckPing_newRequestShouldUseNewConnection() throws Inter
         // First request should succeed
         firstRequest.join();
 
-        // Wait for Ping to close the connection
-        Thread.sleep(200);
+        // Wait for Ping to close the connection, it takes 2 * healthCheckPingPeriod (200ms)
+        // to identify a silent connection to be closed.
+        Thread.sleep(400);
         server.notRespondOnFirstChannel = false;
         sendGetRequest().join();
         assertThat(server.h2ConnectionCount.get()).isEqualTo(2);