diff --git a/docs/PRESENCE-RECOVERY.md b/docs/PRESENCE-RECOVERY.md new file mode 100644 index 0000000..43035ae --- /dev/null +++ b/docs/PRESENCE-RECOVERY.md @@ -0,0 +1,179 @@ +# Presence recovery candidate — QQQ #1300 + +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 + +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. +- 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, 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 + 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. +- 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. +- 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. + +## 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 +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. + +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. + +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/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 new file mode 100644 index 0000000..5171454 --- /dev/null +++ b/docs/acceptance/presence_recovery_browser.mjs @@ -0,0 +1,52 @@ +// 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, profile = 'chromium'] = process.argv.slice(2) +const require = createRequire(path.join(frontend, 'package.json')) +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(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) + } + 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: ${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`) }) + } + await browser.close() +} diff --git a/docs/acceptance/presence_recovery_processes.py b/docs/acceptance/presence_recovery_processes.py new file mode 100644 index 0000000..e406f7a --- /dev/null +++ b/docs/acceptance/presence_recovery_processes.py @@ -0,0 +1,172 @@ +"""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 hashlib +import json +import os +from pathlib import Path +import socket +import shutil +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) + 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() + 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] + 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('-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") + 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) + 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.""" + 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}") + evidence["status"] = "PASS" + except Exception as error: + evidence["failure"] = str(error) + raise + 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() + 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__": + main() diff --git a/docs/acceptance/presence_restore_processes.py b/docs/acceptance/presence_restore_processes.py new file mode 100644 index 0000000..8b22710 --- /dev/null +++ b/docs/acceptance/presence_restore_processes.py @@ -0,0 +1,166 @@ +"""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 hashlib +import json +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) + 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() + 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()] + + 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) + 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) + 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() + 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. + 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 + replacement = join() + 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: + 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: + (output / "result.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8") + + +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..9eaeca9 --- /dev/null +++ b/src/main/resources/presence/redis-recovery.lua @@ -0,0 +1,43 @@ +-- 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 + 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 + 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 + 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. + -- 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 + 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 + 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/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 new file mode 100644 index 0000000..a276c2d --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/collaboration/RedisPresenceRecoveryTest.java @@ -0,0 +1,325 @@ +/* + * 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.PresenceRecoveringException; +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 +{ + /*************************************************************************** + ** 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. + ***************************************************************************/ + 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(); + } + } + } + + + /*************************************************************************** + ** 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. + ***************************************************************************/ + @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..3a1401d 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(); @@ -128,11 +169,11 @@ 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 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(); @@ -303,33 +353,49 @@ 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(45), 10); - assertThat(store.heartbeat("one", "One", "tab", "orders", "1", "VIEWING", null, 1L)).isTrue(); + 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); + 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")) { - 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(); } 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..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,12 +65,13 @@ 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(); 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);