diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs
index 9071ff81e..af176630b 100644
--- a/crates/data-usage/src/data_usage.rs
+++ b/crates/data-usage/src/data_usage.rs
@@ -688,15 +688,14 @@ impl DataUsageCache {
             return;
         }
 
-        let mut leaves = Vec::new();
+        let mut candidates = Vec::new();
         let mut remove = total - limit;
-        add(self, path, &mut leaves);
-        leaves.sort_by_key(|a| a.objects);
+        add(self, path, &mut candidates);
+        candidates.sort_by_key(|a| a.objects);
 
-        while remove > 0 && !leaves.is_empty() {
-            let Some(e) = leaves.first() else {
-                break;
-            };
+        let mut candidate_index = 0;
+        while remove > 0 && candidate_index < candidates.len() {
+            let e = &candidates[candidate_index];
             let candidate = e.path.clone();
             if candidate == *path && !compact_self {
                 break;
@@ -705,7 +704,7 @@ impl DataUsageCache {
             let mut flat = match self.size_recursive(&candidate.key()) {
                 Some(flat) => flat,
                 None => {
-                    leaves.remove(0);
+                    candidate_index += 1;
                     continue;
                 }
             };
@@ -714,8 +713,8 @@ impl DataUsageCache {
             self.delete_recursive(&candidate);
             self.replace_hashed(&candidate, &None, &flat);
 
-            remove -= removing;
-            leaves.remove(0);
+            remove = remove.saturating_sub(removing);
+            candidate_index += 1;
         }
     }
 
@@ -869,23 +868,24 @@ struct Inner {
     path: DataUsageHash,
 }
 
-fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, leaves: &mut Vec<Inner>) {
+fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, candidates: &mut Vec<Inner>) -> usize {
     let e = match data_usage_cache.cache.get(&path.key()) {
         Some(e) => e,
-        None => return,
+        None => return 0,
     };
-    if !e.children.is_empty() {
-        return;
-    }
-
-    let sz = data_usage_cache.size_recursive(&path.key()).unwrap_or_default();
-    leaves.push(Inner {
-        objects: sz.objects,
-        path: path.clone(),
-    });
+    let mut objects = e.objects;
     for ch in e.children.iter() {
-        add(data_usage_cache, &DataUsageHash(ch.clone()), leaves);
+        objects += add(data_usage_cache, &DataUsageHash(ch.clone()), candidates);
     }
+    // Collect internal nodes (with children) as compaction candidates.
+    // Leaf nodes have no children to remove, so compacting them is a no-op.
+    if !e.children.is_empty() {
+        candidates.push(Inner {
+            objects,
+            path: path.clone(),
+        });
+    }
+    objects
 }
 
 fn mark(duc: &DataUsageCache, entry: &DataUsageEntry, found: &mut HashSet<String>) {
@@ -1361,4 +1361,161 @@ mod tests {
         assert_eq!(child_entry.size, 30);
         assert_eq!(child_entry.objects, 3);
     }
+
+    // --- Tests for `add` and `reduce_children_of` (bug fixes) ---
+
+    /// Build a small tree: root -> child1 (leaf), child2 -> grandchild (leaf).
+    fn build_test_tree() -> (DataUsageCache, DataUsageHash) {
+        let root = hash_path("bucket");
+        let c1 = hash_path("bucket/a");
+        let c2 = hash_path("bucket/b");
+        let gc = hash_path("bucket/b/c");
+
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &c1,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                size: 10,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &c2,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                objects: 2,
+                size: 20,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &gc,
+            &Some(c2.clone()),
+            &DataUsageEntry {
+                objects: 3,
+                size: 30,
+                ..Default::default()
+            },
+        );
+        (cache, root)
+    }
+
+    fn build_underflow_test_tree() -> (DataUsageCache, DataUsageHash) {
+        let root = hash_path("bucket");
+        let small = hash_path("bucket/small");
+        let small_a = hash_path("bucket/small/a");
+        let small_b = hash_path("bucket/small/b");
+        let large = hash_path("bucket/large");
+        let large_a = hash_path("bucket/large/a");
+        let large_b = hash_path("bucket/large/b");
+
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(
+            &root,
+            &None,
+            &DataUsageEntry {
+                objects: 100,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(&small, &Some(root.clone()), &DataUsageEntry::default());
+        cache.replace_hashed(
+            &small_a,
+            &Some(small.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &small_b,
+            &Some(small.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(&large, &Some(root.clone()), &DataUsageEntry::default());
+        cache.replace_hashed(
+            &large_a,
+            &Some(large.clone()),
+            &DataUsageEntry {
+                objects: 10,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &large_b,
+            &Some(large.clone()),
+            &DataUsageEntry {
+                objects: 10,
+                ..Default::default()
+            },
+        );
+        (cache, root)
+    }
+
+    #[test]
+    fn test_add_collects_internal_nodes_as_compaction_candidates() {
+        let (cache, root) = build_test_tree();
+        let mut candidates = Vec::new();
+        add(&cache, &root, &mut candidates);
+
+        let mut paths: Vec<String> = candidates.iter().map(|l| l.path.key()).collect();
+        paths.sort();
+        assert_eq!(paths.len(), 2, "add() should find internal nodes with children");
+        assert!(paths.contains(&hash_path("bucket").key()));
+        assert!(paths.contains(&hash_path("bucket/b").key()));
+    }
+
+    #[test]
+    fn test_add_skips_leaf_node() {
+        let mut cache = DataUsageCache::default();
+        let h = hash_path("single-leaf");
+        cache.replace_hashed(
+            &h,
+            &None,
+            &DataUsageEntry {
+                objects: 5,
+                size: 50,
+                ..Default::default()
+            },
+        );
+
+        let mut candidates = Vec::new();
+        add(&cache, &h, &mut candidates);
+        assert!(candidates.is_empty(), "leaf node should not be a compaction candidate");
+    }
+
+    #[test]
+    fn test_reduce_children_of_compacts_internal_node() {
+        let (mut cache, root) = build_test_tree();
+        cache.reduce_children_of(&root, 2, false);
+
+        let entry_c2 = cache.find("bucket/b").unwrap();
+        assert!(entry_c2.compacted, "internal node 'bucket/b' should be compacted");
+        let entry_c1 = cache.find("bucket/a").unwrap();
+        assert!(!entry_c1.compacted, "leaf 'bucket/a' should not be compacted");
+        assert!(cache.find("bucket/b/c").is_none(), "grandchild should be removed");
+    }
+
+    #[test]
+    fn test_reduce_children_of_usize_underflow_saturates() {
+        let (mut cache, root) = build_underflow_test_tree();
+
+        // total children=6, limit=5, remove=1. The smallest candidate removes
+        // two descendants, so plain subtraction would underflow and compact the
+        // next candidate too.
+        cache.reduce_children_of(&root, 5, false);
+
+        assert!(cache.find("bucket/small").is_some_and(|entry| entry.compacted));
+        assert!(cache.find("bucket/small/a").is_none());
+        assert!(cache.find("bucket/small/b").is_none());
+        assert!(cache.find("bucket/large").is_some_and(|entry| !entry.compacted));
+        assert!(cache.find("bucket/large/a").is_some());
+        assert!(cache.find("bucket/large/b").is_some());
+    }
 }
diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs
index b50f5a403..1cb3c116e 100644
--- a/crates/scanner/src/data_usage_define.rs
+++ b/crates/scanner/src/data_usage_define.rs
@@ -410,15 +410,14 @@ impl DataUsageCache {
             return;
         }
 
-        let mut leaves = Vec::new();
+        let mut candidates = Vec::new();
         let mut remove = total - limit;
-        add(self, path, &mut leaves);
-        leaves.sort_by_key(|a| a.objects);
+        add(self, path, &mut candidates);
+        candidates.sort_by_key(|a| a.objects);
 
-        while remove > 0 && !leaves.is_empty() {
-            let Some(e) = leaves.first() else {
-                break;
-            };
+        let mut candidate_index = 0;
+        while remove > 0 && candidate_index < candidates.len() {
+            let e = &candidates[candidate_index];
             let candidate = e.path.clone();
             if candidate == *path && !compact_self {
                 break;
@@ -427,7 +426,7 @@ impl DataUsageCache {
             let mut flat = match self.size_recursive(&candidate.key()) {
                 Some(flat) => flat,
                 None => {
-                    leaves.remove(0);
+                    candidate_index += 1;
                     continue;
                 }
             };
@@ -436,8 +435,8 @@ impl DataUsageCache {
             self.delete_recursive(&candidate);
             self.replace_hashed(&candidate, &None, &flat);
 
-            remove -= removing;
-            leaves.remove(0);
+            remove = remove.saturating_sub(removing);
+            candidate_index += 1;
         }
     }
 
@@ -864,23 +863,25 @@ struct Inner {
     path: DataUsageHash,
 }
 
-fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, leaves: &mut Vec<Inner>) {
+fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, candidates: &mut Vec<Inner>) -> usize {
     let e = match data_usage_cache.cache.get(&path.key()) {
         Some(e) => e,
-        None => return,
+        None => return 0,
     };
-    if !e.children.is_empty() {
-        return;
-    }
-
-    let sz = data_usage_cache.size_recursive(&path.key()).unwrap_or_default();
-    leaves.push(Inner {
-        objects: sz.objects,
-        path: path.clone(),
-    });
+    let mut objects = e.objects;
     for ch in e.children.iter() {
-        add(data_usage_cache, &DataUsageHash(ch.clone()), leaves);
+        objects += add(data_usage_cache, &DataUsageHash(ch.clone()), candidates);
     }
+    // Collect internal nodes (with children) as compaction candidates.
+    // Leaf nodes have no children to remove, so compacting them is a no-op —
+    // total_children_rec returns 0 for leaves, so `remove` would never decrement.
+    if !e.children.is_empty() {
+        candidates.push(Inner {
+            objects,
+            path: path.clone(),
+        });
+    }
+    objects
 }
 
 fn mark(duc: &DataUsageCache, entry: &DataUsageEntry, found: &mut HashSet<String>) {
@@ -1177,6 +1178,196 @@ mod tests {
         assert_eq!(attempts.load(Ordering::SeqCst), 3);
     }
 
+    // --- Tests for `add` function (bug #1: logic inversion) ---
+
+    /// Build a small tree: root -> child1 (leaf), child2 -> grandchild (leaf).
+    /// Returns (cache, root_hash).
+    fn build_test_tree() -> (DataUsageCache, DataUsageHash) {
+        let root = hash_path("bucket");
+        let c1 = hash_path("bucket/a");
+        let c2 = hash_path("bucket/b");
+        let gc = hash_path("bucket/b/c");
+
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &c1,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                size: 10,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &c2,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                objects: 2,
+                size: 20,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &gc,
+            &Some(c2.clone()),
+            &DataUsageEntry {
+                objects: 3,
+                size: 30,
+                ..Default::default()
+            },
+        );
+        (cache, root)
+    }
+
+    fn build_underflow_test_tree() -> (DataUsageCache, DataUsageHash) {
+        let root = hash_path("bucket");
+        let small = hash_path("bucket/small");
+        let small_a = hash_path("bucket/small/a");
+        let small_b = hash_path("bucket/small/b");
+        let large = hash_path("bucket/large");
+        let large_a = hash_path("bucket/large/a");
+        let large_b = hash_path("bucket/large/b");
+
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(
+            &root,
+            &None,
+            &DataUsageEntry {
+                objects: 100,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(&small, &Some(root.clone()), &DataUsageEntry::default());
+        cache.replace_hashed(
+            &small_a,
+            &Some(small.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &small_b,
+            &Some(small.clone()),
+            &DataUsageEntry {
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(&large, &Some(root.clone()), &DataUsageEntry::default());
+        cache.replace_hashed(
+            &large_a,
+            &Some(large.clone()),
+            &DataUsageEntry {
+                objects: 10,
+                ..Default::default()
+            },
+        );
+        cache.replace_hashed(
+            &large_b,
+            &Some(large.clone()),
+            &DataUsageEntry {
+                objects: 10,
+                ..Default::default()
+            },
+        );
+        (cache, root)
+    }
+
+    #[test]
+    fn test_add_collects_internal_nodes_as_compaction_candidates() {
+        // `add()` should collect internal nodes (with children) as compaction candidates.
+        // Leaf nodes have no children to remove, so compacting them is a no-op.
+        let (cache, root) = build_test_tree();
+        let mut candidates = Vec::new();
+        add(&cache, &root, &mut candidates);
+
+        let mut paths: Vec<String> = candidates.iter().map(|l| l.path.key()).collect();
+        paths.sort();
+        // Internal nodes: "bucket" (children: [a, b]) and "bucket/b" (children: [c]).
+        // Leaf nodes "bucket/a" and "bucket/b/c" are NOT collected.
+        assert_eq!(paths.len(), 2, "add() should find internal nodes with children");
+        assert!(paths.contains(&hash_path("bucket").key()));
+        assert!(paths.contains(&hash_path("bucket/b").key()));
+    }
+
+    #[test]
+    fn test_add_returns_empty_for_missing_path() {
+        let cache = DataUsageCache::default();
+        let mut candidates = Vec::new();
+        add(&cache, &hash_path("nonexistent"), &mut candidates);
+        assert!(candidates.is_empty());
+    }
+
+    #[test]
+    fn test_add_skips_leaf_node() {
+        // A leaf node (no children) is not a valid compaction candidate —
+        // total_children_rec returns 0 for leaves, so compacting them has no effect.
+        let mut cache = DataUsageCache::default();
+        let h = hash_path("single-leaf");
+        cache.replace_hashed(
+            &h,
+            &None,
+            &DataUsageEntry {
+                objects: 5,
+                size: 50,
+                ..Default::default()
+            },
+        );
+
+        let mut candidates = Vec::new();
+        add(&cache, &h, &mut candidates);
+        assert!(candidates.is_empty(), "leaf node should not be a compaction candidate");
+    }
+
+    // --- Tests for `reduce_children_of` (bug #2: usize underflow) ---
+
+    #[test]
+    fn test_reduce_children_of_compacts_internal_node() {
+        // Build tree: root -> c1(leaf), c2 -> gc(leaf). total=3, limit=2, remove=1.
+        // Internal nodes (compaction candidates): root, c2.
+        // compact_self=false skips root; c2 (objects=2, 1 child) is the only candidate.
+        // Compacting c2 removes gc (removing=1), satisfying remove=1.
+        let (mut cache, root) = build_test_tree();
+        cache.reduce_children_of(&root, 2, false);
+
+        // "bucket/b" should be compacted (its child gc removed).
+        let entry_c2 = cache.find("bucket/b").unwrap();
+        assert!(entry_c2.compacted, "internal node 'bucket/b' should be compacted");
+        // "bucket/a" (leaf, not a candidate) should remain unchanged.
+        let entry_c1 = cache.find("bucket/a").unwrap();
+        assert!(!entry_c1.compacted, "leaf 'bucket/a' should not be compacted");
+        // "bucket/b/c" was deleted when its parent was compacted.
+        assert!(cache.find("bucket/b/c").is_none(), "grandchild should be removed after parent compaction");
+    }
+
+    #[test]
+    fn test_reduce_children_of_no_op_when_under_limit() {
+        let (mut cache, root) = build_test_tree();
+        let before = cache.cache.len();
+        // limit=10 > total children => no compaction
+        cache.reduce_children_of(&root, 10, false);
+        assert_eq!(cache.cache.len(), before);
+    }
+
+    #[test]
+    fn test_reduce_children_of_usize_underflow_saturates() {
+        let (mut cache, root) = build_underflow_test_tree();
+
+        // total children=6, limit=5, remove=1. The smallest candidate removes
+        // two descendants, so plain subtraction would underflow and compact the
+        // next candidate too.
+        cache.reduce_children_of(&root, 5, false);
+
+        assert!(cache.find("bucket/small").is_some_and(|entry| entry.compacted));
+        assert!(cache.find("bucket/small/a").is_none());
+        assert!(cache.find("bucket/small/b").is_none());
+        assert!(cache.find("bucket/large").is_some_and(|entry| !entry.compacted));
+        assert!(cache.find("bucket/large/a").is_some());
+        assert!(cache.find("bucket/large/b").is_some());
+    }
+
     #[tokio::test]
     async fn test_retry_save_op_times_out_and_returns_error_after_retries() {
         let attempts = Arc::new(AtomicUsize::new(0));