From 293687de5f3e764d169234446989d6d889dc01b7 Mon Sep 17 00:00:00 2001
From: Alex Chi Z <chi@neon.tech>
Date: Wed, 2 Jul 2025 14:38:40 -0700
Subject: [PATCH] revert + add tests

Signed-off-by: Alex Chi Z <chi@neon.tech>
---
 pageserver/src/tenant/upload_queue.rs | 45 +++++++++++++++++++++++++--
 1 file changed, 42 insertions(+), 3 deletions(-)

diff --git a/pageserver/src/tenant/upload_queue.rs b/pageserver/src/tenant/upload_queue.rs
index 27d1dbdf74..e80a90c0e9 100644
--- a/pageserver/src/tenant/upload_queue.rs
+++ b/pageserver/src/tenant/upload_queue.rs
@@ -544,9 +544,12 @@ impl UploadOp {
                 let uname = u.layer_desc().layer_name();
                 !i.references(&uname, umeta) && !index.references(&uname, umeta)
             }
-            (UploadOp::Delete(_), UploadOp::UploadMetadata { .. })
-            | (UploadOp::UploadMetadata { .. }, UploadOp::Delete(_)) => false,
-
+            (UploadOp::Delete(d), UploadOp::UploadMetadata { uploaded: i })
+            | (UploadOp::UploadMetadata { uploaded: i }, UploadOp::Delete(d)) => {
+                d.layers.iter().all(|(dname, dmeta)| {
+                    !i.references(dname, dmeta) && !index.references(dname, dmeta)
+                })
+            }
             // Indexes can never bypass each other. They can coalesce though, and
             // `UploadQueue::next_ready()` currently does this when possible.
             (UploadOp::UploadMetadata { .. }, UploadOp::UploadMetadata { .. }) => false,
@@ -1397,4 +1400,40 @@ mod tests {
 
         Ok(())
     }
+
+    /// Delete should be done after the index_part is uploaded.
+    #[test]
+    fn schedule_upload_index_bypass() -> anyhow::Result<()> {
+        let mut queue = UploadQueue::Uninitialized;
+        let mut index_part = IndexPart::example();
+
+        let tli = make_timeline();
+        let layer0 = make_layer(
+            &tli,
+            "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
+        );
+        index_part
+            .layer_metadata
+            .insert(layer0.layer_desc().layer_name(), layer0.metadata());
+        let queue = queue.initialize_with_current_remote_index_part(&index_part, 0)?;
+        let mut index_part_2 = index_part.clone();
+        index_part_2.layer_metadata.clear();
+
+        let ops = [
+            UploadOp::UploadMetadata {
+                uploaded: Box::new(index_part_2),
+            },
+            UploadOp::Delete(Delete {
+                layers: vec![(layer0.layer_desc().layer_name(), layer0.metadata())],
+            }),
+        ];
+
+        queue.queued_operations.extend(ops.clone());
+
+        let tasks = queue.schedule_ready();
+        assert_same_ops(tasks.iter().map(|t| &t.op), [&ops[0]]);
+        assert_eq!(queue.queued_operations.len(), 1);
+
+        Ok(())
+    }
 }