[RFC]: Refactor and Improve DistributedStorageBackend (and 3FS Adapter)

Summary

PR #2234 extracted 3FS logic from StorageBackend's #ifdef USE_3FS / is_3fs_dir_ branch into a first-class standalone DistributedStorageBackend with a FileSystemAdapter abstraction. This gave 3FS a clean architectural boundary — it is now a peer to BucketStorageBackend and OffsetAllocatorStorageBackend, with no more #ifdef pollution in the shared code path.

However, PR #2234 preserved the per-key file IO model (256 directories, one file per KV block) and the LOCAL_DISK replica semantics (client_id + transport_endpoint). These two assumptions mean 3FS is still not fully utilized:

• IO model: Per-key files create millions of files per bucket at production scale, causing metadata server pressure and expensive ScanMeta walks on restart. 3FS is designed for large shared files, not millions of tiny ones.
• Replica semantics: LocalDiskReplicaData couples data to a specific client process. When that client dies, CleanupStaleHandles erases the master-side metadata — even though the 3FS data is perfectly intact and globally accessible. Cross-node reads still RPC-proxy through the owner node instead of reading 3FS directly.

This mismatch causes three concrete problems:

1. Read amplification: Every cross-node read is a two-hop path (B → RPC → A → 3FS → A → TransferEngine → B) instead of a single pread on B.
2. Owner coupling: CleanupStaleHandles erases all LOCAL_DISK replicas when a client dies (see RFC for warm re-adoption), but 3FS data survives client death — the metadata is destroyed while the data is perfectly intact.
3. Dead code paths: submitFileReadOperation (transfer_task.cpp:1300) calls get_disk_descriptor() on a LocalDiskDescriptor and would throw; the only reason it hasn't crashed is that LOCAL_DISK replicas never reach the client-side read path — they go through batch_get_into_offload_object_internal RPC instead.

This RFC builds on PR #2234's structural foundation and completes the 3FS integration by replacing both the IO model and the replica semantics:

1. A new DISTRIBUTED_DISK replica type (DistributedDiskDescriptor + DistributedDiskReplicaData) that carries file_path (3FS absolute path) + offset + object_size — no client_id, no transport_endpoint. Data ownership is decoupled from any client process.
2. A Master-side global page allocator (BitmapPageAllocator) that coordinates 256 bucket files on 3FS, replacing the per-key file model with a large-file + page-offset model.
3. A client-side direct pread read path (ReadFromDistributedDisk) that bypasses RPC entirely — any node reads any key by pread-ing the 3FS bucket file at the Master-assigned offset.
4. Seven new RPCs for page lifecycle management (AllocateDistributedPage, BatchAllocateDistributedPage, FreeDistributedPage, BatchFreeDistributedPage, GetDistributedPageMapping, BatchGetDistributedPageMapping, NotifyDistributedDiskSuccess).
5. An independent promotion path for DISTRIBUTED_DISK → MEMORY that bypasses LocalDiskSegment and uses holder_id=UUID{} to allow any client to execute the promotion.

Scope — what this RFC does and does not cover:

Scenario	Today	With this RFC	Fixed?
Cross-node read of 3FS data	RPC proxy through owner node	Direct pread on any node	✅
Owner coupling (LOCAL_DISK client_id semantics on 3FS)	Client death erases metadata; data orphaned on 3FS	DISTRIBUTED_DISK has no owner; data survives any client restart	✅
Page allocation coordination	N/A (per-key files)	Master-side BitmapPageAllocator with snapshot HA persistence	✅
Promotion from 3FS to DRAM	Hardcoded LocalDiskSegment path	Independent DistributedPromotionTask path	✅
LOCAL_DISK warm re-adoption	Separate RFC (see References)	Composes; unaffected	➖
DRAM recovery across restart	Lost	Still lost (separate RFC track)	❌
Cross-instance reads during full client downtime	Requires RPC proxy	Requires 3FS mount on reading node (deployment prerequisite)	➖
Live migration of data between 3FS clusters	N/A	Not covered	❌

Motivation

The 3FS value proposition vs. current usage

3FS (HF3FS) provides a POSIX-compatible distributed filesystem mounted identically on every node in the cluster (e.g., /mnt/3fs/mooncake/). A file written by Node A is immediately readable by Node B via a local pread — no network RPC, no knowledge of which node wrote the data. This is the core architectural property that Mooncake should exploit for its offload tier.

The current code exploits none of it:

3FS core value:
  ┌─────────────────────────────────────────────┐
  │  Global shared namespace                     │
  │  /mnt/3fs/mooncake/bucket_00.data            │
  │  Any node can directly pread this path       │
  │  No RPC intermediary, no "which node owns it"│
  └─────────────────────────────────────────────┘

Current code usage:
  ┌─────────────────────────────────────────────┐
  │  Node A writes to 3FS                        │
  │  Master records: LOCAL_DISK(A, size, endpoint)│
  │  Node B wants to read → RPC to Node A        │
  │    → A reads from 3FS → A transfers to B     │
  │                                              │
  │  3FS is right next to B, yet B must go       │
  │  through A to get the data                   │
  └─────────────────────────────────────────────┘

IO model mismatch

The current DistributedStorageBackend uses 256 directories with per-key files — one file per KV cache block. This works for small experiments but has fundamental scalability issues at production scale:

Aspect	Per-key files (current)	Large file + page offset (this RFC)
File count	Millions of files per bucket	1 file per bucket (256 total)
open() cost	Per-read syscall	One-time per bucket (fd pool)
Metadata overhead	3FS metadata server hit per file	Single fstat at init
Delete semantics	unlink() (metadata-heavy)	Bitmap bit-clear (zero I/O)
ScanMeta on restart	Walk entire directory tree	Not needed (Master recovers from snapshot)

Goals

• G1 Introduce DISTRIBUTED_DISK replica type that decouples data ownership from client processes — no client_id, no transport_endpoint.
• G2 Enable any node to read any DISTRIBUTED_DISK key via direct pread on 3FS, with zero RPC intermediation.
• G3 Master coordinates page allocation globally via BitmapPageAllocator, persisted through snapshot HA.
• G4 Feature-flagged (page_mode_) rollout: default off, zero behavior change for existing deployments.
• G5 Full promotion path from DISTRIBUTED_DISK → MEMORY (DRAM) for hot data.
• G6 Compose cleanly with LOCAL_DISK warm re-adoption RFC, Client Graceful Shutdown work, and existing ScanMeta/ReRegister paths.

Non-Goals

• DRAM recovery across restart — tracked separately.
• Removal of LOCAL_DISK path — LOCAL_DISK remains for non-3FS deployments (local SSD without shared filesystem). This RFC adds a new path, it does not replace the old one.
• Async IO for ReadFromDistributedDisk — Phase 1 uses synchronous pread with fd pooling. Async IO (via fileread_pool_ or io_uring) is a follow-up optimization if benchmarking shows the sync path is insufficient.
• Live migration of data between 3FS clusters or from LOCAL_DISK to DISTRIBUTED_DISK.
• Cross-cluster replication — use replica_num >= 2 or a separate replication tier.

Background

Key code paths

Path	Location	Role
LocalDiskReplicaData definition	replica.h:176	Carries client_id, object_size, transport_endpoint
DiskReplicaData definition	replica.h:192	Carries file_path, object_size — no offset field
NotifyOffloadSuccess (hardcodes LOCAL_DISK)	master_service.cpp:3360	Creates LOCAL_DISK replica on offload completion
CleanupStaleHandles (erases on client death)	master_service.cpp:2530	Erases all replicas owned by dead client
submitFileReadOperation (throws for LOCAL_DISK)	transfer_task.cpp:1300	Calls get_disk_descriptor() which throws for LocalDiskDescriptor
FindFirstCompleteReplica (no type priority)	client_service.cpp:3746	Returns first COMPLETE replica, no type discrimination
SelectBestReplica (no DISTRIBUTED_DISK)	real_client.cpp:290	Python entry point; skips unknown types → nullptr
calculate_total_size / allocateSlices	client_buffer.cpp:134,149	Else branch calls get_memory_descriptor() → throws for unknown types
PushPromotionQueue (hardcoded LocalDiskSegment)	master_service.cpp:3427	Only promotes from LOCAL_DISK via LocalDiskSegment
PromotionObjectHeartbeat (only reads LocalDiskSegment)	master_service.cpp:3562	Returns SEGMENT_NOT_FOUND if no LocalDiskSegment, blocking DISTRIBUTED_DISK tasks
DistributedStorageBackend::Init (creates directories)	distributed_storage_backend.cpp:71	Creates 256 subdirectories, not files
DistributedStorageBackend::BatchOffload (per-key writes)	distributed_storage_backend.cpp:159	Writes one file per key
Snapshot-based HA (Master persistence)	master_service.cpp:4801	TryRestoreStateFromSnapshot — the only persistence channel
MetadataSerializer (snapshot segments)	catalog_backed_snapshot_provider.cpp	Serializes metadata_shards_
TaskManagerSerializer (snapshot tasks)	catalog_backed_snapshot_provider.cpp	Serializes promotion task queues

Why LOCAL_DISK is wrong for 3FS

LOCALDiskReplicaData encodes three assumptions that are false for 3FS:

1. client_id: "The data is owned by this specific process." — False. 3FS data is owned by the filesystem, not by any client process.
2. transport_endpoint: "Readers must connect to this RDMA endpoint to access the data." — False. Any node can pread the 3FS file directly.
3. Implicit lifetime coupling: "When the client dies, the data is gone." — False. 3FS data persists independently of client processes.

Using LOCAL_DISK for 3FS is an impedance mismatch that causes the three concrete problems listed in the Summary.

PR #2234: The structural foundation

PR #2234 (merged 2026-06-03) laid the structural groundwork for this RFC. It refactored the 3FS integration from a parasitic "file implementation switch" inside StorageBackend (with #ifdef USE_3FS and is_3fs_dir_ branching) into a first-class standalone DistributedStorageBackend peer to BucketStorageBackend / OffsetAllocatorStorageBackend. Key changes:

• Introduced DistributedStorageBackend as a new StorageBackendInterface implementation.
• Introduced FileSystemAdapter abstract interface with Hf3fsAdapter as the first implementation (USRBIO-based).
• Stripped all 3FS logic from StorageBackend (removed is_3fs_dir_, USRBIOResourceManager, ThreeFSFile, 3fs-virt path probing).
• Added StorageBackendType::kDistributed and wired it through CreateStorageBackend().
• Replaced fragile path heuristic (fs::exists("3fs-virt")) with explicit configuration-driven backend selection.

What PR #2234 did NOT do: it preserved the per-key file IO model. DistributedStorageBackend still creates 256 directories with one file per KV block, and still uses LOCAL_DISK replica semantics (with client_id and transport_endpoint). This RFC builds on PR #2234's architectural skeleton by replacing the IO model (per-key files → large-file + page offset) and the replica semantics (LOCAL_DISK → DISTRIBUTED_DISK).

Why DISK is not the answer either

DiskReplicaData has file_path and object_size but no offset field (replica.h:192). It is designed for per-key files where the entire file is one object. The page-based large-file model requires an offset into a shared bucket file, which DiskDescriptor cannot express. We need a new type.

Design

1. New Replica Type: DISTRIBUTED_DISK

1.1 Data structures

// Client-side descriptor (serialized in RPC, part of Replica::Descriptor)
struct DistributedDiskDescriptor {
    std::string file_path;     // "/mnt/3fs/mooncake/bucket_00.data" (3FS absolute path)
    int64_t offset = 0;        // byte offset = page_index × page_size
    uint64_t object_size = 0;  // actual data size
    YLT_REFL(DistributedDiskDescriptor, file_path, offset, object_size);
};

// Master-side replica data (internal, not serialized over wire)
struct DistributedDiskReplicaData {
    std::string file_path;
    int64_t offset = 0;
    uint64_t object_size = 0;
    // Note: NO client_id, NO transport_endpoint
};

Why not reuse DiskDescriptor? DiskDescriptor has no offset field. Adding one would change its wire format and break backward compatibility. A separate type is cleaner.

1.2 Variant extension

New types must be appended to the end of std::variant — YLT_REFL serializes by index, so inserting in the middle would shift indices and break deserialization of existing snapshots.

// Replica::data_ variant (replica.h)
std::variant<
    MemoryReplicaData,
    DiskReplicaData,
    LocalDiskReplicaData,
    NoFReplicaData,
    DistributedDiskReplicaData   // NEW: appended to end
> data_;

// Replica::Descriptor::descriptor_variant (replica.h)
std::variant<
    MemoryDescriptor,
    NoFDescriptor,
    DiskDescriptor,
    LocalDiskDescriptor,
    DistributedDiskDescriptor    // NEW: appended to end
> descriptor_variant;

1.3 New predicates and accessors

// On Replica
bool is_distributed_disk_replica() const {
    return std::holds_alternative<DistributedDiskReplicaData>(data_);
}
static bool fn_is_distributed_disk_replica(const Replica& r) {
    return r.is_distributed_disk_replica();
}
const DistributedDiskReplicaData& get_distributed_disk_data() const {
    return std::get<DistributedDiskReplicaData>(data_);
}

// On Replica::Descriptor
bool is_distributed_disk_replica() noexcept {
    return std::holds_alternative<DistributedDiskDescriptor>(descriptor_variant);
}
DistributedDiskDescriptor& get_distributed_disk_descriptor() {
    auto* desc = std::get_if<DistributedDiskDescriptor>(&descriptor_variant);
    if (!desc) throw std::runtime_error("Expected DistributedDiskDescriptor");
    return *desc;
}

1.4 New constructor

Replica(std::string file_path, int64_t offset, uint64_t object_size,
        ReplicaStatus status)
    : id_(next_id_.fetch_add(1)),
      data_(DistributedDiskReplicaData{std::move(file_path), offset, object_size}),
      status_(status), refcnt_(0) {
    MasterMetricManager::instance().inc_allocated_file_size(object_size);
}

2. Master-Side Global Page Allocation

2.1 BitmapPageAllocator

Each of the 256 buckets has its own allocator. The allocator uses a bitmap where each bit represents one page (default 64KB). A hint_ cursor enables amortized O(1) sequential allocation.

class BitmapPageAllocator {
public:
    int64_t page_size;       // default 64KB
    int64_t num_pages;       // bucket_size / page_size
    std::vector<uint64_t> bitmap;  // 1=allocated, 0=free
    int64_t hint_;           // search start position
    std::mutex mutex;

    void Init(int64_t ps, int64_t bucket_size);
    int64_t Allocate(int64_t count = 1);       // returns page_index, -1 on failure
    void Free(int64_t page_index, int64_t count = 1);
    void MarkAllocated(int64_t page_index, int64_t count = 1);  // for snapshot restore
    bool IsAllocated(int64_t page_index) const;
    int64_t AllocatedCount() const;
};

Allocate algorithm: Linear scan from hint_ for count consecutive free bits. Wrap around at end. Update hint_ past the allocation. Worst case O(num_pages), but hint_ makes sequential allocations O(1) amortized.

Free: Clear the bits, update hint_ if the freed range is before current hint (enables reuse of freed space without full scan).

2.2 Master data structures

class MasterService {
    static constexpr int kBucketCount = 256;
    std::array<BitmapPageAllocator, kBucketCount> bucket_allocators_;

    // Key: MakeTenantScopedStorageKey(tenant_id, key) = "tenant_id\0object_key"
    // Value: the descriptor (file_path, offset, size)
    std::unordered_map<std::string, DistributedDiskDescriptor> distributed_page_mappings_;
    std::shared_mutex distributed_mapping_mutex_;

    std::string distributed_root_dir_;  // "/mnt/3fs/mooncake"
    int64_t page_size_ = 64 << 10;     // 64KB
};

Tenant isolation: All Master methods carry tenant_id. The in-memory key is MakeTenantScopedStorageKey(tenant_id, key) (format: "tenant_id\0key", types.h:230), ensuring different tenants with the same key name map to different pages.

2.3 Bucket selection

Master and DistributedStorageBackend must use exactly the same formula to select bucket:

int bucket_id = XXH64(key.data(), key.size(), 0) % kBucketCount;

Note: key is the raw object key (without tenant prefix), not storage_key. Both sides use the raw key for XXH64. XXH64 is already a dependency (distributed_storage_backend.cpp:7, CMakeLists.txt:100-113).

2.4 Persistence via snapshot HA

Master has no KV persistence layer. The only persistence channel is snapshot-based HA (TryRestoreStateFromSnapshot, master_service.cpp:4801). Three serializers exist:

• SegmentSerializer → segments (client_local_disk_segment)
• MetadataSerializer → metadata (metadata_shards_)
• TaskManagerSerializer → tasks

distributed_page_mappings_ goes in MetadataSerializer (it is metadata). distributed_promotion_tasks goes in TaskManagerSerializer (it is a task queue).

Snapshot restore → bitmap rebuild:

// After restoring distributed_page_mappings_ from snapshot
for (auto& [key, desc] : distributed_page_mappings_) {
    int bucket_id = XXH64(/* extract raw key from storage_key */) % kBucketCount;
    int64_t page_index = desc.offset / page_size_;
    int64_t page_count = (desc.object_size + page_size_ - 1) / page_size_;
    bucket_allocators_[bucket_id].MarkAllocated(page_index, page_count);
}

PROCESSING replica cleanup on restore: When snapshot restore cleans up non-COMPLETE replicas (master_service.cpp:4976), it must also free the bitmap pages for any DISTRIBUTED_DISK PROCESSING replica:

if (r.is_distributed_disk_replica()) {
    const auto& data = r.get_distributed_disk_data();
    int bucket_id = ParseBucketIdFromPath(data.file_path);
    int64_t page_index = data.offset / page_size_;
    int64_t page_count = (data.object_size + page_size_ - 1) / page_size_;
    bucket_allocators_[bucket_id].Free(page_index, page_count);
}

Old snapshot compatibility: Restoring from a snapshot without distributed_page_mappings_ yields an empty map and zeroed bitmaps — functionally correct. Pre-existing 3FS data becomes orphaned (not recognized by the new code); operators should document this.

2.5 Seven new RPCs

Each RPC requires changes to 4 files: rpc_types.h (request/response structs), master_service.h/.cpp (implementation), rpc_service.h/.cpp (thin wrapper + registration), master_client.h/.cpp (client stub).

RPC	Purpose	Key detail
AllocateDistributedPage	Allocate one page	Returns DistributedDiskDescriptor
BatchAllocateDistributedPage	Allocate pages for N keys	Takes per-key tenant_ids vector (not a single tenant), because AllocateOffloadingBuckets groups by bucket across tenants
FreeDistributedPage	Free one page	Bitmap bit-clear
BatchFreeDistributedPage	Free pages for N keys	Used by DSB on write failure to prevent page leaks
GetDistributedPageMapping	Query one key's page mapping	Returns DistributedDiskDescriptor
BatchGetDistributedPageMapping	Batch query	Must be in Phase 1, not deferred — N keys × 1 RPC each is unacceptable for vLLM batch-get
NotifyDistributedDiskSuccess	Offload completion notification	Creates DISTRIBUTED_DISK replica + cleans up offloading_task + releases source refcnt

NotifyDistributedDiskSuccess implementation (critical correctness detail):

for (size_t i = 0; i < keys.size(); ++i) {
    auto object_id = MakeObjectIdentity(keys[i], tenant_ids[i]);

    // 1. Release source replica refcnt + clean up offloading_task
    //    (mirrors NotifyOffloadSuccess exactly, master_service.cpp:3330-3370)
    {
        MetadataAccessorRW accessor(this, object_id);
        if (accessor.Exists()) {
            auto& obj_metadata = accessor.Get();
            auto& tenant_state = accessor.GetTenantState();
            auto task_it = tenant_state.offloading_tasks.find(object_id.user_key);
            if (task_it != tenant_state.offloading_tasks.end()) {
                auto source = obj_metadata.GetReplicaByID(task_it->second.source_id);
                if (source != nullptr) source->dec_refcnt();
                tenant_state.offloading_tasks.erase(task_it);
            }
        }
    }

    // 2. Create DISTRIBUTED_DISK replica
    Replica replica(descriptors[i].file_path, descriptors[i].offset,
                    descriptors[i].object_size, ReplicaStatus::COMPLETE);
    auto res = AddReplica(client_id, keys[i], tenant_ids[i], replica);
    if (!res && res.error() != ErrorCode::OBJECT_NOT_FOUND) {
        return tl::make_unexpected(res.error());
    }
}

Omitting step 1 causes a permanent refcnt leak on the source replica — it can never be evicted or deleted.

2.6 Master::Remove for DISTRIBUTED_DISK

Current Remove (master_service.cpp:2833) just calls accessor.Erase(). For DISTRIBUTED_DISK, we must free the bitmap pages first:

metadata.VisitReplicas(
    Replica::fn_is_distributed_disk_replica,
    [&](const Replica& r) {
        const auto& data = r.get_distributed_disk_data();
        int bucket_id = XXH64(key.data(), key.size(), 0) % kBucketCount;
        int64_t page_index = data.offset / page_size_;
        int64_t page_count = (data.object_size + page_size_ - 1) / page_size_;
        bucket_allocators_[bucket_id].Free(page_index, page_count);
    });
// Then erase the in-memory mapping (next snapshot save naturally omits it)
auto storage_key = MakeTenantScopedStorageKey(tenant_id, key);
{
    std::unique_lock lock(distributed_mapping_mutex_);
    distributed_page_mappings_.erase(storage_key);
}

3. Promotion: DISTRIBUTED_DISK → MEMORY

Promotion (warming hot data from SSD to DRAM) today is hardcoded to LOCAL_DISK: TryPushPromotionQueue finds a LOCAL_DISK source, pins it with inc_refcnt(), and pushes to LocalDiskSegment::promotion_objects. DISTRIBUTED_DISK replicas have no client_id and no LocalDiskSegment, so this path fails.

3.1 Independent promotion queue

struct TenantState {
    // ... existing fields ...
    std::unordered_map<std::string, PromotionTaskItem> promotion_tasks;       // existing
    std::unordered_map<std::string, DistributedPromotionTaskItem>
        distributed_promotion_tasks;                                          // NEW
    std::mutex distributed_promotion_mutex;                                   // NEW
};

struct DistributedPromotionTaskItem {
    std::string tenant_id;
    std::string key;
    int64_t size;
    std::string file_path;   // 3FS bucket file path (pre-filled)
    int64_t offset;          // page offset (pre-filled)
};

3.2 PushDistributedDiskPromotionQueue

Called from TryPushPromotionQueue when a DISTRIBUTED_DISK source is found (after the existing LOCAL_DISK path):

tl::expected<void, ErrorCode> MasterService::PushDistributedDiskPromotionQueue(
    const ObjectIdentity& object_id, Replica& source_replica) {
    auto& tenant_state = GetTenantState(object_id.tenant_id);
    const auto& data = source_replica.get_distributed_disk_data();

    // 1. Record in-flight task (reuse PromotionTask)
    //    holder_id = UUID{} (all-zero) = "any client can complete this"
    tenant_state.promotion_tasks.emplace(
        object_id.user_key,
        PromotionTask{
            .source_id = source_replica.id(),
            .alloc_id = 0,
            .object_size = data.object_size,
            .start_time = std::chrono::system_clock::now(),
            .holder_id = UUID{},  // all-zero = DISTRIBUTED_DISK
        });
    promotion_in_flight_.fetch_add(1, std::memory_order_relaxed);

    // 2. Push to heartbeat consumption queue
    tenant_state.distributed_promotion_tasks.emplace(
        MakeTenantScopedStorageKey(object_id.tenant_id, object_id.user_key),
        DistributedPromotionTaskItem{
            .tenant_id = object_id.tenant_id,
            .key = object_id.user_key,
            .size = static_cast<int64_t>(data.object_size),
            .file_path = data.file_path,   // pre-filled
            .offset = data.offset,
        });
    return {};
}

3.3 PromotionObjectHeartbeat modification

Current code (master_service.cpp:3562) only reads from LocalDiskSegment::promotion_objects and returns SEGMENT_NOT_FOUND if no segment exists — blocking DISTRIBUTED_DISK tasks.

auto MasterService::PromotionObjectHeartbeat(const UUID& client_id)
    -> tl::expected<std::vector<PromotionTaskItem>, ErrorCode> {
    // ... existing lock setup ...
    std::vector<PromotionTaskItem> result;

    // 1. LOCAL_DISK tasks (existing, but guard with if-exists)
    {
        auto it = client_local_disk_segment.find(client_id);
        if (it != client_local_disk_segment.end()) {
            // ... existing extraction logic ...
        }
        // If not found: skip, do NOT return SEGMENT_NOT_FOUND
    }

    // 2. DISTRIBUTED_DISK tasks (NEW) — any client can execute
    for (auto& shard : metadata_shards_) {
        std::shared_lock shard_lock(shard.mutex);
        for (auto& [tid, tenant_state] : shard.tenants) {
            std::unique_lock lock(tenant_state.distributed_promotion_mutex);
            while (result.size() < promotion_max_per_heartbeat_ &&
                   !tenant_state.distributed_promotion_tasks.empty()) {
                auto node = tenant_state.distributed_promotion_tasks.extract(
                    tenant_state.distributed_promotion_tasks.begin());
                result.push_back(PromotionTaskItem{
                    .tenant_id = node.mapped().tenant_id,
                    .key = node.mapped().key,
                    .size = node.mapped().size,
                });
            }
        }
    }
    return result;  // may be empty, but never SEGMENT_NOT_FOUND
}

3.4 NotifyPromotionSuccess modification

Three changes needed:

1. holder_id gate bypass: DISTRIBUTED_DISK tasks use holder_id=UUID{} (all-zero). The check holder_id != client_id must allow all-zero to pass.
2. staged->mark_complete(): Both LOCAL_DISK and DISTRIBUTED_DISK promotions create a PROCESSING MEMORY replica via PromotionAllocStart. It must be marked COMPLETE — skipping this makes the replica invisible to readers.
3. Cleanup: Erase from both promotion_tasks and distributed_promotion_tasks. Only clean LocalDiskSegment::promotion_objects for non-zero holder_id (LOCAL_DISK path).

auto task_it = tenant_state.promotion_tasks.find(object_id.user_key);
if (task_it == tenant_state.promotion_tasks.end() ||
    task_it->second.alloc_id == 0) {
    // alloc_id == 0 = PromotionAllocStart never completed = error
    return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY);
}

// holder_id all-zero (DISTRIBUTED_DISK) bypasses holder-only gate
if (task_it->second.holder_id != client_id &&
    !IsZeroUUID(task_it->second.holder_id)) {
    return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}

// Source dec_refcnt (both types need this)
auto* source = metadata.GetReplicaByID(task_it->second.source_id);
if (source != nullptr) source->dec_refcnt();

// Mark PROCESSING MEMORY replica as COMPLETE (both types need this!)
bool committed = false;
Replica* staged = metadata.GetReplicaByID(task_it->second.alloc_id);
if (staged != nullptr && staged->is_memory_replica() && staged->is_processing()) {
    staged->mark_complete();
    committed = true;
}

// Save fields BEFORE erase (avoid use-after-erase)
const auto holder_id = task_it->second.holder_id;
tenant_state.promotion_tasks.erase(task_it);
promotion_in_flight_.fetch_sub(1, std::memory_order_relaxed);

// Clean distributed_promotion_tasks (best-effort)
{
    std::unique_lock lock(tenant_state.distributed_promotion_mutex);
    tenant_state.distributed_promotion_tasks.erase(
        MakeTenantScopedStorageKey(object_id.tenant_id, object_id.user_key));
}

// Only clean LocalDiskSegment for LOCAL_DISK (non-zero holder_id)
if (!IsZeroUUID(holder_id)) {
    // ... existing LocalDiskSegment cleanup ...
}

if (!committed) return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY);
return {};

NotifyPromotionFailure follows the same pattern (symmetric).

4. DistributedStorageBackend Rewrite

4.1 Constructor change

DistributedStorageBackend(
    const FileStorageConfig& file_storage_config,
    const DistributedStorageConfig& distributed_config,
    std::unique_ptr<FileSystemAdapter> fs_adapter,
    std::shared_ptr<MasterClient> master_client);  // NEW parameter

CreateStorageBackend (storage_backend.cpp:3219) passes master_client through. The other backend types (file_per_key, bucket, offset_allocator) ignore this parameter.

Injection chain:

Client (holds master_client_)
  → FileStorage constructor passes master_client_
    → CreateStorageBackend(config, master_client)
      → DistributedStorageBackend(config, adapter, master_client)

4.2 Init: 256 large files instead of 256 directories

tl::expected<void, ErrorCode> DistributedStorageBackend::Init() {
    fs_adapter_->Init(root_dir_);

    if (page_mode_) {
        // NEW: Create 256 large bucket files
        for (int i = 0; i < kBucketCount; ++i) {
            std::string path = fmt::format("{}/bucket_{:02x}.data", root_dir_, i);
            int fd = open(path.c_str(), O_RDWR | O_CREAT, 0644);
            if (fd < 0) return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL);
            struct stat st;
            if (fstat(fd, &st) == 0 && st.st_size < bucket_size_) {
                if (ftruncate(fd, bucket_size_) != 0) {
                    close(fd);
                    return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL);
                }
            }
            close(fd);
        }
    } else {
        // EXISTING: 256 subdirectories (unchanged)
        for (int i = 0; i < hash_bucket_count_; ++i) {
            std::string bucket_dir = fmt::format("{}/{:02x}", root_dir_, i);
            std::error_code ec;
            std::filesystem::create_directories(bucket_dir, ec);
            if (ec) return tl::make_unexpected(ErrorCode::FILE_WRITE_FAIL);
        }
    }
    initialized_ = true;
    return {};
}

Key difference: old model GetBucketPath returns a directory; page mode returns a file (bucket_{:02x}.data). The two models are isolated by page_mode_.

4.3 BatchOffload rewrite

tl::expected<int64_t, ErrorCode> DistributedStorageBackend::BatchOffload(
    const std::unordered_map<std::string, std::vector<Slice>>& batch_object,
    complete_handler, eviction_handler) {

    // Parse storage_keys into (tenant_id, key) pairs
    std::vector<std::string> keys, tenant_ids, storage_keys;
    std::vector<int64_t> sizes;
    for (const auto& [storage_key, slices] : batch_object) {
        auto [tenant_id, key] = ParseTenantScopedStorageKey(storage_key);
        keys.push_back(key);
        tenant_ids.push_back(tenant_id);
        storage_keys.push_back(storage_key);
        int64_t total = 0;
        for (const auto& s : slices) total += s.size;
        sizes.push_back(total);
    }

    // Allocate pages from Master (per-key tenant_ids vector!)
    auto descriptors = master_client_->BatchAllocateDistributedPage(
        keys, tenant_ids, sizes);
    if (!descriptors) return tl::make_unexpected(descriptors.error());

    // Group by bucket and write
    std::unordered_map<int, std::vector<size_t>> bucket_keys;
    for (size_t i = 0; i < keys.size(); ++i) {
        int bucket_id = XXH64(keys[i].data(), keys[i].size(), 0) % kBucketCount;
        bucket_keys[bucket_id].push_back(i);
    }

    std::vector<std::string> success_storage_keys;
    std::vector<StorageObjectMetadata> success_metas;

    for (auto& [bucket_id, indices] : bucket_keys) {
        std::string bucket_path = GetBucketPath(bucket_id);
        for (size_t idx : indices) {
            auto& slices = batch_object.at(storage_keys[idx]);
            std::vector<iovec> iovs;
            for (const auto& s : slices) iovs.push_back({s.ptr, s.size});

            auto result = fs_adapter_->VectorWriteFile(
                bucket_path, iovs.data(), iovs.size(), (*descriptors)[idx].offset);
            if (result) {
                success_storage_keys.push_back(storage_keys[idx]);
                success_metas.push_back({bucket_id, (*descriptors)[idx].offset,
                                         (int64_t)keys[idx].size(), *result, ""});
            }
        }
    }

    // CRITICAL: Free pages for failed keys (prevent page leak)
    std::vector<std::string> failed_keys, failed_tenants;
    std::unordered_set<std::string> success_set(
        success_storage_keys.begin(), success_storage_keys.end());
    for (size_t i = 0; i < keys.size(); ++i) {
        if (!success_set.count(storage_keys[i])) {
            failed_keys.push_back(keys[i]);
            failed_tenants.push_back(tenant_ids[i]);
        }
    }
    if (!failed_keys.empty()) {
        master_client_->BatchFreeDistributedPage(failed_keys, failed_tenants);
    }

    if (!success_storage_keys.empty()) {
        complete_handler(success_storage_keys, success_metas);
    }
    return (int64_t)success_storage_keys.size();
}

4.4 BatchLoad rewrite

tl::expected<void, ErrorCode> DistributedStorageBackend::BatchLoad(
    std::unordered_map<std::string, Slice>& batched_slices) {

    for (auto& [storage_key, slice] : batched_slices) {
        auto [tenant_id, key] = ParseTenantScopedStorageKey(storage_key);
        auto desc = master_client_->GetDistributedPageMapping(key, tenant_id);
        if (!desc) return tl::make_unexpected(ErrorCode::KEY_NOT_FOUND);

        struct iovec iov = {slice.ptr, slice.size};
        auto result = fs_adapter_->VectorReadFile(desc->file_path, &iov, 1, desc->offset);
        if (!result) return tl::make_unexpected(result.error());
    }
    return {};
}

Note: Phase 1 uses per-key GetDistributedPageMapping calls. BatchGetDistributedPageMapping must be implemented in the same phase to avoid N serial RPCs for batch-get workloads.

4.5 ScanMeta adaptation

In page mode, ScanMeta returns empty. Master recovers distributed_page_mappings_ from its own snapshot — no client-side re-registration needed.

tl::expected<void, ErrorCode> DistributedStorageBackend::ScanMeta(...) {
    if (page_mode_) return {};  // Master self-recovers
    // ... existing logic for non-page mode ...
}

ReRegisterOffloadedObjects (file_storage.cpp:1026) naturally becomes a no-op when ScanMeta returns empty.

5. FileStorage Adaptation

5.1 complete_handler page-mode branch

auto complete_handler = [this, &task_by_storage_key](
    const std::vector<std::string>& keys,
    std::vector<StorageObjectMetadata>& metadatas) -> ErrorCode {

    if (storage_backend_->IsPageMode()) {
        std::vector<std::string> real_keys, tenant_ids;
        std::vector<DistributedDiskDescriptor> descriptors;

        for (size_t i = 0; i < keys.size(); ++i) {
            auto [tenant_id, key] = ParseTenantScopedStorageKey(keys[i]);
            real_keys.push_back(key);
            tenant_ids.push_back(tenant_id);
            descriptors.push_back({
                GetDistributedBucketPath(metadatas[i].bucket_id),
                metadatas[i].offset,
                (uint64_t)metadatas[i].data_size
            });
        }
        auto result = client_->NotifyDistributedDiskSuccess(
            real_keys, tenant_ids, descriptors);
        if (!result) return result.error();
        return ErrorCode::OK;
    }

    // Existing LOCAL_DISK path (unchanged)
    for (auto& metadata : metadatas) {
        metadata.transport_endpoint = local_rpc_addr_;
    }
    // ... NotifyOffloadSuccess ...
};

5.2 eviction_handler type fix

auto eviction_handler = [this](const std::vector<std::string>& keys,
                               const std::string& tenant_id) {
    if (storage_backend_->IsPageMode()) {
        return client_->BatchEvictDiskReplica(
            keys, tenant_id, ReplicaType::DISTRIBUTED_DISK);
    }
    return client_->BatchEvictDiskReplica(
        keys, tenant_id, ReplicaType::LOCAL_DISK);
};

5.3 IsPageMode() accessor

// StorageBackendInterface (new virtual method)
virtual bool IsPageMode() const { return false; }

// DistributedStorageBackend (override)
bool IsPageMode() const override { return page_mode_; }

6. Client Read Path: Direct pread to 3FS

6.1 FindFirstCompleteReplica type priority

Current code returns the first COMPLETE replica regardless of type. All 5 call sites (Get, Get+offset, BatchGet×2, function definition) are affected.

ErrorCode Client::FindFirstCompleteReplica(
    const std::vector<Replica::Descriptor>& replica_list,
    Replica::Descriptor& replica) {

    const Replica::Descriptor* best_memory = nullptr;
    const Replica::Descriptor* best_nof = nullptr;
    const Replica::Descriptor* best_distributed = nullptr;
    const Replica::Descriptor* best_local_disk = nullptr;
    const Replica::Descriptor* best_disk = nullptr;

    for (const auto& r : replica_list) {
        if (r.status != ReplicaStatus::COMPLETE) continue;
        if (r.is_memory_replica()) {
            if (!best_memory) best_memory = &r;
        } else if (r.is_nof_replica()) {
            if (!best_nof) best_nof = &r;
        } else if (r.is_distributed_disk_replica()) {
            if (!best_distributed) best_distributed = &r;
        } else if (r.is_local_disk_replica()) {
            if (!best_local_disk) best_local_disk = &r;
        } else if (r.is_disk_replica()) {
            if (!best_disk) best_disk = &r;
        }
    }

    const Replica::Descriptor* chosen =
        best_memory ? best_memory :
        best_nof ? best_nof :
        best_distributed ? best_distributed :
        best_local_disk ? best_local_disk :
        best_disk;
    if (chosen) { replica = *chosen; return ErrorCode::OK; }
    return ErrorCode::INVALID_REPLICA;
}

Priority: MEMORY → NOF → DISTRIBUTED_DISK → LOCAL_DISK → DISK

Rationale: 3FS direct pread (local filesystem call) is faster than RPC to a remote node's LOCAL_DISK endpoint.

6.2 ReadFromDistributedDisk

tl::expected<void, ErrorCode> Client::ReadFromDistributedDisk(
    const DistributedDiskDescriptor& desc,
    std::vector<Slice>& slices) {
    int fd = open(desc.file_path.c_str(), O_RDONLY | O_CLOEXEC);
    if (fd < 0) return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL);

    int64_t offset = desc.offset;
    for (auto& slice : slices) {
        ssize_t total = 0;
        while (total < (ssize_t)slice.size) {
            ssize_t ret = ::pread(fd, (char*)slice.ptr + total,
                                  slice.size - total, offset + total);
            if (ret < 0) {
                if (errno == EINTR) continue;
                close(fd);
                return tl::make_unexpected(ErrorCode::FILE_READ_FAIL);
            }
            if (ret == 0) break;
            total += ret;
        }
        if (total != (ssize_t)slice.size) {
            close(fd);
            return tl::make_unexpected(ErrorCode::FILE_READ_FAIL);
        }
        offset += slice.size;
    }
    close(fd);
    return {};
}

Why not go through TransferRead / submitFileReadOperation? That path calls get_disk_descriptor() which throws for non-DiskDescriptor types (transfer_task.cpp:1300). Direct pread is simpler, safer, and avoids the RDMA overhead entirely.

6.3 fd pool (mandatory)

Opening and closing a file per read is wasteful for batch workloads. A process-level fd cache for the 256 bucket files is mandatory:

class DistributedDiskReader {
public:
    int GetFd(const std::string& file_path) {
        std::lock_guard lock(mu_);
        auto it = fd_cache_.find(file_path);
        if (it != fd_cache_.end()) return it->second;
        int fd = open(file_path.c_str(), O_RDONLY | O_CLOEXEC);
        fd_cache_[file_path] = fd;
        return fd;
    }
private:
    std::mutex mu_;
    std::unordered_map<std::string, int> fd_cache_;
};

With 256 bucket files, the fd pool holds at most 256 file descriptors — well within typical ulimit settings.

7. RealClient (Python Entry Point) Adaptation

Python users (vLLM, sglang, etc.) go through RealClient, not Client directly. Three functions must support DISTRIBUTED_DISK before Client::Get is reached:

Function	Location	Without fix	With fix
SelectBestReplica	real_client.cpp:290	Skips DISTRIBUTED_DISK → returns nullptr	Recognizes and selects it
calculate_total_size	client_buffer.cpp:134	Falls to else → get_memory_descriptor() throws	New is_distributed_disk_replica() branch
allocateSlices	client_buffer.cpp:149	Falls to else → get_memory_descriptor() throws	New is_distributed_disk_replica() branch
execute_ranged_read	real_client.cpp	`is_disk_replica()

SelectBestReplica priority: local MEMORY → any MEMORY → local NOF → any NOF → DISTRIBUTED_DISK → LOCAL_DISK → DISK

DISTRIBUTED_DISK doesn't need local/remote distinction — every node sees the same 3FS path.

Configuration

# Feature flag (default off — zero behavior change for existing deployments)
MOONCAKE_DISTRIBUTED_PAGE_MODE=false

# 3FS mount point
MOONCAKE_DISTRIBUTED_ROOT_DIR=/mnt/3fs/mooncake

# Filesystem type
MOONCAKE_DISTRIBUTED_FS_TYPE=hf3fs

# Bucket count (256 = 1 byte hash)
MOONCAKE_DISTRIBUTED_BUCKET_COUNT=256

# Page size (64KB default)
MOONCAKE_DISTRIBUTED_PAGE_SIZE=65536

# Bucket file size (4GB default)
MOONCAKE_DISTRIBUTED_BUCKET_SIZE=4294967296

# Health check
MOONCAKE_DISTRIBUTED_HEALTH_CHECK=true

page_mode_ is read in DistributedStorageConfig::FromEnvironment() (existing env-parsing method). It flows into DistributedStorageBackend constructor and is exposed to FileStorage via StorageBackendInterface::IsPageMode().

Metrics

mooncake_master_distributed_pages_allocated gauge (per bucket)
mooncake_master_distributed_pages_free gauge (per bucket)
mooncake_master_distributed_page_allocations_total counter
mooncake_master_distributed_page_frees_total counter
mooncake_master_distributed_page_allocation_failures_total counter (NO_SPACE)
mooncake_dsb_batch_offload_pages_written counter
mooncake_dsb_batch_offload_page_leaks_prevented counter (failed-key frees)
mooncake_client_distributed_disk_reads_total counter
mooncake_client_distributed_disk_read_latency_seconds histogram

Implementation Plan

The work is split into 6 patches, each independently compilable and verifiable, with page_mode_ feature flag protecting all new paths.

P1: Replica types + BitmapPageAllocator

PR(#2441) by @fcczzz

Zero behavior change. Pure additions.

File	Change
include/replica.h	Add DistributedDiskDescriptor (with YLT_REFL), DistributedDiskReplicaData; append to data_ variant and Descriptor::descriptor_variant at end; add is_distributed_disk_replica() / get_distributed_disk_data() / constructor
include/bitmap_page_allocator.h	New file: BitmapPageAllocator class
src/bitmap_page_allocator.cpp	New file: implementation
tests/test_bitmap_page_allocator.cpp	New file: unit tests

Verification: variant append doesn't break old snapshot deserialization; allocator correctness; all existing tests green.

P2: Master metadata + all RPCs + snapshot

Zero behavior change. New Master capabilities. New RPCs exist but have no callers yet.

7 new RPCs × 4 files each = 28 file changes, plus master_service.h/.cpp for business logic and snapshot integration.

Verification: AllocateDistributedPage → bitmap → correct offset; tenant isolation; page recycling; snapshot persistence; old snapshot compatibility; PROCESSING cleanup frees pages; all existing tests green.

P3: Master promotion path

Zero behavior change. New promotion path for DISTRIBUTED_DISK.

Single file: master_service.cpp (plus master_service.h for declarations).

Verification: PushDistributedDiskPromotionQueue creates both task records; PromotionObjectHeartbeat consumes distributed tasks; NotifyPromotionSuccess with holder_id=UUID{} bypasses gate, marks staged complete, cleans both task maps; NotifyPromotionFailure symmetric; all existing tests green.

P4: DistributedStorageBackend rewrite (builds on PR #2234)

Feature-flagged. Old logic unchanged when page_mode_=false.

This patch modifies the DistributedStorageBackend introduced by PR #2234. PR #2234 extracted 3FS logic from StorageBackend into a standalone backend with FileSystemAdapter abstraction; this patch replaces its per-key file IO model with the large-file + page-offset model, and injects MasterClient for page coordination.

File	Change
distributed_storage_backend.h/.cpp	Constructor takes MasterClient; Init creates bucket files; BatchOffload / BatchLoad rewritten; ScanMeta returns empty in page mode
storage_backend.cpp	CreateStorageBackend passes MasterClient

Verification: page_mode=false → all existing tests pass; page_mode=true → Init creates files, offload writes correct offset, load reads correct offset, failed-key page reclamation, ScanMeta empty.

P5: FileStorage + Client + RealClient

Feature-flagged. Three sub-patches:

• P5a (file_storage.cpp, storage_backend_interface.h): complete_handler page-mode branch, eviction_handler type fix, IsPageMode() accessor.
• P5b (fd_pool.h/.cpp, client_service.cpp): ReadFromDistributedDisk, FindFirstCompleteReplica type priority, fd pool.
• P5c (real_client.cpp, client_buffer.cpp): SelectBestReplica, calculate_total_size, allocateSlices, execute_ranged_read support.

Verification: page_mode=false → all existing tests pass; end-to-end offload → 3FS → any-node-Get; fd pool reuse; EINTR retry; Python entry point works.

P6: Integration tests + flag cutover

End-to-end verification. Performance comparison: LOCAL_DISK+RPC vs DISTRIBUTED_DISK direct pread.

When all patches verified: flip page_mode_ default to true.

Estimated effort

Patch	New files	Modified files	Estimated lines
P1	3	1	~300
P2	0	~12	~800
P3	0	1	~200
P4	0	2	~300
P5a	0	2	~50
P5b	2	1	~250
P5c	0	2	~80
P6	test files	config	~200

Backward Compatibility

• Wire format: DistributedDiskDescriptor is a new variant alternative appended at the end. Old masters/clients that don't know about it simply never encounter it. Old snapshots without the new variant index deserialize correctly (the new index is beyond their variant size).
• Feature flag: page_mode_=false (default) means all new code paths are dead. Existing LOCAL_DISK behavior is completely unchanged.
• Snapshot format: distributed_page_mappings_ is a new field in MetadataSerializer. Old snapshots yield empty maps; new snapshots are not readable by old masters (standard snapshot version semantics).
• RPCs: New RPCs are additive. Old clients don't call them; old masters don't register them. Mixed-version clusters are safe — the feature simply doesn't activate until both sides are upgraded.
• NotifyOffloadSuccess: Unchanged. Old code path continues to work for LOCAL_DISK. NotifyDistributedDiskSuccess is a separate RPC.

Migration & Rollout

Deployment prerequisites

All nodes in the cluster must have 3FS mounted at the same path (e.g., /mnt/3fs/mooncake). This is a deployment constraint, not a code constraint.

Recommended rollout sequence

1. Upgrade master binary on standby instances, then trigger leader failover. New RPCs are registered but unused.
2. Upgrade client library (Mooncake wheel) on all nodes. page_mode_=false by default — zero behavior change.
3. Enable page mode on one test node: MOONCAKE_DISTRIBUTED_PAGE_MODE=true. Verify end-to-end offload → read.
4. Roll out page mode to all nodes. The 3FS bucket files are created on first Init.
5. Monitor via new metrics. Compare read latency: DISTRIBUTED_DISK direct pread vs old LOCAL_DISK RPC proxy.

Rollback safety

• Set MOONCAKE_DISTRIBUTED_PAGE_MODE=false to revert to LOCAL_DISK behavior. 3FS bucket files remain but are unused.
• No wire-format or snapshot changes that would prevent rolling back the master binary.

Alternatives Considered

A: Fix LOCAL_DISK to work correctly with 3FS. Remove client_id coupling, add offset support to DiskDescriptor. Rejected: LOCAL_DISK semantics (owner-coupled, client lifecycle = data lifecycle) are fundamentally wrong for shared storage. Bolting on fixes creates a confusing hybrid. A clean new type is cheaper long-term.

B: Reuse DISK replica type. DiskReplicaData already has file_path and object_size. Rejected: no offset field. Adding one breaks wire compatibility. DISK is also semantically different (local disk cache, not globally shared).

C: Client-side page mapping (no Master coordination). Each client independently manages its own pages via a shared 3FS metadata file. Rejected: concurrent allocation without a coordinator leads to page collisions. Master is already the membership authority; adding page coordination is natural.

D: Per-key files on 3FS (current model, just fix the read path). Keep one file per KV block, but read it directly instead of via RPC. Rejected: millions of files per bucket is unsustainable at production scale (metadata server pressure, open() cost, ScanMeta walk time). The large-file + page-offset model is the right long-term architecture.

E: Async IO from day one. Use io_uring or fileread_pool_ for ReadFromDistributedDisk. Rejected for Phase 1: significant complexity, and synchronous pread with fd pooling may be sufficient. Benchmark first, optimize if needed.

F: Use 3FS's native KV interface (if available). Some distributed filesystems offer KV APIs. Rejected: 3FS exposes POSIX file API; the page-offset model maps cleanly onto it. No need for a non-standard interface.

G: Stop at PR #2234 (standalone backend, no IO model change). PR #2234 gave DistributedStorageBackend a clean architectural boundary, but preserved the per-key file model and LOCAL_DISK replica semantics. This is the current state of the codebase. Rejected as the final destination: per-key files don't scale (millions of files per bucket, metadata server pressure), LOCAL_DISK semantics still couple data to client processes (owner death = metadata loss), and cross-node reads still require RPC proxying. PR #2234 is the necessary structural prerequisite; this RFC is the semantic and IO-model upgrade that makes the architecture actually deliver on 3FS's value proposition.

Risk Assessment

Risk	Impact	Mitigation
Multi-node concurrent page allocation	Page collision	Master-side BitmapPageAllocator with mutex; single point of coordination
Master restart loses page mappings	Data unreadable	Snapshot HA persistence + bitmap rebuild from mappings
3FS mount inconsistency across nodes	Wrong paths	Config validation + health check at startup
YLT_REFL variant index shift on insert	Snapshot deserialization corruption	New types appended to end only; enforced by code review
O_TRUNC in Init clears existing bucket file	Data loss	Init uses fstat + ftruncate (extend only), never O_TRUNC
Page space exhaustion	Offload failures	Return NO_SPACE; degrade to DRAM-only; monitoring alerts on allocation failures
fd resource exhaustion	IO failures	256 fd pool; ulimit check at startup
Bitmap fragmentation	Allocation efficiency degradation	Per-bucket isolation; hint_ cursor for sequential scanning
XXH64 formula mismatch between Master and DSB	Data written to wrong bucket	Same formula on both sides; unit test coverage
BatchOffload partial write leaks pages	Permanent page waste	Failed keys call BatchFreeDistributedPage in cleanup path
NotifyPromotionSuccess skips mark_complete()	PROCESSING MEMORY replica invisible to readers	Both LOCAL_DISK and DISTRIBUTED_DISK paths call staged->mark_complete()
PromotionObjectHeartbeat returns SEGMENT_NOT_FOUND for DISTRIBUTED_DISK	Promotion blocked	Guard with if-exists; never return error for missing LocalDiskSegment
vLLM inference thread blocked on open()	Latency spike	fd pool caches 256 bucket fds; never re-open during steady state

Test Plan

Unit tests:

• test_bitmap_page_allocator.cpp: Init, Allocate, Free, MarkAllocated, full-allocation returns -1, double-free safety.
• Variant backward compatibility: deserialize a snapshot with [Memory, Disk, LocalDisk, NoF] variant — must not crash.

Integration tests:

• End-to-end: Node A offloads → 3FS → Node B reads via direct pread.
• Multi-node concurrent write to same bucket.
• BatchOffload partial failure → failed-key pages freed.
• Master restart → snapshot restore → page mappings complete → PROCESSING replicas' pages freed.
• Tenant isolation: different tenants, same key name → different pages.
• Delete → Free → re-Allocate.
• ScanMeta returns empty in page mode → ReRegister no-op.

Promotion tests:

• Full chain: PushDistributedDiskPromotionQueue → heartbeat → PromotionAllocStart → BatchLoad → NotifyPromotionSuccess → MEMORY replica visible.
• NotifyPromotionFailure → promotion_in_flight decremented + both task maps cleaned.

Performance tests:

• LOCAL_DISK+RPC proxy vs DISTRIBUTED_DISK direct pread latency comparison (target: ≥2× improvement for cross-node reads).
• Batch-get throughput: 100 keys, measure total latency with and without fd pool.

References

• PR #2234 — Structural foundation: extracted 3FS logic into standalone DistributedStorageBackend with FileSystemAdapter abstraction (merged 2026-06-03)
• Issue #2254 — LOCAL_DISK warm re-adoption bug (composes with this RFC)
• RFC #1920 — Rolling Upgrade (snapshot compatibility coordination)
• PR #2215 — OffsetAllocator restart recovery (composes)
• Mooncake paper (arXiv:2407.00079) — system architecture and production trace data
• large_page_plan.md v10.1 — detailed design document (internal)
• patch_plan.md — patch splitting plan (internal)

Before submitting a new issue...

• Make sure you already searched for relevant issues and read the documentation