handle error getting shard files (#1609)

* handle error getting shard files

* note about gc

* lint

* if not metadata, ignore

nucliadb_core/src/lib.rs

@@ -132,6 +132,7 @@ pub fn encapsulate_writer<T>(writer: T) -> Arc<RwLock<T>> {
     Arc::new(RwLock::new(writer))
 }
 
+#[derive(Debug)]
 pub struct IndexFiles {
     pub metadata_files: HashMap<String, Vec<u8>>,
     pub files: Vec<String>,

nucliadb_node/src/replication/replicator.rs

@@ -146,7 +146,8 @@ pub async fn replicate_shard(
 
     debug!("Finished replicating shard: {:?}", shard_state.shard_id);
 
-    // gc after replication to clean up old segments
+    // GC after replication to clean up old segments
+    // We only do this here because GC will not be done otherwise on a secondary
     let sshard = Arc::clone(&shard); // moved shard for gc
     tokio::task::spawn_blocking(move || sshard.gc())
         .await?

nucliadb_node/src/replication/service.rs

@@ -21,7 +21,7 @@ use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use nucliadb_core::tracing::{debug, warn};
+use nucliadb_core::tracing::{debug, error, warn};
 use nucliadb_core::NodeResult;
 use nucliadb_protos::{noderesources, replication};
 use tokio::fs::File;
@@ -232,8 +232,12 @@ impl replication::replication_service_server::ReplicationService for Replication
             .await
             .map_err(|error| {
                 tonic::Status::internal(format!("Error getting shard files: {error:?}"))
-            })
-            .expect("Error getting shard files");
+            });
+            if let Err(error) = shard_files {
+                error!("Error getting shard files: {:?}", error);
+                return;
+            }
+            let shard_files = shard_files.unwrap();
 
             for segment_files in shard_files {
                 for segment_file in segment_files.files {

nucliadb_node/src/shards/shard_writer.rs

@@ -536,6 +536,10 @@ impl ShardWriter {
             vector_read(&self.vector_writer)
                 .get_index_files(ignored_segement_ids.get("vector").unwrap_or(&Vec::new()))?,
         );
+        files.push(
+            relation_read(&self.relation_writer)
+                .get_index_files(ignored_segement_ids.get("relation").unwrap_or(&Vec::new()))?,
+        );
 
         Ok(files)
     }

nucliadb_paragraphs/src/writer.rs

@@ -160,6 +160,12 @@ impl WriterChild for ParagraphWriterService {
         // Should be called along with a lock at a higher level to be safe
         let mut meta_files = HashMap::new();
         let path = self.config.path.join("meta.json");
+        if !path.exists() {
+            return Ok(IndexFiles {
+                metadata_files: meta_files,
+                files: Vec::new(),
+            });
+        }
         meta_files.insert("paragraph/meta.json".to_string(), fs::read(path)?);
 
         let mut files = Vec::new();

nucliadb_relations2/src/writer.rs

@@ -113,6 +113,12 @@ impl WriterChild for RelationsWriterService {
         // Should be called along with a lock at a higher level to be safe
         let mut meta_files = HashMap::new();
         let path = self.config.path.join("meta.json");
+        if !path.exists() {
+            return Ok(IndexFiles {
+                metadata_files: meta_files,
+                files: Vec::new(),
+            });
+        }
         meta_files.insert("text/meta.json".to_string(), fs::read(path)?);
 
         let mut files = Vec::new();

nucliadb_texts/src/writer.rs

@@ -161,6 +161,12 @@ impl WriterChild for TextWriterService {
         // Should be called along with a lock at a higher level to be safe
         let mut meta_files = HashMap::new();
         let path = self.config.path.join("meta.json");
+        if !path.exists() {
+            return Ok(IndexFiles {
+                metadata_files: meta_files,
+                files: Vec::new(),
+            });
+        }
         meta_files.insert("text/meta.json".to_string(), fs::read(path)?);
 
         let mut files = Vec::new();