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(sync): add rejected events cache persistence and integrate with shutdown/startup

Browse Source 
Implement save/restore functionality for rejected events cache and
integrate persistence with relay shutdown/startup lifecycle. Both
purgatory and rejected cache now survive relay restarts.

Key features:
- Serialize rejected events cache to JSON (rejected-events-cache.json)
- Save both hot cache (2min, full events) and cold index (7day, metadata)
- Restore with downtime adjustment (preserves remaining TTL)
- Graceful degradation (missing/corrupted files don't crash)
- File cleanup after successful restore
- Automatic restoration in SyncManager::new()

Integration:
- Shutdown hook saves both purgatory and rejected cache
- Startup hook restores both and re-queues repositories
- Non-fatal errors (logs warnings, continues on failure)

Files:
- src/sync/rejected_index.rs: save_to_disk/restore_from_disk methods
- src/sync/mod.rs: SyncManager integration and auto-restore
- src/main.rs: Shutdown/startup hooks for both caches
- tests/purgatory_persistence.rs: 17 integration tests

Tests: 13 unit tests + 17 integration tests covering full lifecycle
This commit is contained in:
DanConwayDev
2026-01-14 10:19:18 +00:00
parent b6c70f765d
commit b101afa00b
4 changed files with 1586 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+49 -1
View File
@@ -3,7 +3,7 @@ use std::{path::PathBuf, sync::Arc};
use anyhow::Result;
use tokio::signal;
use tracing::{info, Level};
use tracing::{error, info, warn, Level};
use tracing_subscriber::FmtSubscriber;
use ngit_grasp::{
@@ -64,6 +64,22 @@ async fn main() -> Result<()> {
)));
info!("Purgatory initialized for event coordination");
// Restore purgatory state from disk if available
let purgatory_path =
PathBuf::from(config.effective_git_data_path()).join("purgatory-state.json");
if purgatory_path.exists() {
match purgatory.restore_from_disk(&purgatory_path) {
Ok(()) => {
info!("Restored purgatory state from disk");
// Re-queueing will happen later after sync system is created
}
Err(e) => {
warn!("Failed to restore purgatory state: {}, starting empty", e);
}
}
}
// Create Nostr relay with NIP-34 validation
// Returns both the relay and database for direct queries in handlers
if let Ok(relay_with_db) = nostr::builder::create_relay(&config, purgatory.clone()).await {
@@ -88,6 +104,7 @@ async fn main() -> Result<()> {
relay_with_db.write_policy.clone(),
relay_with_db.relay.clone(),
&config,
PathBuf::from(config.effective_git_data_path()),
metrics.as_ref().and_then(|m| m.sync_metrics().cloned()),
);
@@ -100,6 +117,21 @@ async fn main() -> Result<()> {
info!("Proactive sync enabled (will discover relays from stored announcements)");
}
// Re-queue all restored purgatory repos for sync
let restored_identifiers = purgatory.get_all_identifiers();
if !restored_identifiers.is_empty() {
info!(
"Re-queueing {} restored repositories for sync",
restored_identifiers.len()
);
for identifier in restored_identifiers {
purgatory.enqueue_sync_immediate(&identifier);
}
}
// Get a reference to the rejected events index for shutdown persistence
let shutdown_rejected_index = sync_manager.rejected_events_index();
tokio::spawn(async move {
sync_manager.run().await;
});
@@ -190,6 +222,22 @@ async fn main() -> Result<()> {
}
}
// Save purgatory state to disk
let purgatory_save_path = PathBuf::from(&git_data_path).join("purgatory-state.json");
if let Err(e) = shutdown_purgatory.save_to_disk(&purgatory_save_path) {
error!("Failed to save purgatory state: {}", e);
} else {
info!("Purgatory state saved to disk");
}
// Save rejected events cache to disk
let rejected_cache_path = PathBuf::from(&git_data_path).join("rejected-events-cache.json");
if let Err(e) = shutdown_rejected_index.save_to_disk(&rejected_cache_path) {
error!("Failed to save rejected events cache: {}", e);
} else {
info!("Rejected events cache saved to disk");
}
// Cleanup placeholder refs on shutdown
let placeholder_ids = shutdown_purgatory.get_placeholder_event_ids();
if !placeholder_ids.is_empty() {
src/sync/mod.rs
+59 -12
View File
@@ -43,6 +43,7 @@ pub use health::RelayHealthTracker;
use tokio::time::sleep;
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
@@ -581,6 +582,7 @@ impl SyncManager {
/// * `write_policy` - Policy for validating events before storage
/// * `local_relay` - Local relay for submitting synced events (enables WebSocket broadcast)
/// * `config` - Configuration for sync settings
/// * `data_path` - Path to git data directory (for persistence)
/// * `sync_metrics` - Optional pre-registered SyncMetrics (passed from Metrics if metrics are enabled)
pub fn new(
bootstrap_relay_url: Option<String>,
@@ -589,11 +591,42 @@ impl SyncManager {
write_policy: Nip34WritePolicy,
local_relay: LocalRelay,
config: &Config,
data_path: PathBuf,
sync_metrics: Option<SyncMetrics>,
) -> Self {
// Extract purgatory from write_policy for read-only access
let purgatory = write_policy.purgatory().clone();
// Create rejected events index
let rejected_events_index = Arc::new(if let Some(ref metrics) = sync_metrics {
RejectedEventsIndex::with_metrics(
Duration::from_secs(config.rejected_hot_cache_duration_secs),
Duration::from_secs(config.rejected_cold_index_expiry_secs),
metrics.clone(),
)
} else {
RejectedEventsIndex::new(
Duration::from_secs(config.rejected_hot_cache_duration_secs),
Duration::from_secs(config.rejected_cold_index_expiry_secs),
)
});
// Attempt to restore rejected events index from disk
let rejected_index_path = data_path.join("rejected-events-cache.json");
if rejected_index_path.exists() {
match rejected_events_index.restore_from_disk(&rejected_index_path) {
Ok(()) => {
tracing::info!("Restored rejected events index from disk");
}
Err(e) => {
tracing::warn!(
"Failed to restore rejected events index: {}, starting empty",
e
);
}
}
}
Self {
bootstrap_relay_url,
service_domain,
@@ -605,18 +638,7 @@ impl SyncManager {
repo_sync_index: Arc::new(RwLock::new(HashMap::new())),
relay_sync_index: Arc::new(RwLock::new(HashMap::new())),
pending_sync_index: Arc::new(RwLock::new(HashMap::new())),
rejected_events_index: Arc::new(if let Some(ref metrics) = sync_metrics {
RejectedEventsIndex::with_metrics(
Duration::from_secs(config.rejected_hot_cache_duration_secs),
Duration::from_secs(config.rejected_cold_index_expiry_secs),
metrics.clone(),
)
} else {
RejectedEventsIndex::new(
Duration::from_secs(config.rejected_hot_cache_duration_secs),
Duration::from_secs(config.rejected_cold_index_expiry_secs),
)
}),
rejected_events_index,
connections: HashMap::new(),
health_tracker: Arc::new(RelayHealthTracker::new(config)),
next_batch_id: 0,
@@ -637,6 +659,31 @@ impl SyncManager {
self.next_batch_id
}
/// Get a clone of the rejected events index Arc.
///
/// This allows access to the rejected events index for persistence
/// even after the SyncManager has been moved into a task.
///
/// # Returns
/// Arc clone of the rejected events index
pub fn rejected_events_index(&self) -> Arc<RejectedEventsIndex> {
self.rejected_events_index.clone()
}
/// Save rejected events index to disk.
///
/// This is called during shutdown to persist the rejected events cache,
/// allowing us to avoid re-downloading rejected events after restart.
///
/// # Arguments
/// * `path` - Path to save the rejected index file
///
/// # Returns
/// Ok(()) on success, Err if save fails
pub fn save_rejected_index(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
self.rejected_events_index.save_to_disk(path)
}
/// Handle EOSE (End Of Stored Events) for a subscription
///
/// This method:
src/sync/rejected_index.rs
+723 -3
View File
@@ -86,12 +86,14 @@
//! ```
use nostr_sdk::{Event, EventId, PublicKey};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use std::time::{Duration, Instant, SystemTime};
/// Type of event stored in the rejected events index
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum EventType {
/// Repository announcement (kind 30617)
Announcement,
@@ -109,7 +111,7 @@ impl std::fmt::Display for EventType {
}
/// Reason why a repository announcement was rejected
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RejectionReason {
/// Announcement doesn't list this service in clone/web URLs
DoesNotListService,
@@ -141,6 +143,20 @@ struct HotCacheEntry {
cached_at: Instant,
}
/// Serializable version of HotCacheEntry for persistence
///
/// Converts Instant to Duration offset from saved_at time
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SerializableHotCacheEntry {
event: Event,
pubkey: PublicKey,
identifier: String,
event_type: EventType,
reason: RejectionReason,
/// Duration since saved_at when this entry was cached
cached_at_offset_secs: u64,
}
/// Entry in the cold index (metadata only)
///
/// Note: event_id is stored as the HashMap key, not in this struct
@@ -154,6 +170,49 @@ struct ColdIndexEntry {
rejected_at: Instant,
}
/// Serializable version of ColdIndexEntry for persistence
///
/// Converts Instant to Duration offset from saved_at time
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SerializableColdIndexEntry {
pubkey: PublicKey,
identifier: String,
event_type: EventType,
reason: RejectionReason,
/// Duration since saved_at when this entry was rejected
rejected_at_offset_secs: u64,
}
/// Serializable state for hot cache
#[derive(Debug, Serialize, Deserialize)]
struct SerializableHotCache {
expiry_duration_secs: u64,
entries: HashMap<EventId, SerializableHotCacheEntry>,
}
/// Serializable state for cold index
#[derive(Debug, Serialize, Deserialize)]
struct SerializableColdIndex {
expiry_duration_secs: u64,
entries: HashMap<EventId, SerializableColdIndexEntry>,
}
/// Complete rejected cache state for persistence
///
/// Stores both hot cache and cold index with version and timestamp information.
/// All Instant fields are converted to Duration offsets from saved_at.
#[derive(Debug, Serialize, Deserialize)]
struct RejectedCacheState {
/// Version for future compatibility
version: u32,
/// When this state was saved
saved_at: SystemTime,
/// Hot cache entries with full events
hot_cache: SerializableHotCache,
/// Cold index entries with metadata only
cold_index: SerializableColdIndex,
}
/// Hot cache: Stores full events for immediate re-processing
///
/// Events are stored for a short duration (default: 2 minutes) to enable
@@ -603,6 +662,168 @@ impl RejectedEventsIndex {
ids
}
/// Save rejected events cache to disk
///
/// Serializes both hot cache and cold index to JSON, converting Instant timestamps
/// to Duration offsets from the save time. This allows timestamps to be adjusted
/// for downtime when restored.
///
/// # Arguments
///
/// * `path` - File path to write the serialized state to
///
/// # Returns
///
/// Ok(()) on success, or an error if serialization or file write fails
pub fn save_to_disk(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
let saved_at = SystemTime::now();
let now = Instant::now();
// Lock both caches for consistent snapshot
let hot_entries = self.hot_cache.entries.read().unwrap();
let cold_entries = self.cold_index.entries.read().unwrap();
// Convert hot cache entries to serializable format
let serializable_hot_entries: HashMap<EventId, SerializableHotCacheEntry> = hot_entries
.iter()
.map(|(event_id, entry)| {
let cached_at_offset_secs = now.duration_since(entry.cached_at).as_secs();
let serializable_entry = SerializableHotCacheEntry {
event: entry.event.clone(),
pubkey: entry.pubkey,
identifier: entry.identifier.clone(),
event_type: entry.event_type,
reason: entry.reason,
cached_at_offset_secs,
};
(*event_id, serializable_entry)
})
.collect();
// Convert cold index entries to serializable format
let serializable_cold_entries: HashMap<EventId, SerializableColdIndexEntry> = cold_entries
.iter()
.map(|(event_id, entry)| {
let rejected_at_offset_secs = now.duration_since(entry.rejected_at).as_secs();
let serializable_entry = SerializableColdIndexEntry {
pubkey: entry.pubkey,
identifier: entry.identifier.clone(),
event_type: entry.event_type,
reason: entry.reason,
rejected_at_offset_secs,
};
(*event_id, serializable_entry)
})
.collect();
// Create complete state
let state = RejectedCacheState {
version: 1,
saved_at,
hot_cache: SerializableHotCache {
expiry_duration_secs: self.hot_cache.expiry_duration.as_secs(),
entries: serializable_hot_entries,
},
cold_index: SerializableColdIndex {
expiry_duration_secs: self.cold_index.expiry_duration.as_secs(),
entries: serializable_cold_entries,
},
};
// Serialize to JSON and write to file
let json = serde_json::to_string_pretty(&state)?;
std::fs::write(path, json)?;
Ok(())
}
/// Restore rejected events cache from disk
///
/// Loads the serialized state from disk and populates both hot cache and cold index.
/// Adjusts all timestamps by adding the downtime duration (time since save) to maintain
/// correct expiry behavior. Deletes the state file after successful restore.
///
/// # Arguments
///
/// * `path` - File path to read the serialized state from
///
/// # Returns
///
/// Ok(()) on success, or an error if file doesn't exist, is corrupted, or restore fails
pub fn restore_from_disk(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
// Load and parse JSON
let json = std::fs::read_to_string(path)?;
let state: RejectedCacheState = serde_json::from_str(&json)?;
// Calculate downtime (how long the relay was offline)
let now_system = SystemTime::now();
let downtime = now_system
.duration_since(state.saved_at)
.unwrap_or(Duration::ZERO);
let now_instant = Instant::now();
// Lock both caches for restoration
let mut hot_entries = self.hot_cache.entries.write().unwrap();
let mut cold_entries = self.cold_index.entries.write().unwrap();
// Restore hot cache entries
for (event_id, serializable_entry) in state.hot_cache.entries {
// Reconstruct cached_at by extending the offset by downtime
// Original offset (how long ago it was cached when saved)
let original_offset = Duration::from_secs(serializable_entry.cached_at_offset_secs);
// Total offset including downtime
let total_offset = original_offset + downtime;
// cached_at = now - total_offset
let cached_at = now_instant - total_offset;
let entry = HotCacheEntry {
event: serializable_entry.event,
pubkey: serializable_entry.pubkey,
identifier: serializable_entry.identifier,
event_type: serializable_entry.event_type,
reason: serializable_entry.reason,
cached_at,
};
hot_entries.insert(event_id, entry);
}
// Restore cold index entries
for (event_id, serializable_entry) in state.cold_index.entries {
// Reconstruct rejected_at by extending the offset by downtime
let original_offset = Duration::from_secs(serializable_entry.rejected_at_offset_secs);
let total_offset = original_offset + downtime;
// rejected_at = now - total_offset
let rejected_at = now_instant - total_offset;
let entry = ColdIndexEntry {
pubkey: serializable_entry.pubkey,
identifier: serializable_entry.identifier,
event_type: serializable_entry.event_type,
reason: serializable_entry.reason,
rejected_at,
};
cold_entries.insert(event_id, entry);
}
// Release locks before deleting file
drop(hot_entries);
drop(cold_entries);
// Delete the state file after successful restore
std::fs::remove_file(path)?;
Ok(())
}
}
#[cfg(test)]
@@ -956,4 +1177,503 @@ mod tests {
// Cold index now empty
assert_eq!(index.cold_index_len(), 0);
}
// ========================================================================
// Persistence Serialization Tests
// ========================================================================
#[tokio::test]
async fn test_save_and_restore_hot_cache_roundtrip() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let event = create_test_event().await;
let pubkey = event.pubkey;
let identifier = "test-repo".to_string();
// Add event to hot cache
index.add_announcement(
event.clone(),
pubkey,
identifier.clone(),
RejectionReason::DoesNotListService,
);
assert_eq!(index.hot_cache_len(), 1);
assert_eq!(index.cold_index_len(), 1);
// Save to disk
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Create new index and restore
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify state file was deleted after restore
assert!(!state_path.exists());
// Verify hot cache restored
assert_eq!(index2.hot_cache_len(), 1);
assert!(index2.hot_cache.contains(&event.id));
// Verify cold index restored
assert_eq!(index2.cold_index_len(), 1);
assert!(index2.cold_index.contains(&event.id));
// Verify we can retrieve the event
let events = index2
.hot_cache
.get_maintainer_events(&pubkey, &identifier, None);
assert_eq!(events.len(), 1);
assert_eq!(events[0].id, event.id);
}
#[tokio::test]
async fn test_save_and_restore_cold_index_only() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(
Duration::from_millis(50), // Hot cache expires quickly
Duration::from_secs(604800), // Cold index lasts long
);
let event = create_test_event().await;
// Add event
index.add_announcement(
event.clone(),
event.pubkey,
"test-repo".to_string(),
RejectionReason::MaintainerNotYetValid,
);
// Wait for hot cache to expire
std::thread::sleep(Duration::from_millis(60));
index.cleanup_expired_for_type("announcement");
assert_eq!(index.hot_cache_len(), 0);
assert_eq!(index.cold_index_len(), 1);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Restore into new index
let index2 =
RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify only cold index restored (hot cache was empty)
assert_eq!(index2.hot_cache_len(), 0);
assert_eq!(index2.cold_index_len(), 1);
assert!(index2.cold_index.contains(&event.id));
}
#[tokio::test]
async fn test_save_and_restore_both_hot_and_cold() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
// Create two events
let unsigned1 = nostr_sdk::EventBuilder::text_note("event1").build(keys.public_key());
let event1 = keys.sign_event(unsigned1).await.unwrap();
let unsigned2 = nostr_sdk::EventBuilder::text_note("event2").build(keys.public_key());
let event2 = keys.sign_event(unsigned2).await.unwrap();
// Add both events
index.add_announcement(
event1.clone(),
event1.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
index.add_state(
event2.clone(),
event2.pubkey,
"repo2".to_string(),
RejectionReason::Other,
);
assert_eq!(index.hot_cache_len(), 2);
assert_eq!(index.cold_index_len(), 2);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Restore into new index
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify both caches restored
assert_eq!(index2.hot_cache_len(), 2);
assert_eq!(index2.cold_index_len(), 2);
assert!(index2.contains(&event1.id));
assert!(index2.contains(&event2.id));
}
#[tokio::test]
async fn test_save_and_restore_empty_cache() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Save empty cache
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Restore into new index
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify empty state restored
assert_eq!(index2.hot_cache_len(), 0);
assert_eq!(index2.cold_index_len(), 0);
}
#[tokio::test]
async fn test_restore_missing_file() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("nonexistent.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Attempting to restore missing file should return error
let result = index.restore_from_disk(&state_path);
assert!(result.is_err());
// Index should remain empty
assert_eq!(index.hot_cache_len(), 0);
assert_eq!(index.cold_index_len(), 0);
}
#[tokio::test]
async fn test_restore_corrupted_json() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("corrupted.json");
// Write corrupted JSON
std::fs::write(&state_path, "{ invalid json !!!").unwrap();
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Attempting to restore corrupted file should return error
let result = index.restore_from_disk(&state_path);
assert!(result.is_err());
// Index should remain empty
assert_eq!(index.hot_cache_len(), 0);
assert_eq!(index.cold_index_len(), 0);
}
#[tokio::test]
async fn test_file_cleanup_after_successful_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let event = create_test_event().await;
index.add_announcement(
event.clone(),
event.pubkey,
"test-repo".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Restore
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// File should be deleted after successful restore
assert!(!state_path.exists());
}
#[tokio::test]
async fn test_downtime_calculation_preserves_expiry() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let event = create_test_event().await;
index.add_announcement(
event.clone(),
event.pubkey,
"test-repo".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Simulate downtime by sleeping
std::thread::sleep(Duration::from_millis(100));
// Restore
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Event should still be in both caches (downtime accounted for)
assert_eq!(index2.hot_cache_len(), 1);
assert_eq!(index2.cold_index_len(), 1);
assert!(index2.contains(&event.id));
}
#[tokio::test]
async fn test_entries_expired_during_downtime() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
// Create index with very short expiry
let index = RejectedEventsIndex::new(
Duration::from_millis(100), // Hot cache: 100ms
Duration::from_millis(200), // Cold index: 200ms
);
let event = create_test_event().await;
index.add_announcement(
event.clone(),
event.pubkey,
"test-repo".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Simulate downtime longer than hot cache expiry
std::thread::sleep(Duration::from_millis(150));
// Restore
let index2 =
RejectedEventsIndex::new(Duration::from_millis(100), Duration::from_millis(200));
index2.restore_from_disk(&state_path).unwrap();
// Hot cache entry should have expired during downtime
// Cold index should still have it (200ms expiry)
assert_eq!(index2.hot_cache_len(), 1);
assert_eq!(index2.cold_index_len(), 1);
// But when we try to get it, hot cache will see it's expired
let events = index2
.hot_cache
.get_maintainer_events(&event.pubkey, "test-repo", None);
assert_eq!(events.len(), 0); // Expired!
// Cleanup should remove it
let (hot_expired, cold_expired) = index2.cleanup_expired_for_type("announcement");
assert_eq!(hot_expired, 1);
assert_eq!(cold_expired, 0); // Not expired yet
}
#[tokio::test]
async fn test_hot_cache_different_event_types() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
// Create announcement event
let unsigned_ann =
nostr_sdk::EventBuilder::text_note("announcement").build(keys.public_key());
let event_ann = keys.sign_event(unsigned_ann).await.unwrap();
// Create state event
let unsigned_state = nostr_sdk::EventBuilder::text_note("state").build(keys.public_key());
let event_state = keys.sign_event(unsigned_state).await.unwrap();
// Add both types
index.add_announcement(
event_ann.clone(),
event_ann.pubkey,
"test-repo".to_string(),
RejectionReason::DoesNotListService,
);
index.add_state(
event_state.clone(),
event_state.pubkey,
"test-repo".to_string(),
RejectionReason::Other,
);
// Save and restore
index.save_to_disk(&state_path).unwrap();
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify both event types restored
assert_eq!(index2.hot_cache_len(), 2);
assert!(index2.contains(&event_ann.id));
assert!(index2.contains(&event_state.id));
// Verify we can filter by type
let (removed, events) = index2.invalidate_and_get(
&event_ann.pubkey,
"test-repo",
Some(EventType::Announcement),
);
assert_eq!(removed, 1);
assert_eq!(events.len(), 1);
assert_eq!(events[0].id, event_ann.id);
}
#[tokio::test]
async fn test_cold_index_different_rejection_reasons() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
// Create events with different rejection reasons
let unsigned1 = nostr_sdk::EventBuilder::text_note("event1").build(keys.public_key());
let event1 = keys.sign_event(unsigned1).await.unwrap();
let unsigned2 = nostr_sdk::EventBuilder::text_note("event2").build(keys.public_key());
let event2 = keys.sign_event(unsigned2).await.unwrap();
let unsigned3 = nostr_sdk::EventBuilder::text_note("event3").build(keys.public_key());
let event3 = keys.sign_event(unsigned3).await.unwrap();
// Add with different rejection reasons
index.add_announcement(
event1.clone(),
event1.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
index.add_announcement(
event2.clone(),
event2.pubkey,
"repo2".to_string(),
RejectionReason::MaintainerNotYetValid,
);
index.add_announcement(
event3.clone(),
event3.pubkey,
"repo3".to_string(),
RejectionReason::Other,
);
// Save and restore
index.save_to_disk(&state_path).unwrap();
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify all entries restored with their rejection reasons
assert_eq!(index2.cold_index_len(), 3);
assert!(index2.contains(&event1.id));
assert!(index2.contains(&event2.id));
assert!(index2.contains(&event3.id));
}
#[tokio::test]
async fn test_multiple_save_restore_cycles() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
// First cycle
let index1 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let event1 = create_test_event().await;
index1.add_announcement(
event1.clone(),
event1.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
index1.save_to_disk(&state_path).unwrap();
// Second cycle - restore and add more
let index2 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
let event2 = create_test_event().await;
index2.add_announcement(
event2.clone(),
event2.pubkey,
"repo2".to_string(),
RejectionReason::MaintainerNotYetValid,
);
assert_eq!(index2.hot_cache_len(), 2);
index2.save_to_disk(&state_path).unwrap();
// Third cycle - restore again
let index3 =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index3.restore_from_disk(&state_path).unwrap();
// Verify both events survived multiple cycles
assert_eq!(index3.hot_cache_len(), 2);
assert!(index3.contains(&event1.id));
assert!(index3.contains(&event2.id));
}
#[tokio::test]
async fn test_restore_preserves_remaining_ttl() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
// Create index with 2 second hot cache expiry
let index = RejectedEventsIndex::new(Duration::from_secs(2), Duration::from_secs(604800));
let event = create_test_event().await;
index.add_announcement(
event.clone(),
event.pubkey,
"test-repo".to_string(),
RejectionReason::DoesNotListService,
);
// Wait 200ms (small fraction of TTL)
std::thread::sleep(Duration::from_millis(200));
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Immediately restore (minimal downtime)
let index2 = RejectedEventsIndex::new(Duration::from_secs(2), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Event should still be retrievable (has ~1.8s remaining)
let events = index2
.hot_cache
.get_maintainer_events(&event.pubkey, "test-repo", None);
assert_eq!(events.len(), 1);
// Wait 2 seconds (total 2.2s > 2s expiry)
std::thread::sleep(Duration::from_secs(2));
// Now it should be expired
let events = index2
.hot_cache
.get_maintainer_events(&event.pubkey, "test-repo", None);
assert_eq!(events.len(), 0);
}
}
tests/purgatory_persistence.rs
+755
View File
@@ -0,0 +1,755 @@
//! Purgatory Persistence Integration Tests
//!
//! Tests that verify the full purgatory persistence save/restore cycle:
//! - Purgatory save/restore with state events, PR events, and expired events
//! - Rejected cache save/restore with hot cache and cold index entries
//! - Integration with shutdown/startup hooks
//! - Graceful degradation with missing or corrupted files
//! - Time adjustment for downtime
//!
//! # Test Strategy
//!
//! These tests verify end-to-end persistence functionality:
//! 1. Create purgatory/rejected cache instances with various entries
//! 2. Save state to disk
//! 3. Create new instances and restore from disk
//! 4. Verify all data is restored correctly
//! 5. Verify system continues to work after restore
//!
//! # Running Tests
//!
//! ```bash
//! # Run all purgatory persistence tests
//! cargo test --test purgatory_persistence
//!
//! # Run specific test
//! cargo test --test purgatory_persistence test_full_purgatory_save_restore_cycle
//!
//! # With output for debugging
//! cargo test --test purgatory_persistence -- --nocapture
//! ```
mod common;
use ngit_grasp::purgatory::Purgatory;
use ngit_grasp::sync::rejected_index::{EventType, RejectedEventsIndex, RejectionReason};
use nostr_sdk::prelude::*;
use std::time::Duration;
/// Helper to create a test event
async fn create_test_event(keys: &Keys, content: &str) -> Event {
EventBuilder::text_note(content)
.sign_with_keys(keys)
.unwrap()
}
/// Helper to create a state event with specific refs
fn create_state_event_with_refs(
keys: &Keys,
identifier: &str,
refs: &[(&str, &str)],
) -> Result<Event, Box<dyn std::error::Error>> {
let mut tags = vec![Tag::identifier(identifier)];
// Add ref tags
for (ref_name, commit_hash) in refs {
tags.push(Tag::custom(
TagKind::custom("ref"),
vec![ref_name.to_string(), commit_hash.to_string()],
));
}
let event = EventBuilder::new(Kind::from(30618), "")
.tags(tags)
.sign_with_keys(keys)?;
Ok(event)
}
/// Test 1: Full save/restore cycle with state events, PR events, and expired events
#[tokio::test]
async fn test_full_purgatory_save_restore_cycle() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
// Create purgatory instance
let purgatory = Purgatory::new(&git_data_path);
// Create test keys and events
let keys1 = Keys::generate();
let keys2 = Keys::generate();
let keys3 = Keys::generate();
let state_event1 =
create_state_event_with_refs(&keys1, "repo1", &[("main", "abc123")]).unwrap();
let state_event2 =
create_state_event_with_refs(&keys2, "repo2", &[("main", "def456")]).unwrap();
let pr_event1 = create_test_event(&keys3, "PR 1").await;
let pr_event2 = create_test_event(&keys3, "PR 2").await;
// Add state events to purgatory
purgatory.add_state(
state_event1.clone(),
"repo1".to_string(),
keys1.public_key(),
);
purgatory.add_state(
state_event2.clone(),
"repo2".to_string(),
keys2.public_key(),
);
// Add PR events to purgatory
purgatory.add_pr(
pr_event1.clone(),
pr_event1.id.to_hex(),
"commit-abc".to_string(),
);
purgatory.add_pr(
pr_event2.clone(),
pr_event2.id.to_hex(),
"commit-def".to_string(),
);
// Add a PR placeholder (git-data-first scenario)
purgatory.add_pr_placeholder("placeholder-id".to_string(), "commit-xyz".to_string());
// Note: We can't directly test expired events without accessing private fields,
// so we'll focus on testing state and PR events persistence
// Verify initial counts
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 2, "Should have 2 state events");
assert_eq!(
pr_count, 3,
"Should have 3 PR events (2 events + 1 placeholder)"
);
// Save to disk
purgatory.save_to_disk(&state_path).unwrap();
assert!(state_path.exists(), "State file should exist after save");
// Create new purgatory instance and restore
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify state file was deleted after restore
assert!(
!state_path.exists(),
"State file should be deleted after restore"
);
// Verify all data was restored
let (state_count2, pr_count2) = purgatory2.count();
assert_eq!(state_count2, 2, "Should have 2 state events after restore");
assert_eq!(
pr_count2, 3,
"Should have 3 PR events after restore (2 events + 1 placeholder)"
);
// Verify specific state events
let repo1_states = purgatory2.find_state("repo1");
assert_eq!(repo1_states.len(), 1);
assert_eq!(repo1_states[0].event.id, state_event1.id);
let repo2_states = purgatory2.find_state("repo2");
assert_eq!(repo2_states.len(), 1);
assert_eq!(repo2_states[0].event.id, state_event2.id);
// Verify PR events
let pr1 = purgatory2.find_pr(&pr_event1.id.to_hex());
assert!(pr1.is_some());
assert_eq!(pr1.unwrap().commit, "commit-abc");
let pr2 = purgatory2.find_pr(&pr_event2.id.to_hex());
assert!(pr2.is_some());
assert_eq!(pr2.unwrap().commit, "commit-def");
// Verify placeholder
let placeholder = purgatory2.find_pr_placeholder("placeholder-id");
assert_eq!(placeholder, Some("commit-xyz".to_string()));
// Verify re-queueing works - get all identifiers
let identifiers = purgatory2.get_all_identifiers();
assert_eq!(identifiers.len(), 2);
assert!(identifiers.contains(&"repo1".to_string()));
assert!(identifiers.contains(&"repo2".to_string()));
}
/// Test 2: Rejected cache integration - save/restore hot cache and cold index
#[tokio::test]
async fn test_rejected_cache_save_restore_cycle() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
// Create rejected events index
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Create test events
let keys1 = Keys::generate();
let keys2 = Keys::generate();
let event1 = create_test_event(&keys1, "announcement 1").await;
let event2 = create_test_event(&keys2, "announcement 2").await;
let event3 = create_test_event(&keys1, "state 1").await;
// Add announcements to rejected cache
index.add_announcement(
event1.clone(),
event1.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
index.add_announcement(
event2.clone(),
event2.pubkey,
"repo2".to_string(),
RejectionReason::MaintainerNotYetValid,
);
// Add state event to rejected cache
index.add_state(
event3.clone(),
event3.pubkey,
"repo1".to_string(),
RejectionReason::Other,
);
// Verify initial counts
assert_eq!(index.hot_cache_len(), 3);
assert_eq!(index.cold_index_len(), 3);
// Save to disk
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Create new index and restore
let index2 = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify state file was deleted
assert!(!state_path.exists());
// Verify all entries restored
assert_eq!(index2.hot_cache_len(), 3);
assert_eq!(index2.cold_index_len(), 3);
// Verify specific entries
assert!(index2.contains(&event1.id));
assert!(index2.contains(&event2.id));
assert!(index2.contains(&event3.id));
// Verify we can invalidate and get events
let (removed, hot_events) =
index2.invalidate_and_get(&event1.pubkey, "repo1", Some(EventType::Announcement));
assert_eq!(removed, 1);
assert_eq!(hot_events.len(), 1);
assert_eq!(hot_events[0].id, event1.id);
}
/// Test 3: Simulated downtime - verify expiry times are adjusted correctly
#[tokio::test]
async fn test_purgatory_downtime_adjustment() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
let keys = Keys::generate();
let state_event = create_state_event_with_refs(&keys, "repo1", &[("main", "abc123")])
.unwrap();
purgatory.add_state(state_event.clone(), "repo1".to_string(), keys.public_key());
// Save to disk
purgatory.save_to_disk(&state_path).unwrap();
// Simulate downtime
tokio::time::sleep(Duration::from_millis(100)).await;
// Restore
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify event is still there (downtime was accounted for)
let (state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
let repo1_states = purgatory2.find_state("repo1");
assert_eq!(repo1_states.len(), 1);
assert_eq!(repo1_states[0].event.id, state_event.id);
// Verify the event hasn't expired yet (expiry time was adjusted)
// The event should have ~30 minutes minus the downtime
let entry = &repo1_states[0];
let remaining = entry
.expires_at
.saturating_duration_since(std::time::Instant::now());
assert!(
remaining > Duration::from_secs(1700),
"Event should have most of its 30min expiry remaining"
);
}
/// Test 4: Rejected cache downtime adjustment
#[tokio::test]
async fn test_rejected_cache_downtime_adjustment() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
let event = create_test_event(&keys, "test").await;
index.add_announcement(
event.clone(),
event.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Simulate downtime
tokio::time::sleep(Duration::from_millis(100)).await;
// Restore
let index2 = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify event is still in both caches (downtime was accounted for)
assert_eq!(index2.hot_cache_len(), 1);
assert_eq!(index2.cold_index_len(), 1);
assert!(index2.contains(&event.id));
}
/// Test 5: File cleanup - verify state files are deleted after successful restore
#[tokio::test]
async fn test_purgatory_file_cleanup_after_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
let keys = Keys::generate();
let state_event = create_state_event_with_refs(&keys, "repo1", &[("main", "abc123")])
.unwrap();
purgatory.add_state(state_event, "repo1".to_string(), keys.public_key());
// Save to disk
purgatory.save_to_disk(&state_path).unwrap();
assert!(state_path.exists(), "State file should exist after save");
// Restore
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify file was deleted
assert!(
!state_path.exists(),
"State file should be deleted after successful restore"
);
}
/// Test 6: Rejected cache file cleanup
#[tokio::test]
async fn test_rejected_cache_file_cleanup_after_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
let event = create_test_event(&keys, "test").await;
index.add_announcement(
event,
keys.public_key(),
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Restore
let index2 = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify file was deleted
assert!(!state_path.exists());
}
/// Test 7: Graceful degradation - missing purgatory file
#[tokio::test]
async fn test_purgatory_restore_missing_file() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("nonexistent.json");
let purgatory = Purgatory::new(&git_data_path);
// Attempting to restore missing file should return error
let result = purgatory.restore_from_disk(&state_path);
assert!(result.is_err(), "Should error on missing file");
// Purgatory should still be usable (empty state)
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
// Should be able to add events normally
let keys = Keys::generate();
let event = create_test_event(&keys, "test").await;
purgatory.add_state(event, "repo1".to_string(), keys.public_key());
let (state_count, _) = purgatory.count();
assert_eq!(state_count, 1);
}
/// Test 8: Graceful degradation - missing rejected cache file
#[tokio::test]
async fn test_rejected_cache_restore_missing_file() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("nonexistent.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Attempting to restore missing file should return error
let result = index.restore_from_disk(&state_path);
assert!(result.is_err());
// Index should still be usable (empty state)
assert_eq!(index.hot_cache_len(), 0);
assert_eq!(index.cold_index_len(), 0);
// Should be able to add events normally
let keys = Keys::generate();
let event = create_test_event(&keys, "test").await;
index.add_announcement(
event,
keys.public_key(),
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
assert_eq!(index.hot_cache_len(), 1);
assert_eq!(index.cold_index_len(), 1);
}
/// Test 9: Graceful degradation - corrupted purgatory file
#[tokio::test]
async fn test_purgatory_restore_corrupted_file() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("corrupted.json");
// Write corrupted JSON
std::fs::write(&state_path, "{ invalid json !!!").unwrap();
let purgatory = Purgatory::new(&git_data_path);
// Attempting to restore corrupted file should return error
let result = purgatory.restore_from_disk(&state_path);
assert!(result.is_err(), "Should error on corrupted file");
// Purgatory should still be usable
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
/// Test 10: Graceful degradation - corrupted rejected cache file
#[tokio::test]
async fn test_rejected_cache_restore_corrupted_file() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("corrupted.json");
// Write corrupted JSON
std::fs::write(&state_path, "{ invalid json !!!").unwrap();
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Attempting to restore corrupted file should return error
let result = index.restore_from_disk(&state_path);
assert!(result.is_err());
// Index should still be usable
assert_eq!(index.hot_cache_len(), 0);
assert_eq!(index.cold_index_len(), 0);
}
/// Test 11: Empty purgatory save/restore
#[tokio::test]
async fn test_empty_purgatory_save_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
// Save empty purgatory
purgatory.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Restore
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify empty state
let (state_count, pr_count) = purgatory2.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
assert_eq!(purgatory2.expired_count(), 0);
}
/// Test 12: Empty rejected cache save/restore
#[tokio::test]
async fn test_empty_rejected_cache_save_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Save empty cache
index.save_to_disk(&state_path).unwrap();
assert!(state_path.exists());
// Restore
let index2 = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Verify empty state
assert_eq!(index2.hot_cache_len(), 0);
assert_eq!(index2.cold_index_len(), 0);
}
/// Test 13: Multiple state events for same identifier
#[tokio::test]
async fn test_purgatory_multiple_state_events_same_identifier() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
// Create multiple state events for same identifier (different maintainers)
let keys1 = Keys::generate();
let keys2 = Keys::generate();
let event1 = create_state_event_with_refs(&keys1, "repo1", &[("main", "abc123")])
.unwrap();
let event2 = create_state_event_with_refs(&keys2, "repo1", &[("main", "def456")])
.unwrap();
purgatory.add_state(event1.clone(), "repo1".to_string(), keys1.public_key());
purgatory.add_state(event2.clone(), "repo1".to_string(), keys2.public_key());
// Save and restore
purgatory.save_to_disk(&state_path).unwrap();
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify both events restored
let repo1_states = purgatory2.find_state("repo1");
assert_eq!(repo1_states.len(), 2);
let event_ids: Vec<_> = repo1_states.iter().map(|e| e.event.id).collect();
assert!(event_ids.contains(&event1.id));
assert!(event_ids.contains(&event2.id));
}
/// Test 14: Verify system continues to work after restore
#[tokio::test]
async fn test_purgatory_continues_working_after_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
let keys = Keys::generate();
let event1 = create_state_event_with_refs(&keys, "repo1", &[("main", "abc123")])
.unwrap();
purgatory.add_state(event1.clone(), "repo1".to_string(), keys.public_key());
// Save and restore
purgatory.save_to_disk(&state_path).unwrap();
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Add new events after restore
let event2 = create_state_event_with_refs(&keys, "repo2", &[("main", "xyz789")])
.unwrap();
purgatory2.add_state(event2.clone(), "repo2".to_string(), keys.public_key());
// Verify both old and new events work
let (state_count, _) = purgatory2.count();
assert_eq!(state_count, 2);
let repo1_states = purgatory2.find_state("repo1");
assert_eq!(repo1_states.len(), 1);
assert_eq!(repo1_states[0].event.id, event1.id);
let repo2_states = purgatory2.find_state("repo2");
assert_eq!(repo2_states.len(), 1);
assert_eq!(repo2_states[0].event.id, event2.id);
// Verify cleanup still works
let (state_removed, pr_removed) = purgatory2.cleanup();
// Nothing should be expired yet
assert_eq!(state_removed, 0);
assert_eq!(pr_removed, 0);
}
/// Test 15: Verify rejected cache continues working after restore
#[tokio::test]
async fn test_rejected_cache_continues_working_after_restore() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys = Keys::generate();
let event1 = create_test_event(&keys, "event1").await;
index.add_announcement(
event1.clone(),
event1.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
// Save and restore
index.save_to_disk(&state_path).unwrap();
let index2 = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
index2.restore_from_disk(&state_path).unwrap();
// Add new events after restore
let event2 = create_test_event(&keys, "event2").await;
index2.add_announcement(
event2.clone(),
event2.pubkey,
"repo2".to_string(),
RejectionReason::MaintainerNotYetValid,
);
// Verify both old and new events work
assert_eq!(index2.hot_cache_len(), 2);
assert_eq!(index2.cold_index_len(), 2);
assert!(index2.contains(&event1.id));
assert!(index2.contains(&event2.id));
// Verify invalidation still works
let (removed, hot_events) =
index2.invalidate_and_get(&event1.pubkey, "repo1", Some(EventType::Announcement));
assert_eq!(removed, 1);
assert_eq!(hot_events.len(), 1);
assert_eq!(hot_events[0].id, event1.id);
}
/// Test 16: Entries that expired during downtime are properly handled
#[tokio::test]
async fn test_purgatory_entries_expired_during_downtime() {
let temp_dir = tempfile::tempdir().unwrap();
let git_data_path = temp_dir.path().join("git");
let state_path = temp_dir.path().join("purgatory.json");
let purgatory = Purgatory::new(&git_data_path);
let keys = Keys::generate();
let event = create_state_event_with_refs(&keys, "repo1", &[("main", "abc123")])
.unwrap();
purgatory.add_state(event.clone(), "repo1".to_string(), keys.public_key());
// Save to disk
purgatory.save_to_disk(&state_path).unwrap();
// Simulate very long downtime (longer than the 30min default expiry)
// Note: We can't manually set expiry without accessing private fields,
// so this test verifies that the system handles already-expired entries gracefully
// In a real scenario, if downtime > 30 minutes, entries would be expired on restore
// For this test, we'll just verify the restore works and cleanup can be called
let purgatory2 = Purgatory::new(&git_data_path);
purgatory2.restore_from_disk(&state_path).unwrap();
// Event should be restored
let (state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
// Cleanup should work (even if nothing is expired yet)
let (state_removed, _) = purgatory2.cleanup();
// Nothing expired yet since we didn't wait 30 minutes
assert_eq!(state_removed, 0);
let (state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
}
/// Test 17: Rejected cache entries that expired during downtime
#[tokio::test]
async fn test_rejected_cache_entries_expired_during_downtime() {
let temp_dir = tempfile::tempdir().unwrap();
let state_path = temp_dir.path().join("rejected_cache.json");
// Create index with very short expiry
let index = RejectedEventsIndex::new(
Duration::from_millis(50), // Hot cache: 50ms
Duration::from_millis(100), // Cold index: 100ms
);
let keys = Keys::generate();
let event = create_test_event(&keys, "test").await;
index.add_announcement(
event.clone(),
event.pubkey,
"repo1".to_string(),
RejectionReason::DoesNotListService,
);
// Save to disk
index.save_to_disk(&state_path).unwrap();
// Simulate downtime longer than hot cache expiry
tokio::time::sleep(Duration::from_millis(75)).await;
// Restore
let index2 = RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_millis(100));
index2.restore_from_disk(&state_path).unwrap();
// Both should be restored initially
assert_eq!(index2.hot_cache_len(), 1);
assert_eq!(index2.cold_index_len(), 1);
// Note: We can't directly access hot_cache.get_maintainer_events (private method)
// But we can verify the entry is there via contains() and test cleanup
// Verify entry is still tracked
assert!(index2.contains(&event.id));
// Cleanup should remove expired hot cache entry
let (hot_expired, cold_expired) = index2.cleanup_expired_for_type("announcement");
assert_eq!(hot_expired, 1);
assert_eq!(cold_expired, 0); // Cold index still valid
assert_eq!(index2.hot_cache_len(), 0);
assert_eq!(index2.cold_index_len(), 1);
}