diff --git a/.circleci/config.yml b/.circleci/config.yml index f890c28..53a011b 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -11,4 +11,6 @@ workflows: branches: ignore: automation/qbit-rc-pending tags: - only: /^v\d+\.\d+\.\d+$/ + only: + - /^v\d+\.\d+\.\d+$/ + - /publish.*/ diff --git a/.qqq-ci.yml b/.qqq-ci.yml index 36e07fd..7e18622 100644 --- a/.qqq-ci.yml +++ b/.qqq-ci.yml @@ -10,6 +10,7 @@ release_tag_policy: stable test_branch_policy: non-release publish: snapshot: true + feature: true release_candidate: true release: true asciidoc: false diff --git a/README.md b/README.md index f0a85d5..61c1f00 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ # QBit Redis Reusable Redis connections and bounded shared caches for QQQ applications. -**Development candidate; not released or production-certified.** Track -[QQQ #1295](https://github.com/QRun-IO/qqq/issues/1295) and [remaining gates](docs/TODO.md). -See [snapshot preparation](docs/SNAPSHOT-RELEASE.md) for current normal-flow evidence, -publication order and the explicitly deferred optional scenarios. +**Coordinated feature snapshot published; not a stable/GA release.** Track +[QQQ #1295](https://github.com/QRun-IO/qqq/issues/1295) and [completed milestone and follow-ups](docs/TODO.md). +See [published snapshot](docs/SNAPSHOT-RELEASE.md) for current normal-flow evidence, +exact compatible versions and the explicitly deferred optional scenarios. ## Host integration @@ -29,8 +29,21 @@ configure `.withTlsProvider(new QCodeReference(HostRedisTls.class))`; that provi returns an SSLContext at runtime, and hostname verification stays enabled. `.withTls(false)` is an explicit option for local test Redis. Real integration tests verify authenticated TLS, untrusted-certificate rejection, wrong-host rejection, -and no plaintext fallback. Client-certificate authentication, certificate rotation -and failover acceptance remain open; only standalone endpoints are implemented. +and no plaintext fallback. Mutual-TLS tests require a separate trusted client +identity, reject missing or untrusted client certificates, and verify framework +health after a successful authenticated connection. Certificates and keys are +generated only for the disposable fixture. + +The TLS provider is resolved when the Redis runtime starts. Replacing its context +alone does not reload existing pools or subscribers. For certificate replacement, +stop dependent services and restart the Redis runtime through the application +lifecycle, then restart its dependents. Qualification replaces both server and +client certificates on the same Redis endpoint: stale material fails a fresh +handshake, health becomes DOWN, and runtime restart with updated material restores +PONG and health UP while closing the old pool. This is a restart procedure, not +zero-downtime certificate rotation. Live reload, trust-chain/revocation policies, +subscriber recovery across certificate changes and failover acceptance remain open; +only standalone endpoints are implemented. QQQ's application launcher starts the registered runtime service. Startup validates configuration and pings every connection before exposing any of the named pools. @@ -67,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 @@ -137,7 +168,7 @@ Java 21, Maven wrapper and Docker are required for the full current tests: .\mvnw.cmd test ``` -The current candidate uses local QQQ `4.1.0-SNAPSHOT` and Jedis `7.2.0`. Tests use +The current candidate uses the published QQQ `4.1.0-RC.21` and Jedis `7.2.0`. Tests use real disposable Redis containers. They cover sharing, expiry, isolation, outages, registration, failed startup cleanup, credential-error redaction, lifecycle, authentication, bounded pool exhaustion and TLS. TLS tests generate temporary diff --git a/docs/TODO.md b/docs/TODO.md index 9c281e9..4d017e4 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -1,63 +1,26 @@ -# Production readiness +# Main-line snapshot complete -Current checklist, 2026-10-21. Historical evidence is tracked on QQQ #1295. -No production readiness claim or release yet. +The owner narrowed this milestone to the main-line happy path. Published versions, evidence and operating requirements are in [SNAPSHOT-RELEASE.md](SNAPSHOT-RELEASE.md). -## Current user-directed milestone +- [x] Named bounded Redis connections/cache primitives, framework health and admin metrics. +- [x] Shared presence across two JVMs, viewer/editor focus, table projection, departures and access filtering. +- [x] Owned shared session-store integration, application isolation and launcher cleanup. +- [x] Cache invalidation, OAuth cache and JDBC provider isolation repairs. +- [x] Atomic versioned saves and frontend stale-save review/retry behavior. +- [x] Full affected verification, downstream CI and four backend snapshot publishers. +- [x] Remote JAR/POM receipts and fresh-cache standalone consumer with aligned dependency graph. +- [x] Published-artifact browser presence/admin/framework health acceptance using the tested frontend export. +- [x] Release/migration documentation and QQQ Project12 issue reconciliation. -The user explicitly narrowed this milestone to the main-line happy path on2026-10-21. The historical audit list below is retained for traceability, not as an instruction to resume every optional scenario before this snapshot. Current publication steps and evidence are in [SNAPSHOT-RELEASE.md](SNAPSHOT-RELEASE.md). +## Future work retained in QQQ Project12 -- [x] Live browser presence across two independent QQQ JVMs sharing Redis. -- [x] Live Redis dashboard metrics, framework-persisted UP health history and health-record navigation. -- [x] Collaboration normal-flow backend179, UI34 and Chromium16 checks. -- [x] Packaged standalone Maven consumer and optional Redis-absent issued-lease consumer. -- [x] Redis POM license/developer/SCM metadata, correct NOTICE and standard CircleCI setup configuration prepared locally. -- [x] Establish Redis remote repository and CircleCI project (empty; no source uploaded). -- [ ] Capture reviewed source revisions, set develop default branch, and verify CI context access/actual pipeline execution. -- [ ] Publish coordinated core/Redis/collaboration/frontend snapshots and verify remote consumer resolution. +- #1315: host inventory and coordinated deployment. Required before each host adopts the candidate; no production rollout has occurred. +- #1289: complete real-DB attachment rollback/crash/retry qualification. +- #1196: selectable virtual tag/EAV projection. +- #1291/#1302: My Work budgets, measured capacity and sustained failure/load qualification. +- #1300/#1301: data loss/restore and additional Redis failover topologies. +- #1316: additional standalone/ingress/CDN/reverse-proxy SSE configurations. +- #1317: additional admin management/aggregation, client-library boundary and credential rotation/mTLS qualification. +- #1304/#1305/#1307/#1309/#1310/#1312/#1314: tooling, CI, layout and test-dependency maintenance. -Optional loss/restore, failover topology and sustained scale work: QQQ #1300, #1301 and #1302 respectively. No claim that those scenarios are supported by the current candidate. - -## Implemented with targeted verification - -- [x] Named bounded Redis pools, cache primitives, namespaces and runtime lifecycle. -- [x] Framework health checks, pool metrics and subscription readiness. -- [x] Authentication/TLS rejection and bounded pool exhaustion tests. -- [x] Redis presence store/service selection, tenant isolation and explicit unavailable UI. -- [x] Cross-node Pub/Sub hints, bounded dispatch, reconnect/resync and silent-stall detection. -- [x] Framework Redis app and local connection/realtime widgets with permission checks. -- [x] Framework-persisted health history and link to existing check settings/actions. -- [x] Packaged app/widget help and health read/write permission-contract tests. -- [x] Metadata-name collision rejection and atomic rejected realtime namespace updates. -- [x] Source dependency report across 44 canonical qbits, with browser validation. - -## Historical broader audit backlog (not the current milestone definition) - -- [ ] Browser acceptance of Redis app, widgets, help, health navigation and Run check now. -- [ ] Complete connection-management requirements and multi-node presentation semantics. -- [ ] Replace or constrain native-client API exposure and prove resource ownership. -- [x] Candidate session-store Jedis alignment and optional shared Redis provider integration tests. -- [x] Shared session-store runtime owns cleanup/restart of registrations with real Redis lifecycle coverage. -- [x] Actual launcher shutdown/failure/retry acceptance for shared Redis session storage. -- [x] Real shared Redis OAuth HTTP login/logout, bounded outage failure and retry acceptance. -- [x] Same-node delayed authentication cache fill after logout rejects the next HTTP request. -- [ ] Token memoization fill/invalidation race and delayed/lost cross-node logout invalidation acceptance. -- [ ] Qualify #1297 memoization invalidation repair: focused, HTTP and full core unit rerun pass; earlier canary failure remains unexplained with diagnostics added. -- [ ] Qualify #1298 application-scoped OAuth token cache; targeted cluster and 14 HTTP acceptance tests pass. -- [ ] Complete #1299 JDBC ownership: registry/health/launcher/native C3P0 callback isolation tested; deployed-host migration and combined acceptance remain. -- [ ] Deployment migration and complete multi-node HTTP authentication compatibility. -- [ ] Clean-consumer tests with optional Redis/email artifacts absent. -- [ ] Full multi-node collaboration/browser acceptance behind a load balancer. -- [ ] Redis eviction/restart recovery and delayed request/tombstone semantics. -- [ ] Supported topology failover, network partition and lifecycle stress tests. -- [ ] mTLS/credential rotation acceptance for supported deployment configuration. -- [ ] Measured load, latency percentiles and explicit supported capacity. -- [ ] Remaining collaboration issues (including durable attachment cleanup and notification leadership). -- [ ] Complete security review, dependency convergence and compatible full-stack quality gates. -- [ ] QQQ issue resolution, release notes, clean-consumer verification and snapshot/RC when ready. - -Latest full Redis verify: 23 tests, zero failures/errors/skips, 94.50% instruction -coverage and guardrails pass (%TEMP%/redis-script-boundary-verify.log). -Collaboration uses script operations instead of the native client. Public credential -and direct-cache construction types still expose Jedis; complete boundary and -session-store compatibility remain open. Report browser checks: all eight pass. +All issues are in https://github.com/orgs/QRun-IO/projects/12 and carry the relevant QBit labels. No future scenario above is declared tested merely because the normal flow passes. Worktrees and optional research files remain preserved. diff --git a/docs/TOPOLOGY-QUALIFICATION.md b/docs/TOPOLOGY-QUALIFICATION.md new file mode 100644 index 0000000..3f5b062 --- /dev/null +++ b/docs/TOPOLOGY-QUALIFICATION.md @@ -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. diff --git a/pom.xml b/pom.xml index 21a1a44..5787a50 100644 --- a/pom.xml +++ b/pom.xml @@ -37,7 +37,8 @@ 0.1.0-SNAPSHOT 21 - 4.1.0-SNAPSHOT + + 4.1.0-RC.21 7.2.0 @@ -45,7 +46,7 @@ com.kingsrook.qqq qqq-bom-pom - ${qqq.snapshot.version} + ${qqq.version} pom import 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/FixtureEndpoint.java b/src/test/java/com/kingsrook/qbits/redis/FixtureEndpoint.java new file mode 100644 index 0000000..b4349bf --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/FixtureEndpoint.java @@ -0,0 +1,238 @@ +/* + * 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.nio.file.Files; +import java.nio.file.Path; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import com.github.dockerjava.api.command.InspectContainerResponse; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; + + +/******************************************************************************* + ** Where a disposable test container is reachable FROM THIS JVM. + ** + ** The original behavior is Testcontainers' own: the host it reports and the + ** published port. That is kept everywhere except one measured topology: a JVM + ** inside a build container that shares the daemon's default bridge with the + ** fixture (the full Linux verify, which talks to the host daemon through its + ** socket). On the Docker Desktop runner the published port is reached through + ** the host-gateway hop; measured on an idle machine that path showed round trips + ** of about 1 ms with stalls of 0.3 s and about 2 s, against about 0.13 ms and + ** no stalls when dialing the sibling directly on the bridge. These tests use + ** 250 ms to 1 s socket timeouts, so such stalls can expire them; that is a + ** plausible contributor, not shown to be the only cause of the observed + ** "Read timed out" failures. In that one topology the fixture is dialed on its + ** bridge address and its own port. + ** + ** The choice is made from the daemon's own description of the build container's + ** networks (no connection probing). If the build container cannot be identified + ** or is not on the default bridge (a job network, host networking, a remote + ** daemon), the original published endpoint is used unchanged. Timeouts, + ** scenarios, assertions and server configuration are unchanged. + *******************************************************************************/ +final class FixtureEndpoint +{ + private static final String DEFAULT_BRIDGE = "bridge"; + + private static final Pattern MOUNTINFO_CONTAINER_ID = Pattern.compile("/docker/containers/([0-9a-f]{64})/"); + + private static volatile Boolean buildContainerOnDefaultBridge; + + + + /*************************************************************************** + ** Host and port to dial. + ***************************************************************************/ + record Address(String host, int port) + { + } + + + + /*************************************************************************** + ** Utility class. + ***************************************************************************/ + private FixtureEndpoint() + { + } + + + + /*************************************************************************** + ** Host this JVM must dial for the container. + ***************************************************************************/ + static String host(GenericContainer container) + { + return (choose(onBridgeWithBuildContainer(), bridgeIp(container), 6379, container.getHost(), publishedPort(container)).host()); + } + + + + /*************************************************************************** + ** Port matching {@link #host(GenericContainer)}. + ***************************************************************************/ + static int port(GenericContainer container) + { + return (choose(onBridgeWithBuildContainer(), bridgeIp(container), 6379, container.getHost(), publishedPort(container)).port()); + } + + + + /*************************************************************************** + ** A container that lives only on a custom test network is also attached to the + ** default bridge when (and only when) the build container is on that bridge, so + ** that the build container can reach it directly. The test network, and therefore + ** every partition scenario between the fixture servers, is untouched: attach only + ** the container the test client dials, never one whose reachability the test cuts + ** or whose peers must stay isolated from it. No-op in every other topology. + ***************************************************************************/ + static void attachToDefaultBridgeWhenInsideContainer(GenericContainer container) + { + if(!onBridgeWithBuildContainer()) + { + return; + } + var docker = container.getDockerClient(); + var networks = docker.inspectContainerCmd(container.getContainerId()).exec().getNetworkSettings().getNetworks(); + if(networks == null || !networks.containsKey(DEFAULT_BRIDGE)) + { + docker.connectToNetworkCmd().withContainerId(container.getContainerId()).withNetworkId(DEFAULT_BRIDGE).exec(); + } + } + + + + /*************************************************************************** + ** Whether this JVM runs in a container that the daemon reports on the default + ** bridge. Decided once per JVM from the daemon's description of this container. + ***************************************************************************/ + private static boolean onBridgeWithBuildContainer() + { + Boolean known = buildContainerOnDefaultBridge; + if(known == null) + { + known = decideBuildContainerOnDefaultBridge(); + buildContainerOnDefaultBridge = known; + } + return (known); + } + + + + /*************************************************************************** + ** Inspect the build container; anything that cannot be established means the + ** original published endpoint (false). + ***************************************************************************/ + private static boolean decideBuildContainerOnDefaultBridge() + { + if(!Files.exists(Path.of("/.dockerenv"))) + { + return (false); + } + try + { + String id = ownContainerId(Files.readString(Path.of("/proc/self/mountinfo")), System.getenv("HOSTNAME")); + if(id == null) + { + return (false); + } + var networks = DockerClientFactory.instance().client().inspectContainerCmd(id).exec().getNetworkSettings().getNetworks(); + return (networks != null && networks.containsKey(DEFAULT_BRIDGE)); + } + catch(Exception e) + { + return (false); + } + } + + + + /*************************************************************************** + ** The id of the container this JVM runs in: from the mount table when it names + ** the container directory, otherwise the host name (the container id unless + ** overridden); null when neither exists. + ***************************************************************************/ + static String ownContainerId(String mountInfo, String hostName) + { + Matcher matcher = MOUNTINFO_CONTAINER_ID.matcher(mountInfo == null ? "" : mountInfo); + if(matcher.find()) + { + return (matcher.group(1)); + } + return (hostName == null || hostName.isBlank() ? null : hostName.trim()); + } + + + + /*************************************************************************** + ** The container's address on the default bridge, or null when it has none. + ***************************************************************************/ + private static String bridgeIp(GenericContainer container) + { + if(!onBridgeWithBuildContainer()) + { + return (null); + } + InspectContainerResponse inspected = container.getDockerClient().inspectContainerCmd(container.getContainerId()).exec(); + var networks = inspected.getNetworkSettings().getNetworks(); + return (networks == null || networks.get(DEFAULT_BRIDGE) == null ? null : networks.get(DEFAULT_BRIDGE).getIpAddress()); + } + + + + /*************************************************************************** + ** Testcontainers' published port for the Redis port, or null when the build + ** container dials the bridge (so the published port is not consulted). + ***************************************************************************/ + private static Integer publishedPort(GenericContainer container) + { + return (onBridgeWithBuildContainer() ? null : container.getMappedPort(6379)); + } + + + + /*************************************************************************** + ** Pure decision, separately tested. With the build container on the default + ** bridge: the sibling's bridge address and its own port. Otherwise the published + ** endpoint exactly as Testcontainers reports it. + ***************************************************************************/ + static Address choose(boolean buildContainerOnDefaultBridge, String bridgeIp, int internalPort, String publishedHost, Integer publishedPort) + { + if(buildContainerOnDefaultBridge) + { + if(bridgeIp == null || bridgeIp.isBlank()) + { + throw new IllegalStateException("FIXTURE_ENDPOINT_NO_BRIDGE_ADDRESS: the build container cannot reach a sibling without its bridge address"); + } + return (new Address(bridgeIp, internalPort)); + } + if(publishedHost == null || publishedHost.isBlank() || publishedPort == null) + { + throw new IllegalStateException("FIXTURE_ENDPOINT_NOT_PUBLISHED: no published endpoint to dial"); + } + return (new Address(publishedHost, publishedPort)); + } + +} diff --git a/src/test/java/com/kingsrook/qbits/redis/FixtureEndpointTest.java b/src/test/java/com/kingsrook/qbits/redis/FixtureEndpointTest.java new file mode 100644 index 0000000..64174fe --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/FixtureEndpointTest.java @@ -0,0 +1,92 @@ +/* + * 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 org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + + +/******************************************************************************* + ** The topology decision of the sibling-container fixtures. + *******************************************************************************/ +class FixtureEndpointTest +{ + private static final String CONTAINER_ID = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + + + + /*************************************************************************** + ** Build container on the default bridge: dial the sibling bridge IP and its own port, never the published port. + ***************************************************************************/ + @Test + void buildContainerOnTheBridgeUsesTheBridgeAddress() + { + assertEquals(new FixtureEndpoint.Address("172.17.0.9", 6379), FixtureEndpoint.choose(true, "172.17.0.9", 6379, "localhost", 41111)); + assertEquals(new FixtureEndpoint.Address("172.17.0.9", 6379), FixtureEndpoint.choose(true, "172.17.0.9", 6379, null, null)); + } + + + + /*************************************************************************** + ** Every other topology (JVM on the Docker host, remote daemon, job network, host networking, unidentified + ** build container) keeps the original Testcontainers host and published port exactly, whatever the host is. + ***************************************************************************/ + @Test + void everyOtherTopologyKeepsTheOriginalPublishedEndpoint() + { + assertEquals(new FixtureEndpoint.Address("localhost", 41111), FixtureEndpoint.choose(false, "172.17.0.9", 6379, "localhost", 41111)); + assertEquals(new FixtureEndpoint.Address("docker.example.test", 32768), FixtureEndpoint.choose(false, null, 6379, "docker.example.test", 32768)); + assertEquals(new FixtureEndpoint.Address("host.docker.internal", 40000), FixtureEndpoint.choose(false, null, 6379, "host.docker.internal", 40000)); + } + + + + /*************************************************************************** + ** An unknown endpoint is an error, never a guessed address. + ***************************************************************************/ + @Test + void missingEndpointFailsClosed() + { + assertThrows(IllegalStateException.class, () -> FixtureEndpoint.choose(true, null, 6379, "localhost", 41111)); + assertThrows(IllegalStateException.class, () -> FixtureEndpoint.choose(true, " ", 6379, "localhost", 41111)); + assertThrows(IllegalStateException.class, () -> FixtureEndpoint.choose(false, "172.17.0.9", 6379, "localhost", null)); + assertThrows(IllegalStateException.class, () -> FixtureEndpoint.choose(false, "172.17.0.9", 6379, " ", 41111)); + } + + + + /*************************************************************************** + ** The build container id comes from the mount table when present, else the host name. + ***************************************************************************/ + @Test + void ownContainerIdPrefersTheMountTableThenTheHostName() + { + String mountInfo = "700 699 0:50 /docker/containers/" + CONTAINER_ID + "/hostname /etc/hostname rw,relatime - ext4 /dev/sda1 rw"; + assertEquals(CONTAINER_ID, FixtureEndpoint.ownContainerId(mountInfo, "unrelated")); + assertEquals("abc123", FixtureEndpoint.ownContainerId("no container mounts here", " abc123 ")); + assertNull(FixtureEndpoint.ownContainerId("", null)); + assertNull(FixtureEndpoint.ownContainerId(null, " ")); + } + +} diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisAuthenticationTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisAuthenticationTest.java index 68c3c7b..0986406 100644 --- a/src/test/java/com/kingsrook/qbits/redis/RedisAuthenticationTest.java +++ b/src/test/java/com/kingsrook/qbits/redis/RedisAuthenticationTest.java @@ -54,7 +54,7 @@ class RedisAuthenticationTest RotatingCredentials.owner = instance; RotatingCredentials.password = TEST_PASSWORD; new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("realtime", new RedisConnectionConfig() - .withHost(redis.getHost()).withPort(redis.getMappedPort(6379)).withTls(false) + .withHost(FixtureEndpoint.host(redis)).withPort(FixtureEndpoint.port(redis)).withTls(false) .withCredentialsProvider(new QCodeReference(RotatingCredentials.class))) .withRealtime("realtime", "test", "rotation")).produce(instance); var runtime = new RedisRuntimeService(); @@ -136,7 +136,7 @@ class RedisAuthenticationTest RotatingCredentials.owner = instance; RotatingCredentials.password = TEST_PASSWORD; new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("cache", new RedisConnectionConfig() - .withHost(redis.getHost()).withPort(redis.getMappedPort(6379)).withTls(false) + .withHost(FixtureEndpoint.host(redis)).withPort(FixtureEndpoint.port(redis)).withTls(false) .withCredentialsProvider(new QCodeReference(RotatingCredentials.class)))).produce(instance); var runtime = new RedisRuntimeService(); try @@ -217,7 +217,7 @@ class RedisAuthenticationTest .withCommand("redis-server", "--requirepass", TEST_PASSWORD).withExposedPorts(6379)) { redis.start(); - var settings = new RedisConnectionConfig().withHost(redis.getHost()).withPort(redis.getMappedPort(6379)) + var settings = new RedisConnectionConfig().withHost(FixtureEndpoint.host(redis)).withPort(FixtureEndpoint.port(redis)) .withTls(false).withMaximumConnections(1).withMaximumWaitMillis(100).withSocketTimeoutMillis(250) .withCredentialsProvider(new QCodeReference(ValidCredentials.class)); var instance = RedisRuntimeTest.hostInstance(); @@ -240,7 +240,7 @@ class RedisAuthenticationTest assertEquals(0, connection.metrics().active()); long initialClients = clientCount(redis); - var bad = new RedisConnectionConfig().withHost(redis.getHost()).withPort(redis.getMappedPort(6379)) + var bad = new RedisConnectionConfig().withHost(FixtureEndpoint.host(redis)).withPort(FixtureEndpoint.port(redis)) .withTls(false).withCredentialsProvider(new QCodeReference(WrongCredentials.class)); assertRejected(bad); assertEquals(initialClients, clientCount(redis)); 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/RedisPartitionHealingTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java new file mode 100644 index 0000000..9c9b2c9 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisPartitionHealingTest.java @@ -0,0 +1,158 @@ +/* + * 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(); + FixtureEndpoint.attachToDefaultBridgeWhenInsideContainer(primary); + 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(FixtureEndpoint.host(endpoint), FixtureEndpoint.port(endpoint)), + 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()); + } +} 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..e571d02 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/redis/RedisPromotionTest.java @@ -0,0 +1,167 @@ +/* + * 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(); + FixtureEndpoint.attachToDefaultBridgeWhenInsideContainer(primary); + FixtureEndpoint.attachToDefaultBridgeWhenInsideContainer(replica); + var firstInstance = host(replica); + var secondInstance = host(replica); + var firstRuntime = new RedisRuntimeService(); + var secondRuntime = new RedisRuntimeService(); + try(var oldPrimary = new RedisConnection(new HostAndPort(FixtureEndpoint.host(primary), FixtureEndpoint.port(primary)), + 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(FixtureEndpoint.host(endpoint)).withPort(FixtureEndpoint.port(endpoint)).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), () -> diff --git a/src/test/java/com/kingsrook/qbits/redis/RedisTlsTest.java b/src/test/java/com/kingsrook/qbits/redis/RedisTlsTest.java index 6b0a403..3146574 100644 --- a/src/test/java/com/kingsrook/qbits/redis/RedisTlsTest.java +++ b/src/test/java/com/kingsrook/qbits/redis/RedisTlsTest.java @@ -32,6 +32,7 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import redis.clients.jedis.Protocol; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.testcontainers.containers.GenericContainer; @@ -104,6 +105,117 @@ class RedisTlsTest } } + /*************************************************************************** + ** Redis must require a trusted client identity as well as server verification. + ***************************************************************************/ + @Test + void mutualTlsRequiresTrustedClientCertificate() throws Exception + { + try(GenericContainer redis = new GenericContainer<>(DockerImageName.parse("redis:7.4-alpine")).withExposedPorts(6379)) + { + Path server = Files.createDirectory(temporary.resolve("server")); + Path client = Files.createDirectory(temporary.resolve("client")); + String host = redis.getHost(); + String identity = (host.matches("[0-9.]+") || host.contains(":") ? "ip:" : "dns:") + host; + SSLContext trustOnly = certificate(server, identity); + certificate(client, "dns:client-fixture.invalid"); + redis.withCopyFileToContainer(MountableFile.forHostPath(server.resolve("cert.pem"), 0444), "/server-cert.pem") + .withCopyFileToContainer(MountableFile.forHostPath(server.resolve("key.pem"), 0444), "/server-key.pem") + .withCopyFileToContainer(MountableFile.forHostPath(client.resolve("cert.pem"), 0444), "/client-trust.pem") + .withCommand("redis-server", "--port", "0", "--tls-port", "6379", + "--tls-cert-file", "/server-cert.pem", "--tls-key-file", "/server-key.pem", + "--tls-ca-cert-file", "/client-trust.pem", "--tls-auth-clients", "yes", + "--requirepass", RedisAuthenticationTest.TEST_PASSWORD); + redis.start(); + var settings = new RedisConnectionConfig().withHost(host).withPort(redis.getMappedPort(6379)) + .withConnectionTimeoutMillis(1000).withSocketTimeoutMillis(1000) + .withTlsProvider(new QCodeReference(TrustedContext.class)) + .withCredentialsProvider(new QCodeReference(RedisAuthenticationTest.ValidCredentials.class)); + var instance = RedisRuntimeTest.hostInstance(); + new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("presence", settings)).produce(instance); + var runtime = new RedisRuntimeService(); + try + { + TrustedContext.context = trustOnly; + assertThrows(QException.class, () -> runtime.start(instance), "Server trust alone must not satisfy mutual TLS"); + assertThrows(IllegalStateException.class, () -> RedisRuntimeService.connection(instance, "presence")); + TrustedContext.context = clientContext(server, server); + assertThrows(QException.class, () -> runtime.start(instance), "An unrelated client identity must not authenticate"); + assertThrows(IllegalStateException.class, () -> RedisRuntimeService.connection(instance, "presence")); + TrustedContext.context = clientContext(server, client); + runtime.start(instance); + assertEquals("PONG", RedisRuntimeService.connection(instance, "presence").client().ping()); + assertEquals(com.kingsrook.qqq.backend.core.health.HealthStatus.UP, new RedisHealthCheck().run( + new com.kingsrook.qqq.backend.core.health.HealthCheckInput().withQInstance(instance)).getStatus()); + + // Rotate both identities on the same endpoint; existing TLS sessions do not re-handshake. + Path rotatedServer = Files.createDirectory(temporary.resolve("rotated-server")); + Path rotatedClient = Files.createDirectory(temporary.resolve("rotated-client")); + certificate(rotatedServer, identity); + certificate(rotatedClient, "dns:rotated-client-fixture.invalid"); + redis.copyFileToContainer(MountableFile.forHostPath(rotatedServer.resolve("cert.pem"), 0444), "/rotated-server-cert.pem"); + redis.copyFileToContainer(MountableFile.forHostPath(rotatedServer.resolve("key.pem"), 0444), "/rotated-server-key.pem"); + redis.copyFileToContainer(MountableFile.forHostPath(rotatedClient.resolve("cert.pem"), 0444), "/rotated-client-trust.pem"); + var oldClient = RedisRuntimeService.connection(instance, "presence").client(); + var configured = oldClient.sendCommand(Protocol.Command.CONFIG, "SET", + "tls-cert-file", "/rotated-server-cert.pem", "tls-key-file", "/rotated-server-key.pem", + "tls-ca-cert-file", "/rotated-client-trust.pem"); + assertEquals("OK", new String((byte[]) configured, java.nio.charset.StandardCharsets.UTF_8)); + oldClient.getPool().clear(); + assertThrows(redis.clients.jedis.exceptions.JedisException.class, oldClient::ping, + "A fresh connection with stale trust and client keys must fail"); + assertEquals(0, oldClient.getPool().getNumActive()); + var healthInput = new com.kingsrook.qqq.backend.core.health.HealthCheckInput().withQInstance(instance); + assertEquals(com.kingsrook.qqq.backend.core.health.HealthStatus.DOWN, new RedisHealthCheck().run(healthInput).getStatus()); + + TrustedContext.context = clientContext(rotatedServer, rotatedClient); + assertThrows(redis.clients.jedis.exceptions.JedisException.class, oldClient::ping, + "Changing the provider alone must not be mistaken for live TLS reload"); + runtime.stop(); + runtime.afterApplicationStop(); + assertTrue(oldClient.getPool().isClosed()); + assertThrows(IllegalStateException.class, () -> RedisRuntimeService.connection(instance, "presence")); + runtime.start(instance); + var newClient = RedisRuntimeService.connection(instance, "presence").client(); + assertNotSame(oldClient, newClient); + assertEquals("PONG", newClient.ping()); + assertEquals(com.kingsrook.qqq.backend.core.health.HealthStatus.UP, new RedisHealthCheck().run(healthInput).getStatus()); + assertEquals(0, newClient.getPool().getNumActive()); + } + finally + { + runtime.stop(); + runtime.afterApplicationStop(); + TrustedContext.context = null; + } + } + } + + /*************************************************************************** + ** Keep server trust independent of the client key presented to Redis. + ***************************************************************************/ + private static SSLContext clientContext(Path server, Path client) throws Exception + { + KeyStore keys = KeyStore.getInstance("PKCS12"); + try(var input = Files.newInputStream(client.resolve("test.p12"))) + { + keys.load(input, "test-only-password".toCharArray()); + } + var keyManagers = javax.net.ssl.KeyManagerFactory.getInstance(javax.net.ssl.KeyManagerFactory.getDefaultAlgorithm()); + keyManagers.init(keys, "test-only-password".toCharArray()); + KeyStore trust = KeyStore.getInstance(KeyStore.getDefaultType()); + trust.load(null, null); + try(var input = Files.newInputStream(server.resolve("cert.pem"))) + { + trust.setCertificateEntry("server", java.security.cert.CertificateFactory.getInstance("X.509").generateCertificate(input)); + } + var trustManagers = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trustManagers.init(trust); + SSLContext context = SSLContext.getInstance("TLS"); + context.init(keyManagers.getKeyManagers(), trustManagers.getTrustManagers(), null); + return (context); + } + /*************************************************************************** ** JDK keytool avoids internal JDK APIs and external certificate dependencies. ***************************************************************************/ 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()); + } +}