diff --git a/deploy/README.md b/deploy/README.md index bc50f93..47d0d89 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -46,7 +46,6 @@ serve --backfill --backfill-workers=100 --backfill-async-flush-workers=4 ---backfill-max-inflight-bytes=8589934592 --subscribe-read-log-retention-bytes=268435456 --subscribe-block-cache-bytes=67108864 --subscribe-read-batch=1024 diff --git a/docs/configuration.md b/docs/configuration.md index cbc76b8..b5b059c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -170,7 +170,6 @@ keyspace. --backfill-workers=100 --backfill-batch-size=100000 --backfill-async-flush-workers=4 ---backfill-max-inflight-bytes=8589934592 --skip-merge-discovery ``` @@ -182,14 +181,16 @@ dispatch. Requests still go through the relay and follow its redirect, so this does not claim precise pre-request PDS scheduling. Zero selects the 100,000-entry default. -The byte flag bounds the owned complete CAR plus preparation/index state. -Preparation is serialized, copies the downloaded CAR into that budget, and -releases the scratch mmap before emission. Already-prepared repositories emit -concurrently from the owned CAR and loaded MST through fixed 1,024-row stack -batches, which lets one repository use the entire budget without letting -partial indexes deadlock one another. Four bootstrap-only workers detach -complete JSS blocks, compress concurrently, and commit/fsync in preparation -order; zero selects synchronous compression. +Worker count is the only concurrency bound, as it is upstream. Preparation +copies the downloaded CAR into the worker's own parse arena and releases the +scratch mmap before emission, so each repository's bytes are owned rather than +borrowed; repositories prepare and emit concurrently. Four bootstrap-only +workers detach complete JSS blocks, compress concurrently, and commit/fsync in +preparation order; zero selects synchronous compression. + +Stream previously carried a `--backfill-max-inflight-bytes` ceiling with a +blocking acquire. Upstream has no such flag or concept, it deadlocked +experiment 4 and throttled experiment 5 at its cap, and it is gone. Downloads need temporary free space on `--data-dir` equal to the concurrently in-flight CARs. Scratch files are deleted after each attempt and cleared on diff --git a/docs/gotchas.md b/docs/gotchas.md index 83cc35c..1d8f30f 100644 --- a/docs/gotchas.md +++ b/docs/gotchas.md @@ -82,16 +82,19 @@ approximation of the thing that ships. `STREAM_BUILD_HOST` points `admit` at a native amd64 daemon; the image is streamed back so the push uses local credentials and the build host never receives them. -**The pinned simulator cannot stress the in-flight budget.** Its repositories -are ~4 KiB; a 3,000-repository run peaks around 50 KiB in flight. No budget -that a single repository still fits in can therefore be exhausted by concurrent -workers, so any failure requiring budget exhaustion is unreachable offline — -which is exactly how the batch submit/consume deadlock survived 20 passing -suites. Reproducing that class costs about €0.03: a small cloud box crawling -bsky.network with a tight `--backfill-max-inflight-bytes` shows it in 90 -seconds. Reach for a real-network repro when a hypothesis involves memory -pressure or worker contention; do not conclude from a green suite that the -condition does not occur. +**The pinned simulator's repositories are ~4 KiB; real ones average ~10 MB.** +Measured against bsky.network: 15, 3.4, 2.2, 15, 4.6 and 37 MB in one sample. +Any failure that depends on real repository sizes — memory pressure, worker +contention, bandwidth saturation — is structurally unreachable offline. That is +how the batch submit/consume deadlock survived 20 passing suites. + +Reproducing that class costs about €0.03: a small cloud box crawling +bsky.network showed the deadlock in 90 seconds. Stream the image over SSH +(`docker image save | ssh 'docker image load'`) since the registry needs auth, +and write `/etc/docker/daemon.json` with explicit DNS first or BuildKit cannot +resolve. Reach for this whenever a hypothesis involves anything the fixture +cannot express; do not conclude from a green suite that the condition does not +occur. ## Divergences that were never decided diff --git a/docs/invariants.md b/docs/invariants.md index 06cdfd8..ddb0556 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -80,14 +80,19 @@ closed, or answer `5xx` — never return a smaller truthful-looking answer, because a client cannot tell the difference and will never come back for the gap. -**A bounded producer must not outrun the consumer that frees its budget.** -Bootstrap dispatch, the job queue, the in-flight byte budget and the completion -batcher form a cycle: workers acquire budget, the archive's durability boundary -releases it, and only the consuming fiber advances that. Any code that submits -a whole batch before consuming from it reintroduces the deadlock that ended -experiment 4 — the queue holds `worker_count * 2`, a batch is 100,000, and -`Budget.acquire` waits rather than failing. Submit and consume together. See -`bootstrap/engine.zig`. +**A bounded producer must not outrun its consumer.** Bootstrap dispatch, the +bounded job queue and the completion batcher form a cycle: the queue holds +`worker_count * 2` while a batch is 100,000, so submitting a whole batch before +consuming from it blocks the submitting fiber while finished repositories pile +up unconsumed. That is what ended experiment 4, compounded by an in-flight byte +budget whose `acquire` waited rather than failing. The budget is gone — +upstream has no such thing — but the ordering rule stands on its own: submit +and consume together. See `bootstrap/engine.zig`. + +**Do not add a bound upstream does not have without deciding to.** Both halves +of that deadlock were Stream-only mechanisms introduced without justification +in their commit messages, one under a subject claiming to *match* upstream. +Diverging is allowed; diverging by reflex is what cost two experiments. ## See also diff --git a/docs/testing.md b/docs/testing.md index 69fe4b6..f76b024 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -70,18 +70,17 @@ receipt. ## When offline suites cannot reach a failure Some failure modes are structurally unreachable in the fixture, and a green -suite says nothing about them. The clearest example: the pinned simulator's -repositories are ~4 KiB, so concurrent workers can never exhaust an in-flight -budget that a single repository fits in. The batch submit/consume deadlock that -ended experiment 4 survived 20 passing suites for exactly that reason, and four +suite says nothing about them. The pinned simulator's repositories are ~4 KiB; +real ones average ~10 MB. The batch submit/consume deadlock that ended +experiment 4 survived 20 passing suites for exactly that reason, and four attempts to force it out of the simulator failed. A real-network reproduction costs about €0.03. Provision a small cloud box, -stream the image to it, and run against `bsky.network` with the pressure turned -up — for that bug, 50 workers and `--backfill-max-inflight-bytes=268435456` -showed it in 90 seconds, and the before/after was unambiguous. Reach for this -whenever a hypothesis involves memory pressure, worker contention, or repository -sizes the simulator cannot produce. +stream the image to it, and run against `bsky.network` — that bug showed in 90 +seconds and the before/after was unambiguous. It is also the only way to +measure throughput honestly: backfill is bandwidth-bound at ~10 MB/repo, which +no local fixture reproduces. Reach for this whenever a hypothesis involves +memory pressure, worker contention, or repository sizes. `just backfill-batch-contract` pins the weaker property that a dispatch batch far larger than the job queue converges. Its docstring is explicit that it diff --git a/src/internal/bootstrap/byte_budget.zig b/src/internal/bootstrap/byte_budget.zig deleted file mode 100644 index 5dd47fb..0000000 --- a/src/internal/bootstrap/byte_budget.zig +++ /dev/null @@ -1,151 +0,0 @@ -//! Blocking allocator-wide byte budget for bootstrap workers. -//! -//! CAR preparation allocates through `BudgetedAllocator`; HTTP bodies stream -//! to disk, then preparation copies each complete CAR into the budget before -//! releasing its mmap. Emission reuses that owned CAR and the fully loaded MST -//! through a fixed 1,024-row stack batch. The backfill engine admits the -//! upstream worker count for network I/O, serializes only CAR preparation, and -//! lets already-prepared repositories emit concurrently. Thus one preparation -//! can wait for earlier prepared ownership to drain without several partial -//! indexes deadlocking each other. - -const std = @import("std"); - -const Io = std.Io; -const Allocator = std.mem.Allocator; - -pub const Budget = struct { - io: Io, - limit: usize, - used: usize = 0, - peak: usize = 0, - used_gauge: ?*std.atomic.Value(usize) = null, - peak_gauge: ?*std.atomic.Value(usize) = null, - mu: Io.Mutex = .init, - available: Io.Condition = .init, - - pub fn init(io: Io, limit: usize) Budget { - return .{ .io = io, .limit = limit }; - } - - fn acquire(self: *Budget, amount: usize) bool { - if (amount > self.limit) return false; - self.mu.lockUncancelable(self.io); - defer self.mu.unlock(self.io); - while (amount > self.limit - self.used) - self.available.waitUncancelable(self.io, &self.mu); - self.used += amount; - self.peak = @max(self.peak, self.used); - if (self.used_gauge) |gauge| gauge.store(self.used, .monotonic); - if (self.peak_gauge) |gauge| gauge.store(self.peak, .monotonic); - return true; - } - - fn release(self: *Budget, amount: usize) void { - self.mu.lockUncancelable(self.io); - defer self.mu.unlock(self.io); - std.debug.assert(amount <= self.used); - self.used -= amount; - if (self.used_gauge) |gauge| gauge.store(self.used, .monotonic); - self.available.broadcast(self.io); - } - - pub fn gauges(self: *Budget) struct { used: usize, peak: usize, limit: usize } { - self.mu.lockUncancelable(self.io); - defer self.mu.unlock(self.io); - return .{ .used = self.used, .peak = self.peak, .limit = self.limit }; - } -}; - -pub const BudgetedAllocator = struct { - backing: Allocator, - budget: *Budget, - /// Per-worker ceiling. This prevents one oversized response from owning - /// the whole shared budget and then deadlocking while growing itself. - limit: usize = std.math.maxInt(usize), - owned: usize = 0, - - pub fn allocator(self: *BudgetedAllocator) Allocator { - return .{ .ptr = self, .vtable = &.{ - .alloc = alloc, - .resize = resize, - .remap = remap, - .free = free, - } }; - } - - fn alloc(ctx: *anyopaque, len: usize, alignment: std.mem.Alignment, ret_addr: usize) ?[*]u8 { - const self: *BudgetedAllocator = @ptrCast(@alignCast(ctx)); - if (len > self.limit -| self.owned) return null; - if (!self.budget.acquire(len)) return null; - const ptr = self.backing.rawAlloc(len, alignment, ret_addr) orelse { - self.budget.release(len); - return null; - }; - self.owned += len; - return ptr; - } - - fn resize(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, new_len: usize, ret_addr: usize) bool { - const self: *BudgetedAllocator = @ptrCast(@alignCast(ctx)); - if (new_len == memory.len) return true; - if (new_len < memory.len) { - if (!self.backing.rawResize(memory, alignment, new_len, ret_addr)) return false; - self.owned -= memory.len - new_len; - self.budget.release(memory.len - new_len); - return true; - } - const growth = new_len - memory.len; - if (growth > self.limit -| self.owned) return false; - if (!self.budget.acquire(growth)) return false; - if (self.backing.rawResize(memory, alignment, new_len, ret_addr)) { - self.owned += growth; - return true; - } - self.budget.release(growth); - return false; - } - - fn remap(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, new_len: usize, ret_addr: usize) ?[*]u8 { - const self: *BudgetedAllocator = @ptrCast(@alignCast(ctx)); - if (new_len == memory.len) return memory.ptr; - if (new_len < memory.len) { - const ptr = self.backing.rawRemap(memory, alignment, new_len, ret_addr) orelse return null; - self.owned -= memory.len - new_len; - self.budget.release(memory.len - new_len); - return ptr; - } - const growth = new_len - memory.len; - if (growth > self.limit -| self.owned) return null; - if (!self.budget.acquire(growth)) return null; - const ptr = self.backing.rawRemap(memory, alignment, new_len, ret_addr) orelse { - self.budget.release(growth); - return null; - }; - self.owned += growth; - return ptr; - } - - fn free(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, ret_addr: usize) void { - const self: *BudgetedAllocator = @ptrCast(@alignCast(ctx)); - self.backing.rawFree(memory, alignment, ret_addr); - std.debug.assert(memory.len <= self.owned); - self.owned -= memory.len; - self.budget.release(memory.len); - } -}; - -test "budgeted allocator blocks aggregate ownership and releases it" { - const testing = std.testing; - var budget = Budget.init(testing.io, 64); - var wrapped: BudgetedAllocator = .{ .backing = testing.allocator, .budget = &budget }; - const allocator = wrapped.allocator(); - - const first = try allocator.alloc(u8, 40); - const second = try allocator.alloc(u8, 24); - try testing.expect(allocator.alloc(u8, 65) == error.OutOfMemory); - try testing.expectEqual(@as(usize, 64), budget.gauges().peak); - allocator.free(first); - allocator.free(second); - try testing.expectEqual(@as(usize, 0), budget.gauges().used); -} diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index 7987871..e6eaed6 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -2,9 +2,8 @@ //! //! listRepos remains a serial page walk. Whole pages accumulate until the //! configured entry target, then eligible repositories are shuffled and fan -//! out across long-lived workers. Every worker allocates through one shared -//! byte budget, so worker count controls parallelism while `max_inflight_bytes` -//! controls actual transient heap ownership. Repository completion follows the +//! out across long-lived workers. Worker count is the only concurrency bound, +//! as it is upstream. Repository completion follows the //! archive writer's durability boundary independently of the dispatch batch: //! //! repo/not_started -> concurrent archive appends -> archive fsync @@ -17,7 +16,6 @@ const std = @import("std"); const zat = @import("zat"); const archive_mod = @import("../archive.zig"); -const byte_budget = @import("byte_budget.zig"); const crashpoint = @import("../crashpoint.zig"); const fetched_car = @import("fetched_car.zig"); const prepared_fetch = @import("prepared_fetch.zig"); @@ -42,7 +40,6 @@ const rate_limit_retries: u8 = 20; const rate_limit_ceiling_s: u64 = 330; pub const default_workers: usize = 100; pub const default_batch_size: usize = 100_000; -pub const default_max_inflight_bytes: usize = 8 * 1024 * 1024 * 1024; pub const Config = struct { relay_http: []const u8, @@ -59,7 +56,6 @@ pub const Config = struct { /// Target listRepos entries per page-aligned dispatch batch. Zero uses /// Jetstream's production default, not Atmos's smaller library default. batch_size: usize = default_batch_size, - max_inflight_bytes: usize = default_max_inflight_bytes, metrics: ?*metrics_mod.Stats = null, }; @@ -235,7 +231,6 @@ const Work = struct { archive: *archive_mod.Archive, queue: JobQueue, group: Io.Group = .init, - budget: byte_budget.Budget, worker_count: usize, result_mu: Io.Mutex = .init, result_ready: Io.Condition = .init, @@ -252,23 +247,17 @@ const Work = struct { archive: *archive_mod.Archive, trace_context: observability.Context, ) !Work { - if (config.max_inflight_bytes == 0) return error.InvalidBackfillLimits; const worker_count = effectiveWorkerCount(config.workers); const queue_capacity = worker_count * 2; - var work: Work = .{ + const work: Work = .{ .allocator = allocator, .io = io, .config = config, .archive = archive, .queue = .{ .buffer = try allocator.alloc(Job, queue_capacity) }, - .budget = byte_budget.Budget.init(io, config.max_inflight_bytes), .worker_count = worker_count, .trace_context = trace_context, }; - if (config.metrics) |m| { - work.budget.used_gauge = &m.backfill_inflight_bytes; - work.budget.peak_gauge = &m.backfill_inflight_peak_bytes; - } return work; } @@ -337,12 +326,11 @@ const Work = struct { } fn workerLoop(self: *Work) void { - var bounded: byte_budget.BudgetedAllocator = .{ - .backing = self.allocator, - .budget = &self.budget, - .limit = self.config.max_inflight_bytes, - }; - const allocator = bounded.allocator(); + // No in-flight byte ceiling: worker count is the concurrency bound, as + // it is upstream, which has no such budget. The version here blocked in + // acquire() when exhausted, which is half of what deadlocked + // experiment 4 and throttled experiment 5 at its cap. + const allocator = self.allocator; while (self.queue.pop(self.io)) |job| { self.publishGauges(); if (self.getFatal() != null) { @@ -429,9 +417,6 @@ const Work = struct { fn publishGauges(self: *Work) void { const m = self.config.metrics orelse return; - const gauges = self.budget.gauges(); - m.backfill_inflight_bytes.store(gauges.used, .monotonic); - m.backfill_inflight_peak_bytes.store(gauges.peak, .monotonic); m.backfill_queue_depth.store(self.queue.count(self.io), .monotonic); } }; @@ -975,7 +960,6 @@ test "explicit selected backfill resolves identity, bypasses listRepos, and pers .data_dir = data_dir, .selected_repos = &selected, .workers = 1, - .max_inflight_bytes = 1024 * 1024, .metrics = &instrumentation, }, &archive, &store); // A RepoNotFound response is a successful durable completion upstream: @@ -1054,7 +1038,6 @@ test "explicit selected backfill repairs identity metadata without redownloading .data_dir = data_dir, .selected_repos = &selected, .workers = 1, - .max_inflight_bytes = 1024 * 1024, .metrics = &instrumentation, }, &archive, &store); try testing.expectEqual(@as(u64, 0), stats.completed); diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index b55ac5b..c912723 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -60,7 +60,6 @@ pub const Options = struct { /// Upstream default is four; zero explicitly keeps synchronous block /// compression. This applies only while the repository crawl is active. async_flush_workers: usize = default_async_flush_workers, - max_inflight_bytes: usize = engine.default_max_inflight_bytes, /// Debug-only upstream control. Main also enables it automatically for /// max-repo and explicit-DID bootstrap selections. skip_merge_discovery: bool = false, @@ -517,7 +516,6 @@ fn runBootstrap( .max_repos = opts.max_repos, .workers = opts.workers, .batch_size = opts.batch_size, - .max_inflight_bytes = opts.max_inflight_bytes, .metrics = stats, }, dst, repo_st, trace_context }); const bf = switch (try race.await()) { diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index 592c943..2accb38 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -172,8 +172,6 @@ pub const Stats = struct { retry_skipped_host_parked_total: Counter = .init(0), backfill_workers_active: std.atomic.Value(usize) = .init(0), backfill_queue_depth: std.atomic.Value(usize) = .init(0), - backfill_inflight_bytes: std.atomic.Value(usize) = .init(0), - backfill_inflight_peak_bytes: std.atomic.Value(usize) = .init(0), import_jobs_ok_total: Counter = .init(0), import_jobs_error_total: Counter = .init(0), import_job_duration_count: Counter = .init(0), @@ -1085,10 +1083,6 @@ pub fn format( \\stream_backfill_workers_active {d} \\# TYPE stream_backfill_queue_depth gauge \\stream_backfill_queue_depth {d} - \\# TYPE stream_backfill_inflight_bytes gauge - \\stream_backfill_inflight_bytes {d} - \\# TYPE stream_backfill_inflight_peak_bytes gauge - \\stream_backfill_inflight_peak_bytes {d} \\ , .{ stats.retry_resynced_total.load(.monotonic), @@ -1116,8 +1110,6 @@ pub fn format( stats.backfill_failed_total.load(.monotonic), stats.backfill_workers_active.load(.monotonic), stats.backfill_queue_depth.load(.monotonic), - stats.backfill_inflight_bytes.load(.monotonic), - stats.backfill_inflight_peak_bytes.load(.monotonic), }) catch {}; w.print("# TYPE jetstream_compaction_tombstones_collected_total counter\n", .{}) catch {}; @@ -1444,8 +1436,6 @@ test "format renders counters" { _ = stats.retry_skipped_host_parked_total.fetchAdd(10, .monotonic); stats.backfill_workers_active.store(4, .monotonic); stats.backfill_queue_depth.store(23, .monotonic); - stats.backfill_inflight_bytes.store(4096, .monotonic); - stats.backfill_inflight_peak_bytes.store(8192, .monotonic); _ = stats.import_jobs_ok_total.fetchAdd(1, .monotonic); stats.import_phase.store(2, .monotonic); _ = stats.import_rows_mutated_total.fetchAdd(12, .monotonic); @@ -1575,8 +1565,6 @@ test "format renders counters" { try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_repos_total{result=\"complete\"} 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_workers_active 4") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_queue_depth 23") != null); - try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_inflight_bytes 4096") != null); - try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_inflight_peak_bytes 8192") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_build_info{version=") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completed_total 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_discovered_total 2") != null); diff --git a/src/main.zig b/src/main.zig index e446392..69522b6 100644 --- a/src/main.zig +++ b/src/main.zig @@ -626,7 +626,6 @@ pub fn main(init: std.process.Init.Minimal) !void { var backfill_workers: usize = backfill_engine.default_workers; var backfill_batch_size: usize = backfill_engine.default_batch_size; var backfill_async_flush_workers: usize = lifecycle.default_async_flush_workers; - var backfill_max_inflight_bytes: usize = backfill_engine.default_max_inflight_bytes; var skip_merge_discovery = false; var relay_http: []const u8 = "http://localhost:7777"; var compaction_interval_ns: u64 = 4 * std.time.ns_per_hour; @@ -736,8 +735,6 @@ pub fn main(init: std.process.Init.Minimal) !void { backfill_batch_size = if (configured == 0) backfill_engine.default_batch_size else configured; } else if (std.mem.startsWith(u8, arg, "--backfill-async-flush-workers=")) { backfill_async_flush_workers = try std.fmt.parseInt(usize, arg["--backfill-async-flush-workers=".len..], 10); - } else if (std.mem.startsWith(u8, arg, "--backfill-max-inflight-bytes=")) { - backfill_max_inflight_bytes = try std.fmt.parseInt(usize, arg["--backfill-max-inflight-bytes=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--relay-http=")) { relay_http = arg["--relay-http=".len..]; } else if (std.mem.startsWith(u8, arg, "--compaction-interval=")) { @@ -1148,7 +1145,6 @@ pub fn main(init: std.process.Init.Minimal) !void { .workers = backfill_workers, .batch_size = backfill_batch_size, .async_flush_workers = backfill_async_flush_workers, - .max_inflight_bytes = backfill_max_inflight_bytes, .skip_merge_discovery = skip_merge_discovery, .compaction_enabled = compaction_interval_ns != 0, }, diff --git a/src/tests.zig b/src/tests.zig index abde642..23c99a8 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -23,7 +23,6 @@ comptime { _ = @import("internal/bootstrap/merge.zig"); _ = @import("internal/bootstrap/resync.zig"); _ = @import("internal/bootstrap/retry.zig"); - _ = @import("internal/bootstrap/byte_budget.zig"); _ = @import("internal/bootstrap/lifecycle.zig"); _ = @import("internal/archive.zig"); _ = @import("internal/manifest.zig"); diff --git a/tests/backfill_batch_contract.py b/tests/backfill_batch_contract.py index ff38a0a..7c2988a 100644 --- a/tests/backfill_batch_contract.py +++ b/tests/backfill_batch_contract.py @@ -62,13 +62,6 @@ SIM = os.environ.get("STREAM_SIM_URL", "http://127.0.0.1:7777") # 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. @@ -116,7 +109,6 @@ def main(): 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", ], @@ -149,12 +141,10 @@ def main(): 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." ) @@ -172,11 +162,9 @@ def main(): # 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)" + f"{queue_capacity}-slot queue" ) finally: proc.terminate()