Recurse delete fan-out to every directory, not just the top-level orphan

.github/workflows/ci.yml

@@ -115,6 +115,7 @@ jobs:
           - chunk_resilience_e2e   # agent dies mid-copy; leases expire and re-grant
           - deep_e2e               # directory chain deeper than the walker's in-agent limit
           - delete_fanout_e2e      # pathological orphan directory fans out across DELETE shards
+          - delete_fanout_nested_e2e # fan-out applies at every depth, not just the top-level orphan path
           - dirfix_e2e             # DIRFIX over a directory that fans out to entry-lists
           - direct_write_e2e       # copy.direct_write: new files no-temp, updates atomic
           - fanout_e2e             # a small volume must still use the whole fleet

README.md

@@ -63,6 +63,7 @@ tree, and builds the binaries it needs itself:
 | `chunk_resilience_e2e.sh` | agent dies mid-copy; leases expire and re-grant |
 | `deep_e2e.sh` | directory chain deeper than the walker's in-agent limit |
 | `delete_fanout_e2e.sh` | pathological orphan directory fans out across DELETE shards |
+| `delete_fanout_nested_e2e.sh` | fan-out applies at every depth, not just the top-level orphan path |
 | `dirfix_e2e.sh` | DIRFIX over a directory that fans out to entry-lists |
 | `direct_write_e2e.sh` | `copy.direct_write`: new files skip the temp+rename, updates stay atomic |
 | `fanout_e2e.sh` | a small volume must still use the whole fleet |

agent/src/delete.c

@@ -13,12 +13,16 @@
  * unlinked depth-first by this one agent — the delete-pass analogue of
  * walker.c's split_entrylist_stream, but simpler: there is nothing to diff
  * against a destination (the whole subtree is already condemned), so a batch
- * is just names to remove. Only checked at the top level (the orphan path
- * directly named in this shard's paths[]), not recursively at every nested
- * depth — the same scope split_entrylist_stream's own trigger has (it only
- * fires for the directory the walker is currently sitting in). A nested
- * subdirectory that is itself pathological is still removed correctly by
- * whichever shard reaches it, just not further split. */
+ * is just names to remove. remove_object() applies this check at EVERY
+ * directory the removal touches, not just the top-level path named in the
+ * shard's paths[] — rm_dir_contents() routes every entry it finds back
+ * through remove_object() rather than recursing directly, so a directory
+ * that is individually small still gets caught if it is nested many levels
+ * deep under a huge tree of otherwise-small directories (a wide/deep tree
+ * where no single directory looks large from its own parent's point of view
+ * previously never split at all, and got removed serially by one agent
+ * thread — see docs/DESIGN-coordinator.md §2.2 for the incident that
+ * surfaced this). */
 #include "agent.h"
 
 #include <dirent.h>
@@ -38,41 +42,45 @@
  * its own right. */
 #define DELETE_SPLIT_BATCH_DEFAULT 20000
 
-/* depth-first removal of name inside parentfd; returns removed count */
-static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name,
-                        const char *rel)
+static uint64_t remove_object(struct walk_ctx *ctx, int parentfd,
+                              const char *name, const char *rel);
+
+/* depth-first removal of the already-open directory d (fd dirfd(d), path
+ * rel) — every entry goes through remove_object, so a subdirectory found
+ * at ANY depth that turns out to be pathological is streamed out via
+ * stream_delete_split exactly like a top-level orphan, not just the one
+ * named directly in the shard's paths[] (see the fan-out comment at the top
+ * of this file). Does not remove rel itself or consume d; the caller does
+ * both, so remove_object can decide not to (a directory it just handed off
+ * to a split must NOT be rmdir'd here — the split's own cleanup shard owns
+ * that once every batch it produced has drained). */
+static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel)
 {
-    struct stat st;
-    if (fstatat(parentfd, name, &st, AT_SYMLINK_NOFOLLOW) < 0) {
-        if (errno != ENOENT) /* already gone is success */
-            walk_err(ctx, "stat for delete", rel);
-        return 0;
+    uint64_t removed = 0;
+    struct dirent *de;
+    while ((de = readdir(d))) {
+        if (de->d_name[0] == '.' &&
+            (de->d_name[1] == '\0' ||
+             (de->d_name[1] == '.' && de->d_name[2] == '\0')))
+            continue;
+        char crel[PATH_MAX];
+        snprintf(crel, sizeof crel, "%s/%s", rel, de->d_name);
+        removed += remove_object(ctx, dirfd(d), de->d_name, crel);
     }
+    return removed;
+}
+
+/* Fully removes name (file, or directory including every descendant) inside
+ * parentfd — depth-first, via rm_dir_contents. Returns removed count. Only
+ * reached for a directory once remove_object has already decided it is NOT
+ * pathological (under delete_split_threshold), so no further probing here. */
+static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name,
+                        const char *rel, DIR *already_open)
+{
     uint64_t removed = 0;
-    if (S_ISDIR(st.st_mode)) {
-        int fd = openat(parentfd, name,
-                        O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
-        if (fd < 0) {
-            walk_err(ctx, "open for delete", rel);
-            return 0;
-        }
-        DIR *d = fdopendir(fd);
-        if (!d) {
-            close(fd);
-            walk_err(ctx, "fdopendir for delete", rel);
-            return 0;
-        }
-        struct dirent *de;
-        while ((de = readdir(d))) {
-            if (de->d_name[0] == '.' &&
-                (de->d_name[1] == '\0' ||
-                 (de->d_name[1] == '.' && de->d_name[2] == '\0')))
-                continue;
-            char crel[PATH_MAX];
-            snprintf(crel, sizeof crel, "%s/%s", rel, de->d_name);
-            removed += rm_tree(ctx, dirfd(d), de->d_name, crel);
-        }
-        closedir(d);
+    if (already_open) {
+        removed = rm_dir_contents(ctx, already_open, rel);
+        closedir(already_open);
         if (unlinkat(parentfd, name, AT_REMOVEDIR) < 0 && errno != ENOENT) {
             walk_err(ctx, "rmdir", rel);
             return removed;
@@ -204,69 +212,86 @@ int open_parent_beneath(int root_fd, const char *rel, const char **leaf)
     return cur;
 }
 
-/* Removes rel, fanning out to stream_delete_split if it is a directory over
- * delete_split_threshold. Returns the number of objects this shard itself
- * removed (0 if it handed the directory off to a split instead). */
-static uint64_t remove_orphan(struct walk_ctx *ctx, int pfd, const char *leaf,
-                              const char *rel)
+/* Removes name (file or directory) inside parentfd, fanning out to
+ * stream_delete_split instead of recursing if it is a directory over
+ * delete_split_threshold. Called both for a shard's top-level orphan paths
+ * (process_delete) and for every entry rm_dir_contents finds while
+ * descending an already-accepted directory — so a pathologically large
+ * subdirectory found at ANY depth is caught, not just one named directly in
+ * the shard's paths[] (the gap the single-level-only version of this check
+ * left: a tree of many individually-small subdirectories summing to tens of
+ * millions of entries never split at all, since no single directory in the
+ * chain ever looked large from its own parent's point of view). Returns the
+ * number of objects this shard itself removed (0 if it handed the directory
+ * off to a split instead — those objects are counted by the split-produced
+ * shards that actually remove them). */
+static uint64_t remove_object(struct walk_ctx *ctx, int parentfd,
+                              const char *name, const char *rel)
 {
     struct stat st;
-    if (fstatat(pfd, leaf, &st, AT_SYMLINK_NOFOLLOW) < 0) {
-        if (errno != ENOENT)
+    if (fstatat(parentfd, name, &st, AT_SYMLINK_NOFOLLOW) < 0) {
+        if (errno != ENOENT) /* already gone is success */
             walk_err(ctx, "stat for delete", rel);
         return 0;
     }
+    if (!S_ISDIR(st.st_mode))
+        return rm_tree(ctx, parentfd, name, rel, NULL);
+
     uint64_t threshold = ctx->oe->o.delete_split_threshold;
-    if (S_ISDIR(st.st_mode) && threshold) {
-        /* Bounded probe: open the directory once just to count up to
-         * threshold+1 entries and decide pathological without materialising
-         * the whole thing — same idea as walker.c's read_entries_upto,
-         * reimplemented here rather than shared, since delete's probe needs
-         * no stat placeholders or destination-side bookkeeping, only a
-         * count. This fd is closed here regardless of outcome: readdir's
-         * position is tied to the fd's own open file description, so
-         * reusing it for the real stream below would silently resume
-         * partway through instead of at the start — a second, independent
-         * openat for the pathological case is one extra syscall against
-         * what is, by construction, a very large directory. */
-        int pfd2 = openat(pfd, leaf, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
-        if (pfd2 < 0) {
-            walk_err(ctx, "open for delete probe", rel);
-            return 0;
-        }
-        DIR *pd = fdopendir(pfd2);
-        if (!pd) {
-            close(pfd2);
-            walk_err(ctx, "fdopendir for delete probe", rel);
-            return 0;
-        }
-        uint64_t seen = 0;
-        struct dirent *de;
-        errno = 0;
-        while (seen <= threshold && (de = readdir(pd))) {
-            if (de->d_name[0] == '.' &&
-                (de->d_name[1] == '\0' || (de->d_name[1] == '.' && de->d_name[2] == '\0')))
-                continue;
-            seen++;
-        }
-        bool probe_err = errno != 0;
-        closedir(pd); /* closes pfd2 too */
-        if (probe_err) {
-            walk_err(ctx, "probe dir for delete", rel);
-            return 0;
-        }
-        if (seen > threshold) {
-            int fd = openat(pfd, leaf, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
-            if (fd < 0) {
-                walk_err(ctx, "open for delete split", rel);
-                return 0;
-            }
-            stream_delete_split(ctx, rel, fd); /* consumes fd */
+    if (!threshold)
+        return rm_tree(ctx, parentfd, name, rel, NULL);
+
+    int fd = openat(parentfd, name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    if (fd < 0) {
+        walk_err(ctx, "open for delete probe", rel);
+        return 0;
+    }
+    DIR *d = fdopendir(fd);
+    if (!d) {
+        close(fd);
+        walk_err(ctx, "fdopendir for delete probe", rel);
+        return 0;
+    }
+    /* Bounded probe: count up to threshold+1 entries without materialising
+     * the whole directory — same idea as walker.c's read_entries_upto,
+     * reimplemented here rather than shared, since delete's probe needs no
+     * stat placeholders or destination-side bookkeeping, only a count. */
+    uint64_t seen = 0;
+    struct dirent *de;
+    errno = 0;
+    while (seen <= threshold && (de = readdir(d))) {
+        if (de->d_name[0] == '.' &&
+            (de->d_name[1] == '\0' || (de->d_name[1] == '.' && de->d_name[2] == '\0')))
+            continue;
+        seen++;
+    }
+    if (errno) {
+        walk_err(ctx, "probe dir for delete", rel);
+        closedir(d);
+        return 0;
+    }
+    if (seen > threshold) {
+        /* rewinddir so the stream starts from the first entry, not wherever
+         * the probe's readdir left off. stream_delete_split takes ownership
+         * of d's fd (its own fdopendir + closedir) — do not closedir(d)
+         * here too, or the fd double-closes. */
+        rewinddir(d);
+        int dupfd = dup(dirfd(d));
+        closedir(d);
+        if (dupfd < 0) {
+            walk_err(ctx, "dup for delete split", rel);
             return 0;
         }
-        /* Under threshold: fall through to the ordinary recursive removal. */
+        stream_delete_split(ctx, rel, dupfd);
+        return 0;
     }
-    return rm_tree(ctx, pfd, leaf, rel);
+    /* Under threshold: reuse this same handle for the real removal —
+     * rewinddir resets ITS OWN position (unlike dup(), which shares the
+     * original's offset — see this file's history for why that distinction
+     * matters), so no second openat is needed on the common case, which now
+     * runs at every depth instead of once per top-level orphan. */
+    rewinddir(d);
+    return rm_tree(ctx, parentfd, name, rel, d);
 }
 
 void process_delete(const struct shard_item *it)
@@ -304,12 +329,12 @@ void process_delete(const struct shard_item *it)
             continue;
         }
         /* counters: a delete pass reports removals in the orphans column.
-         * remove_orphan returns 0 (not an undercount) when it fans a
+         * remove_object returns 0 (not an undercount) when it fans a
          * pathological directory out instead of removing it directly — those
          * objects are counted by the split-produced shards that actually
          * remove them, same as entry-list fan-out shifts the copy counters
          * onto the children instead of the walker that discovered them. */
-        CTR_ADD(ctx.c.orphans, remove_orphan(&ctx, pfd, leaf, rel));
+        CTR_ADD(ctx.c.orphans, remove_object(&ctx, pfd, leaf, rel));
         close(pfd);
     }
 

coordinator/internal/agentsrv/server.go

@@ -565,12 +565,12 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error {
 		// bumps delete_groups.n_total by one per entry, so it stays in the same
 		// units as n_done (shards, not directory entries — see delete_groups'
 		// doc comment). total_children > 0 only marks LastBatch, the "readdir
-		// hit EOF" signal; the cleanup shard is pre-built regardless of whether
-		// every sibling has already reported done — RecordSplit only actually
-		// inserts it once that's true.
+		// hit EOF" signal; BuildCleanup is only actually called once every
+		// sibling AND any nested child group has reported done — RecordSplit
+		// decides that, not this loop.
 		deleteTotals = append(deleteTotals, store.DeleteGroupTotal{
 			DirRel: string(dr.DirRel), LastBatch: dr.TotalChildren > 0,
-			CleanupShard: deleteCleanupShard(string(dr.DirRel)),
+			BuildCleanup: deleteCleanupShard,
 		})
 	}
 
@@ -761,10 +761,11 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error {
 			// seedDeletePass never does) — maintain delete_groups the same
 			// way completeChunk maintains chunk_groups, seeding a cleanup
 			// shard for the now-possibly-empty directory once every sibling
-			// has reported done.
+			// AND any nested child group (fan-out applies at every depth, not
+			// just relPath's own top level) has reported done.
 			blob, _ := proto.Marshal(r)
 			err = s.st.CompleteDeleteRemainder(shardID, leaseID, passID, relPath, blob,
-				deleteCleanupShard(relPath), r.Counters)
+				deleteCleanupShard(relPath), deleteCleanupShard, r.Counters)
 		default:
 			blob, _ := proto.Marshal(r)
 			err = s.st.CompleteShard(shardID, leaseID, passID, blob, r.Counters)