diff --git a/.env.example b/.env.example index e6d962b..e015275 100644 --- a/.env.example +++ b/.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 +# Default: 500 +# NGIT_SYNC_RECURSIVE_DESCENDANT_LIMIT=500 + # Maximum backoff time in seconds for sync relay reconnection # CLI: --sync-max-backoff-secs # Default: 3600 (1 hour) diff --git a/docs/explanation/architecture.md b/docs/explanation/architecture.md index 245fc94..12ad859 100644 --- a/docs/explanation/architecture.md +++ b/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:** diff --git a/docs/explanation/grasp-02-proactive-sync.md b/docs/explanation/grasp-02-proactive-sync.md index dc32fd1..88213e7 100644 --- a/docs/explanation/grasp-02-proactive-sync.md +++ b/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) diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 938fce2..9b89b81 100644 --- a/docs/reference/configuration.md +++ b/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 diff --git a/nix/module.nix b/nix/module.nix index c3fb1d4..19f7fe3 100644 --- a/nix/module.nix +++ b/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; diff --git a/src/config.rs b/src/config.rs index eb94d89..8967db9 100644 --- a/src/config.rs +++ b/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([ diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 3e4d838..18a6f91 100644 --- a/src/sync/mod.rs +++ b/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, desired_filters: HashSet, fallback_filters: Vec, subscription_ids: Vec, } -impl DescendantLiveCoverage { - fn retain_for_removal_only_change(&mut self, desired: &HashSet) -> 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 { )) } +/// 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, + 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, + 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::::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::::new(); + let mut event_branches = HashMap::>::new(); + let mut coordinate_branches = HashMap::>::new(); + let mut event_seeds = HashMap::>::new(); + let mut coordinate_seeds = HashMap::>::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::::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::>::new(); + let mut next_coordinate_seeds = HashMap::>::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) -> 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, + ) -> 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 = (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); diff --git a/tests/common/relay.rs b/tests/common/relay.rs index 8946964..feaf712 100644 --- a/tests/common/relay.rs +++ b/tests/common/relay.rs @@ -86,6 +86,7 @@ struct RelayOptions { deletion_lifecycle: Option, rejected_hot_cache_duration_secs: Option, relay_max_subscriptions: Option, + sync_recursive_descendant_limit: Option, /// 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, + 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"); diff --git a/tests/sync/descendant_sync.rs b/tests/sync/descendant_sync.rs index 0da058f..672f55b 100644 --- a/tests/sync/descendant_sync.rs +++ b/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::>(); + 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::>(); + 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;