From 17ec1b75c601a0e737533d39dbdfed53edf109b0 Mon Sep 17 00:00:00 2001 From: Steven Rhoods Date: Tue, 11 Aug 2026 22:23:39 +0100 Subject: [PATCH 1/2] Add work-budget-based DELETE fan-out for trees with no single wide directory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior fix (#68) caught a directory that is itself WIDE at any depth, but a real production tree hit a shape that mechanism structurally cannot see: 77 top-level branches, several levels deep, every individual directory comfortably under any threshold. No directory anywhere in the tree was ever wide enough to trip delete_split_threshold, so the whole multi-million-object tree ran serially inside 2 shards. Adds tuning.delete_shard_budget (default 250000, objects removed), mirroring the scan walker's shard_budget/queue_split: agent/src/ delete.c decrements a per-shard budget on every object removed, threaded through the recursive descent. Once exhausted, every not-yet-opened subdirectory is handed off as its own new top-level DELETE shard (ShardSplit.delete_subdirs) instead of being recursed into. This introduced two more completion-ordering bugs, both caught locally by the new delete_fanout_budget_e2e.sh before reaching CI: 1. A handed-off shard's own single-shard "group" was seeding a redundant cleanup rmdir on top of its own ordinary removal — fixed with a NoSelfCleanup flag on DeleteGroupTotal. 2. A directory that was never itself handed off, but merely an ancestor of one deep in an otherwise-inline rm_tree recursion, had no way to know a descendant was still mid-removal elsewhere, and rmdir'd itself prematurely. Fixed by having the agent track this locally (no coordinator round-trip): a handoff anywhere inside an in-progress rm_tree call propagates back up the C call stack, and every ancestor that sees a deferred descendant skips its own rmdir too, reporting the deferred paths once in its own ShardResult (deferred_rmdirs, new proto field). Coordinator-side, registerPendingChildTx now walks every ancestor up to the top-level orphan path (not just the direct parent) the first time anything beneath it is handed off, stopping the climb the moment it reaches an already-tracked level — an actual over-counting bug (pending_ children reaching 69 instead of 10) was caught and fixed during this same investigation before that stop condition was added. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PsdNZLfmAFrMX2VUtkLtmm --- .github/workflows/ci.yml | 1 + README.md | 1 + agent/src/agent.h | 8 + agent/src/chunk.c | 2 +- agent/src/delete.c | 262 ++++++++++++++--- agent/src/dirfix.c | 2 +- agent/src/link.c | 2 +- agent/src/main.c | 2 +- agent/src/msgs.c | 20 +- agent/src/msgs.h | 22 +- agent/src/probe.c | 2 +- agent/src/verify.c | 2 +- agent/src/walker.c | 4 +- coordinator/internal/agentsrv/server.go | 76 ++++- coordinator/internal/model/spec.go | 22 ++ .../internal/model/spec_defaults_test.go | 10 +- .../internal/store/deletefanout_test.go | 12 +- coordinator/internal/store/store.go | 269 ++++++++++++------ docs/ADMIN.md | 18 ++ docs/DESIGN-coordinator.md | 149 +++++++--- proto/drsync.proto | 40 +++ proto/gen/drsyncpb/drsync.pb.go | 131 ++++++--- template.yaml | 1 + test/delete_fanout_budget_e2e.sh | 175 ++++++++++++ webui/console.html | 2 +- 25 files changed, 1005 insertions(+), 230 deletions(-) create mode 100755 test/delete_fanout_budget_e2e.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 72adb34..b2f9119 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -116,6 +116,7 @@ jobs: - deep_e2e # directory chain deeper than the walker's in-agent limit - delete_fanout_e2e # pathological orphan directory fans out across DELETE shards - delete_fanout_nested_e2e # fan-out applies at every depth, not just the top-level orphan path + - delete_fanout_budget_e2e # budget-based fan-out on a tree with no single wide directory - 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 b7e1c21..ddb1b98 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ tree, and builds the binaries it needs itself: | `deep_e2e.sh` | directory chain deeper than the walker's in-agent limit | | `delete_fanout_e2e.sh` | pathological orphan directory fans out across DELETE shards | | `delete_fanout_nested_e2e.sh` | fan-out applies at every depth, not just the top-level orphan path | +| `delete_fanout_budget_e2e.sh` | budget-based fan-out on a tree with no single wide directory | | `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/src/agent.h b/agent/src/agent.h index 6d02941..36b88ce 100644 --- a/agent/src/agent.h +++ b/agent/src/agent.h @@ -287,6 +287,14 @@ struct walk_ctx { struct split_wait *infl[SPLIT_WINDOW]; size_t infl_head, infl_count; unsigned tmp_seq; /* atomic: unique temp names per shard */ + /* DELETE only (delete.c): relative paths of directories this shard + * processed inline but did NOT rmdir, because a descendant somewhere + * beneath them was handed off via queue_delete_subdir and might still be + * in flight elsewhere — see ShardResult.deferred_rmdirs' doc comment + * (proto/drsync.proto) for the coordinator-side completion tracking this + * feeds. */ + char **deferred; + size_t n_deferred, cap_deferred; char err[256]; bool fatal; }; diff --git a/agent/src/chunk.c b/agent/src/chunk.c index 89b3659..d0d5d22 100644 --- a/agent/src/chunk.c +++ b/agent/src/chunk.c @@ -286,7 +286,7 @@ void process_chunk(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); jrn_destroy(&ctx); diff --git a/agent/src/delete.c b/agent/src/delete.c index 69ab3ba..85451f7 100644 --- a/agent/src/delete.c +++ b/agent/src/delete.c @@ -7,22 +7,36 @@ * Every removed object is journaled JR_DELETED; dry-run jobs journal * 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. remove_object() applies this check at EVERY - * directory the removal touches, not just the top-level path named in the - * shard's paths[] — rm_dir_contents() routes every entry it finds back - * through remove_object() rather than recursing directly, so a directory - * that is individually small still gets caught if it is nested many levels - * deep under a huge tree of otherwise-small directories (a wide/deep tree - * where no single directory looks large from its own parent's point of view - * previously never split at all, and got removed serially by one agent - * thread — see docs/DESIGN-coordinator.md §2.2 for the incident that - * surfaced this). */ + * Fan-out (docs/DESIGN-coordinator.md §2.2 DELETE fan-out) has TWO + * independent mechanisms, catching two different pathological shapes: + * + * 1. delete_split_threshold / stream_delete_split: a directory whose OWN + * entry count exceeds the 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. remove_object() applies this check + * at EVERY directory the removal touches, not just the top-level path + * named in the shard's paths[] — rm_dir_contents() routes every entry it + * finds back through remove_object() rather than recursing directly, so + * a directory that is individually small still gets caught if it is + * nested many levels deep under a huge tree of otherwise-small + * directories. Catches a WIDE directory at any depth. + * + * 2. delete_shard_budget / queue_delete_subdir: bounds the total number of + * objects ONE shard removes, regardless of tree shape — the delete-pass + * analogue of the scan walker's queue_split/shard_budget. Once the + * budget (decremented per object removed, threaded through the whole + * recursive descent via walk_ctx.budget, same as the walker) runs out, + * every not-yet-opened subdirectory rm_dir_contents finds is handed off + * as its OWN new top-level DELETE shard (ShardSplit.delete_subdirs) + * instead of being recursed into. Catches a tree that is pathological by + * aggregate DEPTH/BRANCHING even when no single directory anywhere in it + * individually exceeds delete_split_threshold — mechanism 1 alone cannot + * see this shape, since it only ever looks at one directory's own + * immediate entry count. This was a real production incident: a 77-way + * branching orphan tree, each branch several levels deep, with every + * individual directory well under any reasonable threshold, took 6 hours + * on the last 2 (of what should have been many more) shards. */ #include "agent.h" #include @@ -42,19 +56,108 @@ * its own right. */ #define DELETE_SPLIT_BATCH_DEFAULT 20000 -static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, - const char *name, const char *rel); +/* Default work budget (objects removed) for one DELETE shard when the job + * doesn't override it (tuning.delete_shard_budget) — mirrors shard_budget's + * own built-in fallback (walker.c), scaled up the same way + * DELETE_SPLIT_BATCH_DEFAULT is scaled up from ENTRYLIST_BATCH_DEFAULT: + * delete work per object is a plain unlink, not a full stat/diff/copy. */ +#define DELETE_SHARD_BUDGET_DEFAULT 250000 + +/* How many delete_subdirs accumulate before flush_delete_subdir_splits ships + * them as one ShardSplit frame — same batching rationale as queue_split's + * SPLIT_BATCH (walker.c): keeps one frame from growing unbounded when the + * budget runs out inside a directory with many subdirectory siblings still + * unopened. */ +#define DELETE_SUBDIR_SPLIT_BATCH 256 + +static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, const char *name, + const char *rel, bool top_level, bool *deferred_out); + +/* Records rel (a directory this shard processed inline but deliberately did + * NOT rmdir, because something beneath it was handed off) into + * ctx->deferred[] — reusing the same growable-array pattern as + * queue_delete_subdir/ctx->split[], a separate array since both can be + * live at once (a directory can defer its own rmdir while ALSO still + * accumulating not-yet-flushed delete_subdirs from a sibling). Flushed into + * ShardResult.deferred_rmdirs at the very end of process_delete — unlike + * split[], there is no mid-shard flush: the coordinator only needs this list + * once, attached to this shard's own final result, not as a streamed batch. */ +static void queue_deferred_rmdir(struct walk_ctx *ctx, const char *rel) +{ + if (ctx->n_deferred == ctx->cap_deferred) { + size_t cap = ctx->cap_deferred ? ctx->cap_deferred * 2 : 64; + char **nv = realloc(ctx->deferred, cap * sizeof *nv); + if (!nv) { + CTR_ADD(ctx->c.errors, 1); + return; + } + ctx->deferred = nv; + ctx->cap_deferred = cap; + } + ctx->deferred[ctx->n_deferred] = strdup(rel); + if (ctx->deferred[ctx->n_deferred]) + ctx->n_deferred++; +} + +/* Accumulates rel (a wholly unopened subdirectory the exhausted budget is + * handing off) into ctx->split[] — reusing the SAME accumulator fields the + * walker's queue_split uses, safe because only one of walker.c/delete.c + * ever runs per shard (they never share a live walk_ctx). Flushed via + * enc_delete_subdir_split (ShardSplit.delete_subdirs, wire field 8) instead + * of enc_shard_split (ShardSplit.subdirs, field 3) — a delete_subdirs entry + * must become a new KindDelete shard, not a KindDir walk/diff shard, so it + * cannot share the wire field with the walker's own subdirs without the + * coordinator losing that distinction. */ +static void flush_delete_subdir_splits(struct walk_ctx *ctx) +{ + if (!ctx->n_split) + return; + pb_buf b; + pb_init(&b); + enc_delete_subdir_split(&b, ctx->it->shard_id, ctx->split_seq, ctx->split, ctx->n_split); + ship_split(ctx, &b); + for (size_t i = 0; i < ctx->n_split; i++) + free(ctx->split[i]); + ctx->n_split = 0; +} + +static void queue_delete_subdir(struct walk_ctx *ctx, const char *rel) +{ + if (ctx->n_split == ctx->cap_split) { + size_t cap = ctx->cap_split ? ctx->cap_split * 2 : DELETE_SUBDIR_SPLIT_BATCH; + char **nv = realloc(ctx->split, cap * sizeof *nv); + if (!nv) { + CTR_ADD(ctx->c.errors, 1); + return; + } + ctx->split = nv; + ctx->cap_split = cap; + } + ctx->split[ctx->n_split] = strdup(rel); + if (ctx->split[ctx->n_split]) + ctx->n_split++; + if (ctx->n_split >= DELETE_SUBDIR_SPLIT_BATCH) + flush_delete_subdir_splits(ctx); +} /* depth-first removal of the already-open directory d (fd dirfd(d), path * rel) — every entry goes through remove_object, so a subdirectory found - * at ANY depth that turns out to be pathological is streamed out via + * at ANY depth that turns out to be pathological (WIDE — its own entry + * count over delete_split_threshold) is streamed out via * stream_delete_split exactly like a top-level orphan, not just the one * named directly in the shard's paths[] (see the fan-out comment at the top * of this file). Does not remove rel itself or consume d; the caller does * both, so remove_object can decide not to (a directory it just handed off - * to a split must NOT be rmdir'd here — the split's own cleanup shard owns - * that once every batch it produced has drained). */ -static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel) + * to a split — either kind — must NOT be rmdir'd here). top_level=false is + * passed to remove_object for every entry here (never true — top_level only + * applies to a shard's own paths[], see process_delete): this is what makes + * the budget check apply to nested recursion but never skip a shard's own + * assigned top-level orphan path. *deferred_out is set true if ANY entry + * here was itself deferred or handed off (by either fan-out mechanism) — + * propagated up so rm_tree knows not to rmdir rel itself yet (see rm_tree's + * own doc comment for why: something under rel may still be mid-removal on + * another shard entirely). */ +static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel, bool *deferred_out) { uint64_t removed = 0; struct dirent *de; @@ -65,7 +168,7 @@ static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel) continue; char crel[PATH_MAX]; snprintf(crel, sizeof crel, "%s/%s", rel, de->d_name); - removed += remove_object(ctx, dirfd(d), de->d_name, crel); + removed += remove_object(ctx, dirfd(d), de->d_name, crel, false, deferred_out); } return removed; } @@ -73,14 +176,40 @@ static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel) /* Fully removes name (file, or directory including every descendant) inside * parentfd — depth-first, via rm_dir_contents. Returns removed count. Only * reached for a directory once remove_object has already decided it is NOT - * pathological (under delete_split_threshold), so no further probing here. */ + * pathological (under delete_split_threshold), so no further probing here. + * Decrements ctx->budget by one for this object once removed — threaded + * through the whole recursive descent (same field, same semantics as the + * scan walker's shard_budget/queue_split), so remove_object's budget check + * on the next subdirectory sees work already done deeper in this same + * shard, not just at this level. + * + * If already_open (a directory, not a file) and rm_dir_contents reports any + * descendant was deferred/handed-off, this directory's OWN rmdir is skipped + * too (queue_deferred_rmdir records rel instead) and *deferred_out (the + * CALLER's own flag, one level up) is set — the parent recursing into US + * must defer ITS rmdir as well, since it cannot know rel is actually empty + * until every handed-off descendant's own group closes. This is what makes + * the propagation chain reach all the way up to the shard's own top-level + * path even though the handoff itself may have happened many levels deeper + * — see docs/DESIGN-coordinator.md §2.2 for the completion-tracking this + * feeds coordinator-side (registerPendingChildTx's ancestor walk, + * ShardResult.deferred_rmdirs). A file is never deferred — only a directory + * can have a descendant handed off — so deferred_out is left untouched on + * the non-directory path. */ static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name, - const char *rel, DIR *already_open) + const char *rel, DIR *already_open, bool *deferred_out) { uint64_t removed = 0; if (already_open) { - removed = rm_dir_contents(ctx, already_open, rel); + bool child_deferred = false; + removed = rm_dir_contents(ctx, already_open, rel, &child_deferred); closedir(already_open); + if (child_deferred) { + queue_deferred_rmdir(ctx, rel); + if (deferred_out) + *deferred_out = true; + return removed; + } if (unlinkat(parentfd, name, AT_REMOVEDIR) < 0 && errno != ENOENT) { walk_err(ctx, "rmdir", rel); return removed; @@ -92,6 +221,8 @@ static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name, } } jrn_emit(ctx, JR_DELETED, rel, NULL, NULL, 0, NULL); + if (ctx->budget > 0) + ctx->budget--; return removed + 1; } @@ -212,21 +343,34 @@ int open_parent_beneath(int root_fd, const char *rel, const char **leaf) return cur; } -/* Removes name (file or directory) inside parentfd, fanning out to - * stream_delete_split instead of recursing if it is a directory over - * delete_split_threshold. Called both for a shard's top-level orphan paths - * (process_delete) and for every entry rm_dir_contents finds while - * descending an already-accepted directory — so a pathologically large - * subdirectory found at ANY depth is caught, not just one named directly in - * the shard's paths[] (the gap the single-level-only version of this check - * left: a tree of many individually-small subdirectories summing to tens of - * millions of entries never split at all, since no single directory in the - * chain ever looked large from its own parent's point of view). Returns the - * number of objects this shard itself removed (0 if it handed the directory - * off to a split instead — those objects are counted by the split-produced - * shards that actually remove them). */ -static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, - const char *name, const char *rel) +/* Removes name (file or directory) inside parentfd. Two independent + * fan-out checks apply to a directory (see the file-level comment for the + * shapes each catches): + * - WIDE: if its own entry count is over delete_split_threshold, streams + * it out via stream_delete_split instead of recursing. Sets + * *deferred_out — the caller (rm_dir_contents, one level up) must not + * let ITS OWN containing directory rmdir until this handoff's group + * closes. + * - BUDGET: if this shard's ctx->budget is exhausted (top_level=false + * only — never a shard's own assigned paths[] entry, see + * process_delete/rm_dir_contents), hands it off UNOPENED as a new + * top-level DELETE shard via queue_delete_subdir instead of even + * probing it. Also sets *deferred_out. + * Called both for a shard's top-level orphan paths (process_delete, + * top_level=true, deferred_out=NULL — nothing above a top-level path to + * propagate to within this shard; see process_delete for how a top-level + * path's OWN deferred rmdir is recorded) and for every entry + * rm_dir_contents finds while descending an already-accepted directory + * (top_level=false, deferred_out always non-NULL there) — so a + * pathologically WIDE subdirectory found at ANY depth is caught, not just + * one named directly in the shard's paths[]. Returns the number of objects + * this shard itself removed (0 if it handed the directory off to either + * kind of split instead, OR deferred its own rmdir — either way those + * objects/that directory are accounted for elsewhere: by the split-produced + * shards that actually remove them, or by the eventual cleanup shard + * closeDeleteGroupTx seeds once every deferred descendant's group closes). */ +static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, const char *name, + const char *rel, bool top_level, bool *deferred_out) { struct stat st; if (fstatat(parentfd, name, &st, AT_SYMLINK_NOFOLLOW) < 0) { @@ -235,11 +379,18 @@ static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, return 0; } if (!S_ISDIR(st.st_mode)) - return rm_tree(ctx, parentfd, name, rel, NULL); + return rm_tree(ctx, parentfd, name, rel, NULL, deferred_out); + + if (!top_level && ctx->budget <= 0) { + queue_delete_subdir(ctx, rel); + if (deferred_out) + *deferred_out = true; + return 0; + } uint64_t threshold = ctx->oe->o.delete_split_threshold; if (!threshold) - return rm_tree(ctx, parentfd, name, rel, NULL); + return rm_tree(ctx, parentfd, name, rel, NULL, deferred_out); int fd = openat(parentfd, name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); if (fd < 0) { @@ -283,6 +434,8 @@ static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, return 0; } stream_delete_split(ctx, rel, dupfd); + if (deferred_out) + *deferred_out = true; return 0; } /* Under threshold: reuse this same handle for the real removal — @@ -291,7 +444,7 @@ static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, * matters), so no second openat is needed on the common case, which now * runs at every depth instead of once per top-level orphan. */ rewinddir(d); - return rm_tree(ctx, parentfd, name, rel, d); + return rm_tree(ctx, parentfd, name, rel, d, deferred_out); } void process_delete(const struct shard_item *it) @@ -309,6 +462,14 @@ void process_delete(const struct shard_item *it) status = RES_TRANSIENT; goto out; } + /* Work budget for this shard (objects removed, decremented in rm_tree) — + * see the file-level comment's mechanism 2. Applies only to nested + * recursion (remove_object's top_level=false calls), never to this + * shard's own paths[] entries below, so a shard is never refused the + * work it was explicitly granted. */ + ctx.budget = (int64_t)(ctx.oe->o.delete_shard_budget + ? ctx.oe->o.delete_shard_budget + : DELETE_SHARD_BUDGET_DEFAULT); for (size_t i = 0; i < it->n_paths && !ctx.fatal; i++) { const char *rel = it->paths[i]; @@ -333,11 +494,15 @@ void process_delete(const struct shard_item *it) * 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_object(&ctx, pfd, leaf, rel)); + * onto the children instead of the walker that discovered them. + * top_level=true: this is a path this shard was explicitly granted, + * never itself deferred via the budget check (only nested + * subdirectories found during rm_dir_contents's descent are). */ + CTR_ADD(ctx.c.orphans, remove_object(&ctx, pfd, leaf, rel, true, NULL)); close(pfd); } + flush_delete_subdir_splits(&ctx); /* any budget-exhausted hand-offs still batched */ drain_splits(&ctx); /* every DeleteRemainder acked before the shard result (protocol §4.2) */ jrn_flush(&ctx); if (!jrn_wait_acked(&ctx)) { @@ -355,8 +520,11 @@ void process_delete(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, ctx.deferred, ctx.n_deferred); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); + for (size_t i = 0; i < ctx.n_deferred; i++) + free(ctx.deferred[i]); + free(ctx.deferred); jrn_destroy(&ctx); } diff --git a/agent/src/dirfix.c b/agent/src/dirfix.c index 387de54..89cfd3f 100644 --- a/agent/src/dirfix.c +++ b/agent/src/dirfix.c @@ -107,7 +107,7 @@ void process_dirfix(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); jrn_destroy(&ctx); diff --git a/agent/src/link.c b/agent/src/link.c index 554d68e..9a44bd1 100644 --- a/agent/src/link.c +++ b/agent/src/link.c @@ -161,7 +161,7 @@ void process_linkfix(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); jrn_destroy(&ctx); diff --git a/agent/src/main.c b/agent/src/main.c index 23cd97f..6281cd3 100644 --- a/agent/src/main.c +++ b/agent/src/main.c @@ -297,7 +297,7 @@ static void release_shard(struct shard_item *it) memset(&c, 0, sizeof c); pb_buf b; pb_init(&b); - enc_shard_result(&b, it->shard_id, it->lease_id, RES_RELEASED, &c, NULL); + enc_shard_result(&b, it->shard_id, it->lease_id, RES_RELEASED, &c, NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); shard_item_free(it); diff --git a/agent/src/msgs.c b/agent/src/msgs.c index 9b5f182..73f6eb8 100644 --- a/agent/src/msgs.c +++ b/agent/src/msgs.c @@ -138,6 +138,20 @@ void enc_delete_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, pb_free(&dr); } +void enc_delete_subdir_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, + char *const *subdirs, size_t n_subdirs) +{ + pb_put_u64(b, 1, parent_shard_id); + pb_put_u64(b, 2, seq); + for (size_t i = 0; i < n_subdirs; i++) { + pb_buf sub; + pb_init(&sub); + pb_put_bytes(&sub, 1, subdirs[i], strlen(subdirs[i])); + pb_put_msg(b, 8, &sub); + pb_free(&sub); + } +} + void enc_bigfile_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, const struct bigfile *files, size_t n_files) { @@ -204,7 +218,8 @@ static void enc_counters(pb_buf *b, uint32_t field, const struct shard_counters } void enc_shard_result(pb_buf *b, uint64_t shard_id, uint64_t lease_id, int status, - const struct shard_counters *c, const char *error) + const struct shard_counters *c, const char *error, + char *const *deferred_rmdirs, size_t n_deferred_rmdirs) { pb_put_u64(b, 1, shard_id); pb_put_u64(b, 2, lease_id); @@ -212,6 +227,8 @@ void enc_shard_result(pb_buf *b, uint64_t shard_id, uint64_t lease_id, int statu if (c) enc_counters(b, 4, c); pb_put_str(b, 5, error); + for (size_t i = 0; i < n_deferred_rmdirs; i++) + pb_put_bytes(b, 6, deferred_rmdirs[i], strlen(deferred_rmdirs[i])); } void enc_stats(pb_buf *b, const struct stats_snapshot *s) @@ -503,6 +520,7 @@ static bool dec_tuning_opts(const uint8_t *p, size_t n, struct job_options *o) 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; + case 9: o->delete_shard_budget = pb_get_varint(&c); break; default: pb_skip(&c, wt); } } diff --git a/agent/src/msgs.h b/agent/src/msgs.h index b05b216..7e0c271 100644 --- a/agent/src/msgs.h +++ b/agent/src/msgs.h @@ -149,6 +149,13 @@ struct job_options { * 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) */ + /* Work budget for one DELETE shard, in objects removed — the delete-pass + * analogue of shard_budget above, mirroring queue_split rather than + * delete_split_threshold/delete_split_batch: bounds total work per shard + * regardless of tree SHAPE, catching a tree pathological by aggregate + * depth/branching with no single directory individually over + * delete_split_threshold (docs/DESIGN-coordinator.md §2.2). */ + uint64_t delete_shard_budget; uint32_t statx_batch; /* target statx in flight per walker ⇒ io_uring ring depth */ int64_t mtime_slop_ns; bool dry_run; @@ -340,6 +347,15 @@ void enc_entrylist_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, 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 delete_subdirs: wholly UNOPENED subdirectories being + * handed off as their own new independent, top-level DELETE shards — the + * delete-pass analogue of enc_shard_split's subdirs above, not of + * enc_delete_split (see ShardSplit.delete_subdirs' doc comment: this bounds + * one shard's total work regardless of tree shape, catching a tree + * pathological by aggregate depth/branching with no single directory ever + * individually exceeding delete_split_threshold). */ +void enc_delete_subdir_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, + char *const *subdirs, size_t n_subdirs); /* ShardSplit carrying big files: rel_path + size + mtime_ns each. The * coordinator lays them out into ChunkTasks (proto ShardSplit.BigFile). */ struct bigfile { @@ -363,8 +379,12 @@ struct linksighting { }; void enc_linksighting_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq, const struct linksighting *sightings, size_t n_sightings); +/* deferred_rmdirs/n_deferred_rmdirs: DELETE shards only (delete.c) — pass + * NULL/0 for every other kind. See ShardResult.deferred_rmdirs' doc comment + * (proto/drsync.proto) for what these are. */ void enc_shard_result(pb_buf *b, uint64_t shard_id, uint64_t lease_id, int status, - const struct shard_counters *c, const char *error); + const struct shard_counters *c, const char *error, + char *const *deferred_rmdirs, size_t n_deferred_rmdirs); void enc_stats(pb_buf *b, const struct stats_snapshot *s); /* journal (docs/DESIGN-coordinator.md §5): records are varint-length- diff --git a/agent/src/probe.c b/agent/src/probe.c index b76b8af..4612c17 100644 --- a/agent/src/probe.c +++ b/agent/src/probe.c @@ -164,7 +164,7 @@ void process_probe(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &c, - err[0] ? err : NULL); + err[0] ? err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); } diff --git a/agent/src/verify.c b/agent/src/verify.c index 2300626..d6278a1 100644 --- a/agent/src/verify.c +++ b/agent/src/verify.c @@ -182,7 +182,7 @@ void process_verify(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); jrn_destroy(&ctx); diff --git a/agent/src/walker.c b/agent/src/walker.c index 7ed9e31..debde3c 100644 --- a/agent/src/walker.c +++ b/agent/src/walker.c @@ -1037,7 +1037,7 @@ void process_shard(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); free(ctx.split); @@ -1164,7 +1164,7 @@ void process_entrylist(const struct shard_item *it) pb_buf b; pb_init(&b); enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c, - ctx.err[0] ? ctx.err : NULL); + ctx.err[0] ? ctx.err : NULL, NULL, 0); out_push(FR_SHARD_RESULT, &b); lease_remove(it->lease_id); free(ctx.split); diff --git a/coordinator/internal/agentsrv/server.go b/coordinator/internal/agentsrv/server.go index 12d904e..a095fa9 100644 --- a/coordinator/internal/agentsrv/server.go +++ b/coordinator/internal/agentsrv/server.go @@ -520,10 +520,49 @@ 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)+len(sp.DeleteRemainders)) + shards := make([]store.NewShard, 0, len(sp.Subdirs)+len(sp.EntryLists)+ + len(sp.DeleteRemainders)+len(sp.DeleteSubdirs)) for _, d := range sp.Subdirs { shards = append(shards, store.NewShard{Kind: model.KindDir, RelPath: string(d.RelPath)}) } + // deleteTotals accumulates one DeleteGroupTotal per delete_groups-tracked + // completion this split produces — DeleteSubdirs below (a size-1 group + // per handoff) and DeleteRemainders further down (one entry per streamed + // batch) both feed it; RecordSplit processes the whole slice together. + var deleteTotals []store.DeleteGroupTotal + // A wholly unopened subdirectory the delete pass is handing off once its + // shard's work budget ran out (agent/src/delete.c, delete_shard_budget) — + // the delete-pass analogue of Subdirs above, NOT of DeleteRemainders: the + // receiving shard starts its own fresh budget and re-runs the same + // remove_object descent on d.RelPath from scratch, exactly like a + // top-level orphan from seedDeletePass. See ShardSplit.delete_subdirs' + // doc comment for why this exists alongside DeleteRemainder: that + // mechanism only catches a directory that is itself wide; this one bounds + // one shard's total work regardless of tree shape. + // + // RelPath is set to d.RelPath itself (unlike an ordinary seedDeletePass + // shard, whose RelPath is empty) so onShardResult can register this + // handoff against delete_groups: the parent shard that handed d.RelPath + // off does NOT wait for it (it has already moved on to its own + // siblings), so nothing stops it from calling rmdir on ITS OWN directory + // before d.RelPath has actually finished being removed elsewhere — the + // exact same completion-order gap DeleteRemainder had, just via a + // different hand-off path. Registering d.RelPath as a size-1, + // already-done-streaming delete_groups entry lets the SAME + // pending_children chain (registerPendingChildTx/closeDeleteGroupTx) + // hold the parent's own group open until this shard reports done. + for _, d := range sp.DeleteSubdirs { + payload, err := proto.Marshal(&drsyncpb.DeleteBatch{RelPaths: [][]byte{d.RelPath}}) + if err != nil { + return err + } + shards = append(shards, store.NewShard{ + Kind: model.KindDelete, RelPath: string(d.RelPath), Payload: payload}) + deleteTotals = append(deleteTotals, store.DeleteGroupTotal{ + DirRel: string(d.RelPath), LastBatch: true, NoSelfCleanup: true, + BuildCleanup: deleteCleanupShard, + }) + } for _, el := range sp.EntryLists { payload, err := proto.Marshal(&drsyncpb.EntryListShard{ DirRel: string(el.DirRel), Names: el.Names}) @@ -540,7 +579,6 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error { // 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 @@ -755,17 +793,37 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error { switch { case kind == model.KindChunk: err = s.completeChunk(passID, shardID, leaseID, payload, r) - 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 + case kind == model.KindDelete: + // EVERY KindDelete shard's completion goes through here now, not + // just split-produced ones (relPath != "") — a plain top-level + // orphan shard from seedDeletePass (relPath == "") can ALSO carry + // DeferredRmdirs: the production incident this fixed was exactly + // that shape, a top-level shard deferring its own path's rmdir + // because a budget handoff fired somewhere in its own recursion. + // + // relPath != "" means this shard is itself tracked in + // delete_groups — either a DeleteRemainder batch (onShardSplit + // sets RelPath to the directory it's emptying) or a budget + // handoff (ShardSplit.delete_subdirs, RelPath set to the + // handed-off directory itself). Maintain delete_groups the same // way completeChunk maintains chunk_groups, seeding a cleanup // shard for the now-possibly-empty directory once every sibling - // AND any nested child group (fan-out applies at every depth, not - // just relPath's own top level) has reported done. + // AND any nested child group (fan-out applies at every depth) + // has reported done. isHandoff distinguishes the two relPath!="" + // shapes: a budget handoff's own DeleteBatch is exactly one path + // equal to relPath itself (it already removed relPath as an + // ordinary top-level orphan, agent/src/delete.c remove_object + // top_level=true) — seeding a cleanup shard for it too would be + // a redundant rmdir. A DeleteRemainder batch's paths are always + // children UNDER dir_rel, never equal to it, so this check + // never misfires on the ordinary case. + var batch drsyncpb.DeleteBatch + isHandoff := relPath != "" && proto.Unmarshal(payload, &batch) == nil && + len(batch.RelPaths) == 1 && string(batch.RelPaths[0]) == relPath blob, _ := proto.Marshal(r) err = s.st.CompleteDeleteRemainder(shardID, leaseID, passID, relPath, blob, - deleteCleanupShard(relPath), deleteCleanupShard, r.Counters) + !isHandoff, deleteCleanupShard(relPath), deleteCleanupShard, + r.DeferredRmdirs, r.Counters) default: blob, _ := proto.Marshal(r) err = s.st.CompleteShard(shardID, leaseID, passID, blob, r.Counters) diff --git a/coordinator/internal/model/spec.go b/coordinator/internal/model/spec.go index b2e1021..3d846b5 100644 --- a/coordinator/internal/model/spec.go +++ b/coordinator/internal/model/spec.go @@ -138,6 +138,17 @@ type JobSpec struct { // differ. DeleteSplitThreshold uint64 `yaml:"delete_split_threshold"` DeleteSplitBatch uint32 `yaml:"delete_split_batch"` + // Work budget for one DELETE shard, in objects removed — the + // delete-pass analogue of ShardBudget above, mirroring the scan + // walker's queue_split rather than DeleteSplitThreshold/ + // DeleteSplitBatch: bounds one shard's total work regardless of + // tree SHAPE, catching a tree pathological by aggregate + // depth/branching even when no single directory anywhere in it + // individually exceeds DeleteSplitThreshold (docs/DESIGN- + // coordinator.md §2.2 — DeleteSplitThreshold alone cannot see + // this shape, since it only ever looks at one directory's own + // immediate entry count). + DeleteShardBudget uint64 `yaml:"delete_shard_budget"` // 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. @@ -271,6 +282,16 @@ func (s *JobSpec) ApplyDefaults() { if sp.Tuning.DeleteSplitBatch == 0 { sp.Tuning.DeleteSplitBatch = 20_000 } + // Delete-pass analogue of ShardBudget: bounds one shard's total work + // (objects removed) regardless of tree shape — catches a tree + // pathological by aggregate depth/branching that DeleteSplitThreshold + // alone cannot see, since that only ever checks one directory's own + // immediate entry count. Scaled up from ShardBudget's default the same + // way DeleteSplitBatch is scaled up from EntrylistBatch: delete work per + // object is a plain unlink, not a full stat/diff/copy pipeline. + if sp.Tuning.DeleteShardBudget == 0 { + sp.Tuning.DeleteShardBudget = 250_000 + } if sp.Tuning.StatxBatch == 0 { sp.Tuning.StatxBatch = 256 } @@ -456,6 +477,7 @@ func (s *JobSpec) ToJobOptions(jobID uint64, dryRun bool) (*drsyncpb.JobOptions, MtimeSlopNs: sp.Tuning.MtimeSlopNS, DeleteSplitThreshold: sp.Tuning.DeleteSplitThreshold, DeleteSplitBatch: sp.Tuning.DeleteSplitBatch, + DeleteShardBudget: sp.Tuning.DeleteShardBudget, }, 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 b8bc3ca..9decc58 100644 --- a/coordinator/internal/model/spec_defaults_test.go +++ b/coordinator/internal/model/spec_defaults_test.go @@ -42,6 +42,9 @@ func TestDefaultsAppliedToMinimalSpec(t *testing.T) { if sp.Tuning.DeleteSplitBatch != 20_000 { t.Errorf("tuning.delete_split_batch = %d, want 20000", sp.Tuning.DeleteSplitBatch) } + if sp.Tuning.DeleteShardBudget != 250_000 { + t.Errorf("tuning.delete_shard_budget = %d, want 250000", sp.Tuning.DeleteShardBudget) + } if sp.Metadata.Hardlinks != "preserve" { t.Errorf("metadata.hardlinks = %q, want preserve", sp.Metadata.Hardlinks) } @@ -61,9 +64,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) + if o.Tuning.DeleteSplitThreshold != 200_000 || o.Tuning.DeleteSplitBatch != 20_000 || + o.Tuning.DeleteShardBudget != 250_000 { + t.Errorf("resolved JobOptions delete tuning = %d/%d/%d, want 200000/20000/250000", + o.Tuning.DeleteSplitThreshold, o.Tuning.DeleteSplitBatch, o.Tuning.DeleteShardBudget) } } diff --git a/coordinator/internal/store/deletefanout_test.go b/coordinator/internal/store/deletefanout_test.go index bb96d73..ed28ea4 100644 --- a/coordinator/internal/store/deletefanout_test.go +++ b/coordinator/internal/store/deletefanout_test.go @@ -84,7 +84,7 @@ func TestDeleteGroupNestedChildBlocksParentClose(t *testing.T) { // (handed "child" off), so it reports done. This must NOT close parent's // group: parent/child's own group is still open (pending_children=1). if err := s.CompleteDeleteRemainder(parentBatchID, parentBatchLeaseID, passID, parent, nil, - cleanupShard(parent), cleanupShard, nil); err != nil { + true, cleanupShard(parent), cleanupShard, nil, nil); err != nil { t.Fatal(err) } counts, err := s.ShardStateCounts(passID) @@ -110,7 +110,7 @@ func TestDeleteGroupNestedChildBlocksParentClose(t *testing.T) { t.Fatalf("lease mismatch: leased=%v childIDs=%v", childLeased, childIDs) } if err := s.CompleteDeleteRemainder(childLeased[0].ID, childLeased[0].LeaseID, passID, child, nil, - cleanupShard(child), cleanupShard, nil); err != nil { + true, cleanupShard(child), cleanupShard, nil, nil); err != nil { t.Fatal(err) } counts, err = s.ShardStateCounts(passID) @@ -174,7 +174,7 @@ func TestDeleteGroupSeedsCleanupOnceAllChildrenDone(t *testing.T) { for i, r := range leased { if err := s.CompleteDeleteRemainder(r.ID, r.LeaseID, passID, dir, nil, - cleanupShard(dir), cleanupShard, nil); err != nil { + true, cleanupShard(dir), cleanupShard, nil, nil); err != nil { t.Fatalf("complete remainder %d: %v", i, err) } counts, err := s.ShardStateCounts(passID) @@ -240,7 +240,7 @@ func TestDeleteGroupHandlesChildCompletionRacingFinalBatch(t *testing.T) { // 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), cleanupShard, nil); err != nil { + true, cleanupShard(dir), cleanupShard, nil, nil); err != nil { t.Fatal(err) } counts, err := s.ShardStateCounts(passID) @@ -285,7 +285,7 @@ func TestDeleteGroupHandlesChildCompletionRacingFinalBatch(t *testing.T) { 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), cleanupShard, nil); err != nil { + true, cleanupShard(dir), cleanupShard, nil, nil); err != nil { t.Fatal(err) } counts, err = s.ShardStateCounts(passID) @@ -325,7 +325,7 @@ func TestDeleteGroupNeverSeedsCleanupTwice(t *testing.T) { 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), cleanupShard, nil); err != nil { + true, cleanupShard(dir), cleanupShard, nil, nil); err != nil { t.Fatal(err) } counts, err := s.ShardStateCounts(passID) diff --git a/coordinator/internal/store/store.go b/coordinator/internal/store/store.go index 2f715c9..b9f3391 100644 --- a/coordinator/internal/store/store.go +++ b/coordinator/internal/store/store.go @@ -1228,51 +1228,88 @@ func deleteGroupParent(relPath string) (string, bool) { return relPath[:i], true } -// registerPendingChildTx bumps parent's pending_children by one, but ONLY if -// parent itself has an open (existing, not-yet-closed) delete_groups row — -// an ordinary directory that merely happens to contain childRel is not -// itself being tracked, so there is nothing to chain to. Called once per -// child, guarded by the caller checking childIsNew (the child's own -// delete_groups row was just newly INSERTed, not already present) so a -// retried/redelivered split for the same child never double-counts. +// registerPendingChildTx bumps parent's pending_children by one, creating +// parent's delete_groups row first if it doesn't exist yet. It climbs +// further up PAST parent — through every ancestor of childRel, all the way +// to the top-level orphan path itself (deleteGroupParent returns ok=false +// there — no slash left) — but ONLY as long as each level's row is being +// created for the FIRST time by this very call; the moment a level's row +// already existed, that level is already correctly tracking exactly one +// pending child (whichever descendant triggered ITS OWN creation earlier), +// and the climb stops there — bumping any level ABOVE an already-existing +// row would double (or N-times) count every later, unrelated handoff +// elsewhere under that same already-tracked ancestor. This matters for a +// directory being processed INLINE by rm_tree's own recursion (never itself +// handed off via queue_delete_subdir or DeleteRemainder — it's simply too +// small to trigger either mechanism on its own): such a directory has NO +// delete_groups row until something ELSEWHERE in its subtree gets handed +// off, at which point this walk retroactively creates one for it (and, only +// the FIRST time, everything above it up to the top-level orphan) so +// rm_tree can later find it and defer that directory's own rmdir until the +// handoff's group closes — see rm_tree's own pending-check before its +// rmdir call. Each newly-created row along the way is marked +// done_streaming=1, n_total=1, n_done=0 immediately: unlike a +// DeleteRemainder group (many sibling batches) or a budget-handoff leaf (one +// shard, already known), an inline ancestor's "n_total" is always exactly +// one thing — its own rm_tree call finishing — so there is nothing further +// to stream in; n_done is bumped later by CompleteDeleteRemainder processing +// the containing shard's own ShardResult.deferred_rmdirs. func registerPendingChildTx(tx *sql.Tx, passID int64, childRel string) error { - parent, ok := deleteGroupParent(childRel) - if !ok { - return nil - } - n, err := execCountTx(tx, `UPDATE delete_groups SET pending_children = pending_children + 1 - WHERE pass_id = ? AND rel_path = ? AND closed = 0`, passID, parent) - if err != nil { - return err - } - // n == 0: parent has no open delete_groups row (not itself a fan-out - // group — e.g. childRel's immediate parent happens to be a plain - // directory name that never itself split) — nothing to chain to. - // Deliberately not recursive beyond one level: registerPendingChildTx - // only links a child to its DIRECT parent's row; if that parent is - // itself someone else's pending child, ITS OWN closing (checked by - // closeDeleteGroupTx below, which walks back up via deleteGroupParent) - // is what propagates the chain further up, one link at a time. - _ = n - return nil -} - -// closeDeleteGroupTx marks relPath's delete_groups row closed and inserts -// cleanupShard — called once the row's own counters prove it closeable -// (done_streaming=1, n_done>=n_total, pending_children=0). If relPath has a -// parent row, decrements ITS pending_children and recursively re-checks -// whether that unblocks the parent's own closure too — the chain can be -// arbitrarily deep (a pathological directory inside a pathological -// directory inside...), so this walks all the way up rather than handling -// just one level. buildCleanup builds a cleanup NewShard for any dirRel -// string (store stays payload-agnostic — same division of labor as -// CompleteDataChunk's finalizeShard — but an ancestor's own cleanup shard is -// only discovered while walking up the chain here, so the caller hands in a -// builder function instead of one pre-built shard). -func closeDeleteGroupTx(tx *sql.Tx, passID int64, relPath string, cleanupShard NewShard, - buildCleanup func(dirRel string) NewShard) error { - if _, err := insertShardsTx(tx, passID, 0, []NewShard{cleanupShard}); err != nil { - return err + for { + parent, ok := deleteGroupParent(childRel) + if !ok { + return nil + } + n, err := execCountTx(tx, `INSERT OR IGNORE INTO delete_groups + (pass_id, rel_path, n_total, n_done, done_streaming) VALUES (?, ?, 1, 0, 1)`, + passID, parent) + if err != nil { + return err + } + if _, err := tx.Exec(`UPDATE delete_groups SET pending_children = pending_children + 1 + WHERE pass_id = ? AND rel_path = ? AND closed = 0`, passID, parent); err != nil { + return err + } + if n == 0 { + // parent's row already existed — it's already tracking exactly + // one pending child from whatever handoff first created it; + // stop here, do NOT also bump anything further up the chain for + // this (different, later, unrelated) handoff. + return nil + } + childRel = parent + } +} + +// closeDeleteGroupTx marks relPath's delete_groups row closed — called once +// the row's own counters prove it closeable (done_streaming=1, +// n_done>=n_total, pending_children=0). If seedCleanup, also inserts +// cleanupShard: true for an ordinary DeleteRemainder group (individual +// batch shards only ever remove entries INSIDE relPath, never relPath +// itself — something has to). false for a budget-handoff group +// (queue_delete_subdir/ShardSplit.delete_subdirs): the single shard that +// makes up that "group" already IS a top-level orphan removal that removes +// relPath itself as part of its own ordinary completion (agent/src/ +// delete.c remove_object with top_level=true), so a second cleanup shard +// here would be a redundant rmdir of something already gone. If relPath has +// a parent row, decrements ITS pending_children and recursively re-checks +// whether that unblocks the parent's own closure too (ALWAYS with +// seedCleanup=true from here on up — every ancestor in the chain is an +// ordinary DeleteRemainder-tracked directory, only the leaf that started +// the chain can be a budget handoff) — the chain can be arbitrarily deep +// (a pathological directory inside a pathological directory inside...), so +// this walks all the way up rather than handling just one level. +// buildCleanup builds a cleanup NewShard for any dirRel string (store stays +// payload-agnostic — same division of labor as CompleteDataChunk's +// finalizeShard — but an ancestor's own cleanup shard is only discovered +// while walking up the chain here, so the caller hands in a builder +// function instead of one pre-built shard). +func closeDeleteGroupTx(tx *sql.Tx, passID int64, relPath string, seedCleanup bool, + cleanupShard NewShard, buildCleanup func(dirRel string) NewShard) error { + if seedCleanup { + 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 { @@ -1296,7 +1333,11 @@ func closeDeleteGroupTx(tx *sql.Tx, passID int64, relPath string, cleanupShard N if pClosed != 0 || pDoneStreaming == 0 || pDone < pTotal || pPending > 0 { return nil // parent not closeable yet from this side either } - return closeDeleteGroupTx(tx, passID, parent, buildCleanup(parent), buildCleanup) + // Every ancestor walked to from here is an ordinary DeleteRemainder- + // tracked directory (only the LEAF that started this call can be a + // budget-handoff group — see this function's doc comment), so its own + // cleanup shard is always needed: seedCleanup=true unconditionally. + return closeDeleteGroupTx(tx, passID, parent, true, buildCleanup(parent), buildCleanup) } // DeleteGroupTotal records one delete-remainder batch shipped for DirRel — @@ -1309,14 +1350,24 @@ func closeDeleteGroupTx(tx *sql.Tx, passID int64, relPath string, cleanupShard N // NewShard for DirRel — and, via closeDeleteGroupTx's upward walk, for any // ancestor directory this closure unblocks — without store itself knowing // the DeleteBatch payload shape (same division of labor as -// CompleteDataChunk's finalizeShard). Only actually used once every sibling -// delete-remainder shard has reported done AND no pending nested child group -// remains open, by the time LastBatch lands — the same race -// CompleteDeleteRemainder resolves from the other direction. +// CompleteDataChunk's finalizeShard). NoSelfCleanup skips seeding a cleanup +// shard for DirRel ITSELF when this group closes (an ancestor this closure +// unblocks still gets its own cleanup seeded regardless — only DirRel's own +// is skipped): set for a budget-handoff group (ShardSplit.delete_subdirs), +// whose single shard already removes DirRel itself as an ordinary top-level +// orphan (agent/src/delete.c remove_object, top_level=true) — a cleanup +// shard here would be a redundant rmdir of something already gone. Left +// false (the default, seed the cleanup) for an ordinary DeleteRemainder +// group, where individual batch shards only ever remove entries INSIDE +// DirRel, never DirRel itself. Only actually used once every sibling has +// reported done AND no pending nested child group remains open, by the time +// LastBatch lands — the same race CompleteDeleteRemainder resolves from the +// other direction. type DeleteGroupTotal struct { - DirRel string - LastBatch bool - BuildCleanup func(dirRel string) NewShard + DirRel string + LastBatch bool + NoSelfCleanup bool + BuildCleanup func(dirRel string) NewShard } // RecordSplit persists a ShardSplit idempotently: retransmits of the same @@ -1457,7 +1508,8 @@ func (s *Store) RecordSplit(parentShardID int64, seq uint64, shards []NewShard, if closed != 0 || doneStreaming == 0 || nDone < nTotal || pending > 0 { continue // streaming not finished, siblings/nested children outstanding, or already closed } - if err := closeDeleteGroupTx(tx, passID, dt.DirRel, dt.BuildCleanup(dt.DirRel), dt.BuildCleanup); err != nil { + if err := closeDeleteGroupTx(tx, passID, dt.DirRel, !dt.NoSelfCleanup, + dt.BuildCleanup(dt.DirRel), dt.BuildCleanup); err != nil { return nil, err } } @@ -1677,18 +1729,37 @@ func (s *Store) CompleteDataChunk(shardID, leaseID, passID int64, relPath string // directory a removal touches, not just relPath itself — a name in one of // relPath's own batches can turn out to be pathological and get streamed off // as its own delete_groups row; relPath cannot close until that grandchild -// closes too, see delete_groups' doc comment), 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, then -// walks back up the chain (closeDeleteGroupTx) in case this was itself the -// last blocker on relPath's own parent. buildCleanup builds a cleanup -// NewShard for any dirRel string, used for any ancestor closeDeleteGroupTx's -// upward walk unblocks. counters folds the pass-counter accumulation into -// this same transaction — see CompleteShard's doc comment for why. +// closes too, see delete_groups' doc comment), this marks the group closed +// (so a re-delivered final result can never seed a cleanup twice) and, if +// seedCleanup, 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). seedCleanup is false for a +// budget-handoff group (ShardSplit.delete_subdirs — see DeleteGroupTotal. +// NoSelfCleanup): the single shard that makes up that group already removed +// relPath itself as an ordinary top-level orphan, so a cleanup shard here +// would be a redundant rmdir. Then walks back up the chain +// (closeDeleteGroupTx) in case this was itself the last blocker on relPath's +// own parent — every ancestor found that way always gets its own cleanup +// seeded (only a chain's own leaf can ever be a no-self-cleanup budget +// handoff). buildCleanup builds a cleanup NewShard for any dirRel string, +// used for any ancestor closeDeleteGroupTx's upward walk unblocks. counters +// folds the pass-counter accumulation into this same transaction — see +// CompleteShard's doc comment for why. deferredRmdirs (ShardResult. +// deferred_rmdirs) are relative paths of directories THIS SAME shard +// processed inline but deliberately did not rmdir, because a descendant was +// handed off during its own run (agent/src/delete.c rm_tree/ +// queue_deferred_rmdir) — each is closed via bumpAndMaybeCloseDeleteGroupTx +// in this same transaction, regardless of whether relPath itself is "" (an +// ordinary top-level orphan shard from seedDeletePass — deferred_rmdirs is +// exactly how the bug in a real production incident was fixed for THAT +// case: the shard granted a whole orphan tree can itself defer its own +// top-level path's rmdir, same as any inline ancestor deeper in its +// recursion) or non-empty (this shard is itself a DeleteRemainder batch or +// budget-handoff leaf, tracked via relPath's own group below). func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath string, result []byte, - cleanupShard NewShard, buildCleanup func(dirRel string) NewShard, counters *drsyncpb.ShardCounters) error { + seedCleanup bool, cleanupShard NewShard, buildCleanup func(dirRel string) NewShard, + deferredRmdirs [][]byte, counters *drsyncpb.ShardCounters) error { defer s.lockTimed("CompleteDeleteRemainder")() tx, err := s.db.Begin() if err != nil { @@ -1709,24 +1780,59 @@ func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath 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 THIS - // call is what creates the row, register it as relPath's parent's - // pending child too (registerPendingChildTx) — same as RecordSplit does - // for the ordinary case, just reached from the other side of the race. - n, err = execCountTx(tx, `INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, - passID, relPath) - if err != nil { - return err + if relPath != "" { + // 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 THIS + // call is what creates the row, register it as relPath's parent's + // pending child too (registerPendingChildTx) — same as RecordSplit does + // for the ordinary case, just reached from the other side of the race. + n, err = execCountTx(tx, `INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, + passID, relPath) + if err != nil { + return err + } + if n > 0 { + if err := registerPendingChildTx(tx, passID, relPath); err != nil { + return err + } + } + if err := bumpAndMaybeCloseDeleteGroupTx(tx, passID, relPath, seedCleanup, + cleanupShard, buildCleanup); err != nil { + return err + } } - if n > 0 { - if err := registerPendingChildTx(tx, passID, relPath); err != nil { + for _, dr := range deferredRmdirs { + // A directory THIS shard processed inline but did not rmdir — its + // delete_groups row already exists (registerPendingChildTx created it + // retroactively, n_total=1/n_done=0, the moment some descendant of it + // was first handed off) unless this ShardResult somehow arrives before + // any of that registration ever landed, which INSERT OR IGNORE below + // makes safe either way. seedCleanup=true always: this directory + // genuinely still needs its rmdir performed by someone (unlike a + // budget-handoff leaf, nobody has removed it yet). + if _, err := tx.Exec(`INSERT OR IGNORE INTO delete_groups + (pass_id, rel_path, n_total, n_done, done_streaming) VALUES (?, ?, 1, 0, 1)`, + passID, string(dr)); err != nil { + return err + } + if err := bumpAndMaybeCloseDeleteGroupTx(tx, passID, string(dr), true, + buildCleanup(string(dr)), buildCleanup); err != nil { return err } } + return tx.Commit() +} + +// bumpAndMaybeCloseDeleteGroupTx increments relPath's delete_groups n_done +// by one and, if that makes the row closeable (done_streaming=1, +// n_done>=n_total, pending_children=0), closes it (closeDeleteGroupTx) — +// the shared body CompleteDeleteRemainder uses both for relPath's own group +// (when relPath != "") and for each of a shard's deferredRmdirs entries. +func bumpAndMaybeCloseDeleteGroupTx(tx *sql.Tx, passID int64, relPath string, seedCleanup bool, + cleanupShard NewShard, buildCleanup func(dirRel string) NewShard) error { var nTotal, nDone, doneStreaming, pending, 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, pending_children, closed`, @@ -1734,12 +1840,9 @@ func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath return err } if closed != 0 || doneStreaming == 0 || nDone < nTotal || pending > 0 { - return tx.Commit() // already closed, streaming not finished, or siblings/nested children outstanding + return nil // already closed, streaming not finished, or siblings/nested children outstanding } - if err := closeDeleteGroupTx(tx, passID, relPath, cleanupShard, buildCleanup); err != nil { - return err - } - return tx.Commit() + return closeDeleteGroupTx(tx, passID, relPath, seedCleanup, cleanupShard, buildCleanup) } // CompleteFinalizeChunk marks the finalize shard DONE and closes its group. diff --git a/docs/ADMIN.md b/docs/ADMIN.md index 4171ee3..49de78a 100644 --- a/docs/ADMIN.md +++ b/docs/ADMIN.md @@ -506,6 +506,24 @@ 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. +**A deeply-branching orphan tree fans out too, even if no single directory +is ever individually large.** `delete_split_threshold` only catches a +directory that is itself WIDE. A tree that's pathological purely by +aggregate depth/branching — many subdirectories, each individually well +under the threshold, several levels deep — never trips that check +anywhere, and would otherwise run serially inside one shard no matter how +many millions of files it adds up to. `tuning.delete_shard_budget` (default +250 000, objects removed) bounds the total work any ONE delete shard does +regardless of tree shape: once a shard has removed that many objects, every +subdirectory it hasn't opened yet is handed off as its own new shard +instead of being recursed into, the same way `shard_budget` bounds the scan +walker. + +```bash +drsync job submit branchy-orphans.yaml --start \ + --set spec.tuning.delete_shard_budget=50000 +``` + --- ## 6. Monitoring diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index eb7d9b7..6ee3ae6 100644 --- a/docs/DESIGN-coordinator.md +++ b/docs/DESIGN-coordinator.md @@ -148,16 +148,14 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards the top-level orphan path — `rm_dir_contents` routes every entry it finds during descent back through `remove_object` rather than recursing into it directly, so a subdirectory discovered at any depth that turns out to be - pathological is streamed out too, not just removed serially. This matters - because "pathological" is not the same shape as "large": a real incident - hit a destination-only orphan tree where NO single directory (not the top - orphan path, not any one subdirectory) individually exceeded the - threshold, but the tree as a whole summed to ~14M files and directories — - the delete pass finished all its other work and then sat on 2 remaining - shards for 6 hours, each depth-first-recursing an enormous subtree with no - fan-out at all, because the single-level-only check never had a reason to - fire anywhere in that tree — see the "found in production" note below for - the fix and the completion-tracking wrinkle it introduced. + pathological is streamed out too, not just removed serially. But this + mechanism only ever looks at ONE directory's own immediate entry count — + it catches a WIDE directory at any depth, not a tree that is pathological + by aggregate depth/branching with no single directory anywhere in it ever + individually wide. A second, independent mechanism (`tuning. + delete_shard_budget`, below) was added for that shape after it hit + production — see the "found in production" notes further down for both + the incident and the three-round fix. 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 @@ -209,35 +207,110 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards directory failed to disappear from disk. `TestDeleteRemainderPathsJoinDirRel` (`agentsrv/server_test.go`) now pins the join at the `onShardSplit` level. - **Found in production, after the above shipped:** the "2 shards, 14M - files, 6 hours" incident above. Extending the pathological check to every - depth (`remove_object`/`rm_dir_contents`, not just the shard's own - top-level path) fixed the fan-out itself, but exposed a completion-order - bug the flat case structurally could not: a batch shard under `dir_rel` - that ships a name which turns out to be itself pathological reports - itself *done* the instant it hands that name off (correctly — the batch - has no more work of its own), which increments `dir_rel`'s `n_done` — - but the handed-off name is not yet actually removed, only queued as its - OWN separate `delete_groups` row. Before the fix, `dir_rel`'s group could - therefore close and seed its cleanup `rmdir` while a nested child was - still mid-removal: `delete_fanout_nested_e2e.sh` (built specifically - because the existing flat-directory e2e test structurally cannot exercise - this — no single directory in it is ever handed off to a nested group) - caught both a double-removed path (a race where the same name was - streamed twice) and a permanently-orphaned empty directory (the parent's - `rmdir` either never ran or ran too early and something after it - recreated nothing — the parent just silently never got removed). Fixed - with a `pending_children` counter on `delete_groups` (§3): - `registerPendingChildTx` bumps a parent's `pending_children` the first - time a nested child row is created under it (derived purely from the - child's own `rel_path` string — no explicit parent pointer needed, since - `orphandir/sub0000`'s parent is unambiguously `orphandir`); closing a - child's row (`closeDeleteGroupTx`) decrements the parent's count and + **Found in production, round 1 (nested WIDE completion race).** Extending + the pathological check to every depth (`remove_object`/`rm_dir_contents`, + not just the shard's own top-level path) fixed the fan-out itself, but + exposed a completion-order bug the flat case structurally could not: a + batch shard under `dir_rel` that ships a name which turns out to be itself + pathological reports itself *done* the instant it hands that name off + (correctly — the batch has no more work of its own), which increments + `dir_rel`'s `n_done` — but the handed-off name is not yet actually + removed, only queued as its OWN separate `delete_groups` row. Before the + fix, `dir_rel`'s group could therefore close and seed its cleanup `rmdir` + while a nested child was still mid-removal: `delete_fanout_nested_e2e.sh` + (built specifically because the existing flat-directory e2e test + structurally cannot exercise this — no single directory in it is ever + handed off to a nested group) caught both a double-removed path (a race + where the same name was streamed twice) and a permanently-orphaned empty + directory. Fixed with a `pending_children` counter on `delete_groups` + (§3): `registerPendingChildTx` bumps a parent's `pending_children` the + first time a nested child row is created under it (derived purely from + the child's own `rel_path` string — no explicit parent pointer needed, + since `orphandir/sub0000`'s parent is unambiguously `orphandir`); closing + a child's row (`closeDeleteGroupTx`) decrements the parent's count and walks back up recursively, re-checking the parent's own closeability from - the other side — the same both-directions-check pattern `n_done`/ - `done_streaming` already use, just one more hop, and applied at every - level so a pathological directory nested inside a pathological directory - inside another still closes bottom-up correctly. + the other side. + + **Found in production, round 2 (the actual "2 shards, 14M files, 6 hours" + incident).** A real orphan tree hit the shape mechanism 1 fundamentally + cannot see: 77 top-level branches, each several levels deep, with every + individual directory well under any reasonable `delete_split_threshold` — + no directory anywhere in the tree was ever WIDE, so `remove_object`'s + probe never once tripped, and the whole multi-million-object tree ran + serially inside 2 shards. This needed a second, independent mechanism: + **`tuning.delete_shard_budget`** (`agent/src/delete.c`, mirroring the scan + walker's `shard_budget`/`queue_split`) bounds the total number of objects + ONE shard removes, regardless of tree shape. `walk_ctx.budget` is + decremented once per object actually removed (`rm_tree`), threaded + through the whole recursive descent exactly like the walker's own budget. + Once it reaches zero, every not-yet-opened subdirectory `rm_dir_contents` + finds is handed off UNOPENED as its own brand-new top-level DELETE shard + (`queue_delete_subdir` → `ShardSplit.delete_subdirs`, wire field 8 — + deliberately NOT the walker's own `subdirs` field 3, since that always + becomes a `KindDir` walk/diff shard, wrong kind entirely for a directory + that's already condemned) instead of being recursed into — the receiving + shard starts its own fresh budget and re-runs `remove_object` on it from + scratch, same as any top-level orphan from `seedDeletePass`. + `delete_fanout_budget_e2e.sh` (a 10×10 branching tree, 5 files per leaf, + every directory capped at 10 entries — deliberately far under any + threshold mechanism 1 would ever trip) reproduces this shape directly. + + **Found in production, round 3 (two more completion bugs the budget + mechanism itself introduced, both caught locally before reaching CI via + the new e2e script).** A budget handoff creates the SAME completion-order + problem round 1 fixed for the WIDE case, but in a shape the + `pending_children` chain as first built could not yet handle, because a + handed-off directory has no *sibling batches* the way a `DeleteRemainder` + group does — it is exactly one shard, already fully known: + 1. The handed-off shard's own `DeleteBatch` (built in `onShardSplit`) is + registered as a size-one `delete_groups` row (`n_total=1, + done_streaming=1`) via the SAME `DeleteGroupTotal` plumbing + `DeleteRemainder` batches use, but with a new `NoSelfCleanup` flag: the + handoff shard's own completion (an ordinary top-level orphan removal, + `remove_object` with `top_level=true`) already removes the directory + itself, so `closeDeleteGroupTx` must NOT also seed a redundant cleanup + `rmdir` for it the way it does for an ordinary `DeleteRemainder` group + — only ancestors *above* a budget-handoff leaf, discovered while + walking the closing chain upward, ever get their own cleanup seeded. + 2. Once (1) was fixed, a DIFFERENT directory — one that was never itself + handed off, but merely an ancestor of one deep in an otherwise-inline + `rm_tree` recursion (e.g. `orphandir/b00`, whose own `cNN` children got + handed off but `b00` itself stayed under budget and was processed + inline) — was left behind, empty, forever. `rm_tree` had no way to + know a descendant was mid-removal elsewhere, so it happily `rmdir`'d + itself the instant its own (locally visible) contents were gone. Fixed + by having the AGENT track this locally, no coordinator round-trip + needed: when a handoff fires anywhere inside an in-progress `rm_tree` + call, that fact propagates back up through the C call stack + (`*deferred_out`, threaded through `remove_object`/`rm_dir_contents`/ + `rm_tree`) — a directory that sees a deferred descendant skips its own + `rmdir` too (`queue_deferred_rmdir`, recorded in + `walk_ctx.deferred[]`) and propagates the same signal to whoever is + recursing into IT. The accumulated list ships once, attached to the + shard's own final result (`ShardResult.deferred_rmdirs`, proto field + 6) — not a separate streamed message, since the coordinator only needs + it once the whole shard is done. Coordinator-side, + `registerPendingChildTx` now walks EVERY ancestor of a handoff (not + just the direct parent) up to the top-level orphan path, retroactively + creating a `delete_groups` row for each ancestor the FIRST time + anything beneath it is handed off (`n_total=1, done_streaming=1`, + mirroring a budget-handoff leaf's own shape: there is exactly one + thing to wait for — this ancestor's own containing shard eventually + checking in via `deferred_rmdirs`) — critically, the climb STOPS the + moment it reaches a level that already had a row, or every later, + unrelated handoff under an already-tracked ancestor would double (or + N-times) count that ancestor's `pending_children` (an actual bug hit + and fixed during this same investigation: `orphandir`'s own + `pending_children` reached 69 instead of 10 before this stop condition + was added, since every one of the ~90 leaf handoffs was climbing all + the way to the top and bumping it again). `CompleteDeleteRemainder` + processes a completed shard's own `deferred_rmdirs` list in the same + transaction as its ordinary completion, via a shared + `bumpAndMaybeCloseDeleteGroupTx` helper — bumping `n_done` to 1 (always + `seedCleanup=true`: unlike a budget-handoff leaf, nobody has removed + this directory yet, it was deliberately skipped) and walking the + closing chain upward exactly like every other path into + `closeDeleteGroupTx`. ### 2.3 Shard diff --git a/proto/drsync.proto b/proto/drsync.proto index 0fb24ab..90eb594 100644 --- a/proto/drsync.proto +++ b/proto/drsync.proto @@ -283,6 +283,17 @@ message TuningOptions { // 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) + // Work budget for one DELETE shard, in objects removed — the delete-pass + // analogue of shard_budget (field 1) above, mirroring the scan walker's + // queue_split rather than delete_split_threshold/entrylist_batch: bounds + // the total work any one shard does regardless of tree SHAPE, catching a + // tree that is pathological by aggregate depth/branching even when no + // single directory anywhere in it individually exceeds + // delete_split_threshold (docs/DESIGN-coordinator.md §2.2). Once exhausted + // mid-descent, every not-yet-opened subdirectory still queued for removal + // is handed off as its own new top-level DELETE shard + // (ShardSplit.delete_subdirs) instead of being recursed into inline. + uint64 delete_shard_budget = 9; } message JobOptions { @@ -547,6 +558,20 @@ message ShardSplit { repeated BigFile big_files = 5; repeated LinkSighting link_sightings = 6; repeated DeleteRemainder delete_remainders = 7; + // A wholly UNOPENED subdirectory the delete pass is handing off as its own + // independent, top-level DELETE shard once this shard's work budget + // (tuning.delete_shard_budget, agent/src/delete.c) runs out mid-descent — + // the delete-pass analogue of the scan walker's queue_split/subdirs above, + // not of DeleteRemainder: DeleteRemainder streams a directory's OWN entries + // once that one directory looks pathological by its immediate entry count; + // this instead bounds the total work any ONE delete shard does regardless + // of tree shape, catching a tree that is pathological by aggregate + // depth/branching with no single directory (at any level) ever individually + // exceeding delete_split_threshold. rel_path reuses NewShard's shape (a + // bare relative path, nothing else needed — the receiving shard starts a + // fresh budget and re-runs the same remove_object/rm_dir_contents descent + // from scratch on it, exactly like a top-level orphan from seedDeletePass). + repeated NewShard delete_subdirs = 8; } message ShardSplitAck { @@ -603,6 +628,21 @@ message ShardResult { ResultStatus status = 3; ShardCounters counters = 4; string error = 5; + // DELETE shards only (agent/src/delete.c): relative paths of directories + // this shard processed INLINE (rm_tree, never handed off via + // queue_delete_subdir or stream_delete_split — individually too small to + // trigger either) whose own rmdir it deliberately skipped, because a + // descendant somewhere beneath it WAS handed off during this shard's run + // and therefore might still be in flight elsewhere. Every directory's + // contents are still fully removed by the time it lands here — only the + // directory's own rmdir is deferred. The coordinator's delete_groups + // tracking (registerPendingChildTx creates an ancestor row for exactly + // this case) treats each path here as that ancestor's own work finishing + // (n_done=1) — its cleanup shard (the actual deferred rmdir) is seeded + // once BOTH this lands AND every descendant handoff's own group has + // closed (pending_children=0), same closeDeleteGroupTx chain already + // used for every other delete-fan-out completion path. + repeated bytes deferred_rmdirs = 6; } message TaskResult { diff --git a/proto/gen/drsyncpb/drsync.pb.go b/proto/gen/drsyncpb/drsync.pb.go index 4d31e12..704bc9e 100644 --- a/proto/gen/drsyncpb/drsync.pb.go +++ b/proto/gen/drsyncpb/drsync.pb.go @@ -1953,8 +1953,19 @@ type TuningOptions struct { // 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 + // Work budget for one DELETE shard, in objects removed — the delete-pass + // analogue of shard_budget (field 1) above, mirroring the scan walker's + // queue_split rather than delete_split_threshold/entrylist_batch: bounds + // the total work any one shard does regardless of tree SHAPE, catching a + // tree that is pathological by aggregate depth/branching even when no + // single directory anywhere in it individually exceeds + // delete_split_threshold (docs/DESIGN-coordinator.md §2.2). Once exhausted + // mid-descent, every not-yet-opened subdirectory still queued for removal + // is handed off as its own new top-level DELETE shard + // (ShardSplit.delete_subdirs) instead of being recursed into inline. + DeleteShardBudget uint64 `protobuf:"varint,9,opt,name=delete_shard_budget,json=deleteShardBudget,proto3" json:"delete_shard_budget,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *TuningOptions) Reset() { @@ -2043,6 +2054,13 @@ func (x *TuningOptions) GetDeleteSplitBatch() uint32 { return 0 } +func (x *TuningOptions) GetDeleteShardBudget() uint64 { + if x != nil { + return x.DeleteShardBudget + } + 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"` @@ -3546,8 +3564,22 @@ type ShardSplit struct { 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 + // A wholly UNOPENED subdirectory the delete pass is handing off as its own + // independent, top-level DELETE shard once this shard's work budget + // (tuning.delete_shard_budget, agent/src/delete.c) runs out mid-descent — + // the delete-pass analogue of the scan walker's queue_split/subdirs above, + // not of DeleteRemainder: DeleteRemainder streams a directory's OWN entries + // once that one directory looks pathological by its immediate entry count; + // this instead bounds the total work any ONE delete shard does regardless + // of tree shape, catching a tree that is pathological by aggregate + // depth/branching with no single directory (at any level) ever individually + // exceeding delete_split_threshold. rel_path reuses NewShard's shape (a + // bare relative path, nothing else needed — the receiving shard starts a + // fresh budget and re-runs the same remove_object/rm_dir_contents descent + // from scratch on it, exactly like a top-level orphan from seedDeletePass). + DeleteSubdirs []*ShardSplit_NewShard `protobuf:"bytes,8,rep,name=delete_subdirs,json=deleteSubdirs,proto3" json:"delete_subdirs,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ShardSplit) Reset() { @@ -3629,6 +3661,13 @@ func (x *ShardSplit) GetDeleteRemainders() []*ShardSplit_DeleteRemainder { return nil } +func (x *ShardSplit) GetDeleteSubdirs() []*ShardSplit_NewShard { + if x != nil { + return x.DeleteSubdirs + } + 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"` @@ -3882,14 +3921,29 @@ func (x *ShardCounters) GetLinkFallback() uint64 { } type ShardResult struct { - state protoimpl.MessageState `protogen:"open.v1"` - ShardId uint64 `protobuf:"varint,1,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"` - LeaseId uint64 `protobuf:"varint,2,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` - Status ResultStatus `protobuf:"varint,3,opt,name=status,proto3,enum=drsync.v1.ResultStatus" json:"status,omitempty"` - Counters *ShardCounters `protobuf:"bytes,4,opt,name=counters,proto3" json:"counters,omitempty"` - Error string `protobuf:"bytes,5,opt,name=error,proto3" json:"error,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + ShardId uint64 `protobuf:"varint,1,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"` + LeaseId uint64 `protobuf:"varint,2,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` + Status ResultStatus `protobuf:"varint,3,opt,name=status,proto3,enum=drsync.v1.ResultStatus" json:"status,omitempty"` + Counters *ShardCounters `protobuf:"bytes,4,opt,name=counters,proto3" json:"counters,omitempty"` + Error string `protobuf:"bytes,5,opt,name=error,proto3" json:"error,omitempty"` + // DELETE shards only (agent/src/delete.c): relative paths of directories + // this shard processed INLINE (rm_tree, never handed off via + // queue_delete_subdir or stream_delete_split — individually too small to + // trigger either) whose own rmdir it deliberately skipped, because a + // descendant somewhere beneath it WAS handed off during this shard's run + // and therefore might still be in flight elsewhere. Every directory's + // contents are still fully removed by the time it lands here — only the + // directory's own rmdir is deferred. The coordinator's delete_groups + // tracking (registerPendingChildTx creates an ancestor row for exactly + // this case) treats each path here as that ancestor's own work finishing + // (n_done=1) — its cleanup shard (the actual deferred rmdir) is seeded + // once BOTH this lands AND every descendant handoff's own group has + // closed (pending_children=0), same closeDeleteGroupTx chain already + // used for every other delete-fan-out completion path. + DeferredRmdirs [][]byte `protobuf:"bytes,6,rep,name=deferred_rmdirs,json=deferredRmdirs,proto3" json:"deferred_rmdirs,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ShardResult) Reset() { @@ -3957,6 +4011,13 @@ func (x *ShardResult) GetError() string { return "" } +func (x *ShardResult) GetDeferredRmdirs() [][]byte { + if x != nil { + return x.DeferredRmdirs + } + return nil +} + type TaskResult struct { state protoimpl.MessageState `protogen:"open.v1"` TaskId uint64 `protobuf:"varint,1,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"` @@ -5068,7 +5129,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\"\xd8\x02\n" + + "\x0eiops_per_agent\x18\x02 \x01(\x04R\fiopsPerAgent\"\x88\x03\n" + "\rTuningOptions\x12!\n" + "\fshard_budget\x18\x01 \x01(\x04R\vshardBudget\x12.\n" + "\x13dir_split_threshold\x18\x02 \x01(\x04R\x11dirSplitThreshold\x12\x1f\n" + @@ -5078,7 +5139,8 @@ const file_drsync_proto_rawDesc = "" + "\rop_deadline_s\x18\x05 \x01(\rR\vopDeadlineS\x12'\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" + + "\x12delete_split_batch\x18\b \x01(\rR\x10deleteSplitBatch\x12.\n" + + "\x13delete_shard_budget\x18\t \x01(\x04R\x11deleteShardBudget\"\xff\x03\n" + "\n" + "JobOptions\x12\x15\n" + "\x06job_id\x18\x01 \x01(\x04R\x05jobId\x12\x19\n" + @@ -5206,7 +5268,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\"\xd9\x06\n" + + "\aoptions\x18\x02 \x03(\v2\x15.drsync.v1.JobOptionsR\aoptions\"\xa0\a\n" + "\n" + "ShardSplit\x12&\n" + "\x0fparent_shard_id\x18\x01 \x01(\x04R\rparentShardId\x12\x10\n" + @@ -5216,7 +5278,8 @@ const file_drsync_proto_rawDesc = "" + "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\x12R\n" + - "\x11delete_remainders\x18\a \x03(\v2%.drsync.v1.ShardSplit.DeleteRemainderR\x10deleteRemainders\x1a%\n" + + "\x11delete_remainders\x18\a \x03(\v2%.drsync.v1.ShardSplit.DeleteRemainderR\x10deleteRemainders\x12E\n" + + "\x0edelete_subdirs\x18\b \x03(\v2\x1e.drsync.v1.ShardSplit.NewShardR\rdeleteSubdirs\x1a%\n" + "\bNewShard\x12\x19\n" + "\brel_path\x18\x01 \x01(\fR\arelPath\x1a=\n" + "\fNewEntryList\x12\x17\n" + @@ -5263,13 +5326,14 @@ const file_drsync_proto_rawDesc = "" + "verifyFail\x12#\n" + "\rlinks_created\x18\x11 \x01(\x04R\flinksCreated\x12*\n" + "\x11link_anchor_races\x18\x12 \x01(\x04R\x0flinkAnchorRaces\x12#\n" + - "\rlink_fallback\x18\x13 \x01(\x04R\flinkFallback\"\xc0\x01\n" + + "\rlink_fallback\x18\x13 \x01(\x04R\flinkFallback\"\xe9\x01\n" + "\vShardResult\x12\x19\n" + "\bshard_id\x18\x01 \x01(\x04R\ashardId\x12\x19\n" + "\blease_id\x18\x02 \x01(\x04R\aleaseId\x12/\n" + "\x06status\x18\x03 \x01(\x0e2\x17.drsync.v1.ResultStatusR\x06status\x124\n" + "\bcounters\x18\x04 \x01(\v2\x18.drsync.v1.ShardCountersR\bcounters\x12\x14\n" + - "\x05error\x18\x05 \x01(\tR\x05error\"\xcf\x02\n" + + "\x05error\x18\x05 \x01(\tR\x05error\x12'\n" + + "\x0fdeferred_rmdirs\x18\x06 \x03(\fR\x0edeferredRmdirs\"\xcf\x02\n" + "\n" + "TaskResult\x12\x17\n" + "\atask_id\x18\x01 \x01(\x04R\x06taskId\x12\x19\n" + @@ -5497,21 +5561,22 @@ var file_drsync_proto_depIdxs = []int32{ 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 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 + 56, // 40: drsync.v1.ShardSplit.delete_subdirs:type_name -> drsync.v1.ShardSplit.NewShard + 2, // 41: drsync.v1.ShardResult.status:type_name -> drsync.v1.ResultStatus + 46, // 42: drsync.v1.ShardResult.counters:type_name -> drsync.v1.ShardCounters + 2, // 43: drsync.v1.TaskResult.status:type_name -> drsync.v1.ResultStatus + 10, // 44: drsync.v1.TaskResult.src_caps:type_name -> drsync.v1.MountCaps + 10, // 45: drsync.v1.TaskResult.dst_caps:type_name -> drsync.v1.MountCaps + 48, // 46: drsync.v1.TaskResultBatch.results:type_name -> drsync.v1.TaskResult + 8, // 47: drsync.v1.JournalRecord.type:type_name -> drsync.v1.JournalRecord.Type + 9, // 48: drsync.v1.JournalRecord.src:type_name -> drsync.v1.StatInfo + 9, // 49: drsync.v1.JournalRecord.dst:type_name -> drsync.v1.StatInfo + 53, // 50: drsync.v1.StatsReport.latencies:type_name -> drsync.v1.LatencyHistogram + 51, // [51:51] is the sub-list for method output_type + 51, // [51:51] is the sub-list for method input_type + 51, // [51:51] is the sub-list for extension type_name + 51, // [51:51] is the sub-list for extension extendee + 0, // [0:51] is the sub-list for field type_name } func init() { file_drsync_proto_init() } diff --git a/template.yaml b/template.yaml index 75a8464..5e28e8c 100644 --- a/template.yaml +++ b/template.yaml @@ -97,6 +97,7 @@ spec: 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 + delete_shard_budget: 250000 # objects one DELETE shard removes before pushing subdirs back 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_budget_e2e.sh b/test/delete_fanout_budget_e2e.sh new file mode 100755 index 0000000..cff1359 --- /dev/null +++ b/test/delete_fanout_budget_e2e.sh @@ -0,0 +1,175 @@ +#!/usr/bin/env bash +# drsync delete-fanout budget e2e: an orphan tree that is pathological by +# aggregate depth/branching, NOT by any single directory's own entry count — +# every directory anywhere in the tree stays comfortably under tuning. +# delete_split_threshold, so mechanism 1 (stream_delete_split, delete_ +# fanout_e2e.sh / delete_fanout_nested_e2e.sh) never fires anywhere. Only +# tuning.delete_shard_budget (agent/src/delete.c, queue_delete_subdir) can +# catch this shape: once a shard's work budget runs out mid-descent, every +# not-yet-opened subdirectory is handed off as its own new top-level DELETE +# shard, the delete-pass analogue of the scan walker's queue_split. +# +# This is the exact shape a real production incident hit: a 77-way branching +# orphan tree, several levels deep, no single directory large enough to trip +# any reasonable threshold — the delete pass finished all its other work and +# then spent 6 hours on its last 2 (of what should have been many) shards. +set -euo pipefail + +ROOT=$(cd "$(dirname "$0")/.." && pwd) +. "$ROOT/test/lib.sh" +WORK=$(mktemp -d "${TMPDIR:-/tmp}/drsync-delfanoutbudget.XXXXXX") +read -r _CP _HP < <(pick_ports) +CP=${CP:-$_CP}; HP=${HP:-$_HP} +API="http://127.0.0.1:${HP}"; AUTH="Authorization: Bearer delfanoutbudgettok" +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() { + local pat=$1 out + shift + out=$("$@") || return 1 + grep -q -- "$pat" <<<"$out" +} +export DRSYNC_SERVER="$API" DRSYNC_TOKEN=delfanoutbudgettok + +API_TOKEN_FILE="$WORK/api-token" +echo -n delfanoutbudgettok >"$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 ) + +# --- tree: branching but never individually wide ------------------------------ +# orphandir/b00..b09 (10-way branch), each with c00..c09 (10-way branch again), +# each holding 5 files — 10*10*5 = 500 files total. Every directory's own +# immediate entry count is at most 10, comfortably under any threshold used +# below, so mechanism 1 (entry-count streaming) never triggers anywhere in +# this tree — only a work-budget-based handoff can fan this out. +SRC="$WORK/src"; DST="$WORK/dst" +mkdir -p "$SRC/keep" "$DST/keep" +echo keepme > "$SRC/keep/file.txt" +echo keepme > "$DST/keep/file.txt" +for b in $(seq 0 9); do + for c in $(seq 0 9); do + leaf="$DST/orphandir/b$(printf %02d "$b")/c$(printf %02d "$c")" + mkdir -p "$leaf" + for f in $(seq 1 5); do + echo "junk $b $c $f" > "$leaf/f$(printf %02d "$f").txt" + done + done +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 delfanoutbudget-agent -w 4 -C 4 \ + >"$WORK/agent.log" 2>&1 & +APID=$! +sleep 1 + +# delete_split_threshold set high enough that NOTHING in this tree (max 10 +# entries in any one directory) ever trips it — mechanism 1 must stay +# completely inert here. delete_shard_budget=15 is small enough that a shard +# starting at orphandir (10 b-dirs) exhausts its budget a handful of objects +# into the first b-dir's own c-dirs, forcing at least several handoffs. +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 + curl -sf -H "$AUTH" "$API/api/v1/jobs/delfanoutbudget" | 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 fired via the BUDGET mechanism specifically: with no directory +# anywhere in the tree ever over delete_split_threshold, mechanism 1 never +# fires — the only way this tree splits into more than 1 shard at all is +# queue_delete_subdir handing off unopened b/c-directories once a shard's +# budget runs out. Before this fix, this whole tree (10 b-dirs * 10 c-dirs +# * 5 files = 500 objects, comfortably reproducing the "no single dir is +# large" production shape) would run to completion inside the ONE +# top-level shard, start to finish, with zero fan-out. +[[ "$NDEL" -ge 3 ]] || fail "only $NDEL delete shards recorded; budget-based fan-out did not fire " \ + "(want >=3: the top-level shard plus at least one budget-exhausted handoff)" + +# 2. orphandir and everything under it (all 10 b-dirs, all 100 c-dirs, all +# 500 files) is gone. +[[ ! -e "$DST/orphandir" ]] || fail "orphandir 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 +"$DRSYNC" report delfanoutbudget --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; branching-but-not-wide orphan tree fully removed; content intact" +PASS=1 +echo "PASS: budget-based DELETE fan-out (queue_delete_subdir) fired on a tree with no wide directory OK" diff --git a/webui/console.html b/webui/console.html index b5cdf44..e12d189 100644 --- a/webui/console.html +++ b/webui/console.html @@ -992,7 +992,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 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"; + 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 delete_shard_budget: 250000 # objects one DELETE shard removes before pushing subdirs back\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:{}, From c076b1e26bfcd9339d08e3906ca6e5c06558366c Mon Sep 17 00:00:00 2001 From: Steven Rhoods Date: Tue, 11 Aug 2026 23:09:34 +0100 Subject: [PATCH 2/2] Add missing schema migrations for delete_groups.done_streaming/pending_children MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit delete_groups' CREATE TABLE IF NOT EXISTS predates both columns (done_streaming from #67, pending_children from #68/this PR) and is a no-op against an already-existing table, so a coordinator whose data-dir was created by an older binary hit "SQL logic error: no such column: pending_children" on its very first delete pass after upgrading — reported live against exactly that scenario. Neither column had a corresponding ALTER TABLE entry in the migrations slice. Added both, plus a regression test that builds a pre-migration delete_groups table by hand and confirms Open (the real migration path) adds both columns and the table is fully usable afterward. Verified end-to-end against a fresh coordinator combining both fan-out mechanisms (a wide directory nested inside a 77-way branching tree, matching the reported production shape) with the exact tuning values reported: delete_split_threshold=1000, delete_split_batch=2500, delete_shard_budget=1000 — fan-out fires, the tree is fully removed, and content stays intact. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PsdNZLfmAFrMX2VUtkLtmm --- .../internal/store/deletefanout_test.go | 67 +++++++++++++++++++ coordinator/internal/store/store.go | 8 +++ 2 files changed, 75 insertions(+) diff --git a/coordinator/internal/store/deletefanout_test.go b/coordinator/internal/store/deletefanout_test.go index ed28ea4..41e7b2b 100644 --- a/coordinator/internal/store/deletefanout_test.go +++ b/coordinator/internal/store/deletefanout_test.go @@ -1,9 +1,13 @@ package store import ( + "database/sql" + "path/filepath" "testing" "time" + _ "modernc.org/sqlite" + "drsync/coordinator/internal/model" ) @@ -20,6 +24,69 @@ func cleanupShard(dirRel string) NewShard { return NewShard{Kind: model.KindDelete} } +// TestOpenMigratesPreExistingDeleteGroupsTable is the schema-migration +// regression: delete_groups' CREATE TABLE IF NOT EXISTS predates both +// done_streaming (added when n_total's meaning changed from total_children +// to shard count, the fan-out-at-every-depth fix) and pending_children +// (added when a budget-exhausted handoff needed to block its own parent's +// close) — a data-dir created by an OLDER coordinator binary already has a +// delete_groups table without one or both columns, and IF NOT EXISTS is a +// no-op against it. Without an explicit ALTER TABLE migration for each +// column (store.go's migrations slice), the very first delete pass such a +// coordinator ever runs after upgrading fails outright: "SQL logic error: +// no such column: pending_children" (or done_streaming, for an even older +// data-dir) — reported live against a coordinator whose data-dir predated +// this column. This builds exactly that pre-migration table by hand, then +// confirms Open (the real migration path, not a mock) adds both columns +// and delete_groups is fully usable afterward. +func TestOpenMigratesPreExistingDeleteGroupsTable(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.db") + raw, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + // The delete_groups shape as it existed before EITHER done_streaming or + // pending_children were added — same PRIMARY KEY, same WITHOUT ROWID, + // just missing both later columns, exactly like a real pre-upgrade + // data-dir's table. + if _, err := raw.Exec(`CREATE TABLE delete_groups ( + pass_id INTEGER NOT NULL, + rel_path TEXT NOT NULL, + n_total INTEGER NOT NULL DEFAULT 0, + n_done INTEGER NOT NULL DEFAULT 0, + closed INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (pass_id, rel_path) + ) WITHOUT ROWID`); err != nil { + t.Fatal(err) + } + if err := raw.Close(); err != nil { + t.Fatal(err) + } + + s, err := Open(path) + if err != nil { + t.Fatalf("Open on a pre-migration delete_groups table: %v", err) + } + defer s.Close() + + // delete_groups must now accept both columns — the exact query shape + // RecordSplit/CompleteDeleteRemainder run against it in production. + if _, err := s.db.Exec(`INSERT INTO delete_groups + (pass_id, rel_path, n_total, n_done, done_streaming, pending_children, closed) + VALUES (1, 'orphandir', 1, 0, 1, 0, 0)`); err != nil { + t.Fatalf("insert into migrated delete_groups: %v", err) + } + var doneStreaming, pending int + if err := s.db.QueryRow(`SELECT done_streaming, pending_children + FROM delete_groups WHERE pass_id = 1 AND rel_path = 'orphandir'`). + Scan(&doneStreaming, &pending); err != nil { + t.Fatalf("read back migrated columns: %v", err) + } + if doneStreaming != 1 || pending != 0 { + t.Fatalf("done_streaming/pending_children = %d/%d, want 1/0", doneStreaming, pending) + } +} + // TestDeleteGroupNestedChildBlocksParentClose is the nested fan-out // regression: a batch shard under "parent" ships a name ("parent/child") // that turns out to itself be pathological (agent/src/delete.c diff --git a/coordinator/internal/store/store.go b/coordinator/internal/store/store.go index b9f3391..67def5f 100644 --- a/coordinator/internal/store/store.go +++ b/coordinator/internal/store/store.go @@ -525,6 +525,14 @@ var migrations = []string{ `ALTER TABLE passes ADD COLUMN link_anchor_races INTEGER NOT NULL DEFAULT 0`, `ALTER TABLE passes ADD COLUMN link_fallback INTEGER NOT NULL DEFAULT 0`, `ALTER TABLE jobs ADD COLUMN username TEXT NOT NULL DEFAULT ''`, + // delete_groups predates both of these (§2.2 DELETE fan-out landed with + // neither column; done_streaming was added when n_total's meaning changed + // from "total_children" to "shard count", pending_children when fan-out + // started applying at every depth instead of just a shard's own top-level + // path) — a coordinator with a data-dir from before either PR hits "no + // such column" on its very first delete pass without these. + `ALTER TABLE delete_groups ADD COLUMN done_streaming INTEGER NOT NULL DEFAULT 0`, + `ALTER TABLE delete_groups ADD COLUMN pending_children INTEGER NOT NULL DEFAULT 0`, } func (s *Store) Close() error {