Assert cluster_name on import

restatedev · Dec 1, 2024 · 1b817bb · 1b817bb
1 parent 85a316b
commit 1b817bb
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 9 deletions.
diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs
@@ -159,6 +159,7 @@ impl Worker {
             SnapshotRepository::create(
                 snapshots_options,
                 config.common.base_dir().join("pp-snapshots"),
+                config.common.cluster_name().to_owned(),
             )
             .await
             .map_err(BuildError::SnapshotRepository)?,

diff --git a/crates/worker/src/partition/snapshots/repository.rs b/crates/worker/src/partition/snapshots/repository.rs
@@ -60,6 +60,8 @@ pub struct SnapshotRepository {
     prefix: String,
     /// Ingested snapshots staging location.
     staging_dir: PathBuf,
+    /// Expected cluster name for the snapshots in this repository.
+    cluster_name: String,
 }
 
 /// S3 and other stores require a certain minimum size for the parts of a multipart upload. It is an
@@ -101,6 +103,7 @@ impl SnapshotRepository {
     pub async fn create(
         snapshots_options: &SnapshotsOptions,
         staging_dir: PathBuf,
+        cluster_name: String,
     ) -> anyhow::Result<Option<SnapshotRepository>> {
         let mut destination = if let Some(ref destination) = snapshots_options.destination {
             Url::parse(destination).context("Failed parsing snapshot repository URL")?
@@ -160,6 +163,7 @@ impl SnapshotRepository {
             destination,
             prefix,
             staging_dir,
+            cluster_name,
         }))
     }
 
@@ -414,6 +418,13 @@ impl SnapshotRepository {
             ));
         }
 
+        if snapshot_metadata.cluster_name != self.cluster_name {
+            panic!("Snapshot does not match the cluster name of latest snapshot at destination in snapshot id {}! Expected: cluster name=\"{}\", found: \"{}\"",
+                   snapshot_metadata.snapshot_id,
+                   self.cluster_name,
+                   snapshot_metadata.cluster_name);
+        }
+
         // The snapshot ingest directory should be on the same filesystem as the partition store
         // to minimize IO and disk space usage during import.
         let snapshot_dir = TempDir::with_prefix_in(
@@ -729,9 +740,13 @@ mod tests {
             ),
             ..SnapshotsOptions::default()
         };
-        let repository = SnapshotRepository::create(&opts, TempDir::new().unwrap().into_path())
-            .await?
-            .unwrap();
+        let repository = SnapshotRepository::create(
+            &opts,
+            TempDir::new().unwrap().into_path(),
+            "cluster".to_owned(),
+        )
+        .await?
+        .unwrap();
 
         repository.put(&snapshot, source_dir).await?;
 
@@ -766,9 +781,13 @@ mod tests {
             ),
             ..SnapshotsOptions::default()
         };
-        let repository = SnapshotRepository::create(&opts, TempDir::new().unwrap().into_path())
-            .await?
-            .unwrap();
+        let repository = SnapshotRepository::create(
+            &opts,
+            TempDir::new().unwrap().into_path(),
+            "cluster".to_owned(),
+        )
+        .await?
+        .unwrap();
 
         // Write invalid JSON to latest.json
         let latest_path = destination_dir.join(format!("{}/latest.json", PartitionId::MIN));
@@ -829,9 +848,13 @@ mod tests {
         //     ..SnapshotsOptions::default()
         // };
 
-        let repository = SnapshotRepository::create(&opts, TempDir::new().unwrap().into_path())
-            .await?
-            .unwrap();
+        let repository = SnapshotRepository::create(
+            &opts,
+            TempDir::new().unwrap().into_path(),
+            "cluster".to_owned(),
+        )
+        .await?
+        .unwrap();
 
         repository.put(&snapshot1, source_dir.clone()).await?;