Merge #26412457: feat(sync): recurse through related event frontiers

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqszvsfy2lca5ujjnpn3h0kft403m5nnxqj6nsd68m394vlj9c8xjlc4hc37z

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

## Scope

Implements the recursive related-event frontier recommended by nostr:nevent1qqsfdj4fegeq7zjpje7cht6cj9xcsw5a9h2lcx2ju0f4uj80gsvvyhcpz3mhxue69uhhyetvv9ujumn8d96zuer9wc70zgv8.

PR1 and PR2 are merged. This is layer three; participant mailbox discovery remains the final stacked proposal in nostr:nevent1qqsyuueftzc68d5a3x5d0pecs66y0nacv4pc87kkcfw499adyl5t3ycpz3mhxue69uhhyetvv9ujumn8d96zuer9wca2tmqx.

Root-only and one-generation filters miss events which reference an immediate parent, including replies to reactions. This layer traverses accepted local `e`, `E`, and event-valued `q` references plus `a`, `A`, and address-valued `q` references. The resulting frontier feeds existing live coverage when it fits and the existing five-second REQ+EOSE historic rotation when it does not.

## Simplified bounded model

Let `R` be an accepted repository root and `D` an event which directly tags `R`.

- `D` owns one independent recursive subtree allowance, configured by `NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT` / `--sync-recursive-descendant-limit` and defaulting to 500.
- `D` itself is direct/core coverage and does not consume the allowance. Every other event which directly tags `R` starts its own branch.
- Locally stored descendants are walked breadth-first in stable creation-time/event-ID order through event and address references.
- While fewer than 500 recursive descendants are known for a branch, its members can be child-query seeds.
- As soon as the branch reaches 500, none of its members remain in recursive child-query filters. Existing auxiliary live coverage is closed and rebuilt without that branch; historic filters omit it too.
- A recursively discovered event never creates another independent allowance. The existing eight-generation traversal remains an additional safety bound.
- Direct/core sync and unrelated direct-root branches continue normally after one branch fills.

This is intentionally a soft query-frontier limit, not an exact storage cap. Events returned by a request already in flight are still processed normally. A single response could therefore store more than the configured number of siblings; on the next reconciliation the branch is full and none of those events can extend recursive fan-out. The design bounds continued graph expansion without adding receive-path rejection or reservation machinery.

## Relay independence and restart

There is no new cross-relay coordinator, counter, reservation, teardown dependency, or completion dependency. Each source retains its own connection, subscriptions, historic cursor, EOSE handling, pacing, health, and backoff.

The only shared input is the relay's existing local database. Every connection derives its desired frontier from accepted LMDB events during normal reconciliation. An event arriving from one source can therefore make the locally reconstructed branch full for later reconciliation on every source, but no source waits for or coordinates with another.

LMDB is also the restart checkpoint. Startup derives the same bounded frontier from persisted events; a branch already full issues no further descendant queries and receives no fresh allowance. No additional durable cursor or budget state is stored.

## Configuration

The positive integer option is present on all four supported surfaces:

- CLI: `--sync-recursive-descendant-limit`
- environment: `NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT`
- NixOS: `syncRecursiveDescendantLimit`
- example environment and configuration reference

Default: 500. Small values such as 2 or 3 are supported for integration tests and fresh-archive canaries.

## Validation

- Exact head: `9360d419b3bffc5703a90aae330c315ed22459ee`
- Stable PR root: `26412457f1da725298671bbec95d5f1dd2733025a9c1ba3ee25ab3f22e0e697f`
- Final diff: 667 additions, 110 deletions (777 changed lines), down from the discarded 1.3k-addition exact-admission design.
- `nix develop -c cargo test --lib -q`: 758 passed.
- Six focused recursive-frontier unit tests: passed.
- Limit-2 historic/restart integration: passed in 42 seconds.
- Serialized sync suite: all descendant cases passed; overall 95 passed, 1 deliberately ignored, and 2 unrelated Layer-2 timing tests failed once then passed isolated reruns.
- `cargo clippy --workspace --all-targets --all-features -- -D warnings`: passed.
- `cargo fmt --check`, `git diff --check`, and `nix flake check --no-build`: passed.

The real-relay integration constructs `R <- D <- A <- B <- C` with a limit of 2. It observes `A` and `B`, proves `C` is not queried, restarts against the same LMDB, publishes another child of `B`, proves both excluded children remain absent, and proves an unmetered direct/core marker still syncs.

## Why the model changed

An aggregate 4,096 limit truncated unrelated direct sets. A later 32,768 recursive aggregate still allowed one popular direct child to consume the entire repository budget. The discarded exact per-branch design prevented overshoot but added roughly 680 runtime lines of receive-time admission, reservations, and cross-relay lifecycle handling.

This head keeps the useful per-direct-child boundary and deliberately sacrifices an exact stored-event guarantee for a much smaller, reconstructible query-frontier rule. The final PR is a single atomic commit and adds no inter-relay behavior.

## Coverage evidence

The audit found 17 genuine missing events at recursive depths 3–5 in the named `gitworkshop` and `ngit` repositories, demonstrating that the former one-generation rule was incomplete. A superseded unrestricted recursive canary recovered at least 230 saved events, showing material recursive yield; it is not presented as validation of the final tiny-limit boundary.

## Canary deployment

The combined PR3 + stacked PR4 candidate was deployed only to the isolated archive canary at 2026-08-14 16:09:48 UTC. PR3 is exact commit `9360d419b3bffc5703a90aae330c315ed22459ee`; the deployed binary is `/nix/store/wlhqpr0x18njpmj3khwigrzhwj9z48lq-ngit-grasp-2.1.2/bin/ngit-grasp`.

The canary uses persisted LMDB at `/persistent/grasp/gitnostr-com-archive-soft-cap-2`, `NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT=2`, and the normal real-relay archive inventory. At the 2026-08-14 16:13:08 UTC snapshot it was active with PID 1553732 and zero restarts. Since activation, logs contained five started descendant historic queries and six terminal descendant batch transitions, demonstrating startup reconstruction and live REQ+EOSE descendant operation. No application panic, OOM, failed assertion, or process crash was present; the only textual `fatal` match was an unrelated external Git remote TLS hostname error.

The public `gitnostr.com` service remained on its prior binary and PID with zero restarts. This is deliberately described as an early three-minute operational snapshot, not a completed long soak or proof that every naturally occurring branch saturated the limit.

## Review state

Merge-ready. The simplified implementation, deterministic limit-2 restart integration, full validation above, and early isolated real-relay canary support a merge recommendation. Extended canary saturation/restart evidence and production deployment at the safe default of 500 remain operational follow-up checks and are not claimed as completed here.

.env.example

@@ -145,6 +145,12 @@
 # Default: wss://relay.ditto.pub,wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net
 # NGIT_SYNC_PLUS_FALLBACK_RELAYS=wss://relay.ditto.pub,wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net
 
+# Soft limit on recursive query-frontier members below each event which directly
+# tags a repository root. Direct root-tagging events do not consume this limit.
+# CLI: --sync-recursive-descendant-limit <count>
+# Default: 500
+# NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT=500
+
 # Maximum backoff time in seconds for sync relay reconnection
 # CLI: --sync-max-backoff-secs <seconds>
 # Default: 3600 (1 hour)

docs/explanation/architecture.md

@@ -547,6 +547,13 @@ The ngit-grasp relay implements **Proactive Sync of Nostr Events**, which synchr
 - **Rejected events index** - prevents wasteful broad re-fetching while retaining exact IDs for dependency recovery
 - **Desired-source retention** keeps listed GRASP-02 relays retryable until
   repository work is actually confirmed, including StateOnly invitation sync
+- **Bounded recursive related-event coverage** turns accepted descendants into
+  the next remote query frontier through event and address references, with an
+  eight-generation traversal and a configurable soft limit, defaulting to 500,
+  for each subtree rooted at an event which directly tags a repository root.
+  The local database deterministically reconstructs saturated branches after
+  restart; relay connections retain independent progress and require no shared
+  admission coordinator
 
 **Architecture:**
 

docs/explanation/grasp-02-proactive-sync.md

@@ -800,12 +800,13 @@ live in
 - **Function**: `build_root_event_tag_filters(root_events, since)`
 - **Only for `SyncLevel::Full` repos** — purgatory announcements (`StateOnly`) skip this layer
 
-### Direct-Member Descendants
+### Recursive Descendant Frontier
 
 Some collaboration events reference only their immediate parent. Once the
 ordinary Layer 3 filters have discovered an event which directly tags a
-repository root, each source relay is also queried for events whose `e`, `E`,
-or `q` tags reference that direct member.
+repository root, the accepted local graph is traversed through `e`, `E`, `q`,
+`a`, and `A` references. Each source relay is queried for events that reference
+every known member, not only the original root or its direct children.
 
 - complete descendant filters are retained live when they fit after core live
   coverage while preserving the two control-plane slots and at least one
@@ -816,20 +817,34 @@ or `q` tags reference that direct member.
   each constrained relay on the existing five-second maintenance cadence,
   through the ordinary historic queue, pagination, shared ledger, and request
   pacing;
-- ordinary historic batches include the currently known direct-member filters;
+- ordinary historic batches include the complete currently known frontier;
 - fallback filters keep an in-memory cursor, advance it only after successful
   EOSE, and query from the preceding successful upper bound with 15 minutes of
-  overlap; and
-- recovered descendants never enter the frontier, so this is deliberately one
-  additional generation rather than recursive thread traversal.
+  overlap;
+- an event recovered from a source relay is accepted into the ordinary local
+  database, then becomes a parent seed on the next five-second reconciliation
+  tick; and
+- each event which directly tags a repository root owns an independent
+  recursive subtree allowance, configured by
+  `NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT` and defaulting to 500. The direct
+  event itself is unmetered. Once its deterministic breadth-first frontier
+  fills, every member of that branch is removed from future child-query seeds;
+  unrelated direct events continue with their own allowances;
+- already active requests can store events beyond the configured number, but
+  those events do not extend a full branch. Accepting this soft overshoot avoids
+  receive-path accounting and leaves source-relay connections independent; and
+- startup and periodic reconciliation rebuild branch membership in stable
+  breadth-first creation-time/event-ID order from the local database. A branch
+  already full after restart therefore issues no further child queries and
+  does not receive a fresh allowance.
 
 An unexpected auxiliary CLOSED retires the remaining descendant subscriptions
 without rebuilding core coverage and falls back to history. A filter already
 queued or active blocks another fallback filter for that relay; failure leaves
 the same filter and cursor at the head. Reconnect and daily reconciliation
-reconstruct the mode from current session capacity. This keeps the feature
-complete without durable cursors, another capacity ledger, request-class
-priority, or multi-connection sharding.
+reconstruct the mode from current session capacity. The local database is the
+frontier checkpoint, so recursion needs no additional durable cursor, capacity
+coordinator, request-class priority, or multi-connection sharding.
 
 ### Combined Layer 2+3 (SyncLevel-Aware)
 

docs/reference/configuration.md

@@ -358,6 +358,32 @@ the relay retains GRASP-02 sync but does not discover NIP-65 inboxes and omits
 
 ---
 
+#### `NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT`
+
+**Description:** Soft limit on recursive query-frontier members below each event which directly tags a repository root
+**Type:** Positive integer
+**Default:** `500`
+**Required:** No
+
+```bash
+# Exercise branch saturation in a fresh archive or test environment
+NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT=3
+```
+
+Direct repository-root events and events which independently tag a root do not
+consume this allowance. Once the deterministic breadth-first frontier contains
+the configured number of descendants for a branch, none of that branch's
+members are used in further child queries. Already active requests can still
+store more than the configured number, which is why this is a soft limit; those
+extra events do not extend the branch. Startup and historic reconciliation
+derive the same frontier from locally stored events, so a saturated branch does
+not receive a fresh budget. Source relays retain independent connections and
+sync progress; there is no cross-relay counter or coordinator.
+
+The corresponding NixOS option is `syncRecursiveDescendantLimit`.
+
+---
+
 #### `NGIT_SYNC_BOOTSTRAP_RELAY_URL`
 
 **Description:** URL of the bootstrap relay to initially sync events from

nix/module.nix

@@ -151,6 +151,16 @@ let
         '';
       };
 
+      syncRecursiveDescendantLimit = mkOption {
+        type = types.ints.positive;
+        default = 500;
+        description = ''
+          Soft limit on recursive query-frontier members below each event
+          which directly tags a repository root. Direct root-tagging events
+          do not consume the limit.
+        '';
+      };
+
       databaseBackend = mkOption {
         type = types.enum [ "lmdb" "memory" ];
         default = "lmdb";
@@ -524,6 +534,8 @@ let
       NGIT_METRICS_TOP_N_REPOS = toString cfg.metricsTopNRepos;
       NGIT_SYNC_MAX_BACKOFF_SECS = toString cfg.syncMaxBackoffSecs;
       NGIT_SYNC_PLUS_ENABLED = if cfg.syncPlusEnabled then "true" else "false";
+      NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT =
+        toString cfg.syncRecursiveDescendantLimit;
       NGIT_SYNC_DISCONNECT_CHECK_INTERVAL_SECS =
         toString cfg.syncDisconnectCheckIntervalSecs;
       NGIT_SYNC_BASE_BACKOFF_SECS = toString cfg.syncBaseBackoffSecs;

src/config.rs

@@ -424,6 +424,15 @@ pub struct Config {
     )]
     pub sync_plus_fallback_relays: String,
 
+    /// Soft limit on recursive query-frontier members below each event which directly tags
+    /// a repository root. Direct root-tagging events are not counted.
+    #[arg(
+        long,
+        env = "NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT",
+        default_value_t = 500
+    )]
+    pub sync_recursive_descendant_limit: usize,
+
     /// Maximum backoff time in seconds for sync relay reconnection (default: 3600 = 1 hour)
     #[arg(long, env = "NGIT_SYNC_MAX_BACKOFF_SECS", default_value_t = 3600)]
     pub sync_max_backoff_secs: u64,
@@ -908,6 +917,12 @@ impl Config {
             ));
         }
 
+        if self.sync_recursive_descendant_limit == 0 {
+            return Err(anyhow!(
+                "NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT must be greater than 0"
+            ));
+        }
+
         Self::validate_deletion_request_retention_duration(
             self.deletion_request_retention_unused_served_secs,
             "NGIT_DELETION_REQUEST_RETENTION_UNUSED_SERVED_SECS",
@@ -1141,6 +1156,7 @@ impl Config {
             sync_plus_enabled: true,
             user_index_relays: DEFAULT_USER_INDEX_RELAYS.to_string(),
             sync_plus_fallback_relays: DEFAULT_SYNC_PLUS_FALLBACK_RELAYS.to_string(),
+            sync_recursive_descendant_limit: 500,
             sync_max_backoff_secs: 3600,
             sync_disconnect_check_interval_secs: 60,
             sync_base_backoff_secs: 5,
@@ -1350,6 +1366,33 @@ mod tests {
         assert!(!disabled.sync_plus_enabled);
     }
 
+    #[test]
+    fn recursive_descendant_limit_defaults_to_500_and_accepts_small_soak_values() {
+        let default = Config::try_parse_from(["ngit-grasp", "--domain", "example.com"])
+            .expect("recursive descendant default should parse");
+        assert_eq!(default.sync_recursive_descendant_limit, 500);
+
+        let tiny = Config::try_parse_from([
+            "ngit-grasp",
+            "--domain",
+            "example.com",
+            "--sync-recursive-descendant-limit=3",
+        ])
+        .expect("tiny recursive descendant limit should parse");
+        assert_eq!(tiny.sync_recursive_descendant_limit, 3);
+    }
+
+    #[test]
+    fn recursive_descendant_limit_must_be_positive() {
+        let mut config = Config::for_testing();
+        config.sync_recursive_descendant_limit = 0;
+
+        let error = config.validate().expect_err("zero limit must be rejected");
+        assert!(error
+            .to_string()
+            .contains("NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT must be greater than 0"));
+    }
+
     #[test]
     fn user_index_relays_parse_cli_list_and_ignore_empty_entries() {
         let config = Config::try_parse_from([

src/sync/mod.rs

@@ -65,7 +65,9 @@ const MAX_PURGATORY_DEPENDENCY_IDS_PER_QUERY: usize = 100;
 const SEMANTIC_FALLBACK_MIN_REQUESTED_EVENTS: usize = 20;
 const SEMANTIC_FALLBACK_MAX_DELIVERED_PERCENT: usize = 10;
 const DESCENDANT_FALLBACK_OVERLAP_SECS: u64 = 15 * 60;
-
+/// Maximum number of locally known parent/child generations expanded into a
+/// relay's related-event query frontier.
+const MAX_DESCENDANT_FRONTIER_DEPTH: usize = 8;
 fn should_use_semantic_fallback(requested_count: usize, received_count: usize) -> bool {
     requested_count >= SEMANTIC_FALLBACK_MIN_REQUESTED_EVENTS
         && received_count.saturating_mul(100)
@@ -883,27 +885,11 @@ struct DescendantFilterInFlight {
 
 #[derive(Debug)]
 struct DescendantLiveCoverage {
-    /// Individual filters physically covered by the still-open REQs. Desired
-    /// coverage may become a subset while these drain naturally.
-    covered_filters: HashSet<String>,
     desired_filters: HashSet<String>,
     fallback_filters: Vec<Filter>,
     subscription_ids: Vec<SubscriptionId>,
 }
 
-impl DescendantLiveCoverage {
-    fn retain_for_removal_only_change(&mut self, desired: &HashSet<String>) -> bool {
-        if desired == &self.desired_filters {
-            return true;
-        }
-        if desired.is_subset(&self.covered_filters) {
-            self.desired_filters.clone_from(desired);
-            return true;
-        }
-        false
-    }
-}
-
 fn filter_group_fingerprint(filters: &[Filter]) -> String {
     filters
         .iter()
@@ -1020,6 +1006,238 @@ fn descendant_event_coordinate(event: &Event) -> Option<String> {
     ))
 }
 
+/// Derive a bounded transitive frontier from related events already accepted
+/// into the local database.
+///
+/// A newly fetched event becomes a parent on the next reconciliation tick, so
+/// remote traversal progresses without retaining another durable cursor. Both
+/// event IDs and replaceable/addressable coordinates participate because
+/// clients may continue a thread through either reference form.
+async fn recursive_descendant_frontier(
+    database: &SharedDatabase,
+    root_events: &HashSet<EventId>,
+    max_recursive_events_per_branch: usize,
+) -> DescendantFrontier {
+    recursive_descendant_frontier_with_limits(
+        database,
+        root_events,
+        MAX_DESCENDANT_FRONTIER_DEPTH,
+        max_recursive_events_per_branch,
+    )
+    .await
+}
+
+async fn recursive_descendant_frontier_with_limits(
+    database: &SharedDatabase,
+    root_events: &HashSet<EventId>,
+    max_depth: usize,
+    max_recursive_events_per_branch: usize,
+) -> DescendantFrontier {
+    let mut frontier = DescendantFrontier::default();
+    if max_depth == 0 || root_events.is_empty() {
+        return frontier;
+    }
+
+    // Every event directly tagging a root owns an independent recursive
+    // subtree. The direct event itself remains complete core coverage.
+    let mut direct_events = HashMap::<EventId, Event>::new();
+    for filter in filters::tagged_one_of_our_root_event_filters(root_events, None) {
+        match database.query(filter).await {
+            Ok(events) => {
+                for event in events
+                    .iter()
+                    .filter(|event| !root_events.contains(&event.id))
+                {
+                    direct_events
+                        .entry(event.id)
+                        .or_insert_with(|| event.clone());
+                }
+            }
+            Err(error) => {
+                tracing::warn!(
+                    %error,
+                    root_event_count = root_events.len(),
+                    "Failed to derive direct repository thread members"
+                );
+                return DescendantFrontier::default();
+            }
+        }
+    }
+
+    let mut direct_events: Vec<_> = direct_events.into_iter().collect();
+    direct_events.sort_by(|(left_id, left), (right_id, right)| {
+        left.created_at
+            .cmp(&right.created_at)
+            .then_with(|| left_id.cmp(right_id))
+    });
+    let direct_event_ids: HashSet<_> = direct_events.iter().map(|(id, _)| *id).collect();
+    let mut branch_member_counts = HashMap::<EventId, usize>::new();
+    let mut event_branches = HashMap::<EventId, HashSet<EventId>>::new();
+    let mut coordinate_branches = HashMap::<String, HashSet<EventId>>::new();
+    let mut event_seeds = HashMap::<EventId, HashSet<EventId>>::new();
+    let mut coordinate_seeds = HashMap::<String, HashSet<EventId>>::new();
+    for (event_id, event) in direct_events {
+        branch_member_counts.insert(event_id, 0);
+        event_branches.entry(event_id).or_default().insert(event_id);
+        event_seeds.entry(event_id).or_default().insert(event_id);
+        if let Some(coordinate) = descendant_event_coordinate(&event) {
+            coordinate_branches
+                .entry(coordinate.clone())
+                .or_default()
+                .insert(event_id);
+            coordinate_seeds
+                .entry(coordinate)
+                .or_default()
+                .insert(event_id);
+        }
+    }
+
+    if max_depth == 1 || (event_seeds.is_empty() && coordinate_seeds.is_empty()) {
+        frontier.event_ids.extend(event_seeds.into_keys());
+        frontier.coordinates.extend(coordinate_seeds.into_keys());
+        return frontier;
+    }
+
+    for depth in 2..=max_depth {
+        let mut layer_events = HashMap::<EventId, Event>::new();
+        let event_seed_ids: HashSet<_> = event_seeds.keys().copied().collect();
+        let coordinate_seed_ids: HashSet<_> = coordinate_seeds.keys().cloned().collect();
+        let mut layer_filters =
+            filters::tagged_one_of_our_root_event_filters(&event_seed_ids, None);
+        layer_filters.extend(filters::tagged_one_of_our_repo_event_filters(
+            &coordinate_seed_ids,
+            None,
+        ));
+
+        for filter in layer_filters {
+            match database.query(filter).await {
+                Ok(events) => {
+                    for event in events.iter() {
+                        if root_events.contains(&event.id)
+                            || direct_event_ids.contains(&event.id)
+                            || layer_events.contains_key(&event.id)
+                        {
+                            continue;
+                        }
+                        layer_events.insert(event.id, event.clone());
+                    }
+                }
+                Err(error) => {
+                    tracing::warn!(
+                        %error,
+                        depth,
+                        event_seed_count = event_seeds.len(),
+                        coordinate_seed_count = coordinate_seeds.len(),
+                        direct_event_count = direct_event_ids.len(),
+                        "Failed to derive recursive repository thread members"
+                    );
+                    return frontier;
+                }
+            }
+        }
+
+        let mut layer_events: Vec<_> = layer_events.into_iter().collect();
+        layer_events.sort_by(|(left_id, left), (right_id, right)| {
+            left.created_at
+                .cmp(&right.created_at)
+                .then_with(|| left_id.cmp(right_id))
+        });
+
+        let mut next_event_seeds = HashMap::<EventId, HashSet<EventId>>::new();
+        let mut next_coordinate_seeds = HashMap::<String, HashSet<EventId>>::new();
+        for (event_id, event) in layer_events {
+            let (addressable_refs, event_refs) =
+                crate::nostr::policy::RelatedEventPolicy::extract_reference_tags(&event);
+            let mut inherited_branches = HashSet::new();
+            for parent in event_refs {
+                if let Some(branches) = event_branches.get(&parent) {
+                    inherited_branches.extend(branches.iter().copied());
+                }
+            }
+            for parent in addressable_refs {
+                if let Some(branches) = coordinate_branches.get(&parent) {
+                    inherited_branches.extend(branches.iter().copied());
+                }
+            }
+            if let Some(existing_branches) = event_branches.get(&event_id) {
+                inherited_branches.retain(|branch| !existing_branches.contains(branch));
+            }
+
+            let mut admitted_branches = HashSet::new();
+            for branch in inherited_branches {
+                let member_count = branch_member_counts.entry(branch).or_default();
+                if *member_count < max_recursive_events_per_branch {
+                    *member_count += 1;
+                    admitted_branches.insert(branch);
+                }
+            }
+            if admitted_branches.is_empty() {
+                continue;
+            }
+            event_branches
+                .entry(event_id)
+                .or_default()
+                .extend(admitted_branches.iter().copied());
+            if let Some(coordinate) = descendant_event_coordinate(&event) {
+                coordinate_branches
+                    .entry(coordinate)
+                    .or_default()
+                    .extend(admitted_branches.iter().copied());
+            }
+
+            let expandable_branches: HashSet<_> = admitted_branches
+                .into_iter()
+                .filter(|branch| branch_member_counts[branch] < max_recursive_events_per_branch)
+                .collect();
+            if expandable_branches.is_empty() {
+                continue;
+            }
+            next_event_seeds.insert(event_id, expandable_branches.clone());
+            if let Some(coordinate) = descendant_event_coordinate(&event) {
+                next_coordinate_seeds.insert(coordinate, expandable_branches);
+            }
+        }
+
+        if next_event_seeds.is_empty() && next_coordinate_seeds.is_empty() {
+            break;
+        }
+        event_seeds = next_event_seeds;
+        coordinate_seeds = next_coordinate_seeds;
+    }
+
+    let exhausted_branches: HashSet<_> = branch_member_counts
+        .iter()
+        .filter(|(_, member_count)| **member_count >= max_recursive_events_per_branch)
+        .map(|(branch, _)| *branch)
+        .collect();
+    for (event_id, branches) in &event_branches {
+        if branches
+            .iter()
+            .any(|branch| !exhausted_branches.contains(branch))
+        {
+            frontier.event_ids.insert(*event_id);
+        }
+    }
+    for (coordinate, branches) in &coordinate_branches {
+        if branches
+            .iter()
+            .any(|branch| !exhausted_branches.contains(branch))
+        {
+            frontier.coordinates.insert(coordinate.clone());
+        }
+    }
+
+    tracing::debug!(
+        depth_limit = max_depth,
+        event_count = frontier.event_ids.len(),
+        coordinate_count = frontier.coordinates.len(),
+        branch_count = branch_member_counts.len(),
+        exhausted_branch_count = exhausted_branches.len(),
+        "Derived bounded recursive repository thread frontier"
+    );
+    frontier
+}
+
 #[cfg(test)]
 fn descendant_frontier_filters(
     frontier: &DescendantFrontier,
@@ -4320,37 +4538,17 @@ impl SyncManager {
             .await;
     }
 
-    /// Find the existing first-generation members of repository root threads.
-    ///
-    /// Only events that directly reference a root are admitted to this
-    /// frontier. Events fetched by the descendant rotation are deliberately
-    /// not fed back into it, keeping this feature non-recursive.
-    async fn direct_thread_members(&self, root_events: &HashSet<EventId>) -> DescendantFrontier {
-        let mut frontier = DescendantFrontier::default();
-        for filter in filters::tagged_one_of_our_root_event_filters(root_events, None) {
-            match self.database.query(filter).await {
-                Ok(events) => {
-                    for event in events
-                        .iter()
-                        .filter(|event| !root_events.contains(&event.id))
-                    {
-                        frontier.event_ids.insert(event.id);
-                        if let Some(coordinate) = descendant_event_coordinate(event) {
-                            frontier.coordinates.insert(coordinate);
-                        }
-                    }
-                }
-                Err(error) => {
-                    tracing::warn!(
-                        error = %error,
-                        root_event_count = root_events.len(),
-                        "Failed to derive direct repository thread members"
-                    );
-                    return DescendantFrontier::default();
-                }
-            }
-        }
-        frontier
+    /// Find the bounded transitive members of repository root threads.
+    async fn descendant_thread_members(
+        &self,
+        root_events: &HashSet<EventId>,
+    ) -> DescendantFrontier {
+        recursive_descendant_frontier(
+            &self.database,
+            root_events,
+            self.config.sync_recursive_descendant_limit,
+        )
+        .await
     }
 
     async fn close_descendant_live_coverage(
@@ -4391,7 +4589,7 @@ impl SyncManager {
         relay_url: &str,
         target: &algorithms::RelaySyncNeeds,
     ) {
-        let frontier = self.direct_thread_members(&target.root_events).await;
+        let frontier = self.descendant_thread_members(&target.root_events).await;
         let historic_entries =
             tiered_auxiliary_filters(&target.repos, &target.root_events, &frontier, None);
         let live_since = Timestamp::from(
@@ -4408,11 +4606,7 @@ impl SyncManager {
             let desired_live_filters: HashSet<_> =
                 live_filters.iter().map(Filter::as_json).collect();
             if let Some(coverage) = self.descendant_live_coverage.get_mut(relay_url) {
-                let desired_changed = desired_live_filters != coverage.desired_filters;
-                if coverage.retain_for_removal_only_change(&desired_live_filters) {
-                    // Removal-only changes update desired fallback coverage but
-                    // do not churn a shared live REQ. Its obsolete filters die
-                    // at CLOSED/disconnect; additions still rebuild below.
+                if desired_live_filters == coverage.desired_filters {
                     let max_filters = connection.max_filters_per_req();
                     let rotation = self
                         .descendant_sync_rotations
@@ -4424,13 +4618,6 @@ impl SyncManager {
                         rotation.refresh(rotated_filters, max_filters);
                     }
                     coverage.fallback_filters = complete_auxiliary_fallback(&historic_entries);
-                    if desired_changed {
-                        tracing::info!(
-                            relay = %relay_url,
-                            retained_subscription_count = coverage.subscription_ids.len(),
-                            "Retaining removal-only auxiliary live coverage until natural closure"
-                        );
-                    }
                     return;
                 }
             }
@@ -4463,7 +4650,6 @@ impl SyncManager {
                         self.descendant_live_coverage.insert(
                             relay_url.to_string(),
                             DescendantLiveCoverage {
-                                covered_filters: desired_live_filters.clone(),
                                 desired_filters: desired_live_filters,
                                 fallback_filters,
                                 subscription_ids: subscription_ids.clone(),
@@ -7655,7 +7841,7 @@ impl SyncManager {
         {
             filters
         } else {
-            let frontier = self.direct_thread_members(&items.root_events).await;
+            let frontier = self.descendant_thread_members(&items.root_events).await;
             packed_historic_filters(&items, &frontier)
         };
         self.historic_sync_with_options(
@@ -8167,36 +8353,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn maintenance_reconciliation_keeps_subscription_ids_for_removal_only_coverage() {
-        let first_id = EventId::from_byte_array([1; 32]);
-        let second_id = EventId::from_byte_array([2; 32]);
-        let first_filter = Filter::new().event(first_id);
-        let second_filter = Filter::new().event(second_id);
-        let first = first_filter.as_json();
-        let second = second_filter.as_json();
-        let subscription_ids = vec![SubscriptionId::new("live-descendants")];
-        let mut coverage = DescendantLiveCoverage {
-            covered_filters: HashSet::from([first.clone(), second.clone()]),
-            desired_filters: HashSet::from([first, second.clone()]),
-            fallback_filters: vec![first_filter, second_filter.clone()],
-            subscription_ids: subscription_ids.clone(),
-        };
-
-        assert!(coverage.retain_for_removal_only_change(&HashSet::from([second.clone()])));
-        assert_eq!(coverage.subscription_ids, subscription_ids);
-        coverage.fallback_filters = vec![second_filter];
-        assert_eq!(coverage.fallback_filters.len(), 1);
-        assert!(coverage.fallback_filters[0].as_json().contains(&second));
-        assert!(!coverage.fallback_filters[0]
-            .as_json()
-            .contains(&first_id.to_hex()));
-        // A later maintenance tick sees stable desired coverage and still
-        // leaves the physical subscription untouched.
-        assert!(coverage.retain_for_removal_only_change(&HashSet::from([second])));
-        assert_eq!(coverage.subscription_ids, subscription_ids);
-    }
-
     #[test]
     fn retained_list_does_not_mask_an_empty_new_outbox_response() {
         let author_with_retained_list = Keys::generate().public_key();
@@ -8333,6 +8489,147 @@ mod tests {
         assert_eq!(descendant_event_coordinate(&malformed_addressable), None);
     }
 
+    #[tokio::test]
+    async fn descendant_frontier_recurses_through_event_and_coordinate_references() {
+        let keys = Keys::generate();
+        let root = EventBuilder::new(Kind::GitIssue, "root")
+            .finalize(&keys)
+            .expect("build root");
+        let addressable = EventBuilder::new(Kind::Custom(30_023), "first descendant")
+            .tags([Tag::identifier("thread-member"), Tag::event(root.id)])
+            .finalize(&keys)
+            .expect("build addressable descendant");
+        let coordinate = descendant_event_coordinate(&addressable).unwrap();
+        let coordinate_child = EventBuilder::new(Kind::TextNote, "coordinate child")
+            .tag(Tag::custom("a", [coordinate.clone()]))
+            .finalize(&keys)
+            .expect("build coordinate child");
+        let grandchild = EventBuilder::new(Kind::TextNote, "grandchild")
+            .tag(Tag::event(coordinate_child.id))
+            .finalize(&keys)
+            .expect("build grandchild");
+        let unrelated = EventBuilder::new(Kind::TextNote, "unrelated")
+            .finalize(&keys)
+            .expect("build unrelated event");
+        let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded());
+        for event in [
+            &root,
+            &addressable,
+            &coordinate_child,
+            &grandchild,
+            &unrelated,
+        ] {
+            database.save_event(event).await.expect("save test event");
+        }
+
+        let frontier =
+            recursive_descendant_frontier(&database, &HashSet::from([root.id]), 500).await;
+
+        assert_eq!(
+            frontier.event_ids,
+            HashSet::from([addressable.id, coordinate_child.id, grandchild.id])
+        );
+        assert_eq!(frontier.coordinates, HashSet::from([coordinate]));
+        assert!(!frontier.event_ids.contains(&unrelated.id));
+    }
+
+    #[tokio::test]
+    async fn descendant_frontier_stops_at_the_depth_bound() {
+        let keys = Keys::generate();
+        let root = EventBuilder::new(Kind::GitIssue, "root")
+            .finalize(&keys)
+            .expect("build root");
+        let mut chain = Vec::new();
+        let mut parent = root.id;
+        for depth in 1..=MAX_DESCENDANT_FRONTIER_DEPTH + 1 {
+            let child = EventBuilder::new(Kind::TextNote, format!("depth {depth}"))
+                .tag(Tag::event(parent))
+                .finalize(&keys)
+                .expect("build descendant");
+            parent = child.id;
+            chain.push(child);
+        }
+        let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded());
+        database.save_event(&root).await.expect("save root");
+        for event in &chain {
+            database.save_event(event).await.expect("save descendant");
+        }
+
+        let frontier =
+            recursive_descendant_frontier(&database, &HashSet::from([root.id]), 500).await;
+
+        assert_eq!(frontier.event_ids.len(), MAX_DESCENDANT_FRONTIER_DEPTH);
+        assert!(chain[..MAX_DESCENDANT_FRONTIER_DEPTH]
+            .iter()
+            .all(|event| frontier.event_ids.contains(&event.id)));
+        assert!(!frontier
+            .event_ids
+            .contains(&chain[MAX_DESCENDANT_FRONTIER_DEPTH].id));
+    }
+
+    #[tokio::test]
+    async fn descendant_frontier_preserves_direct_members_and_bounds_recursive_events() {
+        let keys = Keys::generate();
+        let root = EventBuilder::new(Kind::GitIssue, "root")
+            .finalize(&keys)
+            .expect("build root");
+        let direct = EventBuilder::new(Kind::TextNote, "direct member")
+            .tag(Tag::event(root.id))
+            .custom_created_at(Timestamp::from_secs(1))
+            .finalize(&keys)
+            .expect("build direct member");
+        let direct_siblings: Vec<Event> = (0..3)
+            .map(|index| {
+                EventBuilder::new(Kind::TextNote, format!("direct sibling {index}"))
+                    .tag(Tag::event(root.id))
+                    .custom_created_at(Timestamp::from_secs(2 + index))
+                    .finalize(&keys)
+                    .expect("build direct sibling")
+            })
+            .collect();
+        let mut children = Vec::new();
+        for created_at in [30, 10, 20] {
+            children.push(
+                EventBuilder::new(Kind::TextNote, format!("created at {created_at}"))
+                    .tag(Tag::event(direct.id))
+                    .custom_created_at(Timestamp::from_secs(created_at))
+                    .finalize(&keys)
+                    .expect("build child"),
+            );
+        }
+        let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded());
+        database
+            .save_event(&direct)
+            .await
+            .expect("save direct member");
+        for event in &direct_siblings {
+            database
+                .save_event(event)
+                .await
+                .expect("save direct sibling");
+        }
+        for event in &children {
+            database.save_event(event).await.expect("save child");
+        }
+
+        let frontier = recursive_descendant_frontier_with_limits(
+            &database,
+            &HashSet::from([root.id]),
+            MAX_DESCENDANT_FRONTIER_DEPTH,
+            2,
+        )
+        .await;
+
+        assert_eq!(frontier.event_ids.len(), 3);
+        assert!(!frontier.event_ids.contains(&direct.id));
+        assert!(direct_siblings
+            .iter()
+            .all(|event| frontier.event_ids.contains(&event.id)));
+        assert!(children
+            .iter()
+            .all(|child| !frontier.event_ids.contains(&child.id)));
+    }
+
     #[test]
     fn descendant_frontier_queries_event_and_coordinate_tag_variants() {
         let since = Timestamp::from_secs(1234);

tests/common/relay.rs

@@ -86,6 +86,7 @@ struct RelayOptions {
     deletion_lifecycle: Option<DeletionLifecycleOptions>,
     rejected_hot_cache_duration_secs: Option<u64>,
     relay_max_subscriptions: Option<usize>,
+    sync_recursive_descendant_limit: Option<usize>,
     /// Run with the production outbound target policy (reject non-global
     /// event-directed sync targets). The fixture default is permissive
     /// because the entire test infrastructure lives on loopback.
@@ -427,6 +428,31 @@ impl TestRelay {
         .await
     }
 
+    /// Start a persistent syncing relay with a small recursive-descendant
+    /// allowance so saturation and restart reconstruction can be exercised.
+    pub async fn start_on_reservation_persistent_sync_with_recursive_limit(
+        reservation: PortReservation,
+        bootstrap_relay_url: Option<String>,
+        disable_negentropy: bool,
+        git_data_path: PathBuf,
+        relay_data_path: PathBuf,
+        recursive_descendant_limit: usize,
+    ) -> Self {
+        Self::start_internal(
+            reservation,
+            RelayOptions {
+                bootstrap_relay_url,
+                disable_negentropy,
+                lmdb_backend: true,
+                git_data_path: Some(git_data_path),
+                relay_data_path: Some(relay_data_path),
+                sync_recursive_descendant_limit: Some(recursive_descendant_limit),
+                ..RelayOptions::default()
+            },
+        )
+        .await
+    }
+
     /// Start a persistent syncing relay with an explicit Sync+ fallback set.
     pub async fn start_on_reservation_persistent_sync_with_fallback(
         reservation: PortReservation,
@@ -632,6 +658,10 @@ impl TestRelay {
             cmd.env("NGIT_RELAY_MAX_SUBSCRIPTIONS", limit.to_string());
         }
 
+        if let Some(limit) = options.sync_recursive_descendant_limit {
+            cmd.env("NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT", limit.to_string());
+        }
+
         // Add negentropy disable flag if requested
         if options.disable_negentropy {
             cmd.env("NGIT_SYNC_DISABLE_NEGENTROPY", "true");

tests/sync/descendant_sync.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 
 use nostr_sdk::prelude::*;
 
-use crate::common::{sync_helpers::*, TestRelay};
+use crate::common::{reserve_port, sync_helpers::*, TestRelay};
 
 async fn wait_for_log(path: &std::path::Path, needle: &str, timeout: Duration) -> bool {
     let deadline = tokio::time::Instant::now() + timeout;
@@ -22,11 +22,132 @@ async fn wait_for_log(path: &std::path::Path, needle: &str, timeout: Duration) -
     }
 }
 
-/// Events which reference a direct repository-thread member, but not the
-/// repository or its root event, are recovered by the scheduled historic
-/// pass. The recovered events do not recursively extend the frontier.
+/// A tiny configured frontier exercises saturation and LMDB reconstruction
+/// without needing a naturally popular production thread.
 #[tokio::test]
-async fn historic_sync_recovers_one_generation_of_parent_only_descendants() {
+async fn historic_recursive_cap_is_enforced_and_reconstructed_after_restart() {
+    const LIMIT: usize = 2;
+
+    let source = TestRelay::start_with_relay_max_subscriptions(4).await;
+    let keys = Keys::generate();
+    let repo_id = "restart-safe-recursive-cap";
+    let source_domains = [source.domain()];
+    let source_refs = source_domains
+        .iter()
+        .map(String::as_str)
+        .collect::<Vec<_>>();
+    let (_announcement, _source_git) =
+        setup_announcement_on_relay(&source, &keys, &source_refs, repo_id).await;
+    let source_client = TestClient::new(source.url(), keys.clone())
+        .await
+        .expect("connect to source relay");
+    let issue = build_layer2_issue_event(&keys, &repo_coord(&keys, repo_id), "Recursive cap root")
+        .expect("build issue");
+    let direct = build_layer3_reply_with_e_tag(&keys, &issue.id, "Direct branch owner")
+        .expect("build direct branch owner");
+    let first = build_layer3_reply_with_e_tag(&keys, &direct.id, "First recursive descendant")
+        .expect("build first recursive descendant");
+    let second = build_layer3_reply_with_e_tag(&keys, &first.id, "Second recursive descendant")
+        .expect("build second recursive descendant");
+    let overflow = build_layer3_reply_with_e_tag(&keys, &second.id, "Beyond bounded frontier")
+        .expect("build descendant beyond bounded frontier");
+    for event in std::iter::once(&issue)
+        .chain(std::iter::once(&direct))
+        .chain([&first, &second, &overflow])
+    {
+        source_client
+            .send_event(event)
+            .await
+            .expect("seed source event");
+    }
+
+    let syncing_git_dir = tempfile::tempdir().expect("create syncing git dir");
+    let syncing_data_dir = tempfile::tempdir().expect("create syncing relay data dir");
+    let syncing = TestRelay::start_on_reservation_persistent_sync_with_recursive_limit(
+        reserve_port(),
+        None,
+        true,
+        syncing_git_dir.path().to_path_buf(),
+        syncing_data_dir.path().to_path_buf(),
+        LIMIT,
+    )
+    .await;
+    let domains = [source.domain(), syncing.domain()];
+    let domain_refs = domains.iter().map(String::as_str).collect::<Vec<_>>();
+    let (_target_announcement, _target_git) =
+        setup_announcement_on_relay(&syncing, &keys, &domain_refs, repo_id).await;
+
+    for descendant in [&first, &second] {
+        assert!(
+            wait_for_event_on_relay(
+                syncing.url(),
+                Filter::new().id(descendant.id),
+                Duration::from_secs(30),
+            )
+            .await,
+            "the bounded frontier should recover {}",
+            descendant.content
+        );
+    }
+    assert!(
+        wait_for_log(
+            &syncing.log_path(),
+            "Started queued descendant fallback query",
+            Duration::from_secs(5),
+        )
+        .await,
+        "the constrained source should reach saturation through historic fallback"
+    );
+    assert!(
+        !wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(overflow.id),
+            Duration::from_secs(10),
+        )
+        .await,
+        "a full branch must not query through its last retained descendant"
+    );
+
+    let syncing = syncing.restart().await;
+    let late_child = build_layer3_reply_with_e_tag(&keys, &second.id, "Late recursive child")
+        .expect("build late recursive child");
+    let progress_marker =
+        build_layer3_reply_with_e_tag(&keys, &issue.id, "Unmetered direct progress marker")
+            .expect("build direct progress marker");
+    source_client.send_event(&late_child).await.unwrap();
+    source_client.send_event(&progress_marker).await.unwrap();
+    assert!(
+        wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(progress_marker.id),
+            Duration::from_secs(20),
+        )
+        .await,
+        "the restarted relay should continue direct/core live sync"
+    );
+    for excluded in [&overflow, &late_child] {
+        assert!(
+            !wait_for_event_on_relay(
+                syncing.url(),
+                Filter::new().id(excluded.id),
+                Duration::from_secs(10),
+            )
+            .await,
+            "restart must reconstruct the full branch and leave {} outside its query frontier",
+            excluded.content
+        );
+    }
+
+    source_client.disconnect().await;
+    syncing.stop().await;
+    source.stop().await;
+}
+
+/// Events which reference a direct repository-thread member, but not the
+/// repository or its root event, are recovered recursively by the scheduled
+/// historic pass while the direct member's branch remains below its allowance.
+#[tokio::test]
+async fn historic_sync_recovers_recursive_parent_only_descendants() {
     let source = TestRelay::start_with_relay_max_subscriptions(4).await;
     let keys = Keys::generate();
     let repo_id = "scheduled-descendant-history";
@@ -92,13 +213,13 @@ async fn historic_sync_recovers_one_generation_of_parent_only_descendants() {
         "the constrained source should exercise EOSE-closing fallback"
     );
     assert!(
-        !wait_for_event_on_relay(
+        wait_for_event_on_relay(
             syncing.url(),
             Filter::new().id(recursive.id),
-            Duration::from_secs(7),
+            Duration::from_secs(20),
         )
         .await,
-        "recovered descendants must not recursively extend the frontier"
+        "a retained recursive descendant must extend its branch frontier"
     );
 
     source_client.disconnect().await;