From 2ddd36e011380f77f49ea00a3105befedddf136e Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 07:47:17 -0500 Subject: [PATCH 1/4] test: qualify controlled Redis replica promotion Co-Authored-By: GPT-6 (Codex) --- docs/TOPOLOGY-QUALIFICATION.md | 64 +++++++ .../qbits/redis/RedisPromotionTest.java | 165 ++++++++++++++++++ .../qbits/redis/RedisRealtimeBusTest.java | 3 +- 3 files changed, 231 insertions(+), 1 deletion(-) create mode 100644 docs/TOPOLOGY-QUALIFICATION.md create mode 100644 src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java 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), () -> -- 2.54.0 From 287d5a6ae112419e5a6434f3677127bf20c40e84 Mon Sep 17 00:00:00 2001 From: James Maes Date: Thu, 29 Oct 2026 09:56:36 -0500 Subject: [PATCH 2/4] feat(redis): qualify writable endpoints and pool recovery Co-Authored-By: Codex (GPT-6) --- README.md | 18 ++ docs/TOPOLOGY-QUALIFICATION.md | 62 +++- .../qbits/redis/RedisConnectionConfig.java | 22 ++ .../qbits/redis/RedisHealthCheck.java | 32 ++- .../qbits/redis/RedisEndpointSwitchTest.java | 270 ++++++++++++++++++ .../qbits/redis/RedisWriteHealthTest.java | 106 +++++++ 6 files changed, 500 insertions(+), 10 deletions(-) create mode 100644 src/test/java/com/kingsrook/qbits/redis/RedisEndpointSwitchTest.java create mode 100644 src/test/java/com/kingsrook/qbits/redis/RedisWriteHealthTest.java diff --git a/README.md b/README.md index 8321186..3a3e4ad 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,24 @@ measured capacity recommendations. `connection.metrics()` reports active, idle, waiting and maximum connections. `redis.connections` integrates with QQQ health checks and reports aggregate pool utilization without endpoints or secrets. +Write-dependent hosts can opt in per connection with +`withWriteHealthKeyPrefix("staging:myapp:health:")`. Reserve that prefix exclusively +for health and permit SET on it in the connection's ACL. Each health run issues +one `SET NX PX 1000` with a random key suffix. It never overwrites an existing +key, and expiry removes probe data even if the reply is lost. The prefix is a +bounded ASCII identifier ending in a colon; do not use application data prefixes. +The default is null (connectivity only), preserving read-only and Pub/Sub-only +connections. Health details report the number of successful `writeChecks`. +A failed opted-in probe reports DOWN through the same framework check/history; +no endpoint, key, credentials or provider exception is exposed. + +This is point-in-time SET acceptance, not replication durability, primary fencing, +or proof that every application command/key is authorized. A writable stale +primary can pass; topology fencing must remain an infrastructure guarantee. +The host decides whether this check gates readiness. Do not enable the probe on +an existing ACL until its dedicated prefix and write permission are configured. +This option is currently an unpublished candidate. + Other qbits consume only `com.kingsrook.qbits.redis.api`: `RedisAccess`, `RedisNamespace`, `RedisCache`, `RedisScript`, `RedisScriptExecutor`, `RedisCredentialsProvider` and `RedisTlsProvider`. `RedisAccess` returns cache or diff --git a/docs/TOPOLOGY-QUALIFICATION.md b/docs/TOPOLOGY-QUALIFICATION.md index 2c4ee7f..db9dbaf 100644 --- a/docs/TOPOLOGY-QUALIFICATION.md +++ b/docs/TOPOLOGY-QUALIFICATION.md @@ -35,7 +35,42 @@ transport and pool counters for diagnosis. No runtime behavior or supported deployment contract changes merely because this fixture passes. -## Remaining acceptance +## Stable endpoint switch evidence + +`RedisEndpointSwitchTest` adds a separate controlled infrastructure-switch case. +Two existing QQQ pools and their subscriptions initially connect through one +loopback TCP relay to the primary. The test verifies command sockets alongside +acknowledged subscription sockets, waits for a marker on the replica, stops the +old primary, and promotes the replica. The relay then changes its target and +closes every accepted old connection. The QQQ instances, pools and buses are +not recreated or reconfigured. + +Bounded read probes tolerate only connection errors while the old sockets are +discarded; this is test polling, not automatic retry of application writes. +After recovery, a single new write is read by the other pool and independently +from the replica. Both subscription reconnect counters must advance, and a +fresh cross-node event must arrive. Two-slot pool bounds and release of borrowed +connections on bus shutdown are checked. The fixture closes its loopback relay, +Redis containers and network. The focused run passed with 907 ms measured from +promotion through fresh event delivery; this single observation is not an SLA. +The subsequent full `./mvnw.cmd verify` run passed all 26 tests with zero +failures, errors or skips, plus packaging and JaCoCo gates. Its endpoint-switch +observation was 6,433 ms; the separate unchanged-endpoint promotion measured +851 ms. Preserve this variability rather than treating the faster focused run +as a deployment recovery target. + +An initial test stopped before promotion because it assumed the initial command +sockets would remain separate after subscriptions borrowed them. The corrected +fixture explicitly issues commands after subscription acknowledgement to ensure +both connection types are established before injecting the switch. + +This tests controlled routing with forced old-socket termination and a stopped +old primary. It does not qualify DNS changes while old sockets remain open, +Sentinel/Cluster discovery, live-primary fencing, partition healing, TLS rotation, +or a production HA endpoint provider. Session durability remains a separate +session-store concern. No production routing or retry behavior changed. + +## Remaining deployment qualification - Select and qualify the actual HA endpoint provider and its routing behavior, including existing pooled connections when the primary changes. @@ -45,8 +80,8 @@ contract changes merely because this fixture passes. 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. +- Adopt the opt-in write-health probe below and define how the host uses its + result for readiness. Existing published health remains connectivity-only. - Record production-representative recovery timing and data-loss limits. Redis replication is asynchronous and permits data-loss windows. The read-only @@ -62,3 +97,24 @@ Run the focused fixture with: 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. + +## Opt-in write-health candidate + +`RedisConnectionConfig.withWriteHealthKeyPrefix` enables a write check in the +existing `redis.connections` framework health check. Each configured connection +must accept a unique `SET NX PX 1000` under its dedicated host-owned prefix. +Random suffixes and NX avoid overwriting existing keys; the one-second expiry +also removes data after a lost reply. No deletion command or application key is +used. The default remains connectivity-only for read-only/Pub/Sub consumers. + +The real Redis fixture verifies that PING still succeeds while a replica rejects +the probe, promotion restores UP, denying SET through ACL makes it DOWN, and +restoring SET permission returns UP. It also checks probe expiry and pool release. +The six focused health/runtime/realtime tests pass. Full verification also passed: 28 tests, zero failures/errors/skips, packaging and JaCoCo gates (1:46). Initial test compilation used unsupported admin methods on the +pooled client; the fixture now uses its owned container's redis-cli instead. + +This change is local and unpublished. Kof22 has not adopted it. The host must +reserve and authorize the prefix, select its readiness policy, and qualify the +deployed route. A writable isolated old primary can still pass: this probe does +not establish fencing, replication durability or permission for arbitrary domain +Lua/scripts. Connectivity and write acceptance are distinct from those guarantees. diff --git a/src/main/java/com/kingsrook/qbits/redis/RedisConnectionConfig.java b/src/main/java/com/kingsrook/qbits/redis/RedisConnectionConfig.java index d2a3cd6..fa11b8d 100644 --- a/src/main/java/com/kingsrook/qbits/redis/RedisConnectionConfig.java +++ b/src/main/java/com/kingsrook/qbits/redis/RedisConnectionConfig.java @@ -39,6 +39,7 @@ public class RedisConnectionConfig implements Serializable private Integer maximumWaitMillis = 500; private QCodeReference credentialsProvider = null; private QCodeReference tlsProvider = null; + private String writeHealthKeyPrefix = null; /*************************************************************************** ** Validate bounds without exposing endpoint or credential values in errors. @@ -61,6 +62,10 @@ public class RedisConnectionConfig implements Serializable { errors.add("Redis TLS provider requires TLS to be enabled"); } + if(writeHealthKeyPrefix != null && !writeHealthKeyPrefix.matches("[A-Za-z0-9:_-]{1,128}:")) + { + errors.add("Redis write health prefix must be a bounded identifier ending in a colon"); + } if(connectionTimeoutMillis == null || connectionTimeoutMillis < 1 || socketTimeoutMillis == null || socketTimeoutMillis < 1 || maximumConnections == null || maximumConnections < 1 || maximumWaitMillis == null || maximumWaitMillis < 1) { @@ -220,4 +225,21 @@ public class RedisConnectionConfig implements Serializable tlsProvider = value; return (this); } + + /*************************************************************************** + ** Optional host-owned namespace for one-second write probes; null means ping only. + ***************************************************************************/ + public String getWriteHealthKeyPrefix() + { + return (writeHealthKeyPrefix); + } + + /*************************************************************************** + ** Reserve a health-only prefix allowed by this connection's Redis ACL. + ***************************************************************************/ + public RedisConnectionConfig withWriteHealthKeyPrefix(String value) + { + writeHealthKeyPrefix = value; + return (this); + } } diff --git a/src/main/java/com/kingsrook/qbits/redis/RedisHealthCheck.java b/src/main/java/com/kingsrook/qbits/redis/RedisHealthCheck.java index f82d304..2e3f137 100644 --- a/src/main/java/com/kingsrook/qbits/redis/RedisHealthCheck.java +++ b/src/main/java/com/kingsrook/qbits/redis/RedisHealthCheck.java @@ -27,7 +27,7 @@ import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome; import com.kingsrook.qqq.backend.core.health.HealthDetails; /******************************************************************************* - ** Observe connectivity without returning keys, credentials or provider exceptions. + ** Observe configured connectivity/write checks without exposing keys or credentials. *******************************************************************************/ public class RedisHealthCheck implements HealthCheck { @@ -45,24 +45,42 @@ public class RedisHealthCheck implements HealthCheck int active = 0; int idle = 0; int waiting = 0; + int writeChecks = 0; try { - for(var connection : connections.values()) + var config = input.getQInstance().getQBits().values().stream().map(bit -> bit.getConfig()) + .filter(RedisQBitConfig.class::isInstance).map(RedisQBitConfig.class::cast).findFirst().orElseThrow(); + for(var entry : connections.entrySet()) { + var connection = entry.getValue(); if(input.getDeadline() != null && !java.time.Instant.now().isBefore(input.getDeadline())) { return (HealthCheckOutcome.unknown("Redis health deadline exceeded")); } connection.client().ping(); + String prefix = config.getConnections().get(entry.getKey()).getWriteHealthKeyPrefix(); + if(prefix != null) + { + if(input.getDeadline() != null && !java.time.Instant.now().isBefore(input.getDeadline())) + { + return (HealthCheckOutcome.unknown("Redis health deadline exceeded")); + } + // NX cannot overwrite existing data; expiry also cleans up after a lost reply. + String result = connection.client().set(prefix + java.util.UUID.randomUUID(), "1", + redis.clients.jedis.params.SetParams.setParams().nx().px(1000)); + if(!"OK".equals(result)) + { + return (HealthCheckOutcome.down("A Redis write health probe was not accepted")); + } + writeChecks++; + } var metrics = connection.metrics(); active += metrics.active(); idle += metrics.idle(); waiting += metrics.waiting(); } var details = new HealthDetails().withValue("connections", connections.size()).withValue("active", active) - .withValue("idle", idle).withValue("waiting", waiting); - var config = input.getQInstance().getQBits().values().stream().map(bit -> bit.getConfig()) - .filter(RedisQBitConfig.class::isInstance).map(RedisQBitConfig.class::cast).findFirst().orElseThrow(); + .withValue("idle", idle).withValue("waiting", waiting).withValue("writeChecks", writeChecks); if(config.getRealtimeTopic() != null) { var bus = com.kingsrook.qqq.backend.core.realtime.RealtimeBus.getIfStarted(input.getQInstance()).orElse(null); @@ -74,11 +92,11 @@ public class RedisHealthCheck implements HealthCheck details.withValue("realtimeQueued", metrics.queued()).withValue("realtimeDropped", metrics.dropped()) .withValue("realtimePublishFailures", metrics.publishFailures()).withValue("realtimeReconnects", metrics.reconnects()); } - return (HealthCheckOutcome.up("Redis connections and configured transport are reachable").withDetails(details)); + return (HealthCheckOutcome.up("Redis connections and configured checks succeeded").withDetails(details)); } catch(RuntimeException failure) { - return (HealthCheckOutcome.down("A Redis connection is unavailable")); + return (HealthCheckOutcome.down("A Redis connection or configured probe is unavailable")); } } } diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisEndpointSwitchTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisEndpointSwitchTest.java new file mode 100644 index 0000000..0161256 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisEndpointSwitchTest.java @@ -0,0 +1,270 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2026. Kingsrook, LLC + * 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 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. + */ +package com.kingsrook.qbits.redis; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +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.HostAndPort; +import redis.clients.jedis.exceptions.JedisConnectionException; +import static org.junit.jupiter.api.Assertions.*; + +/******************************************************************************* + ** A stable external endpoint changes target while existing QQQ clients survive. + *******************************************************************************/ +class RedisEndpointSwitchTest +{ + /*************************************************************************** + ** Fence the old process, promote replicated state and reroute existing sockets. + ***************************************************************************/ + @Test + void existingPoolsAndSubscriptionsRecoverAfterEndpointSwitch() 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(); + try(var relay = new Relay(new HostAndPort(primary.getHost(), primary.getMappedPort(6379)))) + { + var firstHost = host(relay.port()); + var secondHost = host(relay.port()); + var firstRuntime = new RedisRuntimeService(); + var secondRuntime = new RedisRuntimeService(); + try + { + firstRuntime.start(firstHost); + secondRuntime.start(secondHost); + var firstPool = RedisRuntimeService.connection(firstHost, "realtime"); + var secondPool = RedisRuntimeService.connection(secondHost, "realtime"); + assertEquals("OK", firstPool.client().set("endpoint-marker", "replicated")); + assertEquals("replicated", secondPool.client().get("endpoint-marker")); + assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + { + while(!replica.execInContainer("redis-cli", "GET", "endpoint-marker").getStdout().trim().equals("replicated")) + { + Thread.sleep(25); + } + }); + var events = new LinkedBlockingQueue(); + try(var firstBus = new RedisRealtimeBus(firstHost); var secondBus = new RedisRealtimeBus(secondHost)) + { + secondBus.subscribe(events::add); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(!firstBus.isListening() || !secondBus.isListening()) + { + Thread.sleep(25); + } + }); + assertEquals("replicated", firstPool.client().get("endpoint-marker")); + assertEquals("replicated", secondPool.client().get("endpoint-marker")); + assertTrue(relay.socketCount() >= 8, "Two command and two subscription TCP pairs must exist before switching"); + primary.stop(); + long started = System.nanoTime(); + var promotion = replica.execInContainer("redis-cli", "REPLICAOF", "NO", "ONE"); + assertEquals(0, promotion.getExitCode()); + assertEquals("OK", promotion.getStdout().trim()); + relay.switchTo(new HostAndPort(replica.getHost(), replica.getMappedPort(6379))); + assertTimeoutPreemptively(Duration.ofSeconds(15), () -> + { + for(var pool : List.of(firstPool, secondPool)) + { + while(true) + { + try + { + assertEquals("replicated", pool.client().get("endpoint-marker")); + break; + } + catch(JedisConnectionException expectedDuringSwitch) + { + Thread.sleep(25); + } + } + } + while(!firstBus.isListening() || !secondBus.isListening() + || firstBus.metrics().reconnects() == 0 || secondBus.metrics().reconnects() == 0) + { + Thread.sleep(25); + } + }); + assertEquals("OK", firstPool.client().set("endpoint-marker", "after-switch")); + assertEquals("after-switch", secondPool.client().get("endpoint-marker")); + assertEquals("after-switch", replica.execInContainer("redis-cli", "GET", "endpoint-marker").getStdout().trim()); + events.clear(); + firstBus.publish(new RealtimeMessage().withChannel("table:document").withEvent("endpoint-switched")); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(true) + { + var event = events.poll(100, TimeUnit.MILLISECONDS); + if(event != null && "endpoint-switched".equals(event.getEvent())) + { + break; + } + } + }); + System.out.println("ENDPOINT_SWITCH_RECOVERY_MS " + Duration.ofNanos(System.nanoTime() - started).toMillis()); + for(var pool : List.of(firstPool, secondPool)) + { + assertEquals(2, pool.metrics().maximum()); + assertTrue(pool.metrics().active() <= 2); + assertEquals(0, pool.metrics().waiting()); + } + } + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + { + while(firstPool.metrics().active() != 0 || secondPool.metrics().active() != 0) + { + Thread.sleep(25); + } + }); + } + finally + { + firstRuntime.afterApplicationStop(); + secondRuntime.afterApplicationStop(); + } + } + } + } + + /*************************************************************************** + ** Independently configured instances share only the stable loopback endpoint. + ***************************************************************************/ + private static QInstance host(int port) throws Exception + { + var instance = RedisRuntimeTest.hostInstance(); + new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("realtime", + new RedisConnectionConfig().withHost("127.0.0.1").withPort(port).withTls(false) + .withMaximumConnections(2).withConnectionTimeoutMillis(500).withSocketTimeoutMillis(1000)) + .withRealtime("realtime", "test", "endpoint-switch")).produce(instance); + return (instance); + } + + /*************************************************************************** + ** Owned test relay; switching closes existing TCP pairs instead of moving them. + ***************************************************************************/ + private static class Relay implements AutoCloseable + { + private final ServerSocket listener; + private final List sockets = new ArrayList<>(); + private HostAndPort target; + private final Thread acceptor; + + /************************************************************************ + ** Bind only loopback and start a disposable accept loop. + ************************************************************************/ + Relay(HostAndPort target) throws IOException + { + this.target = target; + listener = new ServerSocket(0, 16, InetAddress.getLoopbackAddress()); + acceptor = Thread.ofVirtual().start(() -> + { + while(!listener.isClosed()) + { + try + { + Socket downstream = listener.accept(); + synchronized(this) + { + if(listener.isClosed()) + { + downstream.close(); + continue; + } + sockets.add(downstream); + Socket upstream = new Socket(); + sockets.add(upstream); + upstream.connect(new java.net.InetSocketAddress(this.target.getHost(), this.target.getPort()), 500); + Thread.ofVirtual().start(() -> copy(downstream, upstream)); + Thread.ofVirtual().start(() -> copy(upstream, downstream)); + } + } + catch(IOException expectedOnDisconnect) + { + // Expected while the old primary or the owned listener is stopped. + } + } + }); + } + + /************************************************************************ + ** Forward bytes until either endpoint closes; always close both sockets. + ************************************************************************/ + private static void copy(Socket source, Socket destination) + { + try(source; destination) + { + source.getInputStream().transferTo(destination.getOutputStream()); + } + catch(IOException expectedOnDisconnect) + { + // Socket termination is the fault injected by this fixture. + } + } + + /************************************************************************ + ** Return the stable caller-facing endpoint. + ************************************************************************/ + int port() + { + return (listener.getLocalPort()); + } + + /************************************************************************ + ** Count accepted socket pairs before the first switch. + ************************************************************************/ + synchronized int socketCount() + { + return (sockets.size()); + } + + /************************************************************************ + ** Change only infrastructure routing; terminate all old pooled connections. + ************************************************************************/ + synchronized void switchTo(HostAndPort target) throws IOException + { + this.target = target; + for(Socket socket : sockets) + { + socket.close(); + } + sockets.clear(); + } + + /************************************************************************ + ** Stop the owned listener and all accepted connections. + ************************************************************************/ + @Override + public void close() throws Exception + { + listener.close(); + switchTo(target); + acceptor.join(2000); + assertFalse(acceptor.isAlive()); + } + } +} diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisWriteHealthTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisWriteHealthTest.java new file mode 100644 index 0000000..c64e054 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisWriteHealthTest.java @@ -0,0 +1,106 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. 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 com.kingsrook.qqq.backend.core.health.HealthCheckInput; +import com.kingsrook.qqq.backend.core.health.HealthStatus; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.utility.DockerImageName; +import static org.junit.jupiter.api.Assertions.*; + +/******************************************************************************* + ** Real Redis role and ACL changes must not pass an opted-in write health probe. + *******************************************************************************/ +class RedisWriteHealthTest +{ + /*************************************************************************** + ** Ping succeeds on replicas and SET-denied users; write health must fail. + ***************************************************************************/ + @Test + void detectsReadOnlyAndDeniedWritesWithoutPersistentKeys() throws Exception + { + try(var redis = new GenericContainer<>(DockerImageName.parse("redis:7.4-alpine")).withExposedPorts(6379)) + { + redis.start(); + var instance = RedisRuntimeTest.hostInstance(); + var settings = new RedisConnectionConfig().withHost(redis.getHost()).withPort(redis.getMappedPort(6379)).withTls(false) + .withWriteHealthKeyPrefix("fixture:health:"); + new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("cache", settings)).produce(instance); + var runtime = new RedisRuntimeService(); + try + { + runtime.start(instance); + var client = RedisRuntimeService.connection(instance, "cache").client(); + var input = new HealthCheckInput().withQInstance(instance); + var health = new RedisHealthCheck(); + assertEquals(HealthStatus.UP, health.run(input).getStatus()); + for(String key : client.keys("fixture:health:*")) + { + assertTrue(client.pttl(key) <= 1000); + assertTrue(client.pttl(key) > 0); + } + assertEquals("OK", redis.execInContainer("redis-cli", "REPLICAOF", "127.0.0.1", "1").getStdout().trim()); + assertEquals("PONG", client.ping()); + assertEquals(HealthStatus.DOWN, health.run(input).getStatus()); + assertEquals("OK", redis.execInContainer("redis-cli", "REPLICAOF", "NO", "ONE").getStdout().trim()); + assertEquals(HealthStatus.UP, health.run(input).getStatus()); + assertEquals("OK", redis.execInContainer("redis-cli", "ACL", "SETUSER", "default", "-set").getStdout().trim()); + assertEquals("PONG", client.ping()); + assertEquals(HealthStatus.DOWN, health.run(input).getStatus()); + assertEquals("OK", redis.execInContainer("redis-cli", "ACL", "SETUSER", "default", "+set").getStdout().trim()); + assertEquals(HealthStatus.UP, health.run(input).getStatus()); + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + { + while(!client.keys("fixture:health:*").isEmpty()) + { + Thread.sleep(25); + } + }); + assertEquals(0, RedisRuntimeService.connection(instance, "cache").metrics().active()); + } + finally + { + runtime.afterApplicationStop(); + } + } + } + + /*************************************************************************** + ** Probe namespace is explicit and cannot contain glob patterns or URLs. + ***************************************************************************/ + @Test + void validatesProbePrefix() + { + assertNull(new RedisConnectionConfig().getWriteHealthKeyPrefix()); + for(String prefix : java.util.List.of("", "secret@host:", "health*:", "missing-colon", "a".repeat(129) + ":")) + { + var errors = new java.util.ArrayList(); + new RedisConnectionConfig().withWriteHealthKeyPrefix(prefix).validate(errors); + assertFalse(errors.isEmpty()); + } + var errors = new java.util.ArrayList(); + new RedisConnectionConfig().withWriteHealthKeyPrefix("staging:kof22:health:").validate(errors); + assertTrue(errors.isEmpty()); + } +} -- 2.54.0 From 5f8177844c91668929820738082eb7b01ba13af5 Mon Sep 17 00:00:00 2001 From: James Maes Date: Thu, 29 Oct 2026 11:26:19 -0500 Subject: [PATCH 3/4] test: qualify live primary fencing and partition healing --- docs/TOPOLOGY-QUALIFICATION.md | 31 ++++ .../redis/RedisPartitionHealingTest.java | 157 ++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java diff --git a/docs/TOPOLOGY-QUALIFICATION.md b/docs/TOPOLOGY-QUALIFICATION.md index db9dbaf..3f5b062 100644 --- a/docs/TOPOLOGY-QUALIFICATION.md +++ b/docs/TOPOLOGY-QUALIFICATION.md @@ -72,6 +72,37 @@ session-store concern. No production routing or retry behavior changed. ## Remaining deployment qualification +### Live-primary fencing and partition healing (October 29 candidate) + +`RedisPartitionHealingTest` adds an OSS Valkey 9.1.3 primary/replica fixture. +The original primary requires one connected replica with at most one second +of replication lag before accepting writes. The test disconnects the replica +from its Docker network without stopping the primary. An existing QQQ pool +observes `NOREPLICAS` while the same primary remains running and answers PING; +the rejected write does not create a key. The first focused run observed that +fence after 1,640 ms. This is a local observation, not a guaranteed bound. + +Only after that rejection does the fixture promote the detached replica. +It reconnects the network, demotes the old primary to a replica of the new +primary, restores the one-replica write requirement, and verifies new data +replicates back. The unchanged old QQQ endpoint then rejects writes as READONLY. +The bounded QQQ pool is released, and both containers/network are cleaned up. +One focused test passed with no failures, errors or skips in 24.15 seconds. + +The full Windows and containerized-Linux runs subsequently reported command +socket timeouts during healing, alongside timeouts in existing authentication +and promotion tests. Both paths use Docker Desktop published-port forwarding. +An isolated idle-PING diagnostic measured 300–900 ms replies through that path, +but 0–58 ms on an owned direct Docker network. No runtime timeout, test deadline, +or assertion was relaxed. Full qualification remains pending native CI evidence. + +This fixture targets controlled promotion after verified write fencing. It does not +implement automatic discovery, prove fencing for arbitrary partitions, or +promise zero data loss. Writes can be acknowledged during the replication-lag +window before fencing; applications requiring durable revocation cannot infer +session correctness from this presence-oriented availability test. The actual +deployment still needs its endpoint, TLS/ACL, admission and recovery contracts. + - 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 diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java new file mode 100644 index 0000000..56c4f4f --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java @@ -0,0 +1,157 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. 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 org.junit.jupiter.api.Test; +import org.testcontainers.DockerClientFactory; +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.JedisDataException; +import static org.junit.jupiter.api.Assertions.*; + +/******************************************************************************* + ** Controlled fail-closed promotion: the old primary remains alive but fenced. + *******************************************************************************/ +class RedisPartitionHealingTest +{ + /*************************************************************************** + ** Losing replication stops old writes before promotion; healing demotes it. + ***************************************************************************/ + @Test + void liveOldPrimaryRejectsWritesAcrossPartitionAndHealing() throws Exception + { + try(var network = Network.newNetwork(); + var primary = new GenericContainer<>("valkey/valkey:9.1.3-alpine") + .withNetwork(network).withNetworkAliases("primary").withExposedPorts(6379) + .withCommand("valkey-server", "--min-replicas-to-write", "1", "--min-replicas-max-lag", "1"); + var replica = new GenericContainer<>("valkey/valkey:9.1.3-alpine") + .withNetwork(network).withExposedPorts(6379) + .withCommand("valkey-server", "--replicaof", "primary", "6379")) + { + primary.start(); + replica.start(); + try(var oldPool = connection(primary)) + { + assertTimeoutPreemptively(Duration.ofSeconds(20), () -> + { + while(true) + { + try + { + assertEquals("OK", oldPool.client().set("partition-marker", "before")); + break; + } + catch(JedisDataException waitingForReplica) + { + assertTrue(waitingForReplica.getMessage().contains("NOREPLICAS")); + Thread.sleep(25); + } + } + while(!"before".equals(command(replica, "GET", "partition-marker"))) + { + Thread.sleep(25); + } + }); + + var docker = DockerClientFactory.instance().client(); + docker.disconnectFromNetworkCmd().withContainerId(replica.getContainerId()).withNetworkId(network.getId()).exec(); + long partitionStarted = System.nanoTime(); + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> + { + while(true) + { + try + { + oldPool.client().set("fencing-window", "not-authoritative"); + } + catch(JedisDataException fenced) + { + assertTrue(fenced.getMessage().contains("NOREPLICAS")); + break; + } + Thread.sleep(25); + } + }); + assertTrue(primary.isRunning()); + assertEquals("PONG", oldPool.client().ping()); + assertThrows(JedisDataException.class, () -> oldPool.client().set("must-not-exist", "stale")); + assertNull(oldPool.client().get("must-not-exist")); + System.out.println("LIVE_PRIMARY_FENCING_MS " + Duration.ofNanos(System.nanoTime() - partitionStarted).toMillis()); + + // Promotion is authorized only after the old primary's rejection above. + assertEquals("OK", command(replica, "REPLICAOF", "NO", "ONE")); + assertEquals("OK", command(replica, "SET", "partition-marker", "after")); + docker.connectToNetworkCmd().withContainerId(replica.getContainerId()).withNetworkId(network.getId()).exec(); + var address = docker.inspectContainerCmd(replica.getContainerId()).exec().getNetworkSettings().getNetworks() + .values().iterator().next().getIpAddress(); + assertEquals("OK", command(primary, "REPLICAOF", address, "6379")); + assertEquals("OK", command(replica, "CONFIG", "SET", "min-replicas-to-write", "1")); + assertEquals("OK", command(replica, "CONFIG", "SET", "min-replicas-max-lag", "1")); + assertTimeoutPreemptively(Duration.ofSeconds(20), () -> + { + while(!"after".equals(oldPool.client().get("partition-marker"))) + { + Thread.sleep(25); + } + while(!"OK".equals(command(replica, "SET", "healed-marker", "replicated"))) + { + Thread.sleep(25); + } + while(!"replicated".equals(oldPool.client().get("healed-marker"))) + { + Thread.sleep(25); + } + }); + var readOnly = assertThrows(JedisDataException.class, () -> oldPool.client().set("must-not-exist", "stale")); + assertTrue(readOnly.getMessage().contains("READONLY")); + assertEquals(0, oldPool.metrics().active()); + assertEquals(2, oldPool.metrics().maximum()); + } + } + } + + /*************************************************************************** + ** Bounded actual QQQ pool connected to the original endpoint throughout. + ***************************************************************************/ + private static RedisConnection connection(GenericContainer endpoint) + { + return (new RedisConnection(new HostAndPort(endpoint.getHost(), endpoint.getMappedPort(6379)), + DefaultJedisClientConfig.builder().connectionTimeoutMillis(500).socketTimeoutMillis(1000).build(), + 2, Duration.ofMillis(500))); + } + + /*************************************************************************** + ** Administration of disposable fixture servers, never application hosts. + ***************************************************************************/ + private static String command(GenericContainer server, String... arguments) throws Exception + { + var invocation = new String[arguments.length + 1]; + invocation[0] = "valkey-cli"; + System.arraycopy(arguments, 0, invocation, 1, arguments.length); + var result = server.execInContainer(invocation); + assertEquals(0, result.getExitCode()); + return (result.getStdout().trim()); + } +} -- 2.54.0 From 551f1d5a00312eb1680295a278f2fabe7a6b67e5 Mon Sep 17 00:00:00 2001 From: James Maes Date: Thu, 29 Oct 2026 11:41:07 -0500 Subject: [PATCH 4/4] ci: align Redis feature qualification with released orb0.8 --- .circleci/config.yml | 4 ++-- .qqq-ci.yml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 124a18c..53a011b 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -1,12 +1,12 @@ version: 2.1 setup: true orbs: - qqq-orb: kingsrook/qqq-orb@0.7.18 + qqq-orb: kingsrook/qqq-orb@0.8 workflows: setup: jobs: - qqq-orb/generate_pipeline: - orb_version: 0.7.18 + orb_version: '0.8' filters: branches: ignore: automation/qbit-rc-pending diff --git a/.qqq-ci.yml b/.qqq-ci.yml index 32d1f74..7e18622 100644 --- a/.qqq-ci.yml +++ b/.qqq-ci.yml @@ -1,5 +1,5 @@ profile: qbit -orb_version: 0.7.18 +orb_version: '0.8' default_branch: develop java_version: '21' browsers: false -- 2.54.0