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"