Stream all available blobs in DownloadBlobs and skip missing

linera-core/src/client/mod.rs

@@ -843,15 +843,15 @@ impl<Env: Environment> Client<Env> {
         // block bodies, so they don't need to be fetched from a validator. The
         // chain worker resolves them from `Block::created_blobs()` during
         // `handle_certificate`.
-        let created_blob_ids: BTreeSet<BlobId> = certificates
+        let created_blob_ids = certificates
             .iter()
             .flat_map(|certificate| certificate.value().block().created_blob_ids())
-            .collect();
-        let required_blob_ids: Vec<_> = certificates
+            .collect::<BTreeSet<BlobId>>();
+        let required_blob_ids = certificates
             .iter()
             .flat_map(|certificate| certificate.value().required_blob_ids())
             .filter(|blob_id| !created_blob_ids.contains(blob_id))
-            .collect();
+            .collect::<Vec<_>>();
 
         match self
             .local_node

linera-service/src/proxy/grpc.rs

@@ -632,15 +632,13 @@ where
             .read_blobs(&blob_ids)
             .await
             .map_err(Self::view_error_to_status)?;
-        let stream = futures::stream::iter(blob_ids.into_iter().zip(blobs).map(
-            |(blob_id, maybe_blob)| {
-                let blob = maybe_blob
-                    .map(Arc::unwrap_or_clone)
-                    .ok_or_else(|| Status::not_found(format!("Blob not found {blob_id}")))?;
-                BlobContent::try_from(blob.into_content())
-                    .map_err(|err| Status::internal(err.to_string()))
-            },
-        ));
+        let stream = futures::stream::iter(blobs.into_iter().filter_map(|maybe_blob| {
+            let blob = maybe_blob?;
+            Some(
+                BlobContent::try_from(blob.content().clone())
+                    .map_err(|err| Status::internal(err.to_string())),
+            )
+        }));
         Ok(Response::new(Box::pin(stream)))
     }
 

linera-service/src/proxy/main.rs

@@ -300,16 +300,12 @@ where
                 })));
             }
         };
-        let messages =
-            blob_ids
-                .into_iter()
-                .zip(blobs)
-                .map(|(blob_id, maybe_blob)| match maybe_blob {
-                    Some(blob) => RpcMessage::DownloadBlobResponse(Box::new(
-                        Arc::unwrap_or_clone(blob).into_content(),
-                    )),
-                    None => RpcMessage::Error(Box::new(NodeError::BlobsNotFound(vec![blob_id]))),
-                });
+        let messages = blobs.into_iter().filter_map(|maybe_blob| {
+            let blob = maybe_blob?;
+            Some(RpcMessage::DownloadBlobResponse(Box::new(
+                blob.content().clone(),
+            )))
+        });
         Some(Box::pin(futures::stream::iter(messages)))
     }
 }