fix: reclaim HDFS temporary files left behind by a datanode crash (#9616)

* fix: reclaim HDFS temporary files left behind by a datanode crash

Signed-off-by: lczllx <2181719471@qq.com>

* fix(mito2): stop the temporary file sweep from starving GC jobs

Signed-off-by: lczllx <2181719471@qq.com>

* feat(mito2): add metrics for the temporary file sweep

Signed-off-by: lczllx <2181719471@qq.com>

* refactor(object-store): share one definition of the temporary file name

Signed-off-by: lczllx <2181719471@qq.com>

* fix(mito2): skip the temporary file sweep when no GC job is allowed

Signed-off-by: lczllx <2181719471@qq.com>

* refactor: gate the temporary file sweep behind the hdfs-object-store feature

Signed-off-by: lczllx <2181719471@qq.com>

* fix(mito2): sweep temporary files when a create opens an existing region

Signed-off-by: lczllx <2181719471@qq.com>

* docs(config): document the temporary file sweep in the GC settings

Signed-off-by: lczllx <2181719471@qq.com>

* refactor: align the temporary file sweep code with the surrounding style

Signed-off-by: lczllx <2181719471@qq.com>

---------

Signed-off-by: lczllx <2181719471@qq.com>

src/mito2/AGENTS.md

@@ -34,6 +34,7 @@ snapshot isolation). It implements the `RegionEngine` trait from `store-api`.
 | `schedule` | `src/mito2/src/schedule/` | Local/remote background job scheduling |
 | `remap_manifest` | `src/mito2/src/remap_manifest.rs` | Manifest path remapping for region copy/migration |
 | `config` | `src/mito2/src/config.rs` | `MitoConfig` tuning knobs |
+| `temp_file_sweeper` | `src/mito2/src/temp_file_sweeper.rs` | Reclaims temporary files a crashed writer left in a region, triggered when the region becomes a writable leader or a create opens an existing region (under the `hdfs-object-store` feature) |
 | `test_util` | `src/mito2/src/test_util.rs` | `TestEnv` and builders (under the `test` feature) |
 
 ## Write path

src/mito2/src/gc.rs

@@ -159,6 +159,12 @@ pub struct GcConfig {
     /// expel time is the time when the file is considered as removed, as in removed from the manifest.
     /// This should only occur rarely, as manifest keep tracks in `removed_files` field
     /// unless something goes wrong.
+    ///
+    /// It is also the age threshold of the sweep that reclaims temporary files left behind
+    /// by a crashed HDFS writer, and that sweep runs even if `enable` is false. HDFS advances
+    /// the modification time of a file under construction only as its blocks complete, so
+    /// lowering this value also shrinks the margin between the sweep and files that are
+    /// still being written.
     #[serde(with = "humantime_serde")]
     pub unknown_file_lingering_time: Duration,
     /// Maximum concurrent list operations per GC job.
@@ -167,6 +173,8 @@ pub struct GcConfig {
     /// Maximum concurrent GC jobs.
     /// This is used to limit the number of concurrent GC jobs running on the datanode
     /// to prevent too many concurrent GC jobs from overwhelming the datanode.
+    /// Setting it to 0 disables GC jobs and the temporary file sweep, which is limited
+    /// to the same number of concurrent tasks.
     pub max_concurrent_gc_job: usize,
 }
 
@@ -177,8 +185,9 @@ impl Default for GcConfig {
             // expect long running queries to be finished(or at least be able to notify it's using a deleted file) within a reasonable time
             lingering_time: Some(Duration::from_secs(60 * 60)),
             // 1 day, for unknown expel time, which is when this file get removed from manifest.
-            // Only applies to full-listing GC for active/open regions. A long default avoids
-            // accidentally deleting pre-manifest files (e.g. compaction/flush still in progress).
+            // A long default avoids accidentally deleting pre-manifest files (e.g.
+            // compaction/flush still in progress) or temporary files of a writer that is
+            // still alive; the temporary file sweep uses this value as well.
             unknown_file_lingering_time: Duration::from_secs(24 * 60 * 60),
             max_concurrent_lister_per_gc_job: 32,
             max_concurrent_gc_job: 4,
@@ -806,7 +815,7 @@ impl LocalGcWorker {
     /// Only `.puffin` files (not subdirectories) are included.
     async fn list_region_index_files(&self, region_id: RegionId) -> Result<Vec<Entry>> {
         let region_dir = self.access_layer.build_region_dir(region_id);
-        let index_dir = object_store::util::join_dir(&region_dir, "index");
+        let index_dir = location::index_dir(&region_dir);
 
         let mut lister = match self
             .access_layer

src/mito2/src/lib.rs

@@ -42,6 +42,8 @@ pub mod request;
 pub mod schedule;
 pub mod series_index;
 pub mod sst;
+#[cfg(feature = "hdfs-object-store")]
+mod temp_file_sweeper;
 mod time_provider;
 pub mod wal;
 pub mod worker;

src/mito2/src/metrics.rs

@@ -607,6 +607,27 @@ lazy_static! {
         register_int_counter!("greptime_mito_flush_file_total", "mito flushed file count").unwrap();
 }
 
+#[cfg(feature = "hdfs-object-store")]
+lazy_static! {
+    /// Counter for the number of temporary files deleted by the sweeper.
+    pub static ref SWEEP_FILES_DELETED_TOTAL: IntCounter = register_int_counter!(
+        "greptime_mito_sweep_files_deleted_total",
+        "Total temporary files deleted by the sweeper",
+    ).unwrap();
+
+    /// Counter for the number of failures while sweeping temporary files.
+    pub static ref SWEEP_ERRORS_TOTAL: IntCounter = register_int_counter!(
+        "greptime_mito_sweep_errors_total",
+        "Total failures while sweeping temporary files",
+    ).unwrap();
+
+    /// Counter for the number of temporary file sweeps run.
+    pub static ref SWEEP_RUNS_TOTAL: IntCounter = register_int_counter!(
+        "greptime_mito_sweep_runs_total",
+        "Total temporary file sweeps run",
+    ).unwrap();
+}
+
 /// Stager notifier to collect metrics.
 pub struct StagerMetrics {
     cache_hit: IntCounter,

src/mito2/src/sst/location.rs

@@ -39,6 +39,11 @@ pub fn region_dir_from_table_dir(
     }
 }
 
+/// Generate index dir from region_dir
+pub fn index_dir(region_dir: &str) -> String {
+    util::join_dir(region_dir, "index")
+}
+
 pub fn sst_file_path(table_dir: &str, region_file_id: RegionFileId, path_type: PathType) -> String {
     let region_dir = region_dir_from_table_dir(table_dir, region_file_id.region_id(), path_type);
     util::join_path(
@@ -49,7 +54,7 @@ pub fn sst_file_path(table_dir: &str, region_file_id: RegionFileId, path_type: P
 
 pub fn index_file_path(table_dir: &str, index_id: RegionIndexId, path_type: PathType) -> String {
     let region_dir = region_dir_from_table_dir(table_dir, index_id.file_id.region_id(), path_type);
-    let index_dir = util::join_dir(&region_dir, "index");
+    let index_dir = index_dir(&region_dir);
 
     let filename = if index_id.version == 0 {
         format!("{}.puffin", index_id.file_id.file_id())

src/mito2/src/temp_file_sweeper.rs

@@ -0,0 +1,152 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Reclaims the temporary files that a crashed writer left in a region.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use common_telemetry::info;
+use object_store::temp_file::{SweepReport, sweep_temporary_files, writes_temporary_files};
+use store_api::storage::RegionId;
+use tokio::sync::Semaphore;
+
+use crate::access_layer::{AccessLayer, AccessLayerRef};
+use crate::manifest::storage::manifest_dir;
+use crate::manifest::storage::staging::{staging_blob_path, staging_manifest_path};
+use crate::metrics::{SWEEP_ERRORS_TOTAL, SWEEP_FILES_DELETED_TOTAL, SWEEP_RUNS_TOTAL};
+use crate::sst::location::index_dir;
+
+/// Returns the directories that `region_id` writes non-append files into.
+///
+/// Every one of them can hold the temporary file of a crashed writer, and a
+/// list is not recursive, so the sweep has to name each directory.
+pub(crate) fn region_write_dirs(access_layer: &AccessLayer, region_id: RegionId) -> Vec<String> {
+    let region_dir = access_layer.build_region_dir(region_id);
+    let manifest_path = manifest_dir(&region_dir);
+    vec![
+        region_dir.clone(),
+        index_dir(&region_dir),
+        manifest_path.clone(),
+        staging_blob_path(&manifest_path),
+        staging_manifest_path(&manifest_path),
+    ]
+}
+
+/// Sweeps the temporary files of `region_id` in the background.
+///
+/// The caller is on the path that makes the region its writable leader, so
+/// reclaiming leftovers must not slow it down or fail it: everything happens in
+/// a spawned task that only logs.
+///
+/// `older_than` is the margin the caller takes from GC's
+/// `unknown_file_lingering_time`, which guards the same thing: a file this old
+/// is not one a writer is still filling.
+///
+/// A sweep waits for a permit of `sweep_limiter`, so a `None` means the
+/// datanode allows no concurrent GC job and there is no sweep.
+pub(crate) fn spawn_region_sweep(
+    access_layer: AccessLayerRef,
+    region_id: RegionId,
+    older_than: Duration,
+    sweep_limiter: Option<Arc<Semaphore>>,
+) {
+    // No write to this store leaves a temporary file behind, so don't spend a
+    // task on it: a sweep would still list every directory of the region to
+    // find that out.
+    if !writes_temporary_files(access_layer.object_store()) {
+        return;
+    }
+
+    let Some(sweep_limiter) = sweep_limiter else {
+        return;
+    };
+
+    common_runtime::spawn_global(async move {
+        // A burst of regions becoming writable must not run more sweeps at once
+        // than the datanode allows GC jobs. Wait for a permit instead of giving
+        // up, the sweep has no deadline.
+        let Ok(_permit) = sweep_limiter.acquire_owned().await else {
+            return;
+        };
+
+        let dirs = region_write_dirs(&access_layer, region_id);
+        let report = sweep_temporary_files(access_layer.object_store(), &dirs, older_than).await;
+        record_sweep(&report);
+        if report.deleted > 0 {
+            info!(
+                "Reclaimed {} temporary file(s) of a crashed writer in region {region_id}",
+                report.deleted
+            );
+        }
+    });
+}
+
+fn record_sweep(report: &SweepReport) {
+    SWEEP_RUNS_TOTAL.inc();
+    SWEEP_FILES_DELETED_TOTAL.inc_by(report.deleted as u64);
+    SWEEP_ERRORS_TOTAL.inc_by(report.failed as u64);
+}
+
+#[cfg(test)]
+mod tests {
+    use object_store::ObjectStore;
+    use object_store::services::Memory;
+    use store_api::region_request::PathType;
+
+    use super::*;
+    use crate::test_util::TestEnv;
+
+    fn test_access_layer(env: &TestEnv) -> AccessLayer {
+        AccessLayer::new(
+            "test",
+            PathType::Bare,
+            ObjectStore::new(Memory::default()).unwrap(),
+            env.get_puffin_manager(),
+            env.get_intermediate_manager(),
+        )
+    }
+
+    #[tokio::test]
+    async fn test_region_write_dirs() {
+        let env = TestEnv::new().await;
+        let access_layer = test_access_layer(&env);
+        let region_id = RegionId::new(1, 2);
+
+        let dirs = region_write_dirs(&access_layer, region_id);
+
+        assert_eq!(5, dirs.len());
+        assert!(dirs[0].ends_with("1_0000000002/"));
+        assert!(dirs[1].ends_with("1_0000000002/index/"));
+        assert!(dirs[2].ends_with("1_0000000002/manifest/"));
+        assert!(dirs[3].ends_with("1_0000000002/staging/blob/"));
+        assert!(dirs[4].ends_with("1_0000000002/staging/manifest/"));
+    }
+
+    #[test]
+    fn test_record_sweep() {
+        let runs = SWEEP_RUNS_TOTAL.get();
+        let deleted = SWEEP_FILES_DELETED_TOTAL.get();
+        let errors = SWEEP_ERRORS_TOTAL.get();
+
+        record_sweep(&SweepReport {
+            deleted: 2,
+            failed: 1,
+        });
+
+        assert_eq!(runs + 1, SWEEP_RUNS_TOTAL.get());
+        assert_eq!(deleted + 2, SWEEP_FILES_DELETED_TOTAL.get());
+        assert_eq!(errors + 1, SWEEP_ERRORS_TOTAL.get());
+    }
+}

src/mito2/src/worker.rs

@@ -87,6 +87,8 @@ use crate::sst::file_ref::FileReferenceManagerRef;
 use crate::sst::index::IndexBuildScheduler;
 use crate::sst::index::intermediate::IntermediateManager;
 use crate::sst::index::puffin_manager::PuffinManagerFactory;
+#[cfg(feature = "hdfs-object-store")]
+use crate::temp_file_sweeper;
 use crate::time_provider::{StdTimeProvider, TimeProviderRef};
 use crate::wal::Wal;
 use crate::worker::handle_manifest::RegionEditQueues;
@@ -241,6 +243,13 @@ impl WorkerGroup {
         let compaction_memory_manager =
             Arc::new(new_compaction_memory_manager(compaction_limit_bytes));
         let gc_limiter = Arc::new(GcLimiter::new(config.gc.max_concurrent_gc_job));
+        // A sweep waits for a permit while a GC job takes one without waiting, so
+        // sharing the GC limiter would fail GC jobs and report sweeps as GC jobs.
+        // A datanode that allows no concurrent GC job runs no sweep either: a
+        // limiter of zero would never hand out the permit the task waits for.
+        #[cfg(feature = "hdfs-object-store")]
+        let sweep_limiter = (config.gc.max_concurrent_gc_job > 0)
+            .then(|| Arc::new(Semaphore::new(config.gc.max_concurrent_gc_job)));
 
         let workers = (0..config.num_workers)
             .map(|id| {
@@ -269,6 +278,8 @@ impl WorkerGroup {
                     file_ref_manager: file_ref_manager.clone(),
                     partition_expr_fetcher: partition_expr_fetcher.clone(),
                     flush_semaphore: flush_semaphore.clone(),
+                    #[cfg(feature = "hdfs-object-store")]
+                    sweep_limiter: sweep_limiter.clone(),
                 }
                 .start()
             })
@@ -491,6 +502,13 @@ impl WorkerGroup {
         let compaction_memory_manager =
             Arc::new(new_compaction_memory_manager(compaction_limit_bytes));
         let gc_limiter = Arc::new(GcLimiter::new(config.gc.max_concurrent_gc_job));
+        // A sweep waits for a permit while a GC job takes one without waiting, so
+        // sharing the GC limiter would fail GC jobs and report sweeps as GC jobs.
+        // A datanode that allows no concurrent GC job runs no sweep either: a
+        // limiter of zero would never hand out the permit the task waits for.
+        #[cfg(feature = "hdfs-object-store")]
+        let sweep_limiter = (config.gc.max_concurrent_gc_job > 0)
+            .then(|| Arc::new(Semaphore::new(config.gc.max_concurrent_gc_job)));
         let workers = (0..config.num_workers)
             .map(|id| {
                 WorkerStarter {
@@ -518,6 +536,8 @@ impl WorkerGroup {
                     file_ref_manager: file_ref_manager.clone(),
                     partition_expr_fetcher: partition_expr_fetcher.clone(),
                     flush_semaphore: flush_semaphore.clone(),
+                    #[cfg(feature = "hdfs-object-store")]
+                    sweep_limiter: sweep_limiter.clone(),
                 }
                 .start()
             })
@@ -628,6 +648,10 @@ struct WorkerStarter<S> {
     file_ref_manager: FileReferenceManagerRef,
     partition_expr_fetcher: PartitionExprFetcherRef,
     flush_semaphore: Arc<Semaphore>,
+    #[cfg(feature = "hdfs-object-store")]
+    /// Limiter for the temporary file sweeps, or `None` if the datanode allows
+    /// no concurrent GC job.
+    sweep_limiter: Option<Arc<Semaphore>>,
 }
 
 impl<S: LogStore> WorkerStarter<S> {
@@ -728,6 +752,8 @@ impl<S: LogStore> WorkerStarter<S> {
             partition_expr_fetcher: self.partition_expr_fetcher,
             flush_semaphore: self.flush_semaphore,
             plugins: self.plugins,
+            #[cfg(feature = "hdfs-object-store")]
+            sweep_limiter: self.sweep_limiter,
         };
         let handle = common_runtime::spawn_global(async move {
             worker_thread.run().await;
@@ -1055,6 +1081,10 @@ struct RegionWorkerLoop<S> {
     flush_semaphore: Arc<Semaphore>,
     /// Plugins for flush hooks.
     plugins: Plugins,
+    #[cfg(feature = "hdfs-object-store")]
+    /// Limiter for the temporary file sweeps, or `None` if the datanode allows
+    /// no concurrent GC job.
+    sweep_limiter: Option<Arc<Semaphore>>,
 }
 
 impl<S: LogStore> RegionWorkerLoop<S> {
@@ -1475,6 +1505,10 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                 // changes a writable or staging leader.
                 region.set_role(RegionRole::DowngradingLeader);
             }
+            #[cfg(feature = "hdfs-object-store")]
+            let sweep_older_than = self.config.gc.unknown_file_lingering_time;
+            #[cfg(feature = "hdfs-object-store")]
+            let sweep_limiter = self.sweep_limiter.clone();
             // We need to do this in background as we need the manifest lock.
             common_runtime::spawn_global(async move {
                 match region.set_role_state_gracefully(region_role_state).await {
@@ -1483,6 +1517,17 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                         let _ = sender.send(SetRegionRoleStateResponse::success(
                             SetRegionRoleStateSuccess::mito(last_entry_id),
                         ));
+                        // Only a writable leader reclaims what a writer that died
+                        // before it renamed its temporary file left behind.
+                        #[cfg(feature = "hdfs-object-store")]
+                        if region_role_state == SettableRegionRoleState::Leader {
+                            temp_file_sweeper::spawn_region_sweep(
+                                region.access_layer(),
+                                region_id,
+                                sweep_older_than,
+                                sweep_limiter,
+                            );
+                        }
                     }
                     Err(e) => {
                         error!(e; "Failed to set region {} role state to {:?}", region_id, region_role_state);

src/mito2/src/worker/handle_catchup.rs

@@ -28,6 +28,8 @@ use crate::region::MitoRegion;
 use crate::region::catchup::RegionCatchupTask;
 use crate::region::opener::RegionOpener;
 use crate::request::OptionOutputTx;
+#[cfg(feature = "hdfs-object-store")]
+use crate::temp_file_sweeper;
 use crate::wal::entry_distributor::WalEntryReceiver;
 use crate::worker::RegionWorkerLoop;
 
@@ -69,6 +71,10 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         let catchup_regions = self.catchup_regions.clone();
         let wal = self.wal.clone();
         let allow_stale_entries = self.config.allow_stale_entries;
+        #[cfg(feature = "hdfs-object-store")]
+        let sweep_older_than = self.config.gc.unknown_file_lingering_time;
+        #[cfg(feature = "hdfs-object-store")]
+        let sweep_limiter = self.sweep_limiter.clone();
         common_runtime::spawn_global(async move {
             let mut task = RegionCatchupTask::new(region.clone(), wal, allow_stale_entries)
                 .with_entry_receiver(entry_receiver)
@@ -80,11 +86,22 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                     if request.set_writable {
                         region.set_role(RegionRole::Leader);
                         // Finalize leadership: persist backfilled metadata.
-                        if let Err(err) = region
+                        match region
                             .set_role_state_gracefully(SettableRegionRoleState::Leader)
                             .await
                         {
-                            error!(err; "Failed to set region {region_id} to leader");
+                            Ok(()) => {
+                                #[cfg(feature = "hdfs-object-store")]
+                                temp_file_sweeper::spawn_region_sweep(
+                                    region.access_layer(),
+                                    region_id,
+                                    sweep_older_than,
+                                    sweep_limiter,
+                                );
+                            }
+                            Err(err) => {
+                                error!(err; "Failed to set region {region_id} to leader");
+                            }
                         }
                     }
                     catchup_regions.remove_region(region_id);

src/mito2/src/worker/handle_create.rs

@@ -22,6 +22,8 @@ use store_api::storage::RegionId;
 
 use crate::error::Result;
 use crate::region::opener::{RegionOpener, check_recovered_region};
+#[cfg(feature = "hdfs-object-store")]
+use crate::temp_file_sweeper;
 use crate::worker::RegionWorkerLoop;
 
 impl<S: LogStore> RegionWorkerLoop<S> {
@@ -79,6 +81,16 @@ impl<S: LogStore> RegionWorkerLoop<S> {
 
         let region = opener.create_or_open(&self.config, &self.wal).await?;
 
+        // A Create retried after a crash opens an existing region as a leader
+        // without a role-state change, so sweep here as well.
+        #[cfg(feature = "hdfs-object-store")]
+        temp_file_sweeper::spawn_region_sweep(
+            region.access_layer(),
+            region_id,
+            self.config.gc.unknown_file_lingering_time,
+            self.sweep_limiter.clone(),
+        );
+
         info!(
             "A new region created with requirement {:?}, worker: {}, region: {:?}",
             requirements,

src/object-store/src/factory.rs

@@ -197,6 +197,8 @@ mod tests {
         let store = new_hdfs_object_store(&config).await.unwrap();
         assert_eq!(HDFS_NATIVE_SCHEME, store.info().scheme());
         assert_eq!("/greptimedb/", store.info().root());
+        // The sweep recognizes this store as one whose writes leave temporary files.
+        assert!(crate::temp_file::writes_temporary_files(&store));
     }
 
     #[tokio::test]

src/object-store/src/layers/hdfs.rs

@@ -23,7 +23,8 @@ use opendal::raw::{
     OpWrite, RpCreateDir, RpPresign, RpRename, RpStat, Service, ServiceInfo, Servicer, oio,
 };
 use opendal::{Buffer, Capability, ErrorKind, Metadata, OperationContext, Result};
-use uuid::Uuid;
+
+use crate::temp_file::temporary_path;
 
 /// Adds atomic writes and streaming copies to the native HDFS backend.
 #[derive(Debug, Clone)]
@@ -219,7 +220,10 @@ impl oio::Write for HdfsWriter {
                     Ok(metadata) => metadata,
                     Err(error) => {
                         drop(writer);
-                        let _ = delete_path(inner, context, temporary_path).await;
+                        if let Err(delete_error) = delete_path(inner, context, temporary_path).await
+                        {
+                            warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS close");
+                        }
                         return Err(error);
                     }
                 };
@@ -229,7 +233,9 @@ impl oio::Write for HdfsWriter {
                     .rename(inner, context, temporary_path, target_path)
                     .await
                 {
-                    let _ = delete_path(inner, context, temporary_path).await;
+                    if let Err(delete_error) = delete_path(inner, context, temporary_path).await {
+                        warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS rename");
+                    }
                     return Err(error);
                 }
 
@@ -424,7 +430,9 @@ async fn copy_via_read_write(
         Ok(metadata) => metadata,
         Err(error) => {
             drop(writer);
-            let _ = delete_path(&inner, context, &temporary_path).await;
+            if let Err(delete_error) = delete_path(&inner, context, &temporary_path).await {
+                warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS close");
+            }
             return Err(error);
         }
     };
@@ -434,7 +442,9 @@ async fn copy_via_read_write(
         .rename(&inner, context, &temporary_path, target_path)
         .await
     {
-        let _ = delete_path(&inner, context, &temporary_path).await;
+        if let Err(delete_error) = delete_path(&inner, context, &temporary_path).await {
+            warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS rename");
+        }
         return Err(error);
     }
 
@@ -462,15 +472,6 @@ async fn delete_path(inner: &Servicer, context: &OperationContext, path: &str) -
     deleter.close().await
 }
 
-// TODO(fengjiachun): Clean up temporary files left behind after a process crash.
-fn temporary_path(path: &str) -> String {
-    let suffix = format!(".greptime-{}.tmp", Uuid::new_v4());
-    match path.rsplit_once('/') {
-        Some((parent, name)) => format!("{parent}/.{name}{suffix}"),
-        None => format!(".{path}{suffix}"),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use opendal::services::Fs;
@@ -565,6 +566,25 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_list_reports_temporary_file_of_crashed_writer() {
+        // Regression test for #9591: a temporary file left by a crashed writer
+        // must stay visible to a sweep over the directory of its target.
+        let (_directory, store) = test_store();
+        store.create_dir("data/").await.unwrap();
+        let temporary = temporary_path("data/manifest.json");
+        assert!(temporary.starts_with("data/.manifest.json.greptime-"));
+        assert!(temporary.ends_with(".tmp"));
+        store.write(&temporary, "partial").await.unwrap();
+
+        let entries = store.list("data/").await.unwrap();
+        let entry = entries
+            .iter()
+            .find(|entry| entry.path() == temporary)
+            .unwrap();
+        assert!(entry.metadata().last_modified().is_some());
+    }
+
     #[tokio::test]
     async fn test_copy_fallback_streams_to_target() {
         let (_directory, store) = test_store();

src/object-store/src/lib.rs

@@ -25,6 +25,8 @@ pub mod factory;
 pub mod layers;
 pub mod manager;
 pub mod secure_fs;
+#[cfg(feature = "hdfs-object-store")]
+pub mod temp_file;
 #[cfg(feature = "testing")]
 pub mod test_util;
 pub mod util;

src/object-store/src/temp_file.rs

@@ -0,0 +1,373 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Temporary files of non-append writes, and how to reclaim them.
+
+use std::time::Duration;
+
+use common_telemetry::warn;
+use opendal::raw::Timestamp;
+use uuid::Uuid;
+
+use crate::{ErrorKind, ObjectStore};
+
+/// Prefix of a temporary file name.
+const TEMPORARY_PREFIX: &str = ".";
+
+/// Marker that follows the name of the target in a temporary file name.
+const TEMPORARY_MARKER: &str = ".greptime-";
+
+/// Suffix of a temporary file name.
+const TEMPORARY_SUFFIX: &str = ".tmp";
+
+/// Scheme of the only service that publishes non-append writes with a rename.
+const TEMPORARY_FILE_SCHEME: &str = crate::services::HDFS_NATIVE_SCHEME;
+
+/// Returns true if `store` publishes non-append writes with a rename, which is
+/// what leaves temporary files behind when the writer dies.
+///
+/// Only the HDFS compatibility layer does that; a write to any other service
+/// appears under its final name.
+pub fn writes_temporary_files(store: &ObjectStore) -> bool {
+    store.info().scheme() == TEMPORARY_FILE_SCHEME
+}
+
+/// What a sweep did.
+///
+/// A sweep goes through its directories and files one at a time, and a failure
+/// on either doesn't stop it, so it reports what it reclaimed and what it
+/// couldn't.
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct SweepReport {
+    /// Temporary files deleted.
+    pub deleted: usize,
+    /// List and delete operations that failed.
+    pub failed: usize,
+}
+
+impl SweepReport {
+    fn merge(&mut self, other: SweepReport) {
+        self.deleted += other.deleted;
+        self.failed += other.failed;
+    }
+}
+
+/// Deletes the temporary files in `dirs` that are older than `older_than`.
+///
+/// A non-append write goes to a temporary file next to its target and is
+/// renamed on close, so a writer that dies in between leaves the file behind
+/// with nothing to reclaim it: GC can't parse the name, and the startup cleanup
+/// of [`ATOMIC_WRITE_DIR`](crate::ATOMIC_WRITE_DIR) doesn't reach it.
+///
+/// The caller runs this for a region that just became its writable leader, so a
+/// file that is old enough belongs to a writer that is gone. Failures are
+/// logged and don't stop the sweep.
+pub async fn sweep_temporary_files(
+    store: &ObjectStore,
+    dirs: &[String],
+    older_than: Duration,
+) -> SweepReport {
+    let mut report = SweepReport::default();
+    for dir in dirs {
+        report.merge(sweep_dir(store, dir, older_than).await);
+    }
+    report
+}
+
+async fn sweep_dir(store: &ObjectStore, dir: &str, older_than: Duration) -> SweepReport {
+    let entries = match store.list(dir).await {
+        Ok(entries) => entries,
+        // Not every region has every directory, for example one that was never
+        // repartitioned has no staging directory.
+        Err(error) if error.kind() == ErrorKind::NotFound => return SweepReport::default(),
+        Err(error) => {
+            warn!(error; "Failed to list {dir} for temporary files");
+            return SweepReport {
+                failed: 1,
+                ..Default::default()
+            };
+        }
+    };
+
+    let expired_before = Timestamp::now()
+        .into_inner()
+        .as_millisecond()
+        .saturating_sub(i64::try_from(older_than.as_millis()).unwrap_or(i64::MAX));
+
+    let mut report = SweepReport::default();
+    for entry in entries {
+        if !is_temporary_file(entry.name()) {
+            continue;
+        }
+
+        // HDFS advances the modification time of a file under construction only
+        // as its blocks complete, so an old file is not necessarily abandoned.
+        // That is what the threshold covers.
+        let is_expired = entry
+            .metadata()
+            .last_modified()
+            .map(|last_modified| last_modified.into_inner().as_millisecond() < expired_before)
+            // Without a modification time there is no way to tell whether the
+            // writer is gone, so keep the file.
+            .unwrap_or(false);
+        if !is_expired {
+            continue;
+        }
+
+        // Delete one file at a time: the bulk delete stops at the first
+        // failure, and one file that can't be deleted must not keep the rest of
+        // the directory from being swept.
+        let path = entry.path();
+        match store.delete(path).await {
+            Ok(()) => report.deleted += 1,
+            Err(error) => {
+                warn!(error; "Failed to delete temporary file {path}");
+                report.failed += 1;
+            }
+        }
+    }
+
+    report
+}
+
+/// Returns the path a non-append write to `path` goes to before it is renamed.
+///
+/// A writer that dies in between leaves the file behind, which is what
+/// [`sweep_temporary_files`] reclaims.
+pub(crate) fn temporary_path(path: &str) -> String {
+    let suffix = format!("{TEMPORARY_MARKER}{}{TEMPORARY_SUFFIX}", Uuid::new_v4());
+    match path.rsplit_once('/') {
+        Some((parent, name)) => format!("{parent}/{TEMPORARY_PREFIX}{name}{suffix}"),
+        None => format!("{TEMPORARY_PREFIX}{path}{suffix}"),
+    }
+}
+
+fn is_temporary_file(name: &str) -> bool {
+    // `name` is the last segment of a path, so a directory name ends with `/`.
+    name.starts_with(TEMPORARY_PREFIX)
+        && name.contains(TEMPORARY_MARKER)
+        && name.ends_with(TEMPORARY_SUFFIX)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use opendal::Operator;
+    use opendal::services::Fs;
+    use tempfile::TempDir;
+
+    use super::*;
+    use crate::layers::HdfsCompatibilityLayer;
+    use crate::layers::mock::{MockLayer, MockLayerBuilder, OpDelete, oio};
+    use crate::{Error, Result};
+
+    const EXPIRED: Duration = Duration::from_millis(1);
+
+    fn test_store() -> (TempDir, ObjectStore) {
+        let directory = tempfile::tempdir().unwrap();
+        let store = Operator::new(Fs::default().root(directory.path().to_str().unwrap())).unwrap();
+        (directory, store)
+    }
+
+    fn test_store_of(layer: MockLayer) -> (TempDir, ObjectStore) {
+        let directory = tempfile::tempdir().unwrap();
+        let store = Operator::new(Fs::default().root(directory.path().to_str().unwrap()))
+            .unwrap()
+            .layer(layer);
+        (directory, store)
+    }
+
+    fn test_store_of_failing_list(kind: ErrorKind) -> (TempDir, ObjectStore) {
+        struct FailingLister(ErrorKind);
+
+        impl oio::List for FailingLister {
+            async fn next(&mut self) -> Result<Option<oio::Entry>> {
+                Err(Error::new(self.0, "Failed to list"))
+            }
+        }
+
+        let layer = MockLayerBuilder::default()
+            .lister_factory(Arc::new(move |_, _, _| Box::new(FailingLister(kind))))
+            .build()
+            .unwrap();
+        test_store_of(layer)
+    }
+
+    // A store that fails to delete every path containing `marker`.
+    fn test_store_of_failing_delete(marker: &'static str) -> (TempDir, ObjectStore) {
+        struct FailingDeleter {
+            inner: oio::Deleter,
+            marker: &'static str,
+        }
+
+        impl oio::Delete for FailingDeleter {
+            async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+                if path.contains(self.marker) {
+                    return Err(Error::new(ErrorKind::Unexpected, "Failed to delete"));
+                }
+                self.inner.delete(path, args).await
+            }
+
+            async fn close(&mut self) -> Result<()> {
+                self.inner.close().await
+            }
+        }
+
+        let layer = MockLayerBuilder::default()
+            .deleter_factory(Arc::new(move |inner| {
+                Box::new(FailingDeleter { inner, marker })
+            }))
+            .build()
+            .unwrap();
+        test_store_of(layer)
+    }
+
+    async fn wait_until_expired() {
+        // A file written just before this call can carry the same modification
+        // time in milliseconds as the sweep that follows it, which would make the
+        // sweep keep it.
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+
+    #[test]
+    fn test_is_temporary_file() {
+        assert!(is_temporary_file(
+            ".00000000000000000042.json.greptime-53f0f4a6-3f3e-4a2f-9a2f-6fe4b0d5b5b0.tmp"
+        ));
+        assert!(is_temporary_file(".a5f2.parquet.greptime-53f0f4a6.tmp"));
+
+        assert!(!is_temporary_file("00000000000000000042.json"));
+        assert!(!is_temporary_file(".00000000000000000042.json"));
+        assert!(!is_temporary_file(
+            "00000000000000000042.json.greptime-53f0f4a6.tmp"
+        ));
+        assert!(!is_temporary_file(
+            ".00000000000000000042.json.greptime-53f0f4a6"
+        ));
+        assert!(!is_temporary_file("index/"));
+    }
+
+    #[tokio::test]
+    async fn test_sweep_dir_deletes_expired_temporary_file() {
+        let (_directory, store) = test_store();
+        let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp";
+        store.create_dir("data/").await.unwrap();
+        store.write(temporary, "partial").await.unwrap();
+        store.write("data/42.parquet", "data").await.unwrap();
+
+        wait_until_expired().await;
+        assert_eq!(
+            SweepReport {
+                deleted: 1,
+                failed: 0
+            },
+            sweep_dir(&store, "data/", EXPIRED).await
+        );
+
+        assert!(!store.exists(temporary).await.unwrap());
+        assert!(store.exists("data/42.parquet").await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_sweep_dir_keeps_young_temporary_file() {
+        let (_directory, store) = test_store();
+        let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp";
+        store.create_dir("data/").await.unwrap();
+        store.write(temporary, "partial").await.unwrap();
+
+        assert_eq!(
+            SweepReport::default(),
+            sweep_dir(&store, "data/", Duration::from_secs(24 * 60 * 60)).await
+        );
+
+        assert!(store.exists(temporary).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_sweep_dir_of_missing_directory() {
+        // Fs lists a directory it doesn't have as an empty one, but HDFS reports
+        // it as not found.
+        let (_directory, store) = test_store_of_failing_list(ErrorKind::NotFound);
+
+        // A region that was never repartitioned has no staging directory, so a
+        // missing directory is an expected answer rather than a failure.
+        assert_eq!(
+            SweepReport::default(),
+            sweep_dir(&store, "data/staging/blob/", Duration::ZERO).await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_sweep_dir_reports_failed_list() {
+        let (_directory, store) = test_store_of_failing_list(ErrorKind::Unexpected);
+
+        assert_eq!(
+            SweepReport {
+                deleted: 0,
+                failed: 1
+            },
+            sweep_dir(&store, "data/", Duration::ZERO).await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_sweep_dir_keeps_going_after_failed_delete() {
+        let (_directory, store) = test_store_of_failing_delete("fail");
+        let failing = "data/.fail-42.parquet.greptime-53f0f4a6.tmp";
+        let sweepable = "data/.ok-42.parquet.greptime-53f0f4a6.tmp";
+        store.create_dir("data/").await.unwrap();
+        store.write(failing, "partial").await.unwrap();
+        store.write(sweepable, "partial").await.unwrap();
+
+        wait_until_expired().await;
+        assert_eq!(
+            SweepReport {
+                deleted: 1,
+                failed: 1
+            },
+            sweep_dir(&store, "data/", EXPIRED).await
+        );
+
+        assert!(store.exists(failing).await.unwrap());
+        assert!(!store.exists(sweepable).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_sweep_temporary_files_on_compatibility_layer() {
+        let (_directory, store) = test_store();
+        let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp";
+        store.create_dir("data/").await.unwrap();
+        // What a writer that died before it renamed its file leaves behind.
+        store.write(temporary, "partial").await.unwrap();
+        // The same layer HDFS writes through, on an Fs store whose scheme stays `fs`.
+        let store = store.layer(HdfsCompatibilityLayer::new_for_test());
+
+        wait_until_expired().await;
+        assert_eq!(
+            SweepReport {
+                deleted: 1,
+                failed: 0
+            },
+            sweep_temporary_files(
+                &store,
+                &["data/".to_string(), "data/staging/blob/".to_string()],
+                EXPIRED,
+            )
+            .await
+        );
+
+        assert!(!store.exists(temporary).await.unwrap());
+    }
+}