From ef2c0c11a0014a6b70a087b2ce7ffe65f987cc6b Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 06:42:52 -0500 Subject: [PATCH 1/6] feat: detect lost Redis presence authority and rejoin Report unavailable presence during recovery, invalidate stale leases, and isolate controlled restores by host generation. Add store, HTTP and independent-JVM recovery acceptance. Browser, maintenance enforcement and performance qualification remain open in QQQ #1300. Co-Authored-By: GPT-6 (Codex) --- docs/PRESENCE-RECOVERY.md | 100 ++++++++ .../acceptance/presence_recovery_processes.py | 138 +++++++++++ .../presence/PresenceRecoveringException.java | 36 +++ .../presence/RedisPresenceAudience.java | 2 +- .../presence/RedisPresenceBackend.java | 4 + .../presence/RedisPresenceConfig.java | 24 ++ .../presence/RedisPresenceStore.java | 21 +- src/main/resources/presence/redis-mutate.lua | 21 +- .../resources/presence/redis-recovery.lua | 37 +++ .../resources/presence/redis-snapshot.lua | 16 +- .../CollaborationApiHttpTest.java | 63 +++++ .../RedisPresenceRecoveryTest.java | 232 ++++++++++++++++++ .../RedisPresenceServiceTest.java | 63 ++++- .../collaboration/RedisPresenceStoreTest.java | 111 +++++++-- 14 files changed, 822 insertions(+), 46 deletions(-) create mode 100644 docs/PRESENCE-RECOVERY.md create mode 100644 docs/acceptance/presence_recovery_processes.py create mode 100644 src/main/java/com/kingsrook/qbits/collaboration/presence/PresenceRecoveringException.java create mode 100644 src/main/resources/presence/redis-recovery.lua create mode 100644 src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md new file mode 100644 index 0000000..1254e62 --- /dev/null +++ b/docs/PRESENCE-RECOVERY.md @@ -0,0 +1,100 @@ +# Presence recovery candidate — QQQ #1300 + +This branch is an uncommitted implementation candidate. It is not part of the +published snapshots and has not passed the existing store/service suites. + +## Contract under qualification + +Each deployment/application/tenant/table has a persistent recovery guard in the +same Redis hash slot as its bounded room projections. Each Lua operation reads +that guard atomically with its room work. The guard contains a generation, a +readiness deadline and the latest issued expiry. A missing guard, or complete +loss of table projections before that expiry, starts a fresh generation and a +recovery interval of one lease lifetime. Existing live browsers can reserve and +renew replacement leases during that interval. Snapshots report recovery rather +than a successful empty room until the interval finishes. + +Lease entries carry the guard generation. Recovery removes invalid-generation +projections from touched rooms so they cannot consume all capacity and prevent +re-registration. Existing authorization remains in PresenceService; the guard +does not grant record, tenant or field access. The shared adapter maps recovery +to a sanitized service-unavailable response. Record-version conflict checks are +independent of this advisory state. + +An untouched new namespace is conservatively unknown too. Its first snapshot +therefore starts a recovery interval. Normal idle expiry retains the guard so +later empty snapshots can be known-empty. The guard has no TTL; tenant-retirement +operations and long-term guard storage must be included in operational capacity +qualification. + +## Controlled restore + +The host configuration has a bounded `recoveryGeneration`, default `initial`, +included in both the storage namespace and realtime audience. It belongs in +deployment configuration outside the Redis backup. All nodes must use the same +value. A restored guard can otherwise restore old authority; Redis process IDs +cannot detect arbitrary same-process data restoration. + +The required maintenance procedure is: stop admission and drain all old web +nodes; restore Redis; choose a never-reused generation outside Redis; start only +nodes with that generation; verify recovery and authenticated re-registration; +then reopen traffic. Do not roll back the generation when rolling back an app. +The host tooling that enforces and verifies this sequence is **not implemented** +yet. The branch does not claim transparent arbitrary restore or split-brain +recovery. Native Sentinel/Cluster routing remains the separate Redis issue #1301. + +The storage prefix is v2 so old executables cannot treat generation-bound entries +as old-format authority. This requires a coordinated deployment, not a mixed +version rolling deployment. Existing published v1 snapshots are unchanged. + +## Evidence and remaining work + +- Before the patch, a two-client disposable Redis test failed because total loss + returned an empty snapshot. +- Five focused tests now pass: full loss, loss of volatile projections while the + guard survives, restored guard/lease bytes followed by a host generation + change, bounded generation validation, and rejection of inconsistent lease + lifetimes across nodes. These use independent clients in + one JVM, not a deployed HA cluster. +- Full wrapper verification passed on October 26: 183 tests, no failures, + errors or skips; packaging and coverage gates passed. This includes the + original seven store tests and two-node six-viewer/two-editor scenario. + The admin dashboard reads every table, so the fixture initializes every + dashboard table for both tenants through the real backend and waits for + recovery without writing Redis guard state. +- The service fixture socket read timeout is one second; its previous 200 ms + limit caused a Redis EVALSHA read timeout. Production timeouts are unchanged. +- The full run includes a real HTTP recovery test: real Redis + FLUSHDB produces 503 during recovery, the former lease returns 409 with + PRESENCE_LEASE_EXPIRED, hidden records and unauthorized viewers remain + inaccessible, and a new lease rejoins through the production recovery window. + This uses a loopback HTTP server with fixture authentication, not a browser + or a deployed host. The focused test and the combined full suite both pass. +- Still required: remaining cold/readiness and partial-loss edge cases, + browser re-registration and unavailable + rendering, enforced host maintenance/reset, and measured + per-operation cost before adoption. The implementation uses no INFO call or + lifecycle polling cache; its additional guard reads/writes still need measurement. +- This is an unreleased draft candidate. Publication and production deployment + have not occurred; the remaining acceptance work above is still required. + +## Independent-process acceptance + +`docs/acceptance/presence_recovery_processes.py` passed on October 26 using +two independent `SharedPresenceDevServer` JVMs and one disposable Redis container. +Each user renews through the opposite node. After FLUSHDB, both nodes return 503 +for presence and 409 for the former leases; fresh leases restore both viewers on +both nodes. The harness terminates only its owned processes and container. +This is fixture authentication with independently seeded memory records, not a +deployed HA application or a browser acceptance result. + +Reproduce from the repository root after `./mvnw test-compile`: + +```text +./mvnw dependency:build-classpath -Dmdep.includeScope=test -Dmdep.outputFile=target/recovery-process-classpath.txt +python docs/acceptance/presence_recovery_processes.py --java /path/to/java --repo /path/to/repository +``` + +Requires Python 3, Java 21 and Docker. JVM output and argument files stay under +`target/recovery-processes-*` for diagnosis. The harness uses the production +45-second recovery window, with bounded startup and recovery deadlines. diff --git a/docs/acceptance/presence_recovery_processes.py b/docs/acceptance/presence_recovery_processes.py new file mode 100644 index 0000000..f5f26ae --- /dev/null +++ b/docs/acceptance/presence_recovery_processes.py @@ -0,0 +1,138 @@ +"""Exercise Redis loss through two independent SharedPresenceDevServer JVMs. + +Run after test-compile and dependency:build-classpath with test scope and output +target/recovery-process-classpath.txt. Uses only disposable loopback fixtures. +""" + +import argparse +import json +import os +from pathlib import Path +import socket +import subprocess +import time +import urllib.error +import urllib.request +import uuid + + +def free_port(): + """Reserve a candidate loopback port; startup detects any bind race.""" + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + return listener.getsockname()[1] + + +def request(port, user, method, path, body=None): + """Use the fixture's mock session through the actual HTTP boundary.""" + separator = "&" if "?" in path else "?" + url = f"http://127.0.0.1:{port}/qqq/v1/qbit/collaboration{path}{separator}user={user}" + data = None if body is None else json.dumps(body).encode() + req = urllib.request.Request(url, data=data, method=method, headers={ + "Cookie": f"sessionUUID={user}-recovery-processes", + "Content-Type": "application/json", + }) + try: + response = urllib.request.urlopen(req, timeout=10) + except urllib.error.HTTPError as error: + response = error + with response: + return response.status, json.load(response) + + +def require(status, result): + """Fail with fixture-only response evidence when transport differs.""" + assert result[0] == status, result + return result[1] + + +def settle(ports, leases): + """Renew on the opposite node until both nodes report both people.""" + deadline = time.monotonic() + 60 + sequence = 0 + while time.monotonic() < deadline: + sequence += 1 + for index, (user, lease) in enumerate(leases.items()): + result = request(ports[1 - index], user, "POST", "/presence/leases/renew", { + "connectionId": lease, "table": "person", "pk": "1", + "mode": "VIEWING", "sequence": sequence, + }) + assert result[0] in (200, 503), result + if result[0] == 503: + assert "recovering" in result[1]["message"], result + snapshots = [request(port, "alice", "GET", "/presence?table=person&pk=1") for port in ports] + if all(status == 200 for status, _ in snapshots): + assert all(len(body["viewers"]) == 2 for _, body in snapshots), snapshots + return + assert all(status == 503 for status, _ in snapshots), snapshots + time.sleep(1) + raise AssertionError("Two-process recovery exceeded 60 seconds") + + +def main(): + """Own and clean up the exact child JVMs and disposable Redis container.""" + parser = argparse.ArgumentParser() + parser.add_argument("--java", required=True) + parser.add_argument("--repo", type=Path, required=True) + args = parser.parse_args() + repo = args.repo.resolve() + output = repo / "target" / ("recovery-processes-" + uuid.uuid4().hex[:10]) + output.mkdir() + classpath = os.pathsep.join([str(repo / "target/test-classes"), str(repo / "target/classes"), + (repo / "target/recovery-process-classpath.txt").read_text().strip()]) + container = "qqq-recovery-" + uuid.uuid4().hex[:12] + processes, logs = [], [] + try: + subprocess.run(["docker", "run", "-d", "--name", container, "-p", "127.0.0.1::6379", "redis:7.4-alpine"], check=True, capture_output=True) + binding = subprocess.check_output(["docker", "port", container, "6379"], text=True).strip() + redis_port = int(binding.rsplit(":", 1)[1]) + ports = [free_port(), free_port()] + for port in ports: + argfile = output / f"node-{port}.args" + argfile.write_text('-cp\n"' + classpath.replace("\\", "/") + '"\n' + 'com.kingsrook.qbits.collaboration.devserver.SharedPresenceDevServer\n' + f'{port}\n{redis_port}\n', encoding="utf-8") + log = (output / f"node-{port}.log").open("w") + logs.append(log) + processes.append(subprocess.Popen([args.java, "@" + str(argfile)], cwd=repo, + stdout=log, stderr=subprocess.STDOUT, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))) + deadline = time.monotonic() + 60 + for port, process in zip(ports, processes): + while f"SHARED_PRESENCE_READY {port}" not in (output / f"node-{port}.log").read_text(errors="replace"): + assert process.poll() is None, f"Node exited; see {output}" + assert time.monotonic() < deadline, f"Startup timed out; see {output}" + time.sleep(0.2) + def join(): + """Acquire each person's new lease on a different node.""" + return {user: require(200, request(port, user, "POST", "/presence/leases", { + "table": "person", "pk": "1"}))["connectionId"] for port, user in zip(ports, ("alice", "bob"))} + original = join() + settle(ports, original) + subprocess.run(["docker", "exec", container, "redis-cli", "FLUSHDB"], check=True, capture_output=True) + for port in ports: + require(503, request(port, "alice", "GET", "/presence?table=person&pk=1")) + for user, lease in original.items(): + error = require(409, request(port, user, "POST", "/presence/leases/renew", { + "connectionId": lease, "table": "person", "pk": "1", "mode": "VIEWING", "sequence": 1000})) + assert error["code"] == "PRESENCE_LEASE_EXPIRED", error + replacement = join() + assert all(replacement[user] != original[user] for user in original) + settle(ports, replacement) + print(f"PASS: two independent JVMs recovered two viewers after Redis loss. Logs: {output}") + finally: + for process in processes: + process.terminate() + for process in processes: + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + for log in logs: + log.close() + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + + +if __name__ == "__main__": + main() diff --git a/src/main/java/com/kingsrook/qbits/collaboration/presence/PresenceRecoveringException.java b/src/main/java/com/kingsrook/qbits/collaboration/presence/PresenceRecoveringException.java new file mode 100644 index 0000000..11d4a58 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/collaboration/presence/PresenceRecoveringException.java @@ -0,0 +1,36 @@ +/* + * 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.collaboration.presence; + +/******************************************************************************* + ** Shared presence is rebuilding after unknown or lost Redis state. + *******************************************************************************/ +public final class PresenceRecoveringException extends IllegalStateException +{ + /*************************************************************************** + ** Never include caller identifiers or stored data in this message. + ***************************************************************************/ + public PresenceRecoveringException() + { + super("Presence is recovering; active viewers are rejoining"); + } +} diff --git a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceAudience.java b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceAudience.java index e767049..824c4b4 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceAudience.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceAudience.java @@ -49,6 +49,6 @@ public class RedisPresenceAudience implements RealtimeAudienceProvider public static RedisNamespace namespace(RedisPresenceConfig config) throws Exception { var scope = QCodeLoader.getAdHoc(PresenceScopeProvider.class, config.getScopeProvider()).tenantScope(); - return (new RedisNamespace(config.getDeployment(), config.getApplication(), scope, "presence")); + return (new RedisNamespace(config.getDeployment(), config.getApplication(), scope, "presence:" + config.getRecoveryGeneration())); } } diff --git a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceBackend.java b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceBackend.java index 199428f..2bdb5cf 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceBackend.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceBackend.java @@ -145,6 +145,10 @@ public class RedisPresenceBackend implements PresenceBackend { return (store.snapshot(table, pk)); } + catch(PresenceRecoveringException recovering) + { + throw (new QServiceUnavailableException("Presence is recovering; active viewers are rejoining. Please retry.")); + } catch(RuntimeException failure) { throw (unavailable()); diff --git a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceConfig.java b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceConfig.java index 8724a8d..b761c6a 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceConfig.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceConfig.java @@ -35,6 +35,7 @@ public class RedisPresenceConfig implements Serializable private String application = null; private QCodeReference scopeProvider = null; private Integer capacity = 2000; + private String recoveryGeneration = "initial"; /*************************************************************************** ** Inspect configuration without creating a runtime pool or local registry. @@ -61,6 +62,10 @@ public class RedisPresenceConfig implements Serializable { errors.add("Redis presence capacity must be between 1 and 10000"); } + if(recoveryGeneration == null || !recoveryGeneration.matches("[A-Za-z0-9_-]{1,64}")) + { + errors.add("Redis presence recovery generation must contain 1 to 64 letters, digits, underscores or hyphens"); + } } /*************************************************************************** @@ -147,4 +152,23 @@ public class RedisPresenceConfig implements Serializable capacity = value; return (this); } + + + /*************************************************************************** + ** Host-owned generation, stored outside Redis and changed after restoring data. + ***************************************************************************/ + public String getRecoveryGeneration() + { + return (recoveryGeneration); + } + + + /*************************************************************************** + ** All nodes must use the same new generation before maintenance traffic resumes. + ***************************************************************************/ + public RedisPresenceConfig withRecoveryGeneration(String value) + { + recoveryGeneration = value; + return (this); + } } diff --git a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceStore.java b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceStore.java index 9a8c208..69db925 100644 --- a/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceStore.java +++ b/src/main/java/com/kingsrook/qbits/collaboration/presence/RedisPresenceStore.java @@ -44,8 +44,8 @@ import com.kingsrook.qqq.backend.core.utils.JsonUtils; *******************************************************************************/ public final class RedisPresenceStore { - private static final RedisScript MUTATE = new RedisScript(resource("redis-mutate.lua")); - private static final RedisScript SNAPSHOT = new RedisScript(resource("redis-snapshot.lua")); + private static final RedisScript MUTATE = new RedisScript(resource("redis-recovery.lua") + "\n" + resource("redis-mutate.lua")); + private static final RedisScript SNAPSHOT = new RedisScript(resource("redis-recovery.lua") + "\n" + resource("redis-snapshot.lua")); private final RedisScriptExecutor client; private final RedisNamespace namespace; private final long ttlMillis; @@ -123,7 +123,12 @@ public final class RedisPresenceStore public List snapshot(String table, String pk) { List keys = roomKeys(table, pk, true); - Object result = client.execute(SNAPSHOT, keys, List.of(pk == null ? "" : ":" + encode(pk, 1024))); + Object result = client.execute(SNAPSHOT, keys, List.of(pk == null ? "" : ":" + encode(pk, 1024), + Long.toString(ttlMillis), java.util.UUID.randomUUID().toString())); + if(result instanceof Number number && number.longValue() == -2) + { + throw (new PresenceRecoveringException()); + } Map users = new TreeMap<>(); for(Object value : (List) result) { @@ -160,9 +165,10 @@ public final class RedisPresenceStore { throw (new IllegalArgumentException("Presence sequence must be a nonnegative safe integer")); } - Stored value = new Stored(user, displayName, mode, pk == null ? null : field, 0, sequence, departed, false, false); + Stored value = new Stored(user, displayName, mode, pk == null ? null : field, 0, sequence, departed, false, false, null); Object result = client.execute(MUTATE, roomKeys(table, pk, true), List.of(member, JsonUtils.toJson(value), - sequence == null ? "" : sequence.toString(), departed ? "1" : "0", Long.toString(ttlMillis), Integer.toString(capacity), pk == null ? "1" : "0", operation)); + sequence == null ? "" : sequence.toString(), departed ? "1" : "0", Long.toString(ttlMillis), Integer.toString(capacity), pk == null ? "1" : "0", operation, + java.util.UUID.randomUUID().toString())); if(((Number) result).longValue() == -1) { throw (new PresenceLeaseExpiredException()); @@ -175,7 +181,7 @@ public final class RedisPresenceStore ***************************************************************************/ private List roomKeys(String table, String pk, boolean includeTable) { - String base = "qqq:presence:{" + digest("SHA-256", namespace.key(table)) + "}"; + String base = "qqq:presence:v2:{" + digest("SHA-256", namespace.key(table)) + "}"; List keys = new ArrayList<>(); if(pk != null) { @@ -188,6 +194,7 @@ public final class RedisPresenceStore keys.add(base + ":table:expiry"); keys.add(base + ":table:state"); } + keys.add(base + ":recovery"); return (keys); } @@ -240,7 +247,7 @@ public final class RedisPresenceStore /*************************************************************************** ** Internal lease data, never returned directly to a browser. ***************************************************************************/ - public record Stored(String userId, String displayName, String mode, String fieldName, long at, Long sequence, boolean departed, boolean pending, boolean leased) + public record Stored(String userId, String displayName, String mode, String fieldName, long at, Long sequence, boolean departed, boolean pending, boolean leased, String generation) { } diff --git a/src/main/resources/presence/redis-mutate.lua b/src/main/resources/presence/redis-mutate.lua index 9102888..797927d 100644 --- a/src/main/resources/presence/redis-mutate.lua +++ b/src/main/resources/presence/redis-mutate.lua @@ -1,19 +1,21 @@ -- All room keys share a table hash tag. Redis TIME owns lease expiry. +local clock = redis.call('TIME') +local now = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000) +local guard = presenceRecovery(KEYS, tonumber(ARGV[5]), ARGV[9], now) +local rooms = #KEYS - 1 -- Validate every pair before changing either record or table membership. -for i = 1, #KEYS, 2 do +for i = 1, rooms, 2 do if redis.call('ZCARD', KEYS[i]) ~= redis.call('HLEN', KEYS[i + 1]) then return redis.error_reply('Presence room state is incomplete') end end -local clock = redis.call('TIME') -local now = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000) local member = ARGV[1] local ttl = tonumber(ARGV[5]) local capacity = tonumber(ARGV[6]) local sequence = tonumber(ARGV[3]) local departed = ARGV[4] == '1' -for i = 1, #KEYS, 2 do +for i = 1, rooms, 2 do local expired = redis.call('ZRANGEBYSCORE', KEYS[i], '-inf', now) for _, id in ipairs(expired) do redis.call('ZREM', KEYS[i], id) @@ -24,7 +26,7 @@ end local previous = redis.call('HGET', KEYS[2], member) -- A record lease and its table projection are written atomically. If one is -- missing while the other is live, do not erase surviving ordering/tombstones. -if #KEYS == 4 then +if rooms == 4 then local tablePrevious = redis.call('HGET', KEYS[4], member) if (previous and not tablePrevious) or (tablePrevious and not previous) then return redis.error_reply('Presence room state is incomplete') @@ -34,7 +36,7 @@ local old = previous and cjson.decode(previous) or nil local operation = ARGV[8] or 'legacy' -- Renewal authority is issued by open, not inferred from a caller's sequence. -- Missing/expired/departed leases cannot be recreated by delayed HTTP requests. -if operation == 'renew' and (not old or not old.leased or old.departed) then +if operation == 'renew' and (not old or not old.leased or old.departed or old.generation ~= guard.generation) then return -1 end if operation == 'open' and old then return -1 end @@ -52,7 +54,7 @@ if old then end -- Check every room before writing any membership, avoiding a partial join. -for i = 1, #KEYS, 2 do +for i = 1, rooms, 2 do if redis.call('HEXISTS', KEYS[i + 1], member) == 0 and redis.call('ZCARD', KEYS[i]) >= capacity then return 0 end @@ -64,7 +66,8 @@ data.sequence = sequence or cjson.null data.departed = departed data.pending = operation == 'open' data.leased = operation ~= 'legacy' or (old and old.leased) or false -for i = 1, #KEYS, 2 do +data.generation = guard.generation +for i = 1, rooms, 2 do -- Record scope comes first; table scope must never carry focused-field data. if i > 1 or ARGV[7] == '1' then data.fieldName = cjson.null end redis.call('HSET', KEYS[i + 1], member, cjson.encode(data)) @@ -72,4 +75,6 @@ for i = 1, #KEYS, 2 do redis.call('PEXPIRE', KEYS[i], ttl) redis.call('PEXPIRE', KEYS[i + 1], ttl) end +guard.liveUntil = now + ttl +redis.call('SET', KEYS[#KEYS], cjson.encode(guard)) return 1 diff --git a/src/main/resources/presence/redis-recovery.lua b/src/main/resources/presence/redis-recovery.lua new file mode 100644 index 0000000..9592a89 --- /dev/null +++ b/src/main/resources/presence/redis-recovery.lua @@ -0,0 +1,37 @@ +-- This prelude is executed atomically with the room operation, in the same slot. +-- The guard deliberately outlives idle leases: missing means unknown, not empty. +local function presenceRecovery(keys, ttl, candidate, now) + local encoded = redis.call('GET', keys[#keys]) + local guard + if encoded then + guard = cjson.decode(encoded) + if type(guard.generation) ~= 'string' or type(guard.readyAt) ~= 'number' or type(guard.liveUntil) ~= 'number' or type(guard.ttl) ~= 'number' then + error('Invalid presence recovery guard') + end + if guard.ttl ~= ttl then error('Presence lease lifetime differs across nodes') end + -- All volatile projections disappearing before the last issued expiry is + -- data loss too, even when a persistent guard survives an eviction policy. + if guard.liveUntil > now and redis.call('ZCARD', keys[#keys - 2]) == 0 + and redis.call('HLEN', keys[#keys - 1]) == 0 then + guard = nil + end + end + if not guard then + guard = { generation = candidate, readyAt = now + ttl, liveUntil = 0, ttl = ttl } + redis.call('SET', keys[#keys], cjson.encode(guard)) + end + -- During recovery, discard only projections from an invalidated generation. + -- This also frees capacity for new reservations without waiting for old TTLs. + if now < guard.readyAt then + for i = 1, #keys - 1, 2 do + for _, member in ipairs(redis.call('HKEYS', keys[i + 1])) do + local data = cjson.decode(redis.call('HGET', keys[i + 1], member)) + if data.generation ~= guard.generation then + redis.call('HDEL', keys[i + 1], member) + redis.call('ZREM', keys[i], member) + end + end + end + end + return guard +end diff --git a/src/main/resources/presence/redis-snapshot.lua b/src/main/resources/presence/redis-snapshot.lua index 46e32d0..686a8cf 100644 --- a/src/main/resources/presence/redis-snapshot.lua +++ b/src/main/resources/presence/redis-snapshot.lua @@ -1,12 +1,14 @@ -- A partially evicted room is unavailable, not empty. Do not silently discard it. -for i = 1, #KEYS, 2 do +local clock = redis.call('TIME') +local now = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000) +local guard = presenceRecovery(KEYS, tonumber(ARGV[2]), ARGV[3], now) +local rooms = #KEYS - 1 +for i = 1, rooms, 2 do if redis.call('ZCARD', KEYS[i]) ~= redis.call('HLEN', KEYS[i + 1]) then return redis.error_reply('Presence room state is incomplete') end end -local clock = redis.call('TIME') -local now = tonumber(clock[1]) * 1000 + math.floor(tonumber(clock[2]) / 1000) -for i = 1, #KEYS, 2 do +for i = 1, rooms, 2 do local expired = redis.call('ZRANGEBYSCORE', KEYS[i], '-inf', now) for _, id in ipairs(expired) do redis.call('ZREM', KEYS[i], id) @@ -16,7 +18,7 @@ end local members = redis.call('ZRANGE', KEYS[1], 0, -1) -- Table membership is bounded by the configured capacity. Compare the record's -- projection without key discovery or cross-slot reads. -if #KEYS == 4 then +if rooms == 4 then local projected = 0 for _, id in ipairs(redis.call('ZRANGE', KEYS[3], 0, -1)) do if string.sub(id, -string.len(ARGV[1])) == ARGV[1] then @@ -31,9 +33,13 @@ if #KEYS == 4 then end end local result = {} +if now < guard.readyAt then return -2 end for _, id in ipairs(members) do local data = redis.call('HGET', KEYS[2], id) if not data then return redis.error_reply('Presence room state is incomplete') end + if cjson.decode(data).generation ~= guard.generation then + return redis.error_reply('Presence generation is inconsistent') + end table.insert(result, data) end return result diff --git a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationApiHttpTest.java b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationApiHttpTest.java index 6d86d75..9884ff3 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/CollaborationApiHttpTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/CollaborationApiHttpTest.java @@ -63,6 +63,69 @@ import static org.assertj.core.api.Assertions.assertThat; *******************************************************************************/ class CollaborationApiHttpTest extends PersonalServiceFixture { + /*************************************************************************** + ** Data loss is unavailable over HTTP until a fresh lease has rejoined. + ***************************************************************************/ + @Test + void redisLossRequiresHttpRejoin() throws Exception + { + try(var redis = new org.testcontainers.containers.GenericContainer<>(org.testcontainers.utility.DockerImageName.parse("redis:7.4-alpine")).withExposedPorts(6379)) + { + redis.start(); + new com.kingsrook.qbits.redis.RedisQBitProducer().withQBitConfig(new com.kingsrook.qbits.redis.RedisQBitConfig() + .withConnection("presence", new com.kingsrook.qbits.redis.RedisConnectionConfig() + .withHost(redis.getHost()).withPort(redis.getMappedPort(6379)).withTls(false))).produce(instance); + config.withRedisPresence(new com.kingsrook.qbits.collaboration.presence.RedisPresenceConfig() + .withConnectionName("presence").withDeployment("test").withApplication("http-recovery") + .withScopeProvider(new QCodeReference(RedisPresenceServiceTest.TenantScope.class))); + var runtime = new com.kingsrook.qbits.redis.RedisRuntimeService(); + try + { + runtime.start(instance); + String original = ok("POST", "/presence/leases", "{\"table\":\"document\",\"pk\":\"1\"}").get("connectionId").asText(); + awaitHttpPresence(original); + assertThat(redis.execInContainer("redis-cli", "FLUSHDB").getExitCode()).isZero(); + assertError(503, "GET", "/presence?table=document&pk=1", null, "alice"); + assertError(404, "GET", "/presence?table=document&pk=3", null, "alice"); + String oldRenewal = JSON.writeValueAsString(Map.of("connectionId", original, "table", "document", "pk", "1", "mode", "VIEWING", "sequence", 1000)); + assertThat(json(request("POST", "/presence/leases/renew", oldRenewal, "alice"), 409).path("code").asText()) + .isEqualTo("PRESENCE_LEASE_EXPIRED"); + assertError(404, "POST", "/presence/leases/renew", oldRenewal, "outsider"); + String replacement = ok("POST", "/presence/leases", "{\"table\":\"document\",\"pk\":\"1\"}").get("connectionId").asText(); + assertThat(replacement).isNotEqualTo(original); + awaitHttpPresence(replacement); + assertThat(ok("GET", "/presence?table=document&pk=1", null).path("viewers")).hasSize(1); + } + finally + { + runtime.stop(); + runtime.afterApplicationStop(); + } + } + } + + /*************************************************************************** + ** Keep a real HTTP lease alive while the production recovery window elapses. + ***************************************************************************/ + private void awaitHttpPresence(String lease) throws Exception + { + long deadline = System.nanoTime() + Duration.ofSeconds(55).toNanos(); + long sequence = 0; + while(true) + { + String body = JSON.writeValueAsString(Map.of("connectionId", lease, "table", "document", "pk", "1", "mode", "VIEWING", "sequence", ++sequence)); + var response = request("POST", "/presence/leases/renew", body, "alice"); + if(response.statusCode() == 200) + { + assertThat(json(response, 200).path("viewers")).hasSize(1); + return; + } + assertThat(json(response, 503).path("message").asText()).contains("recovering"); + assertThat(System.nanoTime()).as("HTTP presence recovery deadline").isLessThan(deadline); + Thread.sleep(1000); + } + } + /*************************************************************************** ** Shared presence has an explicit unavailable response, never a local fallback. ***************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java new file mode 100644 index 0000000..d118320 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java @@ -0,0 +1,232 @@ +/* + * 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.collaboration; + + +import java.time.Duration; +import com.kingsrook.qbits.collaboration.presence.PresenceLeaseExpiredException; +import com.kingsrook.qbits.collaboration.presence.RedisPresenceStore; +import com.kingsrook.qbits.collaboration.presence.RedisPresenceAudience; +import com.kingsrook.qbits.collaboration.presence.RedisPresenceConfig; +import com.kingsrook.qbits.collaboration.presence.PresenceScopeProvider; +import com.kingsrook.qbits.redis.RedisConnection; +import com.kingsrook.qbits.redis.api.RedisNamespace; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.testcontainers.containers.GenericContainer; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** Loss recovery must be shared across independent clients, not node-local memory. + *******************************************************************************/ +class RedisPresenceRecoveryTest extends PersonalServiceFixture +{ + /*************************************************************************** + ** Keep a real lease alive while waiting for bounded initial recovery. + ***************************************************************************/ + private void awaitVisible(RedisPresenceStore store, String lease) throws Exception + { + long deadline = System.nanoTime() + Duration.ofSeconds(8).toNanos(); + long sequence = 1; + while(System.nanoTime() < deadline) + { + store.renewLease("alice", "Alice", lease, "document", "1", "EDITING", null, sequence++); + try + { + assertThat(store.snapshot("document", "1")).extracting("userId").containsExactly("alice"); + return; + } + catch(IllegalStateException recovering) + { + assertThat(recovering).hasMessageContaining("recover"); + Thread.sleep(50); + } + } + throw (new AssertionError("Presence did not finish bounded recovery")); + } + + + /*************************************************************************** + ** A cleared dataset is unknown, not known-empty; live clients can repopulate. + ***************************************************************************/ + @ParameterizedTest + @ValueSource(booleans = { false, true }) + void totalLossRequiresSharedRecoveryBeforeKnownEmpty(boolean retainGuard) throws Exception + { + try(GenericContainer redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + var address = new HostAndPort(redis.getHost(), redis.getMappedPort(6379)); + var clientConfig = DefaultJedisClientConfig.builder().build(); + try(var first = new RedisConnection(address, clientConfig, 2, Duration.ofMillis(500)); + var second = new RedisConnection(address, clientConfig, 2, Duration.ofMillis(500))) + { + var namespace = new RedisNamespace("test", "app", "tenant-a", "presence"); + var a = new RedisPresenceStore(first.scripts(), namespace, Duration.ofSeconds(2), 10); + var b = new RedisPresenceStore(second.scripts(), namespace, Duration.ofSeconds(2), 10); + String oldLease = a.openLease("alice", "Alice", "document", "1"); + awaitVisible(b, oldLease); + if(retainGuard) + { + var projections = first.client().keys("*").stream().filter(key -> !key.endsWith(":recovery")).toArray(String[]::new); + first.client().del(projections); + } + else + { + first.client().flushDB(); + } + assertThatThrownBy(() -> b.snapshot("document", "1")) + .isInstanceOf(IllegalStateException.class).hasMessageContaining("recover"); + assertThatThrownBy(() -> a.snapshot("document", null)) + .isInstanceOf(IllegalStateException.class).hasMessageContaining("recover"); + assertThatThrownBy(() -> a.renewLease("alice", "Alice", oldLease, "document", "1", "EDITING", null, 999L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + String replacement = b.openLease("alice", "Alice", "document", "1"); + assertThat(replacement).isNotNull().isNotEqualTo(oldLease); + awaitVisible(a, replacement); + assertThat(b.snapshot("document", null)).extracting("userId").containsExactly("alice"); + assertThat(b.snapshot("document", "unoccupied")).isEmpty(); + } + } + } + + + /*************************************************************************** + ** Restore must happen in maintenance, followed by a fresh host-owned generation. + ** Replaying both old guard and lease bytes cannot authorize the new deployment. + ***************************************************************************/ + @Test + void restoredImageCannotAuthorizeRotatedHostGeneration() throws Exception + { + try(GenericContainer redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + var address = new HostAndPort(redis.getHost(), redis.getMappedPort(6379)); + var clientConfig = DefaultJedisClientConfig.builder().build(); + try(var first = new RedisConnection(address, clientConfig, 2, Duration.ofMillis(500)); + var second = new RedisConnection(address, clientConfig, 2, Duration.ofMillis(500))) + { + var oldNamespace = RedisPresenceAudience.namespace(recoveryConfig("before-restore")); + var oldStore = new RedisPresenceStore(first.scripts(), oldNamespace, Duration.ofSeconds(2), 10); + String oldLease = oldStore.openLease("alice", "Alice", "document", "1"); + awaitVisible(oldStore, oldLease); + var image = new java.util.HashMap(); + for(String key : first.client().keys("*")) + { + image.put(key, first.client().dump(key)); + } + oldStore.leave("alice", oldLease, "document", "1", 999L); + first.client().flushDB(); + for(var entry : image.entrySet()) + { + first.client().restore(entry.getKey(), 0, entry.getValue()); + } + var restoredNamespace = RedisPresenceAudience.namespace(recoveryConfig("after-restore")); + var a = new RedisPresenceStore(first.scripts(), restoredNamespace, Duration.ofSeconds(2), 10); + var b = new RedisPresenceStore(second.scripts(), restoredNamespace, Duration.ofSeconds(2), 10); + assertThatThrownBy(() -> a.renewLease("alice", "Alice", oldLease, "document", "1", "EDITING", null, 1000L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + assertThatThrownBy(() -> b.snapshot("document", "1")) + .isInstanceOf(IllegalStateException.class).hasMessageContaining("recover"); + String fresh = b.openLease("alice", "Alice", "document", "1"); + awaitVisible(a, fresh); + assertThat(b.snapshot("document", "1")).extracting("userId").containsExactly("alice"); + assertThatThrownBy(() -> b.renewLease("alice", "Alice", oldLease, "document", "1", "EDITING", null, 1001L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + } + } + } + + + /*************************************************************************** + ** Generation is bounded deployment configuration, never browser input. + ***************************************************************************/ + @Test + void validatesHostGeneration() + { + var errors = new java.util.ArrayList(); + recoveryConfig("valid-20261026").validate(errors); + assertThat(errors).isEmpty(); + for(String invalid : java.util.Arrays.asList(null, "", " ", "x".repeat(65), "invalid:namespace")) + { + errors.clear(); + recoveryConfig(invalid).validate(errors); + assertThat(errors).anySatisfy(error -> assertThat(error).contains("recovery generation")); + } + } + + + /*************************************************************************** + ** Recovery readiness must not depend on which node's lease policy is shorter. + ***************************************************************************/ + @Test + void refusesInconsistentNodeLeaseLifetimes() throws Exception + { + try(GenericContainer redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + try(var connection = new RedisConnection(new HostAndPort(redis.getHost(), redis.getMappedPort(6379)), + DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) + { + var namespace = new RedisNamespace("test", "app", "ttl-policy", "presence"); + var first = new RedisPresenceStore(connection.scripts(), namespace, Duration.ofSeconds(2), 10); + var second = new RedisPresenceStore(connection.scripts(), namespace, Duration.ofSeconds(3), 10); + assertThatThrownBy(() -> first.snapshot("document", null)) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException.class); + assertThatThrownBy(() -> second.snapshot("document", null)) + .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("lifetime differs"); + } + } + } + + + /*************************************************************************** + ** Use the real host-configured storage/audience namespace derivation. + ***************************************************************************/ + private RedisPresenceConfig recoveryConfig(String generation) + { + return (new RedisPresenceConfig().withConnectionName("presence").withDeployment("test").withApplication("app") + .withScopeProvider(new QCodeReference(FixtureScope.class)).withRecoveryGeneration(generation)); + } + + + /*************************************************************************** + ** Tenant scope belongs to the host; this test uses one fixed authenticated scope. + ***************************************************************************/ + public static class FixtureScope implements PresenceScopeProvider + { + /*************************************************************************** + ** Return the fixture's host-selected tenant, without client-supplied fields. + ***************************************************************************/ + @Override + public String tenantScope() + { + return ("tenant-a"); + } + } +} diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceServiceTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceServiceTest.java index 785359f..83e8862 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceServiceTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceServiceTest.java @@ -23,6 +23,7 @@ package com.kingsrook.qbits.collaboration; import com.kingsrook.qbits.collaboration.presence.PresenceScopeProvider; import com.kingsrook.qbits.collaboration.presence.PresenceService; +import com.kingsrook.qbits.collaboration.presence.RedisPresenceBackend; import com.kingsrook.qbits.collaboration.presence.RedisPresenceConfig; import com.kingsrook.qbits.redis.RedisConnectionConfig; import com.kingsrook.qbits.redis.RedisQBitConfig; @@ -72,6 +73,7 @@ class RedisPresenceServiceTest extends PersonalServiceFixture } firstRuntime.start(instance); secondRuntime.start(second.instance); + warmScopes(instance); firstTransport.start(instance); secondTransport.start(second.instance); org.junit.jupiter.api.Assertions.assertTimeoutPreemptively(java.time.Duration.ofSeconds(5), () -> @@ -240,13 +242,14 @@ class RedisPresenceServiceTest extends PersonalServiceFixture redis.start(); new RedisQBitProducer().withQBitConfig(new RedisQBitConfig().withConnection("presence", new RedisConnectionConfig().withHost(redis.getHost()).withPort(redis.getMappedPort(6379)).withTls(false) - .withConnectionTimeoutMillis(200).withSocketTimeoutMillis(200))).produce(instance); + .withConnectionTimeoutMillis(200).withSocketTimeoutMillis(1000))).produce(instance); config.withRedisPresence(new RedisPresenceConfig().withConnectionName("presence").withDeployment("test") .withApplication("fixture").withScopeProvider(new QCodeReference(TenantScope.class))); var runtime = new RedisRuntimeService(); try { runtime.start(instance); + warmScopes(instance); var service = new PresenceService(); String aliceLease; try(var hub = new com.kingsrook.qqq.middleware.javalin.realtime.RealtimeHub(instance)) @@ -319,6 +322,64 @@ class RedisPresenceServiceTest extends PersonalServiceFixture } } + /*************************************************************************** + ** Wait for every dashboard table in both authenticated tenant namespaces. + ** No Redis fixture writes bypass the production cold-recovery contract. + ***************************************************************************/ + private static void warmScopes(com.kingsrook.qqq.backend.core.model.metadata.QInstance node) throws Exception + { + var previousInstance = QContext.getQInstance(); + var previousSession = QContext.getQSession(); + var a = session("warm-a"); + var b = session("warm-b"); + b.clearSecurityKeyValues(); + b.withSecurityKeyValue("tenant", "b"); + var sessions = java.util.List.of(a, b); + try + { + for(var user : sessions) + { + QContext.init(node, user); + var backend = new RedisPresenceBackend(RedisPresenceConfig.forInstance(node)); + for(String table : node.getTables().keySet()) + { + assertThatThrownBy(() -> backend.snapshot(table, null)) + .isInstanceOf(QServiceUnavailableException.class).hasMessageContaining("recovering"); + } + } + long deadline = System.nanoTime() + java.time.Duration.ofSeconds(50).toNanos(); + for(var user : sessions) + { + QContext.init(node, user); + var backend = new RedisPresenceBackend(RedisPresenceConfig.forInstance(node)); + for(String table : node.getTables().keySet()) + { + while(true) + { + try + { + assertThat(backend.snapshot(table, null)).isEmpty(); + break; + } + catch(QServiceUnavailableException recovering) + { + assertThat(recovering).hasMessageContaining("recovering"); + if(System.nanoTime() >= deadline) + { + throw (new AssertionError("Service recovery did not finish", recovering)); + } + Thread.sleep(50); + } + } + } + } + } + finally + { + QContext.init(previousInstance, previousSession); + } + } + /*************************************************************************** ** Tenant identity comes from the authenticated host session, never the request. ***************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java index 7817587..6c774d1 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java @@ -38,6 +38,44 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; *******************************************************************************/ class RedisPresenceStoreTest { + /*************************************************************************** + ** Prime cold fixture namespaces together, using actual server readiness. + ***************************************************************************/ + private static void awaitReady(RedisPresenceStore... stores) throws Exception + { + for(RedisPresenceStore store : stores) + { + try + { + assertThat(store.snapshot("orders", null)).isEmpty(); + } + catch(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException expected) + { + // The first read initializes the shared recovery interval. + } + } + long deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + for(RedisPresenceStore store : stores) + { + while(true) + { + try + { + assertThat(store.snapshot("orders", null)).isEmpty(); + break; + } + catch(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException expected) + { + if(System.nanoTime() >= deadline) + { + throw (new AssertionError("Fixture recovery did not finish", expected)); + } + Thread.sleep(20); + } + } + } + } + /*************************************************************************** ** Server-issued leases cross nodes but cannot survive departure or data loss. ***************************************************************************/ @@ -53,8 +91,9 @@ class RedisPresenceStoreTest var second = new RedisConnection(address, config, 2, Duration.ofMillis(500))) { var scope = new RedisNamespace("test", "app", "issued", "presence"); - var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(45), 10); - var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(45), 10); + var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(2), 10); + var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(2), 10); + awaitReady(a); String lease = a.openLease("one", "One", "orders", "1"); assertThat(lease).startsWith("lease:"); assertThat(b.snapshot("orders", "1")).isEmpty(); @@ -85,7 +124,8 @@ class RedisPresenceStoreTest .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceLeaseExpiredException.class); assertThatThrownBy(() -> b.heartbeat("one", "One", replacement, "orders", "1", "VIEWING", null, 2L)) .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceLeaseExpiredException.class); - assertThat(first.client().dbSize()).isZero(); + assertThat(first.client().dbSize()).isEqualTo(1); + assertThat(first.client().keys("*:recovery")).hasSize(1); } } } @@ -103,6 +143,7 @@ class RedisPresenceStoreTest DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) { var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "issued-expiry", "presence"), Duration.ofMillis(150), 1); + awaitReady(store); String pending = store.openLease("one", "One", "orders", "1"); assertThat(store.openLease("two", "Two", "orders", "2")).isNull(); assertThat(store.snapshot("orders", null)).isEmpty(); @@ -132,7 +173,7 @@ class RedisPresenceStoreTest ** This is evidence for #1295, not an assertion of the desired final contract. ***************************************************************************/ @Test - void completeDataLossCurrentlyAcceptsDelayedHeartbeat() throws Exception + void totalLossIsRecoveringEvenForLegacyHeartbeat() throws Exception { try(GenericContainer redis = new GenericContainer<>(DockerImageName.parse("redis:7.4-alpine")).withExposedPorts(6379)) { @@ -143,8 +184,9 @@ class RedisPresenceStoreTest var second = new RedisConnection(address, config, 2, Duration.ofMillis(500))) { var scope = new RedisNamespace("test", "app", "recovery", "presence"); - var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(45), 10); - var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(45), 10); + var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(2), 10); + var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(2), 10); + awaitReady(a); a.heartbeat("active", "Active", "active-tab", "orders", "1", "VIEWING", null, 1L); a.heartbeat("departed", "Departed", "old-tab", "orders", "1", "EDITING", null, 1L); b.leave("departed", "old-tab", "orders", "1", 3L); @@ -152,10 +194,13 @@ class RedisPresenceStoreTest // Only this disposable test server is cleared. Both clients stay alive. first.client().flushDB(); - assertThat(b.snapshot("orders", "1")).isEmpty(); + assertThatThrownBy(() -> b.snapshot("orders", "1")) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException.class); assertThat(a.heartbeat("departed", "Departed", "old-tab", "orders", "1", "EDITING", null, 2L)).isTrue(); - assertThat(b.snapshot("orders", "1")).extracting("userId").containsExactly("departed"); - assertThat(b.snapshot("orders", null)).extracting("userId").containsExactly("departed"); + assertThatThrownBy(() -> b.snapshot("orders", "1")) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException.class); + assertThatThrownBy(() -> b.snapshot("orders", null)) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException.class); } } } @@ -173,14 +218,16 @@ class RedisPresenceStoreTest try(var connection = new RedisConnection(new HostAndPort(redis.getHost(), redis.getMappedPort(6379)), DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) { - var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "late", "presence"), Duration.ofMillis(150), 10); + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "late", "presence"), Duration.ofMillis(400), 10); + awaitReady(store); store.heartbeat("departed", "Departed", "tab", "orders", "1", "EDITING", null, 1L); store.leave("departed", "tab", "orders", "1", 3L); store.heartbeat("departed", "Departed", "tab", "orders", "1", "EDITING", null, 2L); assertThat(store.snapshot("orders", "1")).isEmpty(); - // Longer-lived peer proves this does not depend on losing entire keys. - var peer = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "late", "presence"), Duration.ofSeconds(45), 10); + // A later renewal keeps the peer live with the same deployment TTL. + Thread.sleep(250); + var peer = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "late", "presence"), Duration.ofMillis(400), 10); peer.heartbeat("active", "Active", "peer-tab", "orders", "1", "VIEWING", null, 1L); Thread.sleep(250); assertThat(peer.snapshot("orders", "1")).extracting("userId").containsExactly("active"); @@ -205,8 +252,10 @@ class RedisPresenceStoreTest var second = new RedisConnection(address, config, 4, Duration.ofMillis(500))) { var scope = new RedisNamespace("test", "app", "tenant-a", "presence"); - var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(45), 100); - var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(45), 100); + var a = new RedisPresenceStore(first.scripts(), scope, Duration.ofSeconds(2), 100); + var b = new RedisPresenceStore(second.scripts(), scope, Duration.ofSeconds(2), 100); + var tenantB = new RedisPresenceStore(second.scripts(), new RedisNamespace("test", "app", "tenant-b", "presence"), Duration.ofSeconds(2), 100); + awaitReady(a, tenantB); for(int i = 0; i < 6; i++) { assertThat((i % 2 == 0 ? a : b).heartbeat("user-" + i, "Person " + i, "tab-" + i, "orders", "123", @@ -225,7 +274,6 @@ class RedisPresenceStoreTest assertThat(a.snapshot("orders", "123").getFirst().mode()).isEqualTo("VIEWING"); assertThat(a.heartbeat("user-0", "Person 0", "tab-0", "orders", "123", "EDITING", "status", 4L)).isTrue(); assertThat(b.snapshot("orders", "123").getFirst().mode()).isEqualTo("EDITING"); - var tenantB = new RedisPresenceStore(second.scripts(), new RedisNamespace("test", "app", "tenant-b", "presence"), Duration.ofSeconds(45), 100); assertThat(tenantB.snapshot("orders", "123")).isEmpty(); assertThat(a.snapshot("orders", "124")).isEmpty(); @@ -250,6 +298,7 @@ class RedisPresenceStoreTest assertThat(a.snapshot("orders", "124")).isEmpty(); var shortLease = new RedisPresenceStore(first.scripts(), new RedisNamespace("test", "app", "short", "presence"), Duration.ofMillis(150), 10); + awaitReady(shortLease); assertThat(shortLease.heartbeat("user", "Person", "tab", "orders", "123", "EDITING", "status", 1L)).isTrue(); long deadline = System.nanoTime() + Duration.ofSeconds(3).toNanos(); while(!shortLease.snapshot("orders", "123").isEmpty() && System.nanoTime() < deadline) @@ -278,7 +327,8 @@ class RedisPresenceStoreTest try(var connection = new RedisConnection(new HostAndPort(redis.getHost(), redis.getMappedPort(6379)), DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) { - var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "capacity", "presence"), Duration.ofSeconds(45), 1); + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", "capacity", "presence"), Duration.ofSeconds(2), 1); + awaitReady(store); assertThat(store.heartbeat("one", "One", "tab", "orders", "1", "VIEWING", null, 1L)).isTrue(); assertThat(store.heartbeat("two", "Two", "tab", "orders", "2", "VIEWING", null, 1L)).isFalse(); assertThat(store.snapshot("orders", "2")).isEmpty(); @@ -305,7 +355,8 @@ class RedisPresenceStoreTest { for(String suffix : java.util.List.of("state", "expiry")) { - var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", suffix, "presence"), Duration.ofSeconds(45), 10); + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", suffix, "presence"), Duration.ofSeconds(2), 10); + awaitReady(store); assertThat(store.heartbeat("one", "One", "tab", "orders", "1", "VIEWING", null, 1L)).isTrue(); // KEYS is confined to this disposable test server, never the application path. String recordKey = connection.client().keys("*:record:*:" + suffix).iterator().next(); @@ -321,15 +372,27 @@ class RedisPresenceStoreTest } for(String room : java.util.List.of("record", "table")) { - var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", room, "presence"), Duration.ofSeconds(45), 10); - store.heartbeat("one", "One", "tab", "orders", "1", "VIEWING", null, 1L); - store.leave("one", "tab", "orders", "1", 3L); + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", room, "presence"), Duration.ofSeconds(2), 10); + awaitReady(store); + String lease = store.openLease("one", "One", "orders", "1"); + store.renewLease("one", "One", lease, "orders", "1", "VIEWING", null, 1L); + store.leave("one", lease, "orders", "1", 3L); var lostKeys = connection.client().keys(room.equals("record") ? "*:record:*" : "*:table:*"); connection.client().del(lostKeys.toArray(String[]::new)); - assertThatThrownBy(() -> store.snapshot("orders", "1")) - .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); - assertThatThrownBy(() -> store.heartbeat("one", "One", "tab", "orders", "1", "EDITING", null, 2L)) - .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); + if(room.equals("record")) + { + assertThatThrownBy(() -> store.snapshot("orders", "1")) + .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); + assertThatThrownBy(() -> store.renewLease("one", "One", lease, "orders", "1", "EDITING", null, 2L)) + .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); + } + else + { + assertThatThrownBy(() -> store.snapshot("orders", "1")) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException.class); + assertThatThrownBy(() -> store.renewLease("one", "One", lease, "orders", "1", "EDITING", null, 2L)) + .isInstanceOf(com.kingsrook.qbits.collaboration.presence.PresenceLeaseExpiredException.class); + } assertThat(connection.client().keys(room.equals("record") ? "*:record:*" : "*:table:*")).isEmpty(); connection.client().flushDB(); } -- 2.54.0 From 1c195d689e021ee32299517c4c7fba4f1448a08c Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 06:50:30 -0500 Subject: [PATCH 2/6] test: verify browser presence recovery across JVMs Exercise unavailable rendering and automatic lease rejoin through actual Chromium and two backend processes after disposable Redis loss. Retain traces and screenshots; frontend serving remains fixture-only and opt-in. Co-Authored-By: GPT-6 (Codex) --- docs/PRESENCE-RECOVERY.md | 18 +++++++- docs/acceptance/presence_recovery_browser.mjs | 41 +++++++++++++++++++ .../acceptance/presence_recovery_processes.py | 11 ++++- .../devserver/SharedPresenceDevServer.java | 2 +- 4 files changed, 68 insertions(+), 4 deletions(-) create mode 100644 docs/acceptance/presence_recovery_browser.mjs diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md index 1254e62..835a848 100644 --- a/docs/PRESENCE-RECOVERY.md +++ b/docs/PRESENCE-RECOVERY.md @@ -71,8 +71,7 @@ version rolling deployment. Existing published v1 snapshots are unchanged. This uses a loopback HTTP server with fixture authentication, not a browser or a deployed host. The focused test and the combined full suite both pass. - Still required: remaining cold/readiness and partial-loss edge cases, - browser re-registration and unavailable - rendering, enforced host maintenance/reset, and measured + browser-matrix and deployed-host qualification, enforced host maintenance/reset, and measured per-operation cost before adoption. The implementation uses no INFO call or lifecycle polling cache; its additional guard reads/writes still need measurement. - This is an unreleased draft candidate. Publication and production deployment @@ -98,3 +97,18 @@ python docs/acceptance/presence_recovery_processes.py --java /path/to/java --rep Requires Python 3, Java 21 and Docker. JVM output and argument files stay under `target/recovery-processes-*` for diagnosis. The harness uses the production 45-second recovery window, with bounded startup and recovery deadlines. + +## Chromium recovery acceptance + +The same harness accepts `--frontend-repo /path/to/qqq-frontend-next`. It copies +that checkout's existing `out` export into the owned test output directory and +serves it through the real middleware on both JVMs. Playwright is loaded from +the frontend checkout; no API mocks are installed. Both browser users first +see the other viewer, then Redis FLUSHDB triggers the unavailable message and +removes stale people. Both viewer badges return without reloading either page. + +This check passed on October 26 with the local export; traces and screenshots +are under `target/recovery-processes-229f305e83`. The final screenshot was also +visually inspected. This is Chromium only and does not establish provenance +of a newly published frontend artifact or deployed Kof22 acceptance. Fixture +frontend serving is opt-in via `collab.dev.frontend`; its default remains false. diff --git a/docs/acceptance/presence_recovery_browser.mjs b/docs/acceptance/presence_recovery_browser.mjs new file mode 100644 index 0000000..1a39e8c --- /dev/null +++ b/docs/acceptance/presence_recovery_browser.mjs @@ -0,0 +1,41 @@ +// Real browser recovery against the disposable two-JVM harness. No API mocks. +import { createRequire } from 'node:module' +import { execFileSync } from 'node:child_process' +import path from 'node:path' + +const [frontend, first, second, container, output] = process.argv.slice(2) +const require = createRequire(path.join(frontend, 'package.json')) +const { chromium, expect } = require('@playwright/test') +const browser = await chromium.launch({ headless: true }) +const contexts = [] +try { + const pages = [] + for (const [port, user] of [[first, 'alice'], [second, 'bob']]) { + const context = await browser.newContext() + contexts.push(context) + await context.tracing.start({ screenshots: true, snapshots: true }) + const page = await context.newPage() + await page.goto(`http://127.0.0.1:${port}/?createMockSession=true&user=${user}`) + await page.goto(`http://127.0.0.1:${port}/app/person/1`) + pages.push(page) + } + const unavailable = page => page.getByText('Presence temporarily unavailable', { exact: true }) + await expect(pages[0].locator('[data-qqq-id="presence-bob"]')).toBeVisible({ timeout: 90000 }) + await expect(pages[1].locator('[data-qqq-id="presence-alice"]')).toBeVisible({ timeout: 90000 }) + execFileSync('docker', ['exec', container, 'redis-cli', 'FLUSHDB']) + for (const page of pages) await expect(unavailable(page)).toBeVisible({ timeout: 35000 }) + await expect(pages[0].locator('[data-qqq-id="presence-bob"]')).toHaveCount(0) + await expect(pages[1].locator('[data-qqq-id="presence-alice"]')).toHaveCount(0) + await expect(pages[0].locator('[data-qqq-id="presence-bob"]')).toBeVisible({ timeout: 90000 }) + await expect(pages[1].locator('[data-qqq-id="presence-alice"]')).toBeVisible({ timeout: 90000 }) + for (const [index, page] of pages.entries()) { + await expect(unavailable(page)).toHaveCount(0) + await page.screenshot({ path: path.join(output, `recovered-${index}.png`) }) + } + console.log('PASS: Chromium shows recovery unavailable, clears stale people, and rejoins both JVMs without reload.') +} finally { + for (const [index, context] of contexts.entries()) { + await context.tracing.stop({ path: path.join(output, `browser-${index}.zip`) }) + } + await browser.close() +} diff --git a/docs/acceptance/presence_recovery_processes.py b/docs/acceptance/presence_recovery_processes.py index f5f26ae..5545c90 100644 --- a/docs/acceptance/presence_recovery_processes.py +++ b/docs/acceptance/presence_recovery_processes.py @@ -9,6 +9,7 @@ import json import os from pathlib import Path import socket +import shutil import subprocess import time import urllib.error @@ -74,12 +75,16 @@ def main(): parser = argparse.ArgumentParser() parser.add_argument("--java", required=True) parser.add_argument("--repo", type=Path, required=True) + parser.add_argument("--frontend-repo", type=Path) args = parser.parse_args() repo = args.repo.resolve() output = repo / "target" / ("recovery-processes-" + uuid.uuid4().hex[:10]) output.mkdir() classpath = os.pathsep.join([str(repo / "target/test-classes"), str(repo / "target/classes"), (repo / "target/recovery-process-classpath.txt").read_text().strip()]) + if args.frontend_repo: + shutil.copytree(args.frontend_repo.resolve() / "out", output / "next-dashboard") + classpath = str(output) + os.pathsep + classpath container = "qqq-recovery-" + uuid.uuid4().hex[:12] processes, logs = [], [] try: @@ -89,7 +94,7 @@ def main(): ports = [free_port(), free_port()] for port in ports: argfile = output / f"node-{port}.args" - argfile.write_text('-cp\n"' + classpath.replace("\\", "/") + '"\n' + argfile.write_text('-Dcollab.dev.frontend=' + str(bool(args.frontend_repo)).lower() + '\n-cp\n"' + classpath.replace("\\", "/") + '"\n' 'com.kingsrook.qbits.collaboration.devserver.SharedPresenceDevServer\n' f'{port}\n{redis_port}\n', encoding="utf-8") log = (output / f"node-{port}.log").open("w") @@ -103,6 +108,10 @@ def main(): assert process.poll() is None, f"Node exited; see {output}" assert time.monotonic() < deadline, f"Startup timed out; see {output}" time.sleep(0.2) + if args.frontend_repo: + subprocess.run(["node", str(repo / "docs/acceptance/presence_recovery_browser.mjs"), + str(args.frontend_repo.resolve()), str(ports[0]), str(ports[1]), container, str(output)], check=True) + return def join(): """Acquire each person's new lease on a different node.""" return {user: require(200, request(port, user, "POST", "/presence/leases", { diff --git a/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java b/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java index 2f618b0..fbad83a 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java @@ -70,7 +70,7 @@ public class SharedPresenceDevServer var transport = new RedisRealtimeRuntimeService(); new com.kingsrook.qqq.backend.core.health.HealthMetaDataProvider().defineAll(instance, "memory", null); var health = new com.kingsrook.qqq.backend.core.health.HealthRunner(instance, "acceptance-" + port, java.time.Clock.systemUTC()); - var server = CollaborationDevServer.server(instance, port, false); + var server = CollaborationDevServer.server(instance, port, Boolean.getBoolean("collab.dev.frontend")); try { redis.start(instance); -- 2.54.0 From 4b0d6a32d6bd8ecba023857d3661d9f3f5980975 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 07:10:54 -0500 Subject: [PATCH 3/6] perf: avoid repeated table scans during presence recovery Clean the table when its recovery generation is created, retaining lazy cleanup for record rooms. Add guard-loss regression and diagnostic measurements at 1,000 and 10,000 leases; expose browser profiles and await dashboard redirect before navigation. Co-Authored-By: GPT-6 (Codex) --- docs/PRESENCE-RECOVERY.md | 6 +- docs/acceptance/RECOVERY-COST.md | 56 +++++++ docs/acceptance/presence_recovery_browser.mjs | 21 ++- .../acceptance/presence_recovery_processes.py | 3 +- .../resources/presence/redis-recovery.lua | 18 ++- .../RedisPresenceRecoveryCostProbe.java | 150 ++++++++++++++++++ .../RedisPresenceRecoveryTest.java | 32 ++++ 7 files changed, 271 insertions(+), 15 deletions(-) create mode 100644 docs/acceptance/RECOVERY-COST.md create mode 100644 src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryCostProbe.java diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md index 835a848..04204dd 100644 --- a/docs/PRESENCE-RECOVERY.md +++ b/docs/PRESENCE-RECOVERY.md @@ -51,12 +51,12 @@ version rolling deployment. Existing published v1 snapshots are unchanged. - Before the patch, a two-client disposable Redis test failed because total loss returned an empty snapshot. -- Five focused tests now pass: full loss, loss of volatile projections while the +- Six focused tests now pass: full loss, loss of volatile projections while the guard survives, restored guard/lease bytes followed by a host generation change, bounded generation validation, and rejection of inconsistent lease - lifetimes across nodes. These use independent clients in + lifetimes across nodes, plus guard loss across previously unvisited records. These use independent clients in one JVM, not a deployed HA cluster. -- Full wrapper verification passed on October 26: 183 tests, no failures, +- Full wrapper verification passed on October 26: 184 tests, no failures, errors or skips; packaging and coverage gates passed. This includes the original seven store tests and two-node six-viewer/two-editor scenario. The admin dashboard reads every table, so the fixture initializes every diff --git a/docs/acceptance/RECOVERY-COST.md b/docs/acceptance/RECOVERY-COST.md new file mode 100644 index 0000000..74121a0 --- /dev/null +++ b/docs/acceptance/RECOVERY-COST.md @@ -0,0 +1,56 @@ +# Recovery cost diagnostic — October 26, 2026 + +One Java 21 process, one Redis connection, disposable loopback Redis 7.4-alpine, +1,000 issued leases, ten viewers per record, no TLS or HTTP. The probe keeps +all leases alive through the real 45-second recovery window and verifies that +the final table snapshot contains all 1,000 viewers. Each phase samples 200 +renewals with increasing sequence numbers. This is not a capacity or SLA claim. + +| Renewal latency (ms) | During recovery | Ready | +|---|---:|---:| +| p50 | 3.6779 | 0.5134 | +| p95 | 4.1956 | 0.7226 | +| p99 | 4.5447 | 0.9122 | +| Maximum | 6.8271 | 1.4886 | + +Population took 4,640.799 ms. The run passed all correctness checks. This first +implementation scanned touched projections on every request during recovery. + +After changing table cleanup to occur atomically only when creating its recovery +generation (individual record cleanup remains lazy), the same diagnostic passed: + +| Renewal latency (ms) | Optimized recovery | Ready | +|---|---:|---:| +| p50 | 0.6375 | 0.5029 | +| p95 | 0.8150 | 0.5762 | +| p99 | 0.9406 | 0.7500 | +| Maximum | 4.2280 | 1.2251 | + +Optimized population took 1,634.486 ms. All 1,000 leases survived. The focused +recovery/store suite also passed 13 tests, including guard loss with previously +unvisited record rooms. Full optimized verification passed 184 tests with no failures/errors/skips and coverage gates passed. These are +individual local runs and need further qualification before claiming supported +production capacity. They do not isolate every source +of latency, compare against an earlier release, or measure sustained concurrent +load, saturation, TLS, HTTP or browser propagation. + +At 10,000 entries the optimized diagnostic also passed, retaining all leases. +Population took 10,687.2105 ms; the same 200-sample phases measured: + +| Renewal latency (ms) | Optimized recovery, 10,000 entries | Ready | +|---|---:|---:| +| p50 | 0.4978 | 0.5341 | +| p95 | 0.5914 | 0.7108 | +| p99 | 0.7255 | 1.2964 | +| Maximum | 0.7544 | 1.8799 | + +Do not infer that larger tables are faster from these separate local runs. +JIT warmup, scheduling and concurrent machine activity were not controlled. +The useful result is that the optimization avoids a repeated whole-table scan +during recovery and both populated-table runs preserved lease correctness. + +The explicit `RedisPresenceRecoveryCostProbe` test-source main class is not +automatically run by the unit suite and imposes no machine-dependent CI gate. +After test compilation, run it on the test classpath with argument `1000`. +It fails if the real recovery window ends before the cold measurement, rather +than silently reporting ready-state latency as recovery latency. diff --git a/docs/acceptance/presence_recovery_browser.mjs b/docs/acceptance/presence_recovery_browser.mjs index 1a39e8c..5171454 100644 --- a/docs/acceptance/presence_recovery_browser.mjs +++ b/docs/acceptance/presence_recovery_browser.mjs @@ -3,19 +3,30 @@ import { createRequire } from 'node:module' import { execFileSync } from 'node:child_process' import path from 'node:path' -const [frontend, first, second, container, output] = process.argv.slice(2) +const [frontend, first, second, container, output, profile = 'chromium'] = process.argv.slice(2) const require = createRequire(path.join(frontend, 'package.json')) -const { chromium, expect } = require('@playwright/test') -const browser = await chromium.launch({ headless: true }) +const { chromium, firefox, webkit, devices, expect } = require('@playwright/test') +const profiles = { + chromium: [chromium, devices['Desktop Chrome']], + firefox: [firefox, devices['Desktop Firefox']], + webkit: [webkit, devices['Desktop Safari']], + mobile: [chromium, devices['Pixel 7']], + tablet: [webkit, devices['iPad (gen 7)']], +} +if (!profiles[profile]) throw new Error(`Unknown browser profile: ${profile}`) +const [engine, device] = profiles[profile] +const browser = await engine.launch({ headless: true }) const contexts = [] try { const pages = [] for (const [port, user] of [[first, 'alice'], [second, 'bob']]) { - const context = await browser.newContext() + const context = await browser.newContext(device) contexts.push(context) await context.tracing.start({ screenshots: true, snapshots: true }) const page = await context.newPage() await page.goto(`http://127.0.0.1:${port}/?createMockSession=true&user=${user}`) + // The root export redirects after hydration; finish it before record navigation. + await page.waitForURL(url => /^\/app\/?$/.test(url.pathname), { waitUntil: 'domcontentloaded' }) await page.goto(`http://127.0.0.1:${port}/app/person/1`) pages.push(page) } @@ -32,7 +43,7 @@ try { await expect(unavailable(page)).toHaveCount(0) await page.screenshot({ path: path.join(output, `recovered-${index}.png`) }) } - console.log('PASS: Chromium shows recovery unavailable, clears stale people, and rejoins both JVMs without reload.') + console.log(`PASS: ${profile} shows recovery unavailable, clears stale people, and rejoins both JVMs without reload.`) } finally { for (const [index, context] of contexts.entries()) { await context.tracing.stop({ path: path.join(output, `browser-${index}.zip`) }) diff --git a/docs/acceptance/presence_recovery_processes.py b/docs/acceptance/presence_recovery_processes.py index 5545c90..4e01f66 100644 --- a/docs/acceptance/presence_recovery_processes.py +++ b/docs/acceptance/presence_recovery_processes.py @@ -76,6 +76,7 @@ def main(): parser.add_argument("--java", required=True) parser.add_argument("--repo", type=Path, required=True) parser.add_argument("--frontend-repo", type=Path) + parser.add_argument("--browser-profile", choices=("chromium", "firefox", "webkit", "mobile", "tablet"), default="chromium") args = parser.parse_args() repo = args.repo.resolve() output = repo / "target" / ("recovery-processes-" + uuid.uuid4().hex[:10]) @@ -110,7 +111,7 @@ def main(): time.sleep(0.2) if args.frontend_repo: subprocess.run(["node", str(repo / "docs/acceptance/presence_recovery_browser.mjs"), - str(args.frontend_repo.resolve()), str(ports[0]), str(ports[1]), container, str(output)], check=True) + str(args.frontend_repo.resolve()), str(ports[0]), str(ports[1]), container, str(output), args.browser_profile], check=True) return def join(): """Acquire each person's new lease on a different node.""" diff --git a/src/main/resources/presence/redis-recovery.lua b/src/main/resources/presence/redis-recovery.lua index 9592a89..9eaeca9 100644 --- a/src/main/resources/presence/redis-recovery.lua +++ b/src/main/resources/presence/redis-recovery.lua @@ -3,6 +3,7 @@ local function presenceRecovery(keys, ttl, candidate, now) local encoded = redis.call('GET', keys[#keys]) local guard + local created = false if encoded then guard = cjson.decode(encoded) if type(guard.generation) ~= 'string' or type(guard.readyAt) ~= 'number' or type(guard.liveUntil) ~= 'number' or type(guard.ttl) ~= 'number' then @@ -17,18 +18,23 @@ local function presenceRecovery(keys, ttl, candidate, now) end end if not guard then + created = true guard = { generation = candidate, readyAt = now + ttl, liveUntil = 0, ttl = ttl } redis.call('SET', keys[#keys], cjson.encode(guard)) end -- During recovery, discard only projections from an invalidated generation. - -- This also frees capacity for new reservations without waiting for old TTLs. + -- The table is cleaned atomically when its generation is created. Subsequent + -- requests only need to clean the touched record; all new table writes carry + -- the current generation. Avoid rescanning the whole table on every renewal. if now < guard.readyAt then for i = 1, #keys - 1, 2 do - for _, member in ipairs(redis.call('HKEYS', keys[i + 1])) do - local data = cjson.decode(redis.call('HGET', keys[i + 1], member)) - if data.generation ~= guard.generation then - redis.call('HDEL', keys[i + 1], member) - redis.call('ZREM', keys[i], member) + if created or i < #keys - 2 then + for _, member in ipairs(redis.call('HKEYS', keys[i + 1])) do + local data = cjson.decode(redis.call('HGET', keys[i + 1], member)) + if data.generation ~= guard.generation then + redis.call('HDEL', keys[i + 1], member) + redis.call('ZREM', keys[i], member) + end end end end diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryCostProbe.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryCostProbe.java new file mode 100644 index 0000000..90b0db6 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryCostProbe.java @@ -0,0 +1,150 @@ +/* + * 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.collaboration; + +import java.time.Duration; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; +import com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException; +import com.kingsrook.qbits.collaboration.presence.RedisPresenceStore; +import com.kingsrook.qbits.redis.RedisConnection; +import com.kingsrook.qbits.redis.api.RedisNamespace; +import com.kingsrook.qqq.backend.core.utils.JsonUtils; +import org.testcontainers.containers.GenericContainer; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; + +/******************************************************************************* + ** Explicit local diagnostic, not a CI latency gate or deployment capacity claim. + *******************************************************************************/ +public class RedisPresenceRecoveryCostProbe +{ + /*************************************************************************** + ** Compare issued-lease renewal cost during real recovery and after readiness. + ***************************************************************************/ + public static void main(String[] args) throws Exception + { + int count = args.length == 0 ? 1000 : Integer.parseInt(args[0]); + if(count < 10 || count > 10000 || count % 10 != 0) + { + throw (new IllegalArgumentException("Use a multiple of ten between 10 and 10000")); + } + try(var redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + String host = "localhost".equals(redis.getHost()) ? "127.0.0.1" : redis.getHost(); + try(var connection = new RedisConnection(new HostAndPort(host, redis.getMappedPort(6379)), + DefaultJedisClientConfig.builder().socketTimeoutMillis(5000).build(), 1, Duration.ofSeconds(1))) + { + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("cost", "local", "tenant", "presence"), Duration.ofSeconds(45), count); + String[] leases = new String[count]; + long[] sequences = new long[count]; + long start = System.nanoTime(); + for(int i = 0; i < count; i++) + { + leases[i] = store.openLease("user" + i, "Person " + i, "document", Integer.toString(i / 10)); + renew(store, leases, sequences, i); + } + double populationMillis = (System.nanoTime() - start) / 1_000_000.0; + requireRecovering(store); + Map cold = sample(store, leases, sequences); + requireRecovering(store); + long deadline = System.nanoTime() + Duration.ofSeconds(60).toNanos(); + while(true) + { + for(int i = 0; i < count; i++) + { + renew(store, leases, sequences, i); + } + try + { + if(store.snapshot("document", null).size() != count) + { + throw (new AssertionError("Population did not survive recovery")); + } + break; + } + catch(PresenceRecoveringException expected) + { + if(System.nanoTime() >= deadline) + { + throw (new AssertionError("Recovery exceeded deadline")); + } + Thread.sleep(5000); + } + } + var result = new LinkedHashMap(); + result.put("entries", count); + result.put("samplesPerPhase", 200); + result.put("populationMillis", populationMillis); + result.put("recoveryRenewMillis", cold); + result.put("readyRenewMillis", sample(store, leases, sequences)); + result.put("scope", "One JVM, one connection, loopback Redis, no TLS/HTTP; diagnostic only"); + System.out.println(JsonUtils.toJson(result)); + } + } + } + + /*************************************************************************** + ** Sample the same issued leases with increasing sequence numbers. + ***************************************************************************/ + private static Map sample(RedisPresenceStore store, String[] leases, long[] sequences) + { + double[] times = new double[200]; + for(int i = 0; i < times.length; i++) + { + long before = System.nanoTime(); + renew(store, leases, sequences, i % leases.length); + times[i] = (System.nanoTime() - before) / 1_000_000.0; + } + Arrays.sort(times); + return (Map.of("p50", times[99], "p95", times[189], "p99", times[197], "max", times[199])); + } + + /*************************************************************************** + ** Renewal must not silently fail while measuring its elapsed time. + ***************************************************************************/ + private static void renew(RedisPresenceStore store, String[] leases, long[] sequences, int i) + { + if(!store.renewLease("user" + i, "Person " + i, leases[i], "document", Integer.toString(i / 10), "VIEWING", null, ++sequences[i])) + { + throw (new AssertionError("Renewal refused")); + } + } + + /*************************************************************************** + ** Do not label a measurement cold if the actual recovery window has ended. + ***************************************************************************/ + private static void requireRecovering(RedisPresenceStore store) + { + try + { + store.snapshot("document", "0"); + } + catch(PresenceRecoveringException expected) + { + return; + } + throw (new AssertionError("Recovery finished before cold measurement")); + } +} diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java index d118320..12ef1c6 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java @@ -24,6 +24,7 @@ package com.kingsrook.qbits.collaboration; import java.time.Duration; import com.kingsrook.qbits.collaboration.presence.PresenceLeaseExpiredException; +import com.kingsrook.qbits.collaboration.presence.PresenceRecoveringException; import com.kingsrook.qbits.collaboration.presence.RedisPresenceStore; import com.kingsrook.qbits.collaboration.presence.RedisPresenceAudience; import com.kingsrook.qbits.collaboration.presence.RedisPresenceConfig; @@ -116,6 +117,37 @@ class RedisPresenceRecoveryTest extends PersonalServiceFixture } + /*************************************************************************** + ** Guard loss cleans the table once and untouched record rooms on demand. + ***************************************************************************/ + @Test + void missingGuardRetiresUnvisitedRecordsWithoutRemovingNewLeases() throws Exception + { + try(var redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + try(var connection = new RedisConnection(new HostAndPort(redis.getHost(), redis.getMappedPort(6379)), + DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) + { + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "guard-loss", "tenant", "presence"), Duration.ofSeconds(2), 2); + String oldAlice = store.openLease("alice", "Alice", "document", "1"); + String oldBob = store.openLease("bob", "Bob", "document", "2"); + connection.client().del(connection.client().keys("*:recovery").toArray(String[]::new)); + assertThatThrownBy(() -> store.snapshot("document", null)).isInstanceOf(PresenceRecoveringException.class); + String alice = store.openLease("alice", "Alice", "document", "1"); + String bob = store.openLease("bob", "Bob", "document", "2"); + assertThatThrownBy(() -> store.renewLease("alice", "Alice", oldAlice, "document", "1", "VIEWING", null, 99L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + assertThatThrownBy(() -> store.renewLease("bob", "Bob", oldBob, "document", "2", "VIEWING", null, 99L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + assertThat(store.renewLease("bob", "Bob", bob, "document", "2", "VIEWING", null, 1L)).isTrue(); + awaitVisible(store, alice); + assertThat(store.snapshot("document", "1")).extracting("userId").containsExactly("alice"); + } + } + } + + /*************************************************************************** ** Restore must happen in maintenance, followed by a fresh host-owned generation. ** Replaying both old guard and lease bytes cannot authorize the new deployment. -- 2.54.0 From 0062644530a3e94f217312a2fcbf9e0794d9f9e7 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 07:31:33 -0500 Subject: [PATCH 4/6] test: verify controlled presence restore across JVMs Co-Authored-By: GPT-6 (Codex) --- docs/PRESENCE-RECOVERY.md | 56 ++++++-- docs/acceptance/presence_restore_processes.py | 135 ++++++++++++++++++ .../devserver/SharedPresenceDevServer.java | 1 + 3 files changed, 181 insertions(+), 11 deletions(-) create mode 100644 docs/acceptance/presence_restore_processes.py diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md index 04204dd..20c3f0c 100644 --- a/docs/PRESENCE-RECOVERY.md +++ b/docs/PRESENCE-RECOVERY.md @@ -1,7 +1,8 @@ # Presence recovery candidate — QQQ #1300 -This branch is an uncommitted implementation candidate. It is not part of the -published snapshots and has not passed the existing store/service suites. +This is an unreleased implementation candidate in collaboration draft PR47. +The existing store/service suites and CI pass; host adoption and enforced +maintenance remain incomplete. It is not part of the published snapshots. ## Contract under qualification @@ -71,9 +72,10 @@ version rolling deployment. Existing published v1 snapshots are unchanged. This uses a loopback HTTP server with fixture authentication, not a browser or a deployed host. The focused test and the combined full suite both pass. - Still required: remaining cold/readiness and partial-loss edge cases, - browser-matrix and deployed-host qualification, enforced host maintenance/reset, and measured - per-operation cost before adoption. The implementation uses no INFO call or - lifecycle polling cache; its additional guard reads/writes still need measurement. + deployed-host qualification and enforced host maintenance/reset. Local + per-operation cost measurements at 1,000 and 10,000 leases are recorded in + `acceptance/RECOVERY-COST.md`; sustained production capacity is unqualified. + The implementation uses no INFO call or lifecycle polling cache. - This is an unreleased draft candidate. Publication and production deployment have not occurred; the remaining acceptance work above is still required. @@ -98,7 +100,7 @@ Requires Python 3, Java 21 and Docker. JVM output and argument files stay under `target/recovery-processes-*` for diagnosis. The harness uses the production 45-second recovery window, with bounded startup and recovery deadlines. -## Chromium recovery acceptance +## Browser recovery acceptance The same harness accepts `--frontend-repo /path/to/qqq-frontend-next`. It copies that checkout's existing `out` export into the owned test output directory and @@ -107,8 +109,40 @@ the frontend checkout; no API mocks are installed. Both browser users first see the other viewer, then Redis FLUSHDB triggers the unavailable message and removes stale people. Both viewer badges return without reloading either page. -This check passed on October 26 with the local export; traces and screenshots -are under `target/recovery-processes-229f305e83`. The final screenshot was also -visually inspected. This is Chromium only and does not establish provenance -of a newly published frontend artifact or deployed Kof22 acceptance. Fixture -frontend serving is opt-in via `collab.dev.frontend`; its default remains false. +The final fixed-candidate run passed all five profiles on October 26 against +commit `4b0d6a32d6bd8ecba023857d3661d9f3f5980975`: Chromium, Firefox, WebKit, +Pixel 7 emulation and iPad emulation. Source/compiled Lua hashes were checked +before the run and after each profile. Traces and screenshots remain under +each `target/recovery-processes-*` directory. The earlier Chromium screenshot +in `target/recovery-processes-229f305e83` was also visually inspected. + +Select profiles using `--browser-profile chromium|firefox|webkit|mobile|tablet`. +These tests use the local export and do not establish provenance of a newly +published frontend artifact, physical-device behavior or deployed Kof22 +acceptance. Fixture frontend serving is opt-in via `collab.dev.frontend`; +its default remains false. + +## Controlled restore acceptance + +`acceptance/presence_restore_processes.py` exercises a disposable Redis RDB +restore with two independent JVMs. It saves the populated dataset, verifies +both old JVMs have terminated and their listeners have closed, clears the +dataset, reloads the saved image and starts both nodes with a shared fresh +generation supplied outside Redis. It checks recovery responses, rejects old +leases on both nodes, and requires two fresh viewers to rejoin across nodes. +The final run passed on October 26; logs are retained in +`target/restore-processes-aec5288289`. Both leases are explicitly renewed before +saving, and rejection must complete within 40 seconds of that renewal, before +their normal 45-second expiry. Old leases are rejected again after fresh +viewers have rejoined. + +The fixture enables Redis DEBUG RELOAD only on its owned loopback container. +Run after test compilation and classpath preparation: + +```text +python docs/acceptance/presence_restore_processes.py --java /path/to/java --repo /path/to/repository +``` + +This is a protocol acceptance fixture, not host maintenance automation. Actual +host admission closure, replacement prevention and generation configuration +still require implementation and staging validation. diff --git a/docs/acceptance/presence_restore_processes.py b/docs/acceptance/presence_restore_processes.py new file mode 100644 index 0000000..37a2e7c --- /dev/null +++ b/docs/acceptance/presence_restore_processes.py @@ -0,0 +1,135 @@ +"""Restore a disposable Redis image while both fixture JVMs are stopped. + +This verifies the controlled-restore protocol, not production orchestration. +Requires the same compiled test classpath as presence_recovery_processes.py. +""" + +import argparse +import os +from pathlib import Path +import socket +import subprocess +import time +import uuid + +from presence_recovery_processes import free_port, request, require, settle + + +def main(): + """Own a Redis fixture and two generations of independently running nodes.""" + parser = argparse.ArgumentParser() + parser.add_argument("--java", required=True) + parser.add_argument("--repo", type=Path, required=True) + args = parser.parse_args() + repo = args.repo.resolve() + output = repo / "target" / ("restore-processes-" + uuid.uuid4().hex[:10]) + output.mkdir() + classpath = os.pathsep.join([str(repo / "target/test-classes"), str(repo / "target/classes"), + (repo / "target/recovery-process-classpath.txt").read_text().strip()]) + container = "qqq-restore-" + uuid.uuid4().hex[:12] + processes, logs = [], [] + ports = [free_port(), free_port()] + + def redis(*command): + """Run commands only against this owned, disposable Redis container.""" + return subprocess.check_output(["docker", "exec", container, "redis-cli", *command], text=True).strip() + + def stop_nodes(): + """Require process termination and closed old listeners before restore.""" + for process in processes: + if process.poll() is None: + process.terminate() + for process in processes: + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + processes.clear() + for port in ports: + with socket.socket() as connection: + connection.settimeout(1) + assert connection.connect_ex(("127.0.0.1", port)) != 0, "Old listener remains open" + + def start_nodes(generation, redis_port): + """Supply the shared generation outside the Redis image, on both JVMs.""" + for port in ports: + argfile = output / f"{generation}-{port}.args" + argfile.write_text(f'-Dcollab.dev.recoveryGeneration={generation}\n-cp\n"' + + classpath.replace("\\", "/") + '"\n' + 'com.kingsrook.qbits.collaboration.devserver.SharedPresenceDevServer\n' + f'{port}\n{redis_port}\n', encoding="utf-8") + logfile = output / f"{generation}-{port}.log" + log = logfile.open("w") + logs.append(log) + processes.append(subprocess.Popen([args.java, "@" + str(argfile)], cwd=repo, + stdout=log, stderr=subprocess.STDOUT, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))) + deadline = time.monotonic() + 60 + for port, process in zip(ports, processes): + logfile = output / f"{generation}-{port}.log" + while f"SHARED_PRESENCE_READY {port}" not in logfile.read_text(errors="replace"): + assert process.poll() is None, f"Node exited; see {output}" + assert time.monotonic() < deadline, f"Startup timed out; see {output}" + time.sleep(0.2) + + def join(): + """Issue each person's lease on a separate node.""" + return {user: require(200, request(port, user, "POST", "/presence/leases", { + "table": "person", "pk": "1"}))["connectionId"] for port, user in zip(ports, ("alice", "bob"))} + + def reject_old(leases, sequence): + """Restored lease authority must fail on both new-generation nodes.""" + for port in ports: + for user, lease in leases.items(): + error = require(409, request(port, user, "POST", "/presence/leases/renew", { + "connectionId": lease, "table": "person", "pk": "1", + "mode": "VIEWING", "sequence": sequence})) + assert error["code"] == "PRESENCE_LEASE_EXPIRED", error + + try: + # DEBUG RELOAD is enabled only on this disposable loopback fixture. + subprocess.run(["docker", "run", "-d", "--name", container, "-p", "127.0.0.1::6379", + "redis:7.4-alpine", "redis-server", "--enable-debug-command", "yes", "--save", ""], + check=True, capture_output=True) + binding = subprocess.check_output(["docker", "port", container, "6379"], text=True).strip() + redis_port = int(binding.rsplit(":", 1)[1]) + start_nodes("before-restore", redis_port) + original = join() + settle(ports, original) + last_renewal = time.monotonic() + for port, (user, lease) in zip(ports, original.items()): + require(200, request(port, user, "POST", "/presence/leases/renew", { + "connectionId": lease, "table": "person", "pk": "1", + "mode": "VIEWING", "sequence": 100})) + assert redis("SAVE") == "OK" + before = int(redis("DBSIZE")) + assert before > 0 + stop_nodes() + assert redis("FLUSHDB") == "OK" + assert redis("DBSIZE") == "0" + assert redis("DEBUG", "RELOAD", "NOSAVE") == "OK" + assert int(redis("DBSIZE")) > 0, "Saved Redis image was not restored" + start_nodes("restored-" + uuid.uuid4().hex, redis_port) + reject_old(original, 1000) + # Avoid a false pass caused only by the old 45-second lease expiring. + assert time.monotonic() - last_renewal < 40, "Restore exceeded the live-lease proof window" + for port in ports: + result = require(503, request(port, "alice", "GET", "/presence?table=person&pk=1")) + assert "recovering" in result["message"], result + replacement = join() + assert all(replacement[user] != original[user] for user in original) + settle(ports, replacement) + reject_old(original, 1001) + print(f"PASS: restored Redis image cannot authorize old leases on either new-generation JVM. Logs: {output}") + finally: + try: + stop_nodes() + finally: + for log in logs: + log.close() + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + + +if __name__ == "__main__": + main() diff --git a/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java b/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java index fbad83a..8530b32 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/devserver/SharedPresenceDevServer.java @@ -65,6 +65,7 @@ public class SharedPresenceDevServer .filter(CollaborationQBitConfig.class::isInstance).map(CollaborationQBitConfig.class::cast) .findFirst().orElseThrow().withRedisPresence(new RedisPresenceConfig().withConnectionName("presence") .withDeployment("acceptance").withApplication("collaboration") + .withRecoveryGeneration(System.getProperty("collab.dev.recoveryGeneration", "initial")) .withScopeProvider(new QCodeReference(SingleTenant.class))); var redis = new RedisRuntimeService(); var transport = new RedisRealtimeRuntimeService(); -- 2.54.0 From 7d987eca2a7712c6b734bf03e2d2fdad58d5b6f3 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 26 Oct 2026 07:37:47 -0500 Subject: [PATCH 5/6] test: cover shared readiness and partial presence loss Co-Authored-By: GPT-6 (Codex) --- docs/PRESENCE-RECOVERY.md | 15 +++-- .../RedisPresenceRecoveryTest.java | 61 +++++++++++++++++++ .../collaboration/RedisPresenceStoreTest.java | 25 ++++---- 3 files changed, 86 insertions(+), 15 deletions(-) diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md index 20c3f0c..e97dada 100644 --- a/docs/PRESENCE-RECOVERY.md +++ b/docs/PRESENCE-RECOVERY.md @@ -52,10 +52,14 @@ version rolling deployment. Existing published v1 snapshots are unchanged. - Before the patch, a two-client disposable Redis test failed because total loss returned an empty snapshot. -- Six focused tests now pass: full loss, loss of volatile projections while the +- Seven focused tests now pass: full loss, loss of volatile projections while the guard survives, restored guard/lease bytes followed by a host generation change, bounded generation validation, and rejection of inconsistent lease - lifetimes across nodes, plus guard loss across previously unvisited records. These use independent clients in + lifetimes across nodes, guard loss across previously unvisited records, and + shared cold readiness followed by normal idle expiry. The latter verifies + that peer reads do not extend the deadline, the guard remains persistent, + expired projections remain known-empty and expired leases cannot renew. + These use independent clients in one JVM, not a deployed HA cluster. - Full wrapper verification passed on October 26: 184 tests, no failures, errors or skips; packaging and coverage gates passed. This includes the @@ -71,8 +75,11 @@ version rolling deployment. Existing published v1 snapshots are unchanged. inaccessible, and a new lease rejoins through the production recovery window. This uses a loopback HTTP server with fixture authentication, not a browser or a deployed host. The focused test and the combined full suite both pass. -- Still required: remaining cold/readiness and partial-loss edge cases, - deployed-host qualification and enforced host maintenance/reset. Local +- The store suite checks loss of either state or expiry data in both record + and table projections. Server-issued lease renewal must fail unavailable + without changing the surviving projection. It also covers complete loss + of either projection pair and stale departure ordering. +- Still required: deployed-host qualification and enforced host maintenance/reset. Local per-operation cost measurements at 1,000 and 10,000 leases are recorded in `acceptance/RECOVERY-COST.md`; sustained production capacity is unqualified. The implementation uses no INFO call or lifecycle polling cache. diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java index 12ef1c6..a276c2d 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java @@ -47,6 +47,67 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; *******************************************************************************/ class RedisPresenceRecoveryTest extends PersonalServiceFixture { + /*************************************************************************** + ** Peer reads cannot prolong cold recovery; normal expiry stays known-empty. + ***************************************************************************/ + @Test + void coldRecoveryDeadlineAndIdleKnowledgeAreShared() throws Exception + { + try(var redis = new GenericContainer<>("redis:7.4-alpine").withExposedPorts(6379)) + { + redis.start(); + var address = new HostAndPort(redis.getHost(), redis.getMappedPort(6379)); + var config = DefaultJedisClientConfig.builder().build(); + try(var first = new RedisConnection(address, config, 2, Duration.ofMillis(500)); + var second = new RedisConnection(address, config, 2, Duration.ofMillis(500))) + { + var namespace = new RedisNamespace("test", "cold-idle", "tenant", "presence"); + var a = new RedisPresenceStore(first.scripts(), namespace, Duration.ofSeconds(2), 10); + var b = new RedisPresenceStore(second.scripts(), namespace, Duration.ofSeconds(2), 10); + assertThatThrownBy(() -> a.snapshot("document", "1")).isInstanceOf(PresenceRecoveringException.class); + String guardKey = first.client().keys("*:recovery").iterator().next(); + String coldGuard = first.client().get(guardKey); + assertThatThrownBy(() -> b.snapshot("document", null)).isInstanceOf(PresenceRecoveringException.class); + assertThat(first.client().get(guardKey)).isEqualTo(coldGuard); + long deadline = System.nanoTime() + Duration.ofSeconds(6).toNanos(); + while(true) + { + try + { + assertThat(b.snapshot("document", "1")).isEmpty(); + break; + } + catch(PresenceRecoveringException recovering) + { + assertThat(System.nanoTime()).isLessThan(deadline); + assertThat(first.client().get(guardKey)).isEqualTo(coldGuard); + Thread.sleep(50); + } + } + assertThat(a.snapshot("document", null)).isEmpty(); + assertThat(first.client().get(guardKey)).isEqualTo(coldGuard); + assertThat(first.client().pttl(guardKey)).isEqualTo(-1L); + + String lease = a.openLease("alice", "Alice", "document", "1"); + awaitVisible(b, lease); + String activeGuard = first.client().get(guardKey); + deadline = System.nanoTime() + Duration.ofSeconds(6).toNanos(); + while(first.client().keys("*").size() > 1) + { + assertThat(System.nanoTime()).isLessThan(deadline); + Thread.sleep(50); + } + assertThat(first.client().keys("*")).containsExactly(guardKey); + assertThat(a.snapshot("document", "1")).isEmpty(); + assertThat(b.snapshot("document", null)).isEmpty(); + assertThat(first.client().get(guardKey)).isEqualTo(activeGuard); + assertThatThrownBy(() -> b.renewLease("alice", "Alice", lease, "document", "1", "VIEWING", null, 100L)) + .isInstanceOf(PresenceLeaseExpiredException.class); + } + } + } + + /*************************************************************************** ** Keep a real lease alive while waiting for bounded initial recovery. ***************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java index 6c774d1..3a1401d 100644 --- a/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceStoreTest.java @@ -169,8 +169,8 @@ class RedisPresenceStoreTest } /*************************************************************************** - ** Characterize the open recovery gap: all projections lost together look new. - ** This is evidence for #1295, not an assertion of the desired final contract. + ** Total loss stays unknown even when the internal legacy API repopulates it. + ** The shared HTTP service requires server-issued leases instead of this API. ***************************************************************************/ @Test void totalLossIsRecoveringEvenForLegacyHeartbeat() throws Exception @@ -353,21 +353,24 @@ class RedisPresenceStoreTest try(var connection = new RedisConnection(new HostAndPort(redis.getHost(), redis.getMappedPort(6379)), DefaultJedisClientConfig.builder().build(), 2, Duration.ofMillis(500))) { - for(String suffix : java.util.List.of("state", "expiry")) + for(String lostPart : java.util.List.of("record:state", "record:expiry", "table:state", "table:expiry")) { - var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", suffix, "presence"), Duration.ofSeconds(2), 10); + String[] parts = lostPart.split(":"); + boolean record = parts[0].equals("record"); + var store = new RedisPresenceStore(connection.scripts(), new RedisNamespace("test", "app", parts[0] + parts[1], "presence"), Duration.ofSeconds(2), 10); awaitReady(store); - assertThat(store.heartbeat("one", "One", "tab", "orders", "1", "VIEWING", null, 1L)).isTrue(); + String lease = store.openLease("one", "One", "orders", "1"); + assertThat(store.renewLease("one", "One", lease, "orders", "1", "VIEWING", null, 1L)).isTrue(); // KEYS is confined to this disposable test server, never the application path. - String recordKey = connection.client().keys("*:record:*:" + suffix).iterator().next(); - String tableKey = recordKey.substring(0, recordKey.indexOf(":record:")) + ":table:state"; - var before = connection.client().hgetAll(tableKey); - connection.client().del(recordKey); + String lostKey = connection.client().keys(record ? "*:record:*:" + parts[1] : "*:table:" + parts[1]).iterator().next(); + String retainedState = connection.client().keys(record ? "*:table:state" : "*:record:*:state").iterator().next(); + var before = connection.client().hgetAll(retainedState); + connection.client().del(lostKey); assertThatThrownBy(() -> store.snapshot("orders", "1")) .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); - assertThatThrownBy(() -> store.heartbeat("one", "One", "tab", "orders", "1", "EDITING", "status", 2L)) + assertThatThrownBy(() -> store.renewLease("one", "One", lease, "orders", "1", "EDITING", "status", 2L)) .isInstanceOf(redis.clients.jedis.exceptions.JedisException.class).hasMessageContaining("incomplete"); - assertThat(connection.client().hgetAll(tableKey)).isEqualTo(before); + assertThat(connection.client().hgetAll(retainedState)).isEqualTo(before); connection.client().flushDB(); } for(String room : java.util.List.of("record", "table")) -- 2.54.0 From 776ebe3fcb6492881bc9d7ec3f6ca6adfa320200 Mon Sep 17 00:00:00 2001 From: James Maes Date: Thu, 29 Oct 2026 09:56:37 -0500 Subject: [PATCH 6/6] test(collaboration): verify packaged presence recovery Co-Authored-By: Codex (GPT-6) --- docs/PRESENCE-RECOVERY.md | 24 ++++++++++ .../acceptance/presence_recovery_processes.py | 32 +++++++++++-- docs/acceptance/presence_restore_processes.py | 45 ++++++++++++++++--- 3 files changed, 90 insertions(+), 11 deletions(-) diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md index e97dada..43035ae 100644 --- a/docs/PRESENCE-RECOVERY.md +++ b/docs/PRESENCE-RECOVERY.md @@ -153,3 +153,27 @@ python docs/acceptance/presence_restore_processes.py --java /path/to/java --repo This is a protocol acceptance fixture, not host maintenance automation. Actual host admission closure, replacement prevention and generation configuration still require implementation and staging validation. + +To qualify a packaged backend, add `--backend-jar /path/to/qbit-collaboration.jar`. +This excludes production `target/classes` from the JVM classpath, rejects a +duplicate collaboration dependency JAR, and records the artifact SHA-256 in +`target/restore-processes-*/result.json`. The receipt includes the elapsed time +for rejection of old leases (required to remain below their normal expiry), +the restore/rejoin checks and fixture cleanup. Cleanup failure fails the run. +The test repository still supplies compiled fixture classes and its dependency +classpath; this is not a standalone production launcher. + +## Packaged browser qualification + +The process harness accepts `--backend-jar` and `--frontend-jar` for artifact +verification. With a backend JAR, production `target/classes` is omitted from the +runtime classpath; test classes provide only the disposable server fixture. The +frontend JAR is served directly instead of copying a local `out` export. +`--frontend-repo` remains required for Playwright tooling. Dependency classpaths +must not contain a second collaboration or frontend artifact. + +Each run writes `result.json` with the selected artifact SHA-256 hashes, browser +profile, result and cleanup outcome. A successful test with failed fixture cleanup +exits unsuccessfully. This records local artifact qualification, not publication +or deployment provenance. Supply `--browser-profile` to qualify each required +browser independently. diff --git a/docs/acceptance/presence_recovery_processes.py b/docs/acceptance/presence_recovery_processes.py index 4e01f66..e406f7a 100644 --- a/docs/acceptance/presence_recovery_processes.py +++ b/docs/acceptance/presence_recovery_processes.py @@ -5,6 +5,7 @@ target/recovery-process-classpath.txt. Uses only disposable loopback fixtures. """ import argparse +import hashlib import json import os from pathlib import Path @@ -76,14 +77,27 @@ def main(): parser.add_argument("--java", required=True) parser.add_argument("--repo", type=Path, required=True) parser.add_argument("--frontend-repo", type=Path) + parser.add_argument("--frontend-jar", type=Path, help="Serve this packaged frontend instead of a local export") + parser.add_argument("--backend-jar", type=Path, help="Use this backend artifact instead of target/classes") parser.add_argument("--browser-profile", choices=("chromium", "firefox", "webkit", "mobile", "tablet"), default="chromium") args = parser.parse_args() repo = args.repo.resolve() output = repo / "target" / ("recovery-processes-" + uuid.uuid4().hex[:10]) output.mkdir() - classpath = os.pathsep.join([str(repo / "target/test-classes"), str(repo / "target/classes"), - (repo / "target/recovery-process-classpath.txt").read_text().strip()]) - if args.frontend_repo: + if args.frontend_jar and not args.frontend_repo: + parser.error("frontend-jar requires frontend-repo for browser tooling") + dependencies = (repo / "target/recovery-process-classpath.txt").read_text().strip().split(os.pathsep) + assert not any(Path(item).name.startswith(("qbit-collaboration-", "qqq-frontend-next-")) for item in dependencies) + backend = args.backend_jar.resolve(strict=True) if args.backend_jar else repo / "target/classes" + classpath = os.pathsep.join([str(repo / "target/test-classes"), str(backend), *dependencies]) + evidence = {"status": "FAIL", "browser_profile": args.browser_profile, "cleanup": False} + if args.backend_jar: + evidence["backend_jar_sha256"] = hashlib.sha256(backend.read_bytes()).hexdigest() + if args.frontend_jar: + frontend = args.frontend_jar.resolve(strict=True) + evidence["frontend_jar_sha256"] = hashlib.sha256(frontend.read_bytes()).hexdigest() + classpath = str(frontend) + os.pathsep + classpath + elif args.frontend_repo: shutil.copytree(args.frontend_repo.resolve() / "out", output / "next-dashboard") classpath = str(output) + os.pathsep + classpath container = "qqq-recovery-" + uuid.uuid4().hex[:12] @@ -112,6 +126,7 @@ def main(): if args.frontend_repo: subprocess.run(["node", str(repo / "docs/acceptance/presence_recovery_browser.mjs"), str(args.frontend_repo.resolve()), str(ports[0]), str(ports[1]), container, str(output), args.browser_profile], check=True) + evidence["status"] = "PASS" return def join(): """Acquire each person's new lease on a different node.""" @@ -130,6 +145,10 @@ def main(): assert all(replacement[user] != original[user] for user in original) settle(ports, replacement) print(f"PASS: two independent JVMs recovered two viewers after Redis loss. Logs: {output}") + evidence["status"] = "PASS" + except Exception as error: + evidence["failure"] = str(error) + raise finally: for process in processes: process.terminate() @@ -141,7 +160,12 @@ def main(): process.wait(timeout=10) for log in logs: log.close() - subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + cleanup = subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + evidence["cleanup"] = cleanup.returncode == 0 and all(process.poll() is not None for process in processes) + (output / "result.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8") + print(output, flush=True) + if evidence["status"] == "PASS" and not evidence["cleanup"]: + raise RuntimeError("Recovery passed but owned fixture cleanup failed") if __name__ == "__main__": diff --git a/docs/acceptance/presence_restore_processes.py b/docs/acceptance/presence_restore_processes.py index 37a2e7c..8b22710 100644 --- a/docs/acceptance/presence_restore_processes.py +++ b/docs/acceptance/presence_restore_processes.py @@ -5,6 +5,8 @@ Requires the same compiled test classpath as presence_recovery_processes.py. """ import argparse +import hashlib +import json import os from pathlib import Path import socket @@ -20,13 +22,21 @@ def main(): parser = argparse.ArgumentParser() parser.add_argument("--java", required=True) parser.add_argument("--repo", type=Path, required=True) + parser.add_argument("--backend-jar", type=Path, help="Verify this artifact without production target/classes") args = parser.parse_args() repo = args.repo.resolve() output = repo / "target" / ("restore-processes-" + uuid.uuid4().hex[:10]) output.mkdir() - classpath = os.pathsep.join([str(repo / "target/test-classes"), str(repo / "target/classes"), - (repo / "target/recovery-process-classpath.txt").read_text().strip()]) + dependencies = (repo / "target/recovery-process-classpath.txt").read_text().strip().split(os.pathsep) + assert all(Path(item).is_file() and item.endswith('.jar') for item in dependencies) + assert not any(Path(item).name.startswith("qbit-collaboration-") for item in dependencies) + backend = args.backend_jar.resolve(strict=True) if args.backend_jar else repo / "target/classes" + classpath = os.pathsep.join([str(repo / "target/test-classes"), str(backend), *dependencies]) + evidence = {"status": "FAIL", "cleanup": False, "checks": []} + if args.backend_jar: + evidence["backend_jar_sha256"] = hashlib.sha256(backend.read_bytes()).hexdigest() container = "qqq-restore-" + uuid.uuid4().hex[:12] + container_created = False processes, logs = [], [] ports = [free_port(), free_port()] @@ -92,6 +102,7 @@ def main(): subprocess.run(["docker", "run", "-d", "--name", container, "-p", "127.0.0.1::6379", "redis:7.4-alpine", "redis-server", "--enable-debug-command", "yes", "--save", ""], check=True, capture_output=True) + container_created = True binding = subprocess.check_output(["docker", "port", container, "6379"], text=True).strip() redis_port = int(binding.rsplit(":", 1)[1]) start_nodes("before-restore", redis_port) @@ -106,14 +117,18 @@ def main(): before = int(redis("DBSIZE")) assert before > 0 stop_nodes() + evidence["checks"].append("old JVMs stopped and listeners closed before restore") assert redis("FLUSHDB") == "OK" assert redis("DBSIZE") == "0" assert redis("DEBUG", "RELOAD", "NOSAVE") == "OK" assert int(redis("DBSIZE")) > 0, "Saved Redis image was not restored" + evidence["checks"].append("nonempty saved Redis image restored") start_nodes("restored-" + uuid.uuid4().hex, redis_port) reject_old(original, 1000) # Avoid a false pass caused only by the old 45-second lease expiring. - assert time.monotonic() - last_renewal < 40, "Restore exceeded the live-lease proof window" + evidence["old_lease_rejection_seconds"] = time.monotonic() - last_renewal + assert evidence["old_lease_rejection_seconds"] < 40, "Restore exceeded the live-lease proof window" + evidence["checks"].append("both nodes reject old leases before normal expiry") for port in ports: result = require(503, request(port, "alice", "GET", "/presence?table=person&pk=1")) assert "recovering" in result["message"], result @@ -121,14 +136,30 @@ def main(): assert all(replacement[user] != original[user] for user in original) settle(ports, replacement) reject_old(original, 1001) + evidence["checks"].append("both nodes recover with fresh viewers and still reject old leases") + if args.backend_jar: + assert hashlib.sha256(backend.read_bytes()).hexdigest() == evidence["backend_jar_sha256"] + evidence["status"] = "PASS" print(f"PASS: restored Redis image cannot authorize old leases on either new-generation JVM. Logs: {output}") + except Exception as error: + evidence["failure"] = str(error) + raise finally: try: - stop_nodes() + try: + stop_nodes() + finally: + for log in logs: + log.close() + if container_created: + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=True) + evidence["cleanup"] = True + except Exception as error: + evidence["status"] = "FAIL" + evidence["cleanup_failure"] = str(error) + raise finally: - for log in logs: - log.close() - subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + (output / "result.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8") if __name__ == "__main__": -- 2.54.0