feat(ingestion_pipeline): concurrency improvements

* Remove the clone from the node cache
* Run the node cache futures on tokio
* When flattening streams, do so unordered

swiftide/src/ingestion/ingestion_pipeline.rs

@@ -78,24 +78,20 @@ impl IngestionPipeline {
         let cache = Arc::new(cache);
         self.stream = self
             .stream
-            .try_filter(move |node| {
+            .try_filter_map(move |node| {
                 let cache = Arc::clone(&cache);
-                let node = node.clone();
                 let current_span = tracing::Span::current();
                 tokio::spawn(current_span.in_scope(|| async move {
                     if !cache.get(&node).await {
                         cache.set(&node).await;
                         tracing::debug!("Node not in cache, passing through");
-                        true
+                        Some(node)
                     } else {
                         tracing::debug!("Node in cache, skipping");
-                        false
+                        None
                     }
                 }))
-                .unwrap_or_else(|e| {
-                    tracing::error!("Error filtering cached node: {:?}", e);
-                    true
-                })
+                .map_err(anyhow::Error::from)
             })
             .boxed();
         self
@@ -159,7 +155,7 @@ impl IngestionPipeline {
             })
             .err_into::<anyhow::Error>()
             .try_buffer_unordered(self.concurrency)
-            .try_flatten()
+            .try_flatten_unordered(self.concurrency)
             .boxed();
         self
     }
@@ -185,9 +181,8 @@ impl IngestionPipeline {
                 )
                 .map_err(anyhow::Error::from)
             })
-            .err_into::<anyhow::Error>()
             .try_buffer_unordered(self.concurrency)
-            .try_flatten()
+            .try_flatten_unordered(self.concurrency)
             .boxed();
 
         self