diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cd84149..94ba97c 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/README.md b/README.md index 850b031..334df91 100644 --- a/README.md +++ b/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 | diff --git a/agent/Makefile b/agent/Makefile index 2203dc0..2b8cba6 100644 --- a/agent/Makefile +++ b/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 diff --git a/agent/src/agent.h b/agent/src/agent.h index 79330ea..6d02941 100644 --- a/agent/src/agent.h +++ b/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, diff --git a/agent/src/delete.c b/agent/src/delete.c index 5e21e53..567fae4 100644 --- a/agent/src/delete.c +++ b/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 @@ -18,6 +31,13 @@ #include #include +/* 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 + diff --git a/agent/src/msgs.c b/agent/src/msgs.c index c326395..9b5f182 100644 --- a/agent/src/msgs.c +++ b/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); } } diff --git a/agent/src/msgs.h b/agent/src/msgs.h index a2bc3a1..b05b216 100644 --- a/agent/src/msgs.h +++ b/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 { diff --git a/agent/src/split.c b/agent/src/split.c new file mode 100644 index 0000000..a8bb087 --- /dev/null +++ b/agent/src/split.c @@ -0,0 +1,73 @@ +/* Shared ShardSplit outbox machinery: ships a prepared split frame without + * blocking the caller, tracks in-flight acks in a bounded window (SPLIT_ + * WINDOW), and drains every outstanding ack before a shard reports its + * result — the ordering invariant protocol doc §4.2 requires (a shard's + * ShardResult must not be sent until every split it produced is acked). + * + * Used by both the directory walker (subdirs / entry-list batches, + * walker.c) and the delete-pass executor (delete-remainder batches for a + * pathological orphan directory, delete.c) — the ack/backpressure logic is + * identical regardless of what kind of work a split carries, so it lives + * here once instead of twice. */ +#include "agent.h" + +#include +#include +#include + +/* Awaits one in-flight split ack (timeout => fatal so the shard re-runs), then + * unregisters and frees the waiter. */ +static void await_split(struct walk_ctx *ctx, struct split_wait *w) +{ + struct timespec dl; + clock_gettime(CLOCK_REALTIME, &dl); + dl.tv_sec += SPLIT_ACK_TIMEOUT_S; + if (sem_timedwait(&w->sem, &dl) < 0) { + snprintf(ctx->err, sizeof ctx->err, "split ack timeout (seq %llu)", + (unsigned long long)w->seq); + ctx->fatal = true; + } + split_unregister(w); /* removes from the registry and sem_destroys */ + free(w); +} + +/* Ships a prepared ShardSplit frame WITHOUT blocking: the ack is awaited + * later (drain_splits, before the shard result), so consecutive round-trips + * overlap instead of serialising. Blocks only for backpressure when + * SPLIT_WINDOW acks are already outstanding. Consumes seq. */ +void ship_split(struct walk_ctx *ctx, pb_buf *b) +{ + if (ctx->infl_count == SPLIT_WINDOW) { + struct split_wait *old = ctx->infl[ctx->infl_head]; + ctx->infl_head = (ctx->infl_head + 1) % SPLIT_WINDOW; + ctx->infl_count--; + await_split(ctx, old); + } + struct split_wait *w = calloc(1, sizeof *w); + if (!w) { + CTR_ADD(ctx->c.errors, 1); + ctx->fatal = true; + out_push(FR_SHARD_SPLIT, b); /* still recorded (idempotent); just not awaited */ + ctx->split_seq++; + return; + } + w->parent = ctx->it->shard_id; + w->seq = ctx->split_seq; + split_register(w); + out_push(FR_SHARD_SPLIT, b); + ctx->infl[(ctx->infl_head + ctx->infl_count) % SPLIT_WINDOW] = w; + ctx->infl_count++; + ctx->split_seq++; +} + +/* Awaits every outstanding split ack. Must run before reporting the shard + * result so the coordinator has recorded all children first. */ +void drain_splits(struct walk_ctx *ctx) +{ + while (ctx->infl_count > 0) { + struct split_wait *w = ctx->infl[ctx->infl_head]; + ctx->infl_head = (ctx->infl_head + 1) % SPLIT_WINDOW; + ctx->infl_count--; + await_split(ctx, w); + } +} diff --git a/agent/src/walker.c b/agent/src/walker.c index 98c172c..7ed9e31 100644 --- a/agent/src/walker.c +++ b/agent/src/walker.c @@ -24,7 +24,6 @@ #include #define SPLIT_BATCH 4096 -#define SPLIT_ACK_TIMEOUT_S 120 /* Publish in-flight progress every 256 entries. One relaxed atomic store per * 256 entries is noise next to the stat/copy work each entry already costs, @@ -364,63 +363,8 @@ static void copy_symlink(struct walk_ctx *ctx, const char *dir_rel, int sfd, static void handle_orphan(struct walk_ctx *ctx, const char *rel, int dfd, const char *name); -/* Awaits one in-flight split ack (timeout => fatal so the shard re-runs), then - * unregisters and frees the waiter. */ -static void await_split(struct walk_ctx *ctx, struct split_wait *w) -{ - struct timespec dl; - clock_gettime(CLOCK_REALTIME, &dl); - dl.tv_sec += SPLIT_ACK_TIMEOUT_S; - if (sem_timedwait(&w->sem, &dl) < 0) { - snprintf(ctx->err, sizeof ctx->err, "split ack timeout (seq %llu)", - (unsigned long long)w->seq); - ctx->fatal = true; - } - split_unregister(w); /* removes from the registry and sem_destroys */ - free(w); -} - -/* Ships a prepared ShardSplit frame (subdirs or an entry list) WITHOUT blocking: - * the ack is awaited later (drain_splits, before the shard result — the ordering - * invariant of protocol §4.2), so consecutive round-trips overlap instead of - * serialising. Blocks only for backpressure when SPLIT_WINDOW acks are already - * outstanding. Consumes seq. */ -static void ship_split(struct walk_ctx *ctx, pb_buf *b) -{ - if (ctx->infl_count == SPLIT_WINDOW) { - struct split_wait *old = ctx->infl[ctx->infl_head]; - ctx->infl_head = (ctx->infl_head + 1) % SPLIT_WINDOW; - ctx->infl_count--; - await_split(ctx, old); - } - struct split_wait *w = calloc(1, sizeof *w); - if (!w) { - CTR_ADD(ctx->c.errors, 1); - ctx->fatal = true; - out_push(FR_SHARD_SPLIT, b); /* still recorded (idempotent); just not awaited */ - ctx->split_seq++; - return; - } - w->parent = ctx->it->shard_id; - w->seq = ctx->split_seq; - split_register(w); - out_push(FR_SHARD_SPLIT, b); - ctx->infl[(ctx->infl_head + ctx->infl_count) % SPLIT_WINDOW] = w; - ctx->infl_count++; - ctx->split_seq++; -} - -/* Awaits every outstanding split ack. Must run before reporting the shard - * result so the coordinator has recorded all children first. */ -static void drain_splits(struct walk_ctx *ctx) -{ - while (ctx->infl_count > 0) { - struct split_wait *w = ctx->infl[ctx->infl_head]; - ctx->infl_head = (ctx->infl_head + 1) % SPLIT_WINDOW; - ctx->infl_count--; - await_split(ctx, w); - } -} +/* ship_split/drain_splits: shared ShardSplit outbox machinery (split.c), + * also used by delete.c for pathological-orphan-directory fan-out. */ static void flush_splits(struct walk_ctx *ctx) { diff --git a/coordinator/internal/agentsrv/server.go b/coordinator/internal/agentsrv/server.go index 8a88c04..ed1f849 100644 --- a/coordinator/internal/agentsrv/server.go +++ b/coordinator/internal/agentsrv/server.go @@ -520,7 +520,7 @@ func (s *Server) onWorkRequest(ac *agentConn, req *drsyncpb.WorkRequest) error { } func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error { - shards := make([]store.NewShard, 0, len(sp.Subdirs)+len(sp.EntryLists)) + shards := make([]store.NewShard, 0, len(sp.Subdirs)+len(sp.EntryLists)+len(sp.DeleteRemainders)) for _, d := range sp.Subdirs { shards = append(shards, store.NewShard{Kind: model.KindDir, RelPath: string(d.RelPath)}) } @@ -533,6 +533,46 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error { shards = append(shards, store.NewShard{ Kind: model.KindEntryList, RelPath: string(el.DirRel), Payload: payload}) } + // A pathological orphan directory the delete pass is streaming out in + // batches instead of unlinking depth-first on one agent (agent/src/ + // delete.c, delete_split_threshold) — see ShardSplit.DeleteRemainder's + // doc comment. Same DeleteBatch shape seedDeletePass already uses for + // the top-level batches; job_id/pass_no/task_id are filled at grant time + // from the shard row (buildItem), same as every other split-produced + // kind. + var deleteTotals []store.DeleteGroupTotal + for _, dr := range sp.DeleteRemainders { + // dr.Names are bare basenames (agent readdir entries, delete.c + // flush_delete_split) — DeleteBatch.RelPaths must be full paths + // relative to the destination root (same contract seedDeletePass's + // own DeleteBatch shards use), so each name is joined under dir_rel + // here before the split-produced shard is built. Without this a + // split-produced delete shard tries to unlink a bare basename at the + // destination root, silently misses (ENOENT, not an error), and the + // pathological directory's contents are never actually removed even + // though delete_groups correctly tracks every batch as done. + relPaths := make([][]byte, len(dr.Names)) + for i, name := range dr.Names { + relPaths[i] = []byte(string(dr.DirRel) + "/" + string(name)) + } + payload, err := proto.Marshal(&drsyncpb.DeleteBatch{RelPaths: relPaths}) + if err != nil { + return err + } + shards = append(shards, store.NewShard{ + Kind: model.KindDelete, RelPath: string(dr.DirRel), Payload: payload}) + // One DeleteGroupTotal per batch (not just the final one) — RecordSplit + // 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. + deleteTotals = append(deleteTotals, store.DeleteGroupTotal{ + DirRel: string(dr.DirRel), LastBatch: dr.TotalChildren > 0, + CleanupShard: deleteCleanupShard(string(dr.DirRel)), + }) + } var groups []store.NewChunkGroup if len(sp.BigFiles) > 0 { @@ -562,7 +602,7 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error { return err } - ids, err := s.st.RecordSplit(int64(sp.ParentShardId), sp.Seq, shards, groups, sightings, maxGroupScan) + ids, err := s.st.RecordSplit(int64(sp.ParentShardId), sp.Seq, shards, groups, sightings, maxGroupScan, deleteTotals) if err != nil { return err } @@ -681,15 +721,27 @@ func finalizeShard(rel, temp string, gen *drsyncpb.FileGen) store.NewShard { return store.NewShard{Kind: model.KindChunk, RelPath: rel, Payload: payload} } +// deleteCleanupShard builds the one-entry DeleteBatch that removes a +// pathological orphan directory itself, once every split-produced +// delete-remainder shard has finished emptying it (CompleteDeleteRemainder). +// Runs through the ordinary, unmodified delete path (agent/src/delete.c) — +// by the time this is granted, the directory is empty, so it costs one +// fstatat + one rmdir, same as any single-entry delete batch. +func deleteCleanupShard(dirRel string) store.NewShard { + payload, _ := proto.Marshal(&drsyncpb.DeleteBatch{RelPaths: [][]byte{[]byte(dirRel)}}) + return store.NewShard{Kind: model.KindDelete, Payload: payload} +} + func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error { shardID, leaseID := int64(r.ShardId), int64(r.LeaseId) var err error switch r.Status { case drsyncpb.ResultStatus_RESULT_OK: - // One read tells us pass, kind and (for chunks) the ChunkTask, so the - // chunk group can be maintained in the same completion without a second - // round trip. It replaces the passOfShard lookup the OK path already did. - passID, kind, payload, e := s.st.ShardMeta(shardID) + // One read tells us pass, kind, rel_path and (for chunks) the + // ChunkTask, so the chunk/delete group can be maintained in the same + // completion without a second round trip. It replaces the + // passOfShard lookup the OK path already did. + passID, kind, payload, relPath, e := s.st.ShardMeta(shardID) if e != nil { err = e break @@ -700,9 +752,20 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error { if ms := r.GetCounters().GetWallMs(); ms > 0 { s.met.ShardDuration.WithLabelValues(string(kind)).Observe(float64(ms) / 1000) } - if kind == model.KindChunk { + switch { + case kind == model.KindChunk: err = s.completeChunk(passID, shardID, leaseID, payload, r) - } else { + case kind == model.KindDelete && relPath != "": + // A split-produced delete-remainder shard (onShardSplit sets + // RelPath to the directory it's emptying; a top-level batch from + // 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. + blob, _ := proto.Marshal(r) + err = s.st.CompleteDeleteRemainder(shardID, leaseID, passID, relPath, blob, + deleteCleanupShard(relPath), r.Counters) + default: blob, _ := proto.Marshal(r) err = s.st.CompleteShard(shardID, leaseID, passID, blob, r.Counters) } @@ -710,7 +773,7 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error { // A chunk saw the source drift under it: abort the whole file's group. // The half-written temp is reclaimed as .drsync.tmp residue next walk, // and the file is re-diffed next pass. Only chunks emit this status. - passID, kind, payload, e := s.st.ShardMeta(shardID) + passID, kind, payload, _, e := s.st.ShardMeta(shardID) if e != nil { err = e break diff --git a/coordinator/internal/agentsrv/server_test.go b/coordinator/internal/agentsrv/server_test.go index 035dad9..df99a3a 100644 --- a/coordinator/internal/agentsrv/server_test.go +++ b/coordinator/internal/agentsrv/server_test.go @@ -419,6 +419,99 @@ func TestShardSplitForReapedParentIsAcked(t *testing.T) { a.recv(drsyncpb.FrameType_FRAME_HEARTBEAT_ACK, hbAck) } +// TestDeleteRemainderPathsJoinDirRel is the path-joining regression: a +// ShardSplit.DeleteRemainder's Names are bare basenames (agent readdir +// entries, agent/src/delete.c flush_delete_split) — onShardSplit must join +// each one under DirRel before placing it in the split-produced KindDelete +// shard's DeleteBatch.RelPaths, the same full-relative-path shape +// seedDeletePass's own DeleteBatch shards use (passctrl.go). Feeding a bare +// basename straight into DeleteBatch.RelPaths makes the agent try to unlink +// it at the destination root instead of under the orphan directory — a +// silent ENOENT, not an error, so the shard reports success while removing +// nothing. This was caught by local e2e verification (delete_fanout_e2e.sh), +// not by the delete_groups completion-tracking unit tests, since those drive +// store.RecordSplit directly with pre-built NewShards and never exercise +// onShardSplit's own payload construction. +func TestDeleteRemainderPathsJoinDirRel(t *testing.T) { + srv := newTestServer(t) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go srv.Serve(ln) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + a := &fakeAgent{t: t, conn: conn} + + a.send(drsyncpb.FrameType_FRAME_HELLO, &drsyncpb.Hello{ + AgentId: "agent-test", Hostname: "testhost", ProtoMajor: 1, AgentVersion: "0.0.1"}) + a.recv(drsyncpb.FrameType_FRAME_HELLO_ACK, &drsyncpb.HelloAck{}) + + a.send(drsyncpb.FrameType_FRAME_WORK_REQUEST, &drsyncpb.WorkRequest{ShardCredits: 4}) + grant := &drsyncpb.WorkGrant{} + a.recv(drsyncpb.FrameType_FRAME_WORK_GRANT, grant) + root := grant.Items[0].GetShard() + lease := grant.Items[0].LeaseId + + // A pathological orphan directory streamed out as two DeleteRemainder + // batches — the second (final) one carrying TotalChildren, same shape + // agent/src/delete.c's stream_delete_split produces. + a.send(drsyncpb.FrameType_FRAME_SHARD_SPLIT, &drsyncpb.ShardSplit{ + ParentShardId: root.ShardId, Seq: 1, + DeleteRemainders: []*drsyncpb.ShardSplit_DeleteRemainder{ + {DirRel: []byte("big-orphan-dir"), Names: [][]byte{[]byte("f1.txt"), []byte("f2.txt")}}, + }, + }) + a.recv(drsyncpb.FrameType_FRAME_SHARD_SPLIT_ACK, &drsyncpb.ShardSplitAck{}) + a.send(drsyncpb.FrameType_FRAME_SHARD_SPLIT, &drsyncpb.ShardSplit{ + ParentShardId: root.ShardId, Seq: 2, + DeleteRemainders: []*drsyncpb.ShardSplit_DeleteRemainder{ + {DirRel: []byte("big-orphan-dir"), Names: [][]byte{[]byte("f3.txt")}, TotalChildren: 2}, + }, + }) + a.recv(drsyncpb.FrameType_FRAME_SHARD_SPLIT_ACK, &drsyncpb.ShardSplitAck{}) + + // Finish the root shard so its own credits free up, then pull the + // split-produced delete shards and inspect their payloads directly. + a.send(drsyncpb.FrameType_FRAME_SHARD_RESULT, &drsyncpb.ShardResult{ + ShardId: root.ShardId, LeaseId: lease, Status: drsyncpb.ResultStatus_RESULT_OK, + Counters: &drsyncpb.ShardCounters{}, + }) + a.send(drsyncpb.FrameType_FRAME_WORK_REQUEST, &drsyncpb.WorkRequest{ShardCredits: 4}) + grant2 := &drsyncpb.WorkGrant{} + a.recv(drsyncpb.FrameType_FRAME_WORK_GRANT, grant2) + if len(grant2.Items) != 2 { + t.Fatalf("granted %d delete shards, want 2", len(grant2.Items)) + } + + var gotPaths []string + for _, item := range grant2.Items { + batch := item.GetDelete() + if batch == nil { + t.Fatalf("granted item = %+v, want a delete work item", item) + } + for _, p := range batch.RelPaths { + gotPaths = append(gotPaths, string(p)) + } + } + want := map[string]bool{ + "big-orphan-dir/f1.txt": true, "big-orphan-dir/f2.txt": true, "big-orphan-dir/f3.txt": true, + } + if len(gotPaths) != len(want) { + t.Fatalf("relPaths = %v, want exactly %v", gotPaths, want) + } + for _, p := range gotPaths { + if !want[p] { + t.Fatalf("relPaths contains %q (bare basename, not joined under dir_rel) — full set: %v", p, gotPaths) + } + } +} + // TestChunkTempNamePassTagged is the "open temp for finalize" regression. A // chunk temp lives in the destination directory, with no source counterpart, // for the whole multi-host copy — indistinguishable from crash residue to an diff --git a/coordinator/internal/model/spec.go b/coordinator/internal/model/spec.go index 2c56757..b2e1021 100644 --- a/coordinator/internal/model/spec.go +++ b/coordinator/internal/model/spec.go @@ -130,6 +130,14 @@ type JobSpec struct { EntrylistBatch uint32 `yaml:"entrylist_batch"` StatxBatch uint32 `yaml:"statx_batch"` MtimeSlopNS int64 `yaml:"mtime_slop_ns"` + // Delete-pass fan-out: the analogue of DirSplitThreshold/ + // EntrylistBatch for orphan-directory deletion (docs/DESIGN- + // coordinator.md §2.2 DELETE fan-out). Independently tunable since + // delete work per name (a plain unlink) is much cheaper than a + // full stat/diff/copy pipeline, so the shapes that make sense + // differ. + DeleteSplitThreshold uint64 `yaml:"delete_split_threshold"` + DeleteSplitBatch uint32 `yaml:"delete_split_batch"` // Fan-out control. Coordinator-side only: these never reach an agent // (D9 — the agent acts on the resolved per-shard overrides it is // granted, not on policy). See SpreadPolicy. @@ -251,6 +259,18 @@ func (s *JobSpec) ApplyDefaults() { if sp.Tuning.EntrylistBatch == 0 { sp.Tuning.EntrylistBatch = 4_000 } + // Delete-pass fan-out: same shape as dir_split_threshold/entrylist_batch + // above (threshold decides whether to fan out at all; batch decides how + // many shards result), but delete work per name is a plain unlink, not a + // full stat/diff/copy pipeline, so a directory needs to be substantially + // larger before serializing its removal on one agent is worth avoiding + // the fan-out's own coordinator round-trip overhead. + if sp.Tuning.DeleteSplitThreshold == 0 { + sp.Tuning.DeleteSplitThreshold = 200_000 + } + if sp.Tuning.DeleteSplitBatch == 0 { + sp.Tuning.DeleteSplitBatch = 20_000 + } if sp.Tuning.StatxBatch == 0 { sp.Tuning.StatxBatch = 256 } @@ -429,11 +449,13 @@ func (s *JobSpec) ToJobOptions(jobID uint64, dryRun bool) (*drsyncpb.JobOptions, IopsPerAgent: sp.Limits.IOPSPerAgent, }, Tuning: &drsyncpb.TuningOptions{ - ShardBudget: sp.Tuning.ShardBudget, - DirSplitThreshold: sp.Tuning.DirSplitThreshold, - EntrylistBatch: sp.Tuning.EntrylistBatch, - StatxBatch: sp.Tuning.StatxBatch, - MtimeSlopNs: sp.Tuning.MtimeSlopNS, + ShardBudget: sp.Tuning.ShardBudget, + DirSplitThreshold: sp.Tuning.DirSplitThreshold, + EntrylistBatch: sp.Tuning.EntrylistBatch, + StatxBatch: sp.Tuning.StatxBatch, + MtimeSlopNs: sp.Tuning.MtimeSlopNS, + DeleteSplitThreshold: sp.Tuning.DeleteSplitThreshold, + DeleteSplitBatch: sp.Tuning.DeleteSplitBatch, }, DryRun: dryRun, RequireMount: *sp.Probe.RequireMount, diff --git a/coordinator/internal/model/spec_defaults_test.go b/coordinator/internal/model/spec_defaults_test.go index aec192f..b8bc3ca 100644 --- a/coordinator/internal/model/spec_defaults_test.go +++ b/coordinator/internal/model/spec_defaults_test.go @@ -36,6 +36,12 @@ func TestDefaultsAppliedToMinimalSpec(t *testing.T) { if sp.Tuning.ShardBudget != 2_000 { t.Errorf("tuning.shard_budget = %d, want 2000", sp.Tuning.ShardBudget) } + if sp.Tuning.DeleteSplitThreshold != 200_000 { + t.Errorf("tuning.delete_split_threshold = %d, want 200000", sp.Tuning.DeleteSplitThreshold) + } + if sp.Tuning.DeleteSplitBatch != 20_000 { + t.Errorf("tuning.delete_split_batch = %d, want 20000", sp.Tuning.DeleteSplitBatch) + } if sp.Metadata.Hardlinks != "preserve" { t.Errorf("metadata.hardlinks = %q, want preserve", sp.Metadata.Hardlinks) } @@ -55,6 +61,10 @@ func TestDefaultsAppliedToMinimalSpec(t *testing.T) { t.Errorf("resolved JobOptions chunk sizes = %d/%d, want %d/%d", o.Copy.ChunkThreshold, o.Copy.ChunkSize, twentyFourGiB, eightGiB) } + if o.Tuning.DeleteSplitThreshold != 200_000 || o.Tuning.DeleteSplitBatch != 20_000 { + t.Errorf("resolved JobOptions delete-split tuning = %d/%d, want 200000/20000", + o.Tuning.DeleteSplitThreshold, o.Tuning.DeleteSplitBatch) + } } // TestDirectWriteExplicitFalseIsRespected guards the *bool switch: an diff --git a/coordinator/internal/passctrl/reaper_test.go b/coordinator/internal/passctrl/reaper_test.go index 278ce9b..158aa3d 100644 --- a/coordinator/internal/passctrl/reaper_test.go +++ b/coordinator/internal/passctrl/reaper_test.go @@ -75,7 +75,7 @@ func drainReaps(t *testing.T, c *Controller) { // straight from the shards table, so sql.ErrNoRows is the reap signal). func shardGone(t *testing.T, c *Controller, id int64) bool { t.Helper() - _, _, _, err := c.st.ShardMeta(id) + _, _, _, _, err := c.st.ShardMeta(id) if errors.Is(err, sql.ErrNoRows) { return true } diff --git a/coordinator/internal/passctrl/seedlinkfix_test.go b/coordinator/internal/passctrl/seedlinkfix_test.go index 9e91daa..9ff065f 100644 --- a/coordinator/internal/passctrl/seedlinkfix_test.go +++ b/coordinator/internal/passctrl/seedlinkfix_test.go @@ -32,17 +32,17 @@ func TestSeedLinkfixBuildsOneBatchTaskForPendingMembers(t *testing.T) { // that's anchor-only (no member needs one). if _, err := c.st.RecordSplit(shardID, 1, nil, nil, []store.NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } if _, err := c.st.RecordSplit(shardID, 2, nil, nil, []store.NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "b/two", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } if _, err := c.st.RecordSplit(shardID, 3, nil, nil, []store.NewLinkSighting{ {Dev: 1, Ino: 200, RelPath: "c/anchor-only", Nlink: 1, Size: 5, MtimeNs: 2}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } @@ -114,7 +114,7 @@ func TestSeedLinkfixFlushesAtBatchBoundary(t *testing.T) { Nlink: 2, Size: 10, MtimeNs: 1, } if _, err := c.st.RecordSplit(shardID, seq, nil, nil, - []store.NewLinkSighting{anchor}, 0); err != nil { + []store.NewLinkSighting{anchor}, 0, nil); err != nil { t.Fatal(err) } seq++ @@ -123,7 +123,7 @@ func TestSeedLinkfixFlushesAtBatchBoundary(t *testing.T) { Nlink: 2, Size: 10, MtimeNs: 1, } if _, err := c.st.RecordSplit(shardID, seq, nil, nil, - []store.NewLinkSighting{member}, 0); err != nil { + []store.NewLinkSighting{member}, 0, nil); err != nil { t.Fatal(err) } } @@ -228,12 +228,12 @@ func TestAdvanceReapsLinkRegistryOnLinkfixTransition(t *testing.T) { // either way would not distinguish "reaped" from "never had anything"). if _, err := c.st.RecordSplit(shardID, 1, nil, nil, []store.NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/anchor", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } if _, err := c.st.RecordSplit(shardID, 2, nil, nil, []store.NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/member", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } diff --git a/coordinator/internal/passctrl/seedtempreclaim_test.go b/coordinator/internal/passctrl/seedtempreclaim_test.go index d9b61a2..daa36de 100644 --- a/coordinator/internal/passctrl/seedtempreclaim_test.go +++ b/coordinator/internal/passctrl/seedtempreclaim_test.go @@ -47,7 +47,7 @@ func TestSeedTempReclaimOnlyUnfinalized(t *testing.T) { []store.NewChunkGroup{ {RelPath: "a/done.bin", TempName: doneTemp, Size: 8, MtimeNs: 1, NChunks: 1}, {RelPath: "b/dead.bin", TempName: deadTemp, Size: 8, MtimeNs: 2, NChunks: 1}, - }, nil, 0); err != nil { + }, nil, 0, nil); err != nil { t.Fatal(err) } diff --git a/coordinator/internal/store/deletefanout_test.go b/coordinator/internal/store/deletefanout_test.go new file mode 100644 index 0000000..ad985f2 --- /dev/null +++ b/coordinator/internal/store/deletefanout_test.go @@ -0,0 +1,249 @@ +package store + +import ( + "testing" + "time" + + "drsync/coordinator/internal/model" +) + +// deleteRemainderShard is a NewShard shaped like onShardSplit builds for a +// ShardSplit.DeleteRemainder — Kind KindDelete, RelPath set to the directory +// being emptied. This is the signal CompleteDeleteRemainder's caller +// (agentsrv.onShardResult) uses to tell a split-produced delete-remainder +// shard apart from an ordinary top-level one (RelPath empty). +func deleteRemainderShard(dirRel string) NewShard { + return NewShard{Kind: model.KindDelete, RelPath: dirRel} +} + +func cleanupShard(dirRel string) NewShard { + return NewShard{Kind: model.KindDelete} +} + +// TestDeleteGroupSeedsCleanupOnceAllChildrenDone is the ordinary case: a +// directory splits into 3 delete-remainder shards in one ShardSplit (the +// final one carrying LastBatch), all 3 complete, and the cleanup shard for +// the directory itself is seeded exactly once, only after the last one. +func TestDeleteGroupSeedsCleanupOnceAllChildrenDone(t *testing.T) { + s := openTest(t) + _, passID, shardID := seed(t, s) + if _, err := s.LeaseShards("agent-a", 1, time.Minute); err != nil { + t.Fatal(err) + } + + const dir = "big-orphan-dir" + remainders := []NewShard{ + deleteRemainderShard(dir), deleteRemainderShard(dir), deleteRemainderShard(dir), + } + // One DeleteGroupTotal per batch shard (RecordSplit bumps n_total by one + // per entry, same units as n_done) — only the last carries LastBatch. + ids, err := s.RecordSplit(shardID, 1, remainders, nil, nil, 0, []DeleteGroupTotal{ + {DirRel: dir, CleanupShard: cleanupShard(dir)}, + {DirRel: dir, CleanupShard: cleanupShard(dir)}, + {DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}, + }) + if err != nil { + t.Fatal(err) + } + if len(ids) != 3 { + t.Fatalf("RecordSplit produced %d shard ids, want 3", len(ids)) + } + + countsAfterSplit, err := s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + // Only the 3 remainder shards queued yet — no cleanup shard seeded before + // any child has completed. + if countsAfterSplit[model.ShardQueued] != 3 { + t.Fatalf("queued after split = %d, want 3 (no cleanup shard yet)", countsAfterSplit[model.ShardQueued]) + } + + leased, err := s.LeaseShards("agent-a", 3, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased) != 3 { + t.Fatalf("leased %d of 3 remainder shards", len(leased)) + } + + for i, r := range leased { + if err := s.CompleteDeleteRemainder(r.ID, r.LeaseID, passID, dir, nil, + cleanupShard(dir), nil); err != nil { + t.Fatalf("complete remainder %d: %v", i, err) + } + counts, err := s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if i < 2 { + if counts[model.ShardQueued] != 0 { + t.Fatalf("after %d/3 children done: queued = %d, want 0 (cleanup not seeded yet)", + i+1, counts[model.ShardQueued]) + } + } else { + // The 3rd (last) completion must have seeded exactly one new + // queued shard: the cleanup for dir itself. + if counts[model.ShardQueued] != 1 { + t.Fatalf("after all 3 children done: queued = %d, want 1 (the cleanup shard)", + counts[model.ShardQueued]) + } + } + } +} + +// TestDeleteGroupHandlesChildCompletionRacingFinalBatch: a child shard's +// completion is an independent frame from the streaming parent's own +// ShardSplit batches — only the parent's own ShardResult is ordered after +// every split it shipped (protocol §4.2), not the split's own processing +// relative to its children. This drives that race directly: two batches are +// recorded (n_total reaches 2, only the second carrying LastBatch), the +// FIRST batch's shard completes before the SECOND (final) batch is ever +// recorded — CompleteDeleteRemainder's own check finds streaming not yet +// done and doesn't close the group — then the final batch lands and +// RecordSplit notices n_done already reached n_total and seeds the cleanup +// shard itself. +func TestDeleteGroupHandlesChildCompletionRacingFinalBatch(t *testing.T) { + s := openTest(t) + _, passID, shardID := seed(t, s) + if _, err := s.LeaseShards("agent-a", 1, time.Minute); err != nil { + t.Fatal(err) + } + + const dir = "racing-dir" + // First ShardSplit: one remainder batch, NOT the final one — every batch + // (final or not) ships a DeleteRemainder and therefore a DeleteGroupTotal + // entry (server.go onShardSplit), so n_total is bumped to 1 here; only + // LastBatch (from total_children>0 on the wire) is still unset. + ids, err := s.RecordSplit(shardID, 1, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, + []DeleteGroupTotal{{DirRel: dir, CleanupShard: cleanupShard(dir)}}) + if err != nil { + t.Fatal(err) + } + if len(ids) != 1 { + t.Fatalf("RecordSplit produced %d ids, want 1", len(ids)) + } + + leased, err := s.LeaseShards("agent-a", 1, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased) != 1 { + t.Fatalf("leased %d shards, want 1", len(leased)) + } + // This first child completes while n_total is still 1 (streaming not + // finished) — reaches n_done >= n_total by coincidence, but done_streaming + // is still 0, so the group must NOT close yet. + if err := s.CompleteDeleteRemainder(leased[0].ID, leased[0].LeaseID, passID, dir, nil, + cleanupShard(dir), nil); err != nil { + t.Fatal(err) + } + counts, err := s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardQueued] != 0 { + t.Fatalf("queued after the first child completes with streaming unfinished = %d, want 0", + counts[model.ShardQueued]) + } + + // Now the streaming parent's final batch lands (a second ShardSplit, + // different seq — the parent is still mid-stream, this is its EOF batch, + // carrying its own remainder shard plus LastBatch): n_total becomes 2, + // but only 1 child has completed so far, so RecordSplit itself must not + // close the group on this call either. + ids2, err := s.RecordSplit(shardID, 2, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}) + if err != nil { + t.Fatal(err) + } + if len(ids2) != 1 { + t.Fatalf("RecordSplit produced %d ids, want 1", len(ids2)) + } + counts, err = s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued after the final batch lands with 1/2 done = %d, want 1 (only the new remainder, no cleanup yet)", + counts[model.ShardQueued]) + } + + // The second (final) batch's own shard now completes — n_done reaches + // n_total (2) AND done_streaming is set, so THIS completion must seed the + // cleanup shard. + leased2, err := s.LeaseShards("agent-a", 1, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased2) != 1 || leased2[0].ID != ids2[0] { + t.Fatalf("lease mismatch: leased=%v ids2=%v", leased2, ids2) + } + if err := s.CompleteDeleteRemainder(leased2[0].ID, leased2[0].LeaseID, passID, dir, nil, + cleanupShard(dir), nil); err != nil { + t.Fatal(err) + } + counts, err = s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued after both children done and streaming finished = %d, want 1 (the cleanup shard)", + counts[model.ShardQueued]) + } +} + +// TestDeleteGroupNeverSeedsCleanupTwice: the closed flag must survive both +// write paths (CompleteDeleteRemainder and RecordSplit) each independently +// re-checking a group they didn't close — otherwise a retransmit or a +// re-delivered result could seed the cleanup shard more than once, which +// would attempt to remove an already-gone directory twice (harmless per se, +// since delete is ENOENT-tolerant, but wasteful and would double-count). +func TestDeleteGroupNeverSeedsCleanupTwice(t *testing.T) { + s := openTest(t) + _, passID, shardID := seed(t, s) + if _, err := s.LeaseShards("agent-a", 1, time.Minute); err != nil { + t.Fatal(err) + } + + const dir = "dup-dir" + ids, err := s.RecordSplit(shardID, 1, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}) + if err != nil { + t.Fatal(err) + } + leased, err := s.LeaseShards("agent-a", 1, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased) != 1 || leased[0].ID != ids[0] { + t.Fatalf("lease mismatch: leased=%v ids=%v", leased, ids) + } + if err := s.CompleteDeleteRemainder(leased[0].ID, leased[0].LeaseID, passID, dir, nil, + cleanupShard(dir), nil); err != nil { + t.Fatal(err) + } + counts, err := s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued after the only child completes = %d, want 1 (cleanup seeded)", counts[model.ShardQueued]) + } + + // A retransmitted final batch for the same directory (agent outbox + // replay racing a reconnect, same shape RecordSplit already handles for + // every other split kind) must not seed a second cleanup shard. + if _, err := s.RecordSplit(shardID, 2, nil, nil, nil, 0, + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}); err != nil { + t.Fatal(err) + } + counts, err = s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued after a retransmitted final batch = %d, want still 1 (no duplicate cleanup shard)", + counts[model.ShardQueued]) + } +} diff --git a/coordinator/internal/store/link_test.go b/coordinator/internal/store/link_test.go index d2533f4..3fb90ea 100644 --- a/coordinator/internal/store/link_test.go +++ b/coordinator/internal/store/link_test.go @@ -20,7 +20,7 @@ func TestRecordLinkSightingsFirstIsAnchor(t *testing.T) { sightings := []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0); err != nil { + if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0, nil); err != nil { t.Fatal(err) } @@ -49,20 +49,20 @@ func TestRecordLinkSightingsHighBitInode(t *testing.T) { first := []NewLinkSighting{ {Dev: 1, Ino: hugeIno, RelPath: "a/one", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 1, nil, nil, first, 0); err != nil { + if _, err := s.RecordSplit(shardID, 1, nil, nil, first, 0, nil); err != nil { t.Fatalf("RecordSplit with a high-bit inode: %v", err) } second := []NewLinkSighting{ {Dev: 1, Ino: hugeIno, RelPath: "b/two", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 2, nil, nil, second, 0); err != nil { + if _, err := s.RecordSplit(shardID, 2, nil, nil, second, 0, nil); err != nil { t.Fatalf("RecordSplit (second sighting) with a high-bit inode: %v", err) } // A second, independent group at the absolute max value. maxGroup := []NewLinkSighting{ {Dev: maxIno, Ino: maxIno, RelPath: "c/max", Nlink: 2, Size: 8192, MtimeNs: 222}, } - if _, err := s.RecordSplit(shardID, 3, nil, nil, maxGroup, 0); err != nil { + if _, err := s.RecordSplit(shardID, 3, nil, nil, maxGroup, 0, nil); err != nil { t.Fatalf("RecordSplit with dev=ino=MaxUint64: %v", err) } @@ -115,13 +115,13 @@ func TestRecordLinkSightingsSecondIsPendingMember(t *testing.T) { first := []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 1, nil, nil, first, 0); err != nil { + if _, err := s.RecordSplit(shardID, 1, nil, nil, first, 0, nil); err != nil { t.Fatal(err) } second := []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "b/two", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 2, nil, nil, second, 0); err != nil { + if _, err := s.RecordSplit(shardID, 2, nil, nil, second, 0, nil); err != nil { t.Fatal(err) } @@ -151,12 +151,12 @@ func TestRecordLinkSightingsIdempotent(t *testing.T) { sightings := []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 2, Size: 4096, MtimeNs: 111}, } - if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0); err != nil { + if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0, nil); err != nil { t.Fatal(err) } // Same (parent, seq): RecordSplit returns the cached result without // re-running recordLinkSightingsTx. - if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0); err != nil { + if _, err := s.RecordSplit(shardID, 1, nil, nil, sightings, 0, nil); err != nil { t.Fatal(err) } @@ -180,25 +180,25 @@ func TestRecordLinkSightingsMaxGroupScanFallback(t *testing.T) { if _, err := s.RecordSplit(shardID, 1, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 2); err != nil { // maxGroupScan=2 + }, 2, nil); err != nil { // maxGroupScan=2 t.Fatal(err) } if _, err := s.RecordSplit(shardID, 2, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "b/two", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 2); err != nil { + }, 2, nil); err != nil { t.Fatal(err) } // Third sighting pushes members_seen to 3, over the cap of 2. if _, err := s.RecordSplit(shardID, 3, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "c/three", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 2); err != nil { + }, 2, nil); err != nil { t.Fatal(err) } // Fourth sighting of the SAME already-fallen-back group: must not // increment link_fallback again (it counts groups, not sightings). if _, err := s.RecordSplit(shardID, 4, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "d/four", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 2); err != nil { + }, 2, nil); err != nil { t.Fatal(err) } @@ -238,12 +238,12 @@ func TestMarkLinkMembersQueuedIsScopedToPending(t *testing.T) { if _, err := s.RecordSplit(shardID, 1, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/one", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } if _, err := s.RecordSplit(shardID, 2, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "b/two", Nlink: 2, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } @@ -401,13 +401,13 @@ func TestReapLinkRegistryDeletesBothTables(t *testing.T) { // One group of 3 (anchor + 2 pending members). if _, err := s.RecordSplit(shardID, 1, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/anchor", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } if _, err := s.RecordSplit(shardID, 2, nil, nil, []NewLinkSighting{ {Dev: 1, Ino: 100, RelPath: "a/member1", Nlink: 3, Size: 10, MtimeNs: 1}, {Dev: 1, Ino: 100, RelPath: "a/member2", Nlink: 3, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } @@ -425,7 +425,7 @@ func TestReapLinkRegistryDeletesBothTables(t *testing.T) { } if _, err := s.RecordSplit(otherShardIDs[0], 1, nil, nil, []NewLinkSighting{ {Dev: 2, Ino: 200, RelPath: "b/anchor", Nlink: 1, Size: 10, MtimeNs: 1}, - }, 0); err != nil { + }, 0, nil); err != nil { t.Fatal(err) } diff --git a/coordinator/internal/store/store.go b/coordinator/internal/store/store.go index db2acd7..8679849 100644 --- a/coordinator/internal/store/store.go +++ b/coordinator/internal/store/store.go @@ -283,6 +283,33 @@ CREATE TABLE IF NOT EXISTS chunk_groups ( PRIMARY KEY (pass_id, rel_path) ) WITHOUT ROWID; +-- One row per pathological orphan directory being fanned out across DELETE +-- shards (docs/DESIGN-coordinator.md §2.2 DELETE fan-out) — the delete-pass +-- analogue of chunk_groups above. n_total counts split-produced DELETE +-- *shards* (batches) for dir_rel, one per ShardSplit.DeleteRemainder +-- received — NOT total_children, which is the directory's *entry* count and +-- is unrelated in scale (a 300-entry directory at batch size 40 is 8 +-- shards); comparing n_done (also counted in shards) against an entry count +-- would just never converge except by coincidence. total_children only +-- serves as the "readdir hit EOF, this was the last batch" signal, recorded +-- in done_streaming. A child shard's completion can race ahead of that final +-- batch — the two are independent shards/frames, only the *streaming +-- parent's own* ShardResult is ordered after every split it shipped +-- (protocol §4.2) — so n_done is incremented unconditionally on every child +-- completion, and the "seed a cleanup shard for dir_rel" decision is checked +-- at BOTH the write that sets done_streaming and the write that increments +-- n_done, whichever lands last: closeable is (done_streaming=1 AND +-- n_done >= n_total). +CREATE TABLE IF NOT EXISTS delete_groups ( + pass_id INTEGER NOT NULL, + rel_path TEXT NOT NULL, -- the directory being emptied (dir_rel) + n_total INTEGER NOT NULL DEFAULT 0, -- split-produced DELETE shards seen so far + n_done INTEGER NOT NULL DEFAULT 0, -- of those, how many have completed + done_streaming INTEGER NOT NULL DEFAULT 0, -- 1 once the final batch (EOF) has landed + closed INTEGER NOT NULL DEFAULT 0, -- 1 once the cleanup shard has been seeded (idempotency) + PRIMARY KEY (pass_id, rel_path) +) WITHOUT ROWID; + -- link_groups/link_members correlate hardlink-group members across shards -- (docs/DESIGN-hardlinks.md). A group is identified by (dev,ino), which is -- unique only within one pass's walk — this is scratch state, reaped with the @@ -1173,14 +1200,34 @@ func recordLinkSightingsTx(tx *sql.Tx, passID int64, sightings []NewLinkSighting return nil } +// DeleteGroupTotal records one delete-remainder batch shipped for DirRel — +// RecordSplit bumps that group's n_total by one per entry, regardless of +// LastBatch (see delete_groups' doc comment for why n_total counts shards, +// not ShardSplit.DeleteRemainder.total_children's entry count). LastBatch is +// set from total_children > 0 — the "readdir hit EOF for dirRel" signal, only +// known on the final streamed batch, not upfront like a chunk group's +// byte-size-derived n_chunks. CleanupShard is pre-built by the caller (store +// stays payload-agnostic, same division of labor as CompleteDataChunk's +// finalizeShard) and only actually inserted if every sibling delete-remainder +// shard has already reported done by the time LastBatch lands — the same +// race CompleteDeleteRemainder resolves from the other direction; it is only +// read when LastBatch is true. +type DeleteGroupTotal struct { + DirRel string + LastBatch bool + CleanupShard NewShard +} + // RecordSplit persists a ShardSplit idempotently: retransmits of the same // (parent, seq) return the originally assigned ids (protocol doc §4.3). groups -// (for big files whose data-chunk shards are among shards) and sightings -// (nlink>1 files reported this split) are recorded in the same transaction; -// pass nil for either when there are none. maxGroupScan is the job's +// (for big files whose data-chunk shards are among shards), sightings +// (nlink>1 files reported this split), and deleteTotals (final delete-remainder +// batches — see DeleteGroupTotal) are recorded in the same transaction; pass +// nil for any that don't apply. maxGroupScan is the job's // hardlinks_max_group_scan (0 = unlimited). func (s *Store) RecordSplit(parentShardID int64, seq uint64, shards []NewShard, - groups []NewChunkGroup, sightings []NewLinkSighting, maxGroupScan uint64) ([]int64, error) { + groups []NewChunkGroup, sightings []NewLinkSighting, maxGroupScan uint64, + deleteTotals []DeleteGroupTotal) ([]int64, error) { // int64(seq): bit-preserving reinterpret, not a truncation — see // recordLinkSightingsTx's comment. seq is agent-assigned per-parent and in // practice never near 2^63, but binding a uint64 with the high bit set @@ -1267,6 +1314,45 @@ func (s *Store) RecordSplit(parentShardID int64, seq uint64, shards []NewShard, if err := recordLinkSightingsTx(tx, passID, sightings, maxGroupScan); err != nil { return nil, err } + for _, dt := range deleteTotals { + // INSERT OR IGNORE: a child's own completion (CompleteDeleteRemainder) + // may have already created this row via its own INSERT OR IGNORE if + // it raced ahead of this, the streaming parent's final batch — the + // two are independent shards/frames, only the parent's own + // ShardResult is ordered after every split it shipped (protocol + // §4.2), not the split's own processing relative to its children's. + if _, err := tx.Exec(`INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, + passID, dt.DirRel); err != nil { + return nil, err + } + // n_total counts shards (one per batch, this one included), never + // total_children — see delete_groups' doc comment. + var nTotal, nDone, doneStreaming, closed int + if dt.LastBatch { + err = tx.QueryRow(`UPDATE delete_groups SET n_total = n_total + 1, done_streaming = 1 + WHERE pass_id = ? AND rel_path = ? + RETURNING n_total, n_done, done_streaming, closed`, + passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &closed) + } else { + err = tx.QueryRow(`UPDATE delete_groups SET n_total = n_total + 1 + WHERE pass_id = ? AND rel_path = ? + RETURNING n_total, n_done, done_streaming, closed`, + passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &closed) + } + if err != nil { + return nil, err + } + if closed != 0 || doneStreaming == 0 || nDone < nTotal { + continue // streaming not finished, not every sibling has reported done yet, or already closed + } + if _, err := insertShardsTx(tx, passID, 0, []NewShard{dt.CleanupShard}); err != nil { + return nil, err + } + if _, err := tx.Exec(`UPDATE delete_groups SET closed = 1 WHERE pass_id = ? AND rel_path = ?`, + passID, dt.DirRel); err != nil { + return nil, err + } + } blob, _ := json.Marshal(ids) if _, err := tx.Exec(`INSERT INTO splits (parent_shard_id, seq, assigned_ids) VALUES (?,?,?)`, parentShardID, seq64, string(blob)); err != nil { @@ -1419,11 +1505,11 @@ func (s *Store) MarkLinkMembersQueued(passID int64, members []LinkMemberKey) err // ShardMeta returns a leased shard's pass, kind and inner payload — enough for // the result handler to route a completion without a second round trip. Read // before the shard transitions, so a chunk's ChunkTask payload is still there. -func (s *Store) ShardMeta(shardID int64) (passID int64, kind model.ShardKind, payload []byte, err error) { +func (s *Store) ShardMeta(shardID int64) (passID int64, kind model.ShardKind, payload []byte, relPath string, err error) { var k string - err = s.rdb.QueryRow(`SELECT pass_id, kind, payload FROM shards WHERE id = ?`, - shardID).Scan(&passID, &k, &payload) - return passID, model.ShardKind(k), payload, err + err = s.rdb.QueryRow(`SELECT pass_id, kind, payload, rel_path FROM shards WHERE id = ?`, + shardID).Scan(&passID, &k, &payload, &relPath) + return passID, model.ShardKind(k), payload, relPath, err } // CompleteDataChunk marks a data-chunk shard DONE and bumps its group's n_done. @@ -1472,6 +1558,68 @@ func (s *Store) CompleteDataChunk(shardID, leaseID, passID int64, relPath string return true, tx.Commit() } +// CompleteDeleteRemainder marks a split-produced delete-remainder shard DONE +// and bumps its delete_groups row's n_done — the delete-pass analogue of +// CompleteDataChunk. relPath is the directory being emptied (dir_rel), not +// this shard's own identity; see agentsrv.onShardResult for how a +// split-produced KindDelete shard is told apart from an ordinary top-level +// one. When n_done reaches n_total AND done_streaming is set (the streaming +// parent's final DeleteRemainder batch has landed — see RecordSplit) this +// inserts cleanupShard (built by the caller: a single-entry DeleteBatch for +// relPath itself, run through the ordinary unmodified delete path — store +// stays payload-agnostic, same division of labor as CompleteDataChunk's +// finalizeShard) and marks the group closed so a re-delivered final result +// can never seed it twice. counters folds the pass-counter accumulation into +// this same transaction — see CompleteShard's doc comment for why. +func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath string, result []byte, cleanupShard NewShard, counters *drsyncpb.ShardCounters) error { + defer s.lockTimed("CompleteDeleteRemainder")() + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + n, err := execCountTx(tx, `UPDATE shards SET state = ?, result = COALESCE(?, result), updated_at = ? + WHERE id = ? AND state = ? AND lease_id = ?`, + string(model.ShardDone), result, nowMS(), shardID, string(model.ShardLeased), leaseID) + if err != nil { + return err + } + if n == 0 { + return ErrLeaseMismatch // stale/duplicate result; drop it + } + if err := accumulatePassCountersTx(tx, passID, counters); err != nil { + return err + } + + // The row may not exist yet if this child's completion raced ahead of + // RecordSplit's own insert for the batch that produced it — vanishingly + // unlikely (the grant that produced this ShardResult already required + // the row's shard to exist) but INSERT OR IGNORE makes the ordering + // irrelevant either way, same defensive shape as chunk_groups. + if _, err := tx.Exec(`INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, + passID, relPath); err != nil { + return err + } + var nTotal, nDone, doneStreaming, closed int + if err := tx.QueryRow(`UPDATE delete_groups SET n_done = n_done + 1 + WHERE pass_id = ? AND rel_path = ? RETURNING n_total, n_done, done_streaming, closed`, + passID, relPath).Scan(&nTotal, &nDone, &doneStreaming, &closed); err != nil { + return err + } + if closed != 0 || doneStreaming == 0 || nDone < nTotal { + return tx.Commit() // already closed, streaming not finished yet, or siblings outstanding + } + if _, err := insertShardsTx(tx, passID, 0, []NewShard{cleanupShard}); err != nil { + return err + } + if _, err := tx.Exec(`UPDATE delete_groups SET closed = 1 WHERE pass_id = ? AND rel_path = ?`, + passID, relPath); err != nil { + return err + } + return tx.Commit() +} + // CompleteFinalizeChunk marks the finalize shard DONE and closes its group. // counters folds the pass-counter accumulation into this same transaction — // see CompleteShard's doc comment for why. diff --git a/coordinator/internal/store/store_test.go b/coordinator/internal/store/store_test.go index 9881de4..fc6646f 100644 --- a/coordinator/internal/store/store_test.go +++ b/coordinator/internal/store/store_test.go @@ -132,7 +132,7 @@ func TestShardCountsRollupConsistent(t *testing.T) { assertCountsConsistent(t, s, "after lease (1 LEASED)") if _, err := s.RecordSplit(shardID, 1, []NewShard{ {Kind: model.KindDir, RelPath: "a"}, {Kind: model.KindEntryList, RelPath: "b"}, - }, nil, nil, 0); err != nil { + }, nil, nil, 0, nil); err != nil { t.Fatal(err) } assertCountsConsistent(t, s, "after split") @@ -379,7 +379,7 @@ func TestReapDoneShardsDeletesSplits(t *testing.T) { } childIDs, err := s.RecordSplit(shardID, 1, []NewShard{ {Kind: model.KindEntryList, RelPath: "a"}, {Kind: model.KindEntryList, RelPath: "b"}, - }, nil, nil, 0) + }, nil, nil, 0, nil) if err != nil { t.Fatal(err) } @@ -410,7 +410,7 @@ func TestReapDoneShardsDeletesSplits(t *testing.T) { } if _, err := s.RecordSplit(otherParent[0], 1, []NewShard{ {Kind: model.KindEntryList, RelPath: "c"}, - }, nil, nil, 0); err != nil { + }, nil, nil, 0, nil); err != nil { t.Fatal(err) } @@ -1444,11 +1444,11 @@ func TestSplitIdempotency(t *testing.T) { {Kind: model.KindDir, RelPath: "a"}, {Kind: model.KindDir, RelPath: "b"}, } - ids1, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0) + ids1, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0, nil) if err != nil { t.Fatal(err) } - ids2, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0) // retransmit + ids2, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0, nil) // retransmit if err != nil { t.Fatal(err) } @@ -1485,7 +1485,7 @@ func TestRecordSplitPreChecksDoNotBlockOnWriteConnection(t *testing.T) { t.Fatal(err) } subs := []NewShard{{Kind: model.KindDir, RelPath: "a"}} - if _, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0); err != nil { + if _, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0, nil); err != nil { t.Fatal(err) } @@ -1504,7 +1504,7 @@ func TestRecordSplitPreChecksDoNotBlockOnWriteConnection(t *testing.T) { go func() { // Retransmit of the split recorded above: hits the idempotency-check // fast path and must return without ever touching s.db. - _, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0) + _, err := s.RecordSplit(shardID, 7, subs, nil, nil, 0, nil) done <- err }() @@ -1598,7 +1598,7 @@ func TestRecordSplitMissingParentIsIdempotent(t *testing.T) { } subs := []NewShard{{Kind: model.KindDir, RelPath: "a"}} - ids, err := s.RecordSplit(shardID, 9, subs, nil, nil, 0) + ids, err := s.RecordSplit(shardID, 9, subs, nil, nil, 0, nil) if err != nil { t.Fatalf("RecordSplit on reaped parent returned an error, want nil: %v", err) } diff --git a/docs/ADMIN.md b/docs/ADMIN.md index a3afc20..4171ee3 100644 --- a/docs/ADMIN.md +++ b/docs/ADMIN.md @@ -484,6 +484,28 @@ journaled as a single `orphan` record but each removed entry is journaled as a dry-run first (`drsync journal cat myjob --type would_delete`), since dropping one orphaned directory can remove a large tree. +**A very large orphan directory fans out across the fleet, like a huge source +directory does during scanning.** A directory whose own entry count exceeds +`tuning.delete_split_threshold` (default 200 000) is streamed out as batches of +names — `tuning.delete_split_batch` (default 20 000) per batch — instead of +being removed depth-first by whichever one agent found it. Each batch runs as +its own DELETE shard, so the fleet unlinks the directory's contents in +parallel; once every batch has finished, the coordinator seeds one more shard +that removes the now-empty directory itself. Tune it the same way as +`dir_split_threshold`/`entrylist_batch` (§4.5): the threshold decides whether +to fan out at all, the batch size decides how many shards the directory +becomes. + +```bash +drsync job submit huge-orphans.yaml --start \ + --set spec.tuning.delete_split_batch=50000 +``` + +Without this, one pathologically large orphaned tree (a stale multi-million- +file subtree that no longer exists on the source, common right after a large +reorganization) can make the delete pass take as long as the rest of the job +combined, with the whole fleet idle except the one agent working through it. + --- ## 6. Monitoring diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index 134cc47..8d33ed9 100644 --- a/docs/DESIGN-coordinator.md +++ b/docs/DESIGN-coordinator.md @@ -118,6 +118,81 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards filesystem object under each orphan path — so it is expected and correct for the delete pass's live removal count to run well past the prior pass's reported orphan count while still in progress, not a sign of double-counting or runaway deletion. +- **DELETE fan-out.** `rm_tree`, unlike the scan walker, had no split mechanism at + all until this was added: a single huge orphan directory was always removed + depth-first by whichever one agent's shard named it — the delete-pass analogue of + the entry-list problem (a directory too large for one shard), but with no + equivalent fix. On a tree where the orphan set is dominated by a handful of large + stale subtrees (common right after a reorg — a source directory that no longer + exists at all), this made the delete pass take as long as the rest of the job + combined, fleet-wide parallelism notwithstanding, since only one agent thread was + ever doing the actual unlinking. + + The fix mirrors `split_entrylist_stream` (§4.1's entry-list mechanism) more + closely than the plain directory-split (`queue_split`) mechanism, because delete + has nothing to diff against a destination — the whole subtree is already + condemned (D5) — so a batch is just names to remove, not a source/destination + merge. `remove_orphan` (`agent/src/delete.c`) probe-reads a top-level orphan + directory (only the path named directly in the shard's `paths[]`, not every + nested directory reached during recursion — same scope as the entry-list + trigger) up to `tuning.delete_split_threshold` entries; over that, the directory + is streamed via `stream_delete_split` in batches of `tuning.delete_split_batch` + names as `ShardSplit.DeleteRemainder` splits (wire field 7 — additive to the + existing `subdirs`/`entry_lists`/`big_files`/`link_sightings` repeated fields, + no new frame type needed) instead of being unlinked inline. `ship_split`/ + `drain_splits` — the ack/backpressure machinery every split kind already shared + — moved out of `walker.c` into a new `split.c` so `delete.c` can use them too + without duplicating the ordering-invariant-critical code. + + Unlike entry-list sharding, delete fan-out has a real completion problem + entry-list never does: something has to remove the now-empty directory itself + once every split-produced child has finished emptying it, and the coordinator + has no way to know that in advance — the agent streams via `readdir` and only + learns EOF once it hits it (the last `DeleteRemainder` batch for a directory + carries `total_children`, its own entry count; every earlier batch carries 0, + meaning "not yet the last one"). A new `delete_groups` table (§3), the + delete-pass analogue of `chunk_groups`, tracks `n_total`/`n_done`/ + `done_streaming`/`closed` per directory. Because a split-produced child shard's + own completion is an independent frame from the streaming parent's batches — only + the parent's *own* `ShardResult` is ordered after every split it shipped + (protocol §4.2), not the split's own processing relative to its children's — a + child can complete before the coordinator has even recorded the group's true + total. `store.CompleteDeleteRemainder` (child completion) and `store.RecordSplit` + (each batch, not just the final one) each independently check "is this group now + closeable" — whichever one lands last seeds the one-entry cleanup shard that + removes the directory, and `closed` stops the other from ever seeding a second + one. + + **Found in local verification, before this ever reached CI:** two bugs, caught + by `delete_fanout_e2e.sh` (a unit-test-only check would have missed both — see + below). + 1. `n_total` was first wired directly to `total_children` — the directory's + *entry* count (e.g. 300) — while `n_done` counts completed *shards* + (batches, e.g. 8 at batch size 40). Comparing the two never converges + except by coincidence, so the group sat open forever and the DELETE phase + silently "completed" (shard counts drained to zero, nothing was left + `QUEUED`/`LEASED`, so `passctrl.advance()` saw no reason to block) with the + orphan directory's cleanup shard never seeded. Fixed by making `n_total` + count shards (one `RecordSplit` bump per `DeleteRemainder` received, + final batch included) and moving the "streaming finished" signal to its + own `done_streaming` column, set only when `total_children > 0` lands — + closeable is now `done_streaming=1 AND n_done>=n_total`, both sides in the + same unit. + 2. Once that closed correctly, the directory's contents were *still* not + removed: `stream_delete_split` (`agent/src/delete.c`) ships bare + `readdir` basenames (`f0001.txt`), but `onShardSplit` + (`coordinator/internal/agentsrv/server.go`) was passing them straight into + `DeleteBatch.RelPaths` — the same field `seedDeletePass` fills with full + paths relative to the destination root. A split-produced delete shard + then tried to unlink e.g. `f0001.txt` at the destination root, missed + (ENOENT, silently treated as "already gone"), and reported success having + removed nothing. Fixed by joining each name under `dir_rel` in + `onShardSplit` before building the batch. Neither `deletefanout_test.go` + (drives `store.RecordSplit`/`CompleteDeleteRemainder` directly with + pre-built shards, never touching `onShardSplit`'s own payload + construction) nor a code read caught this — it only showed up once a real + directory failed to disappear from disk. `TestDeleteRemainderPathsJoinDirRel` + (`agentsrv/server_test.go`) now pins the join at the `onShardSplit` level. ### 2.3 Shard @@ -153,6 +228,15 @@ agents (id, hostname, state, version, caps BLOB, last_heartbeat, chunk_groups (pass_id, rel_path, temp_name, size, mtime_ns, n_chunks, n_done, state) -- large-file cross-fleet assembly; -- finalize task seeded (same tx as the last data chunk) at n_done==n_chunks +delete_groups (pass_id, rel_path, n_total, n_done, -- §2.2 DELETE fan-out; the + done_streaming, closed) -- delete-pass analogue of + -- chunk_groups. n_total counts split-produced DELETE *shards* (bumped by + -- one per DeleteRemainder batch received, NOT total_children's entry + -- count — see §2.2's "found in local verification" note for why those + -- two must never be compared); done_streaming is set once the final + -- (EOF) batch lands. Cleanup shard seeded once done_streaming=1 AND + -- n_done>=n_total, whichever of CompleteDeleteRemainder/RecordSplit + -- observes that first (closed dedups) link_groups (pass_id, dev, ino, nlink_expected, members_seen, anchor_rel_path, anchor_size, anchor_mtime_ns, anchor_state, updated_at) -- D11 hardlink correlation, pass-scoped; diff --git a/docs/DESIGN-protocol.md b/docs/DESIGN-protocol.md index f1a01c0..31502ba 100644 --- a/docs/DESIGN-protocol.md +++ b/docs/DESIGN-protocol.md @@ -50,7 +50,7 @@ Direction key: `A→C` agent to coordinator, `C→A` coordinator to agent. | 4 | `HeartbeatAck` | C→A | piggybacks control state: pause/resume/drain flags, config-changed epoch | | 5 | `WorkRequest` | A→C | credit-based pull: "I have capacity for N shards / M copy tasks"; sent whenever local queues drop below low-water marks | | 6 | `WorkGrant` | C→A | 0..N work items, each a `Shard`, `EntryListShard`, `ChunkTask`, `DirFixBatch`, `VerifyBatch`, `DeleteBatch`, `ProbeTask` (a mount-probe pinned to this agent, gating pass start), or `LinkTask` (D11, linkat a hardlink-group member to its anchor — on by default, `docs/DESIGN-hardlinks.md`); each carries a lease (id, TTL) | -| 7 | `ShardSplit` | A→C | new shards discovered mid-walk (subdirectories pushed back, or entry-list batches from a huge directory); also carries `LinkSighting`s (D11, nlink>1 files — sent unconditionally, acted on unless the job opted out via `hardlinks: report`); coordinator persists + queues them, acks with assigned shard ids | +| 7 | `ShardSplit` | A→C | new shards discovered mid-walk (subdirectories pushed back, or entry-list batches from a huge directory), or delete-remainder batches from a pathological orphan directory being fanned out during a DELETE pass (`DeleteRemainder`, docs/DESIGN-coordinator.md §2.2); also carries `LinkSighting`s (D11, nlink>1 files — sent unconditionally, acted on unless the job opted out via `hardlinks: report`); coordinator persists + queues them, acks with assigned shard ids | | 8 | `ShardSplitAck` | C→A | ids assigned; until received, the agent must not report the parent shard complete (no lost subtrees) | | 9 | `ShardResult` | A→C | terminal state of a leased shard: counters (entries walked, tasks emitted/completed, bytes copied), orphan count, error summary, nlink>1 stats, wall/IO timings. Status `RESULT_RELEASED` means a draining agent is handing a shard back **unstarted** (not a failure) so the coordinator re-queues it for an active agent | | 10 | `TaskResult` | A→C | terminal state for coordinator-tracked tasks (chunk copies, dirfix batches, verify batches); batched | diff --git a/proto/drsync.proto b/proto/drsync.proto index f34ba4a..0fb24ab 100644 --- a/proto/drsync.proto +++ b/proto/drsync.proto @@ -274,6 +274,15 @@ message TuningOptions { // fanned out. Sets the granularity of that fan-out: a 1.4M-entry directory // becomes ceil(1.4M / entrylist_batch) shards. uint32 entrylist_batch = 6; + // Delete-pass analogue of dir_split_threshold/entrylist_batch: a directory + // being orphan-deleted (agent/src/delete.c) whose own entry count exceeds + // delete_split_threshold is streamed out as new DELETE shards instead of + // being unlinked depth-first by one agent (docs/DESIGN-coordinator.md §2.2 + // DELETE fan-out). Kept independently tunable from the entry-list values: + // delete work per name is a plain unlink, not a full stat/diff/copy + // pipeline, so the shapes that make sense for each can differ. + uint64 delete_split_threshold = 7; + uint32 delete_split_batch = 8; // names per split-produced DELETE shard (0 = built-in default) } message JobOptions { @@ -510,10 +519,34 @@ message ShardSplit { uint64 size = 5; int64 mtime_ns = 6; } - repeated NewShard subdirs = 3; - repeated NewEntryList entry_lists = 4; - repeated BigFile big_files = 5; - repeated LinkSighting link_sightings = 6; + // A pathological orphan directory being deleted (agent/src/delete.c's + // rm_tree, WI_DELETE): once its own entry count exceeds + // delete_split_threshold, the remaining not-yet-removed entries are + // streamed out in batches as new DELETE shards instead of being unlinked + // depth-first by the one agent that found the directory — the delete-pass + // analogue of NewEntryList above. Unlike NewEntryList there is nothing to + // diff against a destination: the whole subtree is already condemned + // (D5), so a batch is just names to remove, deepest-first ordering + // preserved by construction (every name here is a direct child of dir_rel, + // never an ancestor of anything else in this same split). + message DeleteRemainder { + bytes dir_rel = 1; + repeated bytes names = 2; + // Total DELETE shards this directory is being split into, set ONLY on + // the last DeleteRemainder batch shipped for dir_rel (0 on every earlier + // batch — the agent streams via readdir and only knows the true total + // once it hits EOF, unlike a chunk group's byte-size-derived n_chunks, + // known upfront). The coordinator uses this to size delete_groups' + // n_children the one time it sees it, then seeds a cleanup shard for + // dir_rel itself once every child DELETE shard reports done — see + // store.RecordSplit / CompleteShard's delete_groups handling. + uint32 total_children = 3; + } + repeated NewShard subdirs = 3; + repeated NewEntryList entry_lists = 4; + repeated BigFile big_files = 5; + repeated LinkSighting link_sightings = 6; + repeated DeleteRemainder delete_remainders = 7; } message ShardSplitAck { diff --git a/proto/gen/drsyncpb/drsync.pb.go b/proto/gen/drsyncpb/drsync.pb.go index 78351a7..4d31e12 100644 --- a/proto/gen/drsyncpb/drsync.pb.go +++ b/proto/gen/drsyncpb/drsync.pb.go @@ -1944,8 +1944,17 @@ type TuningOptions struct { // fanned out. Sets the granularity of that fan-out: a 1.4M-entry directory // becomes ceil(1.4M / entrylist_batch) shards. EntrylistBatch uint32 `protobuf:"varint,6,opt,name=entrylist_batch,json=entrylistBatch,proto3" json:"entrylist_batch,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Delete-pass analogue of dir_split_threshold/entrylist_batch: a directory + // being orphan-deleted (agent/src/delete.c) whose own entry count exceeds + // delete_split_threshold is streamed out as new DELETE shards instead of + // being unlinked depth-first by one agent (docs/DESIGN-coordinator.md §2.2 + // DELETE fan-out). Kept independently tunable from the entry-list values: + // delete work per name is a plain unlink, not a full stat/diff/copy + // pipeline, so the shapes that make sense for each can differ. + DeleteSplitThreshold uint64 `protobuf:"varint,7,opt,name=delete_split_threshold,json=deleteSplitThreshold,proto3" json:"delete_split_threshold,omitempty"` + DeleteSplitBatch uint32 `protobuf:"varint,8,opt,name=delete_split_batch,json=deleteSplitBatch,proto3" json:"delete_split_batch,omitempty"` // names per split-produced DELETE shard (0 = built-in default) + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *TuningOptions) Reset() { @@ -2020,6 +2029,20 @@ func (x *TuningOptions) GetEntrylistBatch() uint32 { return 0 } +func (x *TuningOptions) GetDeleteSplitThreshold() uint64 { + if x != nil { + return x.DeleteSplitThreshold + } + return 0 +} + +func (x *TuningOptions) GetDeleteSplitBatch() uint32 { + if x != nil { + return x.DeleteSplitBatch + } + return 0 +} + type JobOptions struct { state protoimpl.MessageState `protogen:"open.v1"` JobId uint64 `protobuf:"varint,1,opt,name=job_id,json=jobId,proto3" json:"job_id,omitempty"` @@ -3515,15 +3538,16 @@ func (x *WorkGrant) GetOptions() []*JobOptions { } type ShardSplit struct { - state protoimpl.MessageState `protogen:"open.v1"` - ParentShardId uint64 `protobuf:"varint,1,opt,name=parent_shard_id,json=parentShardId,proto3" json:"parent_shard_id,omitempty"` - Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"` // per-parent sequence for retransmit dedup - Subdirs []*ShardSplit_NewShard `protobuf:"bytes,3,rep,name=subdirs,proto3" json:"subdirs,omitempty"` - EntryLists []*ShardSplit_NewEntryList `protobuf:"bytes,4,rep,name=entry_lists,json=entryLists,proto3" json:"entry_lists,omitempty"` - BigFiles []*ShardSplit_BigFile `protobuf:"bytes,5,rep,name=big_files,json=bigFiles,proto3" json:"big_files,omitempty"` - LinkSightings []*ShardSplit_LinkSighting `protobuf:"bytes,6,rep,name=link_sightings,json=linkSightings,proto3" json:"link_sightings,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + ParentShardId uint64 `protobuf:"varint,1,opt,name=parent_shard_id,json=parentShardId,proto3" json:"parent_shard_id,omitempty"` + Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"` // per-parent sequence for retransmit dedup + Subdirs []*ShardSplit_NewShard `protobuf:"bytes,3,rep,name=subdirs,proto3" json:"subdirs,omitempty"` + EntryLists []*ShardSplit_NewEntryList `protobuf:"bytes,4,rep,name=entry_lists,json=entryLists,proto3" json:"entry_lists,omitempty"` + BigFiles []*ShardSplit_BigFile `protobuf:"bytes,5,rep,name=big_files,json=bigFiles,proto3" json:"big_files,omitempty"` + LinkSightings []*ShardSplit_LinkSighting `protobuf:"bytes,6,rep,name=link_sightings,json=linkSightings,proto3" json:"link_sightings,omitempty"` + DeleteRemainders []*ShardSplit_DeleteRemainder `protobuf:"bytes,7,rep,name=delete_remainders,json=deleteRemainders,proto3" json:"delete_remainders,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ShardSplit) Reset() { @@ -3598,6 +3622,13 @@ func (x *ShardSplit) GetLinkSightings() []*ShardSplit_LinkSighting { return nil } +func (x *ShardSplit) GetDeleteRemainders() []*ShardSplit_DeleteRemainder { + if x != nil { + return x.DeleteRemainders + } + return nil +} + type ShardSplitAck struct { state protoimpl.MessageState `protogen:"open.v1"` ParentShardId uint64 `protobuf:"varint,1,opt,name=parent_shard_id,json=parentShardId,proto3" json:"parent_shard_id,omitempty"` @@ -4802,6 +4833,84 @@ func (x *ShardSplit_LinkSighting) GetMtimeNs() int64 { return 0 } +// A pathological orphan directory being deleted (agent/src/delete.c's +// rm_tree, WI_DELETE): once its own entry count exceeds +// delete_split_threshold, the remaining not-yet-removed entries are +// streamed out in batches as new DELETE shards instead of being unlinked +// depth-first by the one agent that found the directory — the delete-pass +// analogue of NewEntryList above. Unlike NewEntryList there is nothing to +// diff against a destination: the whole subtree is already condemned +// (D5), so a batch is just names to remove, deepest-first ordering +// preserved by construction (every name here is a direct child of dir_rel, +// never an ancestor of anything else in this same split). +type ShardSplit_DeleteRemainder struct { + state protoimpl.MessageState `protogen:"open.v1"` + DirRel []byte `protobuf:"bytes,1,opt,name=dir_rel,json=dirRel,proto3" json:"dir_rel,omitempty"` + Names [][]byte `protobuf:"bytes,2,rep,name=names,proto3" json:"names,omitempty"` + // Total DELETE shards this directory is being split into, set ONLY on + // the last DeleteRemainder batch shipped for dir_rel (0 on every earlier + // batch — the agent streams via readdir and only knows the true total + // once it hits EOF, unlike a chunk group's byte-size-derived n_chunks, + // known upfront). The coordinator uses this to size delete_groups' + // n_children the one time it sees it, then seeds a cleanup shard for + // dir_rel itself once every child DELETE shard reports done — see + // store.RecordSplit / CompleteShard's delete_groups handling. + TotalChildren uint32 `protobuf:"varint,3,opt,name=total_children,json=totalChildren,proto3" json:"total_children,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ShardSplit_DeleteRemainder) Reset() { + *x = ShardSplit_DeleteRemainder{} + mi := &file_drsync_proto_msgTypes[51] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ShardSplit_DeleteRemainder) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ShardSplit_DeleteRemainder) ProtoMessage() {} + +func (x *ShardSplit_DeleteRemainder) ProtoReflect() protoreflect.Message { + mi := &file_drsync_proto_msgTypes[51] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ShardSplit_DeleteRemainder.ProtoReflect.Descriptor instead. +func (*ShardSplit_DeleteRemainder) Descriptor() ([]byte, []int) { + return file_drsync_proto_rawDescGZIP(), []int{35, 4} +} + +func (x *ShardSplit_DeleteRemainder) GetDirRel() []byte { + if x != nil { + return x.DirRel + } + return nil +} + +func (x *ShardSplit_DeleteRemainder) GetNames() [][]byte { + if x != nil { + return x.Names + } + return nil +} + +func (x *ShardSplit_DeleteRemainder) GetTotalChildren() uint32 { + if x != nil { + return x.TotalChildren + } + return 0 +} + var File_drsync_proto protoreflect.FileDescriptor const file_drsync_proto_rawDesc = "" + @@ -4959,7 +5068,7 @@ const file_drsync_proto_rawDesc = "" + "\rFSYNC_BATCHED\x10\x02\"d\n" + "\fLimitOptions\x12.\n" + "\x13bandwidth_per_agent\x18\x01 \x01(\x04R\x11bandwidthPerAgent\x12$\n" + - "\x0eiops_per_agent\x18\x02 \x01(\x04R\fiopsPerAgent\"\xf4\x01\n" + + "\x0eiops_per_agent\x18\x02 \x01(\x04R\fiopsPerAgent\"\xd8\x02\n" + "\rTuningOptions\x12!\n" + "\fshard_budget\x18\x01 \x01(\x04R\vshardBudget\x12.\n" + "\x13dir_split_threshold\x18\x02 \x01(\x04R\x11dirSplitThreshold\x12\x1f\n" + @@ -4967,7 +5076,9 @@ const file_drsync_proto_rawDesc = "" + "statxBatch\x12\"\n" + "\rmtime_slop_ns\x18\x04 \x01(\x03R\vmtimeSlopNs\x12\"\n" + "\rop_deadline_s\x18\x05 \x01(\rR\vopDeadlineS\x12'\n" + - "\x0fentrylist_batch\x18\x06 \x01(\rR\x0eentrylistBatch\"\xff\x03\n" + + "\x0fentrylist_batch\x18\x06 \x01(\rR\x0eentrylistBatch\x124\n" + + "\x16delete_split_threshold\x18\a \x01(\x04R\x14deleteSplitThreshold\x12,\n" + + "\x12delete_split_batch\x18\b \x01(\rR\x10deleteSplitBatch\"\xff\x03\n" + "\n" + "JobOptions\x12\x15\n" + "\x06job_id\x18\x01 \x01(\x04R\x05jobId\x12\x19\n" + @@ -5095,7 +5206,7 @@ const file_drsync_proto_rawDesc = "" + "\x04item\"g\n" + "\tWorkGrant\x12)\n" + "\x05items\x18\x01 \x03(\v2\x13.drsync.v1.WorkItemR\x05items\x12/\n" + - "\aoptions\x18\x02 \x03(\v2\x15.drsync.v1.JobOptionsR\aoptions\"\x9c\x05\n" + + "\aoptions\x18\x02 \x03(\v2\x15.drsync.v1.JobOptionsR\aoptions\"\xd9\x06\n" + "\n" + "ShardSplit\x12&\n" + "\x0fparent_shard_id\x18\x01 \x01(\x04R\rparentShardId\x12\x10\n" + @@ -5104,7 +5215,8 @@ const file_drsync_proto_rawDesc = "" + "\ventry_lists\x18\x04 \x03(\v2\".drsync.v1.ShardSplit.NewEntryListR\n" + "entryLists\x12:\n" + "\tbig_files\x18\x05 \x03(\v2\x1d.drsync.v1.ShardSplit.BigFileR\bbigFiles\x12I\n" + - "\x0elink_sightings\x18\x06 \x03(\v2\".drsync.v1.ShardSplit.LinkSightingR\rlinkSightings\x1a%\n" + + "\x0elink_sightings\x18\x06 \x03(\v2\".drsync.v1.ShardSplit.LinkSightingR\rlinkSightings\x12R\n" + + "\x11delete_remainders\x18\a \x03(\v2%.drsync.v1.ShardSplit.DeleteRemainderR\x10deleteRemainders\x1a%\n" + "\bNewShard\x12\x19\n" + "\brel_path\x18\x01 \x01(\fR\arelPath\x1a=\n" + "\fNewEntryList\x12\x17\n" + @@ -5120,7 +5232,11 @@ const file_drsync_proto_rawDesc = "" + "\brel_path\x18\x03 \x01(\fR\arelPath\x12\x14\n" + "\x05nlink\x18\x04 \x01(\rR\x05nlink\x12\x12\n" + "\x04size\x18\x05 \x01(\x04R\x04size\x12\x19\n" + - "\bmtime_ns\x18\x06 \x01(\x03R\amtimeNs\"w\n" + + "\bmtime_ns\x18\x06 \x01(\x03R\amtimeNs\x1ag\n" + + "\x0fDeleteRemainder\x12\x17\n" + + "\adir_rel\x18\x01 \x01(\fR\x06dirRel\x12\x14\n" + + "\x05names\x18\x02 \x03(\fR\x05names\x12%\n" + + "\x0etotal_children\x18\x03 \x01(\rR\rtotalChildren\"w\n" + "\rShardSplitAck\x12&\n" + "\x0fparent_shard_id\x18\x01 \x01(\x04R\rparentShardId\x12\x10\n" + "\x03seq\x18\x02 \x01(\x04R\x03seq\x12,\n" + @@ -5276,68 +5392,69 @@ func file_drsync_proto_rawDescGZIP() []byte { } var file_drsync_proto_enumTypes = make([]protoimpl.EnumInfo, 9) -var file_drsync_proto_msgTypes = make([]protoimpl.MessageInfo, 51) +var file_drsync_proto_msgTypes = make([]protoimpl.MessageInfo, 52) var file_drsync_proto_goTypes = []any{ - (FrameType)(0), // 0: drsync.v1.FrameType - (EntryType)(0), // 1: drsync.v1.EntryType - (ResultStatus)(0), // 2: drsync.v1.ResultStatus - (Control_Command)(0), // 3: drsync.v1.Control.Command - (AclOptions_Untranslatable)(0), // 4: drsync.v1.AclOptions.Untranslatable - (VerifyOptions_OnMismatch)(0), // 5: drsync.v1.VerifyOptions.OnMismatch - (CopyOptions_ServerSideCopy)(0), // 6: drsync.v1.CopyOptions.ServerSideCopy - (CopyOptions_FsyncMode)(0), // 7: drsync.v1.CopyOptions.FsyncMode - (JournalRecord_Type)(0), // 8: drsync.v1.JournalRecord.Type - (*StatInfo)(nil), // 9: drsync.v1.StatInfo - (*MountCaps)(nil), // 10: drsync.v1.MountCaps - (*Hello)(nil), // 11: drsync.v1.Hello - (*HelloAck)(nil), // 12: drsync.v1.HelloAck - (*MountHealth)(nil), // 13: drsync.v1.MountHealth - (*InflightItem)(nil), // 14: drsync.v1.InflightItem - (*Heartbeat)(nil), // 15: drsync.v1.Heartbeat - (*HeartbeatAck)(nil), // 16: drsync.v1.HeartbeatAck - (*Control)(nil), // 17: drsync.v1.Control - (*ProtocolError)(nil), // 18: drsync.v1.ProtocolError - (*FilterRule)(nil), // 19: drsync.v1.FilterRule - (*AclOptions)(nil), // 20: drsync.v1.AclOptions - (*MetadataOptions)(nil), // 21: drsync.v1.MetadataOptions - (*VerifyOptions)(nil), // 22: drsync.v1.VerifyOptions - (*CopyOptions)(nil), // 23: drsync.v1.CopyOptions - (*LimitOptions)(nil), // 24: drsync.v1.LimitOptions - (*TuningOptions)(nil), // 25: drsync.v1.TuningOptions - (*JobOptions)(nil), // 26: drsync.v1.JobOptions - (*WorkRequest)(nil), // 27: drsync.v1.WorkRequest - (*FileGen)(nil), // 28: drsync.v1.FileGen - (*WalkOverrides)(nil), // 29: drsync.v1.WalkOverrides - (*Shard)(nil), // 30: drsync.v1.Shard - (*EntryListShard)(nil), // 31: drsync.v1.EntryListShard - (*ChunkTask)(nil), // 32: drsync.v1.ChunkTask - (*DirMeta)(nil), // 33: drsync.v1.DirMeta - (*DirFixBatch)(nil), // 34: drsync.v1.DirFixBatch - (*VerifyEntry)(nil), // 35: drsync.v1.VerifyEntry - (*VerifyBatch)(nil), // 36: drsync.v1.VerifyBatch - (*DeleteBatch)(nil), // 37: drsync.v1.DeleteBatch - (*ProbeTask)(nil), // 38: drsync.v1.ProbeTask - (*LinkTask)(nil), // 39: drsync.v1.LinkTask - (*LinkEntry)(nil), // 40: drsync.v1.LinkEntry - (*LinkTaskBatch)(nil), // 41: drsync.v1.LinkTaskBatch - (*WorkItem)(nil), // 42: drsync.v1.WorkItem - (*WorkGrant)(nil), // 43: drsync.v1.WorkGrant - (*ShardSplit)(nil), // 44: drsync.v1.ShardSplit - (*ShardSplitAck)(nil), // 45: drsync.v1.ShardSplitAck - (*ShardCounters)(nil), // 46: drsync.v1.ShardCounters - (*ShardResult)(nil), // 47: drsync.v1.ShardResult - (*TaskResult)(nil), // 48: drsync.v1.TaskResult - (*TaskResultBatch)(nil), // 49: drsync.v1.TaskResultBatch - (*JournalRecord)(nil), // 50: drsync.v1.JournalRecord - (*JournalBatch)(nil), // 51: drsync.v1.JournalBatch - (*JournalAck)(nil), // 52: drsync.v1.JournalAck - (*LatencyHistogram)(nil), // 53: drsync.v1.LatencyHistogram - (*StatsReport)(nil), // 54: drsync.v1.StatsReport - (*WorkRequest_CachedOptions)(nil), // 55: drsync.v1.WorkRequest.CachedOptions - (*ShardSplit_NewShard)(nil), // 56: drsync.v1.ShardSplit.NewShard - (*ShardSplit_NewEntryList)(nil), // 57: drsync.v1.ShardSplit.NewEntryList - (*ShardSplit_BigFile)(nil), // 58: drsync.v1.ShardSplit.BigFile - (*ShardSplit_LinkSighting)(nil), // 59: drsync.v1.ShardSplit.LinkSighting + (FrameType)(0), // 0: drsync.v1.FrameType + (EntryType)(0), // 1: drsync.v1.EntryType + (ResultStatus)(0), // 2: drsync.v1.ResultStatus + (Control_Command)(0), // 3: drsync.v1.Control.Command + (AclOptions_Untranslatable)(0), // 4: drsync.v1.AclOptions.Untranslatable + (VerifyOptions_OnMismatch)(0), // 5: drsync.v1.VerifyOptions.OnMismatch + (CopyOptions_ServerSideCopy)(0), // 6: drsync.v1.CopyOptions.ServerSideCopy + (CopyOptions_FsyncMode)(0), // 7: drsync.v1.CopyOptions.FsyncMode + (JournalRecord_Type)(0), // 8: drsync.v1.JournalRecord.Type + (*StatInfo)(nil), // 9: drsync.v1.StatInfo + (*MountCaps)(nil), // 10: drsync.v1.MountCaps + (*Hello)(nil), // 11: drsync.v1.Hello + (*HelloAck)(nil), // 12: drsync.v1.HelloAck + (*MountHealth)(nil), // 13: drsync.v1.MountHealth + (*InflightItem)(nil), // 14: drsync.v1.InflightItem + (*Heartbeat)(nil), // 15: drsync.v1.Heartbeat + (*HeartbeatAck)(nil), // 16: drsync.v1.HeartbeatAck + (*Control)(nil), // 17: drsync.v1.Control + (*ProtocolError)(nil), // 18: drsync.v1.ProtocolError + (*FilterRule)(nil), // 19: drsync.v1.FilterRule + (*AclOptions)(nil), // 20: drsync.v1.AclOptions + (*MetadataOptions)(nil), // 21: drsync.v1.MetadataOptions + (*VerifyOptions)(nil), // 22: drsync.v1.VerifyOptions + (*CopyOptions)(nil), // 23: drsync.v1.CopyOptions + (*LimitOptions)(nil), // 24: drsync.v1.LimitOptions + (*TuningOptions)(nil), // 25: drsync.v1.TuningOptions + (*JobOptions)(nil), // 26: drsync.v1.JobOptions + (*WorkRequest)(nil), // 27: drsync.v1.WorkRequest + (*FileGen)(nil), // 28: drsync.v1.FileGen + (*WalkOverrides)(nil), // 29: drsync.v1.WalkOverrides + (*Shard)(nil), // 30: drsync.v1.Shard + (*EntryListShard)(nil), // 31: drsync.v1.EntryListShard + (*ChunkTask)(nil), // 32: drsync.v1.ChunkTask + (*DirMeta)(nil), // 33: drsync.v1.DirMeta + (*DirFixBatch)(nil), // 34: drsync.v1.DirFixBatch + (*VerifyEntry)(nil), // 35: drsync.v1.VerifyEntry + (*VerifyBatch)(nil), // 36: drsync.v1.VerifyBatch + (*DeleteBatch)(nil), // 37: drsync.v1.DeleteBatch + (*ProbeTask)(nil), // 38: drsync.v1.ProbeTask + (*LinkTask)(nil), // 39: drsync.v1.LinkTask + (*LinkEntry)(nil), // 40: drsync.v1.LinkEntry + (*LinkTaskBatch)(nil), // 41: drsync.v1.LinkTaskBatch + (*WorkItem)(nil), // 42: drsync.v1.WorkItem + (*WorkGrant)(nil), // 43: drsync.v1.WorkGrant + (*ShardSplit)(nil), // 44: drsync.v1.ShardSplit + (*ShardSplitAck)(nil), // 45: drsync.v1.ShardSplitAck + (*ShardCounters)(nil), // 46: drsync.v1.ShardCounters + (*ShardResult)(nil), // 47: drsync.v1.ShardResult + (*TaskResult)(nil), // 48: drsync.v1.TaskResult + (*TaskResultBatch)(nil), // 49: drsync.v1.TaskResultBatch + (*JournalRecord)(nil), // 50: drsync.v1.JournalRecord + (*JournalBatch)(nil), // 51: drsync.v1.JournalBatch + (*JournalAck)(nil), // 52: drsync.v1.JournalAck + (*LatencyHistogram)(nil), // 53: drsync.v1.LatencyHistogram + (*StatsReport)(nil), // 54: drsync.v1.StatsReport + (*WorkRequest_CachedOptions)(nil), // 55: drsync.v1.WorkRequest.CachedOptions + (*ShardSplit_NewShard)(nil), // 56: drsync.v1.ShardSplit.NewShard + (*ShardSplit_NewEntryList)(nil), // 57: drsync.v1.ShardSplit.NewEntryList + (*ShardSplit_BigFile)(nil), // 58: drsync.v1.ShardSplit.BigFile + (*ShardSplit_LinkSighting)(nil), // 59: drsync.v1.ShardSplit.LinkSighting + (*ShardSplit_DeleteRemainder)(nil), // 60: drsync.v1.ShardSplit.DeleteRemainder } var file_drsync_proto_depIdxs = []int32{ 1, // 0: drsync.v1.StatInfo.type:type_name -> drsync.v1.EntryType @@ -5379,21 +5496,22 @@ var file_drsync_proto_depIdxs = []int32{ 57, // 36: drsync.v1.ShardSplit.entry_lists:type_name -> drsync.v1.ShardSplit.NewEntryList 58, // 37: drsync.v1.ShardSplit.big_files:type_name -> drsync.v1.ShardSplit.BigFile 59, // 38: drsync.v1.ShardSplit.link_sightings:type_name -> drsync.v1.ShardSplit.LinkSighting - 2, // 39: drsync.v1.ShardResult.status:type_name -> drsync.v1.ResultStatus - 46, // 40: drsync.v1.ShardResult.counters:type_name -> drsync.v1.ShardCounters - 2, // 41: drsync.v1.TaskResult.status:type_name -> drsync.v1.ResultStatus - 10, // 42: drsync.v1.TaskResult.src_caps:type_name -> drsync.v1.MountCaps - 10, // 43: drsync.v1.TaskResult.dst_caps:type_name -> drsync.v1.MountCaps - 48, // 44: drsync.v1.TaskResultBatch.results:type_name -> drsync.v1.TaskResult - 8, // 45: drsync.v1.JournalRecord.type:type_name -> drsync.v1.JournalRecord.Type - 9, // 46: drsync.v1.JournalRecord.src:type_name -> drsync.v1.StatInfo - 9, // 47: drsync.v1.JournalRecord.dst:type_name -> drsync.v1.StatInfo - 53, // 48: drsync.v1.StatsReport.latencies:type_name -> drsync.v1.LatencyHistogram - 49, // [49:49] is the sub-list for method output_type - 49, // [49:49] is the sub-list for method input_type - 49, // [49:49] is the sub-list for extension type_name - 49, // [49:49] is the sub-list for extension extendee - 0, // [0:49] is the sub-list for field type_name + 60, // 39: drsync.v1.ShardSplit.delete_remainders:type_name -> drsync.v1.ShardSplit.DeleteRemainder + 2, // 40: drsync.v1.ShardResult.status:type_name -> drsync.v1.ResultStatus + 46, // 41: drsync.v1.ShardResult.counters:type_name -> drsync.v1.ShardCounters + 2, // 42: drsync.v1.TaskResult.status:type_name -> drsync.v1.ResultStatus + 10, // 43: drsync.v1.TaskResult.src_caps:type_name -> drsync.v1.MountCaps + 10, // 44: drsync.v1.TaskResult.dst_caps:type_name -> drsync.v1.MountCaps + 48, // 45: drsync.v1.TaskResultBatch.results:type_name -> drsync.v1.TaskResult + 8, // 46: drsync.v1.JournalRecord.type:type_name -> drsync.v1.JournalRecord.Type + 9, // 47: drsync.v1.JournalRecord.src:type_name -> drsync.v1.StatInfo + 9, // 48: drsync.v1.JournalRecord.dst:type_name -> drsync.v1.StatInfo + 53, // 49: drsync.v1.StatsReport.latencies:type_name -> drsync.v1.LatencyHistogram + 50, // [50:50] is the sub-list for method output_type + 50, // [50:50] is the sub-list for method input_type + 50, // [50:50] is the sub-list for extension type_name + 50, // [50:50] is the sub-list for extension extendee + 0, // [0:50] is the sub-list for field type_name } func init() { file_drsync_proto_init() } @@ -5419,7 +5537,7 @@ func file_drsync_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_drsync_proto_rawDesc), len(file_drsync_proto_rawDesc)), NumEnums: 9, - NumMessages: 51, + NumMessages: 52, NumExtensions: 0, NumServices: 0, }, diff --git a/template.yaml b/template.yaml index f5ff5b3..75a8464 100644 --- a/template.yaml +++ b/template.yaml @@ -95,6 +95,8 @@ spec: shard_budget: 2000 # entries a walker processes before pushing subdirs back dir_split_threshold: 50000 # single-directory size that triggers entry-list sharding entrylist_batch: 4000 # names per entry-list shard (sets a huge dir's fan-out) + delete_split_threshold: 200000 # single orphan-directory size that triggers delete sharding + delete_split_batch: 20000 # names per split-produced DELETE shard statx_batch: 256 # in-flight statx per walker = io_uring ring depth (1–4096) mtime_slop_ns: 1000000 # 1ms slop for cross-filesystem timestamp granularity spread_mode: auto # auto | off | always — coordinator-side walk fan-out diff --git a/test/delete_fanout_e2e.sh b/test/delete_fanout_e2e.sh new file mode 100755 index 0000000..546ad1a --- /dev/null +++ b/test/delete_fanout_e2e.sh @@ -0,0 +1,164 @@ +#!/usr/bin/env bash +# drsync delete-fanout e2e: a single orphan directory whose own entry count +# exceeds tuning.delete_split_threshold is streamed out as DeleteRemainder +# splits (docs/DESIGN-coordinator.md §2.2 DELETE fan-out) instead of being +# unlinked depth-first by one agent — asserted via the coordinator's recorded +# KindDelete shard count during the pass (must be > 1: the original orphan +# path plus at least one split-produced remainder shard) and via full, +# correct removal including the directory itself. +set -euo pipefail + +ROOT=$(cd "$(dirname "$0")/.." && pwd) +. "$ROOT/test/lib.sh" +WORK=$(mktemp -d "${TMPDIR:-/tmp}/drsync-delfanout.XXXXXX") +read -r _CP _HP < <(pick_ports) +CP=${CP:-$_CP}; HP=${HP:-$_HP} +API="http://127.0.0.1:${HP}"; AUTH="Authorization: Bearer delfanouttok" +PASS=0 +cleanup() { + for p in "${APID:-}" "${CPID:-}"; do [[ -n "$p" ]] && kill "$p" 2>/dev/null || true; done + wait 2>/dev/null || true + if [[ $PASS -eq 1 ]]; then rm -rf "$WORK"; else echo "work dir kept: $WORK"; fi +} +trap cleanup EXIT +fail() { echo "FAIL: $*" >&2; exit 1; } +# has — runs cmd, fails (returns 1) if it errors or its +# output doesn't contain pattern. out is declared before the assignment on +# purpose: `local out=$(CMD)` would mask CMD's exit status behind local's own, +# silently dropping the status check pipefail is meant to give us. Same +# helper as e2e.sh/hardlink_e2e.sh define locally (not in lib.sh, since not +# every script needs it). +has() { + local pat=$1 out + shift + out=$("$@") || return 1 + grep -q -- "$pat" <<<"$out" +} +export DRSYNC_SERVER="$API" DRSYNC_TOKEN=delfanouttok + +API_TOKEN_FILE="$WORK/api-token" +echo -n delfanouttok >"$API_TOKEN_FILE" +chmod 600 "$API_TOKEN_FILE" +DRSYNC="$ROOT/bin/drsync" + +# --- build ------------------------------------------------------------------- +make -C "$ROOT/agent" -s +( cd "$ROOT" && go build -o bin/drsyncd ./coordinator/cmd/drsyncd \ + && go build -o bin/drsync ./cli/drsync ) + +# --- trees: a small source, a destination carrying one huge orphan dir ------- +# The orphan directory is built directly on the destination (not synced then +# deleted from source) — it only needs to exist as a destination-only +# subtree for the scan to journal it as an ORPHAN; how it got there is not +# part of what this test exercises. +SRC="$WORK/src"; DST="$WORK/dst" +mkdir -p "$SRC/keep" "$DST/keep" "$DST/orphandir" +echo keepme > "$SRC/keep/file.txt" +echo keepme > "$DST/keep/file.txt" +for i in $(seq 1 300); do echo "junk $i" > "$DST/orphandir/f$(printf %04d "$i").txt"; done + +# --- services ---------------------------------------------------------------- +"$ROOT/bin/drsyncd" -data-dir "$WORK/coord" -listen-agent 127.0.0.1:$CP \ + -listen-http 127.0.0.1:$HP -api-token-file "$API_TOKEN_FILE" -log-level warn \ + >"$WORK/coord.log" 2>&1 & +CPID=$! +wait_coordinator "$API" "$AUTH" || exit 1 +"$ROOT/agent/bin/drsync-agent" -c 127.0.0.1:$CP -i delfanout-agent -w 4 -C 4 \ + >"$WORK/agent.log" 2>&1 & +APID=$! +sleep 1 + +# Small thresholds so the 300-entry orphan directory (well below what a real +# deployment would call pathological) still exercises the fan-out path +# without the test needing to build a huge tree. +cat > "$WORK/job.yaml" </dev/null | head -1) +delete_shard_count() { + python3 - "$DB" <<'PY' +import sqlite3, sys +c = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True) +print(c.execute("select count(*) from shards where kind='delete'").fetchone()[0]) +PY +} +NDEL=0 +DONE=0 +for _ in $(seq 1 120); do + n=$(delete_shard_count) + [[ "${n:-0}" -gt "$NDEL" ]] && NDEL=$n + # The job's own top-level "state" is "COMPLETED" (with the D) — a pass's + # own nested "state" is "COMPLETE" (without it), so this must not just + # grep for the LAST "state" value in the response (that would pick up + # the delete pass's own passView.State instead of the job's). + curl -sf -H "$AUTH" "$API/api/v1/jobs/delfanout" | grep -q '"state":"COMPLETED"' \ + && { DONE=1; break; } + sleep 0.25 +done +[[ "$DONE" -eq 1 ]] \ + || { tail -n 8 "$WORK"/agent.log "$WORK"/coord.log; fail "delete pass did not complete"; } + +# 1. fan-out actually happened: more than just the one top-level orphan shard +# (the original orphandir path) — split-produced remainder shards, plus the +# coordinator-seeded cleanup shard for orphandir itself, must have run too. +[[ "$NDEL" -ge 3 ]] || fail "only $NDEL delete shards recorded; fan-out did not fire " \ + "(want >=3: at least one remainder batch, the original, and the cleanup shard)" + +# 2. orphandir and everything under it is gone — including the directory +# itself, which only the coordinator-seeded cleanup shard removes (nothing +# in the split-produced children ever unlinks their own parent). +[[ ! -e "$DST/orphandir" ]] || fail "orphandir (or its cleanup) was not fully removed" + +# 3. synced content untouched +DIFF=$(diff -r "$SRC" "$DST" 2>&1 || true) +[[ -z "$DIFF" ]] || fail "delete pass damaged synced content:"$'\n'"$DIFF" + +# 4. no errors, nothing parked (report totals, same fields e2e.sh checks) +"$DRSYNC" report delfanout --json > "$WORK/report.json" +python3 - "$WORK/report.json" <<'EOF' || fail "report shows errors or parked shards" +import json, sys +r = json.load(open(sys.argv[1])) +assert r["totals"]["errors"] == 0, r["totals"]["errors"] +assert r["parked_shard_count"] == 0, r["parked_shard_count"] +EOF + +echo "delete shards recorded: $NDEL; orphandir fully removed; content intact" +PASS=1 +echo "PASS: pathological orphan directory fanned out across DELETE shards OK" diff --git a/webui/console.html b/webui/console.html index c9436ea..cfa27c2 100644 --- a/webui/console.html +++ b/webui/console.html @@ -979,7 +979,7 @@

New job

const kindLabel = k => k === "entrylist" ? "large-dir" : k; // ---------- job template (embedded verbatim from ../template.yaml) ---------- - const JOB_TEMPLATE = "# drsync job template \u2014 copy this, edit source/destination, and submit with:\n# drsync job submit template.yaml --start\n#\n# Every value below is the shipped default unless noted, so a minimal job only\n# needs apiVersion, kind, metadata.name and the two paths \u2014 the rest can be\n# deleted to inherit defaults. Sizes accept KiB/MiB/GiB/TiB suffixes (binary);\n# plain integers are bytes or counts. Only the fields shown here are recognised;\n# unknown keys are rejected at submit time. Full reference: docs/DESIGN-jobspec.md\n# and docs/ADMIN.md.\n\napiVersion: drsync/v1\nkind: Job\n\nmetadata:\n name: example-sync # REQUIRED, unique; also the journal directory name\n description: \"example drsync job\" # optional, free text\n\nspec:\n # REQUIRED. Absolute paths, identical on every agent host, and disjoint from\n # each other. These are the roots on the source and destination mounts.\n source:\n path: /mnt/src/data\n destination:\n path: /mnt/dst/data\n\n # Optional include/exclude rules, evaluated in order \u2014 first match wins; no\n # match keeps the entry (implicit `include: \"**\"`). Globs: ? and * stop at /,\n # ** crosses /. Max 64 rules, each pattern <= 255 bytes. Delete this block to\n # copy everything.\n filters:\n - exclude: \"**/.snapshot/**\" # snapshot dirs of either filesystem\n - exclude: \"**/*.tmp\"\n\n passes:\n max: 5 # hard ceiling; convergence usually stops sooner\n schedule: continuous # continuous | manual (operator triggers each pass)\n converge_when: # stop once a pass's delta is under EITHER (OR-combined)\n delta_files_below: 0 # 0 = only the zero-delta fixpoint stops the job\n delta_bytes_below: 0 # e.g. 50GiB to stop while a small delta remains\n\n copy:\n chunk_threshold: 24GiB # files >= this are eligible to split into chunk tasks\n chunk_size: 8GiB # bytes per chunk; a file > this fans out across agents\n buffer_size: 1MiB # copy buffer unit\n preserve_sparse: true # SEEK_HOLE/DATA, with zero-detect fallback\n server_side_copy: auto # auto | off | require (copy_file_range / reflink)\n temp_naming: \".drsync.tmp.\" # prefix for in-progress destination names\n fsync: batched # per_file | batched\n direct_write: true # write NEW files straight to their final name (faster\n # on GPFS/Weka; a crash leaves a partial, re-copied\n # next pass). Updates always use the atomic temp+rename.\n\n metadata:\n owner: true # uid/gid (needs root on the agents)\n mode: true # permission bits\n times: true # atime + mtime, ns precision\n xattrs: true # all readable xattr namespaces\n specials: true # device nodes, FIFOs, sockets (needs root)\n acls:\n posix: true\n nfs4: true\n untranslatable: warn # warn | fail | skip \u2014 how to treat an untranslatable ACL\n hardlinks: preserve # preserve (default) | report \u2014 preserve links\n # nlink>1 files to a shared destination copy;\n # report copies them independently (D3 behavior;\n # set this to opt out) \u2014 docs/DESIGN-hardlinks.md\n hardlinks_max_group_scan: 0 # cap a link group's member count before giving up and\n # falling back to independent copies; 0 = unlimited\n\n # Mount probe: at pass start each agent verifies its source and destination\n # roots before any bulk work runs, gating the whole pass until all agents pass.\n probe:\n require_mount: true # require each root to sit on a real mounted\n # filesystem, so an unmounted volume's leftover\n # stub directory parks the pass instead of syncing\n # into the underlying rootfs. Set false only when a\n # root legitimately lives on the host root filesystem.\n\n verify:\n mode: \"on\" # on | off \u2014 off skips the verify phase entirely\n checksum:\n sample_rate: 0.01 # deterministic fraction of copied files re-checksummed\n on_mismatch: recopy # recopy | fail\n\n deletes:\n mode: mirror # report | mirror \u2014 mirror deletes destination orphans\n # and requires a second explicit gate at pass-trigger time\n\n limits:\n bandwidth_per_agent: 0 # 0 = unlimited; else bytes/s per agent\n iops_per_agent: 0 # 0 = unlimited\n\n # Rarely changed; defaults are sized for a small fleet of fast agents.\n tuning:\n shard_budget: 2000 # entries a walker processes before pushing subdirs back\n dir_split_threshold: 50000 # single-directory size that triggers entry-list sharding\n entrylist_batch: 4000 # names per entry-list shard (sets a huge dir's fan-out)\n statx_batch: 256 # in-flight statx per walker = io_uring ring depth (1\u20134096)\n mtime_slop_ns: 1000000 # 1ms slop for cross-filesystem timestamp granularity\n spread_mode: auto # auto | off | always \u2014 coordinator-side walk fan-out\n spread_target_per_agent: 32 # walk shards per agent to aim for while spreading\n\n # Optional email; inert unless the coordinator has an SMTP config (-smtp-config).\n # Delete this block for no notifications.\n notifications:\n recipients: # required if either flag below is set\n - ops@example.com\n on_pass_complete: false # email as each pass finishes (the convergence trace)\n on_job_complete: false # one summary email when the job reaches COMPLETED\n # (per-pass table includes each pass's duration)\n # Parked-shard alerts are NOT a flag here: as soon as any shard hits its\n # retry ceiling, `recipients` above gets an email automatically (batched\n # per job per check), independent of the two flags \u2014 see DESIGN-jobspec.md \u00a71.2.\n"; + const JOB_TEMPLATE = "# drsync job template \u2014 copy this, edit source/destination, and submit with:\n# drsync job submit template.yaml --start\n#\n# Every value below is the shipped default unless noted, so a minimal job only\n# needs apiVersion, kind, metadata.name and the two paths \u2014 the rest can be\n# deleted to inherit defaults. Sizes accept KiB/MiB/GiB/TiB suffixes (binary);\n# plain integers are bytes or counts. Only the fields shown here are recognised;\n# unknown keys are rejected at submit time. Full reference: docs/DESIGN-jobspec.md\n# and docs/ADMIN.md.\n\napiVersion: drsync/v1\nkind: Job\n\nmetadata:\n name: example-sync # REQUIRED, unique; also the journal directory name\n description: \"example drsync job\" # optional, free text\n\nspec:\n # REQUIRED. Absolute paths, identical on every agent host, and disjoint from\n # each other. These are the roots on the source and destination mounts.\n source:\n path: /mnt/src/data\n destination:\n path: /mnt/dst/data\n\n # Optional include/exclude rules, evaluated in order \u2014 first match wins; no\n # match keeps the entry (implicit `include: \"**\"`). Globs: ? and * stop at /,\n # ** crosses /. Max 64 rules, each pattern <= 255 bytes. Delete this block to\n # copy everything.\n filters:\n - exclude: \"**/.snapshot/**\" # snapshot dirs of either filesystem\n - exclude: \"**/*.tmp\"\n\n passes:\n max: 5 # hard ceiling; convergence usually stops sooner\n schedule: continuous # continuous | manual (operator triggers each pass)\n converge_when: # stop once a pass's delta is under EITHER (OR-combined)\n delta_files_below: 0 # 0 = only the zero-delta fixpoint stops the job\n delta_bytes_below: 0 # e.g. 50GiB to stop while a small delta remains\n\n copy:\n chunk_threshold: 24GiB # files >= this are eligible to split into chunk tasks\n chunk_size: 8GiB # bytes per chunk; a file > this fans out across agents\n buffer_size: 1MiB # copy buffer unit\n preserve_sparse: true # SEEK_HOLE/DATA, with zero-detect fallback\n server_side_copy: auto # auto | off | require (copy_file_range / reflink)\n temp_naming: \".drsync.tmp.\" # prefix for in-progress destination names\n fsync: batched # per_file | batched\n direct_write: true # write NEW files straight to their final name (faster\n # on GPFS/Weka; a crash leaves a partial, re-copied\n # next pass). Updates always use the atomic temp+rename.\n\n metadata:\n owner: true # uid/gid (needs root on the agents)\n mode: true # permission bits\n times: true # atime + mtime, ns precision\n xattrs: true # all readable xattr namespaces\n specials: true # device nodes, FIFOs, sockets (needs root)\n acls:\n posix: true\n nfs4: true\n untranslatable: warn # warn | fail | skip \u2014 how to treat an untranslatable ACL\n hardlinks: preserve # preserve (default) | report \u2014 preserve links\n # nlink>1 files to a shared destination copy;\n # report copies them independently (D3 behavior;\n # set this to opt out) \u2014 docs/DESIGN-hardlinks.md\n hardlinks_max_group_scan: 0 # cap a link group's member count before giving up and\n # falling back to independent copies; 0 = unlimited\n\n # Mount probe: at pass start each agent verifies its source and destination\n # roots before any bulk work runs, gating the whole pass until all agents pass.\n probe:\n require_mount: true # require each root to sit on a real mounted\n # filesystem, so an unmounted volume's leftover\n # stub directory parks the pass instead of syncing\n # into the underlying rootfs. Set false only when a\n # root legitimately lives on the host root filesystem.\n\n verify:\n mode: \"on\" # on | off \u2014 off skips the verify phase entirely\n checksum:\n sample_rate: 0.01 # deterministic fraction of copied files re-checksummed\n on_mismatch: recopy # recopy | fail\n\n deletes:\n mode: mirror # report | mirror \u2014 mirror deletes destination orphans\n # and requires a second explicit gate at pass-trigger time\n\n limits:\n bandwidth_per_agent: 0 # 0 = unlimited; else bytes/s per agent\n iops_per_agent: 0 # 0 = unlimited\n\n # Rarely changed; defaults are sized for a small fleet of fast agents.\n tuning:\n shard_budget: 2000 # entries a walker processes before pushing subdirs back\n dir_split_threshold: 50000 # single-directory size that triggers entry-list sharding\n entrylist_batch: 4000 # names per entry-list shard (sets a huge dir's fan-out)\n delete_split_threshold: 200000 # single orphan-directory size that triggers delete sharding\n delete_split_batch: 20000 # names per split-produced DELETE shard\n statx_batch: 256 # in-flight statx per walker = io_uring ring depth (1\u20134096)\n mtime_slop_ns: 1000000 # 1ms slop for cross-filesystem timestamp granularity\n spread_mode: auto # auto | off | always \u2014 coordinator-side walk fan-out\n spread_target_per_agent: 32 # walk shards per agent to aim for while spreading\n\n # Optional email; inert unless the coordinator has an SMTP config (-smtp-config).\n # Delete this block for no notifications.\n notifications:\n recipients: # required if either flag below is set\n - ops@example.com\n on_pass_complete: false # email as each pass finishes (the convergence trace)\n on_job_complete: false # one summary email when the job reaches COMPLETED\n # (per-pass table includes each pass's duration)\n # Parked-shard alerts are NOT a flag here: as soon as any shard hits its\n # retry ceiling, `recipients` above gets an email automatically (batched\n # per job per check), independent of the two flags \u2014 see DESIGN-jobspec.md \u00a71.2.\n"; // ---------- live state ---------- const S = { jobs:[], queue:{depth:[],parked:[]}, agents:[], rates:{}, glob:{},