From 73a366cbd7be4edf9c74194cd0891c80a15236a5 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Fri, 23 Jan 2026 11:41:10 +0000 Subject: Add structured logging for migration analysis - Add [PARSE_FAIL] logging when event parsing fails - Add [PURGATORY_EXPIRED] logging when repos expire from purgatory - Logs include: kind, event_id, repo, npub, reason - Supports Phase 4 migration scripts (30-extract-*.sh) - All 382 tests pass --- src/purgatory/mod.rs | 103 ++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 86 insertions(+), 17 deletions(-) (limited to 'src/purgatory') diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs index 47798a6..8b75351 100644 --- a/src/purgatory/mod.rs +++ b/src/purgatory/mod.rs @@ -21,6 +21,7 @@ pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; use dashmap::DashMap; use nostr_sdk::prelude::*; +use nostr_sdk::ToBech32; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::collections::HashSet; @@ -608,6 +609,9 @@ impl Purgatory { /// prevent infinite re-sync loops. Events that expire without finding git data /// will be filtered out during future negentropy/REQ sync operations. /// + /// Emits structured `[PURGATORY_EXPIRED]` log entries for each expired event + /// to support migration scripts and operational monitoring. + /// /// # Returns /// Tuple of (num_state_removed, num_pr_removed) pub fn cleanup(&self) -> (usize, usize) { @@ -615,18 +619,24 @@ impl Purgatory { let mut state_removed = 0; // Remove expired state events and mark them as expired - self.state_events.retain(|_, entries| { + self.state_events.retain(|identifier, entries| { let original_len = entries.len(); - // Collect event IDs before removing - let expired_ids: Vec = entries - .iter() - .filter(|entry| entry.expires_at <= now) - .map(|entry| entry.event.id) - .collect(); - // Mark as expired to prevent re-sync - for event_id in expired_ids { - self.mark_expired(event_id); + // Log and collect expired entries before removing + for entry in entries.iter().filter(|e| e.expires_at <= now) { + let npub = entry.author.to_bech32().unwrap_or_else(|_| entry.author.to_hex()); + let event_id_short = &entry.event.id.to_hex()[..12]; + + // Structured log for migration scripts + tracing::warn!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} reason=\"git data not received within 30 minutes\"", + identifier, + npub, + event_id_short, + entry.event.kind.as_u16() + ); + + self.mark_expired(entry.event.id); } // Remove expired entries @@ -636,21 +646,80 @@ impl Purgatory { }); // Remove expired PR events and mark them as expired - let expired_prs: Vec<(String, Option)> = self + let expired_prs: Vec<_> = self .pr_events .iter() .filter(|entry| entry.value().expires_at <= now) .map(|entry| { - let event_id = entry.value().event.as_ref().map(|e| e.id); - (entry.key().clone(), event_id) + let pr_entry = entry.value(); + let event_id_str = entry.key().clone(); + let event_opt = pr_entry.event.clone(); + let commit = pr_entry.commit.clone(); + (event_id_str, event_opt, commit) }) .collect(); let pr_removed = expired_prs.len(); - for (event_id_str, event_id_opt) in expired_prs { - // Mark actual PR events as expired (not placeholders) - if let Some(event_id) = event_id_opt { - self.mark_expired(event_id); + for (event_id_str, event_opt, commit) in expired_prs { + // Log structured entry for PR events (not placeholders) + if let Some(ref event) = event_opt { + let npub = event.pubkey.to_bech32().unwrap_or_else(|_| event.pubkey.to_hex()); + let event_id_short = &event.id.to_hex()[..12]; + + // Extract ALL repo identifiers from 'a' tags + // (PR events can reference multiple repos when there are multiple maintainers) + let repos: Vec = event + .tags + .iter() + .filter_map(|tag| { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() >= 2 && tag_vec[0] == "a" && tag_vec[1].starts_with("30617:") { + // Format: 30617:: + let parts: Vec<&str> = tag_vec[1].split(':').collect(); + if parts.len() >= 3 { + Some(parts[2].to_string()) + } else { + None + } + } else { + None + } + }) + .collect(); + + // Deduplicate while preserving order + let mut seen = std::collections::HashSet::new(); + let unique_repos: Vec = repos + .into_iter() + .filter(|r| seen.insert(r.clone())) + .collect(); + + let repos_to_log = if unique_repos.is_empty() { + vec!["unknown".to_string()] + } else { + unique_repos + }; + + // Structured log for migration scripts - log once per repo + for repo in &repos_to_log { + tracing::warn!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} commit={} reason=\"git data not received within 30 minutes\"", + repo, + npub, + event_id_short, + event.kind.as_u16(), + &commit[..commit.len().min(12)] + ); + } + + self.mark_expired(event.id); + } else { + // Placeholder (git data arrived first, but PR event never came) + tracing::debug!( + "[PURGATORY_EXPIRED] placeholder event_id={} commit={} reason=\"PR event not received within 30 minutes\"", + &event_id_str[..event_id_str.len().min(12)], + &commit[..commit.len().min(12)] + ); } self.pr_events.remove(&event_id_str); } -- cgit v1.2.3 From 51c331f26ad3c8c422b41267e3695c8f2295510e Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Tue, 27 Jan 2026 21:05:50 +0000 Subject: feat: implement OID retry logic for 'not our ref' errors Add retry loop in fetch_oids that handles git's behavior of stopping at the first missing OID. When a 'not our ref' error occurs: - Parse the missing OID from stderr - Remove it from the fetch list and track it as missing - Retry with remaining OIDs until success or all OIDs exhausted This ensures we fetch all available OIDs even when some are missing from the remote, rather than failing the entire batch. Also improves error reporting: - Include URL in all error messages for easier debugging - Log stderr even when domain is already on naughty list --- src/purgatory/sync/context.rs | 163 +++++++++++++++++++++++++----------------- 1 file changed, 96 insertions(+), 67 deletions(-) (limited to 'src/purgatory') diff --git a/src/purgatory/sync/context.rs b/src/purgatory/sync/context.rs index 33c2d12..0df8be0 100644 --- a/src/purgatory/sync/context.rs +++ b/src/purgatory/sync/context.rs @@ -361,59 +361,60 @@ impl SyncContext for RealSyncContext { let naughty_list = self.git_naughty_list.clone(); tokio::task::spawn_blocking(move || -> Result> { - // git fetch ... - fetch all OIDs with full history - let mut args = vec!["fetch", &url]; - args.extend(missing_oids.iter().map(|s| s.as_str())); - - let output = Command::new("git") - .args(&args) - .current_dir(&repo_path) - .output(); - - match output { - Ok(result) if result.status.success() => { - // Count how many OIDs we now have - let fetched: Vec = missing_oids - .iter() - .filter(|oid| crate::git::oid_exists(&repo_path, oid)) - .cloned() - .collect(); - - debug!(fetched_count = fetched.len(), "Successfully fetched OIDs"); - - Ok(fetched) + let mut remaining_oids = missing_oids.clone(); + let mut missing_from_remote: Vec = Vec::new(); + + // Retry loop: keep fetching until success or no OIDs left + loop { + if remaining_oids.is_empty() { + // All OIDs were missing from remote + debug!( + url = %url, + missing_count = missing_from_remote.len(), + "All requested OIDs missing from remote" + ); + return Ok(vec![]); } - Ok(result) => { - let stderr = String::from_utf8_lossy(&result.stderr); - - // Extract domain and classify error for naughty list - if let Some(domain) = extract_domain(&url) { - if let Some(category) = NaughtyListTracker::classify_error(&stderr) { - let is_new = naughty_list.record(&domain, category, stderr.to_string()); - - if is_new { - tracing::warn!( - domain = %domain, - category = %category, - error = %stderr, - "Git remote domain added to naughty list" - ); - } else { - debug!( - domain = %domain, - category = %category, - "Git remote domain still on naughty list" - ); - } + + // git fetch ... - fetch all OIDs with full history + let mut args = vec!["fetch".to_string(), url.clone()]; + args.extend(remaining_oids.iter().cloned()); + + let output = Command::new("git") + .args(&args) + .current_dir(&repo_path) + .output(); + + match output { + Ok(result) if result.status.success() => { + // Fetch succeeded - count how many OIDs we now have + let fetched: Vec = missing_oids + .iter() + .filter(|oid| crate::git::oid_exists(&repo_path, oid)) + .cloned() + .collect(); + + if !missing_from_remote.is_empty() { + debug!( + url = %url, + fetched_count = fetched.len(), + missing_count = missing_from_remote.len(), + missing_oids = ?missing_from_remote, + "Fetch completed after retries - some OIDs were missing from remote" + ); + } else { + debug!(fetched_count = fetched.len(), "Successfully fetched OIDs"); } + + return Ok(fetched); } + Ok(result) => { + let stderr = String::from_utf8_lossy(&result.stderr); - // Check for "not our ref" errors and provide a clearer error message - let error_msg = if stderr.contains("upload-pack: not our ref") { - // Parse out the missing OID from stderr (git only reports one at a time) - let missing_oid = stderr - .lines() - .find_map(|line| { + // Check for "not our ref" error - this is retryable + if stderr.contains("upload-pack: not our ref") { + // Parse out the missing OID from stderr + let missing_oid = stderr.lines().find_map(|line| { if line.contains("not our ref") { // Extract the OID from lines like: // "fatal: remote error: upload-pack: not our ref " @@ -423,32 +424,60 @@ impl SyncContext for RealSyncContext { } }); - let total_requested = missing_oids.len(); + if let Some(ref oid) = missing_oid { + // Remove the missing OID and retry with remaining + remaining_oids.retain(|o| o != oid); + missing_from_remote.push(oid.clone()); - if let Some(oid) = missing_oid { - if total_requested > 1 { - // BUG: Git stops at first missing OID, so we don't know if the others exist - // We need retry logic to fetch remaining OIDs individually - tracing::warn!( + debug!( url = %url, missing_oid = %oid, - total_requested = total_requested, - "Git fetch failed on first missing OID - other requested OIDs may exist but were not fetched. Retry logic needed." + remaining_count = remaining_oids.len(), + "OID not found on remote, retrying with remaining OIDs" ); - format!("remote missing oid {} (BUG: {} other oids not attempted)", oid, total_requested - 1) - } else { - format!("remote missing only oid requested: {}", oid) + + continue; // Retry with remaining OIDs + } + } + + // Non-retryable error - record to naughty list and return error + if let Some(domain) = extract_domain(&url) { + if let Some(category) = NaughtyListTracker::classify_error(&stderr) { + let is_new = + naughty_list.record(&domain, category, stderr.to_string()); + + if is_new { + tracing::warn!( + domain = %domain, + category = %category, + error = %stderr, + "Git remote domain added to naughty list" + ); + } else { + debug!( + domain = %domain, + category = %category, + error = %stderr, + "Git fetch failed (domain on naughty list)" + ); + } } - } else { - format!("git fetch failed: {}", stderr) } - } else { - format!("git fetch failed: {}", stderr) - }; - Err(anyhow::anyhow!("{}", error_msg)) + return Err(anyhow::anyhow!( + "git fetch failed for {}: {}", + url, + stderr + )); + } + Err(e) => { + return Err(anyhow::anyhow!( + "git fetch command error for {}: {}", + url, + e + )) + } } - Err(e) => Err(anyhow::anyhow!("git fetch command error: {}", e)), } }) .await -- cgit v1.2.3 From 847acdecb9c28a5307123b9ee685b769a598cfc1 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Tue, 27 Jan 2026 21:40:46 +0000 Subject: fix: distinguish 0 OIDs fetched from successful fetch in logging When fetch_oids returns Ok(vec![]) (all requested OIDs missing from remote), the log message now says 'Fetch returned no OIDs (not available on remote)' instead of the misleading 'Fetch succeeded' with oids_fetched=0. --- src/purgatory/sync/functions.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) (limited to 'src/purgatory') diff --git a/src/purgatory/sync/functions.rs b/src/purgatory/sync/functions.rs index 65d29af..2b7e71f 100644 --- a/src/purgatory/sync/functions.rs +++ b/src/purgatory/sync/functions.rs @@ -369,7 +369,7 @@ pub async fn sync_identifier_from_url( throttle_manager.complete_request(&domain); let oids_fetched = match fetch_result { - Ok(fetched) => { + Ok(fetched) if !fetched.is_empty() => { debug!( identifier = %identifier, url = %url, @@ -378,6 +378,14 @@ pub async fn sync_identifier_from_url( ); fetched.len() } + Ok(_) => { + debug!( + identifier = %identifier, + url = %url, + "Fetch returned no OIDs (not available on remote)" + ); + 0 + } Err(e) => { debug!( identifier = %identifier, -- cgit v1.2.3 From 6d920cae2704016869500889a92b358d845b69e1 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Tue, 27 Jan 2026 21:42:25 +0000 Subject: improve logging --- src/purgatory/sync/context.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) (limited to 'src/purgatory') diff --git a/src/purgatory/sync/context.rs b/src/purgatory/sync/context.rs index 0df8be0..904f8af 100644 --- a/src/purgatory/sync/context.rs +++ b/src/purgatory/sync/context.rs @@ -403,7 +403,7 @@ impl SyncContext for RealSyncContext { "Fetch completed after retries - some OIDs were missing from remote" ); } else { - debug!(fetched_count = fetched.len(), "Successfully fetched OIDs"); + debug!(url = %url, fetched_count = fetched.len(), "Successfully fetched OIDs"); } return Ok(fetched); @@ -418,7 +418,9 @@ impl SyncContext for RealSyncContext { if line.contains("not our ref") { // Extract the OID from lines like: // "fatal: remote error: upload-pack: not our ref " - line.split("not our ref").nth(1).map(|s| s.trim().to_string()) + line.split("not our ref") + .nth(1) + .map(|s| s.trim().to_string()) } else { None } @@ -464,11 +466,7 @@ impl SyncContext for RealSyncContext { } } - return Err(anyhow::anyhow!( - "git fetch failed for {}: {}", - url, - stderr - )); + return Err(anyhow::anyhow!("git fetch failed for {}: {}", url, stderr)); } Err(e) => { return Err(anyhow::anyhow!( -- cgit v1.2.3 From efc3da477d4edb9d1334718e3e20d197ba711468 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Tue, 27 Jan 2026 21:55:34 +0000 Subject: fix: pass actually fetched OIDs to process_newly_available_git_data Previously, sync_identifier_from_url passed all needed OIDs to process_newly_available_git_data, not just the OIDs that were successfully fetched. This caused incorrect logging (new_oids_count would show all needed OIDs, not just fetched ones). While this didn't break functionality (the actual processing uses can_apply_state which checks the repository on disk), it made debugging confusing. Changes: - Rename oids_fetched to fetched_oids and change type from usize to Vec - Return Vec from match arms instead of counts - Pass fetched_oids (not needed_oids) to process_newly_available_git_data - Return fetched_oids.len() at the end This ensures logging accurately reflects which OIDs were actually fetched from the remote. --- src/purgatory/sync/functions.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) (limited to 'src/purgatory') diff --git a/src/purgatory/sync/functions.rs b/src/purgatory/sync/functions.rs index 2b7e71f..9207d58 100644 --- a/src/purgatory/sync/functions.rs +++ b/src/purgatory/sync/functions.rs @@ -368,7 +368,7 @@ pub async fn sync_identifier_from_url( let fetch_result = ctx.fetch_oids(&target_repo, url, &needed_oids).await; throttle_manager.complete_request(&domain); - let oids_fetched = match fetch_result { + let fetched_oids = match fetch_result { Ok(fetched) if !fetched.is_empty() => { debug!( identifier = %identifier, @@ -376,7 +376,7 @@ pub async fn sync_identifier_from_url( oids_fetched = fetched.len(), "Fetch succeeded" ); - fetched.len() + fetched } Ok(_) => { debug!( @@ -384,7 +384,7 @@ pub async fn sync_identifier_from_url( url = %url, "Fetch returned no OIDs (not available on remote)" ); - 0 + vec![] } Err(e) => { debug!( @@ -393,13 +393,13 @@ pub async fn sync_identifier_from_url( error = %e, "Fetch failed" ); - 0 + vec![] } }; // Try to process any events that can now be satisfied - if oids_fetched > 0 { - let new_oids: HashSet = needed_oids.into_iter().collect(); + if !fetched_oids.is_empty() { + let new_oids: HashSet = fetched_oids.iter().cloned().collect(); if let Err(e) = ctx .process_newly_available_git_data(&target_repo, &new_oids) .await @@ -412,7 +412,7 @@ pub async fn sync_identifier_from_url( } } - oids_fetched + fetched_oids.len() } /// Sync git data for an identifier. -- cgit v1.2.3 From f148b3a0e4b032c0acf835cda6d2935e19b9f67e Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Wed, 28 Jan 2026 21:00:14 +0000 Subject: feat(purgatory): track event source for filtered expiry logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add EventSource enum (Direct/Sync) to purgatory entries to distinguish between user-submitted events and sync-fetched events. This enables: - WARN-level logging for direct submissions that expire (user should know) - DEBUG-level logging for sync-fetched expirations (expected behavior) - Source upgrade from Sync→Direct if user submits after sync - Expiry timer reset on source upgrade (fresh 30-min window for user) The source is included in [PURGATORY_EXPIRED] logs as source=direct or source=sync for easy filtering. --- src/nostr/builder.rs | 2 +- src/nostr/policy/state.rs | 2 +- src/purgatory/mod.rs | 206 ++++++++++++++++++++++++++++++++++------------ src/purgatory/types.rs | 30 +++++++ 4 files changed, 187 insertions(+), 53 deletions(-) (limited to 'src/purgatory') diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index 629c111..9211972 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -399,7 +399,7 @@ impl Nip34WritePolicy { // Add to purgatory self.ctx .purgatory - .add_pr(event.clone(), event.id.to_hex(), commit.clone()); + .add_pr(event.clone(), event.id.to_hex(), commit.clone(), is_synced); WritePolicyResult::Reject { status: true, // Client sees OK diff --git a/src/nostr/policy/state.rs b/src/nostr/policy/state.rs index f94f004..52f0483 100644 --- a/src/nostr/policy/state.rs +++ b/src/nostr/policy/state.rs @@ -207,7 +207,7 @@ impl StatePolicy { // (add_state automatically enqueues for background sync) self.ctx .purgatory - .add_state(event.clone(), state.identifier.clone(), event.pubkey); + .add_state(event.clone(), state.identifier.clone(), event.pubkey, is_synced); tracing::info!( "state event added to purgatory: eventid: {}, identifier: {}", diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs index 8b75351..d442ad8 100644 --- a/src/purgatory/mod.rs +++ b/src/purgatory/mod.rs @@ -17,7 +17,7 @@ pub mod sync; mod types; pub use helpers::{can_apply_state, can_satisfy_state, extract_refs_from_state, get_unpushed_refs}; -pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; +pub use types::{EventSource, PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; use dashmap::DashMap; use nostr_sdk::prelude::*; @@ -58,6 +58,9 @@ struct SerializableStatePurgatoryEntry { created_at_offset_secs: u64, /// Duration offset from saved_at for expires_at expires_at_offset_secs: u64, + /// Source of this event (direct submission vs sync) + #[serde(default)] + source: types::EventSource, } /// Serializable wrapper for `PrPurgatoryEntry` with time offsets. @@ -75,6 +78,9 @@ struct SerializablePrPurgatoryEntry { created_at_offset_secs: u64, /// Duration offset from saved_at for expires_at expires_at_offset_secs: u64, + /// Source of this event (direct submission vs sync) + #[serde(default)] + source: types::EventSource, } /// Serializable purgatory state for disk persistence. @@ -271,11 +277,38 @@ impl Purgatory { /// For sync-triggered events, the SyncManager calls `enqueue_sync_immediate` separately /// to override this delay. /// + /// If an event already exists in purgatory with `Sync` source and the new submission + /// is direct (`!from_sync`), the source is upgraded to `Direct` without extending expiry. + /// /// # Arguments /// * `event` - The state event (kind 30618) to hold /// * `identifier` - The repository identifier from the 'd' tag /// * `author` - The event author's public key - pub fn add_state(&self, event: Event, identifier: String, author: PublicKey) { + /// * `from_sync` - True if this event came from proactive sync (vs user-submitted) + pub fn add_state(&self, event: Event, identifier: String, author: PublicKey, from_sync: bool) { + let source = if from_sync { + types::EventSource::Sync + } else { + types::EventSource::Direct + }; + + // Check if event already exists - if so, potentially upgrade source + if let Some(mut entries) = self.state_events.get_mut(&identifier) { + if let Some(existing) = entries.iter_mut().find(|e| e.event.id == event.id) { + // Upgrade source from Sync to Direct if new submission is direct + if existing.source == types::EventSource::Sync && !from_sync { + existing.source = types::EventSource::Direct; + existing.expires_at = Instant::now() + DEFAULT_EXPIRY; + tracing::debug!( + event_id = %event.id, + identifier = %identifier, + "Upgraded purgatory entry source from Sync to Direct, reset expiry" + ); + } + return; // Event already exists, don't add duplicate + } + } + let now = Instant::now(); let entry = StatePurgatoryEntry { event, @@ -283,6 +316,7 @@ impl Purgatory { author, created_at: now, expires_at: now + DEFAULT_EXPIRY, + source, }; self.state_events @@ -302,11 +336,35 @@ impl Purgatory { /// Automatically enqueues the referenced repository identifier for background sync /// with the default delay (3 minutes), giving time for a git push to arrive. /// + /// If an event already exists in purgatory with `Sync` source and the new submission + /// is direct (`!from_sync`), the source is upgraded to `Direct` without extending expiry. + /// /// # Arguments /// * `event` - The PR event (kind 1617/1618) to hold /// * `event_id` - The event ID (hex string) from the 'e' tag /// * `commit` - The commit SHA from the 'c' tag - pub fn add_pr(&self, event: Event, event_id: String, commit: String) { + /// * `from_sync` - True if this event came from proactive sync (vs user-submitted) + pub fn add_pr(&self, event: Event, event_id: String, commit: String, from_sync: bool) { + let source = if from_sync { + types::EventSource::Sync + } else { + types::EventSource::Direct + }; + + // Check if event already exists - if so, potentially upgrade source + if let Some(mut existing) = self.pr_events.get_mut(&event_id) { + // Upgrade source from Sync to Direct if new submission is direct + if existing.source == types::EventSource::Sync && !from_sync { + existing.source = types::EventSource::Direct; + existing.expires_at = Instant::now() + DEFAULT_EXPIRY; + tracing::debug!( + event_id = %event_id, + "Upgraded PR purgatory entry source from Sync to Direct, reset expiry" + ); + } + return; // Event already exists, don't add duplicate + } + // Extract identifier from the event's `a` tag for sync enqueueing let identifier = crate::git::sync::extract_identifier_from_pr_event(&event); @@ -316,6 +374,7 @@ impl Purgatory { commit, created_at: now, expires_at: now + DEFAULT_EXPIRY, + source, }; self.pr_events.insert(event_id, entry); @@ -329,6 +388,8 @@ impl Purgatory { /// Add a PR placeholder (git data arrived before PR event). /// /// Creates a placeholder entry waiting for the corresponding PR event. + /// Placeholders are always marked as `Direct` source since they originate + /// from git pushes (direct user action). /// /// # Arguments /// * `event_id` - The expected event ID (from git ref name) @@ -340,6 +401,7 @@ impl Purgatory { commit, created_at: now, expires_at: now + DEFAULT_EXPIRY, + source: types::EventSource::Direct, // Git pushes are direct user actions }; self.pr_events.insert(event_id, entry); @@ -626,15 +688,29 @@ impl Purgatory { for entry in entries.iter().filter(|e| e.expires_at <= now) { let npub = entry.author.to_bech32().unwrap_or_else(|_| entry.author.to_hex()); let event_id_short = &entry.event.id.to_hex()[..12]; + let source_str = if entry.source.is_direct() { "direct" } else { "sync" }; // Structured log for migration scripts - tracing::warn!( - "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} reason=\"git data not received within 30 minutes\"", - identifier, - npub, - event_id_short, - entry.event.kind.as_u16() - ); + // Direct submissions log at WARN, synced events at DEBUG + if entry.source.is_direct() { + tracing::warn!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} source={} reason=\"git data not received within 30 minutes\"", + identifier, + npub, + event_id_short, + entry.event.kind.as_u16(), + source_str + ); + } else { + tracing::debug!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} source={} reason=\"git data not received within 30 minutes\"", + identifier, + npub, + event_id_short, + entry.event.kind.as_u16(), + source_str + ); + } self.mark_expired(entry.event.id); } @@ -655,16 +731,18 @@ impl Purgatory { let event_id_str = entry.key().clone(); let event_opt = pr_entry.event.clone(); let commit = pr_entry.commit.clone(); - (event_id_str, event_opt, commit) + let source = pr_entry.source; + (event_id_str, event_opt, commit, source) }) .collect(); let pr_removed = expired_prs.len(); - for (event_id_str, event_opt, commit) in expired_prs { + for (event_id_str, event_opt, commit, source) in expired_prs { // Log structured entry for PR events (not placeholders) if let Some(ref event) = event_opt { let npub = event.pubkey.to_bech32().unwrap_or_else(|_| event.pubkey.to_hex()); let event_id_short = &event.id.to_hex()[..12]; + let source_str = if source.is_direct() { "direct" } else { "sync" }; // Extract ALL repo identifiers from 'a' tags // (PR events can reference multiple repos when there are multiple maintainers) @@ -701,22 +779,37 @@ impl Purgatory { }; // Structured log for migration scripts - log once per repo + // Direct submissions log at WARN, synced events at DEBUG for repo in &repos_to_log { - tracing::warn!( - "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} commit={} reason=\"git data not received within 30 minutes\"", - repo, - npub, - event_id_short, - event.kind.as_u16(), - &commit[..commit.len().min(12)] - ); + if source.is_direct() { + tracing::warn!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} commit={} source={} reason=\"git data not received within 30 minutes\"", + repo, + npub, + event_id_short, + event.kind.as_u16(), + &commit[..commit.len().min(12)], + source_str + ); + } else { + tracing::debug!( + "[PURGATORY_EXPIRED] repo={} npub={} event_id={}... kind={} commit={} source={} reason=\"git data not received within 30 minutes\"", + repo, + npub, + event_id_short, + event.kind.as_u16(), + &commit[..commit.len().min(12)], + source_str + ); + } } self.mark_expired(event.id); } else { // Placeholder (git data arrived first, but PR event never came) + // Placeholders are always Direct source (from git push) tracing::debug!( - "[PURGATORY_EXPIRED] placeholder event_id={} commit={} reason=\"PR event not received within 30 minutes\"", + "[PURGATORY_EXPIRED] placeholder event_id={} commit={} source=direct reason=\"PR event not received within 30 minutes\"", &event_id_str[..event_id_str.len().min(12)], &commit[..commit.len().min(12)] ); @@ -869,6 +962,7 @@ impl Purgatory { author: e.author, created_at_offset_secs: created_offset.as_secs(), expires_at_offset_secs: expires_offset.as_secs(), + source: e.source, } }) .collect(); @@ -891,6 +985,7 @@ impl Purgatory { commit: e.commit.clone(), created_at_offset_secs: created_offset.as_secs(), expires_at_offset_secs: expires_offset.as_secs(), + source: e.source, }; pr_events.insert(event_id, serializable); } @@ -992,6 +1087,7 @@ impl Purgatory { author: e.author, created_at, expires_at, + source: e.source, } }) .collect(); @@ -1017,6 +1113,7 @@ impl Purgatory { commit: e.commit, created_at, expires_at, + source: e.source, }; self.pr_events.insert(event_id, entry); @@ -1074,8 +1171,8 @@ mod tests { .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event.clone(), "test-repo".to_string(), keys.public_key()); - purgatory.add_pr(event, "test-event-id".to_string(), "abc123".to_string()); + purgatory.add_state(event.clone(), "test-repo".to_string(), keys.public_key(), false); + purgatory.add_pr(event, "test-event-id".to_string(), "abc123".to_string(), false); let (state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 1); @@ -1126,7 +1223,7 @@ mod tests { let event = EventBuilder::text_note("state") .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event, "test-repo".to_string(), keys.public_key()); + purgatory.add_state(event, "test-repo".to_string(), keys.public_key(), false); // Now should have pending events assert!(purgatory.has_pending_events("test-repo")); @@ -1156,7 +1253,7 @@ mod tests { .sign_with_keys(&keys) .unwrap(); - purgatory.add_pr(event, "pr-event-id".to_string(), "commit123".to_string()); + purgatory.add_pr(event, "pr-event-id".to_string(), "commit123".to_string(), false); // Now should have pending events for test-repo assert!(purgatory.has_pending_events("test-repo")); @@ -1221,6 +1318,7 @@ fn test_pr_event_vs_placeholder() { event.clone(), "event-id-1".to_string(), "commit-abc".to_string(), + false, ); // Add a placeholder (no event) @@ -1277,8 +1375,9 @@ fn test_cleanup_removes_expired_entries() { state_event.clone(), "test-repo".to_string(), keys.public_key(), + false, ); - purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string()); + purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string(), false); purgatory.add_pr_placeholder("pr-456".to_string(), "commit-def".to_string()); // Verify entries are there @@ -1325,8 +1424,8 @@ fn test_cleanup_preserves_non_expired_entries() { .unwrap(); // Add fresh entries - purgatory.add_state(state_event, "test-repo".to_string(), keys.public_key()); - purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string()); + purgatory.add_state(state_event, "test-repo".to_string(), keys.public_key(), false); + purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string(), false); // Run cleanup let (state_removed, pr_removed) = purgatory.cleanup(); @@ -1356,8 +1455,8 @@ fn test_cleanup_mixed_expired_and_fresh() { .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event1, "test-repo".to_string(), keys.public_key()); - purgatory.add_state(event2, "test-repo".to_string(), keys.public_key()); + purgatory.add_state(event1, "test-repo".to_string(), keys.public_key(), false); + purgatory.add_state(event2, "test-repo".to_string(), keys.public_key(), false); // Expire only the first one if let Some(mut entries) = purgatory.state_events.get_mut("test-repo") { @@ -1374,8 +1473,8 @@ fn test_cleanup_mixed_expired_and_fresh() { .sign_with_keys(&keys) .unwrap(); - purgatory.add_pr(pr1, "pr-1".to_string(), "commit-1".to_string()); - purgatory.add_pr(pr2, "pr-2".to_string(), "commit-2".to_string()); + purgatory.add_pr(pr1, "pr-1".to_string(), "commit-1".to_string(), false); + purgatory.add_pr(pr2, "pr-2".to_string(), "commit-2".to_string(), false); // Expire only first PR if let Some(mut entry) = purgatory.pr_events.get_mut("pr-1") { @@ -1407,8 +1506,8 @@ fn test_remove_expired_legacy_method() { .unwrap(); let pr_event = EventBuilder::text_note("pr").sign_with_keys(&keys).unwrap(); - purgatory.add_state(state_event, "repo".to_string(), keys.public_key()); - purgatory.add_pr(pr_event, "pr-id".to_string(), "commit".to_string()); + purgatory.add_state(state_event, "repo".to_string(), keys.public_key(), false); + purgatory.add_pr(pr_event, "pr-id".to_string(), "commit".to_string(), false); // Expire both if let Some(mut entries) = purgatory.state_events.get_mut("repo") { @@ -1442,8 +1541,8 @@ fn test_expired_event_tracking() { let pr_event_id = pr_event.id; // Add events to purgatory - purgatory.add_state(state_event, "repo".to_string(), keys.public_key()); - purgatory.add_pr(pr_event, "pr-id".to_string(), "commit".to_string()); + purgatory.add_state(state_event, "repo".to_string(), keys.public_key(), false); + purgatory.add_pr(pr_event, "pr-id".to_string(), "commit".to_string(), false); // Events should not be marked as expired yet assert!(!purgatory.is_expired(&state_event_id)); @@ -1495,7 +1594,7 @@ fn test_cleanup_expired_events() { let event2_id = event2.id; // Add and immediately expire event1 - purgatory.add_state(event1, "repo1".to_string(), keys.public_key()); + purgatory.add_state(event1, "repo1".to_string(), keys.public_key(), false); if let Some(mut entries) = purgatory.state_events.get_mut("repo1") { for entry in entries.iter_mut() { entry.expires_at = Instant::now() - Duration::from_secs(1); @@ -1504,7 +1603,7 @@ fn test_cleanup_expired_events() { purgatory.cleanup(); // Add and expire event2 (will be more recent) - purgatory.add_state(event2, "repo2".to_string(), keys.public_key()); + purgatory.add_state(event2, "repo2".to_string(), keys.public_key(), false); if let Some(mut entries) = purgatory.state_events.get_mut("repo2") { for entry in entries.iter_mut() { entry.expires_at = Instant::now() - Duration::from_secs(1); @@ -1546,7 +1645,7 @@ fn test_expired_events_prevent_readdition() { let event_id = event.id; // Add event to purgatory - purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key()); + purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key(), false); // Expire it if let Some(mut entries) = purgatory.state_events.get_mut("repo") { @@ -1566,7 +1665,7 @@ fn test_expired_events_prevent_readdition() { // This simulates what negentropy/REQ+EOSE should do: // Check if event is in event_ids() before adding if !ids.contains(&event_id) { - purgatory.add_state(event, "repo".to_string(), keys.public_key()); + purgatory.add_state(event, "repo".to_string(), keys.public_key(), false); } // Event should NOT be re-added @@ -1609,7 +1708,7 @@ fn test_user_can_resubmit_expired_event() { let event_id = event.id; // Add event to purgatory - purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key()); + purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key(), false); // Expire it if let Some(mut entries) = purgatory.state_events.get_mut("repo") { @@ -1658,8 +1757,8 @@ async fn test_save_and_restore_state_events() { let event1_id = event1.id; let event2_id = event2.id; - purgatory.add_state(event1.clone(), "test-repo".to_string(), keys.public_key()); - purgatory.add_state(event2.clone(), "test-repo".to_string(), keys.public_key()); + purgatory.add_state(event1.clone(), "test-repo".to_string(), keys.public_key(), false); + purgatory.add_state(event2.clone(), "test-repo".to_string(), keys.public_key(), false); // Save to disk purgatory.save_to_disk(&state_file).unwrap(); @@ -1721,6 +1820,7 @@ async fn test_save_and_restore_pr_events() { pr_event.clone(), "pr-event-id".to_string(), "commit-abc".to_string(), + false, ); // Save to disk @@ -1790,7 +1890,7 @@ async fn test_save_and_restore_expired_events() { let event_id = event.id; // Add and expire event - purgatory.add_state(event, "repo".to_string(), keys.public_key()); + purgatory.add_state(event, "repo".to_string(), keys.public_key(), false); if let Some(mut entries) = purgatory.state_events.get_mut("repo") { for entry in entries.iter_mut() { entry.expires_at = Instant::now() - Duration::from_secs(1); @@ -1929,7 +2029,7 @@ async fn test_downtime_calculation() { .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key()); + purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key(), false); // Get original expiry time let original_entries = purgatory.find_state("repo"); @@ -1985,7 +2085,7 @@ async fn test_expiry_times_preserved() { .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key()); + purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key(), false); // Manually set expiry to a specific time in the future let custom_expiry = Instant::now() + Duration::from_secs(600); // 10 minutes @@ -2044,16 +2144,19 @@ async fn test_multiple_state_events_same_identifier() { event1.clone(), "shared-repo".to_string(), keys1.public_key(), + false, ); purgatory.add_state( event2.clone(), "shared-repo".to_string(), keys2.public_key(), + false, ); purgatory.add_state( event3.clone(), "shared-repo".to_string(), keys3.public_key(), + false, ); // Save to disk @@ -2100,6 +2203,7 @@ async fn test_mixed_pr_events_and_placeholders() { pr_event.clone(), "pr-with-event".to_string(), "commit-abc".to_string(), + false, ); // Add PR placeholder @@ -2145,7 +2249,7 @@ async fn test_file_cleanup_after_successful_restore() { let event = EventBuilder::text_note("test") .sign_with_keys(&keys) .unwrap(); - purgatory.add_state(event, "repo".to_string(), keys.public_key()); + purgatory.add_state(event, "repo".to_string(), keys.public_key(), false); // Save to disk purgatory.save_to_disk(&state_file).unwrap(); @@ -2179,8 +2283,8 @@ async fn test_comprehensive_roundtrip() { .sign_with_keys(&keys2) .unwrap(); - purgatory.add_state(state1.clone(), "repo1".to_string(), keys1.public_key()); - purgatory.add_state(state2.clone(), "repo2".to_string(), keys2.public_key()); + purgatory.add_state(state1.clone(), "repo1".to_string(), keys1.public_key(), false); + purgatory.add_state(state2.clone(), "repo2".to_string(), keys2.public_key(), false); // Add PR event let tags = vec![Tag::custom( @@ -2191,7 +2295,7 @@ async fn test_comprehensive_roundtrip() { .tags(tags) .sign_with_keys(&keys1) .unwrap(); - purgatory.add_pr(pr_event.clone(), "pr-1".to_string(), "commit-1".to_string()); + purgatory.add_pr(pr_event.clone(), "pr-1".to_string(), "commit-1".to_string(), false); // Add PR placeholder purgatory.add_pr_placeholder("pr-2".to_string(), "commit-2".to_string()); @@ -2201,7 +2305,7 @@ async fn test_comprehensive_roundtrip() { .sign_with_keys(&keys1) .unwrap(); let expired_id = expired_event.id; - purgatory.add_state(expired_event, "repo3".to_string(), keys1.public_key()); + purgatory.add_state(expired_event, "repo3".to_string(), keys1.public_key(), false); if let Some(mut entries) = purgatory.state_events.get_mut("repo3") { for entry in entries.iter_mut() { entry.expires_at = Instant::now() - Duration::from_secs(1); diff --git a/src/purgatory/types.rs b/src/purgatory/types.rs index 919504b..e37a3e1 100644 --- a/src/purgatory/types.rs +++ b/src/purgatory/types.rs @@ -8,6 +8,28 @@ use nostr_sdk::prelude::*; use serde::{Deserialize, Serialize}; use std::time::Instant; +/// Source of an event entering purgatory. +/// +/// Tracks whether an event was submitted directly by a user or fetched via +/// proactive sync from another relay. This distinction is used for: +/// - Filtered logging: Direct submissions log at WARN level, synced at DEBUG +/// - Operational monitoring: Helps identify user-facing issues vs sync noise +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +pub enum EventSource { + /// Event was published directly to this relay by a user + #[default] + Direct, + /// Event was fetched via proactive sync from another relay + Sync, +} + +impl EventSource { + /// Returns true if this is a direct submission (not synced) + pub fn is_direct(&self) -> bool { + matches!(self, EventSource::Direct) + } +} + /// Default value for Instant fields during deserialization fn instant_now() -> Instant { Instant::now() @@ -86,6 +108,10 @@ pub struct StatePurgatoryEntry { /// Expiry deadline (30 min from creation, may be extended) #[serde(skip, default = "instant_now")] pub expires_at: Instant, + + /// Source of this event (direct submission vs sync) + #[serde(default)] + pub source: EventSource, } /// Entry for a PR event (kind 1617/1618) or placeholder waiting in purgatory. @@ -112,4 +138,8 @@ pub struct PrPurgatoryEntry { /// Expiry deadline (30 min from creation, may be extended) #[serde(skip, default = "instant_now")] pub expires_at: Instant, + + /// Source of this event (direct submission vs sync) + #[serde(default)] + pub source: EventSource, } -- cgit v1.2.3