diff --git a/README.md b/README.md
index 90fcc6c..5917314 100644
--- a/README.md
+++ b/README.md
@@ -148,6 +148,7 @@ Production bootstrap controls:
 --backfill
 --backfill-workers=100
 --backfill-batch-size=100000
+--backfill-async-flush-workers=4
 --backfill-max-inflight-bytes=8589934592
 --skip-merge-discovery
 ```
@@ -156,13 +157,16 @@ Worker count controls concurrent network downloads. Batch size counts every
 `listRepos` entry, including inactive and already-complete repositories;
 Stream accumulates whole 1,000-entry protocol pages until it reaches the target,
 then shuffles eligible repositories across PDS hosts and dispatches them. Zero
-selects the 100,000-entry production default. The byte flag bounds CAR index
-and conversion allocations; conversion is serialized so one repository can use
-the entire budget instead of being rejected by an arbitrary per-repo ceiling.
+selects the 100,000-entry production default. The byte flag bounds complete CAR
+preparation/index ownership. Preparation is serialized, while already prepared
+repositories emit concurrently from the loaded MST through fixed 1,024-row
+stack batches; this lets one repository use the entire budget without allowing
+partial indexes to deadlock one another.
 The zero worker value selects the 100-worker production default. HTTP bodies
-remain concurrent and disk-backed; complete CAR preparation, indexing, and
-conversion are serialized so one repository can use the entire shared 8 GiB
-bound without partial indexes deadlocking one another. Downloads need temporary
+remain concurrent and disk-backed. Four bootstrap-only workers detach complete
+JSS blocks, perform real zstd compression concurrently, and commit/fsync them
+in preparation order; zero explicitly selects synchronous compression.
+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 restart. Bootstrap
 owns `<data-dir>/backfill/repo-scratch`; steady live and failed-repo healing use
diff --git a/deploy/README.md b/deploy/README.md
index 8e00d7f..a21a816 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -23,7 +23,8 @@ just package        # ReleaseSafe x86_64-linux-gnu binary in zig-out/bin/
 --plc=https://plc.directory
 --data-dir=/data
 --backfill
---backfill-workers=16
+--backfill-workers=100
+--backfill-async-flush-workers=4
 --backfill-max-inflight-bytes=8589934592
 ```
 
@@ -32,12 +33,13 @@ usable mirrored SSD/NVMe. Keep RocksDB and the archive on the same RAID1 XFS
 filesystem: segment fsync precedes the RocksDB batch that acknowledges those
 bytes, and the unified filesystem is reflink-snapshotted for backup.
 
-The defaults reserve at most 8 GiB for transient backfill allocations. Six
-workers run effectively under that budget (the requested 16 remains a ceiling),
-with 1.25 GiB as the worst-case allowance for each active worker. The pod still
-needs headroom for RocksDB, archive/compaction buffers, live capture,
-thread stacks, and libc; do not set the container limit equal to the backfill
-budget.
+The defaults reserve at most 8 GiB for transient backfill allocations while
+admitting 100 concurrent disk-backed downloads. CAR preparation is serialized
+to avoid partial-index deadlock, but prepared repositories emit concurrently;
+four bootstrap-only workers compress detached JSS blocks and commit them in
+order. The pod still needs headroom for RocksDB, archive/compaction buffers,
+live capture, thread stacks, zstd frames, and libc; do not set the container
+limit equal to the backfill budget.
 
 ## promotion gate
 
diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md
index 45800d2..8f4809b 100644
--- a/docs/bootstrap-semantic-parity.md
+++ b/docs/bootstrap-semantic-parity.md
@@ -19,7 +19,8 @@ every experiment.
 | Explicit selected backfill and account identity metadata | `--backfill-repos` processes the caller's order serially, bypasses listRepos, cannot combine with max-repos, deletes stale merge-discovery state, resolves each selected DID even when its repo row is already complete, retains declared handle/PDS, and atomically maintains the normalized local handle index. Ordinary whole-network backfill does not perform this extra resolution. | Stream parses and deduplicates the explicit DID list, synthesizes one ordered selected page without touching listRepos, requires the configured real `DidResolver`, deletes the stale bootstrap discovery watermark, and never advances listRepos state. `repo/<did>` v5 persists Handle/PDS with same-batch `handle/<normalized>` transitions. A real loopback PLC + getRepo run fails on any listRepos request and proves terminal status plus index persistence; handle moves/collisions, rebuild, and restart reads are also exercised offline. | closed |
 | Merge latest revision | For every kept rev-bearing row, refresh top-level Rev/UpdatedAt while preserving Backfill.Rev; commit those rows atomically with the per-source merge cursor | merge collects the last kept rev per DID and `commitMergeSource` stages cursor + repo refreshes in one synced RocksDB batch; missing repo rows are skipped rather than synthesized malformed | closed |
 | Backfill dispatch batching | Accumulate whole 1,000-entry listRepos pages until at least 100,000 entries by default, counting inactive and complete rows; shuffle eligible repos; dispatch and drain before checkpoint | configurable and zero/default batch sizes drive the real pinned-simulator request ledger; a deterministic shuffle receipt exercises the production helper; full differential and predicate-kill restart tiers converge | closed |
-| Backfill worker admission | Positive worker counts are physical network concurrency; zero and omitted values select Jetstream's 100-worker production default | held real simulator getRepo requests observe exact peaks of 7/100/100 across three ReleaseSafe processes, then all 104 repos complete; disk-backed fetch remains concurrent while prepared-repo memory is cancelably serialized under the independent byte bound | closed |
+| Backfill worker admission | Positive worker counts are physical network concurrency; zero and omitted values select Jetstream's 100-worker production default | held real simulator getRepo requests observe exact peaks of 7/100/100 across three ReleaseSafe processes, then all 104 repos complete; disk-backed fetch remains concurrent, preparation is cancelably serialized under the byte bound, and prepared repositories emit concurrently | closed |
+| Backfill async flush | Initial handlers append in 1,024-row batches; four bootstrap-only compression workers by default; zero disables; detach blocks, compress concurrently, commit and fsync in preparation order before handler completion | initial emission uses the upstream batch boundary; real 8 MiB columnar blocks reach exactly four simultaneous zstd workers; an independent archive receipt proves async bytes equal synchronous JSS bytes, all eight blocks remain sequence-contiguous, and the committed watermark reaches the final block | closed |
 | Cursor durability | advance after every repo in the accumulated batch is durable; resume cursor drains empty; discovery cursor retains the last non-empty page cursor | the distinct relay/discovery keys commit atomically at the batch boundary; cursor unit receipt plus real after-repo-complete predicate kills converge after same-disk restart | closed |
 | Resource exhaustion | capacity failure is fatal, never a durable per-repo “too large” skip | OOM is lifecycle-fatal and tested | closed |
 | Resync ordering | validate replacement first, then append sync + replacement, then durable completion | `prepareRepo` precedes emission; sync plus replacement rows append in bounded 1,024-row batches, matching upstream. Memory does not require a second repository-sized event array; archive durability precedes RocksDB completion | closed |
diff --git a/docs/configuration-parity.md b/docs/configuration-parity.md
index b3d2ae7..20350a2 100644
--- a/docs/configuration-parity.md
+++ b/docs/configuration-parity.md
@@ -11,7 +11,7 @@ the same runtime mechanism and an offline receipt observes that behavior.
 | max/selected backfill repos | behavioral equivalent | lifecycle and differential receipts; public max-repo flag naming remains open |
 | backfill workers | closed | real held-request processes observe explicit 7 and both zero/omitted defaults at 100 simultaneous getRepo requests; all complete through physical archive convergence |
 | backfill batch size | closed | page-aligned accumulation counts all listRepos entries; configured, zero/default, request ordering, shuffle boundary, and atomic relay/discovery cursor cadence are covered offline |
-| backfill async flush workers | open | Stream still compresses and commits bootstrap blocks synchronously |
+| backfill async flush workers | closed | omitted defaults to four, positive values create that many real bootstrap-only zstd workers, and zero keeps synchronous compression; detached blocks commit/fsync in preparation order before repo completion, with exact four-worker overlap and byte-identical sync/async JSS receipts |
 | skip merge discovery | closed | explicit flag and automatic max/selected-repo enablement match upstream; the pinned simulator observes the normal discovery request and its absence when skipped |
 | failed-repo retry interval, global workers, per-host workers, max delay | closed | `just retry-config-contract` measures explicit and zero/default concurrency through held real HTTP requests, then reopens RocksDB to verify one durable failure and the configured jitter/cap window per row |
 | expensive repo-action rate limiter | closed | real HTTP/status receipt covers enabled and disabled modes |
diff --git a/docs/upstream-bootstrap-spec.md b/docs/upstream-bootstrap-spec.md
index f1aaf43..253a283 100644
--- a/docs/upstream-bootstrap-spec.md
+++ b/docs/upstream-bootstrap-spec.md
@@ -76,9 +76,12 @@ until steady_state. failed-repo retry loop: 4h interval, 16 workers, 4/host,
 - Stream keeps the upstream 1000-repo durability page and requests up to 16
   long-lived download workers on the 16-core target. There is no CAR-size
   exclusion. HTTP bodies stream to per-attempt scratch files on the data
-  volume, then mmap read-only; CAR indexing and conversion use the configurable
-  global allocation budget (8 GiB default). Conversion is serialized so an
-  individual repository may use the entire budget. Resource exhaustion is a
+  volume, then mmap read-only; complete CAR preparation/index ownership uses
+  the configurable global allocation budget (8 GiB default). Preparation is
+  serialized so an individual repository may use the entire budget; prepared
+  repositories emit allocation-free from the loaded MST through upstream's
+  fixed 1,024-row batches and feed four ordered async zstd workers by default.
+  Resource exhaustion is a
   lifecycle-fatal capacity error, never a durable per-repo skip. Scratch is
   cleared before bootstrap or steady retry starts and each attempt unlinks its
   own file. This deliberately trades temporary disk space for bounded RSS.
diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md
index c619666..9919d54 100644
--- a/docs/upstream-harness.md
+++ b/docs/upstream-harness.md
@@ -219,6 +219,13 @@ converge its physical archive, enter steady state, and drain cleanly. This
 detects the former behavior where the advertised count was silently reduced to
 four by a 2 GiB-per-worker formula.
 
+The bootstrap archive has a separate physical compression receipt. Four
+deterministic 8 MiB columnar blocks overlap in four real zstd workers (no sleep
+or compression mock). A second archive run compares synchronous and async JSS
+bytes exactly, then checks all eight committed blocks and the final durability
+watermark. This prevents `--backfill-async-flush-workers` from becoming a
+parsed-but-idle compatibility flag.
+
 ## pinned differential oracle (2026-07-20)
 
 `just differential-oracle` refuses to run unless the upstream checkout is at
diff --git a/src/internal/archive.zig b/src/internal/archive.zig
index 534c9b6..bef19f6 100644
--- a/src/internal/archive.zig
+++ b/src/internal/archive.zig
@@ -26,6 +26,147 @@ const log = std.log.scoped(.stream);
 
 pub const default_max_segment_bytes: usize = 256 * 1024 * 1024;
 
+const AsyncFlushJob = struct {
+    id: u64 = 0,
+    prepared: *writer_mod.PreparedBlock,
+    frame: ?[]u8 = null,
+    failure: ?anyerror = null,
+    compressed: Io.Event = .unset,
+};
+
+const AsyncFlushQueue = struct {
+    buffer: []*AsyncFlushJob,
+    head: usize = 0,
+    len: usize = 0,
+    closed: bool = false,
+    mu: Io.Mutex = .init,
+    not_empty: Io.Condition = .init,
+    not_full: Io.Condition = .init,
+
+    fn put(self: *AsyncFlushQueue, io: Io, job: *AsyncFlushJob) void {
+        self.mu.lockUncancelable(io);
+        defer self.mu.unlock(io);
+        while (self.len == self.buffer.len and !self.closed)
+            self.not_full.waitUncancelable(io, &self.mu);
+        if (self.closed) unreachable;
+        self.buffer[(self.head + self.len) % self.buffer.len] = job;
+        self.len += 1;
+        self.not_empty.signal(io);
+    }
+
+    fn pop(self: *AsyncFlushQueue, io: Io) !*AsyncFlushJob {
+        self.mu.lockUncancelable(io);
+        defer self.mu.unlock(io);
+        while (self.len == 0 and !self.closed)
+            try self.not_empty.wait(io, &self.mu);
+        if (self.len == 0) return error.Closed;
+        const job = self.buffer[self.head];
+        self.head = (self.head + 1) % self.buffer.len;
+        self.len -= 1;
+        self.not_full.signal(io);
+        return job;
+    }
+
+    fn close(self: *AsyncFlushQueue, io: Io) void {
+        self.mu.lockUncancelable(io);
+        defer self.mu.unlock(io);
+        self.closed = true;
+        self.not_empty.broadcast(io);
+        self.not_full.broadcast(io);
+    }
+};
+
+const AsyncFlush = struct {
+    allocator: Allocator,
+    io: Io,
+    queue: AsyncFlushQueue,
+    group: Io.Group = .init,
+    next_job_id: u64 = 0,
+    next_commit_id: u64 = 0,
+    commit_mu: Io.Mutex = .init,
+    commit_ready: Io.Condition = .init,
+    admission_mu: Io.Mutex = .init,
+    outstanding_mu: Io.Mutex = .init,
+    outstanding_ready: Io.Condition = .init,
+    outstanding: usize = 0,
+    workers_active: std.atomic.Value(usize) = .init(0),
+    workers_peak: std.atomic.Value(usize) = .init(0),
+
+    fn init(allocator: Allocator, io: Io, worker_count: usize) !AsyncFlush {
+        std.debug.assert(worker_count > 0);
+        return .{
+            .allocator = allocator,
+            .io = io,
+            .queue = .{ .buffer = try allocator.alloc(*AsyncFlushJob, worker_count) },
+        };
+    }
+
+    fn start(self: *AsyncFlush, worker_count: usize) !void {
+        errdefer {
+            self.queue.close(self.io);
+            self.group.cancel(self.io);
+        }
+        for (0..worker_count) |_| try self.group.concurrent(self.io, workerLoop, .{self});
+    }
+
+    fn deinit(self: *AsyncFlush) void {
+        self.queue.close(self.io);
+        self.group.await(self.io) catch {};
+        self.allocator.free(self.queue.buffer);
+    }
+
+    fn assign(self: *AsyncFlush, job: *AsyncFlushJob) void {
+        job.id = self.next_job_id;
+        self.next_job_id += 1;
+        self.outstanding_mu.lockUncancelable(self.io);
+        self.outstanding += 1;
+        self.outstanding_mu.unlock(self.io);
+    }
+
+    fn submit(self: *AsyncFlush, job: *AsyncFlushJob) void {
+        self.queue.put(self.io, job);
+    }
+
+    fn workerLoop(self: *AsyncFlush) void {
+        while (self.queue.pop(self.io)) |job| {
+            const active = self.workers_active.fetchAdd(1, .acq_rel) + 1;
+            var peak = self.workers_peak.load(.acquire);
+            while (active > peak) {
+                peak = self.workers_peak.cmpxchgWeak(peak, active, .acq_rel, .acquire) orelse break;
+            }
+            job.frame = writer_mod.compressPrepared(self.allocator, job.prepared) catch |err| blk: {
+                job.failure = err;
+                break :blk null;
+            };
+            _ = self.workers_active.fetchSub(1, .acq_rel);
+            job.compressed.set(self.io);
+        } else |_| {}
+    }
+
+    fn waitTurn(self: *AsyncFlush, id: u64) void {
+        self.commit_mu.lockUncancelable(self.io);
+        while (self.next_commit_id != id)
+            self.commit_ready.waitUncancelable(self.io, &self.commit_mu);
+    }
+
+    fn finishTurn(self: *AsyncFlush) void {
+        self.next_commit_id += 1;
+        self.commit_ready.broadcast(self.io);
+        self.commit_mu.unlock(self.io);
+        self.outstanding_mu.lockUncancelable(self.io);
+        self.outstanding -= 1;
+        self.outstanding_ready.broadcast(self.io);
+        self.outstanding_mu.unlock(self.io);
+    }
+
+    fn drain(self: *AsyncFlush) void {
+        self.outstanding_mu.lockUncancelable(self.io);
+        defer self.outstanding_mu.unlock(self.io);
+        while (self.outstanding != 0)
+            self.outstanding_ready.waitUncancelable(self.io, &self.outstanding_mu);
+    }
+};
+
 pub const Archive = struct {
     allocator: Allocator,
     io: Io,
@@ -74,6 +215,9 @@ pub const Archive = struct {
     on_durable_ctx: ?*anyopaque = null,
     on_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void = null,
     stats: ?*metrics.Stats = null,
+    /// Bootstrap-only ordered compression pipeline. Steady-state writes stay
+    /// synchronous, matching upstream's BackfillAsyncFlushWorkers scope.
+    async_flush: ?AsyncFlush = null,
     /// A failed append/flush may leave a partial in-memory transaction. Never
     /// attempt a graceful flush afterward; recovery trusts only complete
     /// frames already on disk and truncates a torn suffix.
@@ -111,12 +255,43 @@ pub const Archive = struct {
     }
 
     pub fn deinit(self: *Archive) void {
+        if (self.async_flush) |*pipeline| {
+            pipeline.admission_mu.lockUncancelable(self.io);
+            pipeline.drain();
+            pipeline.deinit();
+            pipeline.admission_mu.unlock(self.io);
+            self.async_flush = null;
+        }
         self.writer.deinit();
         self.file.close(self.io);
         self.manifest.deinit();
         self.dir.close(self.io);
     }
 
+    /// Enable upstream's bootstrap-only async compression path. A zero worker
+    /// count is the explicit synchronous mode.
+    pub fn enableAsyncFlush(self: *Archive, worker_count: usize) !void {
+        if (worker_count == 0) return;
+        if (self.async_flush != null) return error.AsyncFlushAlreadyEnabled;
+        self.async_flush = try AsyncFlush.init(self.allocator, self.io, worker_count);
+        errdefer {
+            self.async_flush.?.deinit();
+            self.async_flush = null;
+        }
+        try self.async_flush.?.start(worker_count);
+    }
+
+    /// Drain and remove the bootstrap pipeline before merge/steady state.
+    pub fn disableAsyncFlush(self: *Archive) void {
+        if (self.async_flush) |*pipeline| {
+            pipeline.admission_mu.lockUncancelable(self.io);
+            pipeline.drain();
+            pipeline.deinit();
+            pipeline.admission_mu.unlock(self.io);
+            self.async_flush = null;
+        }
+    }
+
     /// scan the directory (the manifest), find the next index and seq, and
     /// finish any interrupted active segment: walk its frames, truncate the
     /// torn tail, seal in place.
@@ -237,6 +412,18 @@ pub const Archive = struct {
         before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
     ) !?u64 {
         if (events.len == 0) return null;
+        if (self.async_flush != null)
+            return @as(?u64, try self.appendBatchAsync(events, now_us, hook_ctx, before_durable));
+        return @as(?u64, try self.appendBatchSync(events, now_us, hook_ctx, before_durable));
+    }
+
+    fn appendBatchSync(
+        self: *Archive,
+        events: []segment.Event,
+        now_us: i64,
+        hook_ctx: ?*anyopaque,
+        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
+    ) !u64 {
         self.mu.lockUncancelable(self.io);
         defer self.mu.unlock(self.io);
         errdefer self.poisoned = true;
@@ -250,13 +437,115 @@ pub const Archive = struct {
             try self.writer.flush(); // partial block: durability over block size
         }
         if (self.writer.bytes().len > self.file_written) {
-            try self.syncToDisk();
+            try self.syncToDisk(null);
             self.last_sync_us = now_us;
             if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLocked();
         }
         return first_seq;
     }
 
+    fn appendBatchAsync(
+        self: *Archive,
+        events: []segment.Event,
+        now_us: i64,
+        hook_ctx: ?*anyopaque,
+        before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void,
+    ) !u64 {
+        const pipeline = &self.async_flush.?;
+        pipeline.admission_mu.lockUncancelable(self.io);
+        var admission_locked = true;
+        defer if (admission_locked) pipeline.admission_mu.unlock(self.io);
+
+        var prepared: std.ArrayList(writer_mod.PreparedBlock) = .empty;
+        defer {
+            for (prepared.items) |*block| block.deinit(self.allocator);
+            prepared.deinit(self.allocator);
+        }
+
+        self.mu.lockUncancelable(self.io);
+        var writer_locked = true;
+        defer if (writer_locked) self.mu.unlock(self.io);
+        errdefer self.poisoned = true;
+        if (self.poisoned) return error.ArchivePoisoned;
+        const first_seq = self.next_seq;
+        for (events) |*ev| {
+            if (try self.appendStampedPreparedLocked(ev)) |block|
+                try prepared.append(self.allocator, block);
+        }
+        if (before_durable) |hook| try hook(hook_ctx.?, first_seq, self.next_seq - 1);
+        if (self.writer.pending.items.len > 0 and now_us - self.last_sync_us >= self.flush_interval_us) {
+            if (try self.writer.prepareFlush()) |block| try prepared.append(self.allocator, block);
+        }
+
+        const jobs = try self.allocator.alloc(AsyncFlushJob, prepared.items.len);
+        defer self.allocator.free(jobs);
+        for (jobs, prepared.items) |*job, *block| {
+            job.* = .{ .prepared = block };
+            pipeline.assign(job);
+        }
+        self.mu.unlock(self.io);
+        writer_locked = false;
+
+        for (jobs) |*job| pipeline.submit(job);
+        pipeline.admission_mu.unlock(self.io);
+        admission_locked = false;
+
+        try self.finishAsyncJobs(jobs);
+        if (jobs.len > 0) {
+            self.mu.lockUncancelable(self.io);
+            self.last_sync_us = now_us;
+            self.mu.unlock(self.io);
+        }
+        try self.rotateIfFullAsync();
+        return first_seq;
+    }
+
+    fn finishAsyncJobs(self: *Archive, jobs: []AsyncFlushJob) !void {
+        const pipeline = &self.async_flush.?;
+        var first_error: ?anyerror = null;
+        for (jobs) |*job| {
+            job.compressed.waitUncancelable(self.io);
+            pipeline.waitTurn(job.id);
+            self.mu.lockUncancelable(self.io);
+            if (job.failure) |err| {
+                self.poisoned = true;
+                if (first_error == null) first_error = err;
+            } else if (self.poisoned) {
+                if (first_error == null) first_error = error.ArchivePoisoned;
+            } else {
+                self.writer.commitPrepared(job.frame.?) catch |err| {
+                    self.poisoned = true;
+                    if (first_error == null) first_error = err;
+                };
+                if (!self.poisoned) self.syncToDisk(job.prepared.max_seq) catch |err| {
+                    self.poisoned = true;
+                    if (first_error == null) first_error = err;
+                };
+            }
+            self.mu.unlock(self.io);
+            if (job.frame) |frame| self.allocator.free(frame);
+            pipeline.finishTurn();
+        }
+        if (first_error) |err| return err;
+    }
+
+    fn rotateIfFullAsync(self: *Archive) !void {
+        const pipeline = &self.async_flush.?;
+        self.mu.lockUncancelable(self.io);
+        const full = self.writer.bytes().len >= self.max_segment_bytes;
+        self.mu.unlock(self.io);
+        if (!full) return;
+
+        pipeline.admission_mu.lockUncancelable(self.io);
+        defer pipeline.admission_mu.unlock(self.io);
+        pipeline.drain();
+        self.mu.lockUncancelable(self.io);
+        defer self.mu.unlock(self.io);
+        errdefer self.poisoned = true;
+        if (self.poisoned) return error.ArchivePoisoned;
+        if (self.writer.bytes().len >= self.max_segment_bytes) try self.rotateLocked();
+    }
+
     fn appendStampedLocked(self: *Archive, event: *segment.Event) !u64 {
         if (self.stamper) |stamp| stamp(self.stamper_ctx.?, event) catch |err| {
             if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
@@ -274,6 +563,23 @@ pub const Archive = struct {
         return seq;
     }
 
+    fn appendStampedPreparedLocked(self: *Archive, event: *segment.Event) !?writer_mod.PreparedBlock {
+        if (self.stamper) |stamp| stamp(self.stamper_ctx.?, event) catch |err| {
+            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
+            return err;
+        };
+        const seq = self.next_seq;
+        event.seq = seq;
+        const prepared = self.writer.appendPrepared(event.*) catch |err| {
+            if (self.stats) |stats| _ = stats.ingest_append_errors_total.fetchAdd(1, .monotonic);
+            return err;
+        };
+        self.next_seq += 1;
+        if (self.stats) |stats| _ = stats.ingest_events_appended_total.fetchAdd(1, .monotonic);
+        if (self.on_append) |hook| try hook(self.on_append_ctx.?, event.*);
+        return prepared;
+    }
+
     pub fn lastSeq(self: *Archive) u64 {
         self.mu.lockUncancelable(self.io);
         defer self.mu.unlock(self.io);
@@ -312,18 +618,43 @@ pub const Archive = struct {
 
     /// compress + write any pending block, fsync, advance the watermark.
     pub fn flushBlock(self: *Archive) !void {
+        if (self.async_flush != null) return self.flushBlockAsync();
         self.mu.lockUncancelable(self.io);
         defer self.mu.unlock(self.io);
         errdefer self.poisoned = true;
         try self.flushBlockLocked();
     }
 
+    fn flushBlockAsync(self: *Archive) !void {
+        const pipeline = &self.async_flush.?;
+        pipeline.admission_mu.lockUncancelable(self.io);
+        var admission_locked = true;
+        defer if (admission_locked) pipeline.admission_mu.unlock(self.io);
+
+        self.mu.lockUncancelable(self.io);
+        var writer_locked = true;
+        defer if (writer_locked) self.mu.unlock(self.io);
+        errdefer self.poisoned = true;
+        if (self.poisoned) return error.ArchivePoisoned;
+        var prepared = (try self.writer.prepareFlush()) orelse return;
+        defer prepared.deinit(self.allocator);
+        var job: AsyncFlushJob = .{ .prepared = &prepared };
+        pipeline.assign(&job);
+        self.mu.unlock(self.io);
+        writer_locked = false;
+
+        pipeline.submit(&job);
+        pipeline.admission_mu.unlock(self.io);
+        admission_locked = false;
+        try self.finishAsyncJobs((&job)[0..1]);
+    }
+
     fn flushBlockLocked(self: *Archive) !void {
         try self.writer.flush();
-        if (self.writer.bytes().len > self.file_written) try self.syncToDisk();
+        if (self.writer.bytes().len > self.file_written) try self.syncToDisk(null);
     }
 
-    fn syncToDisk(self: *Archive) !void {
+    fn syncToDisk(self: *Archive, async_durable_seq: ?u64) !void {
         const all = self.writer.bytes();
         try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
         try segment_io.sync(&self.file, self.io);
@@ -334,7 +665,7 @@ pub const Archive = struct {
         }
         self.synced_block_count = self.writer.block_count;
         // everything up to (not including) pending is now durable
-        const durable_seq = self.next_seq - 1 - self.writer.pending.items.len;
+        const durable_seq = async_durable_seq orelse self.next_seq - 1 - self.writer.pending.items.len;
         if (self.on_durable) |hook| try hook(self.on_durable_ctx.?, durable_seq, durable_seq + 1);
         // Advance only after metadata accepts the same boundary. If the hook
         // fails, the next flush safely rewrites and re-syncs this suffix.
@@ -345,6 +676,11 @@ pub const Archive = struct {
     /// seal the active segment (footer + finalized header) and open the
     /// next. safe from another thread (the compactor's force-rotate).
     pub fn rotate(self: *Archive) !void {
+        if (self.async_flush) |*pipeline| {
+            pipeline.admission_mu.lockUncancelable(self.io);
+            defer pipeline.admission_mu.unlock(self.io);
+            pipeline.drain();
+        }
         self.mu.lockUncancelable(self.io);
         defer self.mu.unlock(self.io);
         errdefer self.poisoned = true;
@@ -428,6 +764,93 @@ pub const parseSegmentIndex = manifest_mod.parseSegmentIndex;
 
 const testing = std.testing;
 
+test "async flush pipeline performs four real zstd compressions concurrently" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    const worker_count = 4;
+    const block_bytes = 8 * 1024 * 1024;
+
+    var pipeline = try AsyncFlush.init(testing.allocator, io, worker_count);
+    defer pipeline.deinit();
+    try pipeline.start(worker_count);
+
+    var prepared: [worker_count]writer_mod.PreparedBlock = undefined;
+    defer for (&prepared) |*block| block.deinit(testing.allocator);
+    var jobs: [worker_count]AsyncFlushJob = undefined;
+    var prng = std.Random.DefaultPrng.init(0xfeed_face_dead_beef);
+    for (&prepared, 0..) |*block, i| {
+        const bytes = try testing.allocator.alloc(u8, block_bytes);
+        prng.random().bytes(bytes);
+        block.* = .{ .bytes = bytes, .max_seq = i + 1 };
+    }
+    for (&jobs, &prepared) |*job, *block| {
+        job.* = .{ .prepared = block };
+        pipeline.assign(job);
+        pipeline.submit(job);
+    }
+    for (&jobs) |*job| job.compressed.waitUncancelable(io);
+    try testing.expectEqual(@as(usize, worker_count), pipeline.workers_peak.load(.acquire));
+    for (&jobs) |*job| {
+        pipeline.waitTurn(job.id);
+        testing.allocator.free(job.frame.?);
+        pipeline.finishTurn();
+    }
+}
+
+test "async flush commits deterministic JSS blocks and durability in preparation order" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var sync_tmp = testing.tmpDir(.{});
+    defer sync_tmp.cleanup();
+    var async_tmp = testing.tmpDir(.{});
+    defer async_tmp.cleanup();
+    var sync_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+    var async_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+    const sync_path_len = try sync_tmp.dir.realPath(io, &sync_path_buf);
+    const async_path_len = try async_tmp.dir.realPath(io, &async_path_buf);
+
+    var synchronous = try Archive.init(testing.allocator, io, sync_path_buf[0..sync_path_len]);
+    defer synchronous.deinit();
+    synchronous.writer.max_events_per_block = 1;
+    synchronous.max_events_per_block = 1;
+    var sync_rows: [8]segment.Event = undefined;
+    for (&sync_rows, 0..) |*event, i| event.* = testEvent(i + 1);
+    _ = try synchronous.appendBatch(&sync_rows, 1);
+
+    var asynchronous = try Archive.init(testing.allocator, io, async_path_buf[0..async_path_len]);
+    defer asynchronous.deinit();
+    const DurableReceipt = struct {
+        values: [8]u64 = undefined,
+        len: usize = 0,
+
+        fn record(ctx: *anyopaque, durable_seq: u64, _: u64) anyerror!void {
+            const self: *@This() = @ptrCast(@alignCast(ctx));
+            if (self.len == self.values.len) return error.TooManyDurableBoundaries;
+            self.values[self.len] = durable_seq;
+            self.len += 1;
+        }
+    };
+    var receipt: DurableReceipt = .{};
+    asynchronous.on_durable_ctx = &receipt;
+    asynchronous.on_durable = DurableReceipt.record;
+    asynchronous.writer.max_events_per_block = 1;
+    asynchronous.max_events_per_block = 1;
+    try asynchronous.enableAsyncFlush(4);
+    var async_rows: [8]segment.Event = undefined;
+    for (&async_rows, 0..) |*event, i| event.* = testEvent(i + 1);
+    _ = try asynchronous.appendBatch(&async_rows, 1);
+
+    try testing.expectEqualSlices(u8, synchronous.writer.bytes(), asynchronous.writer.bytes());
+    try testing.expectEqual(@as(u64, 8), asynchronous.committed_seq.load(.acquire));
+    try testing.expectEqual(@as(u32, 8), asynchronous.writer.block_count);
+    try testing.expectEqual(@as(usize, 8), receipt.len);
+    for (receipt.values, 1..) |durable_seq, expected|
+        try testing.expectEqual(@as(u64, expected), durable_seq);
+    asynchronous.disableAsyncFlush();
+}
+
 fn testEvent(seq_hint: u64) segment.Event {
     return .{
         .seq = 0, // archive assigns
diff --git a/src/internal/bootstrap/byte_budget.zig b/src/internal/bootstrap/byte_budget.zig
index b29f556..054c497 100644
--- a/src/internal/bootstrap/byte_budget.zig
+++ b/src/internal/bootstrap/byte_budget.zig
@@ -1,10 +1,12 @@
 //! Blocking allocator-wide byte budget for bootstrap workers.
 //!
-//! CAR preparation and conversion allocate through `BudgetedAllocator`; HTTP
-//! bodies stream to disk and are memory-mapped. The backfill engine admits the
-//! upstream worker count for network I/O but serializes the prepared-repo
-//! lifetime, so one repository may use the whole shared limit without several
-//! partial indexes deadlocking each other.
+//! CAR preparation allocates through `BudgetedAllocator`; HTTP bodies stream
+//! to disk and are memory-mapped. Emission reuses the fully loaded MST and 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");
 
diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig
index 0ac143f..6357637 100644
--- a/src/internal/bootstrap/engine.zig
+++ b/src/internal/bootstrap/engine.zig
@@ -30,9 +30,10 @@ const Allocator = std.mem.Allocator;
 const log = std.log.scoped(.stream);
 
 const list_page_limit = 1000;
-// Preparation and conversion are serialized after parallel disk-backed
-// downloads, so a single repo may use the whole configured transient budget
-// without deadlocking on allocations retained by other partial indexes.
+// Preparation is serialized after parallel disk-backed downloads, so a single
+// repo may use the whole configured transient budget without deadlocking on
+// other partial indexes. The lock is released once completeness is proven;
+// prepared repositories then emit concurrently into the ordered archive.
 const transient_retries: u8 = 3;
 const rate_limit_retries: u8 = 20;
 const rate_limit_ceiling_s: u64 = 330;
@@ -1043,11 +1044,21 @@ fn downloadOne(
                 const Sink = struct {
                     archive: *archive_mod.Archive,
                     now_us: i64,
+                    batch: [1024]segment.Event = undefined,
+                    batch_len: usize = 0,
                     rows: u64 = 0,
 
+                    fn flush(self: *@This()) anyerror!void {
+                        if (self.batch_len == 0) return;
+                        _ = try self.archive.appendBatch(self.batch[0..self.batch_len], self.now_us);
+                        self.batch_len = 0;
+                    }
+
                     fn emit(self: *@This(), event: segment.Event) anyerror!void {
-                        _ = try self.archive.append(event, self.now_us);
+                        self.batch[self.batch_len] = event;
+                        self.batch_len += 1;
                         self.rows += 1;
+                        if (self.batch_len == self.batch.len) try self.flush();
                     }
                 };
                 var sink: Sink = .{ .archive = archive, .now_us = now_us };
@@ -1061,6 +1072,10 @@ fn downloadOne(
                     },
                     else => return failureResult(result_allocator, did, &final_host, attempts, started_us, .car, @errorName(err)),
                 };
+                sink.flush() catch |err| switch (err) {
+                    error.OutOfMemory, error.ArchiveAppendFailed => return err,
+                    else => return error.ArchiveAppendFailed,
+                };
                 if (instrumentation) |m| m.observeBackfillHandleRepo(@intCast(@max(
                     handle_started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(),
                     0,
diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig
index efdba7a..4b15efe 100644
--- a/src/internal/bootstrap/lifecycle.zig
+++ b/src/internal/bootstrap/lifecycle.zig
@@ -43,6 +43,7 @@ const log = std.log.scoped(.stream);
 /// bootstrap live capture keeps a small scratch tail: nothing subscribes
 /// to it (serving is gated), it only feeds the consumer plumbing
 const capture_tail_bytes: usize = 16 * 1024 * 1024;
+pub const default_async_flush_workers: usize = 4;
 
 pub const Options = struct {
     data_dir: []const u8,
@@ -54,6 +55,9 @@ pub const Options = struct {
     max_repos: usize = 0,
     workers: usize = engine.default_workers,
     batch_size: usize = engine.default_batch_size,
+    /// 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.
@@ -451,6 +455,13 @@ fn runBootstrap(
     try capture.waitConnected();
     log.info("bootstrap: live capture connected; starting repository backfill", .{});
 
+    try dst.enableAsyncFlush(opts.async_flush_workers);
+    if (opts.async_flush_workers > 0)
+        log.info("backfill: async block compression on ({d} workers)", .{opts.async_flush_workers})
+    else
+        log.info("backfill: async block compression off", .{});
+    var async_flush_enabled = opts.async_flush_workers > 0;
+    defer if (async_flush_enabled) dst.disableAsyncFlush();
     const bf = engine.run(allocator, io, .{
         .relay_http = opts.relay_http,
         .plc_url = opts.plc_url,
@@ -466,6 +477,8 @@ fn runBootstrap(
         _ = capture_future.cancel(io);
         return err;
     };
+    dst.disableAsyncFlush();
+    async_flush_enabled = false;
     log.info("backfill: {d} repos complete, {d} failed, {d} rows ({d} dropped)", .{
         bf.completed,
         bf.failed,
diff --git a/src/internal/bootstrap/prepared_fetch.zig b/src/internal/bootstrap/prepared_fetch.zig
index 2e301c6..1ca2dc5 100644
--- a/src/internal/bootstrap/prepared_fetch.zig
+++ b/src/internal/bootstrap/prepared_fetch.zig
@@ -15,11 +15,9 @@ const Allocator = std.mem.Allocator;
 pub const Prepared = struct {
     fetched: fetched_car.FetchedCar,
     repo: repos.PreparedRepo,
-    prepare_mu: ?*Io.Mutex = null,
 
     pub fn deinit(self: *Prepared, io: Io) void {
         self.fetched.deinit(io);
-        if (self.prepare_mu) |mu| mu.unlock(io);
         self.* = undefined;
     }
 };
@@ -82,9 +80,9 @@ pub fn fetch(
                 .ok => |*fetched| blk: {
                     errdefer fetched.deinit(self.io);
                     if (self.prepare_mu) |mu| try mu.lock(self.io);
-                    errdefer if (self.prepare_mu) |mu| mu.unlock(self.io);
+                    defer if (self.prepare_mu) |mu| mu.unlock(self.io);
                     const prepared = try repos.prepareRepo(self.parse_allocator, fetched.bytes);
-                    break :blk .{ .ok = .{ .fetched = fetched.*, .repo = prepared, .prepare_mu = self.prepare_mu } };
+                    break :blk .{ .ok = .{ .fetched = fetched.*, .repo = prepared } };
                 },
                 .rate_limited => |delay| .{ .rate_limited = delay },
                 .not_found => .not_found,
@@ -111,7 +109,7 @@ pub fn fetch(
     fetched_car.waitForTransfer(io, &task.done, &future, &progress) catch |wait_err| {
         // cancel() has joined the attempt. If CPU-bound preparation crossed
         // the deadline before observing cancellation, reclaim the completed
-        // outcome here, including its held serialization mutex.
+        // outcome here.
         if (task.result) |attempt| if (attempt) |value| {
             var owned = value;
             owned.deinit(io);
diff --git a/src/internal/segment_writer.zig b/src/internal/segment_writer.zig
index 123b42e..788aa09 100644
--- a/src/internal/segment_writer.zig
+++ b/src/internal/segment_writer.zig
@@ -27,6 +27,19 @@ pub const Error = error{
     InvalidEvent,
 };
 
+/// Uncompressed, self-contained block detached from an active writer. The
+/// pending-event arena can be reused immediately while compression happens on
+/// another worker.
+pub const PreparedBlock = struct {
+    bytes: []u8,
+    max_seq: u64,
+
+    pub fn deinit(self: *PreparedBlock, allocator: Allocator) void {
+        allocator.free(self.bytes);
+        self.* = undefined;
+    }
+};
+
 /// validate upstream's column-width limits; mirrors segment.ValidateEvent.
 /// spec-legal-but-unrepresentable fields (e.g. 256-byte rkeys) must be
 /// dropped by the caller with a distinct metric, never truncated.
@@ -126,6 +139,19 @@ pub const ActiveWriter = struct {
 
     /// stage one event; flushes a block when the threshold is reached.
     pub fn append(self: *ActiveWriter, ev: segment.Event) Error!void {
+        try self.stage(ev);
+        if (self.pending.items.len >= self.max_events_per_block) try self.flush();
+    }
+
+    /// Stage one event and detach a complete block without compressing it.
+    /// Used by the bootstrap-only async compression pipeline.
+    pub fn appendPrepared(self: *ActiveWriter, ev: segment.Event) Error!?PreparedBlock {
+        try self.stage(ev);
+        if (self.pending.items.len < self.max_events_per_block) return null;
+        return try self.prepareFlush();
+    }
+
+    fn stage(self: *ActiveWriter, ev: segment.Event) Error!void {
         try validateEvent(ev);
         const alloc = self.pending_arena.allocator();
         try self.pending.append(self.allocator, .{
@@ -140,28 +166,36 @@ pub const ActiveWriter = struct {
             .payload = try alloc.dupe(u8, ev.payload),
         });
         self.event_count += 1;
-        if (self.pending.items.len >= self.max_events_per_block) try self.flush();
     }
 
-    /// compress staged events into a frame and append it to the file image.
-    pub fn flush(self: *ActiveWriter) Error!void {
-        if (self.pending.items.len == 0) return;
+    /// Encode and detach staged rows. No borrowed event bytes escape: the
+    /// returned columnar block owns a complete copy.
+    pub fn prepareFlush(self: *ActiveWriter) Error!?PreparedBlock {
+        if (self.pending.items.len == 0) return null;
+        const max_seq = self.pending.items[self.pending.items.len - 1].seq;
         const block = try encodeBlock(self.allocator, self.pending.items);
-        defer self.allocator.free(block);
-        const frame = zstd.compress(self.allocator, block) catch |err| switch (err) {
-            error.OutOfMemory => return error.OutOfMemory,
-            else => return error.CompressFailed,
-        };
-        defer self.allocator.free(frame);
+        self.pending.clearRetainingCapacity();
+        _ = self.pending_arena.reset(.retain_capacity);
+        return .{ .bytes = block, .max_seq = max_seq };
+    }
 
+    /// Commit a previously compressed block in preparation order.
+    pub fn commitPrepared(self: *ActiveWriter, frame: []const u8) Error!void {
         var len_prefix: [8]u8 = undefined;
         std.mem.writeInt(u64, &len_prefix, frame.len, .little);
         try self.buf.appendSlice(self.allocator, &len_prefix);
         try self.buf.appendSlice(self.allocator, frame);
 
         self.block_count += 1;
-        self.pending.clearRetainingCapacity();
-        _ = self.pending_arena.reset(.retain_capacity);
+    }
+
+    /// compress staged events into a frame and append it to the file image.
+    pub fn flush(self: *ActiveWriter) Error!void {
+        var prepared = (try self.prepareFlush()) orelse return;
+        defer prepared.deinit(self.allocator);
+        const frame = try compressPrepared(self.allocator, &prepared);
+        defer self.allocator.free(frame);
+        try self.commitPrepared(frame);
     }
 
     /// the active-file bytes so far (valid jss active segment).
@@ -232,6 +266,13 @@ pub const ActiveWriter = struct {
     }
 };
 
+pub fn compressPrepared(allocator: Allocator, prepared: *const PreparedBlock) Error![]u8 {
+    return zstd.compress(allocator, prepared.bytes) catch |err| switch (err) {
+        error.OutOfMemory => return error.OutOfMemory,
+        else => return error.CompressFailed,
+    };
+}
+
 /// walk an active file's frames sequentially (no footer). returns decoded
 /// blocks one at a time via the iterator.
 pub const ActiveIterator = struct {
diff --git a/src/main.zig b/src/main.zig
index 251b44a..c8f4ad8 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -220,6 +220,7 @@ pub fn main(init: std.process.Init.Minimal) !void {
     var do_backfill = false;
     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";
@@ -265,6 +266,8 @@ pub fn main(init: std.process.Init.Minimal) !void {
         } else if (std.mem.startsWith(u8, arg, "--backfill-batch-size=")) {
             const configured = try std.fmt.parseInt(usize, arg["--backfill-batch-size=".len..], 10);
             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=")) {
@@ -532,6 +535,7 @@ pub fn main(init: std.process.Init.Minimal) !void {
             .selected_repos = backfill_repos,
             .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/tests/upstream_oracle/stream_differential_test.go b/tests/upstream_oracle/stream_differential_test.go
index 906c412..e90be45 100644
--- a/tests/upstream_oracle/stream_differential_test.go
+++ b/tests/upstream_oracle/stream_differential_test.go
@@ -15,6 +15,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -288,6 +289,13 @@ func TestStreamBootstrapControlsOracle(t *testing.T) {
 		} else {
 			require.Contains(t, logs.String(), "merge discovery: 0 post-bootstrap DIDs queued for retry")
 		}
+		if slices.Contains(selectionArgs, "--backfill-async-flush-workers=0") {
+			require.Contains(t, logs.String(), "backfill: async block compression off")
+		} else if slices.Contains(selectionArgs, "--backfill-async-flush-workers=2") {
+			require.Contains(t, logs.String(), "backfill: async block compression on (2 workers)")
+		} else {
+			require.Contains(t, logs.String(), "backfill: async block compression on (4 workers)")
+		}
 		return counter.calls.Load(), counter.eventLog()
 	}
 
@@ -335,6 +343,14 @@ func TestStreamBootstrapControlsOracle(t *testing.T) {
 		require.NotEqual(t, -1, secondList)
 		require.Less(t, secondList, firstGet, events)
 	})
+	t.Run("zero disables async block compression", func(t *testing.T) {
+		calls, _ := run(t, []string{"--backfill", "--skip-merge-discovery", "--backfill-async-flush-workers=0"}, true)
+		require.Equal(t, int64(2), calls)
+	})
+	t.Run("positive async block compression count is applied", func(t *testing.T) {
+		calls, _ := run(t, []string{"--backfill", "--skip-merge-discovery", "--backfill-async-flush-workers=2"}, true)
+		require.Equal(t, int64(2), calls)
+	})
 }
 
 func (b *synchronizedBuffer) Write(p []byte) (int, error) {