From 895359aeb6746b98ff82944e4fca503f4a6e5439 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Fri, 9 Jan 2026 16:27:38 +0000 Subject: feat(sync): add cleanup loops and metrics for rejected events index Add automatic cleanup and Prometheus metrics for the two-tier rejected events index that caches rejected announcements for re-processing. Cleanup loops: - Hot cache: Every 60 seconds (events expire after 2 minutes) - Cold index: Every 24 hours (metadata expires after 7 days) - Background task with graceful shutdown support New Prometheus metrics (7): - Gauges: hot_cache_current, cold_index_current - Counters: hits, misses, hot_expired, cold_expired, invalidated This completes the maintainer announcement re-processing feature, reducing wait time from 24 hours to <1 second when a maintainer's announcement arrives before the repository owner's announcement. Memory is bounded through automatic cleanup, and comprehensive metrics enable monitoring of hit rates, memory usage, and cleanup effectiveness. Changes: - src/sync/metrics.rs: Added 7 metrics with recording methods - src/sync/rejected_index.rs: Added optional metrics support - src/sync/mod.rs: Added cleanup background task Tests: 248 library tests passing, 3 integration tests passing --- src/sync/metrics.rs | 124 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 124 insertions(+) (limited to 'src/sync/metrics.rs') diff --git a/src/sync/metrics.rs b/src/sync/metrics.rs index 7907d8e..a175210 100644 --- a/src/sync/metrics.rs +++ b/src/sync/metrics.rs @@ -40,6 +40,22 @@ pub struct SyncMetrics { relays_connected_total: IntGauge, /// Relays marked as dead relays_dead_total: IntGauge, + + // === Rejected Announcements Index Metrics === + /// Current number of entries in hot cache + rejected_announcements_hot_cache_current: IntGauge, + /// Total hot cache hits (events re-processed from cache) + rejected_announcements_hot_cache_hits_total: IntCounter, + /// Total hot cache misses (events not in cache) + rejected_announcements_hot_cache_misses_total: IntCounter, + /// Total expired entries removed from hot cache + rejected_announcements_hot_cache_expired_total: IntCounter, + /// Current number of entries in cold index + rejected_announcements_cold_index_current: IntGauge, + /// Total cold index entries expired and removed + rejected_announcements_cold_index_expired_total: IntCounter, + /// Total invalidations (maintainer announcements invalidated) + rejected_announcements_invalidated_total: IntCounter, } impl SyncMetrics { @@ -113,6 +129,49 @@ impl SyncMetrics { ))?; registry.register(Box::new(relays_dead_total.clone()))?; + // Rejected announcements metrics + let rejected_announcements_hot_cache_current = IntGauge::with_opts(Opts::new( + "ngit_sync_rejected_announcements_hot_cache_current", + "Current number of entries in hot cache (full events, 2 min expiry)", + ))?; + registry.register(Box::new(rejected_announcements_hot_cache_current.clone()))?; + + let rejected_announcements_hot_cache_hits_total = IntCounter::with_opts(Opts::new( + "ngit_sync_rejected_announcements_hot_cache_hits_total", + "Total hot cache hits (events re-processed from cache)", + ))?; + registry.register(Box::new(rejected_announcements_hot_cache_hits_total.clone()))?; + + let rejected_announcements_hot_cache_misses_total = IntCounter::with_opts(Opts::new( + "ngit_sync_rejected_announcements_hot_cache_misses_total", + "Total hot cache misses (events not in cache when invalidated)", + ))?; + registry.register(Box::new(rejected_announcements_hot_cache_misses_total.clone()))?; + + let rejected_announcements_hot_cache_expired_total = IntCounter::with_opts(Opts::new( + "ngit_sync_rejected_announcements_hot_cache_expired_total", + "Total expired entries removed from hot cache", + ))?; + registry.register(Box::new(rejected_announcements_hot_cache_expired_total.clone()))?; + + let rejected_announcements_cold_index_current = IntGauge::with_opts(Opts::new( + "ngit_sync_rejected_announcements_cold_index_current", + "Current number of entries in cold index (metadata only, 7 day expiry)", + ))?; + registry.register(Box::new(rejected_announcements_cold_index_current.clone()))?; + + let rejected_announcements_cold_index_expired_total = IntCounter::with_opts(Opts::new( + "ngit_sync_rejected_announcements_cold_index_expired_total", + "Total expired entries removed from cold index", + ))?; + registry.register(Box::new(rejected_announcements_cold_index_expired_total.clone()))?; + + let rejected_announcements_invalidated_total = IntCounter::with_opts(Opts::new( + "ngit_sync_rejected_announcements_invalidated_total", + "Total invalidations (maintainer announcements invalidated when owner accepted)", + ))?; + registry.register(Box::new(rejected_announcements_invalidated_total.clone()))?; + Ok(Self { relay_connected, connection_attempts_total, @@ -122,6 +181,13 @@ impl SyncMetrics { relays_tracked_total, relays_connected_total, relays_dead_total, + rejected_announcements_hot_cache_current, + rejected_announcements_hot_cache_hits_total, + rejected_announcements_hot_cache_misses_total, + rejected_announcements_hot_cache_expired_total, + rejected_announcements_cold_index_current, + rejected_announcements_cold_index_expired_total, + rejected_announcements_invalidated_total, }) } @@ -293,6 +359,43 @@ impl SyncMetrics { pub fn get_dead_count(&self) -> i64 { self.relays_dead_total.get() } + + // === Rejected Announcements Recording Methods === + + /// Update hot cache current size gauge. + pub fn update_hot_cache_size(&self, size: usize) { + self.rejected_announcements_hot_cache_current.set(size as i64); + } + + /// Record hot cache hit (event re-processed from cache). + pub fn record_hot_cache_hit(&self) { + self.rejected_announcements_hot_cache_hits_total.inc(); + } + + /// Record hot cache miss (event not in cache when invalidated). + pub fn record_hot_cache_miss(&self) { + self.rejected_announcements_hot_cache_misses_total.inc(); + } + + /// Record hot cache expired entries. + pub fn record_hot_cache_expired(&self, count: usize) { + self.rejected_announcements_hot_cache_expired_total.inc_by(count as u64); + } + + /// Update cold index current size gauge. + pub fn update_cold_index_size(&self, size: usize) { + self.rejected_announcements_cold_index_current.set(size as i64); + } + + /// Record cold index expired entries. + pub fn record_cold_index_expired(&self, count: usize) { + self.rejected_announcements_cold_index_expired_total.inc_by(count as u64); + } + + /// Record invalidation (maintainer announcement invalidated). + pub fn record_invalidation(&self, count: usize) { + self.rejected_announcements_invalidated_total.inc_by(count as u64); + } } #[cfg(test)] @@ -395,4 +498,25 @@ mod tests { let metrics2 = SyncMetrics::register(®istry); assert!(metrics2.is_err()); } + + #[test] + fn test_rejected_announcements_metrics() { + let registry = create_test_registry(); + let metrics = SyncMetrics::register(®istry).unwrap(); + + // Test hot cache metrics + metrics.update_hot_cache_size(10); + metrics.record_hot_cache_hit(); + metrics.record_hot_cache_hit(); + metrics.record_hot_cache_miss(); + metrics.record_hot_cache_expired(5); + + // Test cold index metrics + metrics.update_cold_index_size(100); + metrics.record_cold_index_expired(10); + + // Test invalidation metrics + metrics.record_invalidation(3); + metrics.record_invalidation(2); + } } -- cgit v1.2.3