Serve a catch-up request from `session_id`. Validates the array, selects the delivery path (op-stream vs snapshot + op-stream), and enqueues all resulting frames into the session's delivery channel. Returns `Ok(())` even when the channel is full (frames are silently dropped; the Lite peer will retry on next reconnect).
(&self, req: &ArrayCatchupRequestMsg, session_id: &str)
| 82 | /// channel. Returns `Ok(())` even when the channel is full (frames are |
| 83 | /// silently dropped; the Lite peer will retry on next reconnect). |
| 84 | pub fn serve(&self, req: &ArrayCatchupRequestMsg, session_id: &str) -> crate::Result<()> { |
| 85 | // 1. Validate the array exists. |
| 86 | if self.schemas.schema_hlc(&req.array).is_none() { |
| 87 | warn!( |
| 88 | session = %session_id, |
| 89 | array = %req.array, |
| 90 | "catchup_server: array not found in schema registry — ignoring request" |
| 91 | ); |
| 92 | return Ok(()); |
| 93 | } |
| 94 | |
| 95 | let from_hlc = Hlc::from_bytes(&req.from_hlc_bytes); |
| 96 | |
| 97 | // 2. Determine whether the request falls below the GC boundary. |
| 98 | let gc_boundary = self.ack_registry.min_ack_hlc(&req.array); |
| 99 | let snapshot_hlc_opt = gc_boundary |
| 100 | .filter(|gc| from_hlc < *gc) |
| 101 | .and_then(|_| self.snapshots.latest_for_array(&req.array)); |
| 102 | |
| 103 | if let Some(snapshot) = snapshot_hlc_opt { |
| 104 | // Snapshot path: send snapshot, then op-stream from snapshot_hlc. |
| 105 | let snap_hlc = snapshot.snapshot_hlc; |
| 106 | |
| 107 | // Encode the snapshot header payload. |
| 108 | let (header, chunks) = |
| 109 | split_into_chunks(&snapshot, CHUNK_BYTES).map_err(|e| crate::Error::Storage { |
| 110 | engine: "array_sync".into(), |
| 111 | detail: format!("catchup_server split_into_chunks: {e}"), |
| 112 | })?; |
| 113 | |
| 114 | let header_payload = |
| 115 | zerompk::to_msgpack_vec(&header).map_err(|e| crate::Error::Storage { |
| 116 | engine: "array_sync".into(), |
| 117 | detail: format!("catchup_server header encode: {e}"), |
| 118 | })?; |
| 119 | |
| 120 | let snap_msg = ArraySnapshotMsg { |
| 121 | array: req.array.clone(), |
| 122 | header_payload, |
| 123 | }; |
| 124 | |
| 125 | self.send_frame(session_id, SyncMessageType::ArraySnapshot, &snap_msg); |
| 126 | |
| 127 | for chunk in &chunks { |
| 128 | let chunk_msg = ArraySnapshotChunkMsg { |
| 129 | array: req.array.clone(), |
| 130 | chunk_index: chunk.chunk_index, |
| 131 | total_chunks: chunk.total_chunks, |
| 132 | payload: chunk.payload.clone(), |
| 133 | snapshot_hlc_bytes: chunk.snapshot_hlc.to_bytes(), |
| 134 | }; |
| 135 | self.send_frame(session_id, SyncMessageType::ArraySnapshotChunk, &chunk_msg); |
| 136 | } |
| 137 | |
| 138 | // Continue with ops from snapshot_hlc onward (the subscriber |
| 139 | // is now caught up to snap_hlc; needs ops above it). |
| 140 | self.stream_ops(session_id, &req.array, snap_hlc)?; |
| 141 |