Merge master into 3ca0-announcements-purgatory

5 files changed, 634 insertions, 126 deletions

diff --git a/src/purgatory/helpers.rs b/src/purgatory/helpers.rs
index 193ef99..a9f6e66 100644
--- a/src/purgatory/helpers.rs
+++ b/src/purgatory/helpers.rs
@@ -225,6 +225,117 @@ pub fn get_unpushed_refs(event: &Event, pushed_refs: &[RefPair]) -> Vec<RefPair>
         .collect()
 }
 
+/// Diagnose why a state event doesn't match the push.
+///
+/// Returns a human-readable explanation of the mismatch between the state event
+/// and what would result from applying the push to local refs.
+///
+/// # Arguments
+/// * `event` - The state event to check
+/// * `pushed_updates` - Ref updates in the current push operation
+/// * `local_refs` - Refs already existing locally (ref_name -> SHA)
+///
+/// # Returns
+/// String explaining why the state doesn't match, or None if it matches
+pub fn diagnose_state_mismatch(
+    event: &Event,
+    pushed_updates: &[RefUpdate],
+    local_refs: &HashMap<String, String>,
+) -> Option<String> {
+    let state_refs = extract_refs_from_state(event);
+
+    // Filter local_refs to only branches and tags
+    let mut would_be_state: HashMap<String, String> = local_refs
+        .iter()
+        .filter(|(ref_name, _)| {
+            ref_name.starts_with("refs/heads/") || ref_name.starts_with("refs/tags/")
+        })
+        .map(|(k, v)| (k.clone(), v.clone()))
+        .collect();
+
+    // Apply all pushed updates to create the would-be state
+    for update in pushed_updates {
+        // Only process branches and tags
+        if !update.ref_name.starts_with("refs/heads/") && !update.ref_name.starts_with("refs/tags/")
+        {
+            continue;
+        }
+
+        if update.is_deletion() {
+            would_be_state.remove(&update.ref_name);
+        } else {
+            would_be_state.insert(update.ref_name.clone(), update.new_oid.clone());
+        }
+    }
+
+    // Convert event's state refs to a HashMap for comparison
+    let declared_state: HashMap<String, String> = state_refs
+        .into_iter()
+        .map(|r| (r.ref_name, r.object_sha))
+        .collect();
+
+    // Check if they match
+    if would_be_state == declared_state {
+        return None; // No mismatch
+    }
+
+    // Build diagnostic message
+    let mut reasons = Vec::new();
+
+    // Check for refs in declared state but not in would-be state
+    for (ref_name, declared_sha) in &declared_state {
+        if let Some(would_be_sha) = would_be_state.get(ref_name) {
+            if would_be_sha != declared_sha {
+                let would_be_short = if would_be_sha.len() >= 8 {
+                    &would_be_sha[..8]
+                } else {
+                    would_be_sha.as_str()
+                };
+                let declared_short = if declared_sha.len() >= 8 {
+                    &declared_sha[..8]
+                } else {
+                    declared_sha.as_str()
+                };
+                reasons.push(format!(
+                    "{} would be at {} but state declares {}",
+                    ref_name, would_be_short, declared_short
+                ));
+            }
+        } else {
+            let declared_short = if declared_sha.len() >= 8 {
+                &declared_sha[..8]
+            } else {
+                declared_sha.as_str()
+            };
+            reasons.push(format!(
+                "{} missing (state declares {})",
+                ref_name, declared_short
+            ));
+        }
+    }
+
+    // Check for refs in would-be state but not in declared state
+    for (ref_name, would_be_sha) in &would_be_state {
+        if !declared_state.contains_key(ref_name) {
+            let would_be_short = if would_be_sha.len() >= 8 {
+                &would_be_sha[..8]
+            } else {
+                would_be_sha.as_str()
+            };
+            reasons.push(format!(
+                "{} would exist at {} but state doesn't declare it",
+                ref_name, would_be_short
+            ));
+        }
+    }
+
+    if reasons.is_empty() {
+        Some("Unknown mismatch".to_string())
+    } else {
+        Some(reasons.join("; "))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -695,4 +806,97 @@ mod tests {
         // Should return true - real OID exists, symbolic ref skipped
         assert!(can_apply_state(&event, repo_path));
     }
+
+    #[test]
+    fn test_diagnose_state_mismatch_missing_ref() {
+        // State declares both main and test branches
+        let event = create_test_state_event(
+            "test-repo",
+            vec![("refs/heads/main", "abc123"), ("refs/heads/test", "def456")],
+        );
+
+        // Push only creates test branch
+        let pushed_updates = vec![RefUpdate {
+            old_oid: "0000000000000000000000000000000000000000".to_string(),
+            new_oid: "def456".to_string(),
+            ref_name: "refs/heads/test".to_string(),
+        }];
+
+        // No local refs
+        let local_refs = HashMap::new();
+
+        let diagnosis = diagnose_state_mismatch(&event, &pushed_updates, &local_refs);
+        assert!(diagnosis.is_some());
+        let msg = diagnosis.unwrap();
+        assert!(msg.contains("refs/heads/main"));
+        assert!(msg.contains("missing"));
+    }
+
+    #[test]
+    fn test_diagnose_state_mismatch_wrong_sha() {
+        // State declares main at abc123
+        let event = create_test_state_event("test-repo", vec![("refs/heads/main", "abc123")]);
+
+        // Push updates main to different SHA
+        let pushed_updates = vec![RefUpdate {
+            old_oid: "0000000000000000000000000000000000000000".to_string(),
+            new_oid: "wrong123".to_string(),
+            ref_name: "refs/heads/main".to_string(),
+        }];
+
+        let local_refs = HashMap::new();
+
+        let diagnosis = diagnose_state_mismatch(&event, &pushed_updates, &local_refs);
+        assert!(diagnosis.is_some());
+        let msg = diagnosis.unwrap();
+        assert!(msg.contains("refs/heads/main"));
+        assert!(msg.contains("would be at"));
+        assert!(msg.contains("state declares"));
+    }
+
+    #[test]
+    fn test_diagnose_state_mismatch_extra_ref() {
+        // State declares only main
+        let event = create_test_state_event("test-repo", vec![("refs/heads/main", "abc123")]);
+
+        // Push creates both main and test
+        let pushed_updates = vec![
+            RefUpdate {
+                old_oid: "0000000000000000000000000000000000000000".to_string(),
+                new_oid: "abc123".to_string(),
+                ref_name: "refs/heads/main".to_string(),
+            },
+            RefUpdate {
+                old_oid: "0000000000000000000000000000000000000000".to_string(),
+                new_oid: "def456".to_string(),
+                ref_name: "refs/heads/test".to_string(),
+            },
+        ];
+
+        let local_refs = HashMap::new();
+
+        let diagnosis = diagnose_state_mismatch(&event, &pushed_updates, &local_refs);
+        assert!(diagnosis.is_some());
+        let msg = diagnosis.unwrap();
+        assert!(msg.contains("refs/heads/test"));
+        assert!(msg.contains("doesn't declare"));
+    }
+
+    #[test]
+    fn test_diagnose_state_mismatch_no_mismatch() {
+        // State declares main
+        let event = create_test_state_event("test-repo", vec![("refs/heads/main", "abc123")]);
+
+        // Push creates main at correct SHA
+        let pushed_updates = vec![RefUpdate {
+            old_oid: "0000000000000000000000000000000000000000".to_string(),
+            new_oid: "abc123".to_string(),
+            ref_name: "refs/heads/main".to_string(),
+        }];
+
+        let local_refs = HashMap::new();
+
+        let diagnosis = diagnose_state_mismatch(&event, &pushed_updates, &local_refs);
+        assert!(diagnosis.is_none()); // No mismatch
+    }
 }
diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs
index 9a63bf6..bb6ff54 100644
--- a/src/purgatory/mod.rs
+++ b/src/purgatory/mod.rs
@@ -16,11 +16,12 @@ pub mod persistence;
 pub mod sync;
 mod types;
 
-pub use helpers::{can_apply_state, can_satisfy_state, extract_refs_from_state, get_unpushed_refs};
-pub use types::{AnnouncementPurgatoryEntry, PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
+pub use helpers::{can_apply_state, can_satisfy_state, diagnose_state_mismatch, extract_refs_from_state, get_unpushed_refs};
+pub use types::{AnnouncementPurgatoryEntry, EventSource, 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;
@@ -64,6 +65,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.
@@ -81,6 +85,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 wrapper for `AnnouncementPurgatoryEntry` with time offsets.
@@ -313,11 +320,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,
@@ -325,6 +359,7 @@ impl Purgatory {
             author,
             created_at: now,
             expires_at: now + DEFAULT_EXPIRY,
+            source,
         };
 
         self.state_events
@@ -344,11 +379,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);
 
@@ -358,6 +417,7 @@ impl Purgatory {
             commit,
             created_at: now,
             expires_at: now + DEFAULT_EXPIRY,
+            source,
         };
 
         self.pr_events.insert(event_id, entry);
@@ -371,6 +431,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)
@@ -382,6 +444,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);
@@ -892,6 +955,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_announcement_removed, num_state_removed, num_pr_removed)
     pub fn cleanup(&self) -> (usize, usize, usize) {
@@ -976,18 +1042,38 @@ 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<EventId> = 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];
+                let source_str = if entry.source.is_direct() { "direct" } else { "sync" };
+
+                // Structured log for migration scripts
+                // 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);
             }
 
             // Remove expired entries
@@ -997,21 +1083,103 @@ impl Purgatory {
         });
 
         // Remove expired PR events and mark them as expired
-        let expired_prs: Vec<(String, Option<EventId>)> = 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();
+                let source = pr_entry.source;
+                (event_id_str, event_opt, commit, source)
             })
             .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, 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)
+                let repos: Vec<String> = 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:<owner_pubkey>:<identifier>
+                            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<String> = 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
+                // Direct submissions log at WARN, synced events at DEBUG
+                for repo in &repos_to_log {
+                    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={} source=direct 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);
         }
@@ -1191,6 +1359,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();
@@ -1213,6 +1382,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);
         }
@@ -1355,6 +1525,7 @@ impl Purgatory {
                         author: e.author,
                         created_at,
                         expires_at,
+                        source: e.source,
                     }
                 })
                 .collect();
@@ -1380,6 +1551,7 @@ impl Purgatory {
                 commit: e.commit,
                 created_at,
                 expires_at,
+                source: e.source,
             };
 
             self.pr_events.insert(event_id, entry);
@@ -1439,8 +1611,18 @@ 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 (announcement_count, state_count, pr_count) = purgatory.count();
         assert_eq!(announcement_count, 0);
@@ -1492,7 +1674,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"));
@@ -1522,7 +1704,12 @@ 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"));
@@ -1587,6 +1774,7 @@ fn test_pr_event_vs_placeholder() {
         event.clone(),
         "event-id-1".to_string(),
         "commit-abc".to_string(),
+        false,
     );
 
     // Add a placeholder (no event)
@@ -1643,8 +1831,14 @@ 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(),
+        false,
     );
-    purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string());
     purgatory.add_pr_placeholder("pr-456".to_string(), "commit-def".to_string());
 
     // Verify entries are there
@@ -1691,8 +1885,18 @@ 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();
@@ -1722,8 +1926,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") {
@@ -1740,8 +1944,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") {
@@ -1773,8 +1977,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") {
@@ -1808,8 +2012,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));
@@ -1861,7 +2065,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);
@@ -1870,7 +2074,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);
@@ -1912,7 +2116,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") {
@@ -1932,7 +2136,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
@@ -1975,7 +2179,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") {
@@ -2024,8 +2228,18 @@ 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();
@@ -2087,6 +2301,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
@@ -2156,7 +2371,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);
@@ -2295,7 +2510,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");
@@ -2351,7 +2566,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
@@ -2410,16 +2625,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
@@ -2466,6 +2684,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
@@ -2511,7 +2730,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();
@@ -2697,8 +2916,18 @@ 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(
@@ -2709,7 +2938,12 @@ 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());
@@ -2719,7 +2953,12 @@ 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/sync/context.rs b/src/purgatory/sync/context.rs
index ece8cd6..8297515 100644
--- a/src/purgatory/sync/context.rs
+++ b/src/purgatory/sync/context.rs
@@ -375,94 +375,121 @@ impl SyncContext for RealSyncContext {
         let naughty_list = self.git_naughty_list.clone();
 
         tokio::task::spawn_blocking(move || -> Result<Vec<String>> {
-            // git fetch <remote> <sha1> <sha2> ... - 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<String> = 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<String> = 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 <remote> <sha1> <sha2> ... - 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<String> = 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!(url = %url, 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 <oid>"
-                                    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
                                 }
                             });
 
-                        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
diff --git a/src/purgatory/sync/functions.rs b/src/purgatory/sync/functions.rs
index 65d29af..9207d58 100644
--- a/src/purgatory/sync/functions.rs
+++ b/src/purgatory/sync/functions.rs
@@ -368,15 +368,23 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
     let fetch_result = ctx.fetch_oids(&target_repo, url, &needed_oids).await;
     throttle_manager.complete_request(&domain);
 
-    let oids_fetched = match fetch_result {
-        Ok(fetched) => {
+    let fetched_oids = match fetch_result {
+        Ok(fetched) if !fetched.is_empty() => {
             debug!(
                 identifier = %identifier,
                 url = %url,
                 oids_fetched = fetched.len(),
                 "Fetch succeeded"
             );
-            fetched.len()
+            fetched
+        }
+        Ok(_) => {
+            debug!(
+                identifier = %identifier,
+                url = %url,
+                "Fetch returned no OIDs (not available on remote)"
+            );
+            vec![]
         }
         Err(e) => {
             debug!(
@@ -385,13 +393,13 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
                 error = %e,
                 "Fetch failed"
             );
-            0
+            vec![]
         }
     };
 
     // Try to process any events that can now be satisfied
-    if oids_fetched > 0 {
-        let new_oids: HashSet<String> = needed_oids.into_iter().collect();
+    if !fetched_oids.is_empty() {
+        let new_oids: HashSet<String> = fetched_oids.iter().cloned().collect();
         if let Err(e) = ctx
             .process_newly_available_git_data(&target_repo, &new_oids)
             .await
@@ -404,7 +412,7 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
         }
     }
 
-    oids_fetched
+    fetched_oids.len()
 }
 
 /// Sync git data for an identifier.
diff --git a/src/purgatory/types.rs b/src/purgatory/types.rs
index d891bc9..1af5c4e 100644
--- a/src/purgatory/types.rs
+++ b/src/purgatory/types.rs
@@ -10,6 +10,28 @@ use std::collections::HashSet;
 use std::path::PathBuf;
 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()
@@ -88,6 +110,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.
@@ -114,6 +140,10 @@ 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,
 }
 
 /// Entry for a repository announcement (kind 30617) waiting in purgatory.