diff --git a/docs/TOPOLOGY-QUALIFICATION.md b/docs/TOPOLOGY-QUALIFICATION.md new file mode 100644 index 0000000..2c4ee7f --- /dev/null +++ b/docs/TOPOLOGY-QUALIFICATION.md @@ -0,0 +1,64 @@ +# Redis topology qualification — QQQ #1301 + +The implemented client accepts one explicit host and port per connection. It +does not discover Sentinel primaries or route Redis Cluster slots. Promotion +and endpoint ownership remain infrastructure responsibilities. This follow-up +keeps topology qualification in qbit-redis; collaboration consumes its shared +connection and realtime primitives. + +## Controlled promotion evidence + +`RedisPromotionTest` runs a real Redis 7.4 primary and replica on an isolated +container network. Two independently configured QQQ instances each use the +replica endpoint with a two-slot pool: one subscription and one command slot. +This deliberately tests role transition at an unchanged endpoint, not routing +from an old primary endpoint to a new one. + +The test observes a marker replicated from the primary, rejects writes while +the endpoint is read-only, stops the old primary and verifies writes there +fail. It promotes the replica, verifies both node pools can read/write there, +disconnects both subscription sockets, and requires acknowledged subscriptions +and delivery of a fresh cross-node event. Pool bounds and release of active +connections after shutdown are checked. Recovery from the promotion command +through reconnect and event delivery has a 25-second fixture deadline; the +measured duration is logged. This is not a production recovery-time guarantee. + +The initial focused test passed on October 26, 2026. In the first full run, +promotion through fresh event delivery measured 683 ms, but an existing +heartbeat test timed out waiting eight seconds for its first pulse. The +isolated heartbeat test then passed without changing its deadline or runtime; +the cause is not established. The diagnostic full rerun measured 1,216 ms for +promotion/reconnect/delivery and passed all 25 tests, packaging and coverage +gates. The original timeout remains unexplained. These timing observations are not a +latency distribution or an SLA. Heartbeat failures now include sanitized +transport and pool counters for diagnosis. +No runtime behavior or supported deployment +contract changes merely because this fixture passes. + +## Remaining acceptance + +- Select and qualify the actual HA endpoint provider and its routing behavior, + including existing pooled connections when the primary changes. +- Exercise partitions, old-primary fencing and healing. Stopping a process in + this test does not prove fencing of a live, isolated old primary. +- Verify the session-store's actual create, refresh, revocation and expiry + semantics through failover. A replicated marker is not a session test. +- Qualify TLS/authentication, endpoint changes, subscriber continuity and pool + pressure together in the deployment topology. +- Define host write-readiness when an endpoint is a replica. The current health + check observes connectivity and subscription acknowledgement, not writability. +- Record production-representative recovery timing and data-loss limits. + +Redis replication is asynchronous and permits data-loss windows. The read-only +replica default prevents ordinary accidental replica writes; it is not a +split-brain fencing mechanism. See the official [Redis replication documentation](https://redis.io/docs/latest/operate/oss_and_stack/management/replication/). +Presence availability and durable session correctness require separate evidence. + +Run the focused fixture with: + +```text +./mvnw -Dtest=RedisPromotionTest test +``` + +The test owns and removes its containers and network. It does not connect to +an existing deployment, change host configuration, or promote a live service. diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java new file mode 100644 index 0000000..a6563e1 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java @@ -0,0 +1,165 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2026. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * 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 + * + * https://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. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +package com.kingsrook.qbits.redis; + +import java.time.Duration; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.realtime.RealtimeMessage; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.exceptions.JedisException; +import static org.junit.jupiter.api.Assertions.*; + +/******************************************************************************* + ** Controlled promotion at an explicit endpoint, not discovery or durable sessions. + *******************************************************************************/ +class RedisPromotionTest +{ + /*************************************************************************** + ** Two unchanged node pools can use a promoted replica and resubscribe there. + ***************************************************************************/ + @Test + void explicitReplicaEndpointSupportsControlledPromotion() throws Exception + { + try(var network = Network.newNetwork(); + var primary = new GenericContainer<>("redis:7.4-alpine").withNetwork(network).withNetworkAliases("primary").withExposedPorts(6379); + var replica = new GenericContainer<>("redis:7.4-alpine").withNetwork(network).withExposedPorts(6379) + .withCommand("redis-server", "--replicaof", "primary", "6379")) + { + primary.start(); + replica.start(); + var firstInstance = host(replica); + var secondInstance = host(replica); + var firstRuntime = new RedisRuntimeService(); + var secondRuntime = new RedisRuntimeService(); + try(var oldPrimary = new RedisConnection(new HostAndPort(primary.getHost(), primary.getMappedPort(6379)), + DefaultJedisClientConfig.builder().connectionTimeoutMillis(500).socketTimeoutMillis(500).build(), 2, Duration.ofMillis(500))) + { + try + { + firstRuntime.start(firstInstance); + secondRuntime.start(secondInstance); + var firstConnection = RedisRuntimeService.connection(firstInstance, "realtime"); + var secondConnection = RedisRuntimeService.connection(secondInstance, "realtime"); + assertEquals("OK", oldPrimary.client().set("promotion-marker", "before")); + assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + { + while(!"before".equals(firstConnection.client().get("promotion-marker"))) + { + Thread.sleep(25); + } + }); + assertThrows(JedisException.class, () -> firstConnection.client().set("replica-write", "forbidden")); + assertNull(oldPrimary.client().get("replica-write")); + + var events = new LinkedBlockingQueue(); + try(var first = new RedisRealtimeBus(firstInstance); var second = new RedisRealtimeBus(secondInstance)) + { + second.subscribe(events::add); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(!first.isListening() || !second.isListening()) + { + Thread.sleep(25); + } + }); + primary.stop(); + assertThrows(JedisException.class, () -> oldPrimary.client().set("old-primary-write", "forbidden")); + long promotionStarted = System.nanoTime(); + var promoted = replica.execInContainer("redis-cli", "REPLICAOF", "NO", "ONE"); + assertEquals(0, promoted.getExitCode()); + assertEquals("OK", promoted.getStdout().trim()); + assertEquals("before", secondConnection.client().get("promotion-marker")); + assertEquals("OK", secondConnection.client().set("promotion-marker", "after")); + assertEquals("after", firstConnection.client().get("promotion-marker")); + + var disconnected = replica.execInContainer("redis-cli", "CLIENT", "KILL", "TYPE", "pubsub"); + assertEquals(0, disconnected.getExitCode()); + assertEquals("2", disconnected.getStdout().trim()); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(!first.isListening() || !second.isListening() + || first.metrics().reconnects() == 0 || second.metrics().reconnects() == 0) + { + Thread.sleep(25); + } + }); + events.clear(); + first.publish(new RealtimeMessage().withChannel("table:document").withEvent("after-promotion")); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(true) + { + var message = events.poll(100, TimeUnit.MILLISECONDS); + if(message != null && "after-promotion".equals(message.getEvent())) + { + break; + } + } + }); + var recoveryTime = Duration.ofNanos(System.nanoTime() - promotionStarted); + assertTrue(recoveryTime.compareTo(Duration.ofSeconds(25)) < 0); + java.util.logging.Logger.getLogger(RedisPromotionTest.class.getName()) + .info("Controlled promotion, subscription reconnect and event delivery: " + recoveryTime.toMillis() + " ms"); + for(var connection : java.util.List.of(firstConnection, secondConnection)) + { + assertEquals(2, connection.metrics().maximum()); + assertTrue(connection.metrics().active() <= 2); + assertEquals(0, connection.metrics().waiting()); + } + } + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + { + while(firstConnection.metrics().active() != 0 || secondConnection.metrics().active() != 0) + { + Thread.sleep(25); + } + }); + } + finally + { + firstRuntime.afterApplicationStop(); + secondRuntime.afterApplicationStop(); + } + } + } + } + + + /*************************************************************************** + ** One subscription slot and one command slot per independently configured node. + ***************************************************************************/ + private static QInstance host(GenericContainer endpoint) throws Exception + { + var instance = RedisRuntimeTest.hostInstance(); + var settings = new RedisConnectionConfig().withHost(endpoint.getHost()).withPort(endpoint.getMappedPort(6379)).withTls(false) + .withMaximumConnections(2).withConnectionTimeoutMillis(500).withSocketTimeoutMillis(1000); + new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("realtime", settings) + .withRealtime("realtime", "test", "promotion")).produce(instance); + return (instance); + } +} diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisRealtimeBusTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisRealtimeBusTest.java index 6cdda6c..a27a283 100644 --- a/src/test/java/com/kingsrook/qbits/redis/RedisRealtimeBusTest.java +++ b/src/test/java/com/kingsrook/qbits/redis/RedisRealtimeBusTest.java @@ -88,7 +88,8 @@ class RedisRealtimeBusTest { Thread.sleep(10); } - }); + }, () -> "No heartbeat received; transport=" + bus.metrics() + + ", pool=" + RedisRuntimeService.connection(instance, "realtime").metrics()); // Keep sockets open but suppress server command progress beyond the read deadline. redis.execInContainer("redis-cli", "CLIENT", "PAUSE", "17000", "ALL"); assertTimeoutPreemptively(Duration.ofSeconds(17), () ->