Fan out orphan-directory deletion across the fleet

.github/workflows/ci.yml

@@ -114,6 +114,7 @@ jobs:
           - chunk_abort_reclaim_e2e # abandoned chunk group is reclaimed
           - 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
           - 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

@@ -62,6 +62,7 @@ tree, and builds the binaries it needs itself:
 | `chunk_abort_reclaim_e2e.sh` | a chunk group abandoned mid-assembly is reclaimed |
 | `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 |
 | `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/Makefile

@@ -2,7 +2,7 @@ CC      ?= gcc
 CFLAGS  ?= -std=c11 -O2 -g -Wall -Wextra -Wshadow -D_GNU_SOURCE -Ivendor
 LDFLAGS ?= -pthread -lzstd -lssl -lcrypto
 
-SRC := src/pb.c src/msgs.c src/wire.c src/tls.c src/state.c src/uring.c src/copy.c src/poolsize.c src/ucopy.c src/chunk.c src/xattr.c src/jrn.c src/delete.c src/verify.c src/filter.c src/probe.c src/dirfix.c src/link.c src/tempname.c src/walker.c src/main.c
+SRC := src/pb.c src/msgs.c src/wire.c src/tls.c src/state.c src/uring.c src/copy.c src/poolsize.c src/ucopy.c src/chunk.c src/xattr.c src/jrn.c src/split.c src/delete.c src/verify.c src/filter.c src/probe.c src/dirfix.c src/link.c src/tempname.c src/walker.c src/main.c
 OBJ := $(SRC:.c=.o)
 BIN := bin/drsync-agent
 
@@ -106,9 +106,9 @@ bin/estat_test: test/estat_test.c src/uring.o src/jrn.o src/msgs.o src/pb.o
 link-test: bin/link_test
 	./bin/link_test
 
-bin/link_test: test/link_test.c src/link.o src/delete.o src/jrn.o src/msgs.o src/pb.o src/state.o src/uring.o
+bin/link_test: test/link_test.c src/link.o src/delete.o src/split.o src/jrn.o src/msgs.o src/pb.o src/state.o src/uring.o
 	@mkdir -p bin
-	$(CC) $(CFLAGS) -o $@ test/link_test.c src/link.o src/delete.o src/jrn.o src/msgs.o src/pb.o src/state.o src/uring.o -pthread -lzstd
+	$(CC) $(CFLAGS) -o $@ test/link_test.c src/link.o src/delete.o src/split.o src/jrn.o src/msgs.o src/pb.o src/state.o src/uring.o -pthread -lzstd
 
 clean:
 	rm -f $(OBJ) $(BIN) bin/filter_test bin/fidelity_test bin/xattr_copy_test bin/ucopy_test bin/tempname_test bin/poolsize_test bin/state_test bin/opts_test bin/estat_test bin/link_test

agent/src/agent.h

@@ -291,6 +291,19 @@ struct walk_ctx {
     bool                     fatal;
 };
 
+/* How long ship_split's ack-wait blocks before giving up and marking the
+ * shard fatal (so it re-runs) — a coordinator that never acks within this
+ * window is treated as unreachable, not as "still working on it". */
+#define SPLIT_ACK_TIMEOUT_S 120
+
+/* Shared ShardSplit outbox machinery (split.c): ships a prepared split frame
+ * without blocking (backpressure only once SPLIT_WINDOW acks are
+ * outstanding), and drains every outstanding ack before a shard reports its
+ * result (protocol doc §4.2's ordering invariant). Used by the directory
+ * walker (walker.c) and the delete-pass executor (delete.c). */
+void ship_split(struct walk_ctx *ctx, pb_buf *b);
+void drain_splits(struct walk_ctx *ctx);
+
 void jrn_init(struct walk_ctx *ctx);
 /* thread-safe append; auto-flushes at the batch size threshold */
 void jrn_emit(struct walk_ctx *ctx, int type, const char *rel_path,

agent/src/delete.c

@@ -5,7 +5,20 @@
  * fd-anchored beneath the destination root (openat2 semantics via the same
  * open_beneath discipline as the walker), never by absolute path.
  * Every removed object is journaled JR_DELETED; dry-run jobs journal
- * JR_WOULD_DELETE and remove nothing. */
+ * JR_WOULD_DELETE and remove nothing.
+ *
+ * Fan-out (docs/DESIGN-coordinator.md §2.2 DELETE fan-out): a directory whose
+ * own entry count exceeds delete_split_threshold is streamed out as new
+ * DELETE shards (delete_split_batch names per shard) instead of being
+ * 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. */
 #include "agent.h"
 
 #include <dirent.h>
@@ -18,6 +31,13 @@
 #include <time.h>
 #include <unistd.h>
 
+/* Default names per split-produced DELETE shard when the job doesn't override
+ * it (tuning.delete_split_batch). Delete work per name is a plain unlink, not
+ * a full stat/diff/copy pipeline, so this can run higher than
+ * ENTRYLIST_BATCH_DEFAULT before one shard's batch becomes a bottleneck in
+ * 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)
@@ -67,6 +87,91 @@ static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name,
     return removed + 1;
 }
 
+/* Ships one batch of names still to remove under rel as a DeleteRemainder
+ * split. total is 0 for every batch except the last one streamed for rel,
+ * which carries the true count the coordinator uses to know when every
+ * sibling shard has reported done (store.RecordSplit / CompleteDeleteRemainder
+ * — the total is only known once readdir hits EOF, unlike a chunk group's
+ * upfront byte-size-derived n_chunks). */
+static void flush_delete_split(struct walk_ctx *ctx, const char *rel,
+                               char *const *names, size_t n, uint32_t total)
+{
+    pb_buf b;
+    pb_init(&b);
+    enc_delete_split(&b, ctx->it->shard_id, ctx->split_seq, rel, names, n, total);
+    ship_split(ctx, &b);
+}
+
+/* Streams a pathological directory's entries out as DeleteRemainder splits
+ * instead of removing them depth-first in this shard. Mirrors walker.c's
+ * split_entrylist_stream: names are shipped in batches as readdir yields
+ * them (peak memory one batch, not the whole directory), and every batch
+ * after the count check has already committed to streaming, so there is no
+ * "go back to rm_tree" fallback mid-stream — once a directory is judged
+ * pathological, it is fully handed off.
+ *
+ * The directory rel itself is NOT removed here: it cannot be, until every
+ * split-produced child shard has finished emptying it, which this shard has
+ * no way to know (they run on other agents, other leases, later). The
+ * coordinator seeds a cleanup shard for rel once delete_groups' n_done
+ * reaches the n_total this function's final batch reports (store.go
+ * CompleteDeleteRemainder). fd is consumed (closed) either way. */
+static void stream_delete_split(struct walk_ctx *ctx, const char *rel, int fd)
+{
+    DIR *d = fdopendir(fd);
+    if (!d) {
+        close(fd);
+        walk_err(ctx, "fdopendir for delete split", rel);
+        return;
+    }
+    size_t batch_max = ctx->oe->o.delete_split_batch
+                          ? ctx->oe->o.delete_split_batch
+                          : DELETE_SPLIT_BATCH_DEFAULT;
+    char **batch = calloc(batch_max, sizeof *batch);
+    if (!batch) {
+        closedir(d);
+        walk_err(ctx, "oom delete split stream", rel);
+        return;
+    }
+    size_t nb = 0;
+    uint32_t total = 0;
+    struct dirent *de;
+    errno = 0;
+    while (!ctx->fatal && (de = readdir(d))) {
+        if (de->d_name[0] == '.' &&
+            (de->d_name[1] == '\0' || (de->d_name[1] == '.' && de->d_name[2] == '\0')))
+            continue;
+        /* d_name is only valid until the next readdir, and a batch spans
+         * many of them, so each name is copied and released once shipped. */
+        batch[nb] = strdup(de->d_name);
+        if (!batch[nb]) {
+            CTR_ADD(ctx->c.errors, 1);
+            break;
+        }
+        nb++;
+        total++;
+        if (nb >= batch_max) {
+            flush_delete_split(ctx, rel, batch, nb, 0);
+            for (size_t i = 0; i < nb; i++)
+                free(batch[i]);
+            nb = 0;
+        }
+        errno = 0;
+    }
+    if (errno)
+        walk_err(ctx, "read dir for delete split", rel);
+    closedir(d); /* also closes fd */
+    /* Final batch always ships (even if empty — total must still reach the
+     * coordinator so an already-empty pathological directory's cleanup can
+     * be seeded), carrying the true total this stream produced. */
+    if (!ctx->fatal) {
+        flush_delete_split(ctx, rel, batch, nb, total);
+    }
+    for (size_t i = 0; i < nb; i++)
+        free(batch[i]);
+    free(batch);
+}
+
 /* split rel into (parent dir fd under root, leaf name); -1 on failure */
 int open_parent_beneath(int root_fd, const char *rel, const char **leaf)
 {
@@ -99,12 +204,77 @@ 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)
+{
+    struct stat st;
+    if (fstatat(pfd, leaf, &st, AT_SYMLINK_NOFOLLOW) < 0) {
+        if (errno != ENOENT)
+            walk_err(ctx, "stat for delete", rel);
+        return 0;
+    }
+    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 */
+            return 0;
+        }
+        /* Under threshold: fall through to the ordinary recursive removal. */
+    }
+    return rm_tree(ctx, pfd, leaf, rel);
+}
+
 void process_delete(const struct shard_item *it)
 {
     struct timespec t0, t1;
     clock_gettime(CLOCK_MONOTONIC, &t0);
 
-    struct walk_ctx ctx = { .it = it };
+    struct walk_ctx ctx = { .it = it, .split_seq = 1 };
     jrn_init(&ctx);
     int status = RES_OK;
     ctx.oe = opts_get(it->job_id);
@@ -115,7 +285,7 @@ void process_delete(const struct shard_item *it)
         goto out;
     }
 
-    for (size_t i = 0; i < it->n_paths; i++) {
+    for (size_t i = 0; i < it->n_paths && !ctx.fatal; i++) {
         const char *rel = it->paths[i];
         if (!rel[0] || strstr(rel, "..")) { /* defense in depth */
             walk_err(&ctx, "refusing suspicious delete path", rel);
@@ -133,16 +303,26 @@ void process_delete(const struct shard_item *it)
                 walk_err(&ctx, "open parent for delete", rel);
             continue;
         }
-        /* counters: a delete pass reports removals in the orphans column */
-        CTR_ADD(ctx.c.orphans, rm_tree(&ctx, pfd, leaf, rel));
+        /* counters: a delete pass reports removals in the orphans column.
+         * remove_orphan 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));
         close(pfd);
     }
 
+    drain_splits(&ctx); /* every DeleteRemainder acked before the shard result (protocol §4.2) */
     jrn_flush(&ctx);
     if (!jrn_wait_acked(&ctx)) {
         snprintf(ctx.err, sizeof ctx.err, "journal ack timeout");
         status = RES_TRANSIENT;
     }
+    if (ctx.fatal && ctx.err[0] == '\0')
+        snprintf(ctx.err, sizeof ctx.err, "delete split failed");
+    if (ctx.fatal)
+        status = RES_TRANSIENT;
 out:
     clock_gettime(CLOCK_MONOTONIC, &t1);
     ctx.c.wall_ms = (uint64_t)((t1.tv_sec - t0.tv_sec) * 1000 +

agent/src/msgs.c

@@ -118,6 +118,26 @@ void enc_entrylist_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
     pb_free(&el);
 }
 
+void enc_delete_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
+                      const char *dir_rel, char *const *names, size_t n_names,
+                      uint32_t total_children)
+{
+    pb_put_u64(b, 1, parent_shard_id);
+    pb_put_u64(b, 2, seq);
+    /* one ShardSplit.DeleteRemainder (field 7): dir_rel (1) + repeated
+     * names (2) + total_children (3, 0 unless this is the final batch for
+     * dir_rel — see the proto doc comment) */
+    pb_buf dr;
+    pb_init(&dr);
+    pb_put_bytes(&dr, 1, dir_rel, strlen(dir_rel));
+    for (size_t i = 0; i < n_names; i++)
+        pb_put_bytes(&dr, 2, names[i], strlen(names[i]));
+    if (total_children)
+        pb_put_u64(&dr, 3, total_children);
+    pb_put_msg(b, 7, &dr);
+    pb_free(&dr);
+}
+
 void enc_bigfile_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
                        const struct bigfile *files, size_t n_files)
 {
@@ -481,6 +501,8 @@ static bool dec_tuning_opts(const uint8_t *p, size_t n, struct job_options *o)
         case 3: o->statx_batch = (uint32_t)pb_get_varint(&c); break;
         case 4: o->mtime_slop_ns = (int64_t)pb_get_varint(&c); break;
         case 6: o->entrylist_batch = (uint32_t)pb_get_varint(&c); break;
+        case 7: o->delete_split_threshold = pb_get_varint(&c); break;
+        case 8: o->delete_split_batch = (uint32_t)pb_get_varint(&c); break;
         default: pb_skip(&c, wt);
         }
     }

agent/src/msgs.h

@@ -142,6 +142,13 @@ struct job_options {
     uint64_t shard_budget;
     uint64_t dir_split_threshold;
     uint32_t entrylist_batch; /* names per entry-list shard (0 = built-in default) */
+    /* Delete-pass analogue of the two above (docs/DESIGN-coordinator.md §2.2
+     * DELETE fan-out): a directory being orphan-deleted whose own entry
+     * count exceeds delete_split_threshold is streamed out as new DELETE
+     * shards, delete_split_batch names per shard, instead of being unlinked
+     * depth-first by one agent. */
+    uint64_t delete_split_threshold;
+    uint32_t delete_split_batch; /* names per split-produced DELETE shard (0 = built-in default) */
     uint32_t statx_batch; /* target statx in flight per walker ⇒ io_uring ring depth */
     int64_t  mtime_slop_ns;
     bool     dry_run;
@@ -323,6 +330,16 @@ void enc_shard_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
  * (the source-side slice of a directory over dir_split_threshold). */
 void enc_entrylist_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
                          const char *dir_rel, char *const *names, size_t n_names);
+/* ShardSplit carrying one DeleteRemainder: a dir_rel plus a batch of names
+ * still to remove under it (the delete-pass analogue of EntryListShard
+ * above — a directory being orphan-deleted over delete_split_threshold).
+ * total_children is 0 for every batch except the last one streamed for
+ * dir_rel, which carries the true split-produced shard count (only known
+ * once readdir hits EOF) so the coordinator can tell when every sibling has
+ * reported done and seed a cleanup shard for dir_rel itself. */
+void enc_delete_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
+                      const char *dir_rel, char *const *names, size_t n_names,
+                      uint32_t total_children);
 /* ShardSplit carrying big files: rel_path + size + mtime_ns each. The
  * coordinator lays them out into ChunkTasks (proto ShardSplit.BigFile). */
 struct bigfile {