From 4be4df8543f284d9ec3a672b105d575ac6f12926 Mon Sep 17 00:00:00 2001 From: Steven Rhoods Date: Sun, 9 Aug 2026 22:03:50 +0100 Subject: [PATCH] Recurse delete fan-out to every directory, not just the top-level orphan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production incident: a delete pass finished all its other work and then spent 6 hours on its last 2 shards, together accounting for ~14M files and directories. Root cause: DELETE fan-out only probed the path named directly in a shard's paths[] for delete_split_threshold — everything rm_tree found while recursing beneath that was removed depth-first with no further splitting, regardless of size. A tree where no single directory (at any depth) individually exceeded the threshold, but the whole subtree summed into the millions, never split at all. agent/src/delete.c: remove_object (renamed from remove_orphan) is now called for every directory entry rm_dir_contents finds during descent, not just the shard's own top-level paths — so a pathological subdirectory nested arbitrarily deep gets streamed out via stream_delete_split too, same as a top-level one. Reuses the probe's own directory handle (rewinddir, or dup+close on the pathological path) instead of a second openat, so this doesn't double the syscall cost at every level. This surfaced a real completion-ordering bug the flat case couldn't: a batch shard that hands a name off to its own nested delete_groups group reports itself done immediately (correctly — it has no more work), which could let its own parent's group close and rmdir before the nested child had actually finished being removed. Fixed with a new pending_children counter on delete_groups: registerPendingChildTx links a newly-created nested group to its parent (derived from the rel_path string, no explicit column needed); closeDeleteGroupTx decrements it on the child's own closure and walks back up the chain, re-checking closeability at each level. Added delete_fanout_nested_e2e.sh: a tree of 40 subdirectories (20 files each, none individually over threshold) that only fans out correctly if every directory is checked during descent, not just the one named in the shard's own paths[] — the existing flat-directory e2e test structurally cannot exercise a nested hand-off at all. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PsdNZLfmAFrMX2VUtkLtmm --- .github/workflows/ci.yml | 1 + README.md | 1 + agent/src/delete.c | 213 +++++++++-------- coordinator/internal/agentsrv/server.go | 13 +- .../internal/store/deletefanout_test.go | 127 +++++++++- coordinator/internal/store/store.go | 220 ++++++++++++++---- docs/DESIGN-coordinator.md | 70 +++++- test/delete_fanout_nested_e2e.sh | 175 ++++++++++++++ 8 files changed, 649 insertions(+), 171 deletions(-) create mode 100755 test/delete_fanout_nested_e2e.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 94ba97c..72adb34 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -115,6 +115,7 @@ jobs: - chunk_resilience_e2e # agent dies mid-copy; leases expire and re-grant - deep_e2e # directory chain deeper than the walker's in-agent limit - delete_fanout_e2e # pathological orphan directory fans out across DELETE shards + - delete_fanout_nested_e2e # fan-out applies at every depth, not just the top-level orphan path - dirfix_e2e # DIRFIX over a directory that fans out to entry-lists - direct_write_e2e # copy.direct_write: new files no-temp, updates atomic - fanout_e2e # a small volume must still use the whole fleet diff --git a/README.md b/README.md index 334df91..b7e1c21 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ tree, and builds the binaries it needs itself: | `chunk_resilience_e2e.sh` | agent dies mid-copy; leases expire and re-grant | | `deep_e2e.sh` | directory chain deeper than the walker's in-agent limit | | `delete_fanout_e2e.sh` | pathological orphan directory fans out across DELETE shards | +| `delete_fanout_nested_e2e.sh` | fan-out applies at every depth, not just the top-level orphan path | | `dirfix_e2e.sh` | DIRFIX over a directory that fans out to entry-lists | | `direct_write_e2e.sh` | `copy.direct_write`: new files skip the temp+rename, updates stay atomic | | `fanout_e2e.sh` | a small volume must still use the whole fleet | diff --git a/agent/src/delete.c b/agent/src/delete.c index 567fae4..69ab3ba 100644 --- a/agent/src/delete.c +++ b/agent/src/delete.c @@ -13,12 +13,16 @@ * unlinked depth-first by this one agent — the delete-pass analogue of * walker.c's split_entrylist_stream, but simpler: there is nothing to diff * against a destination (the whole subtree is already condemned), so a batch - * is just names to remove. Only checked at the top level (the orphan path - * directly named in this shard's paths[]), not recursively at every nested - * depth — the same scope split_entrylist_stream's own trigger has (it only - * fires for the directory the walker is currently sitting in). A nested - * subdirectory that is itself pathological is still removed correctly by - * whichever shard reaches it, just not further split. */ + * is just names to remove. remove_object() applies this check at EVERY + * directory the removal touches, not just the top-level path named in the + * shard's paths[] — rm_dir_contents() routes every entry it finds back + * through remove_object() rather than recursing directly, so a directory + * that is individually small still gets caught if it is nested many levels + * deep under a huge tree of otherwise-small directories (a wide/deep tree + * where no single directory looks large from its own parent's point of view + * previously never split at all, and got removed serially by one agent + * thread — see docs/DESIGN-coordinator.md §2.2 for the incident that + * surfaced this). */ #include "agent.h" #include @@ -38,41 +42,45 @@ * its own right. */ #define DELETE_SPLIT_BATCH_DEFAULT 20000 -/* depth-first removal of name inside parentfd; returns removed count */ -static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name, - const char *rel) +static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, + const char *name, const char *rel); + +/* depth-first removal of the already-open directory d (fd dirfd(d), path + * rel) — every entry goes through remove_object, so a subdirectory found + * at ANY depth that turns out to be pathological is streamed out via + * stream_delete_split exactly like a top-level orphan, not just the one + * named directly in the shard's paths[] (see the fan-out comment at the top + * of this file). Does not remove rel itself or consume d; the caller does + * both, so remove_object can decide not to (a directory it just handed off + * to a split must NOT be rmdir'd here — the split's own cleanup shard owns + * that once every batch it produced has drained). */ +static uint64_t rm_dir_contents(struct walk_ctx *ctx, DIR *d, const char *rel) { - struct stat st; - if (fstatat(parentfd, name, &st, AT_SYMLINK_NOFOLLOW) < 0) { - if (errno != ENOENT) /* already gone is success */ - walk_err(ctx, "stat for delete", rel); - return 0; + uint64_t removed = 0; + struct dirent *de; + while ((de = readdir(d))) { + if (de->d_name[0] == '.' && + (de->d_name[1] == '\0' || + (de->d_name[1] == '.' && de->d_name[2] == '\0'))) + continue; + char crel[PATH_MAX]; + snprintf(crel, sizeof crel, "%s/%s", rel, de->d_name); + removed += remove_object(ctx, dirfd(d), de->d_name, crel); } + return removed; +} + +/* Fully removes name (file, or directory including every descendant) inside + * parentfd — depth-first, via rm_dir_contents. Returns removed count. Only + * reached for a directory once remove_object has already decided it is NOT + * pathological (under delete_split_threshold), so no further probing here. */ +static uint64_t rm_tree(struct walk_ctx *ctx, int parentfd, const char *name, + const char *rel, DIR *already_open) +{ uint64_t removed = 0; - if (S_ISDIR(st.st_mode)) { - int fd = openat(parentfd, name, - O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (fd < 0) { - walk_err(ctx, "open for delete", rel); - return 0; - } - DIR *d = fdopendir(fd); - if (!d) { - close(fd); - walk_err(ctx, "fdopendir for delete", rel); - return 0; - } - struct dirent *de; - while ((de = readdir(d))) { - if (de->d_name[0] == '.' && - (de->d_name[1] == '\0' || - (de->d_name[1] == '.' && de->d_name[2] == '\0'))) - continue; - char crel[PATH_MAX]; - snprintf(crel, sizeof crel, "%s/%s", rel, de->d_name); - removed += rm_tree(ctx, dirfd(d), de->d_name, crel); - } - closedir(d); + if (already_open) { + removed = rm_dir_contents(ctx, already_open, rel); + closedir(already_open); if (unlinkat(parentfd, name, AT_REMOVEDIR) < 0 && errno != ENOENT) { walk_err(ctx, "rmdir", rel); return removed; @@ -204,69 +212,86 @@ int open_parent_beneath(int root_fd, const char *rel, const char **leaf) return cur; } -/* Removes rel, fanning out to stream_delete_split if it is a directory over - * delete_split_threshold. Returns the number of objects this shard itself - * removed (0 if it handed the directory off to a split instead). */ -static uint64_t remove_orphan(struct walk_ctx *ctx, int pfd, const char *leaf, - const char *rel) +/* Removes name (file or directory) inside parentfd, fanning out to + * stream_delete_split instead of recursing if it is a directory over + * delete_split_threshold. Called both for a shard's top-level orphan paths + * (process_delete) and for every entry rm_dir_contents finds while + * descending an already-accepted directory — so a pathologically large + * subdirectory found at ANY depth is caught, not just one named directly in + * the shard's paths[] (the gap the single-level-only version of this check + * left: a tree of many individually-small subdirectories summing to tens of + * millions of entries never split at all, since no single directory in the + * chain ever looked large from its own parent's point of view). Returns the + * number of objects this shard itself removed (0 if it handed the directory + * off to a split instead — those objects are counted by the split-produced + * shards that actually remove them). */ +static uint64_t remove_object(struct walk_ctx *ctx, int parentfd, + const char *name, const char *rel) { struct stat st; - if (fstatat(pfd, leaf, &st, AT_SYMLINK_NOFOLLOW) < 0) { - if (errno != ENOENT) + if (fstatat(parentfd, name, &st, AT_SYMLINK_NOFOLLOW) < 0) { + if (errno != ENOENT) /* already gone is success */ walk_err(ctx, "stat for delete", rel); return 0; } + if (!S_ISDIR(st.st_mode)) + return rm_tree(ctx, parentfd, name, rel, NULL); + uint64_t threshold = ctx->oe->o.delete_split_threshold; - if (S_ISDIR(st.st_mode) && threshold) { - /* Bounded probe: open the directory once just to count up to - * threshold+1 entries and decide pathological without materialising - * the whole thing — same idea as walker.c's read_entries_upto, - * reimplemented here rather than shared, since delete's probe needs - * no stat placeholders or destination-side bookkeeping, only a - * count. This fd is closed here regardless of outcome: readdir's - * position is tied to the fd's own open file description, so - * reusing it for the real stream below would silently resume - * partway through instead of at the start — a second, independent - * openat for the pathological case is one extra syscall against - * what is, by construction, a very large directory. */ - int pfd2 = openat(pfd, leaf, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (pfd2 < 0) { - walk_err(ctx, "open for delete probe", rel); - return 0; - } - DIR *pd = fdopendir(pfd2); - if (!pd) { - close(pfd2); - walk_err(ctx, "fdopendir for delete probe", rel); - return 0; - } - uint64_t seen = 0; - struct dirent *de; - errno = 0; - while (seen <= threshold && (de = readdir(pd))) { - if (de->d_name[0] == '.' && - (de->d_name[1] == '\0' || (de->d_name[1] == '.' && de->d_name[2] == '\0'))) - continue; - seen++; - } - bool probe_err = errno != 0; - closedir(pd); /* closes pfd2 too */ - if (probe_err) { - walk_err(ctx, "probe dir for delete", rel); - return 0; - } - if (seen > threshold) { - int fd = openat(pfd, leaf, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (fd < 0) { - walk_err(ctx, "open for delete split", rel); - return 0; - } - stream_delete_split(ctx, rel, fd); /* consumes fd */ + if (!threshold) + return rm_tree(ctx, parentfd, name, rel, NULL); + + int fd = openat(parentfd, name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (fd < 0) { + walk_err(ctx, "open for delete probe", rel); + return 0; + } + DIR *d = fdopendir(fd); + if (!d) { + close(fd); + walk_err(ctx, "fdopendir for delete probe", rel); + return 0; + } + /* Bounded probe: count up to threshold+1 entries without materialising + * the whole directory — same idea as walker.c's read_entries_upto, + * reimplemented here rather than shared, since delete's probe needs no + * stat placeholders or destination-side bookkeeping, only a count. */ + uint64_t seen = 0; + struct dirent *de; + errno = 0; + while (seen <= threshold && (de = readdir(d))) { + if (de->d_name[0] == '.' && + (de->d_name[1] == '\0' || (de->d_name[1] == '.' && de->d_name[2] == '\0'))) + continue; + seen++; + } + if (errno) { + walk_err(ctx, "probe dir for delete", rel); + closedir(d); + return 0; + } + if (seen > threshold) { + /* rewinddir so the stream starts from the first entry, not wherever + * the probe's readdir left off. stream_delete_split takes ownership + * of d's fd (its own fdopendir + closedir) — do not closedir(d) + * here too, or the fd double-closes. */ + rewinddir(d); + int dupfd = dup(dirfd(d)); + closedir(d); + if (dupfd < 0) { + walk_err(ctx, "dup for delete split", rel); return 0; } - /* Under threshold: fall through to the ordinary recursive removal. */ + stream_delete_split(ctx, rel, dupfd); + return 0; } - return rm_tree(ctx, pfd, leaf, rel); + /* Under threshold: reuse this same handle for the real removal — + * rewinddir resets ITS OWN position (unlike dup(), which shares the + * original's offset — see this file's history for why that distinction + * matters), so no second openat is needed on the common case, which now + * runs at every depth instead of once per top-level orphan. */ + rewinddir(d); + return rm_tree(ctx, parentfd, name, rel, d); } void process_delete(const struct shard_item *it) @@ -304,12 +329,12 @@ void process_delete(const struct shard_item *it) continue; } /* counters: a delete pass reports removals in the orphans column. - * remove_orphan returns 0 (not an undercount) when it fans a + * remove_object returns 0 (not an undercount) when it fans a * pathological directory out instead of removing it directly — those * objects are counted by the split-produced shards that actually * remove them, same as entry-list fan-out shifts the copy counters * onto the children instead of the walker that discovered them. */ - CTR_ADD(ctx.c.orphans, remove_orphan(&ctx, pfd, leaf, rel)); + CTR_ADD(ctx.c.orphans, remove_object(&ctx, pfd, leaf, rel)); close(pfd); } diff --git a/coordinator/internal/agentsrv/server.go b/coordinator/internal/agentsrv/server.go index ed1f849..12d904e 100644 --- a/coordinator/internal/agentsrv/server.go +++ b/coordinator/internal/agentsrv/server.go @@ -565,12 +565,12 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error { // bumps delete_groups.n_total by one per entry, so it stays in the same // units as n_done (shards, not directory entries — see delete_groups' // doc comment). total_children > 0 only marks LastBatch, the "readdir - // hit EOF" signal; the cleanup shard is pre-built regardless of whether - // every sibling has already reported done — RecordSplit only actually - // inserts it once that's true. + // hit EOF" signal; BuildCleanup is only actually called once every + // sibling AND any nested child group has reported done — RecordSplit + // decides that, not this loop. deleteTotals = append(deleteTotals, store.DeleteGroupTotal{ DirRel: string(dr.DirRel), LastBatch: dr.TotalChildren > 0, - CleanupShard: deleteCleanupShard(string(dr.DirRel)), + BuildCleanup: deleteCleanupShard, }) } @@ -761,10 +761,11 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error { // seedDeletePass never does) — maintain delete_groups the same // way completeChunk maintains chunk_groups, seeding a cleanup // shard for the now-possibly-empty directory once every sibling - // has reported done. + // AND any nested child group (fan-out applies at every depth, not + // just relPath's own top level) has reported done. blob, _ := proto.Marshal(r) err = s.st.CompleteDeleteRemainder(shardID, leaseID, passID, relPath, blob, - deleteCleanupShard(relPath), r.Counters) + deleteCleanupShard(relPath), deleteCleanupShard, r.Counters) default: blob, _ := proto.Marshal(r) err = s.st.CompleteShard(shardID, leaseID, passID, blob, r.Counters) diff --git a/coordinator/internal/store/deletefanout_test.go b/coordinator/internal/store/deletefanout_test.go index ad985f2..bb96d73 100644 --- a/coordinator/internal/store/deletefanout_test.go +++ b/coordinator/internal/store/deletefanout_test.go @@ -20,6 +20,111 @@ func cleanupShard(dirRel string) NewShard { return NewShard{Kind: model.KindDelete} } +// 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 +// remove_object checks EVERY directory a removal touches, not just the one +// named in a shard's paths[]) and gets streamed off as its own delete_groups +// row instead of being removed inline. The batch shard that shipped it +// still reports itself done as soon as the hand-off completes — that is +// correct, the batch has no more work — but "parent/child" is NOT yet gone, +// so "parent" must not close (its cleanup rmdir would race an ENOTEMPTY +// directory, or worse, close successfully while the real removal is still +// in flight) until "parent/child"'s own group closes too. +func TestDeleteGroupNestedChildBlocksParentClose(t *testing.T) { + s := openTest(t) + _, passID, shardID := seed(t, s) + if _, err := s.LeaseShards("agent-a", 1, time.Minute); err != nil { + t.Fatal(err) + } + + const parent = "parent" + const child = "parent/child" + + // parent's only (and therefore final) batch: one remainder shard whose + // job is to remove the single name "child" — which, in the real agent, + // turns out to be itself pathological, so it never actually gets + // removed by this shard; it gets streamed off as its own group instead + // (modeled below by directly recording a split for "child" whose parent + // shard is this same remainder shard). + parentIDs, err := s.RecordSplit(shardID, 1, []NewShard{deleteRemainderShard(parent)}, nil, nil, 0, + []DeleteGroupTotal{{DirRel: parent, LastBatch: true, BuildCleanup: cleanupShard}}) + if err != nil { + t.Fatal(err) + } + if len(parentIDs) != 1 { + t.Fatalf("RecordSplit produced %d ids, want 1", len(parentIDs)) + } + parentBatchID := parentIDs[0] + parentLeased, err := s.LeaseShards("agent-a", 1, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(parentLeased) != 1 || parentLeased[0].ID != parentBatchID { + t.Fatalf("lease mismatch: leased=%v parentBatchID=%d", parentLeased, parentBatchID) + } + parentBatchLeaseID := parentLeased[0].LeaseID + + // The parent batch shard, while running, discovers "child" is itself + // pathological and ships it as a nested split — its own parent_shard_id + // is parentBatchID (the remainder shard that found it), not shardID (the + // original walk/delete shard) — same as agent/src/delete.c's + // remove_object calling stream_delete_split for a nested directory. + childIDs, err := s.RecordSplit(parentBatchID, 1, []NewShard{deleteRemainderShard(child)}, nil, nil, 0, + []DeleteGroupTotal{{DirRel: child, LastBatch: true, BuildCleanup: cleanupShard}}) + if err != nil { + t.Fatal(err) + } + if len(childIDs) != 1 { + t.Fatalf("RecordSplit produced %d ids, want 1", len(childIDs)) + } + + // The parent batch shard itself now completes (it was leased BEFORE the + // nested split above, so it's not re-leased here) — it did its job + // (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 { + t.Fatal(err) + } + counts, err := s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + // Only "child"'s own remainder shard should be queued — parent's cleanup + // must NOT have been seeded yet, since parent/child's group is still open. + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued after parent's batch completes (child still pending) = %d, want 1 (only child's own remainder shard, no parent cleanup)", + counts[model.ShardQueued]) + } + + // Now child's own (only) remainder shard completes — its group closes, + // which must decrement parent's pending_children and, since parent's own + // counters were already satisfied, close parent's group too in the same + // transaction (closeDeleteGroupTx's upward walk). + childLeased, err := s.LeaseShards("agent-a", 1, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(childLeased) != 1 || childLeased[0].ID != childIDs[0] { + 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 { + t.Fatal(err) + } + counts, err = s.ShardStateCounts(passID) + if err != nil { + t.Fatal(err) + } + // Both cleanup shards must now be queued: child's own, and parent's + // (unblocked by child's closure propagating up). + if counts[model.ShardQueued] != 2 { + t.Fatalf("queued after child's group closes = %d, want 2 (child's cleanup AND parent's, unblocked by the chain)", + counts[model.ShardQueued]) + } +} + // TestDeleteGroupSeedsCleanupOnceAllChildrenDone is the ordinary case: a // directory splits into 3 delete-remainder shards in one ShardSplit (the // final one carrying LastBatch), all 3 complete, and the cleanup shard for @@ -38,9 +143,9 @@ func TestDeleteGroupSeedsCleanupOnceAllChildrenDone(t *testing.T) { // One DeleteGroupTotal per batch shard (RecordSplit bumps n_total by one // per entry, same units as n_done) — only the last carries LastBatch. ids, err := s.RecordSplit(shardID, 1, remainders, nil, nil, 0, []DeleteGroupTotal{ - {DirRel: dir, CleanupShard: cleanupShard(dir)}, - {DirRel: dir, CleanupShard: cleanupShard(dir)}, - {DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}, + {DirRel: dir, BuildCleanup: cleanupShard}, + {DirRel: dir, BuildCleanup: cleanupShard}, + {DirRel: dir, LastBatch: true, BuildCleanup: cleanupShard}, }) if err != nil { t.Fatal(err) @@ -69,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), nil); err != nil { + cleanupShard(dir), cleanupShard, nil); err != nil { t.Fatalf("complete remainder %d: %v", i, err) } counts, err := s.ShardStateCounts(passID) @@ -116,7 +221,7 @@ func TestDeleteGroupHandlesChildCompletionRacingFinalBatch(t *testing.T) { // entry (server.go onShardSplit), so n_total is bumped to 1 here; only // LastBatch (from total_children>0 on the wire) is still unset. ids, err := s.RecordSplit(shardID, 1, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, - []DeleteGroupTotal{{DirRel: dir, CleanupShard: cleanupShard(dir)}}) + []DeleteGroupTotal{{DirRel: dir, BuildCleanup: cleanupShard}}) if err != nil { t.Fatal(err) } @@ -135,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), nil); err != nil { + cleanupShard(dir), cleanupShard, nil); err != nil { t.Fatal(err) } counts, err := s.ShardStateCounts(passID) @@ -153,7 +258,7 @@ func TestDeleteGroupHandlesChildCompletionRacingFinalBatch(t *testing.T) { // but only 1 child has completed so far, so RecordSplit itself must not // close the group on this call either. ids2, err := s.RecordSplit(shardID, 2, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, - []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}) + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, BuildCleanup: cleanupShard}}) if err != nil { t.Fatal(err) } @@ -180,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), nil); err != nil { + cleanupShard(dir), cleanupShard, nil); err != nil { t.Fatal(err) } counts, err = s.ShardStateCounts(passID) @@ -208,7 +313,7 @@ func TestDeleteGroupNeverSeedsCleanupTwice(t *testing.T) { const dir = "dup-dir" ids, err := s.RecordSplit(shardID, 1, []NewShard{deleteRemainderShard(dir)}, nil, nil, 0, - []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}) + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, BuildCleanup: cleanupShard}}) if err != nil { t.Fatal(err) } @@ -220,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), nil); err != nil { + cleanupShard(dir), cleanupShard, nil); err != nil { t.Fatal(err) } counts, err := s.ShardStateCounts(passID) @@ -235,7 +340,7 @@ func TestDeleteGroupNeverSeedsCleanupTwice(t *testing.T) { // replay racing a reconnect, same shape RecordSplit already handles for // every other split kind) must not seed a second cleanup shard. if _, err := s.RecordSplit(shardID, 2, nil, nil, nil, 0, - []DeleteGroupTotal{{DirRel: dir, LastBatch: true, CleanupShard: cleanupShard(dir)}}); err != nil { + []DeleteGroupTotal{{DirRel: dir, LastBatch: true, BuildCleanup: cleanupShard}}); 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 8679849..2f715c9 100644 --- a/coordinator/internal/store/store.go +++ b/coordinator/internal/store/store.go @@ -299,14 +299,30 @@ CREATE TABLE IF NOT EXISTS chunk_groups ( -- completion, and the "seed a cleanup shard for dir_rel" decision is checked -- at BOTH the write that sets done_streaming and the write that increments -- n_done, whichever lands last: closeable is (done_streaming=1 AND --- n_done >= n_total). +-- n_done>=n_total AND pending_children=0). +-- +-- pending_children handles nesting: fan-out is checked at EVERY directory a +-- removal touches, not just the one named in a shard's paths[] (agent/src/ +-- delete.c remove_object), so a name inside dir_rel's own batch can itself +-- turn out to be pathological and get streamed off as ITS OWN delete_groups +-- row instead of being removed inline. The batch shard that shipped that +-- name still reports itself done as soon as it finishes handing the name +-- off (correctly — the batch itself has no more work), which bumps dir_rel's +-- n_done — but the name itself is NOT yet gone, so dir_rel must not close +-- until that grandchild group also closes. registerPendingChild bumps +-- dir_rel's pending_children when a nested rel_path whose parent is an open +-- delete_groups row is first seen (RecordSplit); the child's own closing +-- transition decrements it and re-checks the parent's closeability from the +-- other direction — same both-sides-check pattern n_done/done_streaming +-- already use for the flat case. CREATE TABLE IF NOT EXISTS delete_groups ( - pass_id INTEGER NOT NULL, - rel_path TEXT NOT NULL, -- the directory being emptied (dir_rel) - n_total INTEGER NOT NULL DEFAULT 0, -- split-produced DELETE shards seen so far - n_done INTEGER NOT NULL DEFAULT 0, -- of those, how many have completed - done_streaming INTEGER NOT NULL DEFAULT 0, -- 1 once the final batch (EOF) has landed - closed INTEGER NOT NULL DEFAULT 0, -- 1 once the cleanup shard has been seeded (idempotency) + pass_id INTEGER NOT NULL, + rel_path TEXT NOT NULL, -- the directory being emptied (dir_rel) + n_total INTEGER NOT NULL DEFAULT 0, -- split-produced DELETE shards seen so far + n_done INTEGER NOT NULL DEFAULT 0, -- of those, how many have completed + done_streaming INTEGER NOT NULL DEFAULT 0, -- 1 once the final batch (EOF) has landed + pending_children INTEGER NOT NULL DEFAULT 0, -- open nested delete_groups rows spawned from a batch of this one + closed INTEGER NOT NULL DEFAULT 0, -- 1 once the cleanup shard has been seeded (idempotency) PRIMARY KEY (pass_id, rel_path) ) WITHOUT ROWID; @@ -1200,22 +1216,107 @@ func recordLinkSightingsTx(tx *sql.Tx, passID int64, sightings []NewLinkSighting return nil } +// deleteGroupParent returns relPath's parent directory and true, or ("", +// false) if relPath is already top-level (no "/") — a top-level orphan's +// delete_groups row has no parent row to chain to (seedDeletePass's own +// top-level DeleteBatch shards don't create one either). +func deleteGroupParent(relPath string) (string, bool) { + i := strings.LastIndexByte(relPath, '/') + if i < 0 { + return "", false + } + 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. +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 + } + if _, err := tx.Exec(`UPDATE delete_groups SET closed = 1 WHERE pass_id = ? AND rel_path = ?`, + passID, relPath); err != nil { + return err + } + parent, ok := deleteGroupParent(relPath) + if !ok { + return nil + } + var pTotal, pDone, pDoneStreaming, pPending, pClosed int + err := tx.QueryRow(`UPDATE delete_groups SET pending_children = pending_children - 1 + WHERE pass_id = ? AND rel_path = ? AND closed = 0 + RETURNING n_total, n_done, done_streaming, pending_children, closed`, + passID, parent).Scan(&pTotal, &pDone, &pDoneStreaming, &pPending, &pClosed) + if errors.Is(err, sql.ErrNoRows) { + return nil // parent has no open row (not itself tracked, or already closed) — nothing to propagate + } + if err != nil { + return err + } + 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) +} + // DeleteGroupTotal records one delete-remainder batch shipped for DirRel — // RecordSplit bumps that group's n_total by one per entry, regardless of // LastBatch (see delete_groups' doc comment for why n_total counts shards, // not ShardSplit.DeleteRemainder.total_children's entry count). LastBatch is // set from total_children > 0 — the "readdir hit EOF for dirRel" signal, only // known on the final streamed batch, not upfront like a chunk group's -// byte-size-derived n_chunks. CleanupShard is pre-built by the caller (store -// stays payload-agnostic, same division of labor as CompleteDataChunk's -// finalizeShard) and only actually inserted if every sibling delete-remainder -// shard has already reported done by the time LastBatch lands — the same -// race CompleteDeleteRemainder resolves from the other direction; it is only -// read when LastBatch is true. +// byte-size-derived n_chunks. BuildCleanup lets store build a cleanup +// 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. type DeleteGroupTotal struct { DirRel string LastBatch bool - CleanupShard NewShard + BuildCleanup func(dirRel string) NewShard } // RecordSplit persists a ShardSplit idempotently: retransmits of the same @@ -1321,35 +1422,42 @@ func (s *Store) RecordSplit(parentShardID int64, seq uint64, shards []NewShard, // two are independent shards/frames, only the parent's own // ShardResult is ordered after every split it shipped (protocol // §4.2), not the split's own processing relative to its children's. - if _, err := tx.Exec(`INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, - passID, dt.DirRel); err != nil { + // n (0 or 1 rows affected) tells us whether THIS call is what + // created dt.DirRel's row — only then does it register as a pending + // child of dt.DirRel's own parent (registerPendingChildTx), so a + // retried/redelivered split for the same nested directory never + // double-bumps pending_children. + n, err := execCountTx(tx, `INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, + passID, dt.DirRel) + if err != nil { return nil, err } + if n > 0 { + if err := registerPendingChildTx(tx, passID, dt.DirRel); err != nil { + return nil, err + } + } // n_total counts shards (one per batch, this one included), never // total_children — see delete_groups' doc comment. - var nTotal, nDone, doneStreaming, closed int + var nTotal, nDone, doneStreaming, pending, closed int if dt.LastBatch { err = tx.QueryRow(`UPDATE delete_groups SET n_total = n_total + 1, done_streaming = 1 WHERE pass_id = ? AND rel_path = ? - RETURNING n_total, n_done, done_streaming, closed`, - passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &closed) + RETURNING n_total, n_done, done_streaming, pending_children, closed`, + passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &pending, &closed) } else { err = tx.QueryRow(`UPDATE delete_groups SET n_total = n_total + 1 WHERE pass_id = ? AND rel_path = ? - RETURNING n_total, n_done, done_streaming, closed`, - passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &closed) + RETURNING n_total, n_done, done_streaming, pending_children, closed`, + passID, dt.DirRel).Scan(&nTotal, &nDone, &doneStreaming, &pending, &closed) } if err != nil { return nil, err } - if closed != 0 || doneStreaming == 0 || nDone < nTotal { - continue // streaming not finished, not every sibling has reported done yet, or already closed - } - if _, err := insertShardsTx(tx, passID, 0, []NewShard{dt.CleanupShard}); err != nil { - return nil, err + if closed != 0 || doneStreaming == 0 || nDone < nTotal || pending > 0 { + continue // streaming not finished, siblings/nested children outstanding, or already closed } - if _, err := tx.Exec(`UPDATE delete_groups SET closed = 1 WHERE pass_id = ? AND rel_path = ?`, - passID, dt.DirRel); err != nil { + if err := closeDeleteGroupTx(tx, passID, dt.DirRel, dt.BuildCleanup(dt.DirRel), dt.BuildCleanup); err != nil { return nil, err } } @@ -1563,15 +1671,24 @@ func (s *Store) CompleteDataChunk(shardID, leaseID, passID int64, relPath string // CompleteDataChunk. relPath is the directory being emptied (dir_rel), not // this shard's own identity; see agentsrv.onShardResult for how a // split-produced KindDelete shard is told apart from an ordinary top-level -// one. When n_done reaches n_total AND done_streaming is set (the streaming -// parent's final DeleteRemainder batch has landed — see RecordSplit) this -// inserts cleanupShard (built by the caller: a single-entry DeleteBatch for -// relPath itself, run through the ordinary unmodified delete path — store -// stays payload-agnostic, same division of labor as CompleteDataChunk's -// finalizeShard) and marks the group closed so a re-delivered final result -// can never seed it twice. counters folds the pass-counter accumulation into +// one. When n_done reaches n_total, done_streaming is set (the streaming +// parent's final DeleteRemainder batch has landed — see RecordSplit), AND no +// pending nested child group remains open (fan-out applies at every +// 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. -func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath string, result []byte, cleanupShard NewShard, counters *drsyncpb.ShardCounters) error { +func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath string, result []byte, + cleanupShard NewShard, buildCleanup func(dirRel string) NewShard, counters *drsyncpb.ShardCounters) error { defer s.lockTimed("CompleteDeleteRemainder")() tx, err := s.db.Begin() if err != nil { @@ -1596,25 +1713,30 @@ func (s *Store) CompleteDeleteRemainder(shardID, leaseID, passID int64, relPath // RecordSplit's own insert for the batch that produced it — vanishingly // unlikely (the grant that produced this ShardResult already required // the row's shard to exist) but INSERT OR IGNORE makes the ordering - // irrelevant either way, same defensive shape as chunk_groups. - if _, err := tx.Exec(`INSERT OR IGNORE INTO delete_groups (pass_id, rel_path) VALUES (?, ?)`, - passID, relPath); err != nil { + // 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 } - var nTotal, nDone, doneStreaming, closed int + if n > 0 { + if err := registerPendingChildTx(tx, passID, relPath); err != nil { + return err + } + } + 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, closed`, - passID, relPath).Scan(&nTotal, &nDone, &doneStreaming, &closed); err != nil { + WHERE pass_id = ? AND rel_path = ? RETURNING n_total, n_done, done_streaming, pending_children, closed`, + passID, relPath).Scan(&nTotal, &nDone, &doneStreaming, &pending, &closed); err != nil { return err } - if closed != 0 || doneStreaming == 0 || nDone < nTotal { - return tx.Commit() // already closed, streaming not finished yet, or siblings outstanding + if closed != 0 || doneStreaming == 0 || nDone < nTotal || pending > 0 { + return tx.Commit() // already closed, streaming not finished, or siblings/nested children outstanding } - if _, err := insertShardsTx(tx, passID, 0, []NewShard{cleanupShard}); err != nil { - return err - } - if _, err := tx.Exec(`UPDATE delete_groups SET closed = 1 WHERE pass_id = ? AND rel_path = ?`, - passID, relPath); err != nil { + if err := closeDeleteGroupTx(tx, passID, relPath, cleanupShard, buildCleanup); err != nil { return err } return tx.Commit() diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index 2e6ad84..eb7d9b7 100644 --- a/docs/DESIGN-coordinator.md +++ b/docs/DESIGN-coordinator.md @@ -133,18 +133,32 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards closely than the plain directory-split (`queue_split`) mechanism, because delete has nothing to diff against a destination — the whole subtree is already condemned (D5) — so a batch is just names to remove, not a source/destination - merge. `remove_orphan` (`agent/src/delete.c`) probe-reads a top-level orphan - directory (only the path named directly in the shard's `paths[]`, not every - nested directory reached during recursion — same scope as the entry-list - trigger) up to `tuning.delete_split_threshold` entries; over that, the directory - is streamed via `stream_delete_split` in batches of `tuning.delete_split_batch` - names as `ShardSplit.DeleteRemainder` splits (wire field 7 — additive to the - existing `subdirs`/`entry_lists`/`big_files`/`link_sightings` repeated fields, - no new frame type needed) instead of being unlinked inline. `ship_split`/ + merge. `remove_object` (`agent/src/delete.c`) probe-reads a directory up to + `tuning.delete_split_threshold` entries; over that, it is streamed via + `stream_delete_split` in batches of `tuning.delete_split_batch` names as + `ShardSplit.DeleteRemainder` splits (wire field 7 — additive to the existing + `subdirs`/`entry_lists`/`big_files`/`link_sightings` repeated fields, no new + frame type needed) instead of being unlinked inline. `ship_split`/ `drain_splits` — the ack/backpressure machinery every split kind already shared — moved out of `walker.c` into a new `split.c` so `delete.c` can use them too without duplicating the ordering-invariant-critical code. + **This check applies at EVERY directory a removal touches, not just the + path named directly in a shard's `paths[]`.** The first version only probed + 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. + Unlike entry-list sharding, delete fan-out has a real completion problem entry-list never does: something has to remove the now-empty directory itself once every split-produced child has finished emptying it, and the coordinator @@ -195,6 +209,36 @@ 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 + 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. + ### 2.3 Shard ``` @@ -250,13 +294,17 @@ chunk_groups (pass_id, rel_path, temp_name, size, mtime_ns, n_chunks, n_done, state) -- large-file cross-fleet assembly; -- finalize task seeded (same tx as the last data chunk) at n_done==n_chunks delete_groups (pass_id, rel_path, n_total, n_done, -- §2.2 DELETE fan-out; the - done_streaming, closed) -- delete-pass analogue of + done_streaming, pending_children, closed) -- delete-pass analogue of -- chunk_groups. n_total counts split-produced DELETE *shards* (bumped by -- one per DeleteRemainder batch received, NOT total_children's entry -- count — see §2.2's "found in local verification" note for why those -- two must never be compared); done_streaming is set once the final - -- (EOF) batch lands. Cleanup shard seeded once done_streaming=1 AND - -- n_done>=n_total, whichever of CompleteDeleteRemainder/RecordSplit + -- (EOF) batch lands. pending_children counts open nested delete_groups + -- rows spawned from one of this row's own batches (fan-out applies at + -- every depth, not just rel_path's own top level — see §2.2's "found in + -- production" note); closeDeleteGroupTx walks this chain bottom-up. + -- Cleanup shard seeded once done_streaming=1 AND n_done>=n_total AND + -- pending_children=0, whichever of CompleteDeleteRemainder/RecordSplit -- observes that first (closed dedups) link_groups (pass_id, dev, ino, nlink_expected, members_seen, anchor_rel_path, anchor_size, anchor_mtime_ns, diff --git a/test/delete_fanout_nested_e2e.sh b/test/delete_fanout_nested_e2e.sh new file mode 100755 index 0000000..d2ab21d --- /dev/null +++ b/test/delete_fanout_nested_e2e.sh @@ -0,0 +1,175 @@ +#!/usr/bin/env bash +# drsync delete-fanout nested e2e: an orphan directory whose OWN entry count +# is small, but whose descendants sum to far more than tuning. +# delete_split_threshold, must still fan out — at whichever depth a +# directory is itself pathological, not just at the top level named in the +# shard's paths[] (docs/DESIGN-coordinator.md §2.2 DELETE fan-out). This is +# a distinct shape from delete_fanout_e2e.sh (one large flat directory): a +# real incident showed a wide/deep tree of many individually-small +# subdirectories — none of which looked large from its own parent's point of +# view — never split at all under the single-level-only version of this +# check, and got removed serially by one agent thread over several hours. +set -euo pipefail + +ROOT=$(cd "$(dirname "$0")/.." && pwd) +. "$ROOT/test/lib.sh" +WORK=$(mktemp -d "${TMPDIR:-/tmp}/drsync-delfanoutnest.XXXXXX") +read -r _CP _HP < <(pick_ports) +CP=${CP:-$_CP}; HP=${HP:-$_HP} +API="http://127.0.0.1:${HP}"; AUTH="Authorization: Bearer delfanoutnesttok" +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=delfanoutnesttok + +API_TOKEN_FILE="$WORK/api-token" +echo -n delfanoutnesttok >"$API_TOKEN_FILE" +chmod 600 "$API_TOKEN_FILE" +DRSYNC="$ROOT/bin/drsync" + +# --- build ------------------------------------------------------------------- +make -C "$ROOT/agent" -s +( cd "$ROOT" && go build -o bin/drsyncd ./coordinator/cmd/drsyncd \ + && go build -o bin/drsync ./cli/drsync ) + +# --- trees: a small source, a destination carrying a wide orphan tree -------- +# orphandir/sub0000 .. orphandir/sub0039: 40 subdirectories, 20 files each +# (800 files total). orphandir itself has only 40 entries (all directories) +# and each subN has only 20 — both individually well under the threshold +# below, so neither the top-level probe nor a naive per-subdirectory probe +# would trigger on either level alone; only summing across the whole +# subtree reveals it's pathological, which is exactly why every directory +# in the descent must be checked, not just the one named in paths[]. +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 d in $(seq 0 39); do + sub="$DST/orphandir/sub$(printf %04d "$d")" + mkdir -p "$sub" + for f in $(seq 1 20); do + echo "junk $d $f" > "$sub/f$(printf %04d "$f").txt" + 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 delfanoutnest-agent -w 4 -C 4 \ + >"$WORK/agent.log" 2>&1 & +APID=$! +sleep 1 + +# threshold=10 sits below BOTH orphandir's own entry count (40 subN +# directories) and each subN's own entry count (20 files) — so the +# regression check is direct: every subN directory must independently fan +# out when rm_dir_contents reaches it during descent, not just orphandir +# itself at the shard's own top level. Proves both levels split in one run. +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/delfanoutnest" | 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 more than once: the top-level orphandir shard alone +# finding 40 sub-directories would produce only a couple of remainder +# batches (batch=8) plus a cleanup shard if it were the only pathological +# directory — but 40 subN directories each individually over threshold +# (20 > 10) must ALSO each fan out when reached during descent, which +# multiplies the shard count well past what a single-level check could +# ever produce. This is the actual regression check: before the fix, this +# number would be tiny (orphandir's own ~6 batches + cleanup, nothing +# from any subN) and the subdirectories' 800 files would be removed +# serially inside one shard instead. +[[ "$NDEL" -ge 20 ]] || fail "only $NDEL delete shards recorded; nested fan-out did not fire " \ + "(want >=20: each of 40 subN dirs over threshold must independently split, " \ + "not just orphandir itself)" + +# 2. orphandir and everything under it (all 40 subN dirs, all 800 files) is +# gone — including every subN directory itself, each removed only by its +# own split's coordinator-seeded cleanup shard. +[[ ! -e "$DST/orphandir" ]] || fail "orphandir (or its nested cleanup) was not fully removed" + +# 3. synced content untouched +DIFF=$(diff -r "$SRC" "$DST" 2>&1 || true) +[[ -z "$DIFF" ]] || fail "delete pass damaged synced content:"$'\n'"$DIFF" + +# 4. no errors, nothing parked +"$DRSYNC" report delfanoutnest --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; nested orphan tree fully removed; content intact" +PASS=1 +echo "PASS: nested pathological subdirectories fanned out across DELETE shards OK"