Add work-budget-based DELETE fan-out for trees with no single wide directory

.github/workflows/ci.yml

@@ -116,6 +116,7 @@ jobs:
           - deep_e2e               # directory chain deeper than the walker's in-agent limit
           - delete_fanout_e2e      # pathological orphan directory fans out across DELETE shards
           - delete_fanout_nested_e2e # fan-out applies at every depth, not just the top-level orphan path
+          - delete_fanout_budget_e2e # budget-based fan-out on a tree with no single wide directory
           - dirfix_e2e             # DIRFIX over a directory that fans out to entry-lists
           - direct_write_e2e       # copy.direct_write: new files no-temp, updates atomic
           - fanout_e2e             # a small volume must still use the whole fleet

README.md

@@ -64,6 +64,7 @@ tree, and builds the binaries it needs itself:
 | `deep_e2e.sh` | directory chain deeper than the walker's in-agent limit |
 | `delete_fanout_e2e.sh` | pathological orphan directory fans out across DELETE shards |
 | `delete_fanout_nested_e2e.sh` | fan-out applies at every depth, not just the top-level orphan path |
+| `delete_fanout_budget_e2e.sh` | budget-based fan-out on a tree with no single wide directory |
 | `dirfix_e2e.sh` | DIRFIX over a directory that fans out to entry-lists |
 | `direct_write_e2e.sh` | `copy.direct_write`: new files skip the temp+rename, updates stay atomic |
 | `fanout_e2e.sh` | a small volume must still use the whole fleet |

agent/src/agent.h

@@ -287,6 +287,14 @@ struct walk_ctx {
     struct split_wait       *infl[SPLIT_WINDOW];
     size_t                   infl_head, infl_count;
     unsigned                 tmp_seq; /* atomic: unique temp names per shard */
+    /* DELETE only (delete.c): relative paths of directories this shard
+     * processed inline but did NOT rmdir, because a descendant somewhere
+     * beneath them was handed off via queue_delete_subdir and might still be
+     * in flight elsewhere — see ShardResult.deferred_rmdirs' doc comment
+     * (proto/drsync.proto) for the coordinator-side completion tracking this
+     * feeds. */
+    char                   **deferred;
+    size_t                   n_deferred, cap_deferred;
     char                     err[256];
     bool                     fatal;
 };

agent/src/chunk.c

@@ -286,7 +286,7 @@ void process_chunk(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     jrn_destroy(&ctx);

agent/src/dirfix.c

@@ -107,7 +107,7 @@ void process_dirfix(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     jrn_destroy(&ctx);

agent/src/link.c

@@ -161,7 +161,7 @@ void process_linkfix(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     jrn_destroy(&ctx);

agent/src/main.c

@@ -297,7 +297,7 @@ static void release_shard(struct shard_item *it)
     memset(&c, 0, sizeof c);
     pb_buf b;
     pb_init(&b);
-    enc_shard_result(&b, it->shard_id, it->lease_id, RES_RELEASED, &c, NULL);
+    enc_shard_result(&b, it->shard_id, it->lease_id, RES_RELEASED, &c, NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     shard_item_free(it);

agent/src/msgs.c

@@ -138,6 +138,20 @@ void enc_delete_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
     pb_free(&dr);
 }
 
+void enc_delete_subdir_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
+                             char *const *subdirs, size_t n_subdirs)
+{
+    pb_put_u64(b, 1, parent_shard_id);
+    pb_put_u64(b, 2, seq);
+    for (size_t i = 0; i < n_subdirs; i++) {
+        pb_buf sub;
+        pb_init(&sub);
+        pb_put_bytes(&sub, 1, subdirs[i], strlen(subdirs[i]));
+        pb_put_msg(b, 8, &sub);
+        pb_free(&sub);
+    }
+}
+
 void enc_bigfile_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
                        const struct bigfile *files, size_t n_files)
 {
@@ -204,14 +218,17 @@ static void enc_counters(pb_buf *b, uint32_t field, const struct shard_counters
 }
 
 void enc_shard_result(pb_buf *b, uint64_t shard_id, uint64_t lease_id, int status,
-                      const struct shard_counters *c, const char *error)
+                      const struct shard_counters *c, const char *error,
+                      char *const *deferred_rmdirs, size_t n_deferred_rmdirs)
 {
     pb_put_u64(b, 1, shard_id);
     pb_put_u64(b, 2, lease_id);
     pb_put_u64(b, 3, (uint64_t)status);
     if (c)
         enc_counters(b, 4, c);
     pb_put_str(b, 5, error);
+    for (size_t i = 0; i < n_deferred_rmdirs; i++)
+        pb_put_bytes(b, 6, deferred_rmdirs[i], strlen(deferred_rmdirs[i]));
 }
 
 void enc_stats(pb_buf *b, const struct stats_snapshot *s)
@@ -503,6 +520,7 @@ static bool dec_tuning_opts(const uint8_t *p, size_t n, struct job_options *o)
         case 6: o->entrylist_batch = (uint32_t)pb_get_varint(&c); break;
         case 7: o->delete_split_threshold = pb_get_varint(&c); break;
         case 8: o->delete_split_batch = (uint32_t)pb_get_varint(&c); break;
+        case 9: o->delete_shard_budget = pb_get_varint(&c); break;
         default: pb_skip(&c, wt);
         }
     }

agent/src/msgs.h

@@ -149,6 +149,13 @@ struct job_options {
      * depth-first by one agent. */
     uint64_t delete_split_threshold;
     uint32_t delete_split_batch; /* names per split-produced DELETE shard (0 = built-in default) */
+    /* Work budget for one DELETE shard, in objects removed — the delete-pass
+     * analogue of shard_budget above, mirroring queue_split rather than
+     * delete_split_threshold/delete_split_batch: bounds total work per shard
+     * regardless of tree SHAPE, catching a tree pathological by aggregate
+     * depth/branching with no single directory individually over
+     * delete_split_threshold (docs/DESIGN-coordinator.md §2.2). */
+    uint64_t delete_shard_budget;
     uint32_t statx_batch; /* target statx in flight per walker ⇒ io_uring ring depth */
     int64_t  mtime_slop_ns;
     bool     dry_run;
@@ -340,6 +347,15 @@ void enc_entrylist_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
 void enc_delete_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
                       const char *dir_rel, char *const *names, size_t n_names,
                       uint32_t total_children);
+/* ShardSplit carrying delete_subdirs: wholly UNOPENED subdirectories being
+ * handed off as their own new independent, top-level DELETE shards — the
+ * delete-pass analogue of enc_shard_split's subdirs above, not of
+ * enc_delete_split (see ShardSplit.delete_subdirs' doc comment: this bounds
+ * one shard's total work regardless of tree shape, catching a tree
+ * pathological by aggregate depth/branching with no single directory ever
+ * individually exceeding delete_split_threshold). */
+void enc_delete_subdir_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
+                             char *const *subdirs, size_t n_subdirs);
 /* ShardSplit carrying big files: rel_path + size + mtime_ns each. The
  * coordinator lays them out into ChunkTasks (proto ShardSplit.BigFile). */
 struct bigfile {
@@ -363,8 +379,12 @@ struct linksighting {
 };
 void enc_linksighting_split(pb_buf *b, uint64_t parent_shard_id, uint64_t seq,
                             const struct linksighting *sightings, size_t n_sightings);
+/* deferred_rmdirs/n_deferred_rmdirs: DELETE shards only (delete.c) — pass
+ * NULL/0 for every other kind. See ShardResult.deferred_rmdirs' doc comment
+ * (proto/drsync.proto) for what these are. */
 void enc_shard_result(pb_buf *b, uint64_t shard_id, uint64_t lease_id, int status,
-                      const struct shard_counters *c, const char *error);
+                      const struct shard_counters *c, const char *error,
+                      char *const *deferred_rmdirs, size_t n_deferred_rmdirs);
 void enc_stats(pb_buf *b, const struct stats_snapshot *s);
 
 /* journal (docs/DESIGN-coordinator.md §5): records are varint-length-

agent/src/probe.c

@@ -164,7 +164,7 @@ void process_probe(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &c,
-                     err[0] ? err : NULL);
+                     err[0] ? err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
 }

agent/src/verify.c

@@ -182,7 +182,7 @@ void process_verify(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     jrn_destroy(&ctx);

agent/src/walker.c

@@ -1037,7 +1037,7 @@ void process_shard(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     free(ctx.split);
@@ -1164,7 +1164,7 @@ void process_entrylist(const struct shard_item *it)
     pb_buf b;
     pb_init(&b);
     enc_shard_result(&b, it->shard_id, it->lease_id, status, &ctx.c,
-                     ctx.err[0] ? ctx.err : NULL);
+                     ctx.err[0] ? ctx.err : NULL, NULL, 0);
     out_push(FR_SHARD_RESULT, &b);
     lease_remove(it->lease_id);
     free(ctx.split);

coordinator/internal/agentsrv/server.go

@@ -520,10 +520,49 @@ func (s *Server) onWorkRequest(ac *agentConn, req *drsyncpb.WorkRequest) error {
 }
 
 func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error {
-	shards := make([]store.NewShard, 0, len(sp.Subdirs)+len(sp.EntryLists)+len(sp.DeleteRemainders))
+	shards := make([]store.NewShard, 0, len(sp.Subdirs)+len(sp.EntryLists)+
+		len(sp.DeleteRemainders)+len(sp.DeleteSubdirs))
 	for _, d := range sp.Subdirs {
 		shards = append(shards, store.NewShard{Kind: model.KindDir, RelPath: string(d.RelPath)})
 	}
+	// deleteTotals accumulates one DeleteGroupTotal per delete_groups-tracked
+	// completion this split produces — DeleteSubdirs below (a size-1 group
+	// per handoff) and DeleteRemainders further down (one entry per streamed
+	// batch) both feed it; RecordSplit processes the whole slice together.
+	var deleteTotals []store.DeleteGroupTotal
+	// A wholly unopened subdirectory the delete pass is handing off once its
+	// shard's work budget ran out (agent/src/delete.c, delete_shard_budget) —
+	// the delete-pass analogue of Subdirs above, NOT of DeleteRemainders: the
+	// receiving shard starts its own fresh budget and re-runs the same
+	// remove_object descent on d.RelPath from scratch, exactly like a
+	// top-level orphan from seedDeletePass. See ShardSplit.delete_subdirs'
+	// doc comment for why this exists alongside DeleteRemainder: that
+	// mechanism only catches a directory that is itself wide; this one bounds
+	// one shard's total work regardless of tree shape.
+	//
+	// RelPath is set to d.RelPath itself (unlike an ordinary seedDeletePass
+	// shard, whose RelPath is empty) so onShardResult can register this
+	// handoff against delete_groups: the parent shard that handed d.RelPath
+	// off does NOT wait for it (it has already moved on to its own
+	// siblings), so nothing stops it from calling rmdir on ITS OWN directory
+	// before d.RelPath has actually finished being removed elsewhere — the
+	// exact same completion-order gap DeleteRemainder had, just via a
+	// different hand-off path. Registering d.RelPath as a size-1,
+	// already-done-streaming delete_groups entry lets the SAME
+	// pending_children chain (registerPendingChildTx/closeDeleteGroupTx)
+	// hold the parent's own group open until this shard reports done.
+	for _, d := range sp.DeleteSubdirs {
+		payload, err := proto.Marshal(&drsyncpb.DeleteBatch{RelPaths: [][]byte{d.RelPath}})
+		if err != nil {
+			return err
+		}
+		shards = append(shards, store.NewShard{
+			Kind: model.KindDelete, RelPath: string(d.RelPath), Payload: payload})
+		deleteTotals = append(deleteTotals, store.DeleteGroupTotal{
+			DirRel: string(d.RelPath), LastBatch: true, NoSelfCleanup: true,
+			BuildCleanup: deleteCleanupShard,
+		})
+	}
 	for _, el := range sp.EntryLists {
 		payload, err := proto.Marshal(&drsyncpb.EntryListShard{
 			DirRel: string(el.DirRel), Names: el.Names})
@@ -540,7 +579,6 @@ func (s *Server) onShardSplit(ac *agentConn, sp *drsyncpb.ShardSplit) error {
 	// the top-level batches; job_id/pass_no/task_id are filled at grant time
 	// from the shard row (buildItem), same as every other split-produced
 	// kind.
-	var deleteTotals []store.DeleteGroupTotal
 	for _, dr := range sp.DeleteRemainders {
 		// dr.Names are bare basenames (agent readdir entries, delete.c
 		// flush_delete_split) — DeleteBatch.RelPaths must be full paths
@@ -755,17 +793,37 @@ func (s *Server) onShardResult(ac *agentConn, r *drsyncpb.ShardResult) error {
 		switch {
 		case kind == model.KindChunk:
 			err = s.completeChunk(passID, shardID, leaseID, payload, r)
-		case kind == model.KindDelete && relPath != "":
-			// A split-produced delete-remainder shard (onShardSplit sets
-			// RelPath to the directory it's emptying; a top-level batch from
-			// seedDeletePass never does) — maintain delete_groups the same
+		case kind == model.KindDelete:
+			// EVERY KindDelete shard's completion goes through here now, not
+			// just split-produced ones (relPath != "") — a plain top-level
+			// orphan shard from seedDeletePass (relPath == "") can ALSO carry
+			// DeferredRmdirs: the production incident this fixed was exactly
+			// that shape, a top-level shard deferring its own path's rmdir
+			// because a budget handoff fired somewhere in its own recursion.
+			//
+			// relPath != "" means this shard is itself tracked in
+			// delete_groups — either a DeleteRemainder batch (onShardSplit
+			// sets RelPath to the directory it's emptying) or a budget
+			// handoff (ShardSplit.delete_subdirs, RelPath set to the
+			// handed-off directory itself). Maintain delete_groups the same
 			// way completeChunk maintains chunk_groups, seeding a cleanup
 			// shard for the now-possibly-empty directory once every sibling
-			// AND any nested child group (fan-out applies at every depth, not
-			// just relPath's own top level) has reported done.
+			// AND any nested child group (fan-out applies at every depth)
+			// has reported done. isHandoff distinguishes the two relPath!=""
+			// shapes: a budget handoff's own DeleteBatch is exactly one path
+			// equal to relPath itself (it already removed relPath as an
+			// ordinary top-level orphan, agent/src/delete.c remove_object
+			// top_level=true) — seeding a cleanup shard for it too would be
+			// a redundant rmdir. A DeleteRemainder batch's paths are always
+			// children UNDER dir_rel, never equal to it, so this check
+			// never misfires on the ordinary case.
+			var batch drsyncpb.DeleteBatch
+			isHandoff := relPath != "" && proto.Unmarshal(payload, &batch) == nil &&
+				len(batch.RelPaths) == 1 && string(batch.RelPaths[0]) == relPath
 			blob, _ := proto.Marshal(r)
 			err = s.st.CompleteDeleteRemainder(shardID, leaseID, passID, relPath, blob,
-				deleteCleanupShard(relPath), deleteCleanupShard, r.Counters)
+				!isHandoff, deleteCleanupShard(relPath), deleteCleanupShard,
+				r.DeferredRmdirs, r.Counters)
 		default:
 			blob, _ := proto.Marshal(r)
 			err = s.st.CompleteShard(shardID, leaseID, passID, blob, r.Counters)

coordinator/internal/model/spec.go

@@ -138,6 +138,17 @@ type JobSpec struct {
 			// differ.
 			DeleteSplitThreshold uint64 `yaml:"delete_split_threshold"`
 			DeleteSplitBatch     uint32 `yaml:"delete_split_batch"`
+			// Work budget for one DELETE shard, in objects removed — the
+			// delete-pass analogue of ShardBudget above, mirroring the scan
+			// walker's queue_split rather than DeleteSplitThreshold/
+			// DeleteSplitBatch: bounds one shard's total work regardless of
+			// tree SHAPE, catching a tree pathological by aggregate
+			// depth/branching even when no single directory anywhere in it
+			// individually exceeds DeleteSplitThreshold (docs/DESIGN-
+			// coordinator.md §2.2 — DeleteSplitThreshold alone cannot see
+			// this shape, since it only ever looks at one directory's own
+			// immediate entry count).
+			DeleteShardBudget uint64 `yaml:"delete_shard_budget"`
 			// Fan-out control. Coordinator-side only: these never reach an agent
 			// (D9 — the agent acts on the resolved per-shard overrides it is
 			// granted, not on policy). See SpreadPolicy.
@@ -271,6 +282,16 @@ func (s *JobSpec) ApplyDefaults() {
 	if sp.Tuning.DeleteSplitBatch == 0 {
 		sp.Tuning.DeleteSplitBatch = 20_000
 	}
+	// Delete-pass analogue of ShardBudget: bounds one shard's total work
+	// (objects removed) regardless of tree shape — catches a tree
+	// pathological by aggregate depth/branching that DeleteSplitThreshold
+	// alone cannot see, since that only ever checks one directory's own
+	// immediate entry count. Scaled up from ShardBudget's default the same
+	// way DeleteSplitBatch is scaled up from EntrylistBatch: delete work per
+	// object is a plain unlink, not a full stat/diff/copy pipeline.
+	if sp.Tuning.DeleteShardBudget == 0 {
+		sp.Tuning.DeleteShardBudget = 250_000
+	}
 	if sp.Tuning.StatxBatch == 0 {
 		sp.Tuning.StatxBatch = 256
 	}
@@ -456,6 +477,7 @@ func (s *JobSpec) ToJobOptions(jobID uint64, dryRun bool) (*drsyncpb.JobOptions,
 			MtimeSlopNs:          sp.Tuning.MtimeSlopNS,
 			DeleteSplitThreshold: sp.Tuning.DeleteSplitThreshold,
 			DeleteSplitBatch:     sp.Tuning.DeleteSplitBatch,
+			DeleteShardBudget:    sp.Tuning.DeleteShardBudget,
 		},
 		DryRun:       dryRun,
 		RequireMount: *sp.Probe.RequireMount,

coordinator/internal/model/spec_defaults_test.go

@@ -42,6 +42,9 @@ func TestDefaultsAppliedToMinimalSpec(t *testing.T) {
 	if sp.Tuning.DeleteSplitBatch != 20_000 {
 		t.Errorf("tuning.delete_split_batch = %d, want 20000", sp.Tuning.DeleteSplitBatch)
 	}
+	if sp.Tuning.DeleteShardBudget != 250_000 {
+		t.Errorf("tuning.delete_shard_budget = %d, want 250000", sp.Tuning.DeleteShardBudget)
+	}
 	if sp.Metadata.Hardlinks != "preserve" {
 		t.Errorf("metadata.hardlinks = %q, want preserve", sp.Metadata.Hardlinks)
 	}
@@ -61,9 +64,10 @@ func TestDefaultsAppliedToMinimalSpec(t *testing.T) {
 		t.Errorf("resolved JobOptions chunk sizes = %d/%d, want %d/%d",
 			o.Copy.ChunkThreshold, o.Copy.ChunkSize, twentyFourGiB, eightGiB)
 	}
-	if o.Tuning.DeleteSplitThreshold != 200_000 || o.Tuning.DeleteSplitBatch != 20_000 {
-		t.Errorf("resolved JobOptions delete-split tuning = %d/%d, want 200000/20000",
-			o.Tuning.DeleteSplitThreshold, o.Tuning.DeleteSplitBatch)
+	if o.Tuning.DeleteSplitThreshold != 200_000 || o.Tuning.DeleteSplitBatch != 20_000 ||
+		o.Tuning.DeleteShardBudget != 250_000 {
+		t.Errorf("resolved JobOptions delete tuning = %d/%d/%d, want 200000/20000/250000",
+			o.Tuning.DeleteSplitThreshold, o.Tuning.DeleteSplitBatch, o.Tuning.DeleteShardBudget)
 	}
 }