laantungir/ngit-grasp
1
0
Fork 0
mirror of https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git synced 2026-10-05 15:08:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(purgatory): track event source for filtered expiry logging

Browse Source 
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.
This commit is contained in:
DanConwayDev
2026-01-28 21:00:14 +00:00
parent e9daa340ce
commit f148b3a0e4
4 changed files with 187 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/nostr/builder.rs
+1 -1
View File
@@ -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
src/nostr/policy/state.rs
+1 -1
View File
@@ -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: {}",
src/purgatory/mod.rs
+155 -51
View File
@@ -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);
src/purgatory/types.rs
+30
View File
@@ -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,
}