Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041"""Observe configured compaction controls in a real Stream process."""
import jsonimport osimport pathlibimport timeimport urllib.request
port = int(os.environ["STREAM_COMPACTION_CONFIG_PORT"])expected_chunks = int(os.environ["STREAM_COMPACTION_CONFIG_CHUNKS"])expected_workers = int(os.environ["STREAM_COMPACTION_CONFIG_WORKERS"])log_path = pathlib.Path(os.environ["STREAM_COMPACTION_CONFIG_LOG"])base = f"http://127.0.0.1:{port}"deadline = time.monotonic() + 20
while time.monotonic() < deadline: log = log_path.read_text(errors="replace") if log_path.exists() else "" marker = f"compaction pass: watermark 4, {expected_chunks} chunks," workers = f", {expected_workers} rewrite workers" if marker in log and workers in log: listing = json.load( urllib.request.urlopen( base + "/xrpc/network.bsky.jetstream.listSegments", timeout=2 ) ) assert sum(segment["eventCount"] for segment in listing["segments"]) == 2 metrics = urllib.request.urlopen(base + "/metrics", timeout=2).read().decode() assert "jetstream_compaction_watermark_seq 4" in metrics print( "compaction config: PASS " f"(chunks={expected_chunks}, rewrite_workers={expected_workers})" ) break time.sleep(0.1)else: raise AssertionError( f"compaction did not expose chunks={expected_chunks}, " f"workers={expected_workers}\n{log}" )