fix(scanner): close lifecycle review gaps

crates/lifecycle/src/core.rs

@@ -54,6 +54,7 @@ const ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS: &str =
     "Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions";
 const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration";
 const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
+const ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY: &str = "Rule ID must not be empty";
 const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
 const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering";
 const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
@@ -402,10 +403,13 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     NoncurrentVersionTransitionOps::validate(transition)?;
                 }
             }
-            if let Some(id) = &r.id
-                && id.len() > 255
-            {
-                return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
+            if let Some(id) = &r.id {
+                if id.is_empty() {
+                    return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY));
+                }
+                if id.len() > 255 {
+                    return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
+                }
             }
             r.validate()?;
             if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref()
@@ -3730,6 +3734,31 @@ mod tests {
             .expect("empty prefix with filter should be valid");
     }
 
+    #[tokio::test]
+    async fn validate_rejects_empty_rule_id() {
+        let lc = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: Some(LifecycleExpiration {
+                    days: Some(30),
+                    ..Default::default()
+                }),
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some(String::new()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: None,
+            }],
+        };
+
+        let error = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
+        assert_eq!(error.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY);
+    }
+
     // --- TASK-004 tests: ExpiredObjectAllVersions ---
 
     #[tokio::test]

crates/scanner/src/scanner_folder.rs

@@ -675,6 +675,9 @@ pub struct FolderScanner {
     skip_heal: Arc<std::sync::atomic::AtomicBool>,
     local_disk: Arc<Disk>,
     pending_heals_changed: bool,
+    pending_size_reconciliation_keys: HashSet<String>,
+    pending_size_reconciliation_scopes: HashSet<String>,
+    pending_size_reconciliation_truncated: bool,
     #[cfg(test)]
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
 }
@@ -690,6 +693,10 @@ fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
         + std::mem::size_of::<u32>()
 }
 
+fn size_reconciliation_scope_key(bucket: &str, object: &str) -> String {
+    format!("{}:{}|{}:{}", bucket.len(), bucket, object.len(), object)
+}
+
 fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
     info.size_reconciliation.retain(|key, entry| {
         if entry.first_seen == 0 || entry.first_seen > now {
@@ -805,27 +812,17 @@ impl FolderScanner {
     /// so an incremental publication cannot lose a debt or its resolution.
     fn apply_size_reconciliation(&mut self, summary: &SizeSummary) {
         let now = Self::now_secs();
-        // Keep an unresolved identity in place while refreshing its object
-        // scope. This lets repeated observations increment `attempts`; only
-        // debts absent from the current pass are considered resolved.
-        let current_keys = summary
-            .size_reconciliation
-            .iter()
-            .map(|entry| entry.key.clone())
-            .collect::<HashSet<_>>();
+        self.pending_size_reconciliation_keys
+            .extend(summary.size_reconciliation.iter().map(|entry| entry.key.clone()));
+        self.pending_size_reconciliation_scopes.extend(
+            summary
+                .reconciliation_scopes
+                .iter()
+                .map(|scope| size_reconciliation_scope_key(&scope.bucket, &scope.object)),
+        );
+        self.pending_size_reconciliation_truncated |= summary.size_reconciliation_truncated;
+
         for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
-            prune_size_reconciliation(info, now);
-
-            if !summary.size_reconciliation_truncated {
-                for scope in &summary.reconciliation_scopes {
-                    let scope_bucket = item_actions::bounded_reconciliation_field(&scope.bucket);
-                    let scope_object = item_actions::bounded_reconciliation_field(&scope.object);
-                    info.size_reconciliation.retain(|key, entry| {
-                        entry.bucket != scope_bucket || entry.object != scope_object || current_keys.contains(key)
-                    });
-                }
-            }
-
             for incoming in &summary.size_reconciliation {
                 if let Some(existing) = info.size_reconciliation.get_mut(&incoming.key) {
                     existing.reason = incoming.reason.clone();
@@ -845,6 +842,21 @@ impl FolderScanner {
                 entry.attempts = 1;
                 info.size_reconciliation.insert(entry.key.clone(), entry);
             }
+        }
+    }
+
+    fn finish_size_reconciliation_batch(&mut self) {
+        let now = Self::now_secs();
+        let current_keys = std::mem::take(&mut self.pending_size_reconciliation_keys);
+        let scopes = std::mem::take(&mut self.pending_size_reconciliation_scopes);
+        let truncated = std::mem::replace(&mut self.pending_size_reconciliation_truncated, false);
+
+        for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
+            if !truncated {
+                info.size_reconciliation.retain(|key, entry| {
+                    !scopes.contains(&size_reconciliation_scope_key(&entry.bucket, &entry.object)) || current_keys.contains(key)
+                });
+            }
             prune_size_reconciliation(info, now);
         }
     }
@@ -2207,6 +2219,7 @@ impl FolderScanner {
             }
         }
 
+        self.finish_size_reconciliation_batch();
         done_folder();
         let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
         emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
@@ -2292,6 +2305,9 @@ pub async fn scan_data_folder(
         skip_heal,
         local_disk,
         pending_heals_changed: false,
+        pending_size_reconciliation_keys: HashSet::new(),
+        pending_size_reconciliation_scopes: HashSet::new(),
+        pending_size_reconciliation_truncated: false,
         #[cfg(test)]
         list_path_raw_options_observer: None,
     };

crates/scanner/src/scanner_folder/item_actions.rs

@@ -249,10 +249,6 @@ fn resolve_sizes(object_infos: &[ObjectInfo]) -> Vec<SizeResolution> {
 }
 
 fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule_id: &str) -> bool {
-    if rule_id.is_empty() {
-        return false;
-    }
-
     let filter_has_size = |filter: &s3s::dto::LifecycleRuleFilter| {
         filter.object_size_greater_than.is_some()
             || filter.object_size_less_than.is_some()
@@ -264,7 +260,13 @@ fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule
     lifecycle
         .rules
         .iter()
-        .find(|rule| rule.id.as_deref() == Some(rule_id))
+        .find(|rule| {
+            if rule_id.is_empty() {
+                rule.id.as_deref().is_none_or(str::is_empty)
+            } else {
+                rule.id.as_deref() == Some(rule_id)
+            }
+        })
         .and_then(|rule| rule.filter.as_ref())
         .is_some_and(filter_has_size)
 }
@@ -606,7 +608,10 @@ impl ScannerItem {
                         size_summary.actions_accounting_unknown(oi);
                         continue;
                     }
-                    SizeResolution::Corrupt { .. } => continue,
+                    SizeResolution::Corrupt { .. } => {
+                        size_summary.actions_accounting_unknown(oi);
+                        continue;
+                    }
                 };
 
                 let size = self.heal_actions(oi, accounting_size, size_summary).await;
@@ -1596,7 +1601,7 @@ mod tests {
             },
             &size_filtered
         ));
-        assert!(!lifecycle_rule_has_size_filter(
+        assert!(lifecycle_rule_has_size_filter(
             &BucketLifecycleConfiguration {
                 rules: vec![s3s::dto::LifecycleRule {
                     status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),

crates/scanner/src/scanner_folder/tests.rs

@@ -326,6 +326,9 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         skip_heal: Arc::new(AtomicBool::new(false)),
         local_disk: disk,
         pending_heals_changed: false,
+        pending_size_reconciliation_keys: HashSet::new(),
+        pending_size_reconciliation_scopes: HashSet::new(),
+        pending_size_reconciliation_truncated: false,
         list_path_raw_options_observer: None,
     };