"""Exercise public snapshot APIs with active writers, quotas, a killed server and a live CLI.""" from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager import json import os from pathlib import Path import secrets import subprocess import sys import tempfile import threading import time from service_fixture import cli, Client, RatDBError EXE = Path(sys.argv[1]).resolve() PASSWORD = secrets.token_urlsafe(24) @contextmanager def serve(path, *limits): proc = subprocess.Popen([EXE, "serve", path, "--bind", "127.0.0.1:0", *limits], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: ready = proc.stdout.readline() assert "listening on" in ready, ready port = int(ready.split("listening on ")[1].split()[0].rsplit(":", 1)[1]) def connect(): c = Client("admin", PASSWORD, port=port) c.select_warren("crm") c.select_den("contacts") return c yield proc, port, connect finally: if proc.poll() is None: proc.kill() proc.communicate(timeout=20) def rejects(code, call, *args): try: call(*args) except RatDBError as error: assert error.code == code, (error.code, str(error)) else: raise AssertionError(f"expected {code}") def shell_roundtrip(port): with subprocess.Popen([EXE, "batch", f"127.0.0.1:{port}"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={**os.environ, "RATDB_PASSWORD": PASSWORD}) as child: def command(line): child.stdin.write(line + "\n") child.stdin.flush() reply=json.loads(child.stdout.readline()) assert reply["ok"],reply return reply["data"] try: command("use crm/contacts") h=command("snapshot begin team=old")["snapshot"] first=command(f"snapshot page {h} 1") second=command(f"snapshot page {h} 1 {first['next_cursor']}") assert first["documents"][0]["id"] < second["documents"][0]["id"] assert command("snapshot end "+h)["closed"] command("quit") finally: child.stdin.close() if child.poll() is None: child.wait(timeout=10) def churn_run(path): n=1000 original=[{"id":f"{i:04}", "team":"old", "revision":0} for i in range(n)] with serve(path) as (proc,port,connect), connect() as setup: for off in range(0,n,100): setup.transaction(setup._request("stats")["generation"], [ {"op":"put","warren":"crm","den":"contacts","id":v["id"],"document":v} for v in original[off:off+100]]) setup.create_index("team") shell_roundtrip(port) with connect() as changed_filter: h=changed_filter.begin_snapshot({"field":"team","value":"old"})["snapshot"] try: changed_filter._request("snapshot_page", snapshot=h, cursor=None, limit=1, filter={"field":"team","value":"new"}) except RatDBError as e: assert e.code=="bad_frame" else: raise AssertionError("accepted filter replacement on bound cursor") with connect() as r1, connect() as r2: v1=r1.begin_snapshot() v2=r2.begin_snapshot({"field":"team","value":"old"}) assert v1["generation"]==v2["generation"] p1=r1.snapshot_page(v1["snapshot"],20) p2=r2.snapshot_page(v2["snapshot"],20) rejects("snapshot_invalid",r2.snapshot_page,v1["snapshot"],20,p1["next_cursor"]) rejects("snapshot_invalid",r1.snapshot_page,v1["snapshot"],20,p2["next_cursor"]) rejects("snapshot_invalid",r1.snapshot_page,v1["snapshot"],20,None) r1.select_den("other") rejects("snapshot_invalid",r1.snapshot_page,v1["snapshot"],20,p1["next_cursor"]) r1.select_den("contacts") # At least one complete commit from BOTH writers precedes each next pair of pages. gate=threading.Barrier(3,timeout=30) def writer(side): elapsed=0 with connect() as c: for k in range(49): gate.wait() start=time.perf_counter_ns() i=20*(k+1)+side*5 c.update(f"{i:04}",{"team":"new","revision":k+1}) c.delete(f"{i+1:04}") c.put(f"z{side}-{k}",{"team":"old","revision":99}) elapsed+=time.perf_counter_ns()-start gate.wait() return elapsed values1=p1["documents"][:] values2=p2["documents"][:] reading=0 with ThreadPoolExecutor(max_workers=2) as pool: tasks=[pool.submit(writer,side) for side in (0,1)] for k in range(49): gate.wait() gate.wait() start=time.perf_counter_ns() p1=r1.snapshot_page(v1["snapshot"],20,p1["next_cursor"]) p2=r2.snapshot_page(v2["snapshot"],20,p2["next_cursor"]) reading+=time.perf_counter_ns()-start assert p1["generation"]==v1["generation"]==p2["generation"] values1.extend(p1["documents"]) values2.extend(p2["documents"]) writing=sum(job.result() for job in tasks) assert values1==original==values2 assert p1["next_cursor"] is None and p2["next_cursor"] is None assert setup.get("0020")["team"]=="new" and setup.get("0021") is None print(f"File-backed TCP: 294 acknowledged writes, 2 complete historical traversals; " f"mean write {writing/294/1e6:.3f} ms, mean 20-row page {reading/98/1e6:.3f} ms") r1.end_snapshot(v1["snapshot"]) # Kill while another thread repeatedly issues historical pages. Synchronize # on a successful page to prove the historical session is active. live=r1.begin_snapshot() ack=threading.Event() def reading_until_exit(): cursor=None try: while True: page=r1.snapshot_page(live["snapshot"],1,cursor) cursor=page["next_cursor"] ack.set() if cursor is None: return except (ConnectionError,OSError): pass with ThreadPoolExecutor(max_workers=1) as pool: job=pool.submit(reading_until_exit) assert ack.wait(10) proc.kill() job.result(timeout=20) with serve(path) as (_,_,connect), connect() as db: rejects("snapshot_invalid",db.snapshot_page,live["snapshot"],1,None) assert db._request("check")["valid"] assert db.get("0985")["team"]=="new" and db.get("0986") is None assert db.get("z1-48")["revision"]==99 def limits_run(path): with serve(path,"--snapshot-count","1","--snapshot-ttl-ms","300", "--snapshot-page-size","3","--snapshot-examined","2", "--snapshot-response-bytes","512") as (_,_,connect), connect() as r, connect() as w: h=r.begin_snapshot({"field":"no-match","value":7})["snapshot"] rejects("snapshot_limit",w.begin_snapshot) rejects("operation_failed",r.snapshot_page,h,4) p=r.snapshot_page(h,3) assert p["documents"]==[] and p["examined"]==2 and p["next_cursor"] time.sleep(.45) # New admission succeeds even though the idle first client is still connected. second=w.begin_snapshot()["snapshot"] rejects("snapshot_expired",r.snapshot_page,h,1,p["next_cursor"]) w.end_snapshot(second) empty=r.begin_snapshot({"field":"id","value":"absent"})["snapshot"] assert r.snapshot_page(empty,1)["next_cursor"] is None again=w.begin_snapshot()["snapshot"] w.end_snapshot(again) r.end_snapshot(empty) for _ in range(12): h=r.begin_snapshot()["snapshot"] r.end_snapshot(h) h=w.begin_snapshot()["snapshot"] w.close() deadline=time.monotonic()+5 while True: try: again=r.begin_snapshot() break except RatDBError as e: assert e.code=="snapshot_limit" and time.monotonic()12288 for stage in stages): child.kill() break child.communicate(timeout=10) assert child.returncode and not target.exists() assert path.read_bytes()==before cli(EXE,"migrate-indexed",path,target) cli(EXE,"check",target) with tempfile.TemporaryDirectory() as tmp: path=Path(tmp)/"snapshot.ratdb" cli(EXE,"init","--indexed",path) cli(EXE,"setup",path,"admin",password=PASSWORD) churn_run(path) limits_run(path) interrupted_compaction(path) print("Snapshot Python and shell lifecycles, quotas, concurrent cursors and crash recovery passed")