From 11b1655638b5a328662187a27f85a84df60fc759 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Fri, 2 Jan 2026 13:53:47 +0000 Subject: sync: use purgatory don't save new events destined for purgatory events directly to db or serve on websockets don't download events already in purgatory via negentropy sync --- src/nostr/builder.rs | 5 +++++ src/purgatory/mod.rs | 29 +++++++++++++++++++++++++++++ src/sync/mod.rs | 52 +++++++++++++++++++++++++++++++++++++++------------- 3 files changed, 73 insertions(+), 13 deletions(-) (limited to 'src') diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index 3d7a0d8..da81e64 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -69,6 +69,11 @@ impl Nip34WritePolicy { } } + /// Get a reference to the purgatory for read-only access + pub fn purgatory(&self) -> &std::sync::Arc { + &self.ctx.purgatory + } + /// 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()); diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs index 18a55d5..2987f15 100644 --- a/src/purgatory/mod.rs +++ b/src/purgatory/mod.rs @@ -19,6 +19,7 @@ pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; use dashmap::DashMap; use nostr_sdk::prelude::*; +use std::collections::HashSet; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -253,6 +254,34 @@ impl Purgatory { self.pr_events.remove(event_id); } + /// Get all event IDs currently stored in purgatory. + /// + /// Returns a HashSet of all event IDs for both state events and PR events + /// held in purgatory. Useful for negentropy sync to avoid fetching events + /// that are already in purgatory awaiting git data. + /// + /// # Returns + /// HashSet of event IDs (as EventId) for all events in purgatory + pub fn event_ids(&self) -> HashSet { + let mut ids = HashSet::new(); + + // Collect state event IDs + for entry in self.state_events.iter() { + for state_entry in entry.value().iter() { + ids.insert(state_entry.event.id); + } + } + + // Collect PR event IDs (only actual events, not placeholders) + for entry in self.pr_events.iter() { + if let Some(ref event) = entry.value().event { + ids.insert(event.id); + } + } + + ids + } + /// Get all PR placeholder event IDs (git-data-first entries without events). /// /// Returns event IDs for entries where git data arrived before the PR event. diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 65affc6..dcdbe3a 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -159,6 +159,8 @@ pub enum ProcessResult { Saved, /// Event already existed in database Duplicate, + /// Event added to Purgatory + Purgatory, /// Event rejected by write policy Rejected, } @@ -377,6 +379,8 @@ pub struct SyncManager { database: SharedDatabase, /// Write policy for validating incoming events write_policy: Nip34WritePolicy, + /// Purgatory for read-only access to events awaiting git data + purgatory: Arc, /// Local relay for submitting synced events (enables broadcast to WebSocket subscribers) local_relay: LocalRelay, /// Configuration reference for sync settings @@ -425,11 +429,15 @@ impl SyncManager { config: &Config, sync_metrics: Option, ) -> Self { + // Extract purgatory from write_policy for read-only access + let purgatory = write_policy.purgatory().clone(); + Self { bootstrap_relay_url, service_domain, database, write_policy, + purgatory, local_relay, config: config.clone(), repo_sync_index: Arc::new(RwLock::new(HashMap::new())), @@ -1621,6 +1629,7 @@ impl SyncManager { match result { WritePolicyResult::Accept => { // Save event to database + if let Err(e) = database.save_event(event).await { tracing::error!( event_id = %event.id, @@ -1644,14 +1653,24 @@ impl SyncManager { ); ProcessResult::Saved } - WritePolicyResult::Reject { message, .. } => { - tracing::debug!( - event_id = %event.id, - relay = %relay_url, - reason = %message, - "Event rejected by write policy" - ); - ProcessResult::Rejected + WritePolicyResult::Reject { message, status } => { + if status { + tracing::debug!( + event_id = %event.id, + kind = %event.kind.as_u16(), + reason = %message, + "Event added to purgatory" + ); + ProcessResult::Purgatory + } else { + tracing::debug!( + event_id = %event.id, + relay = %relay_url, + reason = %message, + "Event rejected by write policy" + ); + ProcessResult::Rejected + } } } } @@ -2172,21 +2191,28 @@ impl SyncManager { let mut all_remote_ids = Vec::new(); let mut failed_count = 0; + // Get event IDs currently in purgatory to avoid fetching them + let purgatory_ids = self.purgatory.event_ids(); + for (idx, result) in diff_results { match result { Ok(reconciliation) => { - let remote_count = reconciliation.remote.len(); - let local_count = reconciliation.local.len(); + let remote_excluding_purgatory_ids: HashSet = reconciliation + .remote + .difference(&purgatory_ids) + .cloned() + .collect(); + let remote_count = remote_excluding_purgatory_ids.len(); tracing::debug!( relay = %relay_url, filter_idx = idx, remote_count = remote_count, - local_count = local_count, - remote_ids = ?reconciliation.remote, + local_count = reconciliation.local.len(), + remote_ids = ?remote_excluding_purgatory_ids, "[DIAG TRACE] ✓ Negentropy diff results for filter {}", idx ); if remote_count > 0 { - all_remote_ids.extend(reconciliation.remote.into_iter()); + all_remote_ids.extend(remote_excluding_purgatory_ids.into_iter()); } } Err(e) => { -- cgit v1.2.3