WIP: Recover presence leases after Redis state loss #47
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
+36
@@ -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");
|
||||
}
|
||||
}
|
||||
@@ -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()));
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<PresenceRegistry.Viewer> snapshot(String table, String pk)
|
||||
{
|
||||
List<String> 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<String, Stored> 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<String> 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<String> 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)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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.
|
||||
***************************************************************************/
|
||||
|
||||
@@ -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<String, Object> 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<String, Object>();
|
||||
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<String, Object> 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"));
|
||||
}
|
||||
}
|
||||
@@ -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<String, byte[]>();
|
||||
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<String>();
|
||||
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");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
***************************************************************************/
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
+2
-1
@@ -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);
|
||||
|
||||
Reference in new issue
Block a user