fix(meta): fix race condition during metadata backup (#24269) (#24335)

Co-authored-by: zwang28 <[email protected]>

Author: risingwave-ci

Cargo.lock

@@ -72,7 +72,7 @@ impl BackupService for BackupServiceImpl {
         _request: Request<GetMetaSnapshotManifestRequest>,
     ) -> Result<Response<GetMetaSnapshotManifestResponse>, Status> {
         Ok(Response::new(GetMetaSnapshotManifestResponse {
-            manifest: Some(self.backup_manager.manifest().deref().into()),
+            manifest: Some(self.backup_manager.manifest().await.deref().into()),
         }))
     }
 }

src/meta/service/src/backup_service.rs

@@ -323,7 +323,7 @@ impl NotificationServiceImpl {
             .on_current_version(|version| version.into())
             .await;
         let hummock_write_limits = self.hummock_manager.write_limits().await;
-        let meta_backup_manifest_id = self.backup_manager.manifest().manifest_id;
+        let meta_backup_manifest_id = self.backup_manager.manifest().await.manifest_id;
         let cluster_resource = self.get_cluster_resource().await;
 
         Ok(MetaSnapshot {

src/meta/service/src/notification_service.rs

@@ -216,6 +216,7 @@ impl BackupManager {
             .load()
             .0
             .manifest()
+            .await
             .snapshot_metadata
             .len();
         if current_number > MAX_META_SNAPSHOT_NUM {
@@ -266,7 +267,7 @@ impl BackupManager {
                     .notify_hummock_without_version(
                         Operation::Update,
                         Info::MetaBackupManifestId(MetaBackupManifestId {
-                            id: self.backup_store.load().0.manifest().manifest_id,
+                            id: self.backup_store.load().0.manifest().await.manifest_id,
                         }),
                     );
                 self.latest_job_info.store(Arc::new((
@@ -308,26 +309,27 @@ impl BackupManager {
             .notify_hummock_without_version(
                 Operation::Update,
                 Info::MetaBackupManifestId(MetaBackupManifestId {
-                    id: self.backup_store.load().0.manifest().manifest_id,
+                    id: self.backup_store.load().0.manifest().await.manifest_id,
                 }),
             );
         Ok(())
     }
 
     /// List id of all objects required by backups.
-    pub fn list_pinned_object_ids(&self) -> HashSet<HummockRawObjectId> {
+    pub async fn list_pinned_object_ids(&self) -> HashSet<HummockRawObjectId> {
         self.backup_store
             .load()
             .0
             .manifest()
+            .await
             .snapshot_metadata
             .iter()
             .flat_map(|s| s.objects.iter().copied())
             .collect()
     }
 
-    pub fn manifest(&self) -> Arc<MetaSnapshotManifest> {
-        self.backup_store.load().0.manifest()
+    pub async fn manifest(&self) -> Arc<MetaSnapshotManifest> {
+        self.backup_store.load().0.manifest().await
     }
 }
 

src/meta/src/backup_restore/backup_manager.rs

@@ -146,7 +146,7 @@ async fn restore_impl(
         Some(b) => b,
     };
     let target_id = opts.meta_snapshot_id;
-    let snapshot_list = &backup_store.manifest().snapshot_metadata;
+    let snapshot_list = &backup_store.manifest().await.snapshot_metadata;
     let snapshot = match snapshot_list.iter().find(|m| m.id == target_id) {
         None => {
             return Err(BackupError::Other(anyhow::anyhow!(

src/meta/src/backup_restore/restore.rs

@@ -38,7 +38,7 @@ impl LoaderV2 {
 #[async_trait::async_trait]
 impl Loader<MetadataV2> for LoaderV2 {
     async fn load(&self, target_id: MetaSnapshotId) -> BackupResult<MetaSnapshot<MetadataV2>> {
-        let snapshot_list = &self.backup_store.manifest().snapshot_metadata;
+        let snapshot_list = &self.backup_store.manifest().await.snapshot_metadata;
         let mut target_snapshot: MetaSnapshotV2 = self.backup_store.get(target_id).await?;
         tracing::debug!(
             "snapshot {} before rewrite:\n{}",

src/meta/src/backup_restore/restore_impl/v2.rs

@@ -318,11 +318,10 @@ impl HummockManager {
         if object_ids.is_empty() {
             return Ok(0);
         }
-        // It's crucial to get pinned_by_metadata_backup only after object_ids.
-        let pinned_by_metadata_backup = backup_manager
-            .as_ref()
-            .map(|b| b.list_pinned_object_ids())
-            .unwrap_or_default();
+        let pinned_by_metadata_backup = match backup_manager.as_ref() {
+            Some(b) => b.list_pinned_object_ids().await,
+            None => HashSet::default(),
+        };
         // It's crucial to collect_min_uncommitted_object_id (i.e. `min_object_id`) only after LIST object store (i.e. `object_ids`).
         // Because after getting `min_object_id`, new compute nodes may join and generate new uncommitted objects that are not covered by `min_sst_id`.
         // By getting `min_object_id` after `object_ids`, it's ensured `object_ids` won't include any objects from those new compute nodes.
@@ -516,7 +515,7 @@ impl HummockManager {
             return Ok(());
         };
         // Objects pinned by either meta backup or time travel should be filtered out.
-        let backup_pinned: HashSet<_> = backup_manager.list_pinned_object_ids();
+        let backup_pinned: HashSet<_> = backup_manager.list_pinned_object_ids().await;
         // The version_pinned is obtained after the candidate object_ids for deletion, which is new enough for filtering purpose.
         let version_pinned = {
             let versioning = self.versioning.read().await;

src/meta/src/hummock/manager/gc.rs

@@ -12,7 +12,6 @@ anyhow = "1"
 async-trait = "0.1"
 bytes = { version = "1", features = ["serde"] }
 itertools = { workspace = true }
-parking_lot = { workspace = true }
 prost = { workspace = true }
 risingwave_common = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
@@ -22,6 +21,7 @@ risingwave_pb = { workspace = true }
 serde = { workspace = true }
 serde_json = "1"
 thiserror = { workspace = true }
+tokio = { version = "0.2", package = "madsim-tokio" }
 twox-hash = "2"
 
 [dev-dependencies]

src/storage/backup/Cargo.toml

@@ -21,6 +21,7 @@ use risingwave_object_store::object::object_metrics::ObjectStoreMetrics;
 use risingwave_object_store::object::{
     InMemObjectStore, MonitoredObjectStore, ObjectError, ObjectStoreImpl, ObjectStoreRef,
 };
+use tokio::sync::RwLock;
 
 use crate::meta_snapshot::{MetaSnapshot, Metadata};
 use crate::{
@@ -42,7 +43,7 @@ pub trait MetaSnapshotStorage: 'static + Sync + Send {
     async fn get<S: Metadata>(&self, id: MetaSnapshotId) -> BackupResult<MetaSnapshot<S>>;
 
     /// Gets local snapshot manifest.
-    fn manifest(&self) -> Arc<MetaSnapshotManifest>;
+    async fn manifest(&self) -> Arc<MetaSnapshotManifest>;
 
     /// Refreshes local snapshot manifest.
     async fn refresh_manifest(&self) -> BackupResult<()>;
@@ -55,7 +56,7 @@ pub trait MetaSnapshotStorage: 'static + Sync + Send {
 pub struct ObjectStoreMetaSnapshotStorage {
     path: String,
     store: ObjectStoreRef,
-    manifest: Arc<parking_lot::RwLock<Arc<MetaSnapshotManifest>>>,
+    manifest: Arc<RwLock<Arc<MetaSnapshotManifest>>>,
 }
 
 // TODO #6482: purge stale snapshots that is not in manifest.
@@ -70,13 +71,18 @@ impl ObjectStoreMetaSnapshotStorage {
         Ok(instance)
     }
 
-    async fn update_manifest(&self, new_manifest: MetaSnapshotManifest) -> BackupResult<()> {
+    async fn update_manifest(
+        &self,
+        update: impl FnOnce(MetaSnapshotManifest) -> MetaSnapshotManifest,
+    ) -> BackupResult<()> {
+        let mut guard = self.manifest.write().await;
+        let new_manifest = update((**guard).clone());
         let bytes =
             serde_json::to_vec(&new_manifest).map_err(|e| BackupError::Encoding(e.into()))?;
         self.store
             .upload(&self.get_manifest_path(), bytes.into())
             .await?;
-        *self.manifest.write() = Arc::new(new_manifest);
+        *guard = Arc::new(new_manifest);
         Ok(())
     }
 
@@ -124,19 +130,17 @@ impl MetaSnapshotStorage for ObjectStoreMetaSnapshotStorage {
     ) -> BackupResult<()> {
         let path = self.get_snapshot_path(snapshot.id);
         self.store.upload(&path, snapshot.encode()?.into()).await?;
-
-        // update manifest last
-        let mut new_manifest = (**self.manifest.read()).clone();
-        new_manifest.manifest_id += 1;
-        new_manifest
-            .snapshot_metadata
-            .push(MetaSnapshotMetadata::new(
+        self.update_manifest(|mut manifest: MetaSnapshotManifest| {
+            manifest.manifest_id += 1;
+            manifest.snapshot_metadata.push(MetaSnapshotMetadata::new(
                 snapshot.id,
                 snapshot.metadata.hummock_version_ref(),
                 snapshot.format_version,
                 remarks,
             ));
-        self.update_manifest(new_manifest).await?;
+            manifest
+        })
+        .await?;
         Ok(())
     }
 
@@ -146,13 +150,13 @@ impl MetaSnapshotStorage for ObjectStoreMetaSnapshotStorage {
         MetaSnapshot::decode(&data)
     }
 
-    fn manifest(&self) -> Arc<MetaSnapshotManifest> {
-        self.manifest.read().clone()
+    async fn manifest(&self) -> Arc<MetaSnapshotManifest> {
+        self.manifest.read().await.clone()
     }
 
     async fn refresh_manifest(&self) -> BackupResult<()> {
         if let Some(manifest) = self.get_manifest().await? {
-            let mut guard = self.manifest.write();
+            let mut guard = self.manifest.write().await;
             if manifest.manifest_id > guard.manifest_id {
                 *guard = Arc::new(manifest);
             }
@@ -161,15 +165,15 @@ impl MetaSnapshotStorage for ObjectStoreMetaSnapshotStorage {
     }
 
     async fn delete(&self, ids: &[MetaSnapshotId]) -> BackupResult<()> {
-        // update manifest first
         let to_delete: HashSet<MetaSnapshotId> = HashSet::from_iter(ids.iter().cloned());
-        let mut new_manifest = (**self.manifest.read()).clone();
-        new_manifest.manifest_id += 1;
-        new_manifest
-            .snapshot_metadata
-            .retain(|m| !to_delete.contains(&m.id));
-        self.update_manifest(new_manifest).await?;
-
+        self.update_manifest(|mut manifest: MetaSnapshotManifest| {
+            manifest.manifest_id += 1;
+            manifest
+                .snapshot_metadata
+                .retain(|m| !to_delete.contains(&m.id));
+            manifest
+        })
+        .await?;
         let paths = ids
             .iter()
             .map(|id| self.get_snapshot_path(*id))

src/storage/backup/src/storage.rs

@@ -143,7 +143,7 @@ impl BackupReader {
             let expect_manifest_id = expect_manifest_id.unwrap();
             // Use the same store throughout one run.
             let current_store = backup_reader.store.load_full();
-            let previous_id = current_store.0.manifest().manifest_id;
+            let previous_id = current_store.0.manifest().await.manifest_id;
             if expect_manifest_id <= previous_id {
                 continue;
             }
@@ -161,6 +161,7 @@ impl BackupReader {
             let manifest: HashSet<MetaSnapshotId> = current_store
                 .0
                 .manifest()
+                .await
                 .snapshot_metadata
                 .iter()
                 .map(|s| s.id)
@@ -192,6 +193,7 @@ impl BackupReader {
         let Some(snapshot_metadata) = current_store
             .0
             .manifest()
+            .await
             .snapshot_metadata
             .iter()
             .find(|v| {