pageserver: assert that uploads don't modify indexed layers

pageserver/src/tenant/remote_timeline_client.rs

@@ -1943,6 +1943,28 @@ impl RemoteTimelineClient {
                 return;
             }
 
+            // Assert that we don't modify a layer that's referenced by the current index.
+            if cfg!(debug_assertions) {
+                let modified = match &task.op {
+                    UploadOp::UploadLayer(layer, layer_metadata, _) => {
+                        vec![(layer.layer_desc().layer_name(), layer_metadata.clone())]
+                    }
+                    UploadOp::Delete(delete) => delete.layers.clone(),
+                    // These don't modify layers.
+                    UploadOp::UploadMetadata { .. } => Vec::new(),
+                    UploadOp::Barrier(_) => Vec::new(),
+                    UploadOp::Shutdown => Vec::new(),
+                };
+                if let Ok(queue) = self.upload_queue.lock().unwrap().initialized_mut() {
+                    for (name, metadata) in modified {
+                        debug_assert!(
+                            !queue.clean.0.references(&name, &metadata),
+                            "layer {name} modified while referenced by index",
+                        );
+                    }
+                }
+            }
+
             let upload_result: anyhow::Result<()> = match &task.op {
                 UploadOp::UploadLayer(ref layer, ref layer_metadata, mode) => {
                     if let Some(OpType::FlushDeletion) = mode {

pageserver/src/tenant/remote_timeline_client/index.rs

@@ -143,6 +143,17 @@ impl IndexPart {
     pub(crate) fn example() -> Self {
         Self::empty(TimelineMetadata::example())
     }
+
+    /// Returns true if the index contains a reference to the given layer (i.e. file path).
+    ///
+    /// TODO: either LayerName or PersistentLayerKey should probably contain information about the
+    /// shard and generation, instead of passing in LayerFileMetadata too.
+    pub fn references(&self, name: &LayerName, metadata: &LayerFileMetadata) -> bool {
+        let Some(index_metadata) = self.layer_metadata.get(name) else {
+            return false;
+        };
+        metadata.shard == index_metadata.shard && metadata.generation == index_metadata.generation
+    }
 }
 
 /// Metadata gathered for each of the layer files.