track insertion time instead of data time for eviction

Author: ygndotgg

src/storage/object_storage.rs

@@ -63,7 +63,7 @@ use crate::storage::TARGETS_ROOT_DIRECTORY;
 use crate::storage::field_stats::DATASET_STATS_STREAM_NAME;
 use crate::storage::field_stats::calculate_field_stats;
 use crate::storage::field_stats::extract_datetime_from_parquet_path_regex;
-use crate::sync::ACTIVE_OBJECT_STORE_SYNC_FILES;
+use crate::sync::{ACTIVE_OBJECT_STORE_SYNC_FILES, SyncFileEntry};
 use crate::sync::FLUSH_AND_CONVERT_RUNTIME;
 
 use super::{
@@ -1090,14 +1090,21 @@ async fn process_parquet_files(
         let parquet_paths: Vec<PathBuf> = upload_context
             .stream
             .parquet_files()
-            .into_iter()
-            .filter(|p| !guard.contains(p))
+            .into_par_iter()
+            .filter(|p| !guard.contains_key(p))
             .collect();
 
         let mut ret = Vec::with_capacity(parquet_paths.len());
         ret.clone_from(&parquet_paths);
-        guard.extend(parquet_paths);
-        tracing::info!(ACTIVE_OBJECT_STORE_SYNC_FILES=?ACTIVE_OBJECT_STORE_SYNC_FILES);
+        for path in parquet_paths {
+            guard.insert(
+                path,
+                SyncFileEntry {
+                    tracked_instant: Instant::now(),
+                    tracked_utc: Utc::now(),
+                },
+            );
+        }
         ret
     };
 
@@ -1283,11 +1290,23 @@ async fn cleanup_uploaded_staged_files(uploaded_files: Vec<UploadedParquetFile>)
             guard.remove(path);
         }
 
-        // check if file has been in hashset for more than 5 minutes
-        let now = Utc::now();
-        guard.retain(|f| {
-            !extract_datetime_from_parquet_path_regex(f)
-                .is_ok_and(|ts| (now - ts).num_minutes() >= 5)
+        // Use monotonic time to ensure the 5-minute eviction window is immune to system clock adjustments.
+        let now = Instant::now();
+        guard.retain(|path, entry| {
+            let elapsed = now.duration_since(entry.tracked_instant);
+            // 5 min eviction window
+            if elapsed >= Duration::from_secs(300) {
+                // Added log for observability and debugging
+                warn!(
+                    "Removing stale sync file: {} (elapsed: {}s, tracked_at: {})",
+                    path.display(),
+                    elapsed.as_secs(),
+                    entry.tracked_utc.to_rfc3339(),
+                );
+                false
+            } else {
+                true
+            }
         });
     }
 

src/sync.rs

@@ -16,8 +16,7 @@
  *
  */
 
-use chrono::{TimeDelta, Timelike};
-use datafusion::common::HashSet;
+use chrono::{DateTime, TimeDelta, Timelike, Utc};
 use futures::FutureExt;
 use once_cell::sync::Lazy;
 use std::collections::HashMap;
@@ -39,8 +38,17 @@ pub static FLUSH_AND_CONVERT_RUNTIME: Lazy<Runtime> =
 static LOCAL_SYNC_RUNNING: AtomicBool = AtomicBool::new(false);
 static REMOTE_SYNC_RUNNING: AtomicBool = AtomicBool::new(false);
 
-pub static ACTIVE_OBJECT_STORE_SYNC_FILES: Lazy<Arc<RwLock<HashSet<PathBuf>>>> =
-    Lazy::new(|| Arc::new(RwLock::new(HashSet::new())));
+// tracks metadata for files being synced to object storage
+#[derive(Clone, Debug)]
+pub struct SyncFileEntry {
+    // Monotonic instant when file was added to tracking
+    pub tracked_instant: std::time::Instant,
+    // Wall-clock UTC timestamp for observability
+    pub tracked_utc: DateTime<Utc>,
+}
+
+pub static ACTIVE_OBJECT_STORE_SYNC_FILES: Lazy<Arc<RwLock<HashMap<PathBuf, SyncFileEntry>>>> =
+    Lazy::new(|| Arc::new(RwLock::new(HashMap::new())));
 /// RAII guard that clears a sync-running flag on drop, so a panic inside the
 /// sync body cannot leave the flag stuck at `true` and wedge future ticks.
 struct SyncRunningGuard(&'static AtomicBool);