diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index ec96265..d97f2fe 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -406,6 +406,18 @@ const Work = struct { } } + /// Take whatever is ready without waiting. The submit loop uses this so it + /// can release memory before it blocks on a full queue; waiting there would + /// be the deadlock this exists to avoid. + fn takeReadyNow(self: *Work) ?std.ArrayList(Result) { + self.result_mu.lockUncancelable(self.io); + defer self.result_mu.unlock(self.io); + if (self.results.items.len == 0) return null; + const out = self.results; + self.results = .empty; + return out; + } + fn takeReadyThrough(self: *Work, target: usize) ?std.ArrayList(Result) { self.result_mu.lockUncancelable(self.io); defer self.result_mu.unlock(self.io); @@ -627,6 +639,21 @@ const CompletionBatcher = struct { } }; +fn consumeReady( + work: *Work, + completions: *CompletionBatcher, + maybe_results: ?std.ArrayList(Result), +) !void { + var results = maybe_results orelse return; + defer results.deinit(work.allocator); + for (results.items) |result| { + if (result.status == .complete and result.completion_requires_durability) + try completions.queue(result) + else + try completions.persistImmediate(result); + } +} + fn finishWorkBatch( work: *Work, completions: *CompletionBatcher, @@ -639,18 +666,28 @@ fn finishWorkBatch( if (shuffle) { shuffleJobs(jobs, @bitCast(Io.Timestamp.now(work.io, .awake).toMicroseconds())); } - for (jobs) |job| try work.submit(job); + // Submit and consume together. The job queue is bounded at + // worker_count * 2, so a whole-network batch (100,000) is far larger than + // it: submitting the whole batch first blocks this fiber on a full queue + // while finished repositories pile up in `results`, each still owning its + // CAR against the in-flight budget. The budget then exhausts, workers can + // no longer allocate, they stop draining the queue, and the submit this + // fiber is blocked on never completes. Consuming as we go hands those + // repositories to the completion batcher, which releases them at the + // archive's durability boundary and keeps the budget circulating. + // + // Every test corpus is smaller than the queue, which is why this only + // ever appeared at whole-network scale. + for (jobs) |job| { + if (work.getFatal()) |err| return err; + try consumeReady(work, completions, work.takeReadyNow()); + try work.submit(job); + } const target = work.submittedCount(); while (true) { if (work.getFatal()) |err| return err; - var results = work.takeReadyThrough(target) orelse break; - defer results.deinit(work.allocator); - for (results.items) |result| { - if (result.status == .complete and result.completion_requires_durability) - try completions.queue(result) - else - try completions.persistImmediate(result); - } + const results = work.takeReadyThrough(target) orelse break; + try consumeReady(work, completions, results); } if (work.getFatal()) |err| return err; try completions.drain(work.archive); diff --git a/tests/backfill_batch_contract.py b/tests/backfill_batch_contract.py new file mode 100644 index 0000000..ff38a0a --- /dev/null +++ b/tests/backfill_batch_contract.py @@ -0,0 +1,191 @@ +# /// script +# dependencies = [] +# /// +"""backfill-batch contract: a dispatch batch larger than the job queue must +still make durable progress. + +This is the defect that ended the 27 July 2026 whole-network experiment, and +the one that made the 23 July run look like a dispatch-boundary metric problem. + +The job queue is bounded at `worker_count * 2`. `finishWorkBatch` used to +submit every job in the batch before consuming a single result, so with a +whole-network batch (100,000) against a 400-slot queue the submitting fiber +blocked on a full queue while finished repositories accumulated in `results`, +each still owning its CAR against the in-flight budget. The budget exhausted, +workers could no longer allocate, they stopped draining the queue, and the +submit the fiber was blocked on never completed. Nothing was ever queued for +completion: 12,686 repositories emitted, zero durable, every row still +`not_started`. + +Two conditions are required, which is why nothing caught it: + +1. more jobs in one batch than the queue holds -- every other test corpus is + smaller than the queue, and +2. an in-flight budget small enough that retained results exhaust it -- the + default 8 GiB is never reached by a small fixture. + +WHAT THIS TEST DOES NOT DO: it does not reproduce the production deadlock, and +it passes on the broken code. The pinned simulator's repositories are ~4 KiB, +so concurrent demand cannot exceed any budget that a single repository still +fits in -- measured, the whole 3,000-repository run peaks at ~50 KiB in flight. +The deadlock needs per-repository memory large enough that concurrent workers +exhaust a multi-GB budget while the submit loop stays blocked, which only real +repositories produce. This is the same fixture limit that keeps the +differential-oracle row partial in docs/semantic-parity.md, in a second and +distinct way. + +What it does pin is the weaker property that a dispatch batch many times larger +than the job queue converges, and that every repository handled becomes durable +rather than merely counted. That is worth holding, but do not read a pass here +as evidence that the deadlock is fixed. + +run: uv run tests/backfill_batch_contract.py (or `just backfill-batch-contract`) +""" +import os +import pathlib +import re +import shutil +import subprocess +import sys +import time +import urllib.error +import urllib.request + +PORT = 6103 +BASE = f"http://127.0.0.1:{PORT}" +DATA = pathlib.Path("data-backfill-batch") +BIN = "./zig-out/bin/stream" +SIM = os.environ.get("STREAM_SIM_URL", "http://127.0.0.1:7777") + +# The deadlock needs concurrent demand to exceed the budget, so it needs many +# workers holding memory at once -- one worker can never exhaust it, which is +# why a single-worker version of this test passes on the broken code. +WORKERS = 16 +REPOS = int(os.environ.get("STREAM_BATCH_REPOS", "3000")) +# Measured on this corpus: ~580 KiB per repository in flight. 16 concurrent +# workers therefore demand ~9 MiB at once, so a 4 MiB budget cannot satisfy +# them all and some must wait -- which is the production condition. One +# repository still fits with ~7x headroom, so a failure here is the deadlock +# and not a budget too small for a single repo. Re-measure if the simulator +# corpus changes. +INFLIGHT_BYTES = 64 * 1024 + +# Generous relative to the work: 64 tiny simulator repositories converge in +# seconds. Exceeding this means no progress is being made, not slow progress. +TIMEOUT_S = 180 + +SAMPLE_RE = re.compile(r"^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+([^\s]+)$") +LABEL_RE = re.compile(r'(\w+)="([^"]*)"') + + +def metrics(): + with urllib.request.urlopen(BASE + "/metrics", timeout=5) as r: + body = r.read().decode() + out = {} + for line in body.splitlines(): + if line.startswith("#"): + continue + m = SAMPLE_RE.match(line.strip()) + if not m: + continue + name, labels, value = m.groups() + key = name + if labels: + pairs = dict(LABEL_RE.findall(labels)) + if "status" in pairs: + key = f'{name}{{status="{pairs["status"]}"}}' + elif "result" in pairs: + key = f'{name}{{result="{pairs["result"]}"}}' + try: + out[key] = float(value) + except ValueError: + pass + return out + + +def main(): + shutil.rmtree(DATA, ignore_errors=True) + log_path = pathlib.Path("/tmp/backfill-batch-contract.log") + log = open(log_path, "w") + proc = subprocess.Popen( + [ + BIN, + f"--port={PORT}", + f"--data-dir={DATA}", + f"--relay-url={SIM}", + f"--plc-url={SIM}", + f"--max-backfill-repos={REPOS}", + f"--backfill-workers={WORKERS}", + f"--backfill-max-inflight-bytes={INFLIGHT_BYTES}", + "--compaction-interval=0", + "--retry-interval=0", + ], + stdout=log, + stderr=log, + ) + + deadline = time.time() + TIMEOUT_S + observed = {} + try: + while time.time() < deadline: + if proc.poll() is not None: + raise SystemExit( + f"stream exited early with {proc.returncode}; see {log_path}" + ) + try: + observed = metrics() + except (urllib.error.URLError, ConnectionError, TimeoutError): + time.sleep(0.5) + continue + complete = observed.get('stream_backfill_repos_durable{status="complete"}', 0) + handled_now = observed.get( + "jetstream_backfill_handle_repo_duration_seconds_count", 0 + ) + if handled_now > WORKERS * 2 * 8 and complete >= handled_now: + break + time.sleep(0.5) + else: + queued = observed.get("jetstream_backfill_completion_queued_total", 0) + handled = observed.get( + "jetstream_backfill_handle_repo_duration_seconds_count", 0 + ) + inflight = observed.get("stream_backfill_inflight_bytes", 0) + raise SystemExit( + "batch never reached durable completion within " + f"{TIMEOUT_S}s: handled={handled:.0f} queued_for_completion={queued:.0f} " + f"durable_complete={observed.get('stream_backfill_repos_durable{status=\"complete\"}', 0):.0f} " + f"inflight_bytes={inflight:.0f} (budget {INFLIGHT_BYTES}). " + "A full queue with unconsumed results is the batch deadlock." + ) + + # Anti-vacuity: the batch must actually have exceeded the queue, or + # this test proves nothing about the condition it exists to pin. + handled = observed.get("jetstream_backfill_handle_repo_duration_seconds_count", 0) + assert handled > WORKERS * 2 * 8, f"only {handled:.0f} repositories were handled" + queue_capacity = WORKERS * 2 + assert REPOS > queue_capacity * 4, ( + f"{REPOS} jobs vs {queue_capacity}-slot queue is not an overflow" + ) + + complete = observed['stream_backfill_repos_durable{status="complete"}'] + # Not every discovered repository is eligible, so the bar is that every + # repository actually handled became durable -- not an exact count. + assert complete >= handled, f"durable complete {complete:.0f} < handled {handled:.0f}" + + peak = observed.get("stream_backfill_inflight_peak_bytes", 0) + print( + f"ok: {complete:.0f}/{REPOS} durably complete with a " + f"{queue_capacity}-slot queue and a {INFLIGHT_BYTES // 1024} KiB budget " + f"(peak in-flight {peak / 1024:.0f} KiB)" + ) + finally: + proc.terminate() + try: + proc.wait(timeout=30) + except subprocess.TimeoutExpired: + proc.kill() + log.close() + + +if __name__ == "__main__": + sys.exit(main())