fix: add FOR UPDATE row locking to delete_metrics_splits

Mirror the CTE + FOR UPDATE pattern from delete_splits to prevent
stale-state races. Without row locking, a concurrent
mark_metrics_splits_for_deletion can commit between the state read
and the DELETE, causing spurious FailedPrecondition errors and retry
churn.

The new query locks the target rows before reading their state,
reports not-deletable (Staged/Published) and not-found splits
separately, and only deletes when all requested splits are in
MarkedForDeletion state.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs

@@ -2625,23 +2625,31 @@ impl MetastoreService for PostgresqlMetastore {
         if request.split_ids.is_empty() {
             return Ok(EmptyResponse {});
         }
+        let index_uid: IndexUid = request.index_uid().clone();
 
         info!(
             index_uid = %request.index_uid(),
             split_ids = ?request.split_ids,
             "deleting metrics splits"
         );
 
-        // Only delete splits that are marked for deletion
-        // Match the non-metrics delete_splits pattern: distinguish
-        // "not found" (warn + succeed) from "not deletable" (FailedPrecondition).
+        // Match the non-metrics delete_splits pattern: CTE with FOR UPDATE
+        // to lock rows before reading state, avoiding stale-state races under
+        // concurrent mark_metrics_splits_for_deletion. Distinguishes "not found"
+        // (warn + succeed) from "not deletable" (FailedPrecondition).
         const DELETE_SPLITS_QUERY: &str = r#"
             WITH input_splits AS (
                 SELECT input_splits.split_id, metrics_splits.split_state
                 FROM UNNEST($2::text[]) AS input_splits(split_id)
-                LEFT JOIN metrics_splits
-                    ON metrics_splits.index_uid = $1
-                    AND metrics_splits.split_id = input_splits.split_id
+                LEFT JOIN (
+                    SELECT split_id, split_state
+                    FROM metrics_splits
+                    WHERE
+                        index_uid = $1
+                        AND split_id = ANY($2)
+                    FOR UPDATE
+                ) AS metrics_splits
+                USING (split_id)
             ),
             deleted AS (
                 DELETE FROM metrics_splits
@@ -2680,7 +2688,7 @@ impl MetastoreService for PostgresqlMetastore {
             .bind(&request.split_ids)
             .fetch_one(&self.connection_pool)
             .await
-            .map_err(|sqlx_error| convert_sqlx_err(&request.index_uid().index_id, sqlx_error))?;
+            .map_err(|sqlx_error| convert_sqlx_err(&index_uid.index_id, sqlx_error))?;
 
         if !not_deletable_ids.is_empty() {
             let message = format!(