From fcc63f8ec6e4e43510c9281cfc58d023b377b240 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 27 Jul 2026 17:00:15 -0500 Subject: [PATCH] bootstrap: break the batch submit/consume deadlock A dispatch batch is 100,000 jobs; the job queue holds worker_count * 2 (400). finishWorkBatch submitted every job before consuming a single result, which is a three-way circular wait: main fiber blocked in submit(), waiting for a worker to pop a job workers blocked in Budget.acquire(), which *waits* when exhausted budget released only when the main fiber consumes results Nothing was ever queued for completion. Reproduced on a ccx23 against the real network in 90 seconds with 50 workers and a 256 MiB budget: 1,403 repositories handled, completion_queued_total 0, queue_depth pinned at capacity, workers pinned at 50 -- the same shape as the 27 July whole-network run at 1/50th the machine. Consuming while submitting releases budget as repositories reach the archive's durability boundary, so workers keep draining the queue. The accompanying contract test does NOT reproduce this and passes on the broken code; the pinned simulator's repositories are ~4 KiB, so concurrent demand cannot exceed a budget a single repository fits in. It pins the weaker property that a batch far larger than the queue converges. Its docstring says so. Co-Authored-By: Claude Opus 5 (1M context) --- src/internal/bootstrap/engine.zig | 55 +++++++-- tests/backfill_batch_contract.py | 191 ++++++++++++++++++++++++++++++ 2 files changed, 237 insertions(+), 9 deletions(-) create mode 100644 tests/backfill_batch_contract.py 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()) -- 2.54.0