fix(mito2): report stale compaction execution instead of region closed

When a compaction finishes but its execution no longer matches the
current one, the region may have been reopened or truncated, or the
compaction was superseded. Reporting RegionClosed to waiters is
misleading; introduce a neutral StaleCompactionExecution error (same
Cancelled status code) for this case.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

src/mito2/src/error.rs

@@ -554,6 +554,16 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "Stale compaction execution for region {}, the region may have been reopened, truncated or the compaction was superseded",
+        region_id
+    ))]
+    StaleCompactionExecution {
+        region_id: RegionId,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Region {} is truncated", region_id))]
     RegionTruncated {
         region_id: RegionId,
@@ -1434,6 +1444,7 @@ impl ErrorExt for Error {
             FlushRegion { source, .. } | BuildIndexAsync { source, .. } => source.status_code(),
             RegionDropped { .. } => StatusCode::Cancelled,
             RegionClosed { .. } => StatusCode::Cancelled,
+            StaleCompactionExecution { .. } => StatusCode::Cancelled,
             RegionTruncated { .. } => StatusCode::Cancelled,
             RejectWrite { .. } => StatusCode::StorageUnavailable,
             CompactRegion { source, .. } => source.status_code(),

src/mito2/src/worker/handle_compaction.rs

@@ -20,7 +20,7 @@ use store_api::storage::RegionId;
 
 use crate::compaction::{CompactionExecution, CompactionPickFinished};
 use crate::config::IndexBuildMode;
-use crate::error::{RegionClosedSnafu, RegionNotFoundSnafu};
+use crate::error::{RegionNotFoundSnafu, StaleCompactionExecutionSnafu};
 use crate::metrics::COMPACTION_REQUEST_COUNT;
 use crate::region::MitoRegionRef;
 use crate::request::{
@@ -116,7 +116,7 @@ impl<S> RegionWorkerLoop<S> {
             }
         };
         if !self.is_current_compaction_execution(&region, &request.execution) {
-            request.on_failure(RegionClosedSnafu { region_id }.build());
+            request.on_failure(StaleCompactionExecutionSnafu { region_id }.build());
             return;
         }
         let execution = request.execution.clone();