Method handle_snapshot_chunk

nodedb/src/control/array_sync/snapshot_assembly.rs:75–192

Buffer a snapshot chunk and, when complete, assemble and apply all ops.

(
        &self,
        msg: &ArraySnapshotChunkMsg,
    )

    /// Buffer a snapshot chunk and, when complete, assemble and apply all ops.
    pub async fn handle_snapshot_chunk(
        &self,
        msg: &ArraySnapshotChunkMsg,
    ) -> Result<InboundOutcome, Option<ArrayRejectMsg>> {
        let key = (msg.array.clone(), msg.snapshot_hlc_bytes);

        let assembled: Option<(SnapshotHeader, Vec<SnapshotChunk>)> = {
            let mut snapshots = match self.snapshots().lock() {
                Ok(g) => g,
                Err(_) => {
                    error!(array = %msg.array, "array_inbound: snapshot mutex poisoned (chunk)");
                    return Err(None);
                }
            };
            let entry = snapshots
                .entry(key.clone())
                .or_insert_with(SnapshotAssembly::new);

            let chunk = SnapshotChunk {
                array: msg.array.clone(),
                chunk_index: msg.chunk_index,
                total_chunks: msg.total_chunks,
                payload: msg.payload.clone(),
                snapshot_hlc: Hlc::from_bytes(&msg.snapshot_hlc_bytes),
            };
            entry.chunks.insert(msg.chunk_index, chunk);

            let total = msg.total_chunks as usize;
            if let Some(h) = &entry.header {
                if entry.chunks.len() == total {
                    let header = h.clone();
                    let chunks_vec: Vec<SnapshotChunk> = entry.chunks.values().cloned().collect();
                    Some((header, chunks_vec))
                } else {
                    None
                }
            } else {
                None
            }
        };

        let Some((header, mut chunks)) = assembled else {
            let snapshots = match self.snapshots().lock() {
                Ok(g) => g,
                Err(_) => return Err(None),
            };
            let received = snapshots
                .get(&key)
                .map(|e| e.chunks.len() as u32)
                .unwrap_or(0);
            return Ok(InboundOutcome::SnapshotPartial {
                received,
                total: msg.total_chunks,
            });
        };

        // Assemble snapshot.
        let snapshot = match assemble_chunks(&header, &mut chunks) {