Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741# /// script# dependencies = ["websockets"]# ///"""Strict power-loss oracle for the production Linux Stream binary.
The process uses RocksDB and ext4 on a kernel NBD device. The nbdkit backendkeeps acknowledged writes in a volatile image and copies dirty extents to thedurable image only for FLUSH/FUA. At a canonical upstream lifecycle boundarythis driver kills nbdkit first, then Stream, reconstructs the device solelyfrom the durable image, repairs ext4's journal, and requires a clean restart.
The runtime target is offline: it refuses image pulls and expects the pinnedsimulator on the host. Build the one-time tool image with`just powerloss-image`, then run `just powerloss-oracle` while `just simulator`is active."""
import asyncioimport datetimeimport jsonimport osimport pathlibimport shutilimport socketimport subprocessimport tempfileimport timeimport urllib.errorimport urllib.parseimport urllib.request
import websockets
ROOT = pathlib.Path(__file__).resolve().parents[1]IMAGE = os.environ.get("STREAM_POWERLOSS_IMAGE", "stream-powerloss-oracle:ubuntu24.04")BIN = pathlib.Path(os.environ.get("STREAM_POWERLOSS_BIN", "/tmp/stream-powerloss-linux/bin/stream")).resolve()FIXTURE = pathlib.Path(os.environ.get("STREAM_POWERLOSS_FIXTURE", "/tmp/stream-powerloss-fixture.jss")).resolve()IMPORT_CSV = ROOT / "tests" / "timestamp-import.csv"DEVICE = os.environ.get("STREAM_POWERLOSS_DEVICE", "/dev/nbd3")PORT = int(os.environ.get("STREAM_POWERLOSS_PORT", "16099"))UPSTREAM_HOST = os.environ.get("STREAM_POWERLOSS_UPSTREAM_HOST", "host.docker.internal")CONTAINER = f"stream-powerloss-oracle-{os.getpid()}"SIZE = 768 * 1024 * 1024BASE = f"http://127.0.0.1:{PORT}"
CASES = [ ("after-repo-complete", 1), ("after-repo-complete", 2), # The durability-ordering seam itself: rows fsynced, completion not yet # acknowledged. Recovery must lose nothing and simply re-complete the repo. ("after-archive-durable-before-completion-commit", 1), ("after-merge-dst-flush-before-source-commit", 1), ("after-merge-dst-seal-before-discovery", 1), ("after-merge-discovery-before-cleanup", 1), ("after-merge-cleanup-complete", 1), ("after-bootstrap-live-close-before-seal", 1), ("after-steady-phase-before-steady-run", 1),]
COMPACTION_CASES = [ "after-segment-rewrite-temp-written", "after-segment-rewrite-temp-synced", "after-segment-rewrite-renamed", "after-segment-rewrite-dir-synced", "after-compaction-rewrite-before-watermark", "after-compaction-chunk-watermark",]
PATCH_CASES = [ "after-segment-patch-temp-written", "after-segment-patch-temp-synced", "after-segment-patch-renamed", "after-segment-patch-dir-synced",]
FIXTURE_EVENTS = 2500COMPACTED_EVENTS = 1507IMPORTED_TIMESTAMP_US = 1_704_164_645_000_000IMPORTED_TIMESTAMP_RFC3339 = "2024-01-02T03:04:05.000000Z"
def run(args, *, check=True, capture=True, timeout=60): result = subprocess.run( [str(a) for a in args], text=True, stdout=subprocess.PIPE if capture else None, stderr=subprocess.STDOUT if capture else None, timeout=timeout, ) if check and result.returncode != 0: raise AssertionError(f"command failed ({result.returncode}): {' '.join(map(str, args))}\n{result.stdout}") return result
def dexec(*args, check=True, capture=True, timeout=60): return run(["docker", "exec", CONTAINER, *args], check=check, capture=capture, timeout=timeout)
def dexec_sh(script, *, check=True, timeout=60): return dexec("sh", "-lc", script, check=check, timeout=timeout)
def read_state(path, *, errors=None): """Read a file the container wrote into the bind-mounted /state.
The container runs privileged as root, so on Linux those files really are root-owned and an unprivileged host user cannot read them. Docker Desktop hides this by remapping ownership to the invoking user, which is why this only ever failed off a Mac (2026-08-10: PermissionError on /state/receipt). Fall back to reading through the container, which owns them. """ try: return path.read_text(errors=errors) if errors else path.read_text() except PermissionError: result = dexec("cat", f"/state/{path.name}", check=False) if result.returncode != 0: raise return result.stdout
def docker_image_ready(): result = run(["docker", "image", "inspect", IMAGE], check=False) if result.returncode != 0: raise SystemExit(f"strict power-loss image missing: {IMAGE}; run `just powerloss-image` once while online")
def require_simulator(): result = dexec( "curl", "-fsS", "--max-time", "3", f"http://{UPSTREAM_HOST}:7777/xrpc/com.atproto.sync.listRepos?limit=1", check=False, ) if result.returncode != 0: raise SystemExit("pinned simulator is not reachable; start it with `just simulator`") body = json.loads(result.stdout) assert body.get("repos"), "simulator returned no repositories"
def wait_log(path, marker, timeout=300): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if path.exists(): text = read_state(path, errors="replace") if marker in text: return text alive = dexec_sh("test -f /state/stream.pid && kill -0 $(cat /state/stream.pid)", check=False) if alive.returncode != 0: raise AssertionError(f"Stream exited before {marker!r}\n{read_state(path, errors="replace") if path.exists() else ""}") time.sleep(0.1) raise AssertionError(f"timed out waiting for {marker!r}\n{read_state(path, errors="replace") if path.exists() else ""}")
def http_json(path): try: with urllib.request.urlopen(BASE + path, timeout=3) as response: return response.status, json.loads(response.read()) except urllib.error.HTTPError as error: return error.code, None except OSError: return None, None
def import_json(path): request = urllib.request.Request( BASE + path, headers={"Authorization": "Bearer powerloss-test-token"}, ) try: with urllib.request.urlopen(request, timeout=3) as response: return response.status, json.loads(response.read()) except urllib.error.HTTPError as error: return error.code, None except OSError: return None, None
def http_status(path): try: with urllib.request.urlopen(BASE + path, timeout=3) as response: response.read() return response.status except urllib.error.HTTPError as error: return error.code except OSError: return None
def http_text(path): try: with urllib.request.urlopen(BASE + path, timeout=3) as response: return response.status, response.read().decode() except urllib.error.HTTPError as error: return error.code, error.read().decode(errors="replace") except OSError: return None, ""
def post_import(): request = urllib.request.Request( BASE + "/xrpc/network.bsky.jetstream.importTimestamps", data=json.dumps({"path": "job.csv"}).encode(), headers={ "Authorization": "Bearer powerloss-test-token", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(request, timeout=5) as response: body = json.loads(response.read()) assert response.status == 200 and body.get("job"), body return body["job"]
def metric_value(name): status, text = http_text("/metrics") if status != 200: return None prefix = name + " " for line in text.splitlines(): if line.startswith(prefix): return int(float(line[len(prefix):])) return None
def wait_serving(log_path, timeout=300): deadline = time.monotonic() + timeout while time.monotonic() < deadline: status, body = http_json("/xrpc/network.bsky.jetstream.listSegments") if status == 200: return body alive = dexec_sh("test -f /state/stream.pid && kill -0 $(cat /state/stream.pid)", check=False) if alive.returncode != 0: raise AssertionError(f"recovery exited before serving\n{read_state(log_path, errors="replace")}") time.sleep(0.2) raise AssertionError(f"recovery did not serve\n{read_state(log_path, errors="replace")}")
async def replay(count, *, require_unique=False): previous = 0 fingerprints = {} url = f"ws://127.0.0.1:{PORT}/xrpc/network.bsky.jetstream.subscribeEvents?cursor=0" async with websockets.connect(url) as websocket: for _ in range(count): event = json.loads(await asyncio.wait_for(websocket.recv(), timeout=15))["payload"] seq = event.get("seq") assert isinstance(seq, int) and seq > previous, (previous, event) assert event.get("did", "").startswith("did:"), event if require_unique: semantic = dict(event) semantic.pop("seq", None) semantic.pop("cursor", None) fingerprint = json.dumps(semantic, sort_keys=True, separators=(",", ":")) assert fingerprint not in fingerprints, ( f"duplicate recovered event at seq {seq}; " f"first local seq {fingerprints.get(fingerprint)}; event={semantic}" ) fingerprints[fingerprint] = seq previous = seq
def start_backend_detached(): run([ "docker", "exec", "-d", CONTAINER, "sh", "-lc", "exec env STRICT_NBD_DURABLE=/state/durable.img " "STRICT_NBD_LIVE=/state/live.img " f"STRICT_NBD_SIZE={SIZE} " "STRICT_NBD_RECEIPT=/state/receipt " "nbdkit -f -i 127.0.0.1 -p 10809 --tls=off python /oracle/strict_nbd.py " ">/state/nbdkit.log 2>&1", ]) for _ in range(100): if dexec_sh("pgrep -r RSD -x nbdkit >/dev/null", check=False).returncode == 0: return time.sleep(0.05) raise AssertionError("nbdkit did not start")
def connect_device(): dexec_sh(f"nbd-client -d {DEVICE} >/dev/null 2>&1 || true") for _ in range(100): attached = dexec( "nbd-client", "127.0.0.1", "10809", DEVICE, "-b", "4096", check=False, ) if attached.returncode == 0 and dexec("nbd-client", "-c", DEVICE, check=False).returncode == 0: return time.sleep(0.05) detail = dexec_sh("dmesg | tail -20", check=False).stdout raise AssertionError(f"failed to attach {DEVICE}\n{detail}")
def detach_and_unmount(): """Drop the nbd device first, then the mount.
Order is load-bearing. Once nbdkit is dead the kernel device has no server, so any I/O to it — umount's included — parks in uninterruptible sleep, where the inner `timeout` cannot reach it and even `docker exec` never returns (a full gate run died this way, 2026-08-09/10). Detaching first makes that pending I/O fail with EIO instead of hanging. Every call site bounds its own exec too: D-state is not interruptible, so the only real protection is not entering it. """ dexec_sh(f"timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False, timeout=20) dexec_sh("timeout 5 umount -l /mnt/strict 2>/dev/null || true", check=False, timeout=20) dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False, timeout=20)
def stop_device(*, graceful=False): if graceful: dexec_sh("sync", check=False) dexec_sh("pkill -9 -x nbdkit 2>/dev/null || true", check=False, timeout=20) detach_and_unmount()
def calibrate(state): dexec_sh("rm -rf /state/*") start_backend_detached() connect_device() dexec_sh( f"python3 - <<'PY'\n" "import os\n" f"fd=os.open('{DEVICE}', os.O_RDWR)\n" "os.pwrite(fd, b'OLD'+bytes(4093), 16*1024*1024)\n" "os.fsync(fd)\n" "os.close(fd)\n" "open('/tmp/new.block','wb').write(b'NEW'+bytes(4093))\n" "PY\n" f"before=$(wc -l </state/receipt); dd if=/tmp/new.block of={DEVICE} bs=4096 seek=4096 count=1 oflag=direct conv=notrunc status=none; " "after=$(wc -l </state/receipt); test \"$before\" = \"$after\"; " f"test \"$(dd if={DEVICE} bs=4096 skip=4096 count=1 iflag=direct status=none | head -c3)\" = NEW" ) dexec_sh("pkill -9 -x nbdkit; sleep .2", check=False) dexec_sh(f"timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False) dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False) start_backend_detached() connect_device() recovered = dexec_sh( f"dd if={DEVICE} bs=4096 skip=4096 count=1 iflag=direct status=none | head -c3" ).stdout assert recovered == "OLD", f"strict cache calibration recovered {recovered!r}, not OLD" assert len(read_state(state / "receipt").splitlines()) == 1 stop_device() print("powerloss: calibrated durable FLUSH vs discarded volatile write")
def stream_args(point=None, ordinal=1): args = ( "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " f"--relay-url=http://{UPSTREAM_HOST}:7777 " f"--plc-url=http://{UPSTREAM_HOST}:7777 " "--max-backfill-repos=25 --compaction-interval=0 --retry-interval=0 --no-verify" ) if point: args += f" --power-cutpoint={point}:{ordinal}" return args
def start_stream(log_name, point=None, ordinal=1): command = f"echo $$ >/state/stream.pid; exec {stream_args(point, ordinal)} >/state/{log_name} 2>&1" run(["docker", "exec", "-d", CONTAINER, "sh", "-lc", command])
def mutation_stream_args(kind, point=None): interval = "1s" if kind == "compaction" else "0" args = ( "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " "--relay-url=http://127.0.0.1:1 " "--plc-url=http://127.0.0.1:1 --retry-interval=0 --no-verify " f"--compaction-interval={interval} " "--timestamp-import-dir=/mnt/strict/imports " "--timestamp-import-token=powerloss-test-token" ) if point: args += f" --power-cutpoint={point}:1" return args
def start_mutation_stream(log_name, kind, point=None): command = ( f"echo $$ >/state/stream.pid; exec {mutation_stream_args(kind, point)} " f">/state/{log_name} 2>&1" ) run(["docker", "exec", "-d", CONTAINER, "sh", "-lc", command])
def prepare_mutation_filesystem(): dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") start_backend_detached() connect_device() dexec_sh( f"mkfs.ext4 -F {DEVICE} >/state/mkfs.log 2>&1; " f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict; " "mkdir -p /mnt/strict/data/segments /mnt/strict/imports; " "cp /oracle/fixture.jss /mnt/strict/data/segments/seg_0000000000.jss; " "cp /oracle/timestamp-import.csv /mnt/strict/imports/job.csv; sync" )
def cut_power_and_recover(): dexec_sh("pkill -9 -x nbdkit; kill -9 $(cat /state/stream.pid)", check=False) detach_and_unmount() start_backend_detached() connect_device() fsck = dexec_sh(f"e2fsck -fy {DEVICE}", check=False, timeout=120) assert fsck.returncode in (0, 1), f"e2fsck failed after mutation cut:\n{fsck.stdout}" dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict")
def stop_mutation_case(): dexec_sh("kill -TERM $(cat /state/stream.pid) 2>/dev/null || true", check=False) for _ in range(100): if dexec_sh("kill -0 $(cat /state/stream.pid) 2>/dev/null", check=False).returncode != 0: break time.sleep(0.05) dexec_sh("kill -9 $(cat /state/stream.pid) 2>/dev/null || true", check=False) stop_device(graceful=True)
def sealed_archive_identity(): encoded = dexec_sh( "python3 - <<'PY'\n" "import glob, hashlib, json\n" "total = 0\n" "digest = hashlib.sha256()\n" "sealed = 0\n" "for path in sorted(glob.glob('/mnt/strict/data/segments/*.jss')):\n" " data = open(path, 'rb').read()\n" " if len(data) < 22 or data[:4] != b'jss0' or int.from_bytes(data[4:12], 'little') == 0:\n" " continue\n" " sealed += 1\n" " total += int.from_bytes(data[18:22], 'little')\n" " digest.update(path.rsplit('/', 1)[-1].encode() + b'\\0')\n" " digest.update(data)\n" "print(json.dumps({'segments': sealed, 'events': total, 'sha256': digest.hexdigest()}))\n" "PY" ).stdout return json.loads(encoded)
def wait_compacted(log_path, timeout=120): deadline = time.monotonic() + timeout while time.monotonic() < deadline: listing = wait_serving(log_path, timeout=5) events = sum(item.get("eventCount", 0) for item in listing.get("segments", [])) if metric_value("jetstream_compaction_watermark_seq") == FIXTURE_EVENTS and events == COMPACTED_EVENTS: return listing time.sleep(0.2) raise AssertionError(f"compaction did not converge\n{read_state(log_path, errors="replace")}")
def run_compaction_powerloss_case(state, point): prepare_mutation_filesystem() start_mutation_stream("mutation-first.log", "compaction", point) marker = f"power cutpoint {point} ordinal 1: ready" first_log = wait_log(state / "mutation-first.log", marker, timeout=120) assert first_log.count(marker) == 1 assert len(read_state(state / "receipt").splitlines()) > 0 cut_power_and_recover() start_mutation_stream("mutation-recovery.log", "compaction") wait_compacted(state / "mutation-recovery.log") dexec_sh("test ! -e /mnt/strict/data/segments/seg_0000000000.jss.tmp") asyncio.run(replay(COMPACTED_EVENTS, require_unique=True)) print(f"powerloss: {point} converged to {COMPACTED_EVENTS} compacted events") stop_mutation_case()
async def verify_timestamp_import(): targeted = 0 untouched = 0 url = f"ws://127.0.0.1:{PORT}/xrpc/network.bsky.jetstream.subscribeEvents?cursor=0" async with websockets.connect(url) as websocket: for _ in range(FIXTURE_EVENTS): event = json.loads(await asyncio.wait_for(websocket.recv(), timeout=15))["payload"] materialization = event["$type"].endswith("#commit") and event.get("operation") != "delete" if materialization and event.get("did") == "did:plc:streamsampleaccount" and event.get("collection") == "app.bsky.feed.post": # the imported display timestamp surfaces as the RFC 3339 `time` assert event.get("time") == IMPORTED_TIMESTAMP_RFC3339, event targeted += 1 elif materialization: expected_us = 1_700_000_000_000_000 + event["seq"] * 1000 expected = datetime.datetime.fromtimestamp( expected_us / 1e6, datetime.UTC ).strftime("%Y-%m-%dT%H:%M:%S.%fZ") assert event.get("time") == expected, event untouched += 1 assert targeted > 0 and untouched > 0, (targeted, untouched)
def wait_import_complete(log_path, timeout=120): deadline = time.monotonic() + timeout while time.monotonic() < deadline: status, body = import_json("/xrpc/network.bsky.jetstream.getImportStatus") if status == 200 and body.get("state") == "complete": return body alive = dexec_sh("kill -0 $(cat /state/stream.pid)", check=False) if alive.returncode != 0: raise AssertionError(f"import recovery exited\n{read_state(log_path, errors="replace")}") time.sleep(0.2) raise AssertionError(f"timestamp import did not complete\n{read_state(log_path, errors="replace")}")
def wait_patch_marker(log_path, marker, job, timeout=120): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if log_path.exists(): text = read_state(log_path, errors="replace") if marker in text: return text status, body = import_json( "/xrpc/network.bsky.jetstream.getImportStatus?job=" + urllib.parse.quote(job) ) if status == 200 and body.get("state") == "failed": raise AssertionError(f"timestamp import failed before {marker}: {body}") alive = dexec_sh("kill -0 $(cat /state/stream.pid)", check=False) if alive.returncode != 0: raise AssertionError(f"Stream exited before {marker}\n{read_state(log_path, errors="replace")}") time.sleep(0.1) raise AssertionError(f"timed out waiting for {marker}\n{read_state(log_path, errors="replace")}")
def run_patch_powerloss_case(state, point): prepare_mutation_filesystem() start_mutation_stream("patch-first.log", "patch", point) wait_serving(state / "patch-first.log") job = post_import() marker = f"power cutpoint {point} ordinal 1: ready" first_log = wait_patch_marker(state / "patch-first.log", marker, job, timeout=120) assert first_log.count(marker) == 1 cut_power_and_recover() start_mutation_stream("patch-recovery.log", "patch") wait_serving(state / "patch-recovery.log") result = wait_import_complete(state / "patch-recovery.log") assert result.get("segmentsApplied", 0) > 0, result dexec_sh("test ! -e /mnt/strict/data/segments/seg_0000000000.jss.tmp") asyncio.run(verify_timestamp_import()) print(f"powerloss: {point} timestamp import recovered") stop_mutation_case()
def run_case(state, point, ordinal): label = f"{point}-ordinal-{ordinal}" dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") start_backend_detached() connect_device() dexec_sh( f"mkfs.ext4 -F {DEVICE} >/state/mkfs.log 2>&1; " f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict; mkdir /mnt/strict/data" ) start_stream("first.log", point, ordinal) marker = f"power cutpoint {point} ordinal {ordinal}: ready" first_log = wait_log(state / "first.log", marker) assert first_log.count(marker) == 1, "cutpoint receipt was not unique" if point == "after-repo-complete" and ordinal == 1: # The first canonical seam holds the bootstrap process in place, so # this is a deterministic process-level proof that the listener never # exposes archive/subscribe readiness before steady state. Repeating # the request catches a transient open-then-close gate. assert http_status("/healthz") == 200 for _ in range(3): assert http_status("/xrpc/network.bsky.jetstream.listSegments") == 503 flushes_before = len(read_state(state / "receipt").splitlines()) assert flushes_before > 0, "filesystem never emitted a durability flush"
# Ordering is the power cut: storage disappears before the process can run # a defer, close RocksDB, or flush a segment. dexec_sh("pkill -9 -x nbdkit; kill -9 $(cat /state/stream.pid)", check=False) detach_and_unmount()
start_backend_detached() connect_device() fsck = dexec_sh(f"e2fsck -fy {DEVICE}", check=False, timeout=120) assert fsck.returncode in (0, 1), f"e2fsck failed after {label}:\n{fsck.stdout}" dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict") dexec_sh("test -d /mnt/strict/data/meta.rocksdb && test -d /mnt/strict/data/segments") start_stream("recovery.log") listing = wait_serving(state / "recovery.log") segments = listing.get("segments", []) events = sum(segment.get("eventCount", 0) for segment in segments) assert segments and events > 0, listing assert http_status("/healthz") == 200 asyncio.run(replay(min(events, 100))) dexec_sh( "test ! -e /mnt/strict/data/backfill; " "test ! -e /mnt/strict/data/phase; test ! -e /mnt/strict/data/repos.log" ) recovery_log = read_state(state / "recovery.log", errors="replace") assert "OutOfMemory" not in recovery_log assert "fatal event handling failure" not in recovery_log print(f"powerloss: {label} recovered ({events} events, {flushes_before} pre-cut flushes)")
dexec_sh("kill -TERM $(cat /state/stream.pid) 2>/dev/null || true", check=False) for _ in range(100): if dexec_sh("kill -0 $(cat /state/stream.pid) 2>/dev/null", check=False).returncode != 0: break time.sleep(0.05) dexec_sh("kill -9 $(cat /state/stream.pid) 2>/dev/null || true", check=False) stop_device(graceful=True)
def run_restart_after_cleanup_guard_case(state): """Crash before dir fsync, restart on cached state, then lose power.
This is upstream's two-crash guard-path schedule. The second process sees the live tree as absent, but must fsync the data directory again before it durably deletes the merge cursors. Otherwise the subsequent power loss can resurrect the source tree with cursor zero and duplicate survivors. """ label = "restart-after-cleanup-guard" dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") start_backend_detached() connect_device() dexec_sh( f"mkfs.ext4 -F {DEVICE} >/state/mkfs.log 2>&1; " f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict; mkdir /mnt/strict/data" )
local_point = "after_cleanup_remove_before_sync" start_stream("guard-first.log", local_point) wait_log(state / "guard-first.log", f"power cutpoint {local_point} ordinal 1: ready") dexec_sh("test ! -e /mnt/strict/data/backfill") before = sealed_archive_identity() before_count = before["events"] assert before_count > 0, "guard fixture had no sealed destination rows" flushes_before_guard = len(read_state(state / "receipt").splitlines()) dexec_sh("kill -9 $(cat /state/stream.pid)", check=False)
canonical_point = "after-merge-cleanup-complete" start_stream("guard-second.log", canonical_point) marker = f"power cutpoint {canonical_point} ordinal 1: ready" second_log = wait_log(state / "guard-second.log", marker) assert second_log.count(marker) == 1 flushes_after_guard = len(read_state(state / "receipt").splitlines()) assert flushes_after_guard > flushes_before_guard, \ "restart-after-cleanup guard emitted no durability flush"
dexec_sh("pkill -9 -x nbdkit; kill -9 $(cat /state/stream.pid)", check=False) detach_and_unmount()
start_backend_detached() connect_device() fsck = dexec_sh(f"e2fsck -fy {DEVICE}", check=False, timeout=120) assert fsck.returncode in (0, 1), f"e2fsck failed after {label}:\n{fsck.stdout}" dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict") start_stream("guard-recovery.log") listing = wait_serving(state / "guard-recovery.log") segments = listing.get("segments", []) events = sum(segment.get("eventCount", 0) for segment in segments) assert segments and events > 0, listing assert events == before_count, \ f"guard recovery changed sealed event count: before={before_count} after={events}" after = sealed_archive_identity() assert after == before, f"guard recovery changed sealed archive bytes: before={before} after={after}" asyncio.run(replay(events)) dexec_sh( "test ! -e /mnt/strict/data/backfill; " "test ! -e /mnt/strict/data/phase; test ! -e /mnt/strict/data/repos.log" ) print( f"powerloss: {label} recovered " f"({events} events, {flushes_after_guard - flushes_before_guard} guard flushes)" )
dexec_sh("kill -TERM $(cat /state/stream.pid) 2>/dev/null || true", check=False) for _ in range(100): if dexec_sh("kill -0 $(cat /state/stream.pid) 2>/dev/null", check=False).returncode != 0: break time.sleep(0.05) dexec_sh("kill -9 $(cat /state/stream.pid) 2>/dev/null || true", check=False) stop_device(graceful=True)
def main(): docker_image_ready() if not BIN.is_file(): raise SystemExit(f"Linux Stream binary missing: {BIN}") if not FIXTURE.is_file(): raise SystemExit(f"power-loss JSS fixture missing: {FIXTURE}") with socket.socket() as sock: try: sock.bind(("127.0.0.1", PORT)) except OSError as error: raise SystemExit(f"power-loss oracle port {PORT} is busy: {error}")
state_dir = pathlib.Path(tempfile.mkdtemp(prefix="stream-powerloss-")) scope = os.environ.get("STREAM_POWERLOSS_SCOPE", "all") if scope not in ("all", "lifecycle", "guard", "mutation"): raise SystemExit("STREAM_POWERLOSS_SCOPE must be all, lifecycle, guard, or mutation") mutation_kind = os.environ.get("STREAM_POWERLOSS_MUTATION_KIND", "all") if mutation_kind not in ("all", "compaction", "patch"): raise SystemExit("STREAM_POWERLOSS_MUTATION_KIND must be all, compaction, or patch") succeeded = False try: run([ "docker", "run", "-d", "--pull=never", "--privileged", "--name", CONTAINER, # host.docker.internal is a Docker Desktop nicety; on Linux docker # it does not resolve unless mapped explicitly. Without this the # suite reports "pinned simulator is not reachable" on any Linux # runner even when the simulator is serving on the host. "--add-host", "host.docker.internal:host-gateway", "-p", f"127.0.0.1:{PORT}:6099", "-v", f"{BIN}:/oracle/stream:ro", "-v", f"{FIXTURE}:/oracle/fixture.jss:ro", "-v", f"{IMPORT_CSV}:/oracle/timestamp-import.csv:ro", "-v", f"{ROOT / 'tests' / 'strict_nbd.py'}:/oracle/strict_nbd.py:ro", "-v", f"{state_dir}:/state", IMAGE, "sleep", "infinity", ]) if scope != "mutation": require_simulator() calibrate(state_dir) if scope in ("all", "lifecycle"): for point, ordinal in CASES: run_case(state_dir, point, ordinal) if scope in ("all", "lifecycle", "guard"): run_restart_after_cleanup_guard_case(state_dir) if scope in ("all", "mutation"): if mutation_kind != "patch": for point in COMPACTION_CASES: run_compaction_powerloss_case(state_dir, point) if mutation_kind != "compaction": for point in PATCH_CASES: run_patch_powerloss_case(state_dir, point) print(f"powerloss: selected {scope} schedules recovered") succeeded = True finally: dexec_sh("pkill -9 stream 2>/dev/null || true; pkill -9 nbdkit 2>/dev/null || true", check=False) detach_and_unmount() run(["docker", "rm", "-f", CONTAINER], check=False) if succeeded: shutil.rmtree(state_dir, ignore_errors=True) else: print(f"powerloss: preserved failure evidence at {state_dir}")
if __name__ == "__main__": main()