Struct InstanceLeader 

pub struct InstanceLeader { /* private fields */ }

Primary leader implementation for the distributed KVBM system.

InstanceLeader coordinates block onboarding across local and remote instances. It owns a G2 (host memory) BlockManager and an optional G3 (disk) BlockManager, a set of workers for executing physical transfers, and a parallel worker abstraction for multi-rank RDMA operations.

Key responsibilities:

• Block matching: finding which requested sequence hashes are already cached locally (via BlockAccessor policies).
• Session management: creating, attaching, and driving onboard sessions between endpoint (source) and controller (destination) roles.
• Remote connectivity: exchanging serialized layout metadata with peer instances so workers can perform RDMA transfers.
• Velo RPC: registering handlers via VeloLeaderService so remote leaders can initiate sessions and exchange metadata.

Implementations

impl InstanceLeader

pub fn g2_manager(&self) -> &Arc<BlockManager<G2>>

Get a reference to the G2 BlockManager.

pub fn g3_manager(&self) -> Option<&Arc<BlockManager<G3>>>

Get a reference to the optional G3 BlockManager.

pub fn registry(&self) -> &BlockRegistry

Get the block registry.

pub fn messenger(&self) -> &Arc<Messenger>

Get a reference to the Nova instance.

This provides access to the Nova distributed system for features like event coordination and cross-instance communication.

pub fn runtime(&self) -> Handle

Get the tokio runtime handle from Nova.

This handle should be used for spawning background tasks that need to run on the KVBM runtime’s executor (e.g., offload engine pipelines).

pub fn has_parallel_worker(&self) -> bool

Check if a parallel_worker is configured.

The parallel_worker is required for local transfer operations (e.g., offloading blocks between tiers).

pub fn parallel_worker(&self) -> Option<Arc<dyn ParallelWorkers>>

Get the parallel worker for distributed operations.

The parallel worker fans out operations to all workers and aggregates results. It implements ObjectBlockOps for coordinated object storage uploads.

pub fn object_client(&self) -> Option<Arc<dyn ObjectBlockOps>>

Get the object storage client for G4 operations.

Returns Some if object storage is configured, None otherwise. The client is used by InitiatorSession for G4 parallel search.

pub fn add_remote_leader(&self, instance_id: InstanceId)

Add a remote leader to the search list.

Remote leaders are queried during find_matches_with_options when search_remote == true. This method allows adding remote leaders after construction (e.g., when instance IDs are only known after cluster setup).

pub fn set_remote_leaders(&self, instance_ids: Vec<InstanceId>)

Set all remote leaders at once.

pub fn remote_leaders(&self) -> Vec<InstanceId>

Get the list of remote leader instance IDs.

pub fn scan_blocks( &self, sequence_hashes: &[SequenceHash], touch: bool, ) -> ScanBlocksResult

Scan for all blocks matching any of the given sequence hashes.

Unlike find_matches, this:

• Does NOT stop on first miss
• Returns blocks from both G2 and G3 tiers separately
• Acquires blocks from pools (caller owns until dropped via RAII)
• Returns sorted_matches ordered by SequenceHash::position()

Arguments

• sequence_hashes - Hashes to scan for
• touch - Whether to update frequency tracking (for MultiLRU eviction policy)

Algorithm

1. Scan G2 manager for candidates
2. Scan G3 manager for remaining candidates
3. Build sorted_matches from both, sorted by position (lowest to highest)

pub fn scan_with_policy<F, T>( &self, hashes: &[SequenceHash], touch: bool, policy: F, ) -> Vec<T>

where F: FnOnce(&[SequenceHash], &mut PolicyContext<'_, T>),

Scan blocks using a custom policy that controls iteration and yields results.

This provides maximum flexibility for implementing custom scanning strategies. The policy receives access to a BlockAccessor for acquiring blocks and a PolicyContext for yielding results incrementally.

Arguments

• hashes - Sequence hashes to scan
• touch - Whether to update frequency tracking on block access
• policy - Function that implements the scanning strategy

Design

The accessor does NOT hold locks between calls. Each .find() call is independent. This enables:

• Custom iteration patterns (sorted, BTree scan, binary search, etc.)
• Yielding results incrementally (e.g., contiguous subsequences)
• Future parallel execution (accessor is Send + Sync)

Example: Simple linear scan

let blocks = leader.scan_with_policy(&hashes, true, |hashes, ctx| {
    for hash in hashes {
        if let Some(block) = ctx.accessor().find(*hash) {
            ctx.yield_item(block);
        }
    }
});

Example: Find contiguous subsequences

let runs: Vec<Vec<TieredBlock>> = leader.scan_with_policy(&hashes, true, |hashes, ctx| {
    let mut run = Vec::new();
    let mut last_pos: Option<u64> = None;

    for hash in hashes.iter().sorted_by_key(|h| h.position()) {
        if let Some(block) = ctx.accessor().find(*hash) {
            let pos = block.position();
            if last_pos.map_or(true, |p| pos == p + 1) {
                run.push(block);
            } else {
                if !run.is_empty() { ctx.yield_item(std::mem::take(&mut run)); }
                run.push(block);
            }
            last_pos = Some(pos);
        } else if !run.is_empty() {
            ctx.yield_item(std::mem::take(&mut run));
            last_pos = None;
        }
    }
    if !run.is_empty() { ctx.yield_item(run); }
});

pub fn builder() -> InstanceLeaderBuilder

pub fn register_handlers(&self) -> Result<()>

Register Nova handlers for leader-to-leader communication.

This must be called after construction to enable distributed onboarding.

pub fn release_session(&self, session_id: SessionId)

Release a completed session, dropping any held blocks.

This is optional - sessions will naturally be cleaned up when the InstanceLeader is dropped. Call this explicitly if you need to release blocks earlier.

pub fn create_controllable_session( &self, sequence_hashes: &[SequenceHash], ) -> Result<ControllableSessionResult>

Create a controllable session for local blocks.

This is the “Decode side” of the inverted control pattern:

1. Search local G2 and G3 for matches
2. Create a ControllableSession that holds the blocks
3. Return session_id to be sent to Prefill out-of-band

By default, G3→G2 staging starts immediately (auto_stage=true).

pub fn create_controllable_session_with_options( &self, sequence_hashes: &[SequenceHash], options: ControllableSessionOptions, ) -> Result<ControllableSessionResult>

Create a controllable session with custom options.

Use this when you need to control auto-staging behavior.

pub async fn attach_session( &self, remote_instance: InstanceId, session_id: SessionId, ) -> Result<SessionHandle>

Attach to a remote session. Returns a SessionHandle that uses SessionMessage for communication.

Arguments

• remote_instance - The instance hosting the session
• session_id - The session to attach to

Example

let handle = leader.attach_session(remote_id, session_id).await?;
let state = handle.wait_for_ready().await?;
handle.trigger_staging().await?;

pub fn create_endpoint_session( &self, sequence_hashes: &[SequenceHash], ) -> Result<(SessionId, ServerSessionHandle)>

Create an endpoint session that a remote peer can attach to.

This searches local G2/G3 for blocks matching the given sequence hashes and creates a session that exposes them for remote RDMA pull.

Returns (session_id, handle) where:

• session_id - Send to remote peer for attachment
• handle - Use to control the session (send layer notifications, close)

Example

// Create session for sequence hashes
let (session_id, handle) = leader.create_endpoint_session(&hashes)?;

// Send session_id to remote peer out-of-band
// Remote attaches via: remote_leader.attach_session(local_id, session_id)

// For layerwise transfer, notify when layers are ready
handle.notify_layers_ready(0..1).await?;

pub fn create_endpoint_session_for_blocks( &self, blocks: BlockHolder<G2>, sequence_hashes: &[SequenceHash], layout_handles: &[LayoutHandle], ) -> Result<(SessionId, ServerSessionHandle)>

Create an endpoint session for specific pre-allocated blocks.

Unlike create_endpoint_session, this doesn’t search - it uses the provided blocks directly. Useful when the caller already has blocks to expose (e.g., after prefill computation).

Arguments

• blocks - Blocks to expose for RDMA pull
• sequence_hashes - Sequence hashes for the blocks (must match block count)
• layout_handles - Layout handles for the blocks (must match block count)

Example

// After prefill computation, expose blocks for Decode to pull
let (session_id, handle) = leader.create_endpoint_session_for_blocks(
    prefill_blocks,
    &hashes,
    &layout_handles,
)?;

pub fn has_remote_metadata(&self, instance: InstanceId) -> bool

Check if metadata for a remote instance has been loaded.

Returns true if import_remote_metadata has been successfully called for the given instance.

pub fn worker_count(&self) -> usize

Get the number of workers attached to this leader.

pub async fn export_worker_metadata(&self) -> Result<Vec<SerializedLayout>>

Export metadata from all workers.

Returns a Vec<SerializedLayout> where each element corresponds to a worker in rank order. This metadata can be sent to remote instances to enable RDMA transfers.

Returns

Vector of serialized layouts, one per worker

pub async fn import_remote_metadata( &self, remote_instance: InstanceId, metadata: Vec<SerializedLayout>, ) -> Result<()>

Import metadata from a remote instance’s workers.

This imports layout metadata from a remote instance, enabling RDMA transfers to pull data from that instance. Metadata is imported rank-by-rank:

• local worker 0 imports remote worker 0’s metadata
• local worker 1 imports remote worker 1’s metadata
• etc.

Arguments

• remote_instance - The instance ID of the remote leader
• metadata - Vector of SerializedLayout from remote workers (one per worker)

Errors

Returns an error if:

• No parallel worker configured
• Metadata was already imported for this instance
• Worker count mismatch between local and remote
• Individual worker metadata import fails

Trait Implementations

impl Clone for InstanceLeader

fn clone(&self) -> InstanceLeader

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Leader for InstanceLeader

fn find_matches_with_options( &self, sequence_hashes: &[SequenceHash], options: FindMatchesOptions, ) -> Result<FindMatchesResult>

Find matching blocks with custom options.

fn find_matches( &self, sequence_hashes: &[SequenceHash], ) -> Result<FindMatchesResult>

Find matching blocks with default options.