Add purgatory persistence to survive relay restarts

Implement save/restore functionality for both purgatory state and
rejected events cache. Events are now saved to disk on graceful
shutdown and restored on startup, preventing data loss during
relay restarts.

Key features:
- Purgatory state persisted to JSON (state events, PR events, expired events)
- Rejected events cache persisted (hot cache + cold index)
- Downtime adjustment preserves remaining TTL
- Graceful degradation on missing/corrupted files
- Automatic re-queueing of restored repositories
- Comprehensive test coverage (45 tests)
This commit is contained in:
DanConwayDev
2026-01-14 10:46:30 +00:00
7 changed files with 2725 additions and 20 deletions
+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() {
+923 -2
View File
@@ -12,6 +12,7 @@
//! - **Separate stores**: State events and PR events use different indexing strategies
mod helpers;
pub mod persistence;
pub mod sync;
mod types;
@@ -20,10 +21,12 @@ pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
use dashmap::DashMap;
use nostr_sdk::prelude::*;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::time::{Duration, Instant, SystemTime};
pub use sync::SyncQueueEntry;
@@ -38,6 +41,63 @@ const DEFAULT_SYNC_DELAY: Duration = Duration::from_secs(180);
/// Used for batching burst arrivals during negentropy sync.
const IMMEDIATE_SYNC_DELAY: Duration = Duration::from_millis(500);
/// Serializable wrapper for `StatePurgatoryEntry` with time offsets.
///
/// Stores `Instant` fields as `Duration` offsets from the `saved_at` timestamp
/// in `PurgatoryState`, allowing state to be persisted and restored across restarts.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SerializableStatePurgatoryEntry {
/// The nostr state event (kind 30618) awaiting git data
event: Event,
/// The repository identifier from the event's 'd' tag
identifier: String,
/// Event author pubkey
author: PublicKey,
/// Duration offset from saved_at for created_at
created_at_offset_secs: u64,
/// Duration offset from saved_at for expires_at
expires_at_offset_secs: u64,
}
/// Serializable wrapper for `PrPurgatoryEntry` with time offsets.
///
/// Stores `Instant` fields as `Duration` offsets from the `saved_at` timestamp
/// in `PurgatoryState`, allowing state to be persisted and restored across restarts.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SerializablePrPurgatoryEntry {
/// The nostr PR event, if received (None = git data arrived first)
event: Option<Event>,
/// The expected commit SHA from 'c' tag (if event exists)
/// or the actual commit pushed (if git arrived first)
commit: String,
/// Duration offset from saved_at for created_at
created_at_offset_secs: u64,
/// Duration offset from saved_at for expires_at
expires_at_offset_secs: u64,
}
/// Serializable purgatory state for disk persistence.
///
/// Contains all purgatory data needed to restore state across restarts:
/// - State events (indexed by identifier)
/// - PR events (indexed by event ID)
/// - Expired events (to prevent re-sync loops)
/// - Version number for future compatibility
/// - Saved timestamp for downtime calculation
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PurgatoryState {
/// Version number for state format (currently 1)
version: u32,
/// When this state was saved to disk
saved_at: SystemTime,
/// State events indexed by repository identifier
state_events: HashMap<String, Vec<SerializableStatePurgatoryEntry>>,
/// PR events indexed by event ID (hex string)
pr_events: HashMap<String, SerializablePrPurgatoryEntry>,
/// Expired event IDs with their expiry timestamps
expired_events: HashMap<String, SystemTime>,
}
/// Main purgatory structure holding events awaiting git data.
///
/// Provides thread-safe concurrent access to two separate stores:
@@ -667,6 +727,260 @@ impl Purgatory {
pub fn sync_queue_size(&self) -> usize {
self.sync_queue.len()
}
/// Get all repository identifiers currently in purgatory.
///
/// Returns a list of all unique repository identifiers that have state events
/// in purgatory. This is useful for re-queueing repositories after restore.
///
/// # Returns
/// Vector of repository identifiers (e.g., "owner/repo")
///
/// # Example
/// ```no_run
/// use ngit_grasp::purgatory::Purgatory;
/// use std::path::PathBuf;
///
/// let purgatory = Purgatory::new(PathBuf::from("/tmp/git"));
/// let identifiers = purgatory.get_all_identifiers();
/// for id in identifiers {
/// println!("Repository in purgatory: {}", id);
/// }
/// ```
pub fn get_all_identifiers(&self) -> Vec<String> {
self.state_events
.iter()
.map(|entry| entry.key().clone())
.collect()
}
/// Save purgatory state to disk.
///
/// Serializes the current purgatory state (state_events, pr_events, expired_events)
/// to JSON and saves it to the specified path. Time-based fields (`Instant`) are
/// converted to duration offsets from the current `SystemTime` for persistence.
///
/// Note: The sync_queue is NOT persisted - it will be rebuilt when events are
/// restored from disk.
///
/// # Arguments
/// * `path` - Path to save the state file
///
/// # Returns
/// Ok(()) on success, Err on failure
///
/// # Example
/// ```no_run
/// use ngit_grasp::purgatory::Purgatory;
/// use std::path::PathBuf;
///
/// let purgatory = Purgatory::new(PathBuf::from("/tmp/git"));
/// purgatory.save_to_disk(&PathBuf::from("/tmp/purgatory.json")).unwrap();
/// ```
pub fn save_to_disk(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
let saved_at = SystemTime::now();
let now_instant = Instant::now();
// Convert state_events to serializable format
let mut state_events = HashMap::new();
for entry in self.state_events.iter() {
let identifier = entry.key().clone();
let entries: Vec<SerializableStatePurgatoryEntry> = entry
.value()
.iter()
.map(|e| {
let created_offset =
persistence::instant_to_offset(e.created_at, saved_at, now_instant);
let expires_offset =
persistence::instant_to_offset(e.expires_at, saved_at, now_instant);
SerializableStatePurgatoryEntry {
event: e.event.clone(),
identifier: e.identifier.clone(),
author: e.author,
created_at_offset_secs: created_offset.as_secs(),
expires_at_offset_secs: expires_offset.as_secs(),
}
})
.collect();
state_events.insert(identifier, entries);
}
// Convert pr_events to serializable format
let mut pr_events = HashMap::new();
for entry in self.pr_events.iter() {
let event_id = entry.key().clone();
let e = entry.value();
let created_offset =
persistence::instant_to_offset(e.created_at, saved_at, now_instant);
let expires_offset =
persistence::instant_to_offset(e.expires_at, saved_at, now_instant);
let serializable = SerializablePrPurgatoryEntry {
event: e.event.clone(),
commit: e.commit.clone(),
created_at_offset_secs: created_offset.as_secs(),
expires_at_offset_secs: expires_offset.as_secs(),
};
pr_events.insert(event_id, serializable);
}
// Convert expired_events to serializable format
// We use SystemTime instead of Instant offsets for expired events since
// we don't need high precision for cleanup timing
let mut expired_events = HashMap::new();
for entry in self.expired_events.iter() {
let event_id = entry.key().to_hex();
// Convert Instant to SystemTime (approximate)
let expired_at_instant = *entry.value();
let elapsed_since_expire = now_instant.saturating_duration_since(expired_at_instant);
let expired_at_system = saved_at - elapsed_since_expire;
expired_events.insert(event_id, expired_at_system);
}
// Create state structure
let state = PurgatoryState {
version: 1,
saved_at,
state_events,
pr_events,
expired_events,
};
// Serialize to JSON and write to file
let json = serde_json::to_string_pretty(&state)?;
std::fs::write(path, json)?;
tracing::info!(
path = %path.display(),
state_events = state.state_events.len(),
pr_events = state.pr_events.len(),
expired_events = state.expired_events.len(),
"Saved purgatory state to disk"
);
Ok(())
}
/// Restore purgatory state from disk.
///
/// Loads a previously saved purgatory state from the specified path and populates
/// the current purgatory instance. Adjusts time-based fields to account for downtime
/// between save and restore.
///
/// After successful restore, the state file is deleted to prevent accidental
/// double-restore.
///
/// # Arguments
/// * `path` - Path to the saved state file
///
/// # Returns
/// Ok(()) on success, Err if file doesn't exist or is corrupted
///
/// # Example
/// ```no_run
/// use ngit_grasp::purgatory::Purgatory;
/// use std::path::PathBuf;
///
/// let purgatory = Purgatory::new(PathBuf::from("/tmp/git"));
/// match purgatory.restore_from_disk(&PathBuf::from("/tmp/purgatory.json")) {
/// Ok(()) => println!("State restored successfully"),
/// Err(e) => eprintln!("Failed to restore state: {}", e),
/// }
/// ```
pub fn restore_from_disk(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
// Read and parse state file
let json = std::fs::read_to_string(path)?;
let state: PurgatoryState = serde_json::from_str(&json)?;
// Verify version
if state.version != 1 {
return Err(format!("Unsupported state version: {}", state.version).into());
}
let now_instant = Instant::now();
// Restore state_events
for (identifier, entries) in state.state_events {
let restored_entries: Vec<StatePurgatoryEntry> = entries
.into_iter()
.map(|e| {
let created_at = persistence::offset_to_instant(
Duration::from_secs(e.created_at_offset_secs),
state.saved_at,
now_instant,
);
let expires_at = persistence::offset_to_instant(
Duration::from_secs(e.expires_at_offset_secs),
state.saved_at,
now_instant,
);
StatePurgatoryEntry {
event: e.event,
identifier: e.identifier,
author: e.author,
created_at,
expires_at,
}
})
.collect();
self.state_events.insert(identifier, restored_entries);
}
// Restore pr_events
for (event_id, e) in state.pr_events {
let created_at = persistence::offset_to_instant(
Duration::from_secs(e.created_at_offset_secs),
state.saved_at,
now_instant,
);
let expires_at = persistence::offset_to_instant(
Duration::from_secs(e.expires_at_offset_secs),
state.saved_at,
now_instant,
);
let entry = PrPurgatoryEntry {
event: e.event,
commit: e.commit,
created_at,
expires_at,
};
self.pr_events.insert(event_id, entry);
}
// Restore expired_events
for (event_id_hex, expired_at_system) in state.expired_events {
if let Ok(event_id) = EventId::from_hex(&event_id_hex) {
// Convert SystemTime back to Instant (approximate)
let elapsed_since_expire = SystemTime::now()
.duration_since(expired_at_system)
.unwrap_or(Duration::ZERO);
let expired_at_instant = now_instant - elapsed_since_expire;
self.expired_events.insert(event_id, expired_at_instant);
}
}
tracing::info!(
path = %path.display(),
state_events = self.state_events.len(),
pr_events = self.pr_events.len(),
expired_events = self.expired_events.len(),
saved_at = ?state.saved_at,
"Restored purgatory state from disk"
);
// Delete state file after successful restore
std::fs::remove_file(path)?;
tracing::debug!(path = %path.display(), "Deleted state file after restore");
Ok(())
}
}
#[cfg(test)]
@@ -1249,3 +1563,610 @@ fn test_user_can_resubmit_expired_event() {
// - Skip the expired check for user-submitted events
// - Allow the event to be re-added to purgatory or accepted if git data now exists
}
// ============================================================================
// Persistence Serialization Tests
// ============================================================================
#[tokio::test]
async fn test_save_and_restore_state_events() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add multiple state events for the same identifier
let event1 = EventBuilder::text_note("state event 1")
.sign_with_keys(&keys)
.unwrap();
let event2 = EventBuilder::text_note("state event 2")
.sign_with_keys(&keys)
.unwrap();
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());
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Verify file exists
assert!(state_file.exists());
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify file was deleted after restore
assert!(!state_file.exists());
// Verify state events were restored
let (state_count, _) = purgatory2.count();
assert_eq!(state_count, 2);
let restored_entries = purgatory2.find_state("test-repo");
assert_eq!(restored_entries.len(), 2);
// Verify event IDs match
let restored_ids: Vec<EventId> = restored_entries.iter().map(|e| e.event.id).collect();
assert!(restored_ids.contains(&event1_id));
assert!(restored_ids.contains(&event2_id));
// Verify identifiers and authors match
for entry in &restored_entries {
assert_eq!(entry.identifier, "test-repo");
assert_eq!(entry.author, keys.public_key());
}
}
#[tokio::test]
async fn test_save_and_restore_pr_events() {
use nostr_sdk::{Kind, Tag, TagKind};
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add PR event with actual event
let tags = vec![Tag::custom(
TagKind::Custom("a".into()),
vec!["30617:abc123:test-repo".to_string()],
)];
let pr_event = EventBuilder::new(Kind::from(1618), "PR content")
.tags(tags)
.sign_with_keys(&keys)
.unwrap();
let pr_event_id = pr_event.id;
purgatory.add_pr(
pr_event.clone(),
"pr-event-id".to_string(),
"commit-abc".to_string(),
);
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify PR event was restored
let (_, pr_count) = purgatory2.count();
assert_eq!(pr_count, 1);
let restored_entry = purgatory2.find_pr("pr-event-id").unwrap();
assert!(restored_entry.event.is_some());
assert_eq!(restored_entry.event.unwrap().id, pr_event_id);
assert_eq!(restored_entry.commit, "commit-abc");
}
#[tokio::test]
async fn test_save_and_restore_pr_placeholders() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
// Add PR placeholder (git data arrived first)
purgatory.add_pr_placeholder("placeholder-id".to_string(), "commit-def".to_string());
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify placeholder was restored
let (_, pr_count) = purgatory2.count();
assert_eq!(pr_count, 1);
let restored_entry = purgatory2.find_pr("placeholder-id").unwrap();
assert!(restored_entry.event.is_none()); // Still a placeholder
assert_eq!(restored_entry.commit, "commit-def");
// Verify it's findable as a placeholder
assert_eq!(
purgatory2.find_pr_placeholder("placeholder-id"),
Some("commit-def".to_string())
);
}
#[tokio::test]
async fn test_save_and_restore_expired_events() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
let event = EventBuilder::text_note("test")
.sign_with_keys(&keys)
.unwrap();
let event_id = event.id;
// Add and expire event
purgatory.add_state(event, "repo".to_string(), keys.public_key());
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);
}
}
purgatory.cleanup();
// Verify event is marked as expired
assert!(purgatory.is_expired(&event_id));
assert_eq!(purgatory.expired_count(), 1);
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify expired event was restored
assert!(purgatory2.is_expired(&event_id));
assert_eq!(purgatory2.expired_count(), 1);
// Verify it's included in event_ids()
let ids = purgatory2.event_ids();
assert!(ids.contains(&event_id));
}
#[tokio::test]
async fn test_save_and_restore_empty_purgatory() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
// Save empty purgatory
purgatory.save_to_disk(&state_file).unwrap();
// Verify file exists
assert!(state_file.exists());
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify purgatory is still empty
let (state_count, pr_count) = purgatory2.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
assert_eq!(purgatory2.expired_count(), 0);
}
#[tokio::test]
async fn test_restore_missing_file() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("nonexistent.json");
let purgatory = Purgatory::new(PathBuf::new());
// Attempting to restore from missing file should error
let result = purgatory.restore_from_disk(&state_file);
assert!(result.is_err());
// Purgatory should remain empty
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
#[tokio::test]
async fn test_restore_corrupted_json() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("corrupted.json");
// Write invalid JSON
std::fs::write(&state_file, "{ this is not valid json }").unwrap();
let purgatory = Purgatory::new(PathBuf::new());
// Attempting to restore corrupted file should error
let result = purgatory.restore_from_disk(&state_file);
assert!(result.is_err());
// Purgatory should remain empty
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
#[tokio::test]
async fn test_restore_unsupported_version() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("wrong_version.json");
// Write state with unsupported version
let state = r#"{
"version": 999,
"saved_at": {"secs_since_epoch": 1000000000, "nanos_since_epoch": 0},
"state_events": {},
"pr_events": {},
"expired_events": {}
}"#;
std::fs::write(&state_file, state).unwrap();
let purgatory = Purgatory::new(PathBuf::new());
// Attempting to restore unsupported version should error
let result = purgatory.restore_from_disk(&state_file);
assert!(result.is_err());
assert!(result
.unwrap_err()
.to_string()
.contains("Unsupported state version"));
}
#[tokio::test]
async fn test_downtime_calculation() {
use tempfile::tempdir;
use tokio::time::sleep;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add state event
let event = EventBuilder::text_note("test")
.sign_with_keys(&keys)
.unwrap();
purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key());
// Get original expiry time
let original_entries = purgatory.find_state("repo");
let original_entry = &original_entries[0];
let original_expires_at = original_entry.expires_at;
let original_remaining = original_expires_at.saturating_duration_since(Instant::now());
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Simulate downtime (100ms)
sleep(Duration::from_millis(100)).await;
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Get restored expiry time
let restored_entries = purgatory2.find_state("repo");
let restored_entry = &restored_entries[0];
let restored_expires_at = restored_entry.expires_at;
let restored_remaining = restored_expires_at.saturating_duration_since(Instant::now());
// Remaining time should be approximately the same (accounting for downtime)
// Allow 2000ms tolerance for test execution time and sleep duration
let diff = if restored_remaining > original_remaining {
restored_remaining.as_millis() - original_remaining.as_millis()
} else {
original_remaining.as_millis() - restored_remaining.as_millis()
};
assert!(
diff < 2000,
"Downtime calculation should preserve remaining TTL. Original: {}ms, Restored: {}ms, Diff: {}ms",
original_remaining.as_millis(),
restored_remaining.as_millis(),
diff
);
}
#[tokio::test]
async fn test_expiry_times_preserved() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add state event
let event = EventBuilder::text_note("test")
.sign_with_keys(&keys)
.unwrap();
purgatory.add_state(event.clone(), "repo".to_string(), keys.public_key());
// Manually set expiry to a specific time in the future
let custom_expiry = Instant::now() + Duration::from_secs(600); // 10 minutes
if let Some(mut entries) = purgatory.state_events.get_mut("repo") {
for entry in entries.iter_mut() {
entry.expires_at = custom_expiry;
}
}
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Get restored expiry time
let restored_entries = purgatory2.find_state("repo");
let restored_entry = &restored_entries[0];
let restored_remaining = restored_entry
.expires_at
.saturating_duration_since(Instant::now());
// Should be approximately 600 seconds (allow 3 second tolerance for test execution)
assert!(
restored_remaining.as_secs() >= 597 && restored_remaining.as_secs() <= 603,
"Expected ~600s remaining, got {}s",
restored_remaining.as_secs()
);
}
#[tokio::test]
async fn test_multiple_state_events_same_identifier() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys1 = Keys::generate();
let keys2 = Keys::generate();
let keys3 = Keys::generate();
// Add multiple state events for the same identifier from different authors
let event1 = EventBuilder::text_note("maintainer 1")
.sign_with_keys(&keys1)
.unwrap();
let event2 = EventBuilder::text_note("maintainer 2")
.sign_with_keys(&keys2)
.unwrap();
let event3 = EventBuilder::text_note("maintainer 3")
.sign_with_keys(&keys3)
.unwrap();
purgatory.add_state(
event1.clone(),
"shared-repo".to_string(),
keys1.public_key(),
);
purgatory.add_state(
event2.clone(),
"shared-repo".to_string(),
keys2.public_key(),
);
purgatory.add_state(
event3.clone(),
"shared-repo".to_string(),
keys3.public_key(),
);
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify all three events were restored
let restored_entries = purgatory2.find_state("shared-repo");
assert_eq!(restored_entries.len(), 3);
// Verify all authors are present
let authors: Vec<PublicKey> = restored_entries.iter().map(|e| e.author).collect();
assert!(authors.contains(&keys1.public_key()));
assert!(authors.contains(&keys2.public_key()));
assert!(authors.contains(&keys3.public_key()));
}
#[tokio::test]
async fn test_mixed_pr_events_and_placeholders() {
use nostr_sdk::{Kind, Tag, TagKind};
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add PR event with actual event
let tags = vec![Tag::custom(
TagKind::Custom("a".into()),
vec!["30617:abc123:test-repo".to_string()],
)];
let pr_event = EventBuilder::new(Kind::from(1618), "PR content")
.tags(tags)
.sign_with_keys(&keys)
.unwrap();
purgatory.add_pr(
pr_event.clone(),
"pr-with-event".to_string(),
"commit-abc".to_string(),
);
// Add PR placeholder
purgatory.add_pr_placeholder("pr-placeholder".to_string(), "commit-def".to_string());
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify both were restored correctly
let (_, pr_count) = purgatory2.count();
assert_eq!(pr_count, 2);
// Verify PR event
let pr_entry = purgatory2.find_pr("pr-with-event").unwrap();
assert!(pr_entry.event.is_some());
assert_eq!(pr_entry.commit, "commit-abc");
// Verify placeholder
let placeholder_entry = purgatory2.find_pr("pr-placeholder").unwrap();
assert!(placeholder_entry.event.is_none());
assert_eq!(placeholder_entry.commit, "commit-def");
assert_eq!(
purgatory2.find_pr_placeholder("pr-placeholder"),
Some("commit-def".to_string())
);
}
#[tokio::test]
async fn test_file_cleanup_after_successful_restore() {
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys = Keys::generate();
// Add some data
let event = EventBuilder::text_note("test")
.sign_with_keys(&keys)
.unwrap();
purgatory.add_state(event, "repo".to_string(), keys.public_key());
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
assert!(state_file.exists());
// Restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// File should be deleted after successful restore
assert!(!state_file.exists());
}
#[tokio::test]
async fn test_comprehensive_roundtrip() {
use nostr_sdk::{Kind, Tag, TagKind};
use tempfile::tempdir;
let temp_dir = tempdir().unwrap();
let state_file = temp_dir.path().join("purgatory_state.json");
let purgatory = Purgatory::new(PathBuf::new());
let keys1 = Keys::generate();
let keys2 = Keys::generate();
// Add multiple state events
let state1 = EventBuilder::text_note("state 1")
.sign_with_keys(&keys1)
.unwrap();
let state2 = EventBuilder::text_note("state 2")
.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());
// Add PR event
let tags = vec![Tag::custom(
TagKind::Custom("a".into()),
vec!["30617:abc123:repo1".to_string()],
)];
let pr_event = EventBuilder::new(Kind::from(1618), "PR")
.tags(tags)
.sign_with_keys(&keys1)
.unwrap();
purgatory.add_pr(pr_event.clone(), "pr-1".to_string(), "commit-1".to_string());
// Add PR placeholder
purgatory.add_pr_placeholder("pr-2".to_string(), "commit-2".to_string());
// Add and expire an event
let expired_event = EventBuilder::text_note("expired")
.sign_with_keys(&keys1)
.unwrap();
let expired_id = expired_event.id;
purgatory.add_state(expired_event, "repo3".to_string(), keys1.public_key());
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);
}
}
purgatory.cleanup();
// Verify initial state
let (state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 2); // state1, state2 (expired_event was cleaned up)
assert_eq!(pr_count, 2); // pr-1, pr-2
assert_eq!(purgatory.expired_count(), 1); // expired_event
// Save to disk
purgatory.save_to_disk(&state_file).unwrap();
// Create new purgatory and restore
let purgatory2 = Purgatory::new(PathBuf::new());
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify all data was restored correctly
let (state_count2, pr_count2) = purgatory2.count();
assert_eq!(state_count2, 2);
assert_eq!(pr_count2, 2);
assert_eq!(purgatory2.expired_count(), 1);
// Verify state events
assert_eq!(purgatory2.find_state("repo1").len(), 1);
assert_eq!(purgatory2.find_state("repo2").len(), 1);
// Verify PR events
assert!(purgatory2.find_pr("pr-1").unwrap().event.is_some());
assert!(purgatory2.find_pr("pr-2").unwrap().event.is_none());
// Verify expired event
assert!(purgatory2.is_expired(&expired_id));
}
+198
View File
@@ -0,0 +1,198 @@
//! Persistence utilities for purgatory state.
//!
//! This module provides conversion functions between `Instant` (which cannot be
//! serialized) and `Duration` offsets from a reference `SystemTime`. This allows
//! purgatory state to be persisted to disk and restored across restarts.
//!
//! ## Time Handling
//!
//! - `Instant` is monotonic but cannot be serialized
//! - `SystemTime` can be serialized but may go backwards (NTP, user changes)
//! - We use `SystemTime` for persistence and convert to/from `Instant` at runtime
//! - Downtime is accounted for when restoring state (elapsed time is preserved)
use std::time::{Duration, Instant, SystemTime};
/// Convert an `Instant` to a `Duration` offset from a reference `SystemTime`.
///
/// This allows storing an `Instant` as a serializable offset that can be
/// restored later, accounting for system downtime.
///
/// # Arguments
/// * `instant` - The `Instant` to convert
/// * `reference_time` - The reference `SystemTime` (typically SystemTime::now())
/// * `reference_instant` - The corresponding `Instant` (typically Instant::now())
///
/// # Returns
/// Duration offset from the reference time
///
/// # Example
/// ```
/// use std::time::{Duration, Instant, SystemTime};
/// use ngit_grasp::purgatory::persistence::instant_to_offset;
///
/// let now_system = SystemTime::now();
/// let now_instant = Instant::now();
/// let future = now_instant + Duration::from_secs(60);
///
/// let offset = instant_to_offset(future, now_system, now_instant);
/// assert!(offset.as_secs() >= 60);
/// ```
pub fn instant_to_offset(
instant: Instant,
_reference_time: SystemTime,
reference_instant: Instant,
) -> Duration {
if instant >= reference_instant {
// Future instant - return positive offset
instant.duration_since(reference_instant)
} else {
// Past instant - this shouldn't happen in normal operation,
// but we handle it by returning zero duration
Duration::ZERO
}
}
/// Convert a `Duration` offset back to an `Instant`, accounting for downtime.
///
/// This restores an `Instant` from a serialized offset, adjusting for the time
/// that has elapsed since the state was saved.
///
/// # Arguments
/// * `offset` - The duration offset from the saved reference time
/// * `saved_at` - The `SystemTime` when the state was saved
/// * `reference_instant` - The current `Instant` (typically Instant::now())
///
/// # Returns
/// The restored `Instant`, adjusted for downtime
///
/// # Example
/// ```
/// use std::time::{Duration, Instant, SystemTime};
/// use ngit_grasp::purgatory::persistence::offset_to_instant;
///
/// let saved_at = SystemTime::now();
/// let offset = Duration::from_secs(60);
/// let now_instant = Instant::now();
///
/// let restored = offset_to_instant(offset, saved_at, now_instant);
/// // restored will be approximately now_instant + 60 seconds
/// ```
pub fn offset_to_instant(
offset: Duration,
saved_at: SystemTime,
reference_instant: Instant,
) -> Instant {
// Calculate how much time has elapsed since the state was saved
let now_system = SystemTime::now();
let elapsed_since_save = now_system
.duration_since(saved_at)
.unwrap_or(Duration::ZERO);
// The original deadline was: saved_at + offset
// Time remaining = (saved_at + offset) - now_system
// = offset - elapsed_since_save
if offset > elapsed_since_save {
// Deadline is still in the future
let remaining = offset - elapsed_since_save;
reference_instant + remaining
} else {
// Deadline has already passed or is right now
reference_instant
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread;
use std::time::Duration;
#[test]
fn test_instant_to_offset_future() {
let now_system = SystemTime::now();
let now_instant = Instant::now();
let future = now_instant + Duration::from_secs(60);
let offset = instant_to_offset(future, now_system, now_instant);
// Should be approximately 60 seconds (within tolerance)
assert!(offset.as_secs() >= 59 && offset.as_secs() <= 61);
}
#[test]
fn test_instant_to_offset_past() {
let now_system = SystemTime::now();
let past_instant = Instant::now();
// Simulate some time passing
thread::sleep(Duration::from_millis(10));
let now_instant = Instant::now();
let offset = instant_to_offset(past_instant, now_system, now_instant);
// Past instants return zero duration
assert_eq!(offset, Duration::ZERO);
}
#[test]
fn test_offset_to_instant_with_time_remaining() {
let saved_at = SystemTime::now();
let offset = Duration::from_secs(60);
// Simulate a very short downtime (< 10ms)
thread::sleep(Duration::from_millis(5));
let now_instant = Instant::now();
let restored = offset_to_instant(offset, saved_at, now_instant);
// Should be approximately 60 seconds in the future
let remaining = restored.duration_since(now_instant);
assert!(
remaining.as_secs() >= 59 && remaining.as_secs() <= 61,
"Expected ~60s, got {}s",
remaining.as_secs()
);
}
#[test]
fn test_offset_to_instant_deadline_passed() {
// Simulate state saved 70 seconds ago with 60 second offset
let saved_at = SystemTime::now() - Duration::from_secs(70);
let offset = Duration::from_secs(60);
let now_instant = Instant::now();
let restored = offset_to_instant(offset, saved_at, now_instant);
// Deadline has passed, should be now or in the past
let remaining = restored.saturating_duration_since(now_instant);
assert_eq!(remaining, Duration::ZERO);
}
#[test]
fn test_round_trip_conversion() {
let now_system = SystemTime::now();
let now_instant = Instant::now();
let future = now_instant + Duration::from_secs(120);
// Convert to offset
let offset = instant_to_offset(future, now_system, now_instant);
// Immediately convert back (minimal downtime)
let restored = offset_to_instant(offset, now_system, now_instant);
// Should be very close to the original future instant
let diff = if restored > future {
restored.duration_since(future)
} else {
future.duration_since(restored)
};
// Allow for small timing differences (< 100ms)
assert!(
diff < Duration::from_millis(100),
"Round trip should preserve instant within 100ms, got {}ms",
diff.as_millis()
);
}
}
+18 -2
View File
@@ -5,8 +5,14 @@
//! problem where either the nostr event or git push can arrive first.
use nostr_sdk::prelude::*;
use serde::{Deserialize, Serialize};
use std::time::Instant;
/// Default value for Instant fields during deserialization
fn instant_now() -> Instant {
Instant::now()
}
/// A reference name and its target object.
///
/// Used to identify specific git refs (branches, tags) that a state event
@@ -59,7 +65,10 @@ impl RefUpdate {
/// State events declare the current state of a repository but may arrive
/// before the corresponding git data has been pushed. This entry holds
/// the event and associated metadata until the git data arrives.
#[derive(Debug, Clone)]
///
/// Note: `Instant` fields cannot be serialized directly. Use the `persistence`
/// module to convert to/from serializable wrapper types.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StatePurgatoryEntry {
/// The nostr state event (kind 30618) awaiting git data
pub event: Event,
@@ -71,9 +80,11 @@ pub struct StatePurgatoryEntry {
pub author: PublicKey,
/// When this entry was added to purgatory
#[serde(skip, default = "instant_now")]
pub created_at: Instant,
/// Expiry deadline (30 min from creation, may be extended)
#[serde(skip, default = "instant_now")]
pub expires_at: Instant,
}
@@ -82,7 +93,10 @@ pub struct StatePurgatoryEntry {
/// PR events reference specific commits but may arrive before the git push
/// containing those commits. Alternatively, a git push may arrive first,
/// creating a placeholder entry waiting for the corresponding PR event.
#[derive(Debug, Clone)]
///
/// Note: `Instant` fields cannot be serialized directly. Use the `persistence`
/// module to convert to/from serializable wrapper types.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PrPurgatoryEntry {
/// The nostr PR event, if received (None = git data arrived first)
pub event: Option<Event>,
@@ -92,8 +106,10 @@ pub struct PrPurgatoryEntry {
pub commit: String,
/// When this entry was added to purgatory
#[serde(skip, default = "instant_now")]
pub created_at: Instant,
/// Expiry deadline (30 min from creation, may be extended)
#[serde(skip, default = "instant_now")]
pub expires_at: Instant,
}
+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:
+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);
}
}
+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);
}