From 55ed09ef730e86624cdcbf860a60652f712b12d2 Mon Sep 17 00:00:00 2001
From: Abhijeet Mohanty <abhmohanty@microsoft.com>
Date: Sat, 6 Jun 2026 21:37:31 -0400
Subject: [PATCH 1/2] fix(cosmos): HTTP/2 PING handler must observe
 child-stream read activity, not just parent-pipeline frames

Follow-up to #49095.

Root cause: Http2PingHandler is installed on the parent (TCP) channel and only updated lastReadActivityNanos from parent-pipeline channelRead -- which sees PING ACK / SETTINGS / GOAWAY but NOT HEADERS / DATA frames. With reactor-netty's Http2MultiplexHandler in the parent pipeline, all application H2 traffic is demuxed onto per-stream child channels, so a connection serving thousands of QPS still looked idle to the PING handler, causing unnecessary PINGs and measurable P95 overhead under load.

Fix:
- New package-private AttributeKey<AtomicLong> LAST_CHILD_STREAM_READ_ACTIVITY_NANOS seeded on the parent channel in handlerAdded.
- Http2PingCloseRewrapHandler (already installed in every child stream's pipeline) overrides channelReadComplete to stamp the parent attribute via accumulateAndGet(now, Math::max) -- once per read cycle, monotonic.
- Http2PingHandler.maybeSendPing now uses effectiveLastReadActivityNanos = max(field, child attr) for both the idle-threshold check AND the outstanding-PING early-return: if child-stream reads arrived after an outstanding PING, clear pingOutstandingSinceNanos and reset consecutiveFailures (defense against middleboxes that drop PING-ACK but pass application H2 frames).
- Names use lastReadActivity / read activity -- both signals are inbound reads, never writes.

Tests: two new unit tests in Http2PingHandlerTest covering the suppression path and the outstanding-PING reset path; both bypass real time via reflection.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../http/Http2PingHandlerTest.java            | 119 ++++++++++++++++-
 sdk/cosmos/azure-cosmos/CHANGELOG.md          |   1 +
 .../http/Http2PingCloseRewrapHandler.java     |  42 ++++++
 .../implementation/http/Http2PingHandler.java | 123 ++++++++++++++++--
 4 files changed, 270 insertions(+), 15 deletions(-)

diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java
index 004cde861d87..a7403ce5bf57 100644
--- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java
@@ -16,6 +16,7 @@
 import java.lang.reflect.Method;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -25,7 +26,7 @@
  * Covers state transitions that do not require advancing real time -- the handler's
  * timeout / threshold logic reads {@code System.nanoTime()}, so the tests sidestep
  * the clock by pre-setting {@code pingOutstandingSinceNanos} (or
- * {@code lastActivityNanos}) via reflection:
+ * {@code lastReadActivityNanos}) via reflection:
  * <ul>
  *   <li>PING ACK with matching payload resets the failure counter (RFC 9113 §6.7 payload echo).</li>
  *   <li>PING ACK with mismatched payload is ignored (late ACK after timeout cannot mask degradation).</li>
@@ -209,10 +210,14 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 
         ChannelHandlerContext ctx = channel.pipeline().context(handler);
         long longAgo = System.nanoTime() - TimeUnit.SECONDS.toNanos(60);
+        AtomicLong childRead = channel.attr(Http2PingHandler.LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
 
         // First tick: force idle -> PING send attempt -> write fails.
         // Expect consecutiveFailures=1, channel still open, task still scheduled.
-        setField(handler, "lastActivityNanos", longAgo);
+        // Rewind both lastReadActivityNanos AND the child-stream-read-activity attribute
+        // (seeded in handlerAdded to now()) so effectiveLastReadActivityNanos is stale.
+        setField(handler, "lastReadActivityNanos", longAgo);
+        childRead.set(longAgo);
         invokeMaybeSendPing(handler, ctx);
         channel.runPendingTasks(); // run the write + listener on the embedded event loop
 
@@ -222,7 +227,12 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         assertThat((Object) getField(handler, "pingTask")).isNotNull();
 
         // Second tick: write fails again -> threshold reached -> task cancelled, channel closed.
-        setField(handler, "lastActivityNanos", longAgo);
+        // Also rewind lastPingSendNanos so the send-throttle (added with the child-stream
+        // activity fix to prevent rapid-fire PING storms after write failure) does not
+        // block the second send within the same pingInterval.
+        setField(handler, "lastReadActivityNanos", longAgo);
+        setField(handler, "lastPingSendNanos", longAgo);
+        childRead.set(longAgo);
         invokeMaybeSendPing(handler, ctx);
         channel.runPendingTasks();
 
@@ -247,10 +257,17 @@ public void ackTimeout_incrementsConsecutiveFailuresAndClosesAtThreshold() throw
         ChannelHandlerContext ctx = channel.pipeline().context(handler);
 
         long pastTimeout = System.nanoTime() - TimeUnit.SECONDS.toNanos(5);
+        long evenOlder = pastTimeout - TimeUnit.SECONDS.toNanos(1);
 
         // First tick: outstanding PING has aged past timeout -> failures=1, channel still open.
+        // Also rewind lastReadActivityNanos AND the child-stream-read-activity attribute to
+        // BEFORE the outstanding PING so the new "activity after outstanding PING" early-return
+        // (which would otherwise clear the outstanding state) does not fire.
         setField(handler, "pingsSent", 1);
         setField(handler, "pingOutstandingSinceNanos", pastTimeout);
+        setField(handler, "lastReadActivityNanos", evenOlder);
+        AtomicLong childRead = channel.attr(Http2PingHandler.LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+        childRead.set(evenOlder);
         invokeMaybeSendPing(handler, ctx);
         channel.runPendingTasks();
 
@@ -263,6 +280,8 @@ public void ackTimeout_incrementsConsecutiveFailuresAndClosesAtThreshold() throw
         // -> failures=threshold -> task cancelled, channel closed.
         setField(handler, "pingsSent", 2);
         setField(handler, "pingOutstandingSinceNanos", pastTimeout);
+        setField(handler, "lastReadActivityNanos", evenOlder);
+        childRead.set(evenOlder);
         invokeMaybeSendPing(handler, ctx);
         channel.runPendingTasks();
 
@@ -274,6 +293,100 @@ public void ackTimeout_incrementsConsecutiveFailuresAndClosesAtThreshold() throw
         channel.finishAndReleaseAll();
     }
 
+    @Test(groups = "unit")
+    public void childStreamReadActivity_suppressesIdlePing() throws Exception {
+        // Regression test for the v4.81.0-beta.1 fix: with parent-pipeline reads
+        // stale (would breach the idle threshold) but child-stream reads recent
+        // (via the LAST_CHILD_STREAM_READ_ACTIVITY_NANOS attribute updated by
+        // Http2PingCloseRewrapHandler.channelReadComplete on each H2 child stream),
+        // the handler must NOT send a PING.
+        //
+        // Without the fix (Http2PingHandler reading only lastReadActivityNanos), a
+        // saturated connection serving thousands of QPS via H2 child streams would
+        // still look "idle" to the handler -- because HEADERS/DATA frames are
+        // demuxed by Http2MultiplexHandler and never surface on the parent's
+        // channelRead -- and PINGs would fire on every interval, adding measurable
+        // latency overhead under load (the symptom that motivated this PR).
+        Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+        ChannelHandlerContext ctx = channel.pipeline().context(handler);
+
+        // Parent-pipeline read-timestamp is stale (60s ago).
+        long longAgo = System.nanoTime() - TimeUnit.SECONDS.toNanos(60);
+        setField(handler, "lastReadActivityNanos", longAgo);
+
+        // Child-stream read-activity attribute is fresh: simulate Http2PingCloseRewrapHandler
+        // having just stamped the parent attribute after demuxing an inbound HEADERS/DATA batch.
+        AtomicLong childReadActivity = channel.attr(Http2PingHandler.LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+        assertThat(childReadActivity).as("handlerAdded should have seeded the child-read-activity attribute").isNotNull();
+        childReadActivity.set(System.nanoTime());
+
+        invokeMaybeSendPing(handler, ctx);
+        channel.runPendingTasks();
+
+        // PING must NOT have fired: effectiveLastReadActivityNanos = max(stale, fresh) = fresh.
+        assertThat((int) getField(handler, "pingsSent"))
+            .as("recent child-stream reads must suppress idle-PING")
+            .isZero();
+        assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+        assertThat(channel.isOpen()).isTrue();
+
+        channel.finishAndReleaseAll();
+    }
+
+    @Test(groups = "unit")
+    public void childStreamReadActivityAfterOutstandingPing_resetsFailureState() throws Exception {
+        // Regression test for the v4.81.0-beta.1 fix: when a PING is outstanding
+        // (pingOutstandingSinceNanos != 0) AND child-stream reads flow AFTER the
+        // PING was sent, the handler must clear the outstanding state and reset
+        // consecutiveFailures -- the connection is demonstrably healthy (H2 codec is
+        // still delivering application data), even though the PING ACK itself was
+        // never observed on the parent pipeline.
+        //
+        // Without the fix, a middlebox that drops PING-ACK frames (but lets
+        // application H2 frames through) would accumulate consecutiveFailures up to
+        // the threshold and close a healthy, actively-serving connection.
+        Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+        ChannelHandlerContext ctx = channel.pipeline().context(handler);
+
+        // PING sent 5s ago is still outstanding; 2 prior consecutive failures recorded.
+        long pingSentAt = System.nanoTime() - TimeUnit.SECONDS.toNanos(5);
+        setField(handler, "pingsSent", 7);
+        setField(handler, "pingOutstandingSinceNanos", pingSentAt);
+        setField(handler, "consecutiveFailures", 2);
+
+        // Parent-pipeline read-timestamp is even OLDER than the outstanding PING so
+        // it cannot itself trigger the early-return -- proving the reset is driven
+        // by the child-stream attribute specifically.
+        long parentLastRead = System.nanoTime() - TimeUnit.SECONDS.toNanos(10);
+        setField(handler, "lastReadActivityNanos", parentLastRead);
+
+        // Child-stream read-activity stamped AFTER the PING was sent (Http2PingCloseRewrapHandler
+        // saw inbound HEADERS/DATA on a child stream just now).
+        AtomicLong childReadActivity = channel.attr(Http2PingHandler.LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+        assertThat(childReadActivity).isNotNull();
+        childReadActivity.set(System.nanoTime());
+
+        invokeMaybeSendPing(handler, ctx);
+        channel.runPendingTasks();
+
+        // Early-return cleared outstanding state and the failure counter; no new
+        // PING was sent; channel is still healthy and open.
+        assertThat((long) getField(handler, "pingOutstandingSinceNanos"))
+            .as("child-stream reads after outstanding PING must clear pingOutstandingSinceNanos")
+            .isZero();
+        assertThat((int) getField(handler, "consecutiveFailures"))
+            .as("child-stream reads after outstanding PING must reset consecutiveFailures")
+            .isZero();
+        assertThat((int) getField(handler, "pingsSent"))
+            .as("early-return must NOT send a new PING")
+            .isEqualTo(7);
+        assertThat(channel.isOpen()).isTrue();
+
+        channel.finishAndReleaseAll();
+    }
+
     private static void invokeMaybeSendPing(Http2PingHandler handler, ChannelHandlerContext ctx) throws Exception {
         Method m = Http2PingHandler.class.getDeclaredMethod("maybeSendPing", ChannelHandlerContext.class);
         m.setAccessible(true);
diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md
index 56bc01bfba61..e4135c6c931a 100644
--- a/sdk/cosmos/azure-cosmos/CHANGELOG.md
+++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md
@@ -12,6 +12,7 @@
 #### Bugs Fixed
 * Fixed region name normalization for preferred and excluded regions — non-canonical inputs (e.g., `"westus3"`, `"WEST US 3"`) are now mapped to the canonical form. Also fixed a case-sensitive exclude-region check in PPCB reevaluate logic. - See [PR 49090](https://github.com/Azure/azure-sdk-for-java/pull/49090)
 * Fixed `UnsupportedOperationException` when using `readManyByPartitionKeys` for empty pages. - See [PR 49311](https://github.com/Azure/azure-sdk-for-java/pull/49311)
+* Fixed HTTP/2 PING keepalive handler (introduced in [PR 49095](https://github.com/Azure/azure-sdk-for-java/pull/49095)) so it observes child-stream HEADERS/DATA reads via `Http2PingCloseRewrapHandler.channelReadComplete`, preventing spurious PINGs (and spurious closes) on connections actively serving requests through `Http2MultiplexHandler`.
 
 #### Other Changes
 * Added HTTP/2 PING keepalive (default ON) for Gateway service endpoints to detect silently-broken connections. - See [PR 49095](https://github.com/Azure/azure-sdk-for-java/pull/49095)
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingCloseRewrapHandler.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingCloseRewrapHandler.java
index 02f6cd67dbcf..c647abc78c32 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingCloseRewrapHandler.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingCloseRewrapHandler.java
@@ -7,6 +7,8 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 /**
  * Per-request HTTP/2 child-stream handler that translates a parent-TCP-channel close
  * driven by {@link Http2PingHandler} into a typed {@link Http2PingTimeoutChannelClosedException}.
@@ -40,6 +42,46 @@ final class Http2PingCloseRewrapHandler extends ChannelInboundHandlerAdapter {
 
     private Http2PingCloseRewrapHandler() {}
 
+    /**
+     * Records inbound H2 stream reads on the parent channel so the PING handler
+     * can recognize a busy connection as non-idle and skip the PING send.
+     * <p>
+     * {@code channelReadComplete} fires at the end of each inbound read cycle on
+     * the child channel -- so this method stamps the parent attribute exactly once
+     * per batch of HEADERS/DATA frames demuxed by {@link io.netty.handler.codec.http2.Http2MultiplexHandler},
+     * not once per frame. Under normal load this is plenty of granularity to
+     * suppress unnecessary PINGs without becoming a hot-path allocator.
+     * <p>
+     * Edge cases:
+     * <ul>
+     *   <li><b>Backpressure / autoRead off</b>: {@link io.netty.handler.codec.http2.Http2MultiplexHandler} may
+     *       buffer frames in the child's inbound queue and defer dispatch. In that
+     *       case {@code channelReadComplete} may fire later than the wire-arrival
+     *       time, but it WILL fire when the read cycle ends. Worst case: one benign
+     *       extra PING during the brief buffering window.</li>
+     *   <li><b>Non-H2 channels</b>: this handler is only installed on H2 child
+     *       streams (where {@code ch.parent() != null}), so the parent reference
+     *       above is non-null in normal operation. Defensive null-check kept for
+     *       safety if the install topology ever changes.</li>
+     * </ul>
+     * Uses {@code accumulateAndGet(now, Math::max)} so concurrent updates from
+     * sibling child streams remain monotonic (in practice child streams sharing a
+     * parent run on the same event loop, so contention is nil; the monotonic
+     * semantics document intent and future-proof against multi-event-loop
+     * scenarios).
+     */
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        Channel parent = ctx.channel().parent();
+        if (parent != null) {
+            AtomicLong holder = parent.attr(Http2PingHandler.LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+            if (holder != null) {
+                holder.accumulateAndGet(System.nanoTime(), Math::max);
+            }
+        }
+        super.channelReadComplete(ctx);
+    }
+
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         Channel parent = ctx.channel().parent();
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java
index 82782d14c71b..ff483977d68d 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java
@@ -16,6 +16,7 @@
 
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Manual HTTP/2 PING keepalive handler installed on the parent (TCP) channel.
@@ -62,6 +63,27 @@ public class Http2PingHandler extends ChannelDuplexHandler {
     public static final AttributeKey<Boolean> PING_TIMEOUT_CLOSED =
         AttributeKey.valueOf("cosmos.http2PingTimeoutClosed");
 
+    /**
+     * Parent-channel attribute holding the {@link System#nanoTime()} of the most recent
+     * inbound H2 stream read (HEADERS/DATA), as observed by
+     * {@link Http2PingCloseRewrapHandler#channelReadComplete} on each H2 child stream.
+     * <p>
+     * Why this exists: {@link io.netty.handler.codec.http2.Http2MultiplexHandler} dispatches H2 stream frames to
+     * child channels, so request/response read traffic never surfaces on the parent
+     * pipeline's {@link #channelRead}. Without this signal, {@link #lastReadActivityNanos}
+     * only observes PING ACKs / SETTINGS / GOAWAY on the parent, which means a
+     * connection serving thousands of QPS would still look "idle" to the PING handler
+     * and be PINGed unnecessarily on every interval -- adding measurable latency
+     * overhead under load.
+     * <p>
+     * The attribute is initialized in {@link #handlerAdded} and updated via
+     * {@code accumulateAndGet(now, Math::max)} so writes are monotonic. The PING
+     * handler reads it via {@link #effectiveLastReadActivityNanos(ChannelHandlerContext)}
+     * to compute idleness as {@code max(parent-pipeline reads, child-stream reads)}.
+     */
+    static final AttributeKey<AtomicLong> LAST_CHILD_STREAM_READ_ACTIVITY_NANOS =
+        AttributeKey.valueOf("cosmos.http2LastChildStreamReadActivityNanos");
+
     private final long pingIntervalNanos;
     private final long pingTimeoutNanos;
     private final int failureThreshold;
@@ -70,12 +92,28 @@ public class Http2PingHandler extends ChannelDuplexHandler {
     // (handlerAdded, channelRead, scheduled task, writeAndFlush listener), so no
     // synchronization or volatile is needed.
     //
-    // nanoTime of the last connection-level activity (inbound frame or PING send).
-    // Note: H2 stream frames (HEADERS/DATA -- i.e. actual HTTP responses) are
-    // dispatched by Http2MultiplexHandler to child channels and never surface on
-    // the parent pipeline, so this does NOT track request/response traffic; it
-    // tracks PING ACKs, SETTINGS, GOAWAY, and our own PING sends.
-    private long lastActivityNanos;
+    // nanoTime of the last PARENT-PIPELINE inbound read (PING ACK, SETTINGS,
+    // GOAWAY). Writes (including our own PING sends) are intentionally NOT
+    // counted here: the purpose of the PING is to detect a peer that has stopped
+    // responding, so only inbound frames are positive evidence of liveness.
+    //
+    // H2 stream frames (HEADERS/DATA -- actual HTTP request/response payloads)
+    // are dispatched by Http2MultiplexHandler to child channels and never surface
+    // here; child-stream reads are tracked separately via the
+    // LAST_CHILD_STREAM_READ_ACTIVITY_NANOS parent-channel attribute (written by
+    // Http2PingCloseRewrapHandler.channelReadComplete on each child). Read-idleness
+    // is computed as max(this, child-attribute) via effectiveLastReadActivityNanos()
+    // so a saturated parent that's serving thousands of QPS on child streams is
+    // correctly recognized as "not read-idle".
+    private long lastReadActivityNanos;
+
+    // nanoTime when we most recently SENT a PING (0 = never). Used as a throttle
+    // in maybeSendPing so a PING send failure (which clears
+    // pingOutstandingSinceNanos in the writeAndFlush listener) does not race the
+    // next 500ms tick into a rapid-fire PING storm. Kept separate from
+    // lastReadActivityNanos so the "did the peer respond?" signal stays pure.
+    private long lastPingSendNanos;
+
     private long pingOutstandingSinceNanos;  // nanoTime when PING was sent; 0 = no outstanding PING
     private int consecutiveFailures;         // incremented on timeout, reset on ACK
     private int pingsSent;
@@ -90,11 +128,19 @@ public Http2PingHandler(int pingIntervalSeconds, int pingTimeoutSeconds, int fai
         this.pingIntervalNanos = TimeUnit.SECONDS.toNanos(Math.max(1, pingIntervalSeconds));
         this.pingTimeoutNanos = TimeUnit.SECONDS.toNanos(Math.max(1, pingTimeoutSeconds));
         this.failureThreshold = Math.max(1, failureThreshold);
-        this.lastActivityNanos = System.nanoTime();
+        this.lastReadActivityNanos = System.nanoTime();
     }
 
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) {
+        // Seed the child-stream read-activity attribute on the parent channel so
+        // the child-side rewrap handler can update it via accumulateAndGet without
+        // having to null-check on every frame. setIfAbsent is a no-op if a prior
+        // handler (e.g. a re-install after handlerRemoved) already created it.
+        ctx.channel()
+            .attr(LAST_CHILD_STREAM_READ_ACTIVITY_NANOS)
+            .setIfAbsent(new AtomicLong(System.nanoTime()));
+
         // Schedule periodic check -- runs on the channel's event loop (single-threaded, no sync needed)
         // Check at interval/2 (min 500ms) to bound worst-case PING send delay to ~1.5× interval.
         long checkIntervalMs = Math.max(500, TimeUnit.NANOSECONDS.toMillis(pingIntervalNanos) / 2);
@@ -124,7 +170,9 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        lastActivityNanos = System.nanoTime();
+        // Parent-pipeline inbound frame (PING ACK, SETTINGS, GOAWAY). Counts as
+        // read activity: it is positive evidence the peer is alive and responding.
+        lastReadActivityNanos = System.nanoTime();
         if (msg instanceof Http2PingFrame) {
             Http2PingFrame ping = (Http2PingFrame) msg;
             // RFC 9113 §6.7: PING ACK echoes the 8-byte payload we sent.
@@ -163,6 +211,22 @@ private void maybeSendPing(ChannelHandlerContext ctx) {
             return;
         }
 
+        // If a previous PING is still outstanding but actual child-stream reads
+        // have flowed AFTER the PING was sent, the connection is demonstrably
+        // healthy -- the ACK frame itself may have been dropped (e.g. middlebox
+        // filtering PING ACKs, load balancer rebalancing) but the H2 codec is
+        // still delivering application data. Clear the outstanding state to
+        // avoid charging a spurious failure (and eventually a spurious close)
+        // against a connection that is actively serving requests.
+        if (pingOutstandingSinceNanos != 0) {
+            long effective = effectiveLastReadActivityNanos(ctx);
+            if (effective > pingOutstandingSinceNanos) {
+                pingOutstandingSinceNanos = 0;
+                consecutiveFailures = 0;
+                return;
+            }
+        }
+
         // If a previous PING is still outstanding, check whether it has timed out
         if (pingOutstandingSinceNanos != 0) {
             long waitingNanos = System.nanoTime() - pingOutstandingSinceNanos;
@@ -189,10 +253,20 @@ private void maybeSendPing(ChannelHandlerContext ctx) {
             return;
         }
 
-        long idleNanos = System.nanoTime() - lastActivityNanos;
+        // Idleness anchor: latest moment we either heard from the peer (read) or
+        // probed it ourselves (own PING). Including lastPingSendNanos prevents a
+        // PING send FAILURE (which clears pingOutstandingSinceNanos in the
+        // writeAndFlush listener) from racing the next 500ms tick into a
+        // rapid-fire PING storm. lastPingSendNanos is intentionally excluded from
+        // effectiveLastReadActivityNanos() so the "did the peer respond?" signal
+        // used by the outstanding-PING early-return above stays pure.
+        long now = System.nanoTime();
+        long idleAnchor = Math.max(effectiveLastReadActivityNanos(ctx), lastPingSendNanos);
+        long idleNanos = now - idleAnchor;
         if (idleNanos >= pingIntervalNanos) {
             int count = ++pingsSent;
-            pingOutstandingSinceNanos = System.nanoTime();
+            pingOutstandingSinceNanos = now;
+            lastPingSendNanos = now;
             ctx.writeAndFlush(new DefaultHttp2PingFrame(count))
                 .addListener(f -> {
                     if (f.isSuccess()) {
@@ -221,8 +295,6 @@ private void maybeSendPing(ChannelHandlerContext ctx) {
                         }
                     }
                 });
-            // Reset activity timestamp so we don't send another PING immediately
-            lastActivityNanos = System.nanoTime();
         }
     }
 
@@ -233,6 +305,33 @@ private void cancelPingTask() {
         }
     }
 
+    /**
+     * Returns the effective last-READ-activity timestamp for idleness and the
+     * outstanding-PING early-return check. This is the max of
+     * (a) {@link #lastReadActivityNanos} (parent-pipeline inbound frames: PING ACK,
+     *     SETTINGS, GOAWAY) and
+     * (b) the {@link #LAST_CHILD_STREAM_READ_ACTIVITY_NANOS} parent-channel attribute
+     *     value (inbound H2 stream reads -- HEADERS/DATA -- observed by
+     *     {@link Http2PingCloseRewrapHandler#channelReadComplete} on child streams).
+     * <p>
+     * Read-only: writes (our own PING sends) are deliberately NOT included here
+     * because the purpose of the PING is to detect a peer that has stopped
+     * responding -- only inbound frames are positive evidence of liveness. The
+     * separate {@link #lastPingSendNanos} field tracks our own probes for the
+     * send-throttle in {@link #maybeSendPing}; it is combined with this value
+     * there but not here.
+     * <p>
+     * The attribute should never be null in practice (it is seeded in
+     * {@link #handlerAdded}), but defensive null-handling is cheap and avoids an NPE
+     * if the channel was somehow torn down between attribute seeding and the
+     * scheduled task firing.
+     */
+    private long effectiveLastReadActivityNanos(ChannelHandlerContext ctx) {
+        AtomicLong childReadActivity = ctx.channel().attr(LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+        long childNanos = childReadActivity != null ? childReadActivity.get() : 0L;
+        return Math.max(lastReadActivityNanos, childNanos);
+    }
+
     /**
      * Returns {@code true} when the manual HTTP/2 PING keepalive should be active for
      * a client built with the given {@link Http2ConnectionConfig}. This is the single

From 26f1615f914a4d9bd4b766ebb0181ad5a52d123b Mon Sep 17 00:00:00 2001
From: Abhijeet Mohanty <abhmohanty@microsoft.com>
Date: Sun, 7 Jun 2026 13:03:51 -0400
Subject: [PATCH 2/2] perf: install Http2PingCloseRewrapHandler in
 doOnConnected, not doOnRequest

The previous .doOnRequest() chained onto the HTTP/2 HttpClient builder
wrapped every outgoing request's Mono in a Reactor operator. That paid
per-request costs even when the rewrap handler was already installed:

* MonoPeekTerminal wrap + per-request subscriber allocation
  (InternalMonoOperator.subscribe +2.60% in alloc profile)
* per-request BiConsumer lambda invocation
  (Http2Pool\$\...run +1.16% in alloc profile)
* per-request channel pipeline String-key walk via pipeline.get(...)

Long-run async-profiler diff vs v4.80.0 on D2 attributed the PR's
-7.6% QPS / +14.5% mean / +166% tail-max regression primarily to a
+10.2% allocation rate -- with the operators above dominating the delta.

Move the install into the existing .doOnConnected() block, which
already runs once per HTTP/2 child stream (the customHeaderCleaner
install right above it relies on the same contract). The rewrap
handler is @Sharable, so installing once per stream is correct and
amortizes the install + pipeline-walk away from the hot path.

The ch.parent() != null guard is kept so the install only targets
child streams, never the parent TCP channel (which uses
Http2ParentChannelExceptionHandler instead).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../http/ReactorNettyClient.java              | 25 +++++++++++++------
 1 file changed, 17 insertions(+), 8 deletions(-)

diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
index e95989c5aa3c..3e6008405869 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
@@ -223,8 +223,7 @@ private void configureChannelPipelineHandlers() {
                             // Duplicate handler name is the only possible cause.
                         }
                     }
-                }))
-                .doOnRequest((req, conn) -> {
+
                     // Install a @Sharable head-of-pipeline rewrap handler on each H2
                     // child-stream pipeline. When Http2PingHandler closes the parent
                     // (TCP) channel after consecutive PING-ACK timeouts or PING-send
@@ -237,21 +236,31 @@ private void configureChannelPipelineHandlers() {
                     // stack maps that to GATEWAY_HTTP2_PING_TIMEOUT_CHANNEL_CLOSED so
                     // ClientRetryPolicy can suppress region mark-down.
                     //
+                    // doOnConnected fires once per child stream channel (the existing
+                    // customHeaderCleaner install above relies on the same contract), so
+                    // this install is amortized to one-time per stream rather than
+                    // per-request -- avoids the operator-wrap + per-request pipeline
+                    // walk that the prior .doOnRequest()-based install paid on every
+                    // HTTP call.
+                    //
                     // Gate on ch.parent() != null so this only runs on H2 child streams
-                    // (H1.1 connections have null parent and never need the rewrap).
-                    Channel ch = conn.channel();
-                    if (ch.parent() != null && ch.pipeline().get(Http2PingCloseRewrapHandler.HANDLER_NAME) == null) {
+                    // (the parent TCP channel uses Http2ParentChannelExceptionHandler
+                    // installed above; H1.1 connections never enter this branch since
+                    // the entire enclosing block is guarded by isH2Enabled).
+                    Channel ch = connection.channel();
+                    if (ch.parent() != null
+                        && channelPipeline.get(Http2PingCloseRewrapHandler.HANDLER_NAME) == null) {
                         try {
-                            ch.pipeline().addFirst(
+                            channelPipeline.addFirst(
                                 Http2PingCloseRewrapHandler.HANDLER_NAME,
                                 Http2PingCloseRewrapHandler.INSTANCE);
                         } catch (IllegalArgumentException ignored) {
                             // TOCTOU race: between the get()==null check above and addFirst(),
-                            // a concurrent doOnRequest may have installed the handler.
+                            // a concurrent doOnConnected may have installed the handler.
                             // Duplicate handler name is the only possible cause.
                         }
                     }
-                });
+                }));
         }
     }