From e2ab25c808f4d875dc4557ee2efb89dff62e1922 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 27 Jul 2026 20:51:13 -0500 Subject: [PATCH] bootstrap: delete the in-flight byte budget Upstream has no such flag, no such module, and no in-flight accounting anywhere in its ingest path. Ours blocked in acquire() when exhausted, which was half of what deadlocked experiment 4, and then sat pinned at its cap for all of experiment 5 while the machine was 86% idle. It arrived in 94e50a0 "make full-network backfill production-ready" with an empty commit body -- no reason recorded, then or since. Worker count is the concurrency bound now, as it is upstream. Removes byte_budget.zig, --backfill-max-inflight-bytes, and the two gauges that would otherwise report 0 forever. Co-Authored-By: Claude Opus 5 (1M context) --- deploy/README.md | 1 - docs/configuration.md | 19 ++-- docs/gotchas.md | 23 ++-- docs/invariants.md | 21 ++-- docs/testing.md | 17 ++- src/internal/bootstrap/byte_budget.zig | 151 ------------------------- src/internal/bootstrap/engine.zig | 33 ++---- src/internal/bootstrap/lifecycle.zig | 2 - src/internal/metrics.zig | 12 -- src/main.zig | 4 - src/tests.zig | 1 - tests/backfill_batch_contract.py | 14 +-- 12 files changed, 53 insertions(+), 245 deletions(-) delete mode 100644 src/internal/bootstrap/byte_budget.zig 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() -- knotmirror 0.0.1