calimero_network/handlers/commands/
request_blob.rs

1use core::time::Duration;
2
3use actix::{Context, Handler, Message, ResponseFuture};
4use borsh::BorshDeserialize;
5use calimero_network_primitives::{
6    blob_types::{BlobChunk, BlobRequest, BlobResponse},
7    messages::{NetworkEvent, RequestBlob},
8    stream::{Message as StreamMessage, Stream, CALIMERO_BLOB_PROTOCOL},
9};
10use eyre::{eyre, Context as EyreContext};
11use futures_util::{SinkExt, StreamExt};
12use tokio::time::timeout;
13use tracing::debug;
14
15use crate::NetworkManager;
16
17// Timeout for individual operations during blob transfer
18const BLOB_TRANSFER_TIMEOUT: Duration = Duration::from_secs(60); // 1 minute per operation
19const CHUNK_RECEIVE_TIMEOUT: Duration = Duration::from_secs(30); // 30 seconds per chunk
20
21impl Handler<RequestBlob> for NetworkManager {
22    type Result = ResponseFuture<<RequestBlob as Message>::Result>;
23
24    fn handle(&mut self, request: RequestBlob, _ctx: &mut Context<Self>) -> Self::Result {
25        debug!(
26            blob_id = %request.blob_id,
27            context_id = %request.context_id,
28            peer_id = %request.peer_id,
29            auth = ?request.auth,
30            "Requesting blob from peer using binary chunk protocol"
31        );
32
33        let mut stream_control = self.swarm.behaviour().stream.new_control();
34        let event_dispatcher = self.event_dispatcher.clone();
35
36        Box::pin(async move {
37            // Wrap the entire blob transfer in a timeout
38            let transfer_result = timeout(BLOB_TRANSFER_TIMEOUT, async {
39                // Open a stream to the peer
40                let libp2p_stream = match stream_control
41                    .open_stream(request.peer_id, CALIMERO_BLOB_PROTOCOL)
42                    .await
43                {
44                    Ok(stream) => stream,
45                    Err(e) => {
46                        // Emit failure event
47                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
48                            blob_id: request.blob_id,
49                            context_id: request.context_id,
50                            from_peer: request.peer_id,
51                            error: format!("Failed to open stream: {e}"),
52                        });
53                        return Err(e).wrap_err("Failed to open stream to peer");
54                    }
55                };
56
57                // Convert to Calimero stream
58                let mut stream = Stream::new(libp2p_stream);
59
60                // Send blob request
61                let blob_request = BlobRequest {
62                    blob_id: request.blob_id,
63                    context_id: request.context_id,
64                    auth: request.auth,
65                };
66
67                let request_data = match serde_json::to_vec(&blob_request) {
68                    Ok(data) => data,
69                    Err(e) => {
70                        // Emit failure event
71                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
72                            blob_id: request.blob_id,
73                            context_id: request.context_id,
74                            from_peer: request.peer_id,
75                            error: format!("Failed to serialize request: {e}"),
76                        });
77                        return Err(e).wrap_err("Failed to serialize blob request");
78                    }
79                };
80
81                if let Err(e) = stream.send(StreamMessage::new(request_data)).await {
82                    // Emit failure event
83                    let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
84                        blob_id: request.blob_id,
85                        context_id: request.context_id,
86                        from_peer: request.peer_id,
87                        error: format!("Failed to send request: {e}"),
88                    });
89                    return Err(e).wrap_err("Failed to send blob request");
90                }
91
92                // Wait for initial response with timeout
93                let response_msg = match timeout(CHUNK_RECEIVE_TIMEOUT, stream.next()).await {
94                    Ok(Some(Ok(msg))) => msg,
95                    Ok(Some(Err(e))) => {
96                        // Emit failure event
97                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
98                            blob_id: request.blob_id,
99                            context_id: request.context_id,
100                            from_peer: request.peer_id,
101                            error: format!("Failed to receive response: {e}"),
102                        });
103                        return Err(e).wrap_err("Failed to receive response");
104                    }
105                    Ok(None) => {
106                        // Emit failure event
107                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
108                            blob_id: request.blob_id,
109                            context_id: request.context_id,
110                            from_peer: request.peer_id,
111                            error: "Stream closed without response".to_owned(),
112                        });
113                        return Err(eyre!("Stream closed without response"));
114                    }
115                    Err(_) => {
116                        // Timeout occurred
117                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
118                            blob_id: request.blob_id,
119                            context_id: request.context_id,
120                            from_peer: request.peer_id,
121                            error: "Timeout waiting for response".to_owned(),
122                        });
123                        return Err(eyre!("Timeout waiting for response"));
124                    }
125                };
126
127                let blob_response: BlobResponse = match serde_json::from_slice(&response_msg.data) {
128                    Ok(response) => response,
129                    Err(e) => {
130                        // Emit failure event
131                        let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
132                            blob_id: request.blob_id,
133                            context_id: request.context_id,
134                            from_peer: request.peer_id,
135                            error: format!("Failed to deserialize response: {e}"),
136                        });
137                        return Err(e).wrap_err("Failed to deserialize blob response");
138                    }
139                };
140
141                debug!(
142                    blob_id = %request.blob_id,
143                    peer_id = %request.peer_id,
144                    found = blob_response.found,
145                    size = ?blob_response.size,
146                    "Received initial blob response"
147                );
148
149                if blob_response.found {
150                    debug!(
151                        blob_id = %request.blob_id,
152                        context_id = %request.context_id,
153                        peer_id = %request.peer_id,
154                        size = ?blob_response.size,
155                        "Blob found, starting streaming download"
156                    );
157
158                    // Collect chunks efficiently (no pre-allocation based on expected size)
159                    let mut collected_data = Vec::new();
160                    let mut chunk_count = 0;
161                    let start_time = std::time::Instant::now();
162
163                    debug!(
164                        blob_id = %request.blob_id,
165                        peer_id = %request.peer_id,
166                        expected_size = ?blob_response.size,
167                        "Starting blob download"
168                    );
169
170                    // Process chunks and collect data
171                    loop {
172                        debug!(
173                            blob_id = %request.blob_id,
174                            peer_id = %request.peer_id,
175                            chunk_number = chunk_count + 1,
176                            "Waiting for next chunk"
177                        );
178
179                        let chunk_msg = match timeout(CHUNK_RECEIVE_TIMEOUT, stream.next()).await {
180                            Ok(Some(Ok(msg))) => {
181                                debug!(
182                                    blob_id = %request.blob_id,
183                                    peer_id = %request.peer_id,
184                                    chunk_number = chunk_count + 1,
185                                    message_size = msg.data.len(),
186                                    "Received raw chunk message"
187                                );
188                                msg
189                            },
190                            Ok(Some(Err(e))) => {
191                                let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
192                                    blob_id: request.blob_id,
193                                    context_id: request.context_id,
194                                    from_peer: request.peer_id,
195                                    error: format!("Stream error while receiving chunk: {e}"),
196                                });
197                                return Err(e).wrap_err("Stream error while receiving chunk");
198                            }
199                            Ok(None) => {
200                                // Stream closed - this is natural EOF
201                                debug!(
202                                    blob_id = %request.blob_id,
203                                    peer_id = %request.peer_id,
204                                    total_chunks = chunk_count,
205                                    total_size = collected_data.len(),
206                                    "Stream closed, blob transfer complete"
207                                );
208                                break;
209                            }
210                            Err(_) => {
211                                let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
212                                    blob_id: request.blob_id,
213                                    context_id: request.context_id,
214                                    from_peer: request.peer_id,
215                                    error: format!("Timeout waiting for chunk {}", chunk_count + 1),
216                                });
217                                return Err(eyre!("Timeout waiting for chunk"));
218                            }
219                        };
220
221                        let blob_chunk = match BlobChunk::try_from_slice(&chunk_msg.data) {
222                            Ok(chunk) => {
223                                let chunk_size: usize = chunk.data.len();
224                                debug!("Successfully parsed borsh chunk: blob_id={}, peer_id={}, chunk_number={}, chunk_size={}",
225                                    request.blob_id,
226                                    request.peer_id,
227                                    chunk_count + 1,
228                                    chunk_size
229                                );
230                                chunk
231                            },
232                            Err(e) => {
233                                // Log raw data for debugging
234                                debug!(
235                                    blob_id = %request.blob_id,
236                                    peer_id = %request.peer_id,
237                                    chunk_number = chunk_count + 1,
238                                    raw_data_size = chunk_msg.data.len(),
239                                    raw_data_hex = hex::encode(&chunk_msg.data[..core::cmp::min(100, chunk_msg.data.len())]),
240                                    error = %e,
241                                    "Failed to parse chunk"
242                                );
243                                let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
244                                    blob_id: request.blob_id,
245                                    context_id: request.context_id,
246                                    from_peer: request.peer_id,
247                                    error: format!("Failed to parse chunk: {e}"),
248                                });
249                                return Err(eyre::eyre!("Failed to parse blob chunk: {}", e));
250                            }
251                        };
252
253                        // Get chunk size before moving the data
254                        let chunk_size = blob_chunk.data.len();
255                        // Add chunk data to collection (move semantics, no clone)
256                        collected_data.extend(blob_chunk.data);
257                        chunk_count += 1;
258
259                        debug!(
260                            blob_id = %request.blob_id,
261                            peer_id = %request.peer_id,
262                            chunk_number = chunk_count,
263                            chunk_size = chunk_size,
264                            total_received = collected_data.len(),
265                            "Received blob chunk"
266                        );
267
268                        // Continue receiving chunks until stream closes
269                    }
270
271                    // Calculate and log transfer statistics
272                    let total_time = start_time.elapsed();
273                    let transfer_rate = if total_time.as_secs() > 0 {
274                        collected_data.len() as f64 / total_time.as_secs_f64() / (1024.0 * 1024.0) // MiB/s
275                    } else {
276                        0.0
277                    };
278
279                    debug!(
280                        blob_id = %request.blob_id,
281                        peer_id = %request.peer_id,
282                        total_size = collected_data.len(),
283                        total_chunks = chunk_count,
284                        transfer_time_secs = total_time.as_secs_f64(),
285                        transfer_rate_mibs = transfer_rate,
286                        "Blob download completed successfully"
287                    );
288
289                    // Emit success event for NodeManager to handle storage
290                    let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloaded {
291                        blob_id: request.blob_id,
292                        context_id: request.context_id,
293                        data: collected_data.clone(),
294                        from_peer: request.peer_id,
295                    });
296
297                    // Return the collected data
298                    Ok(Some(collected_data))
299                } else {
300                    // Emit failure event - blob not found
301                    let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
302                        blob_id: request.blob_id,
303                        context_id: request.context_id,
304                        from_peer: request.peer_id,
305                        error: "Blob not found on peer".to_owned(),
306                    });
307                    Ok(None)
308                }
309            }).await;
310
311            // Handle timeout result
312            if let Ok(result) = transfer_result {
313                result
314            } else {
315                // Overall transfer timeout
316                let _ignored = event_dispatcher.dispatch(NetworkEvent::BlobDownloadFailed {
317                    blob_id: request.blob_id,
318                    context_id: request.context_id,
319                    from_peer: request.peer_id,
320                    error: format!(
321                        "Blob transfer timed out after {} seconds",
322                        BLOB_TRANSFER_TIMEOUT.as_secs()
323                    ),
324                });
325                Err(eyre!("Blob transfer timed out"))
326            }
327        })
328    }
329}
330
331#[cfg(test)]
332mod tests {
333    use super::*;
334
335    #[test]
336    fn test_borsh_serialization() {
337        let test_cases = vec![
338            // Basic case
339            BlobChunk {
340                data: vec![1, 2, 3, 4, 5],
341            },
342            // Empty data
343            BlobChunk { data: vec![] },
344            // Large data
345            BlobChunk {
346                data: vec![42; 1024], // 1KB of data
347            },
348            // Single byte chunk
349            BlobChunk { data: vec![1] },
350        ];
351
352        for (i, chunk) in test_cases.into_iter().enumerate() {
353            let bytes = borsh::to_vec(&chunk)
354                .unwrap_or_else(|e| panic!("Failed to serialize chunk {i}: {e}"));
355
356            let parsed = BlobChunk::try_from_slice(&bytes)
357                .unwrap_or_else(|e| panic!("Failed to deserialize chunk {i}: {e}"));
358
359            assert_eq!(chunk.data, parsed.data, "Data should match for chunk {i}");
360
361            // Verify the serialized size is correct
362            assert_eq!(
363                bytes.len(),
364                // Borsh format: vec length (u32) + vec data
365                4 + chunk.data.len(),
366                "Serialized size should match expected for chunk {i}"
367            );
368        }
369    }
370}
calimero_network/handlers/commands/request_blob.rs

calimero_network/handlers/commands/
request_blob.rs
