QRun-IO/qbit-redis
3
0 Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

WIP: test: qualify controlled Redis replica promotion #4

Draft
KofTwentyTwo wants to merge 4 commits from test/redis-topology-1301 into test/redis-mtls-1317
pull from: test/redis-topology-1301
merge into: QRun-IO:test/redis-mtls-1317
Conversation 0 Commits 4 Files Changed 11
+919 -11
11 changed files with 919 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.circleci/config.yml
+2 -2
View File
@@ -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
.qqq-ci.yml
+1 -1
View File
@@ -1,5 +1,5 @@
profile: qbit
orb_version: 0.7.18
orb_version: '0.8'
default_branch: develop
java_version: '21'
browsers: false
README.md
+18
View File
@@ -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
docs/TOPOLOGY-QUALIFICATION.md
+151
View File
@@ -0,0 +1,151 @@
# 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.
## 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
### 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
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.
- 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
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.
## 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.
src/main/java/com/kingsrook/qbits/redis/RedisConnectionConfig.java
+22
View File
@@ -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);
}
}
src/main/java/com/kingsrook/qbits/redis/RedisHealthCheck.java
+25 -7
View File
@@ -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"));
}
}
}
src/test/java/com/kingsrook/qbits/redis/RedisEndpointSwitchTest.java
+270
View File
@@ -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<RealtimeMessage>();
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<Socket> 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());
}
}
}
src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java
+157
View File
@@ -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());
}
}
src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java
+165
View File
@@ -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<RealtimeMessage>();
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);
}
}
src/test/java/com/kingsrook/qbits/redis/RedisRealtimeBusTest.java
+2 -1
View File
@@ -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), () ->
src/test/java/com/kingsrook/qbits/redis/RedisWriteHealthTest.java
+106
View File
@@ -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<String>();
new RedisConnectionConfig().withWriteHealthKeyPrefix(prefix).validate(errors);
assertFalse(errors.isEmpty());
}
var errors = new java.util.ArrayList<String>();
new RedisConnectionConfig().withWriteHealthKeyPrefix("staging:kof22:health:").validate(errors);
assertTrue(errors.isEmpty());
}
}