Struct FlowStateManager 

pub struct FlowStateManager {
    in_memory: KvBackendRef,
    merge_lock: Mutex<()>,
}

The manager of FlowStateKey. Since state size changes frequently, we store it in memory.

This is only used in distributed mode. When meta-srv use heartbeat to update the flow stat report and frontned use get to get the latest flow stat report.

Per-flownode reports are stored under __flow/state/node/{node_id} keys in the in-memory KV (not in a separate in-process map), so a metasrv leader change — which resets the in-memory KV — automatically clears all per-node state without leaving stale entries behind.

Fields

in_memory: KvBackendRef

merge_lock: Mutex<()>

Serializes the critical section of FlowStateManager::merge (write per-node key -> scan -> aggregate -> write global key). It holds no long-lived state; per-node reports live in the in-memory KV.

Implementations

impl FlowStateManager

pub fn new(in_memory: KvBackendRef) -> Self

pub async fn get(&self) -> Result<Option<FlowStateValue>>

pub async fn put(&self, value: FlowStateValue) -> Result<()>

pub async fn merge(&self, node_id: u64, incoming: FlowStateValue) -> Result<()>

Merges a flow state report from a single flownode into the global view.

node_id is the operator-configured flownode id (unique within the cluster; reported as HeartbeatRequest.header.member_id, the canonical metasrv identity of a flownode, and equal to peer.id).

Reports are tracked per node under __flow/state/node/{node_id} in the in-memory KV. A new report from the same node unconditionally replaces that node’s previous entry: reports are processed strictly in arrival order, so no epoch comparison is made and a wall-clock rollback after a node restart cannot permanently drop the node’s later reports. The global FlowStateValue is then aggregated over all per-node entries: for the same flow, last_exec_time_map takes the max reported timestamp and state_size takes the max reported size across nodes (a flow normally runs on a single active flownode, so max is a safe approximation). The aggregated value is written into the in-memory KV under the global key __flow/state via the same path as put, keeping get() behavior unchanged.

Per-node keys are cleared automatically when the in-memory KV is reset on a leader change. Known limitation: entries of dropped flows are not proactively removed here. Once a flow is dropped its metadata is gone, so the flows table join simply can’t see the stale entry; it becomes user-invisible until the owning flownode reports again (or stops heartbeating forever, in which case the stale flow id lingers in the aggregate but is never joined against any flow metadata).