Local compaction cancellation can leave newly written SST files outside the manifest
C-bug
### What type of bug is this?
Other
### What subsystems are affected?
Distributed Cluster, Storage Engine, Datanode
### Minimal reproduce step
Analyzed source version: 4f5dccf6bb1a9174918a8641b78dddad028588d6.
There is a cancellation window in local compaction after new SST files have been physically written but before the compaction edit is committed to the region manifest. If cancellation is accepted in that window, the task reports `CompactionCancelled` and drops the only local `MergeOutput` that names the new files. Those files are therefore outside the manifest and outside the normal file lifecycle.
The source-level sequence is:
1. A local compaction writes one or more output SST/index files and records them in `MergeOutput.files_to_add`.
2. Before the task enters the non-cancellable commit stage, another region lifecycle operation requests compaction cancellation.
3. The compaction task observes the cancellation before `update_manifest(merge_output).await`.
4. The task reports `CompactionCancelled` and drops `merge_output`.
5. The manifest is not updated, `version_control.apply_edit(...)` is not run, and the file purger/reference path never receives the new `FileMeta`.
The normal production cancellation source is the enter-staging path:
```rust
// src/mito2/src/worker/handle_enter_staging.rs:102-118
match self.compaction_scheduler.request_cancel(region_id) {
RequestCancelResult::CancelIssued
| RequestCancelResult::AlreadyCancelling
| RequestCancelResult::TooLateToCancel => {
// keep the DDL pending until the current task finishes or acknowledges cancellation.
self.compaction_scheduler
.add_ddl_request_to_pending(...);
return;
}
RequestCancelResult::NotRunning => {}
}
```
The local compaction state accepts cancellation until `mark_commit_started()` succeeds:
```rust
// src/mito2/src/compaction.rs:704-725
pub(crate) fn mark_commit_started(&self) -> bool {
let mut commit_started = self.commit_started.lock().unwrap();
if self.cancel_handle.is_cancelled() {
return false;
}
*commit_started = true;
true
}
pub(crate) fn request_cancel(&self) -> RequestCancelResult {
let commit_started = self.commit_started.lock().unwrap();
if *commit_started {
return RequestCancelResult::TooLateToCancel;
}
if self.cancel_handle.is_cancelled() {
return RequestCancelResult::AlreadyCancelling;
}
self.cancel_handle.cancel();
RequestCancelResult::CancelIssued
}
```
The compaction task then has a vulnerable transition: `merge_output` exists, `on_sst_files_written` may already observe it, but `mark_commit_started()` can still reject commit and convert the run into a cancellation result:
```rust
// src/mito2/src/compaction/task.rs:319-337
let notify = match CancellableFuture::new(
async { self.handle_compaction().await },
cancel_handle,
)
.await
{
Ok(Ok(merge_output)) => {
self.invoke_sst_hook(&merge_output).await;
// Stop accepting cancellation once we are about to publish the compaction edit.
if !self.state.mark_commit_started() {
BackgroundNotify::CompactionCancelled(...)
} else {
match self.update_manifest(merge_output).await { ... }
}
}
Err(_) => {
BackgroundNotify::CompactionCancelled(...)
}
...
}
```
The output files are created before the manifest edit is written:
```rust
// src/mito2/src/compaction/compactor.rs:461-487
let sst_infos = compaction_region
.access_layer
.write_sst(
SstWriteRequest {
op_type: OperationType::Compact,
...
},
&write_opts,
&mut metrics,
)
.await?;
// src/mito2/src/compaction/compactor.rs:499-531
let output_files = sst_infos
.iter()
.map(|sst_info| FileMeta {
file_id: sst_info.file_id,
...
})
.collect::<Vec<_>>();
```
The ownership/publication step happens later in `update_manifest()`:
```rust
// src/mito2/src/compaction/compactor.rs:700-724
async fn update_manifest(
&self,
compaction_region: &CompactionRegion,
merge_output: MergeOutput,
) -> Result<(RegionEdit, ManifestVersion)> {
let edit = RegionEdit {
files_to_add: merge_output.files_to_add,
files_to_remove: merge_output.files_to_remove,
...
};
let action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit.clone()));
// TODO: We might leak files if we fail to update manifest. We can add a cleanup task to remove them later.
let manifest_version = compaction_region
.manifest_ctx
.update_manifest_for_compaction(action_list)
.await?;
Ok((edit, manifest_version))
}
```
After successful compaction, the worker applies the edit and the version path creates file handles that notify the purger/reference owner:
```rust
// src/mito2/src/worker/handle_compaction.rs:84-88
region.version_control.apply_edit(
Some(request.edit.clone()),
&[],
region.file_purger.clone(),
);
// src/mito2/src/sst/file.rs:613-619
fn new(
meta: FileMeta,
file_purger: FilePurgerRef,
primary_key_range: Option<(Bytes, Bytes)>,
) -> FileHandleInner {
file_purger.new_file(&meta);
...
}
```
The cancellation path does not receive or clean `MergeOutput`:
```rust
// src/mito2/src/worker/handle_compaction.rs:147-167
pub(crate) async fn handle_compaction_cancelled(
&mut self,
region_id: RegionId,
request: CompactionCancelled,
) {
request.on_success();
let mut pending_ddls = match self.regions.get_region(region_id) {
Some(_) => {
self.compaction_scheduler
.on_compaction_cancelled(region_id)
.await
}
None => Vec::new(),
};
self.handle_ddl_requests(&mut pending_ddls).await;
}
// src/mito2/src/compaction.rs:935-949
fn on_cancel(mut self) -> Vec<SenderDdlRequest> {
for waiter in self.waiters.drain(..) {
waiter.send(CompactionCancelledSnafu.fail());
}
...
std::mem::take(&mut self.pending_ddl_requests)
}
```
A deterministic whitebox validation test is included below. It pauses exactly after the compaction output is observed as written and before `mark_commit_started()`, then requests cancellation through the real `LocalCompactionState`. The current behavior is `CompactionCancelled`, no manifest update call, and no manifest ownership for the written output.
<details>
<summary>Whitebox test patch for <code>src/mito2/src/compaction/task.rs</code></summary>
```diff
diff --git a/src/mito2/src/compaction/task.rs b/src/mito2/src/compaction/task.rs
index a8da92196..814d7b4c1 100644
--- a/src/mito2/src/compaction/task.rs
+++ b/src/mito2/src/compaction/task.rs
@@ -389,10 +389,39 @@ impl CompactionTask for CompactionTaskImpl {
#[cfg(test)]
mod tests {
- use store_api::storage::FileId;
-
- use crate::compaction::picker::PickerOutput;
+ use std::fmt;
+ use std::sync::atomic::{AtomicUsize, Ordering};
+ use std::sync::{Arc, Mutex};
+ use std::time::{Duration, Instant};
+
+ use common_base::Plugins;
+ use common_memory_manager::OnExhaustedPolicy;
+ use store_api::ManifestVersion;
+ use store_api::metadata::RegionMetadataRef;
+ use store_api::storage::{FileId, RegionId};
+ use tokio::sync::{Notify, mpsc};
+ use tokio::time::timeout;
+
+ use super::CompactionTaskImpl;
+ use crate::cache::CacheManager;
+ use crate::compaction::compactor::{
+ CompactionRegion, CompactionVersion, Compactor, MergeOutput,
+ };
+ use crate::compaction::memory_manager::new_compaction_memory_manager;
+ use crate::compaction::picker::{CompactionTask, PickerOutput};
use crate::compaction::test_util::new_file_handle;
+ use crate::compaction::{LocalCompactionState, RequestCancelResult};
+ use crate::config::MitoConfig;
+ use crate::engine::region_hook::{RegionHook, RegionHookRef, SstFileInfo};
+ use crate::error::Result;
+ use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
+ use crate::region::options::RegionOptions;
+ use crate::request::{BackgroundNotify, WorkerRequest};
+ use crate::sst::file::FileMeta;
+ use crate::sst::version::SstVersion;
+ use crate::test_util::memtable_util::metadata_for_test;
+ use crate::test_util::scheduler_util::SchedulerEnv;
+ use crate::worker::WorkerListener;
#[test]
fn test_picker_output_with_expired_ssts() {
@@ -439,6 +468,228 @@ mod tests {
assert!(picker_output.expired_ssts.is_empty());
}
+ fn dummy_file_meta(region_id: RegionId) -> FileMeta {
+ FileMeta {
+ region_id,
+ file_id: FileId::random(),
+ file_size: 1024,
+ ..Default::default()
+ }
+ }
+
+ async fn new_test_compaction_region(hook: RegionHookRef) -> CompactionRegion {
+ let env = SchedulerEnv::new().await;
+ let metadata = metadata_for_test();
+ let manifest_ctx = env.mock_manifest_context(metadata.clone()).await;
+ let plugins = Plugins::new();
+ plugins.insert(hook);
+
+ CompactionRegion {
+ region_id: RegionId::new(1, 1),
+ region_options: RegionOptions::default(),
+ engine_config: Arc::new(MitoConfig::default()),
+ region_metadata: metadata.clone(),
+ cache_manager: Arc::new(CacheManager::default()),
+ access_layer: env.access_layer.clone(),
+ manifest_ctx,
+ current_version: CompactionVersion {
+ metadata,
+ options: RegionOptions::default(),
+ ssts: Arc::new(SstVersion::new()),
+ compaction_time_window: None,
+ },
+ file_purger: None,
+ ttl: None,
+ max_parallelism: 1,
+ plugins,
+ }
+ }
+
+ struct PausingSstHook {
+ reached: Arc<Notify>,
+ release: Arc<Notify>,
+ observed_files: Arc<Mutex<Vec<FileId>>>,
+ }
+
+ impl fmt::Debug for PausingSstHook {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PausingSstHook").finish()
+ }
+ }
+
+ #[async_trait::async_trait]
+ impl RegionHook for PausingSstHook {
+ async fn on_sst_files_written(
+ &self,
+ _region_id: RegionId,
+ _region_metadata: &RegionMetadataRef,
+ files: &[SstFileInfo<'_>],
+ ) {
+ self.observed_files
+ .lock()
+ .unwrap()
+ .extend(files.iter().map(|file| file.file_meta.file_id));
+ self.reached.notify_one();
+ self.release.notified().await;
+ }
+ }
+
+ struct WrittenOutputCompactor {
+ output_file: FileMeta,
+ written_files: Arc<Mutex<Vec<FileId>>>,
+ update_manifest_calls: Arc<AtomicUsize>,
+ }
+
+ #[async_trait::async_trait]
+ impl Compactor for WrittenOutputCompactor {
+ async fn merge_ssts(
+ &self,
+ _compaction_region: &CompactionRegion,
+ _picker_output: PickerOutput,
+ ) -> Result<MergeOutput> {
+ // Models the production compactor after `write_sst()` has created
+ // an external object and returned its FileMeta in MergeOutput.
+ self.written_files
+ .lock()
+ .unwrap()
+ .push(self.output_file.file_id);
+
+ Ok(MergeOutput {
+ files_to_add: vec![self.output_file.clone()],
+ files_to_remove: Vec::new(),
+ compaction_time_window: Some(3600),
+ sst_infos: Vec::new(),
+ })
+ }
+
+ async fn update_manifest(
+ &self,
+ compaction_region: &CompactionRegion,
+ merge_output: MergeOutput,
+ ) -> Result<(RegionEdit, ManifestVersion)> {
+ self.update_manifest_calls.fetch_add(1, Ordering::SeqCst);
+
+ let edit = RegionEdit {
+ files_to_add: merge_output.files_to_add,
+ files_to_remove: merge_output.files_to_remove,
+ timestamp_ms: Some(chrono::Utc::now().timestamp_millis()),
+ compaction_time_window: None,
+ flushed_entry_id: None,
+ flushed_sequence: None,
+ committed_sequence: None,
+ };
+ let action_list =
+ RegionMetaActionList::with_action(RegionMetaAction::Edit(edit.clone()));
+ let manifest_version = compaction_region
+ .manifest_ctx
+ .update_manifest_for_compaction(action_list)
+ .await?;
+
+ Ok((edit, manifest_version))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cancel_after_sst_hook_drops_written_outputs_before_manifest_commit() {
+ common_telemetry::init_default_ut_logging();
+
+ let hook_reached = Arc::new(Notify::new());
+ let hook_release = Arc::new(Notify::new());
+ let observed_files = Arc::new(Mutex::new(Vec::new()));
+ let hook: RegionHookRef = Arc::new(PausingSstHook {
+ reached: hook_reached.clone(),
+ release: hook_release.clone(),
+ observed_files: observed_files.clone(),
+ });
+
+ let compaction_region = new_test_compaction_region(hook).await;
+ let region_id = compaction_region.region_id;
+ let manifest_ctx = compaction_region.manifest_ctx.clone();
+ let output_file = dummy_file_meta(region_id);
+ let output_file_id = output_file.file_id;
+ let written_files = Arc::new(Mutex::new(Vec::new()));
+ let update_manifest_calls = Arc::new(AtomicUsize::new(0));
+
+ assert!(
+ manifest_ctx
+ .manifest()
+ .await
+ .files
+ .get(&output_file_id)
+ .is_none()
+ );
+ assert!(!written_files.lock().unwrap().contains(&output_file_id));
+
+ let state = LocalCompactionState::new(Default::default());
+ let (request_sender, mut request_receiver) = mpsc::channel(4);
+ let mut task = CompactionTaskImpl {
+ state: state.clone(),
+ compaction_region,
+ request_sender,
+ waiters: Vec::new(),
+ start_time: Instant::now(),
+ listener: WorkerListener::default(),
+ compactor: Arc::new(WrittenOutputCompactor {
+ output_file,
+ written_files: written_files.clone(),
+ update_manifest_calls: update_manifest_calls.clone(),
+ }),
+ picker_output: PickerOutput {
+ outputs: Vec::new(),
+ expired_ssts: Vec::new(),
+ time_window_size: 3600,
+ max_file_size: None,
+ },
+ memory_manager: Arc::new(new_compaction_memory_manager(1024 * 1024)),
+ memory_policy: OnExhaustedPolicy::Fail,
+ estimated_memory_bytes: 1,
+ };
+
+ let runner = tokio::spawn(async move {
+ task.run().await;
+ });
+
+ timeout(Duration::from_secs(5), hook_reached.notified())
+ .await
+ .expect("compaction should pause after observing written SST files");
+
+ assert_eq!(written_files.lock().unwrap().clone(), vec![output_file_id]);
+ assert_eq!(observed_files.lock().unwrap().clone(), vec![output_file_id]);
+ assert_eq!(update_manifest_calls.load(Ordering::SeqCst), 0);
+
+ assert_eq!(state.request_cancel(), RequestCancelResult::CancelIssued);
+ hook_release.notify_one();
+
+ let worker_request = timeout(Duration::from_secs(5), request_receiver.recv())
+ .await
+ .expect("compaction should notify the worker after cancellation")
+ .expect("worker request sender should stay open");
+
+ match worker_request.request {
+ WorkerRequest::Background {
+ region_id: notified_region,
+ notify: BackgroundNotify::CompactionCancelled(_),
+ } => assert_eq!(notified_region, region_id),
+ other => panic!("expected cancelled compaction notification, got {other:?}"),
+ }
+
+ runner
+ .await
+ .expect("compaction task should finish after reporting cancellation");
+
+ assert_eq!(written_files.lock().unwrap().clone(), vec![output_file_id]);
+ assert!(
+ manifest_ctx
+ .manifest()
+ .await
+ .files
+ .get(&output_file_id)
+ .is_none(),
+ "cancelled compaction dropped the written output before manifest ownership"
+ );
+ assert_eq!(update_manifest_calls.load(Ordering::SeqCst), 0);
+ }
+
// Note: Testing remove_expired() directly requires extensive mocking of:
// - manifest_ctx (ManifestContext)
// - request_sender (mpsc::Sender<WorkerRequestWithTime>)
```
</details>
The test can be run with:
```bash
cargo test -p mito2 --lib test_cancel_after_sst_hook_drops_written_outputs_before_manifest_commit -- --nocapture
```
Observed result:
```text
running 1 test
test compaction::task::tests::test_cancel_after_sst_hook_drops_written_outputs_before_manifest_commit ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 948 filtered out; finished in 0.10s
```
The test is a bug-existence validation test: it asserts the current bad behavior, not the desired fixed behavior. A post-fix regression test should invert the final invariant so that written outputs are either committed to the manifest or handed to a durable cleanup owner before cancellation is acknowledged.
### What did you expect to see?
Once compaction has physically written new SST/index files, cancellation should not be acknowledged until the system has preserved ownership of those outputs.
Any of the following would satisfy the invariant:
1. Reject cancellation after any compaction output file is written, then continue through `update_manifest_for_compaction`.
2. Accept cancellation, but first transfer all already-written `FileMeta` values to a durable cleanup/purge path and wait until that cleanup is durably scheduled.
3. Move the non-cancellable commit gate earlier so `on_sst_files_written` cannot be observed without either a later `on_manifest_updated` or an explicit abort/rollback notification.
In short: a file physically created by compaction should either become manifest-owned or cleanup-owned before the task reports successful cancellation.
### What did you see instead?
The current state machine can acknowledge `CompactionCancelled` after an SST output has been created and observed, but before `update_manifest()` is called. In that path, `MergeOutput.files_to_add` is dropped and the newly written file has no surviving immediate owner.
Concrete consequences:
- The region manifest does not contain the output file id.
- `version_control.apply_edit(...)` is not run for the output file.
- `FilePurger::new_file(...)` is not reached for the output file.
- The manifest's removed-files list also does not contain the output file, so fast GC has no manifest-tracked removal entry to process.
- If a `RegionHook` is installed, `on_sst_files_written` can observe the file while `on_manifest_updated` never follows for that same compaction edit.
The impact is bounded: this does not prove table-visible data corruption, because normal reads should follow manifest/version state. The concrete issue is external-state cleanup consistency: compaction can leave orphaned SST/index objects outside both the manifest and the ordinary file lifecycle.
Full-listing GC may eventually find such orphan files, but the code comments describe full listing as the expensive mode used to find files not tracked in the manifest:
```rust
// src/mito2/src/gc.rs:199-206
/// Whether to perform full file listing during GC.
/// When set to false, GC will only delete files that are tracked in the manifest's removed_files,
/// which can significantly improve performance by avoiding expensive list operations.
/// When set to true, GC will perform a full listing to find and delete orphan files
/// (files not tracked in the manifest).
///
/// Set to false for regular GC operations to optimize performance.
/// Set to true periodically or when you need to clean up orphan files.
```
The default GC scheduler options also make this a delayed/optional mitigation rather than a cancellation-safe ownership edge:
```rust
// src/meta-srv/src/gc/options.rs:72-87
impl Default for GcSchedulerOptions {
fn default() -> Self {
Self {
enable: false,
...
// Perform full file listing every 24 hours to find orphan files
full_file_listing_interval: Duration::from_secs(60 * 60 * 24),
...
}
}
}
```
### What operating system did you use?
Ubuntu 22.04 x64
### What version of GreptimeDB did you use?
Source checkout at commit 4f5dccf6bb1a9174918a8641b78dddad028588d6.
### Relevant log output and stack trace
```bash
No crash or stack trace is involved.
Key validation output:
$ cargo test -p mito2 --lib test_cancel_after_sst_hook_drops_written_outputs_before_manifest_commit -- --nocapture
running 1 test
test compaction::task::tests::test_cancel_after_sst_hook_drops_written_outputs_before_manifest_commit ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 948 filtered out; finished in 0.10s
```
1 条评论