Assert cluster_name on import

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)?,

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(