[feature](be) Add complete-block cache write handoff and admission

### What problem does this PR solve?

Issue Number: #67611 (tracking)

Related PR: #67292, #67579, #67612

Problem Summary: Expose complete-block submission, tracked owned-buffer handoff, and low-priority spare-capacity admission together as one cache-writer component. Reuse existing epoch checks and asynchronous disk writes. Background hole filling can pass a completed block without another data copy while normal foreground cache writes keep priority.

### Release note

Internal complete-block and zero-copy handoff for best-effort cache population.

### Check List (For Author)

- Test: Unit Test — component coverage is included with this change. The integrated source snapshot d5c427fe337 passed 198 ASAN tests from 11 suites via run-be-ut.sh -j100. This individual intermediate commit was not built separately; target-branch validation is recorded on the PR. Repository clang-format v16, build hygiene and diff checks passed on the integrated tree. No cluster or clang-tidy run.
- Behavior changed: Yes — introduces the opt-in component; this synchronization preserves existing defaults and IO/writeback semantics.
- Does this need documentation: Yes — component scope and current behavior are tracked in #67611 and the related PR descriptions.

be/src/io/cache/async_cache_write_manager.cpp

@@ -19,6 +19,7 @@
 
 #include <algorithm>
 #include <array>
+#include <cstring>
 #include <exception>
 #include <limits>
 #include <optional>
@@ -33,6 +34,7 @@
 #include "cpp/sync_point.h"
 #include "io/cache/async_cache_write_manager_metrics.h"
 #include "io/cache/block_file_cache.h"
+#include "io/cache/inflight_write_buffer_index.h"
 #include "runtime/thread_context.h"
 #include "util/countdown_latch.h"
 #include "util/defer_op.h"
@@ -430,7 +432,8 @@ Status AsyncCacheWriteManager::start() {
     return Status::OK();
 }
 
-bool AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task) {
+bool AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task,
+                                        AsyncCacheWriteAdmissionMode admission_mode) {
     task.validate();
     const int64_t submit_start_us = MonotonicMicros();
     Defer record_submit_latency {
@@ -469,7 +472,9 @@ bool AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task) {
         }
 
         const bool has_capacity = pending_bytes <= max_pending_bytes - task_buffer_bytes;
-        if (!has_capacity && _queue.empty()) {
+        if (!has_capacity &&
+            (admission_mode == AsyncCacheWriteAdmissionMode::REQUIRE_SPARE_CAPACITY ||
+             _queue.empty())) {
             _metrics->record_task_rejected(Metrics::RejectionReason::BACKPRESSURE);
             return false;
         }
@@ -493,6 +498,93 @@ bool AsyncCacheWriteManager::try_submit(AsyncCacheWriteTask task) {
     return true;
 }
 
+AsyncCacheWriteBlockSubmitResult AsyncCacheWriteManager::try_submit_block(
+        AsyncCacheWriteBlockRequest request) {
+    DORIS_CHECK(request.data.data != nullptr);
+    DORIS_CHECK(request.data.size > 0);
+    DORIS_CHECK(request.buffer_size > 0);
+    DORIS_CHECK(request.data.size <= request.buffer_size);
+    if (!check_write_epoch(request.write_epoch)) {
+        return AsyncCacheWriteBlockSubmitResult::STALE_EPOCH;
+    }
+
+    AsyncCacheWriteBufferPtr buffer;
+    if (!allocate_tracked_buffer(request.buffer_size, &buffer).ok()) {
+        return AsyncCacheWriteBlockSubmitResult::BUFFER_ALLOCATION_FAILED;
+    }
+    std::memcpy(buffer->data(), request.data.data, request.data.size);
+
+    return try_submit_owned_block(AsyncCacheWriteOwnedBlockRequest {
+            .cache_hash = request.cache_hash,
+            .file_offset = request.file_offset,
+            .write_size = request.data.size,
+            .buffer = std::move(buffer),
+            .admission_ctx = std::move(request.admission_ctx),
+            .write_epoch = std::move(request.write_epoch),
+            .inflight_index = request.inflight_index,
+    });
+}
+
+AsyncCacheWriteBlockSubmitResult AsyncCacheWriteManager::try_submit_owned_block(
+        AsyncCacheWriteOwnedBlockRequest request) {
+    DORIS_CHECK(request.buffer != nullptr);
+    DORIS_CHECK(request.write_size > 0);
+    DORIS_CHECK(request.write_size <= request.buffer->size());
+    if (!check_write_epoch(request.write_epoch)) {
+        return AsyncCacheWriteBlockSubmitResult::STALE_EPOCH;
+    }
+
+    const int64_t submit_ts_us = MonotonicMicros();
+    AsyncCacheWriteTask task {
+            .cache_hash = request.cache_hash,
+            .file_offset = request.file_offset,
+            .write_size = request.write_size,
+            .buffer = request.buffer,
+            .admission_ctx = std::move(request.admission_ctx),
+            .submit_ts_us = submit_ts_us,
+            .write_epoch = std::move(request.write_epoch),
+            .on_finalized = nullptr,
+    };
+    std::shared_ptr<InflightWriteBufferEntry> entry;
+    if (request.inflight_index != nullptr) {
+        entry = std::make_shared<InflightWriteBufferEntry>(request.buffer, request.file_offset,
+                                                           request.write_size, submit_ts_us);
+        TEST_SYNC_POINT_CALLBACK(
+                "CachedRemoteFileReader::_submit_async_write_tasks:before_inflight_insert", &task);
+        auto existing = request.inflight_index->insert_if_absent(request.cache_hash,
+                                                                 request.file_offset, entry);
+        if (existing != nullptr) {
+            return AsyncCacheWriteBlockSubmitResult::ALREADY_INFLIGHT;
+        }
+        task.on_finalized = [cache_hash = request.cache_hash, offset = request.file_offset,
+                             inflight_index = request.inflight_index,
+                             entry](const AsyncCacheWriteTask&) {
+            inflight_index->remove_if(cache_hash, offset, entry);
+        };
+    }
+
+    if (!try_submit(std::move(task), request.admission_mode)) {
+        if (entry != nullptr) {
+            request.inflight_index->remove_if(request.cache_hash, request.file_offset, entry);
+            request.inflight_index->record_backpressure_rollback();
+        }
+        return AsyncCacheWriteBlockSubmitResult::REJECTED;
+    }
+    return AsyncCacheWriteBlockSubmitResult::SUBMITTED;
+}
+
+bool AsyncCacheWriteManager::can_accept_without_eviction(size_t buffer_size) const {
+    DORIS_CHECK(buffer_size > 0);
+    std::lock_guard lock(_queue_mutex);
+    if (!_started.load(std::memory_order_acquire) || !_accepting.load(std::memory_order_acquire)) {
+        return false;
+    }
+    DORIS_CHECK(_task_buffer_size == 0 || _task_buffer_size == buffer_size);
+    const size_t max_pending_bytes = _options.load(std::memory_order_acquire)->max_pending_bytes;
+    const size_t pending_bytes = _pending_bytes.load(std::memory_order_relaxed);
+    return buffer_size <= max_pending_bytes && pending_bytes <= max_pending_bytes - buffer_size;
+}
+
 Status AsyncCacheWriteManager::allocate_tracked_buffer(size_t size,
                                                        AsyncCacheWriteBufferPtr* buffer) {
     DORIS_CHECK(buffer != nullptr);

be/src/io/cache/async_cache_write_manager.h

@@ -31,12 +31,14 @@
 #include "common/status.h"
 #include "io/cache/file_cache_common.h"
 #include "runtime/memory/mem_tracker_limiter.h"
+#include "util/slice.h"
 #include "util/threadpool.h"
 
 namespace doris::io {
 
 class BlockFileCache;
 class AsyncCacheWriteEpochRegistry;
+class InflightWriteBufferIndex;
 
 /// Cache admission attributes captured on the query thread and replayed by a write worker.
 struct CacheAdmissionContext {
@@ -106,6 +108,55 @@ struct AsyncCacheWriteEpoch {
     std::shared_ptr<AsyncCacheWriteEpochToken> key_token;
 };
 
+enum class AsyncCacheWriteBlockSubmitResult : uint8_t {
+    /// The manager accepted ownership of a complete block buffer.
+    SUBMITTED,
+    /// The cache invalidation fence changed before submission.
+    STALE_EPOCH,
+    /// The copy-based submission could not allocate its tracked block buffer.
+    BUFFER_ALLOCATION_FAILED,
+    /// The same cache block already has a buffer indexed by an accepted write task.
+    ALREADY_INFLIGHT,
+    /// The manager is unavailable or its admission policy found no queue capacity.
+    REJECTED,
+};
+
+enum class AsyncCacheWriteAdmissionMode : uint8_t {
+    /// Normal work may replace the oldest queued task when pending capacity is full.
+    ALLOW_OLDEST_QUEUED_EVICTION,
+    /// Lower-priority work is accepted only from currently unused pending capacity.
+    REQUIRE_SPARE_CAPACITY,
+};
+
+/// Transient source bytes and immutable metadata for one complete cache-block submission.
+/// `data` is copied before try_submit_block() returns. `buffer_size` is the fixed cache block
+/// allocation size; only the physical EOF block may provide a shorter data slice.
+struct AsyncCacheWriteBlockRequest {
+    UInt128Wrapper cache_hash;
+    size_t file_offset {0};
+    Slice data;
+    size_t buffer_size {0};
+    CacheAdmissionContext admission_ctx;
+    AsyncCacheWriteEpoch write_epoch;
+    /// Optional index whose lifetime must cover every accepted write task.
+    InflightWriteBufferIndex* inflight_index {nullptr};
+};
+
+/// Owned complete cache-block payload. The buffer must come from allocate_tracked_buffer() on the
+/// target manager and is transferred without another data copy.
+struct AsyncCacheWriteOwnedBlockRequest {
+    UInt128Wrapper cache_hash;
+    size_t file_offset {0};
+    size_t write_size {0};
+    AsyncCacheWriteBufferPtr buffer;
+    CacheAdmissionContext admission_ctx;
+    AsyncCacheWriteEpoch write_epoch;
+    AsyncCacheWriteAdmissionMode admission_mode {
+            AsyncCacheWriteAdmissionMode::ALLOW_OLDEST_QUEUED_EVICTION};
+    /// Optional index whose lifetime must cover every accepted write task.
+    InflightWriteBufferIndex* inflight_index {nullptr};
+};
+
 /// One cache-block write. The sole production submitter allocates every `buffer` with exactly
 /// `file_cache_each_block_size` bytes. `write_size` is the valid prefix starting at `file_offset`;
 /// only the physical EOF block may use less than the full buffer. `write_epoch` prevents a worker
@@ -163,15 +214,36 @@ public:
     Status start();
 
     /// Admit `task` into the memory-bounded FIFO without waiting for disk I/O. Because all tasks
-    /// have one fixed cache-block buffer capacity, a full queue displaces exactly one oldest queued
-    /// task. Active tasks are never displaced. After a runtime limit decrease, submissions continue
-    /// to replace the oldest queued task without increasing pending bytes, even while existing
-    /// pending bytes exceed the new limit. This call can briefly wait for the queue mutex and
-    /// finalizes a displaced task before returning.
+    /// have one fixed cache-block buffer capacity, normal admission to a full queue displaces
+    /// exactly one oldest queued task. REQUIRE_SPARE_CAPACITY instead rejects a full-queue
+    /// submission so lower-priority work cannot displace normal writes. Active tasks are never
+    /// displaced. After a runtime limit decrease, normal submissions continue to replace the
+    /// oldest queued task without increasing pending bytes, even while existing pending bytes
+    /// exceed the new limit. This call can briefly wait for the queue mutex and finalizes a
+    /// displaced task before returning.
     /// @return true if ownership was transferred to the queue; false when workers have not been
     /// started, during shutdown, or on backpressure. A rejected task's finalization callback is
     /// not invoked.
-    bool try_submit(AsyncCacheWriteTask task);
+    bool try_submit(AsyncCacheWriteTask task,
+                    AsyncCacheWriteAdmissionMode admission_mode =
+                            AsyncCacheWriteAdmissionMode::ALLOW_OLDEST_QUEUED_EVICTION);
+
+    /// Copy and submit one complete cache block through the same epoch, inflight-deduplication,
+    /// memory-accounting, and queue path used by remote cache-miss reads.
+    /// @return ALREADY_INFLIGHT when another accepted task already owns this block; in that case
+    /// the request data is not copied.
+    AsyncCacheWriteBlockSubmitResult try_submit_block(AsyncCacheWriteBlockRequest request);
+
+    /// Submit an already owned complete cache block without copying its payload. This is the
+    /// handoff used after background hole filling completes a tracked block buffer.
+    /// The manager retains the buffer after this call only when SUBMITTED is returned.
+    AsyncCacheWriteBlockSubmitResult try_submit_owned_block(
+            AsyncCacheWriteOwnedBlockRequest request);
+
+    /// Return whether a fixed-size task fits in currently unused pending capacity. The result is a
+    /// point-in-time scheduling hint: a concurrent foreground submission may consume the capacity
+    /// before the caller hands off its task.
+    bool can_accept_without_eviction(size_t buffer_size) const;
 
     /// Allocate `size` payload bytes charged to the manager tracker and return them in `buffer`.
     Status allocate_tracked_buffer(size_t size, AsyncCacheWriteBufferPtr* buffer);

be/src/io/cache/cached_remote_file_reader_async_write.cpp

@@ -503,62 +503,35 @@ void CachedRemoteFileReader::_submit_async_write_tasks(const AsyncReadPlan& plan
         DORIS_CHECK(read_block.range.size() <= cache_block_size);
         DORIS_CHECK(read_block.range.left >= remote_left);
         DORIS_CHECK(read_block.range.right <= remote_right);
-        if (!manager->check_write_epoch(plan.write_epoch)) {
-            ++stats.async_cache_write_drop_stale_epoch;
-            continue;
-        }
-
-        AsyncCacheWriteBufferPtr tracked_buffer;
-        Status status = manager->allocate_tracked_buffer(cache_block_size, &tracked_buffer);
-        if (!status.ok()) {
-            ++stats.async_cache_write_buffer_alloc_fail;
-            continue;
-        }
-        DORIS_CHECK(tracked_buffer->size() == cache_block_size);
-        DORIS_CHECK(read_block.range.size() <= tracked_buffer->size());
         const size_t remote_buffer_offset = read_block.range.left - remote_left;
         DORIS_CHECK(remote_buffer_offset <= remote_size);
         DORIS_CHECK(read_block.range.size() <= remote_size - remote_buffer_offset);
-        memcpy(tracked_buffer->data(), remote_buffer.get() + remote_buffer_offset,
-               read_block.range.size());
-
-        AsyncCacheWriteTask task {
+        AsyncCacheWriteBlockRequest request {
                 .cache_hash = _cache_hash,
                 .file_offset = read_block.range.left,
-                .write_size = read_block.range.size(),
-                .buffer = tracked_buffer,
+                .data = Slice(remote_buffer.get() + remote_buffer_offset, read_block.range.size()),
+                .buffer_size = cache_block_size,
                 .admission_ctx = admission_context,
-                .submit_ts_us = MonotonicMicros(),
                 .write_epoch = plan.write_epoch,
-                .on_finalized = nullptr,
+                .inflight_index = config::enable_async_file_cache_write_inflight_write_buffer_index
+                                          ? inflight_index
+                                          : nullptr,
         };
-        std::shared_ptr<InflightWriteBufferEntry> entry;
-        if (config::enable_async_file_cache_write_inflight_write_buffer_index) {
-            entry = std::make_shared<InflightWriteBufferEntry>(
-                    tracked_buffer, read_block.range.left, read_block.range.size(),
-                    task.submit_ts_us);
-            TEST_SYNC_POINT_CALLBACK(
-                    "CachedRemoteFileReader::_submit_async_write_tasks:before_inflight_insert",
-                    &task);
-            auto existing =
-                    inflight_index->insert_if_absent(_cache_hash, read_block.range.left, entry);
-            if (existing) {
-                continue;
-            }
-            task.on_finalized = [cache_hash = _cache_hash, offset = read_block.range.left,
-                                 inflight_index, entry](const AsyncCacheWriteTask&) {
-                inflight_index->remove_if(cache_hash, offset, entry);
-            };
-        }
-
-        if (!manager->try_submit(std::move(task))) {
-            if (entry) {
-                inflight_index->remove_if(_cache_hash, read_block.range.left, entry);
-                inflight_index->record_backpressure_rollback();
-            }
-            ++stats.async_cache_write_rejected;
-        } else {
+        switch (manager->try_submit_block(std::move(request))) {
+        case AsyncCacheWriteBlockSubmitResult::SUBMITTED:
             ++stats.async_cache_write_submitted;
+            break;
+        case AsyncCacheWriteBlockSubmitResult::STALE_EPOCH:
+            ++stats.async_cache_write_drop_stale_epoch;
+            break;
+        case AsyncCacheWriteBlockSubmitResult::BUFFER_ALLOCATION_FAILED:
+            ++stats.async_cache_write_buffer_alloc_fail;
+            break;
+        case AsyncCacheWriteBlockSubmitResult::REJECTED:
+            ++stats.async_cache_write_rejected;
+            break;
+        case AsyncCacheWriteBlockSubmitResult::ALREADY_INFLIGHT:
+            break;
         }
     }
 }

be/test/io/cache/async_cache_write_manager_test.cpp

@@ -30,6 +30,7 @@
 #include <mutex>
 #include <new>
 #include <string>
+#include <string_view>
 #include <thread>
 #include <vector>
 
@@ -173,6 +174,249 @@ private:
     bool _released {false};
 };
 
+TEST_F(AsyncCacheWriteManagerTest, SubmitBlockCopiesDeduplicatesAndRollsBackBackpressure) {
+    auto cache = create_cache("async_write_manager_submit_block");
+    auto* manager = cache->async_write_manager();
+    auto* index = cache->inflight_write_buffer_index();
+    ASSERT_NE(manager, nullptr);
+    ASSERT_NE(index, nullptr);
+    auto options = manager->options();
+    options.worker_count = 1;
+    options.max_pending_bytes = 4096;
+    ASSERT_TRUE(manager->update_options(options).ok());
+
+    OneShotSyncPointGate worker_gate;
+    auto* sync_point = SyncPoint::get_instance();
+    SyncPoint::CallbackGuard guard;
+    sync_point->set_call_back(
+            "AsyncCacheWriteManager::_persist_task:before_get_or_set",
+            [&](auto&&) { worker_gate.arrive_and_wait(); }, &guard);
+    sync_point->enable_processing();
+    Defer clear_sync_point {[&]() {
+        worker_gate.release();
+        sync_point->disable_processing();
+        sync_point->clear_all_call_backs();
+    }};
+
+    const auto hash = BlockFileCache::hash("submit_block");
+    std::string payload(4096, 'p');
+    const std::string expected = payload;
+    const auto make_request = [&](const UInt128Wrapper& request_hash) {
+        return AsyncCacheWriteBlockRequest {
+                .cache_hash = request_hash,
+                .file_offset = 0,
+                .data = Slice(payload),
+                .buffer_size = 4096,
+                .admission_ctx = {},
+                .write_epoch = manager->current_write_epoch(request_hash),
+                .inflight_index = index,
+        };
+    };
+
+    EXPECT_EQ(manager->try_submit_block(make_request(hash)),
+              AsyncCacheWriteBlockSubmitResult::SUBMITTED);
+    ASSERT_TRUE(worker_gate.wait_until_arrived());
+    payload.assign(payload.size(), 'x');
+    auto inflight = index->lookup(hash, 0);
+    ASSERT_NE(inflight, nullptr);
+    EXPECT_EQ(std::string_view(inflight->buffer->data(), inflight->buffer_size), expected);
+
+    EXPECT_EQ(manager->try_submit_block(make_request(hash)),
+              AsyncCacheWriteBlockSubmitResult::ALREADY_INFLIGHT);
+    const auto rejected_hash = BlockFileCache::hash("submit_block_rejected");
+    EXPECT_EQ(manager->try_submit_block(make_request(rejected_hash)),
+              AsyncCacheWriteBlockSubmitResult::REJECTED);
+    EXPECT_EQ(index->lookup(rejected_hash, 0), nullptr);
+
+    worker_gate.release();
+    for (int attempt = 0; attempt < 100 && index->lookup(hash, 0) != nullptr; ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    ASSERT_EQ(index->lookup(hash, 0), nullptr);
+    ReadStatistics read_stats;
+    CacheContext context;
+    context.stats = &read_stats;
+    FileBlocks blocks;
+    bool fully_covered = false;
+    ASSERT_TRUE(cache->get_downloaded_blocks_if_fully_covered(hash, 0, expected.size(), context,
+                                                              &blocks, &fully_covered)
+                        .ok());
+    ASSERT_TRUE(fully_covered);
+    ASSERT_EQ(blocks.size(), 1);
+    std::string actual(expected.size(), '\0');
+    ASSERT_TRUE(blocks.front()->read(Slice(actual), 0).ok());
+    EXPECT_EQ(actual, expected);
+}
+
+TEST_F(AsyncCacheWriteManagerTest, SubmitBlockRejectsStaleEpochAndAllocationFailure) {
+    auto cache = create_cache("async_write_manager_submit_block_failures");
+    auto* manager = cache->async_write_manager();
+    ASSERT_NE(manager, nullptr);
+    const auto hash = BlockFileCache::hash("submit_block_failures");
+    const std::string payload(4096, 'f');
+    auto stale_epoch = manager->current_write_epoch(hash);
+    manager->invalidate_pending_writes(hash);
+
+    AsyncCacheWriteBlockRequest stale_request {
+            .cache_hash = hash,
+            .file_offset = 0,
+            .data = Slice(payload),
+            .buffer_size = 4096,
+            .admission_ctx = {},
+            .write_epoch = std::move(stale_epoch),
+            .inflight_index = nullptr,
+    };
+    EXPECT_EQ(manager->try_submit_block(std::move(stale_request)),
+              AsyncCacheWriteBlockSubmitResult::STALE_EPOCH);
+
+    auto* sync_point = SyncPoint::get_instance();
+    SyncPoint::CallbackGuard guard;
+    sync_point->set_call_back(
+            "AsyncCacheWriteManager::allocate_tracked_buffer:inject_failure",
+            [](auto&& args) {
+                auto* status = try_any_cast<Status*>(args[0]);
+                *status = Status::MemoryAllocFailed("injected block allocation failure");
+            },
+            &guard);
+    sync_point->enable_processing();
+    Defer clear_sync_point {[&]() {
+        sync_point->disable_processing();
+        sync_point->clear_all_call_backs();
+    }};
+    AsyncCacheWriteBlockRequest allocation_request {
+            .cache_hash = hash,
+            .file_offset = 0,
+            .data = Slice(payload),
+            .buffer_size = 4096,
+            .admission_ctx = {},
+            .write_epoch = manager->current_write_epoch(hash),
+            .inflight_index = nullptr,
+    };
+    EXPECT_EQ(manager->try_submit_block(std::move(allocation_request)),
+              AsyncCacheWriteBlockSubmitResult::BUFFER_ALLOCATION_FAILED);
+}
+
+TEST_F(AsyncCacheWriteManagerTest, SubmitOwnedBlockPreservesBufferIdentity) {
+    auto cache = create_cache("async_write_manager_submit_owned_block");
+    auto* manager = cache->async_write_manager();
+    auto* index = cache->inflight_write_buffer_index();
+    ASSERT_NE(manager, nullptr);
+    ASSERT_NE(index, nullptr);
+
+    OneShotSyncPointGate worker_gate;
+    auto* sync_point = SyncPoint::get_instance();
+    SyncPoint::CallbackGuard guard;
+    sync_point->set_call_back(
+            "AsyncCacheWriteManager::_persist_task:before_get_or_set",
+            [&](auto&&) { worker_gate.arrive_and_wait(); }, &guard);
+    sync_point->enable_processing();
+    Defer clear_sync_point {[&]() {
+        worker_gate.release();
+        sync_point->disable_processing();
+        sync_point->clear_all_call_backs();
+    }};
+
+    const auto hash = BlockFileCache::hash("submit_owned_block");
+    AsyncCacheWriteBufferPtr buffer;
+    ASSERT_TRUE(manager->allocate_tracked_buffer(4096, &buffer).ok());
+    memset(buffer->data(), 'o', buffer->size());
+    const auto* buffer_identity = buffer.get();
+
+    EXPECT_EQ(manager->try_submit_owned_block(AsyncCacheWriteOwnedBlockRequest {
+                      .cache_hash = hash,
+                      .file_offset = 0,
+                      .write_size = 4096,
+                      .buffer = std::move(buffer),
+                      .admission_ctx = {},
+                      .write_epoch = manager->current_write_epoch(hash),
+                      .inflight_index = index,
+              }),
+              AsyncCacheWriteBlockSubmitResult::SUBMITTED);
+    EXPECT_EQ(buffer, nullptr);
+    ASSERT_TRUE(worker_gate.wait_until_arrived());
+
+    auto inflight = index->lookup(hash, 0);
+    ASSERT_NE(inflight, nullptr);
+    EXPECT_EQ(inflight->buffer.get(), buffer_identity);
+    EXPECT_EQ(std::string_view(inflight->buffer->data(), inflight->buffer_size),
+              std::string(4096, 'o'));
+
+    worker_gate.release();
+    for (int attempt = 0; attempt < 100 && index->lookup(hash, 0) != nullptr; ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    EXPECT_EQ(index->lookup(hash, 0), nullptr);
+}
+
+TEST_F(AsyncCacheWriteManagerTest, ReportsOnlyUnusedPendingCapacity) {
+    auto cache = create_cache("async_write_manager_spare_capacity");
+    auto* manager = cache->async_write_manager();
+    ASSERT_NE(manager, nullptr);
+    auto options = manager->options();
+    options.worker_count = 1;
+    options.max_pending_bytes = 8192;
+    ASSERT_TRUE(manager->update_options(options).ok());
+
+    OneShotSyncPointGate worker_gate;
+    auto* sync_point = SyncPoint::get_instance();
+    SyncPoint::CallbackGuard guard;
+    sync_point->set_call_back(
+            "AsyncCacheWriteManager::_persist_task:before_get_or_set",
+            [&](auto&&) { worker_gate.arrive_and_wait(); }, &guard);
+    sync_point->enable_processing();
+    Defer clear_sync_point {[&]() {
+        worker_gate.release();
+        sync_point->disable_processing();
+        sync_point->clear_all_call_backs();
+    }};
+
+    const auto submit = [&](std::string_view key) {
+        const auto hash = BlockFileCache::hash(std::string(key));
+        const std::string payload(4096, 'c');
+        return manager->try_submit_block(AsyncCacheWriteBlockRequest {
+                .cache_hash = hash,
+                .file_offset = 0,
+                .data = Slice(payload),
+                .buffer_size = 4096,
+                .admission_ctx = {},
+                .write_epoch = manager->current_write_epoch(hash),
+                .inflight_index = nullptr,
+        });
+    };
+
+    EXPECT_TRUE(manager->can_accept_without_eviction(4096));
+    EXPECT_EQ(submit("spare_capacity_active"), AsyncCacheWriteBlockSubmitResult::SUBMITTED);
+    ASSERT_TRUE(worker_gate.wait_until_arrived());
+    EXPECT_TRUE(manager->can_accept_without_eviction(4096));
+    EXPECT_EQ(submit("spare_capacity_queued"), AsyncCacheWriteBlockSubmitResult::SUBMITTED);
+    EXPECT_FALSE(manager->can_accept_without_eviction(4096));
+
+    const auto background_hash = BlockFileCache::hash("spare_capacity_background");
+    AsyncCacheWriteBufferPtr background_buffer;
+    ASSERT_TRUE(manager->allocate_tracked_buffer(4096, &background_buffer).ok());
+    memset(background_buffer->data(), 'b', background_buffer->size());
+    EXPECT_EQ(manager->try_submit_owned_block(AsyncCacheWriteOwnedBlockRequest {
+                      .cache_hash = background_hash,
+                      .file_offset = 0,
+                      .write_size = 4096,
+                      .buffer = std::move(background_buffer),
+                      .admission_ctx = {},
+                      .write_epoch = manager->current_write_epoch(background_hash),
+                      .admission_mode = AsyncCacheWriteAdmissionMode::REQUIRE_SPARE_CAPACITY,
+                      .inflight_index = nullptr,
+              }),
+              AsyncCacheWriteBlockSubmitResult::REJECTED);
+    EXPECT_EQ(manager->pending_count(), 2);
+
+    worker_gate.release();
+    for (int attempt = 0; attempt < 100 && manager->pending_count() != 0; ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    EXPECT_EQ(manager->pending_count(), 0);
+    EXPECT_TRUE(manager->can_accept_without_eviction(4096));
+    EXPECT_FALSE(is_cache_range_downloaded(cache.get(), background_hash));
+}
+
 TEST(AsyncCacheWriteConfigTest, ResolveMaxPendingBytes) {
     constexpr int64_t kMiB = 1024 * 1024;
     constexpr int64_t kGiB = 1024 * kMiB;