fix(mito2): remove stale compaction status when next compaction is throttled (#8618)

I hereby agree to the terms of the [GreptimeDB
CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md).

## Refer to a related PR or issue link (optional)

Fixes #8617

## What's changed and what's your intention?

**Summary:** Fixes a mito2 compaction scheduler bug where a region
permanently stops compacting after a finished compaction when
`min_compaction_interval` is configured to a non-zero value.

**Root cause:**

1. `CompactionScheduler::on_compaction_finished` falls through without
removing the region's `CompactionStatus` when there is no pending manual
request and no pending DDL, leaving a status with `active_compaction =
None` in `region_status`.
2. Back in `RegionWorkerLoop::handle_compaction_finished`,
`is_compacting()` is now `false`, but the follow-up
`schedule_next_compaction` (whose `Ok(None)` branch is the only normal
path that removes the status) is gated by `min_compaction_interval`.
Since the scheduling timestamp was already updated when the task was
scheduled, a compaction that finishes within the interval skips this
cleanup, and the stale status remains forever.
3. From then on, every compaction trigger hits the "status exists"
branch of `CompactionScheduler::schedule_compaction`, which does not
check whether a compaction is actually active: regular requests are
merged into `status.waiters` (never woken until close/drop/failure) and
manual `StrictWindow` requests are parked in `status.pending_request`
(never executed). The region never compacts again until restart and SSTs
keep accumulating.

Note this only manifests with `min_compaction_interval > 0`; with the
default `0` the gate always passes and `schedule_next_compaction` cleans
up the status.

**How this PR works:**

- Adds `CompactionScheduler::remove_inactive_status(region_id)`, which
removes the status only when it has no active compaction. A
`debug_assert!` guards the invariant that such a status has no
waiters/pending requests/pending DDLs (they can only be enqueued while a
compaction is active, and the worker loop is strictly sequential).
- Calls it from `handle_compaction_finished` when the
`min_compaction_interval` gate skips `schedule_next_compaction`,
restoring the invariant that a status in the map always has an active
compaction. The throttling semantics are unchanged: the next trigger
still has to wait for the interval to pass.

**Tests:**

- `test_remove_inactive_status_keeps_running_compaction`: status with an
active compaction (or no status) is never removed.
- `test_compaction_reschedulable_when_next_compaction_throttled`:
reproduces the reported flow — schedule, finish (stale status remains),
remove it as the throttled worker path does, and verify the region can
be scheduled for compaction again.

**Compatibility:** no API, persisted-format, or wire-format changes;
internal scheduler behavior only.

## PR Checklist
Please convert it to a draft if some of the following conditions are not
met.

- [x] I have written the necessary rustdoc comments.
- [x] I have added the necessary unit tests and integration tests.
- [ ] This PR requires documentation updates.
- [x] API changes are backward compatible.
- [x] Schema or data changes are backward compatible.

src/mito2/src/compaction.rs

@@ -343,6 +343,32 @@ impl CompactionScheduler {
             .unwrap_or(false)
     }
 
+    /// Removes the compaction status of the region if it has no active compaction.
+    ///
+    /// The worker calls this when it decides not to schedule the next compaction
+    /// after a compaction finishes (e.g. throttled by `min_compaction_interval`).
+    /// Otherwise a stale status without an active compaction would remain in the
+    /// map and all subsequent compaction requests would be swallowed as if the
+    /// region were still compacting.
+    ///
+    /// Returns whether the status is removed.
+    pub(crate) fn remove_inactive_status(&mut self, region_id: RegionId) -> bool {
+        let Some(status) = self.region_status.get(&region_id) else {
+            return false;
+        };
+        if status.active_compaction.is_some() {
+            return false;
+        }
+        // A status without an active compaction must have been drained by
+        // `on_compaction_finished`: no waiters, pending requests or DDLs can be
+        // enqueued while no compaction is running.
+        debug_assert!(status.waiters.is_empty());
+        debug_assert!(status.pending_request.is_none());
+        debug_assert!(status.pending_ddl_requests.is_empty());
+        self.region_status.remove(&region_id);
+        true
+    }
+
     /// Schedules next compaction upon a finished compaction.
     /// Returns whether the compaction is scheduled.
     pub(crate) async fn schedule_next_compaction(
@@ -2523,6 +2549,115 @@ mod tests {
         assert!(!scheduler.region_status.contains_key(&region_id));
     }
 
+    #[tokio::test]
+    async fn test_remove_inactive_status_keeps_running_compaction() {
+        let env = SchedulerEnv::new().await;
+        let (tx, _rx) = mpsc::channel(4);
+        let mut scheduler = env.mock_compaction_scheduler(tx);
+        let builder = VersionControlBuilder::new();
+        let version_control = Arc::new(builder.build());
+        let region_id = builder.region_id();
+
+        // No status at all.
+        assert!(!scheduler.remove_inactive_status(region_id));
+
+        scheduler.region_status.insert(
+            region_id,
+            CompactionStatus::new(region_id, version_control, env.access_layer.clone()),
+        );
+        scheduler
+            .region_status
+            .get_mut(&region_id)
+            .unwrap()
+            .start_local_task();
+
+        // A status with an active compaction must not be removed.
+        assert!(!scheduler.remove_inactive_status(region_id));
+        assert!(scheduler.region_status.contains_key(&region_id));
+    }
+
+    #[tokio::test]
+    async fn test_compaction_reschedulable_when_next_compaction_throttled() {
+        let job_scheduler = Arc::new(VecScheduler::default());
+        let env = SchedulerEnv::new().await.scheduler(job_scheduler.clone());
+        let (tx, _rx) = mpsc::channel(4);
+        let mut scheduler = env.mock_compaction_scheduler(tx);
+        let mut builder = VersionControlBuilder::new();
+        let region_id = builder.region_id();
+        let end = 1000 * 1000;
+        // Five overlapping L0 files are enough for the regular picker to create a task.
+        let version_control = Arc::new(
+            builder
+                .push_l0_file(0, end)
+                .push_l0_file(10, end)
+                .push_l0_file(50, end)
+                .push_l0_file(80, end)
+                .push_l0_file(90, end)
+                .build(),
+        );
+        let manifest_ctx = env
+            .mock_manifest_context(version_control.current().version.metadata.clone())
+            .await;
+        let (schema_metadata_manager, kv_backend) = mock_schema_metadata_manager();
+        schema_metadata_manager
+            .register_region_table_info(
+                region_id.table_id(),
+                "test_table",
+                "test_catalog",
+                "test_schema",
+                None,
+                kv_backend,
+            )
+            .await;
+
+        let scheduled = scheduler
+            .schedule_compaction(
+                region_id,
+                Options::Regular(Default::default()),
+                &version_control,
+                &env.access_layer,
+                OptionOutputTx::none(),
+                &manifest_ctx,
+                schema_metadata_manager.clone(),
+                1,
+            )
+            .await
+            .unwrap();
+        assert!(scheduled);
+        assert_eq!(1, job_scheduler.num_jobs());
+
+        // The compaction finishes with no pending request/DDL. The status remains
+        // in the map with no active compaction; the worker then skips
+        // `schedule_next_compaction` because `min_compaction_interval` has not
+        // passed and must remove the stale status instead.
+        let pending_ddls = scheduler
+            .on_compaction_finished(region_id, &manifest_ctx, schema_metadata_manager.clone())
+            .await;
+        assert!(pending_ddls.is_empty());
+        assert!(!scheduler.is_compacting(region_id));
+        assert!(scheduler.region_status.contains_key(&region_id));
+
+        assert!(scheduler.remove_inactive_status(region_id));
+        assert!(!scheduler.region_status.contains_key(&region_id));
+
+        // The region can be scheduled for compaction again.
+        let scheduled = scheduler
+            .schedule_compaction(
+                region_id,
+                Options::Regular(Default::default()),
+                &version_control,
+                &env.access_layer,
+                OptionOutputTx::none(),
+                &manifest_ctx,
+                schema_metadata_manager,
+                1,
+            )
+            .await
+            .unwrap();
+        assert!(scheduled);
+        assert_eq!(2, job_scheduler.num_jobs());
+    }
+
     #[tokio::test]
     async fn test_on_compaction_finished_next_schedule_error_cleans_status() {
         let env = SchedulerEnv::new()

src/mito2/src/worker/handle_compaction.rs

@@ -141,6 +141,13 @@ impl<S> RegionWorkerLoop<S> {
             {
                 region.update_schedule_compaction_millis();
             }
+        } else {
+            // The minimal compaction interval has not passed, so the next compaction
+            // is not scheduled now and the region has no running compaction. Removes
+            // its compaction status, otherwise the stale status would make the
+            // scheduler swallow all subsequent compaction requests and the region
+            // would never compact again.
+            self.compaction_scheduler.remove_inactive_status(region_id);
         }
     }