Struct MeshSyncManager 

pub struct MeshSyncManager { /* private fields */ }

Mesh sync manager for coordinating state synchronization

Implementations

impl MeshSyncManager

pub fn new(stores: Arc<StateStores>, self_name: String) -> Self

pub fn register_tree_state_subscriber( &self, subscriber: Arc<dyn TreeStateSubscriber>, )

pub fn register_worker_state_subscriber( &self, subscriber: Arc<dyn WorkerStateSubscriber>, )

pub fn self_name(&self) -> &str

Get the node name (actor) for this sync manager

pub fn sync_worker_state( &self, worker_id: String, model_id: String, url: String, health: bool, load: f64, spec: Vec<u8>, )

Sync worker state to mesh stores

pub fn remove_worker_state(&self, worker_id: &str)

Remove worker state from mesh stores

pub fn sync_policy_state( &self, model_id: String, policy_type: String, config: Vec<u8>, )

Sync policy state to mesh stores

pub fn remove_policy_state(&self, model_id: &str)

Remove policy state from mesh stores

pub fn get_worker_state(&self, worker_id: &str) -> Option<WorkerState>

Get worker state from mesh stores

pub fn get_all_worker_states(&self) -> Vec<WorkerState>

Get all worker states from mesh stores

pub fn get_policy_state(&self, model_id: &str) -> Option<PolicyState>

Get policy state from mesh stores

pub fn get_all_policy_states(&self) -> Vec<PolicyState>

Get all policy states from mesh stores

pub fn apply_remote_worker_state( &self, state: WorkerState, actor: Option<String>, )

Apply worker state update from remote node The actor should be extracted from the state update context (e.g., from StateUpdate message)

pub fn apply_remote_policy_state( &self, state: PolicyState, actor: Option<String>, )

Apply policy state update from remote node The actor should be extracted from the state update context (e.g., from StateUpdate message)

pub fn update_rate_limit_membership(&self)

Update rate-limit hash ring with current membership

pub fn handle_node_failure(&self, failed_nodes: &[String])

Handle node failure and transfer rate-limit ownership

pub fn sync_rate_limit_inc(&self, key: String, delta: i64)

Sync rate-limit counter increment (only if this node is an owner)

pub fn apply_remote_rate_limit_counter(&self, log: &OperationLog)

Apply remote rate-limit counter update (merge CRDT)

pub fn apply_remote_rate_limit_counter_value( &self, key: String, counter_value: i64, )

Apply remote rate-limit counter snapshot encoded as raw i64.

pub fn apply_remote_rate_limit_counter_value_with_actor( &self, key: String, actor: String, counter_value: i64, )

pub fn apply_remote_rate_limit_counter_value_with_actor_and_timestamp( &self, key: String, actor: String, counter_value: i64, timestamp: u64, )

pub fn get_rate_limit_value(&self, key: &str) -> Option<i64>

Get rate-limit value (aggregate from all owners)

pub fn get_global_rate_limit_config(&self) -> Option<RateLimitConfig>

Get global rate limit configuration from AppStore

pub fn check_global_rate_limit(&self) -> (bool, i64, u64)

Check if global rate limit is exceeded Returns (is_exceeded, current_count, limit)

pub fn reset_global_rate_limit_counter(&self)

Reset global rate limit counter (called periodically for time window reset)

pub fn sync_tree_insert_hash( &self, model_id: &str, path_hash: u64, tenant: &str, )

Sync tree operation to mesh stores.

This is called on every request (hot path). The operation is appended to the pending buffer for delta sync — the collector serializes and sends it to peers. We do NOT read/deserialize/serialize the full TreeState here, because that is O(tree_size) per request and caused multi-GB memory usage at 200+ rps.

The policy store version is bumped so the generation-based collector detects the change, but the config blob is NOT updated on every call. It is rebuilt lazily by the collector when a full-state fallback is needed. Lightweight sync: accepts a pre-computed hash + tenant, avoiding the 80k+ String allocation from TreeKey::Text on every request.

pub fn sync_tree_operation( &self, model_id: String, operation: TreeOperation, ) -> Result<(), String>

pub fn get_tree_state(&self, model_id: &str) -> Option<TreeState>

Get tree state for a model from mesh stores. Reads from tree_configs (populated by periodic checkpoint from live tree).

pub fn get_all_tree_states(&self) -> Vec<TreeState>

pub fn apply_remote_tree_operation( &self, model_id: String, tree_state: TreeState, actor: Option<String>, )

Apply remote tree operation to local stores. This is called when receiving full tree state updates from other nodes.

Writes to tree_configs (plain DashMap) instead of the CRDT policy store to avoid operation log memory accumulation.

Uses DashMap::entry() for atomic read-modify-write on tree_configs to avoid the TOCTOU gap between get() and insert().

pub fn apply_remote_tree_delta( &self, delta: TreeStateDelta, actor: Option<String>, )

Apply a delta (incremental operations) from a remote node. Merges the delta operations into the existing local tree state, avoiding the cost of replacing the entire tree state on every sync.

Uses DashMap::entry() for atomic read-modify-write on tree_configs to avoid the TOCTOU gap between get() and insert().

pub fn apply_remote_tenant_delta( &self, delta: TenantDelta, _actor: Option<String>, )

Apply a lightweight tenant delta from a remote node. Updates the local radix tree directly via subscribers without going through the CRDT or the full TreeState machinery.

pub fn checkpoint_tree_states(&self)

Checkpoint tree state by exporting compact snapshots from the live radix tree via subscribers.

Called periodically (~every 10s) to keep tree_configs fresh for the periodic structure snapshot (every 30 gossip rounds).

Uses TreeStateSubscriber::export_tree_snapshot to obtain a compact kv_index::snapshot::TreeSnapshot that preserves shared prefixes. This is much smaller than the flat TreeState produced by export_tree_state (~2-4 MB vs ~40 MB for 2048 entries sharing 80% prefixes) and avoids accumulating full prompt text in memory.

Trait Implementations

impl Clone for MeshSyncManager

fn clone(&self) -> MeshSyncManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for MeshSyncManager

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.