* HOW TO RUN: - * 1. Group "manual-thinclient-network-delay" — NOT included in CI. - * 2. Docker container with --cap-add=NET_ADMIN, JDK 21, .m2 mounted. + * 1. Group "manual-http-network-fault" — NOT included in standard CI test suites. + * 2. Runs natively on Linux VMs (with sudo) or in Docker (with --cap-add=NET_ADMIN). * 3. Tests self-manage tc netem (add/remove delay) — no manual intervention. - * 4. See NETWORK_DELAY_TESTING_README.md for full setup and run instructions. + * 4. Tests self-skip if tc is not available (e.g., on Windows or non-privileged Linux). + * 5. See NETWORK_DELAY_TESTING_README.md for full setup and run instructions. *
* DESIGN: * - No creates during tests. One seed item created in beforeClass (via shared container). @@ -70,12 +71,13 @@ public class Http2ConnectionLifecycleTests extends FaultInjectionTestBase { private CosmosAsyncContainer cosmosAsyncContainer; private TestObject seedItem; - private static final String TEST_GROUP = "manual-thinclient-network-delay"; + private static final String TEST_GROUP = "manual-http-network-fault"; // 3 minutes per test — enough for warmup + delay + retries + cross-region failover + recovery read private static final long TEST_TIMEOUT = 180_000; - // Hardcode eth0 — Docker always uses eth0. detectNetworkInterface() fails during active delay - // because `tc qdisc show dev eth0` hangs, and the fallback returns `eth0@if23` which tc rejects. - private static final String NETWORK_INTERFACE = "eth0"; + // Network interface detected at runtime — eth0 for Docker, or the default route interface on CI VMs. + private String networkInterface; + // Prefix for privileged commands: empty string in Docker (runs as root), "sudo " on CI VMs. + private String sudoPrefix; @Factory(dataProvider = "clientBuildersWithGatewayAndHttp2") public Http2ConnectionLifecycleTests(CosmosClientBuilder clientBuilder) { @@ -85,6 +87,29 @@ public Http2ConnectionLifecycleTests(CosmosClientBuilder clientBuilder) { @BeforeClass(groups = {TEST_GROUP}, timeOut = TIMEOUT) public void beforeClass() { + // Detect whether we're running as root (Docker) or need sudo (CI VM) + this.sudoPrefix = "root".equals(System.getProperty("user.name")) ? "" : "sudo "; + + // Detect the default-route network interface + this.networkInterface = detectNetworkInterface(); + logger.info("Network interface: {}, sudo: {}", this.networkInterface, !this.sudoPrefix.isEmpty()); + + // Verify tc (traffic control) is available — fail-fast if not + try { + Process p = Runtime.getRuntime().exec(new String[]{"sh", "-c", sudoPrefix + "tc qdisc show dev " + networkInterface}); + int exit = p.waitFor(); + if (exit != 0) { + try (BufferedReader err = new BufferedReader(new InputStreamReader(p.getErrorStream()))) { + String errMsg = err.readLine(); + fail("tc not available on " + networkInterface + " (exit=" + exit + "): " + errMsg); + } + } + } catch (AssertionError e) { + throw e; + } catch (Exception e) { + fail("tc not available: " + e.getMessage()); + } + System.setProperty("COSMOS.THINCLIENT_ENABLED", "true"); // Seed one item using a temporary client. The shared container is created by @BeforeSuite. @@ -118,7 +143,10 @@ public void beforeMethod() { */ @AfterMethod(groups = {TEST_GROUP}, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) public void afterMethod() { - removeNetworkDelay(); + if (sudoPrefix != null) { + removeNetworkDelay(); + removePacketDrop(); + } safeClose(this.client); this.client = null; this.cosmosAsyncContainer = null; @@ -127,8 +155,12 @@ public void afterMethod() { @AfterClass(groups = {TEST_GROUP}, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) public void afterClass() { - removeNetworkDelay(); + if (sudoPrefix != null) { + removeNetworkDelay(); + removePacketDrop(); + } System.clearProperty("COSMOS.THINCLIENT_ENABLED"); + System.clearProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS"); } // ======================================================================== @@ -265,19 +297,18 @@ private void assertNoGatewayTimeout(CosmosDiagnostics diagnostics, String contex } /** - * Applies a tc netem delay to all outbound traffic on the Docker container's network interface. + * Applies a tc netem delay to all outbound traffic on the network interface. * This delays ALL packets (including TCP handshake, HTTP/2 frames, and TLS records) by the * specified duration, causing reactor-netty's ReadTimeoutHandler to fire on H2 stream channels * when the delay exceeds the configured responseTimeout. * - *
Requires {@code --cap-add=NET_ADMIN} on the Docker container. Fails the test immediately - * if the {@code tc} command is not available or returns a non-zero exit code.
+ *Requires root (Docker with {@code --cap-add=NET_ADMIN}) or passwordless sudo (CI VM). + * Fails the test immediately if {@code tc} command fails.
* * @param delayMs the delay in milliseconds to inject (e.g., 8000 for an 8-second delay) */ private void addNetworkDelay(int delayMs) { - String iface = NETWORK_INTERFACE; - String cmd = String.format("tc qdisc add dev %s root netem delay %dms", iface, delayMs); + String cmd = String.format("%stc qdisc add dev %s root netem delay %dms", sudoPrefix, networkInterface, delayMs); logger.info(">>> Adding network delay: {}", cmd); try { Process p = Runtime.getRuntime().exec(new String[]{"sh", "-c", cmd}); @@ -285,10 +316,10 @@ private void addNetworkDelay(int delayMs) { if (exit != 0) { try (BufferedReader err = new BufferedReader(new InputStreamReader(p.getErrorStream()))) { String errMsg = err.readLine(); - logger.warn("tc add failed (exit={}): {}", exit, errMsg); + fail("tc add failed (exit=" + exit + "): " + errMsg); } } else { - logger.info(">>> Network delay active: {}ms on {}", delayMs, iface); + logger.info(">>> Network delay active: {}ms on {}", delayMs, networkInterface); } } catch (Exception e) { logger.error("Failed to add network delay", e); @@ -297,7 +328,7 @@ private void addNetworkDelay(int delayMs) { } /** - * Removes any tc netem qdisc from the Docker container's network interface, restoring + * Removes any tc netem qdisc from the network interface, restoring * normal network behavior. This is called in {@code finally} blocks after each test and * in {@code @AfterMethod} and {@code @AfterClass} as a safety net. * @@ -305,8 +336,7 @@ private void addNetworkDelay(int delayMs) { * Does not fail the test on error — the priority is cleanup, not assertion. */ private void removeNetworkDelay() { - String iface = NETWORK_INTERFACE; - String cmd = String.format("tc qdisc del dev %s root netem", iface); + String cmd = String.format("%stc qdisc del dev %s root", sudoPrefix, networkInterface); logger.info(">>> Removing network delay: {}", cmd); try { Process p = Runtime.getRuntime().exec(new String[]{"sh", "-c", cmd}); @@ -314,7 +344,10 @@ private void removeNetworkDelay() { if (exit == 0) { logger.info(">>> Network delay removed"); } else { - logger.warn("tc del returned exit={} (may already be removed)", exit); + try (BufferedReader err = new BufferedReader(new InputStreamReader(p.getErrorStream()))) { + String errMsg = err.readLine(); + logger.warn("tc del returned exit={}: {} (may already be removed)", exit, errMsg); + } } } catch (Exception e) { logger.warn("Failed to remove network delay: {}", e.getMessage()); @@ -364,10 +397,10 @@ public void connectionReuseAfterRealNettyTimeout() throws Exception { .isNotNull(); assertContainsGatewayTimeout(delayedDiagnostics, "delayed read"); - // Brief pause to let TCP retransmission settle after netem qdisc deletion - Thread.sleep(1000); + // Brief pause to let TCP retransmission settle after netem qdisc deletion. + Thread.sleep(3000); - // Recovery read — assert no timeout, low latency, and same parent channel + // Recovery read — delay is removed, should succeed cleanly CosmosDiagnostics recoveryDiagnostics = this.performDocumentOperation( this.cosmosAsyncContainer, OperationType.Read, seedItem, false); CosmosDiagnosticsContext recoveryCtx = recoveryDiagnostics.getDiagnosticsContext(); @@ -379,7 +412,7 @@ public void connectionReuseAfterRealNettyTimeout() throws Exception { h2ParentChannelIdBeforeDelay.equals(h2ParentChannelIdAfterDelay), recoveryCtx.getDuration().toMillis()); AssertionsForClassTypes.assertThat(recoveryCtx.getDuration()) - .as("Recovery read should complete within 10s (allows one 6s ReadTimeout retry + TCP stabilization)") + .as("Recovery read should complete within 10s (delay removed, no retries expected)") .isLessThan(Duration.ofSeconds(10)); assertThat(h2ParentChannelIdAfterDelay) .as("H2 parent NioSocketChannel should survive ReadTimeoutException on Http2StreamChannel") @@ -757,4 +790,306 @@ public void parentChannelSurvivesE2ECancelWithoutReadTimeout() throws Exception .as("H2 stream channels are never reused (RFC 9113 §5.1.1) — stream ID should differ from warmup") .isNotEqualTo(warmupStreamChannelId); } + + // ======================================================================== + // Connection Max Lifetime Tests + // ======================================================================== + + /** + * Proves that a connection is rotated after maxLifeTime expires. + * Sets COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS=15 (short lifetime for testing). + * Establishes a connection, captures parentChannelId, waits for the lifetime + background + * sweep interval to elapse, then performs another read and asserts the parentChannelId changed. + */ + @Test(groups = {TEST_GROUP}, timeOut = TEST_TIMEOUT) + public void connectionRotatedAfterMaxLifetimeExpiry() throws Exception { + System.setProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS", "15"); + try { + safeClose(this.client); + this.client = getClientBuilder().buildAsyncClient(); + this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(this.client); + + String initialParentChannelId = establishH2ConnectionAndGetParentChannelId(); + logger.info("Initial parentChannelId: {}", initialParentChannelId); + + long startTime = System.currentTimeMillis(); + long waitMs = 50_000; + String latestParentChannelId = initialParentChannelId; + + while (System.currentTimeMillis() - startTime < waitMs) { + Thread.sleep(5_000); + latestParentChannelId = readAndGetParentChannelId(); + logger.info("Elapsed={}s parentChannelId={} (changed={})", + (System.currentTimeMillis() - startTime) / 1000, + latestParentChannelId, + !latestParentChannelId.equals(initialParentChannelId)); + if (!latestParentChannelId.equals(initialParentChannelId)) { + break; + } + } + + logger.info("RESULT: initial={}, final={}, ROTATED={}", + initialParentChannelId, latestParentChannelId, + !initialParentChannelId.equals(latestParentChannelId)); + assertThat(latestParentChannelId) + .as("After max lifetime (15s + jitter), connection should be rotated to a new parentChannelId") + .isNotEqualTo(initialParentChannelId); + } finally { + System.clearProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS"); + } + } + + /** + * Proves that per-connection jitter staggers eviction — not all connections expire at once. + * Creates multiple H2 parent connections via concurrent requests, sets a short maxLifeTime (15s), + * then observes that connections are evicted at different times. + */ + @Test(groups = {TEST_GROUP}, timeOut = TEST_TIMEOUT) + public void perConnectionJitterStaggersEviction() throws Exception { + System.setProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS", "15"); + try { + safeClose(this.client); + this.client = getClientBuilder().buildAsyncClient(); + this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(this.client); + + int concurrentRequests = 100; + Set+ * Installed on the parent TCP channel (not child H2 streams). Periodically sends + * HTTP/2 PING frames and tracks last ACK timestamp. The connection pool's + * eviction predicate reads {@link #LAST_PING_ACK_NANOS} to determine liveness. + *
+ * Lifecycle: one instance per parent H2 connection. The handler is guarded by
+ * {@link #HANDLER_INSTALLED} to prevent duplicate installation when multiple
+ * child streams are opened on the same parent.
+ */
+public class Http2PingHealthHandler extends ChannelDuplexHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(Http2PingHealthHandler.class);
+
+ /**
+ * Nano timestamp of the last PING ACK received on this parent channel.
+ * Updated ONLY when an Http2PingFrame with ack=true arrives — NOT on arbitrary reads.
+ * Http2FrameCodec propagates PING ACK frames to downstream handlers (addLast position).
+ * When the network is blackholed, no PING ACKs arrive, this goes stale, eviction triggers.
+ */
+ public static final AttributeKey