fix(cubestore): Metastore logs are uploaded without the corresponding snapshot (#6222)

Author: waralexrom

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -4412,7 +4412,7 @@ mod tests {
     use super::table::AggregateColumn;
     use super::*;
     use crate::config::Config;
-    use crate::remotefs::LocalDirRemoteFs;
+    use crate::remotefs::{LocalDirRemoteFs, RemoteFs};
     use futures_timer::Delay;
     use rocksdb::IteratorMode;
     use std::thread::sleep;
@@ -5452,6 +5452,87 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn upload_logs_without_snapshots() {
+        let config = Config::test("upload_logs_without_snapshots");
+
+        let _ = fs::remove_dir_all(config.local_dir());
+        let _ = fs::remove_dir_all(config.remote_dir());
+
+        let services = config.configure().await;
+
+        services.start_processing_loops().await.unwrap();
+        let rocks_meta_store = services.rocks_meta_store.as_ref().unwrap();
+        let remote_fs = services
+            .injector
+            .get_service::<dyn RemoteFs>("original_remote_fs")
+            .await;
+        services
+            .meta_store
+            .create_schema("foo1".to_string(), false)
+            .await
+            .unwrap();
+        rocks_meta_store.run_upload().await.unwrap();
+        services
+            .meta_store
+            .create_schema("foo".to_string(), false)
+            .await
+            .unwrap();
+        rocks_meta_store.run_upload().await.unwrap();
+        let uploaded = remote_fs.list("metastore-").await.unwrap();
+        assert!(uploaded.is_empty());
+
+        rocks_meta_store.upload_check_point().await.unwrap();
+
+        services
+            .meta_store
+            .create_schema("bar".to_string(), false)
+            .await
+            .unwrap();
+
+        rocks_meta_store.run_upload().await.unwrap();
+
+        let uploaded = remote_fs.list("metastore-").await.unwrap();
+
+        let logs_uploaded = uploaded
+            .into_iter()
+            .filter(|n| n.contains("-logs"))
+            .collect::<Vec<_>>();
+
+        assert_eq!(logs_uploaded.len(), 1);
+
+        rocks_meta_store.run_upload().await.unwrap();
+
+        let uploaded = remote_fs.list("metastore-").await.unwrap();
+
+        let logs_uploaded = uploaded
+            .into_iter()
+            .filter(|n| n.contains("-logs"))
+            .collect::<Vec<_>>();
+
+        assert_eq!(logs_uploaded.len(), 1);
+
+        services
+            .meta_store
+            .create_schema("bar2".to_string(), false)
+            .await
+            .unwrap();
+
+        rocks_meta_store.run_upload().await.unwrap();
+
+        let uploaded = remote_fs.list("metastore-").await.unwrap();
+
+        let logs_uploaded = uploaded
+            .into_iter()
+            .filter(|n| n.contains("-logs"))
+            .collect::<Vec<_>>();
+
+        assert_eq!(logs_uploaded.len(), 2);
+
+        let _ = fs::remove_dir_all(config.local_dir());
+        let _ = fs::remove_dir_all(config.remote_dir());
+    }
+
     #[tokio::test]
     async fn log_replay_ordering() {
         {

rust/cubestore/cubestore/src/metastore/rocks_store.rs

@@ -511,6 +511,7 @@ pub struct RocksStore {
     pub(crate) write_completed_notify: Arc<Notify>,
     last_upload_seq: Arc<RwLock<u64>>,
     last_check_seq: Arc<RwLock<u64>>,
+    snapshot_uploaded: Arc<RwLock<bool>>,
     snapshots_upload_stopped: Arc<AsyncMutex<bool>>,
     pub(crate) cached_tables: Arc<Mutex<Option<Arc<Vec<TablePath>>>>>,
     rw_loop_tx: std::sync::mpsc::SyncSender<
@@ -579,6 +580,7 @@ impl RocksStore {
             listeners: Arc::new(RwLock::new(listeners)),
             metastore_fs,
             last_checkpoint_time: Arc::new(RwLock::new(SystemTime::now())),
+            snapshot_uploaded: Arc::new(RwLock::new(false)),
             write_notify: Arc::new(Notify::new()),
             write_completed_notify: Arc::new(Notify::new()),
             last_upload_seq: Arc::new(RwLock::new(db_arc.latest_sequence_number())),
@@ -766,13 +768,15 @@ impl RocksStore {
                 seq_numbers.iter().max().map(|v| *v),
             )
         };
-
         if max.is_some() {
-            let checkpoint_time = self.last_checkpoint_time.read().await;
-            let dir_name = format!("{}-logs", self.get_store_path(&checkpoint_time));
-            self.metastore_fs
-                .upload_log(&dir_name, min.unwrap(), &serializer)
-                .await?;
+            let snapshot_uploaded = self.snapshot_uploaded.read().await;
+            if *snapshot_uploaded {
+                let checkpoint_time = self.last_checkpoint_time.read().await;
+                let dir_name = format!("{}-logs", self.get_store_path(&checkpoint_time));
+                self.metastore_fs
+                    .upload_log(&dir_name, min.unwrap(), &serializer)
+                    .await?;
+            }
             let mut seq = self.last_upload_seq.write().await;
             *seq = max.unwrap();
             self.write_completed_notify.notify_waiters();
@@ -785,11 +789,10 @@ impl RocksStore {
         {
             info!("Uploading {} check point", self.details.get_name());
             self.upload_check_point().await?;
+            let mut check_seq = self.last_check_seq.write().await;
+            *check_seq = last_db_seq;
         }
 
-        let mut check_seq = self.last_check_seq.write().await;
-        *check_seq = last_db_seq;
-
         info!(
             "Persisting {} snapshot: done ({:?})",
             self.details.get_name(),
@@ -813,6 +816,8 @@ impl RocksStore {
             self.metastore_fs
                 .upload_checkpoint(remote_path, checkpoint_path)
                 .await?;
+            let mut snapshot_uploaded = self.snapshot_uploaded.write().await;
+            *snapshot_uploaded = true;
             self.write_completed_notify.notify_waiters();
         }
         Ok(())