Method stream_ops

nodedb/src/control/array_sync/catchup.rs:154–220

Stream all ops for `array` with `hlc >= from_hlc` as batched delta frames, advancing the subscriber cursor after each batch.

(&self, session_id: &str, array: &str, from_hlc: Hlc)

    /// Stream all ops for `array` with `hlc >= from_hlc` as batched delta
    /// frames, advancing the subscriber cursor after each batch.
    fn stream_ops(&self, session_id: &str, array: &str, from_hlc: Hlc) -> crate::Result<()> {
        // scan_from returns all ops with hlc >= from_hlc across all arrays;
        // we filter to the target array here.
        let ops_all = self
            .op_log
            .scan_from(from_hlc)
            .map_err(|e| crate::Error::Storage {
                engine: "array_sync".into(),
                detail: format!("catchup_server scan_from: {e}"),
            })?;

        let mut batch_payloads: Vec<Vec<u8>> = Vec::with_capacity(BATCH_SIZE);
        let mut last_hlc = from_hlc;

        for op_result in ops_all {
            let op = op_result.map_err(|e| crate::Error::Storage {
                engine: "array_sync".into(),
                detail: format!("catchup_server scan_from iter: {e}"),
            })?;
            if op.header.array != array {
                continue;
            }

            match op_codec::encode_op(&op) {
                Ok(payload) => {
                    last_hlc = op.header.hlc;
                    batch_payloads.push(payload);
                }
                Err(e) => {
                    warn!(
                        session = %session_id,
                        array = %array,
                        error = %e,
                        "catchup_server: encode_op failed — skipping op"
                    );
                    continue;
                }
            }

            if batch_payloads.len() >= BATCH_SIZE {
                let batch = ArrayDeltaBatchMsg {
                    array: array.to_owned(),
                    op_payloads: std::mem::take(&mut batch_payloads),
                };
                self.send_frame(session_id, SyncMessageType::ArrayDeltaBatch, &batch);
                self.cursors.mark_sent(session_id, array, last_hlc);
                debug!(
                    session = %session_id,
                    array = %array,
                    last_hlc = ?last_hlc,
                    "catchup_server: flushed batch"
                );
            }
        }

        // Flush any remaining ops.
        if !batch_payloads.is_empty() {
            let batch = ArrayDeltaBatchMsg {