From e922e14e3ec4b898c111b2100cd63dddbe2fcdb1 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Fri, 13 Feb 2026 19:59:36 +0000 Subject: feat: add SyncLevel to sync system for purgatory announcement state-only sync Purgatory announcements need state events (kind 30618) synced from external relays, but not full L2/L3 events (patches, issues, PRs) which would be rejected anyway. This implements the SyncLevel concept from the design doc (decision #6): - Add SyncLevel enum (Full vs StateOnly) to RepoSyncNeeds - When announcement enters purgatory during sync, register in RepoSyncIndex with SyncLevel::StateOnly - Add build_sync_level_aware_filters() that partitions repos by level: StateOnly repos only get state event filters (kind 30618) - Update derive_relay_targets to track state_only_repos separately - Update compute_actions to handle both repo sets - SelfSubscriber always uses SyncLevel::Full (promoted repos) --- src/sync/self_subscriber.rs | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) (limited to 'src/sync/self_subscriber.rs') diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs index 3cc408d..db16c62 100644 --- a/src/sync/self_subscriber.rs +++ b/src/sync/self_subscriber.rs @@ -16,7 +16,7 @@ use nostr_sdk::Timestamp; use tokio::sync::broadcast::error::RecvError; use tokio::sync::{broadcast, mpsc}; -use super::{AddFilters, RepoSyncIndex, RepoSyncNeeds}; +use super::{AddFilters, RepoSyncIndex, RepoSyncNeeds, SyncLevel}; // ============================================================================= // LoopControl - Result of notification processing @@ -58,6 +58,7 @@ impl PendingUpdates { let entry = self.repos.entry(repo_id).or_insert_with(|| RepoSyncNeeds { relays: HashSet::new(), root_events: HashSet::new(), + sync_level: SyncLevel::Full, }); entry.relays.extend(relays); entry.root_events.extend(root_events); @@ -475,6 +476,7 @@ impl SelfSubscriber { .or_insert_with(|| RepoSyncNeeds { relays: HashSet::new(), root_events: HashSet::new(), + sync_level: SyncLevel::Full, }); entry.relays.extend(needs.relays); entry.root_events.extend(needs.root_events); @@ -499,21 +501,26 @@ impl SelfSubscriber { continue; } - // Build filters for these repos - let filters = crate::sync::filters::build_layer2_and_layer3_filters( + // Build filters for these repos (sync-level-aware) + let filters = crate::sync::filters::build_sync_level_aware_filters( &needs.repos, + &needs.state_only_repos, &needs.root_events, None, ); // Log before moving values - let repo_count = needs.repos.len(); + let repo_count = needs.repos.len() + needs.state_only_repos.len(); let event_count = needs.root_events.len(); + // Combine all repos into pending items + let mut all_repos = needs.repos; + all_repos.extend(needs.state_only_repos); + let action = AddFilters { relay_url: relay_url.clone(), items: crate::sync::PendingItems { - repos: needs.repos, + repos: all_repos, root_events: needs.root_events, }, filters, -- cgit v1.2.3 From e22021f0b248ebcf3bd09210d59b2cdb4701032f Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Wed, 18 Feb 2026 19:41:29 +0000 Subject: fix: simplify purgatory sync - fix SelfSubscriber sync_level upgrade and negentropy fallback Three targeted fixes for purgatory announcement sync: 1. SelfSubscriber sync_level upgrade: After or_insert_with in process_batch, always set entry.sync_level = SyncLevel::Full so that when a promoted announcement is broadcast via notify_event and SelfSubscriber receives it, an existing StateOnly entry gets upgraded to Full and PR event subscriptions are triggered immediately (not delayed up to 24h). 2. Negentropy fallback filter split: In handle_eose, when falling back from negentropy to REQ+EOSE, split batch_repos by SyncLevel and call build_sync_level_aware_filters instead of build_layer2_and_layer3_filters. Prevents StateOnly (purgatory) repos from getting Layer 2 #a/#A/#q filters prematurely, which caused nostr-sdk client deduplication to permanently drop PR events after orphan rejection. 3. Recompute sync filters after announcement batch EOSE: Add recompute_new_sync_filters_for_relay calls at all three batch-completion paths in handle_eose for generic filter (announcement) batches. This triggers state-only subscriptions for any purgatory repos registered during that batch, fixing the 24h delay before state event sync starts. 4. User-submitted purgatory announcements: Add repo_sync_index field to PolicyContext with setter/getter, wire in main.rs after SyncManager creation, and register in AcceptPurgatory handler so user-submitted announcements get StateOnly sync started immediately. 5. Update archive tests: test_archive_without_state_events_does_not_sync_git updated to reflect that StateOnly subscription now proactively fetches state events from source relays. test_archive_read_only_creates_bare_repo un-ignored as it now works end-to-end. --- src/main.rs | 7 +++++ src/nostr/builder.rs | 54 +++++++++++++++++++++++++++++++++++++ src/nostr/policy/mod.rs | 19 +++++++++++++ src/sync/mod.rs | 66 ++++++++++++++++++++++++++++++++++++++++++--- src/sync/self_subscriber.rs | 4 +++ tests/archive_read_only.rs | 63 +++++++++++++++++++++++++++---------------- 6 files changed, 187 insertions(+), 26 deletions(-) (limited to 'src/sync/self_subscriber.rs') diff --git a/src/main.rs b/src/main.rs index ab6ede7..ebe05a3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -132,6 +132,13 @@ async fn main() -> Result<()> { // Get a reference to the rejected events index for shutdown persistence let shutdown_rejected_index = sync_manager.rejected_events_index(); + // Wire repo_sync_index into write policy so user-submitted purgatory announcements + // get registered for state event sync immediately (Fix 3). + let repo_sync_index = sync_manager.repo_sync_index(); + relay_with_db + .write_policy + .set_repo_sync_index(repo_sync_index); + tokio::spawn(async move { sync_manager.run().await; }); diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index aff12a6..8d1e461 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -17,6 +17,7 @@ use crate::nostr::policy::{ AnnouncementPolicy, AnnouncementResult, PolicyContext, PrEventPolicy, ReferenceResult, RelatedEventPolicy, StatePolicy, StateResult, }; +use crate::sync::{RepoSyncIndex, RepoSyncNeeds, SyncLevel}; /// Type alias for the shared database used by the relay pub type SharedDatabase = Arc; @@ -98,6 +99,14 @@ impl Nip34WritePolicy { self.ctx.set_local_relay(relay); } + /// Set the repo sync index so that user-submitted purgatory announcements can + /// be registered for state event sync immediately. + /// + /// This must be called after SyncManager is created. + pub fn set_repo_sync_index(&self, index: RepoSyncIndex) { + self.ctx.set_repo_sync_index(index); + } + /// Handle repository announcement event async fn handle_announcement(&self, event: &Event) -> WritePolicyResult { let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()); @@ -146,6 +155,51 @@ impl Nip34WritePolicy { "Accepted announcement to purgatory: {} (waiting for git data)", event_id_str ); + + // Register repo in repo_sync_index with StateOnly level so that + // state event sync starts promptly via the next batch EOSE recompute. + // This handles user-submitted purgatory announcements - the SelfSubscriber + // only sees DB events, so it won't pick these up automatically. + if let Some(repo_sync_index) = self.ctx.get_repo_sync_index() { + if let Ok(announcement) = + RepositoryAnnouncement::from_event(event.clone()) + { + use std::collections::HashSet; + let repo_id = format!( + "30617:{}:{}", + event.pubkey, + announcement.identifier + ); + + // Extract relay URLs from the announcement event tags + let relays: HashSet = event + .tags + .iter() + .flat_map(|tag| { + let tag_vec = tag.as_slice(); + if !tag_vec.is_empty() && tag_vec[0] == "relays" { + tag_vec[1..].iter().map(|s| s.to_string()).collect::>() + } else { + vec![] + } + }) + .collect(); + + let mut index = repo_sync_index.write().await; + index.entry(repo_id.clone()).or_insert_with(|| RepoSyncNeeds { + relays, + root_events: HashSet::new(), + sync_level: SyncLevel::StateOnly, + }); + drop(index); + + tracing::debug!( + repo_id = %repo_id, + "Registered purgatory announcement in repo_sync_index as StateOnly" + ); + } + } + WritePolicyResult::Reject { status: true, // Client sees OK message: "purgatory: won't be served until git data arrives".into(), diff --git a/src/nostr/policy/mod.rs b/src/nostr/policy/mod.rs index 1566b6c..c958586 100644 --- a/src/nostr/policy/mod.rs +++ b/src/nostr/policy/mod.rs @@ -20,6 +20,7 @@ pub use crate::git::sync::AlignmentResult; use super::SharedDatabase; use crate::purgatory::Purgatory; +use crate::sync::RepoSyncIndex; use nostr_relay_builder::LocalRelay; use std::sync::Arc; @@ -34,6 +35,8 @@ pub struct PolicyContext { pub local_relay: Arc>>, /// Configuration reference for policy settings (includes blacklists) pub config: crate::config::Config, + /// Repo sync index for registering purgatory announcements (set after SyncManager creation) + pub repo_sync_index: Arc>>, } impl PolicyContext { @@ -51,6 +54,7 @@ impl PolicyContext { purgatory, local_relay: Arc::new(std::sync::RwLock::new(None)), config, + repo_sync_index: Arc::new(std::sync::RwLock::new(None)), } } @@ -68,4 +72,19 @@ impl PolicyContext { let guard = self.local_relay.read().unwrap(); guard.clone() } + + /// Set the repo sync index after SyncManager has been created. + /// + /// This allows purgatory announcements submitted by users to be registered + /// in the sync index so state event sync starts promptly. + pub fn set_repo_sync_index(&self, index: RepoSyncIndex) { + let mut guard = self.repo_sync_index.write().unwrap(); + *guard = Some(index); + } + + /// Get a clone of the repo sync index if it has been set. + pub fn get_repo_sync_index(&self) -> Option { + let guard = self.repo_sync_index.read().unwrap(); + guard.clone() + } } diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 519017b..916e2b0 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -700,6 +700,14 @@ impl SyncManager { self.rejected_events_index.save_to_disk(path) } + /// Get a clone of the repo sync index Arc. + /// + /// This allows the write policy to register user-submitted purgatory announcements + /// in the sync index so that state event sync starts promptly. + pub fn repo_sync_index(&self) -> RepoSyncIndex { + self.repo_sync_index.clone() + } + /// Handle EOSE (End Of Stored Events) for a subscription /// /// This method: @@ -951,9 +959,29 @@ impl SyncManager { // Create REQ+EOSE subscriptions using original semantic filters // This queries by kind/author/tags instead of by ID, which may - // succeed even when ID-based queries fail - let fallback_filters = filters::build_layer2_and_layer3_filters( - &batch_repos, + // succeed even when ID-based queries fail. + // Split batch_repos by SyncLevel to avoid sending Layer 2 filters + // (#a/#A/#q) for StateOnly (purgatory) repos - those PRs would be + // rejected as orphan and then silently dropped by nostr-sdk deduplication. + let (full_repos, state_only_repos) = { + let repo_index = self.repo_sync_index.read().await; + let mut full = HashSet::new(); + let mut state_only = HashSet::new(); + for repo_ref in &batch_repos { + match repo_index.get(repo_ref).map(|n| n.sync_level) { + Some(SyncLevel::StateOnly) => { + state_only.insert(repo_ref.clone()); + } + _ => { + full.insert(repo_ref.clone()); + } + } + } + (full, state_only) + }; + let fallback_filters = filters::build_sync_level_aware_filters( + &full_repos, + &state_only_repos, &batch_root_events, None, ); @@ -1033,12 +1061,24 @@ impl SyncManager { { let mut completed_batch = batches.remove(idx); completed_batch.failed = true; // Mark as failed + let is_generic = + completed_batch.items.repos.is_empty() + && completed_batch.items.root_events.is_empty(); if batches.is_empty() { pending.remove(&relay_url_for_fallback); } drop(pending); self.confirm_batch(&relay_url_for_fallback, completed_batch) .await; + // For generic filter (announcement) batches, recompute filters + // so any purgatory repos registered during this batch get + // state-only subscriptions triggered. + if is_generic { + self.recompute_new_sync_filters_for_relay( + &relay_url_for_fallback, + ) + .await; + } } } return; @@ -1132,12 +1172,24 @@ impl SyncManager { if let Some(batches) = pending.get_mut(&relay_url_for_retry) { if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id) { let completed_batch = batches.remove(idx); + let is_generic = + completed_batch.items.repos.is_empty() + && completed_batch.items.root_events.is_empty(); if batches.is_empty() { pending.remove(&relay_url_for_retry); } drop(pending); self.confirm_batch(&relay_url_for_retry, completed_batch) .await; + // For generic filter (announcement) batches, recompute filters + // so any purgatory repos registered during this batch get + // state-only subscriptions triggered. + if is_generic { + self.recompute_new_sync_filters_for_relay( + &relay_url_for_retry, + ) + .await; + } } } return; @@ -1148,6 +1200,8 @@ impl SyncManager { // 3. Batch complete - extract and remove let completed_batch = batches.remove(batch_idx); + let is_generic = completed_batch.items.repos.is_empty() + && completed_batch.items.root_events.is_empty(); // Clean up empty relay entry if batches.is_empty() { @@ -1159,6 +1213,12 @@ impl SyncManager { // 4. Confirm the batch (moves items to RelayState) self.confirm_batch(relay_url, completed_batch).await; + + // 5. For generic filter (announcement) batches, recompute sync filters so any + // purgatory repos registered during this batch get state-only subscriptions triggered. + if is_generic { + self.recompute_new_sync_filters_for_relay(relay_url).await; + } } /// Confirm a completed batch by moving items to RelayState diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs index db16c62..70c3dbf 100644 --- a/src/sync/self_subscriber.rs +++ b/src/sync/self_subscriber.rs @@ -478,6 +478,10 @@ impl SelfSubscriber { root_events: HashSet::new(), sync_level: SyncLevel::Full, }); + // Upgrade sync_level to Full - this handles the case where the entry + // already exists as StateOnly (purgatory announcement) and is now being + // promoted (git data arrived and the event was broadcast via notify_event). + entry.sync_level = SyncLevel::Full; entry.relays.extend(needs.relays); entry.root_events.extend(needs.root_events); diff --git a/tests/archive_read_only.rs b/tests/archive_read_only.rs index e388ae5..069b3b7 100644 --- a/tests/archive_read_only.rs +++ b/tests/archive_read_only.rs @@ -55,7 +55,6 @@ use std::time::Duration; /// 5. Verify bare repository is created and git data is synced /// 6. Verify git pushes are rejected (read-only mode) #[tokio::test] -#[ignore] // Requires SyncLevel implementation (Phase 3) - purgatory announcements don't trigger per-repo sync yet async fn test_archive_read_only_creates_bare_repo() { // 1. Start source relay let source_relay = TestRelay::start().await; @@ -264,24 +263,24 @@ async fn test_archive_read_only_creates_bare_repo() { source_relay.stop().await; } -/// Test that archive mode without state events does NOT sync git data. +/// Test that archive mode proactively syncs state events and git data +/// when the source relay has state events available. /// -/// This verifies the security model: archive mode only syncs git data -/// when there are state events to validate against. +/// With StateOnly sync now implemented, purgatory announcements subscribe +/// to state events from the relays listed in the announcement. This means +/// the archive relay will: +/// 1. Sync the announcement → purgatory → register as StateOnly in repo_sync_index +/// 2. Subscribe to state events (kind 30618) on source relay +/// 3. Receive the state event → purgatory sync triggered +/// 4. Fetch git data from source relay's clone URL /// -/// With announcement purgatory, the flow is: -/// 1. Send announcement to source relay (goes to purgatory) -/// 2. Send state event to source relay (goes to purgatory) -/// 3. Push git data to source relay (promotes announcement and state event) -/// 4. Start archive relay with sync from source -/// 5. Archive relay syncs the promoted announcement -/// 6. Verify git data is NOT synced (archive has no state event to authorize git fetch) +/// This test verifies the full sync chain works end-to-end for archive mode. #[tokio::test] -async fn test_archive_without_state_events_does_not_sync_git() { +async fn test_archive_syncs_state_events_and_git_data_via_state_only_subscription() { // 1. Start source relay let source_relay = TestRelay::start().await; let keys = Keys::generate(); - let identifier = "archive-no-state-repo"; + let identifier = "archive-state-only-sync-repo"; // Pre-allocate archive relay port let archive_port = TestRelay::find_free_port(); @@ -295,6 +294,7 @@ async fn test_archive_without_state_events_does_not_sync_git() { let npub = keys.public_key().to_bech32().expect("Failed to get npub"); // 3. Create and send announcement listing BOTH relays + // The archive relay will subscribe to state events on BOTH listed relays let announcement = create_repo_announcement( &keys, &[&source_relay.domain(), &archive_domain], @@ -337,6 +337,8 @@ async fn test_archive_without_state_events_does_not_sync_git() { ) .expect("Failed to create state event"); + let state_event_id = state_event.id; + source_client .send_event(&state_event) .await @@ -348,9 +350,12 @@ async fn test_archive_without_state_events_does_not_sync_git() { push_to_relay(temp_dir.path(), &source_relay.domain(), &npub, identifier) .expect("Push to source should succeed"); - tokio::time::sleep(Duration::from_millis(500)).await; + // Wait for state event to be promoted on source relay + wait_for_event_served(source_relay.url(), &state_event_id, Duration::from_secs(5)) + .await + .expect("State event should be served on source relay after push"); - // 6. Start archive relay (without state event - we don't send state event to archive) + // 6. Start archive relay - StateOnly subscription will proactively fetch state events let archive_relay = TestRelay::start_with_archive_and_sync( archive_port, Some(source_relay.url().to_string()), @@ -360,15 +365,28 @@ async fn test_archive_without_state_events_does_not_sync_git() { ) .await; - // Wait for sync + // Wait for sync connection wait_for_sync_connection(archive_relay.url(), 1, Duration::from_secs(5)) .await .expect("Sync connection should establish"); - // Give time for sync to fetch announcement - tokio::time::sleep(Duration::from_secs(3)).await; + // 7. Wait for state event to be served on archive relay + // The StateOnly subscription fetches the state event from source relay, + // which then triggers purgatory sync and git data fetch. + let found = wait_for_event_served( + archive_relay.url(), + &state_event_id, + Duration::from_secs(30), // Allow time for sync + git fetch + ) + .await; + + assert!( + found.is_ok(), + "State event should be served on archive after StateOnly subscription fetches it: {:?}", + found.err() + ); - // 7. Verify bare repository was created (announcement was synced and accepted to purgatory) + // 8. Verify bare repository was created let repo_path = archive_relay .git_data_path() .join(format!("{}/{}.git", npub, identifier)); @@ -378,8 +396,7 @@ async fn test_archive_without_state_events_does_not_sync_git() { "Bare repository should be created for archive announcement" ); - // 8. Verify git data was NOT synced (no state events on archive to trigger git fetch) - // Check that the commit does NOT exist in the archive relay's repo + // 9. Verify git data was synced via the state event chain let output = tokio::process::Command::new("git") .args(["cat-file", "-t", &commit_hash]) .current_dir(&repo_path) @@ -389,8 +406,8 @@ async fn test_archive_without_state_events_does_not_sync_git() { let commit_exists = output.map(|o| o.status.success()).unwrap_or(false); assert!( - !commit_exists, - "Git data should NOT be synced without state events (security: validates against Nostr state)" + commit_exists, + "Git data should be synced via StateOnly subscription → state event → git fetch chain" ); // Cleanup -- cgit v1.2.3 From f62ef12fb84e2210f9a0a67a5e1e574a8ee66c16 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Mon, 23 Feb 2026 12:48:26 +0000 Subject: refactor: replace inline purgatory sync registration with timer-only approach Remove the redundant inline kind-30617 registration block from the sync event loop and the three is_generic/recompute_new_sync_filters_for_relay calls from confirm_batch error paths. The purgatory announcement sync timer (run_purgatory_announcement_sync) is now the sole registration path. Consolidate NGIT_SYNC_BATCH_WINDOW_MS and NGIT_PURGATORY_SYNC_INTERVAL_MS into a single NGIT_TEST=1 flag that sets both timers to 200ms, replacing two ad-hoc env vars with one reusable test-mode flag. --- src/sync/mod.rs | 103 ++++++---------------------------- src/sync/self_subscriber.rs | 12 ++-- tests/common/relay.rs | 2 +- tests/sync/maintainer_reprocessing.rs | 2 +- 4 files changed, 26 insertions(+), 93 deletions(-) (limited to 'src/sync/self_subscriber.rs') diff --git a/src/sync/mod.rs b/src/sync/mod.rs index ed5b6e7..44efbf0 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -399,21 +399,24 @@ async fn run_daily_timer( /// Background task that periodically syncs purgatory announcements into repo_sync_index. /// -/// Runs every 5 seconds. For each announcement currently in purgatory, ensures there -/// is a `StateOnly` entry in `repo_sync_index`. New entries trigger `handle_new_sync_filters` -/// which connects to the relay URLs listed in the announcement and subscribes to state -/// events (kind 30618). +/// Runs every 5 seconds by default (200ms when `NGIT_TEST=1`). +/// For each announcement currently in purgatory, ensures there is a `StateOnly` entry in +/// `repo_sync_index`. New entries trigger `handle_new_sync_filters` which connects to the +/// relay URLs listed in the announcement and subscribes to state events (kind 30618). /// -/// This covers two cases: -/// - Sync-path announcements: registered inline during event processing, but this -/// provides a safety net in case the inline registration was missed. +/// This is the sole registration path for purgatory announcements: +/// - Sync-path announcements: registered here within one interval of arriving. /// - User-submitted purgatory announcements: the SelfSubscriber never sees them -/// (they're rejected from DB), so this timer is the primary registration path. +/// (they're rejected from DB), so this timer is the only registration path. async fn run_purgatory_announcement_sync( sync_manager: Arc>, mut shutdown_rx: broadcast::Receiver<()>, ) { - let interval = Duration::from_secs(5); + let interval = if std::env::var("NGIT_TEST").as_deref() == Ok("1") { + Duration::from_millis(200) + } else { + Duration::from_secs(5) + }; loop { tokio::select! { _ = tokio::time::sleep(interval) => { @@ -1084,24 +1087,12 @@ impl SyncManager { { let mut completed_batch = batches.remove(idx); completed_batch.failed = true; // Mark as failed - let is_generic = - completed_batch.items.repos.is_empty() - && completed_batch.items.root_events.is_empty(); if batches.is_empty() { pending.remove(&relay_url_for_fallback); } drop(pending); self.confirm_batch(&relay_url_for_fallback, completed_batch) .await; - // For generic filter (announcement) batches, recompute filters - // so any purgatory repos registered during this batch get - // state-only subscriptions triggered. - if is_generic { - self.recompute_new_sync_filters_for_relay( - &relay_url_for_fallback, - ) - .await; - } } } return; @@ -1195,24 +1186,12 @@ impl SyncManager { if let Some(batches) = pending.get_mut(&relay_url_for_retry) { if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id) { let completed_batch = batches.remove(idx); - let is_generic = - completed_batch.items.repos.is_empty() - && completed_batch.items.root_events.is_empty(); if batches.is_empty() { pending.remove(&relay_url_for_retry); } drop(pending); self.confirm_batch(&relay_url_for_retry, completed_batch) .await; - // For generic filter (announcement) batches, recompute filters - // so any purgatory repos registered during this batch get - // state-only subscriptions triggered. - if is_generic { - self.recompute_new_sync_filters_for_relay( - &relay_url_for_retry, - ) - .await; - } } } return; @@ -1223,8 +1202,6 @@ impl SyncManager { // 3. Batch complete - extract and remove let completed_batch = batches.remove(batch_idx); - let is_generic = completed_batch.items.repos.is_empty() - && completed_batch.items.root_events.is_empty(); // Clean up empty relay entry if batches.is_empty() { @@ -1236,12 +1213,6 @@ impl SyncManager { // 4. Confirm the batch (moves items to RelayState) self.confirm_batch(relay_url, completed_batch).await; - - // 5. For generic filter (announcement) batches, recompute sync filters so any - // purgatory repos registered during this batch get state-only subscriptions triggered. - if is_generic { - self.recompute_new_sync_filters_for_relay(relay_url).await; - } } /// Confirm a completed batch by moving items to RelayState @@ -1370,7 +1341,7 @@ impl SyncManager { /// to be batched and create Layer 2/3 filters before we mark sync complete. /// /// The 6-second delay is based on: - /// - Self-subscriber batch window: 5 seconds (configurable via NGIT_SYNC_BATCH_WINDOW_MS) + /// - Self-subscriber batch window: 5 seconds (200ms when `NGIT_TEST=1`) /// - Buffer for processing: 1 second /// /// Called after each batch is confirmed to detect completion. @@ -1785,7 +1756,6 @@ impl SyncManager { let eose_tx = self.eose_tx.as_ref().unwrap().clone(); let metrics_clone = self.metrics.clone(); let pending_sync_index = Arc::clone(&self.pending_sync_index); - let repo_sync_index = Arc::clone(&self.repo_sync_index); let health_tracker = Arc::clone(&self.health_tracker); let rejected_events_index = Arc::clone(&self.rejected_events_index); @@ -1827,50 +1797,13 @@ impl SyncManager { // For sync-triggered events that go to purgatory, trigger immediate sync // (instead of the default 3-minute delay for user-submitted events) + // + // Note: announcement events (kind 30617) are registered in repo_sync_index + // by the purgatory announcement sync timer (run_purgatory_announcement_sync) + // rather than inline here. if result == ProcessResult::Purgatory { - // Announcement events (kind 30617) - register in RepoSyncIndex with StateOnly - // so that state events (kind 30618) are synced for this purgatory announcement - if event.kind == Kind::GitRepoAnnouncement { - if let Some(identifier) = event.tags.iter().find_map(|tag| { - let tag_vec = tag.as_slice(); - if tag_vec.len() >= 2 && tag_vec[0] == "d" { - Some(tag_vec[1].to_string()) - } else { - None - } - }) { - let repo_id = format!("30617:{}:{}", event.pubkey, identifier); - - // Extract relay URLs from the purgatory entry - let relays = write_policy - .purgatory() - .find_announcement(&event.pubkey, &identifier) - .map(|entry| entry.relays) - .unwrap_or_default(); - - tracing::info!( - event_id = %event.id, - repo_id = %repo_id, - relay_count = relays.len(), - "Registering purgatory announcement in RepoSyncIndex with StateOnly level" - ); - - // Register in RepoSyncIndex with StateOnly level - let mut index = repo_sync_index.write().await; - let entry = index - .entry(repo_id) - .or_insert_with(|| RepoSyncNeeds { - relays: HashSet::new(), - root_events: HashSet::new(), - sync_level: SyncLevel::StateOnly, - }); - entry.relays.extend(relays); - // Don't upgrade sync_level if already Full - // (e.g., if announcement was promoted before this runs) - } - } // State events (kind 30618) - extract identifier and trigger immediate sync - else if event.kind.as_u16() == 30618 { + if event.kind.as_u16() == 30618 { if let Some(identifier) = event.tags.iter().find_map(|tag| { let tag_vec = tag.clone().to_vec(); if tag_vec.len() >= 2 && tag_vec[0] == "d" { diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs index 70c3dbf..ab10c49 100644 --- a/src/sync/self_subscriber.rs +++ b/src/sync/self_subscriber.rs @@ -126,14 +126,14 @@ impl SelfSubscriber { /// Get batch window from environment or use default /// - /// Reads `NGIT_SYNC_BATCH_WINDOW_MS` environment variable. + /// When `NGIT_TEST=1` is set, uses 200ms for faster test execution. /// Default: 5000ms (5 seconds) fn get_batch_window() -> Duration { - std::env::var("NGIT_SYNC_BATCH_WINDOW_MS") - .ok() - .and_then(|s| s.parse::().ok()) - .map(Duration::from_millis) - .unwrap_or(Duration::from_millis(5000)) + if std::env::var("NGIT_TEST").as_deref() == Ok("1") { + Duration::from_millis(200) + } else { + Duration::from_millis(5000) + } } /// Process a relay pool notification diff --git a/tests/common/relay.rs b/tests/common/relay.rs index 0ec9a2e..b1e96cf 100644 --- a/tests/common/relay.rs +++ b/tests/common/relay.rs @@ -204,7 +204,7 @@ impl TestRelay { .env("NGIT_GIT_DATA_PATH", git_data_dir.path()) .env("NGIT_DATABASE_BACKEND", "memory") // Force in-memory database for isolation .env("NGIT_OWNER_NPUB", &test_npub) - .env("NGIT_SYNC_BATCH_WINDOW_MS", "200") // Fast batch window for tests (200ms instead of 5s default) + .env("NGIT_TEST", "1") // Enable test mode: fast timers (200ms batch window, 200ms purgatory sync) .env("NGIT_SYNC_STARTUP_DELAY_SECS", "0") // No startup delay for faster tests .env("NGIT_SYNC_STARTUP_JITTER_MS", "0") // No jitter for tests .env("NGIT_SYNC_DISCONNECT_CHECK_INTERVAL_SECS", "1") // Fast reconnect attempts for tests diff --git a/tests/sync/maintainer_reprocessing.rs b/tests/sync/maintainer_reprocessing.rs index 61d8e14..ff1eb43 100644 --- a/tests/sync/maintainer_reprocessing.rs +++ b/tests/sync/maintainer_reprocessing.rs @@ -377,7 +377,7 @@ async fn test_multiple_maintainers_all_reprocessed() { println!("relay_b started at {}", relay_b.url()); // Give relay_b's SyncManager time to complete the initial negentropy sync with relay_a. - // The negentropy sync completes within ~200ms (NGIT_SYNC_BATCH_WINDOW_MS=200), but we + // The negentropy sync completes within ~200ms (NGIT_TEST=1 sets batch window to 200ms), but we // allow extra time for slow CI environments. tokio::time::sleep(Duration::from_secs(3)).await; println!("✓ relay_b synced from relay_a (maintainer announcements should be in hot cache)"); -- cgit v1.2.3