From 245637ecb74ca44ba54588e72eb7d31c70d3ebf3 Mon Sep 17 00:00:00 2001
From: Beinan Wang <beinanwang@microsoft.com>
Date: Tue, 20 Oct 2026 23:04:20 +0000
Subject: [PATCH 1/2] fix: respect owned table admission on legacy compaction
 route

---
 .../src/routes/rollouts.rs                    | 51 +++++++++++++++++++
 1 file changed, 51 insertions(+)

diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs
index 6e07183..f0f4494 100644
--- a/crates/lance-context-server/src/routes/rollouts.rs
+++ b/crates/lance-context-server/src/routes/rollouts.rs
@@ -637,6 +637,10 @@ pub async fn compact_rollout(
     Path(name): Path<String>,
     Json(req): Json<CompactRequest>,
 ) -> Result<Json<CompactResponse>, AppError> {
+    // This legacy HTTP path has no durable execution or commit authorizer.
+    // Once a table is draining or owned, only the master's fenced maintenance
+    // path may publish a rewrite. Check before opening or locking its handle.
+    state.merge_executions.legacy_allowed(&name)?;
     let store_lock = state.get_or_open_rollout_store(&name).await?;
 
     let config = if req.target_rows_per_fragment.is_some() || req.materialize_deletions.is_some() {
@@ -798,6 +802,53 @@ mod tests {
     use super::*;
     use crate::state::AppState;
 
+    #[tokio::test]
+    async fn legacy_compaction_cannot_bypass_owned_or_draining_admission() {
+        let dir = TempDir::new().unwrap();
+        let mut state = AppState::new_for_test(dir.path().to_path_buf()).await;
+        state.merge_executions.rollout.owned_targets = vec!["owned".into()];
+        state.merge_executions.rollout.drain_targets = vec!["draining".into()];
+        let state = Arc::new(state);
+        for name in ["owned", "draining", "legacy"] {
+            create_rollout_store(
+                State(state.clone()),
+                Json(CreateRolloutStoreRequest {
+                    name: name.into(),
+                    storage_options: None,
+                }),
+            )
+            .await
+            .unwrap();
+            let handle = state.get_or_open_rollout_store(name).await.unwrap();
+            // A live reader must not make rejection wait for the write lock.
+            let reader = handle.read().await;
+            let version = reader.version();
+            if name != "legacy" {
+                let result = tokio::time::timeout(
+                    std::time::Duration::from_secs(1),
+                    compact_rollout(
+                        State(state.clone()),
+                        Path(name.into()),
+                        Json(CompactRequest::default()),
+                    ),
+                )
+                .await
+                .expect("protected maintenance must reject before acquiring a store lock");
+                assert!(matches!(result, Err(AppError::Overloaded(_))));
+                assert_eq!(reader.version(), version);
+            } else {
+                drop(reader);
+                compact_rollout(
+                    State(state.clone()),
+                    Path(name.into()),
+                    Json(CompactRequest::default()),
+                )
+                .await
+                .expect("unowned legacy compaction remains available");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn blob_miss_does_not_require_exclusive_store_lock() {
         let dir = TempDir::new().unwrap();

From 2e955865465746c50373e287ede347ccee77c02c Mon Sep 17 00:00:00 2001
From: Beinan Wang <beinanwang@microsoft.com>
Date: Tue, 20 Oct 2026 23:14:19 +0000
Subject: [PATCH 2/2] test: verify compaction route responses explicitly

---
 crates/lance-context-server/src/routes/rollouts.rs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs
index f0f4494..8df9514 100644
--- a/crates/lance-context-server/src/routes/rollouts.rs
+++ b/crates/lance-context-server/src/routes/rollouts.rs
@@ -810,7 +810,7 @@ mod tests {
         state.merge_executions.rollout.drain_targets = vec!["draining".into()];
         let state = Arc::new(state);
         for name in ["owned", "draining", "legacy"] {
-            create_rollout_store(
+            let (status, Json(_)) = create_rollout_store(
                 State(state.clone()),
                 Json(CreateRolloutStoreRequest {
                     name: name.into(),
@@ -819,6 +819,7 @@ mod tests {
             )
             .await
             .unwrap();
+            assert_eq!(status, StatusCode::CREATED);
             let handle = state.get_or_open_rollout_store(name).await.unwrap();
             // A live reader must not make rejection wait for the write lock.
             let reader = handle.read().await;
@@ -838,13 +839,14 @@ mod tests {
                 assert_eq!(reader.version(), version);
             } else {
                 drop(reader);
-                compact_rollout(
+                let Json(compacted) = compact_rollout(
                     State(state.clone()),
                     Path(name.into()),
                     Json(CompactRequest::default()),
                 )
                 .await
                 .expect("unowned legacy compaction remains available");
+                assert_eq!(compacted.fragments_removed, 0);
             }
         }
     }