Method deliver_to_session

nodedb/src/control/array_sync/outbound/fanout.rs:137–183

Deliver one op to a single session via the multi-shard merger. The merger buffers ops from all vShards and drains them in HLC order, ensuring subscribers see a consistent stream regardless of which shard the op originated from.

(&self, session_id: &str, op: &ArrayOp, op_hlc: Hlc, _op_payload: &[u8])

    /// ensuring subscribers see a consistent stream regardless of which shard
    /// the op originated from.
    fn deliver_to_session(&self, session_id: &str, op: &ArrayOp, op_hlc: Hlc, _op_payload: &[u8]) {
        let cursor = match self.cursors.get(session_id, &op.header.array) {
            Some(c) => c,
            None => {
                // Session has not registered for this array yet — skip.
                return;
            }
        };

        // Check if this op has already been delivered.
        if !cursor::should_send(op_hlc, cursor.last_pushed_hlc) {
            debug!(
                session = %session_id,
                array = %op.header.array,
                op_hlc = ?op_hlc,
                "array_fanout: op already delivered, skipping"
            );
            return;
        }

        // Check if the subscriber cursor has fallen behind the GC boundary.
        let snapshot_hlc = self
            .snapshot_hlcs
            .read()
            .ok()
            .and_then(|m| m.get(&op.header.array).copied())
            .unwrap_or(Hlc::ZERO);

        if snapshot_trigger::check_and_trigger(
            session_id,
            &op.header.array,
            cursor.last_pushed_hlc,
            snapshot_hlc,
            &self.delivery,
        ) {
            // Subscriber needs catch-up — do not send op stream frames.
            return;
        }

        // Route through the multi-shard merger for HLC-ordered delivery.
        let merger = self.mergers.get_or_create(session_id, &op.header.array);
        merger.push_op(self.shard_id, op.clone(), &self.delivery);

        // Advance the cursor so subsequent ops from any shard know the
        // current frontier.
        cursor::mark_sent(&self.cursors, session_id, &op.header.array, op_hlc);
    }
}

impl ArrayApplyObserver for ArrayFanout {