Repository: risingwavelabs/risingwave
File: src/storage/backup/src/storage.rs (Rust)

Code before:

    async fn get<S: Metadata>(&self, id: MetaSnapshotId) -> BackupResult<MetaSnapshot<S>> {
        let path = self.get_snapshot_path(id);
        let data = self.store.read(&path, ..).await?;
        MetaSnapshot::decode(&data)
    }

    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().await;
            if manifest.manifest_id > guard.manifest_id {
                *guard = Arc::new(manifest);
            }
        }
        Ok(())
    }

    async fn delete(&self, ids: &[MetaSnapshotId]) -> BackupResult<()> {
        // update manifest first
        let to_delete: HashSet<MetaSnapshotId> = HashSet::from_iter(ids.iter().cloned());
        {
            let mut gurad = self.manifest.write().await;
            let mut new_manifest = (**gurad).clone();
            new_manifest.manifest_id += 1;
            new_manifest
                .snapshot_metadata
                .retain(|m| !to_delete.contains(&m.id));
            self.update_manifest(&new_manifest).await?;
            *gurad = Arc::new(new_manifest);
        }
        let paths = ids
            .iter()
            .map(|id| self.get_snapshot_path(*id))
            .collect_vec();
        self.store.delete_objects(&paths).await?;
        Ok(())
    }
}

impl From<ObjectError> for BackupError {
    fn from(e: ObjectError) -> Self {
        BackupError::BackupStorage(e.into())
    }
}

// #[cfg(test)]
pub async fn unused() -> ObjectStoreMetaSnapshotStorage {

Review comment (suggestion):
```suggestion
            let mut guard = self.manifest.write().await;
```

Code after:
        Ok(())
    }

    async fn get<S: Metadata>(&self, id: MetaSnapshotId) -> BackupResult<MetaSnapshot<S>> {
        let path = self.get_snapshot_path(id);
        let data = self.store.read(&path, ..).await?;
        MetaSnapshot::decode(&data)
    }

    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().await;
            if manifest.manifest_id > guard.manifest_id {
                *guard = Arc::new(manifest);
            }
        }
        Ok(())
    }

    async fn delete(&self, ids: &[MetaSnapshotId]) -> BackupResult<()> {
        let to_delete: HashSet<MetaSnapshotId> = HashSet::from_iter(ids.iter().cloned());
        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))
            .collect_vec();
        self.store.delete_objects(&paths).await?;
        Ok(())
    }
}

impl From<ObjectError> for BackupError {
    fn from(e: ObjectError) -> Self {
        BackupError::BackupStorage(e.into())
    }
}

// #[cfg(test)]
pub async fn unused() -> ObjectStoreMetaSnapshotStorage {
    ObjectStoreMetaSnapshotStorage::new(