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

chore: cargo fmt

Browse Source
This commit is contained in:
DanConwayDev
2026-01-09 19:58:41 +00:00
parent 6dd9fcd539
commit b28a356cb4
11 changed files with 462 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.rs
+21 -12
View File
@@ -141,12 +141,20 @@ pub struct Config {
/// Stores full event objects for immediate re-processing when dependencies resolve.
/// Too short (<30s): Miss events from slow relays
/// Too long (>5min): Waste memory
#[arg(long, env = "NGIT_REJECTED_HOT_CACHE_DURATION_SECS", default_value_t = 120)]
#[arg(
long,
env = "NGIT_REJECTED_HOT_CACHE_DURATION_SECS",
default_value_t = 120
)]
pub rejected_hot_cache_duration_secs: u64,
/// Cold index expiry in seconds for rejected announcements (default: 604800 = 7 days)
/// Stores metadata only to prevent repeated downloads of rejected events.
#[arg(long, env = "NGIT_REJECTED_COLD_INDEX_EXPIRY_SECS", default_value_t = 604800)]
#[arg(
long,
env = "NGIT_REJECTED_COLD_INDEX_EXPIRY_SECS",
default_value_t = 604800
)]
pub rejected_cold_index_expiry_secs: u64,
}
@@ -190,10 +198,7 @@ impl Config {
// Validate it's a valid nsec
Keys::parse(&nsec).context("Invalid nsec in relay owner key file")?;
tracing::info!(
"Loaded relay owner key from {}",
key_path.display()
);
tracing::info!("Loaded relay owner key from {}", key_path.display());
return Ok(nsec);
}
@@ -202,8 +207,7 @@ impl Config {
let nsec = keys.secret_key().to_bech32()?;
// Save to file
fs::write(&key_path, &nsec)
.context("Failed to write relay owner key file")?;
fs::write(&key_path, &nsec).context("Failed to write relay owner key file")?;
tracing::info!(
"Generated new relay owner key and saved to {}",
@@ -215,7 +219,9 @@ impl Config {
/// Get the relay owner's Keys object
pub fn relay_owner_keys(&self) -> Result<Keys> {
let nsec = self.relay_owner_nsec.as_ref()
let nsec = self
.relay_owner_nsec
.as_ref()
.context("relay_owner_nsec not set (should be set by Config::load())")?;
Keys::parse(nsec).context("Invalid relay_owner_nsec")
}
@@ -251,8 +257,11 @@ impl Config {
pub fn for_testing() -> Self {
// Generate a test key deterministically for consistent tests
let keys = Keys::generate();
let nsec = keys.secret_key().to_bech32().expect("Failed to generate test nsec");
let nsec = keys
.secret_key()
.to_bech32()
.expect("Failed to generate test nsec");
Self {
domain: "localhost:8080".to_string(),
relay_owner_nsec: Some(nsec),
@@ -348,7 +357,7 @@ mod tests {
let config = Config::for_testing();
let keys = config.relay_owner_keys().expect("Should have valid keys");
let npub = config.relay_owner_npub().expect("Should derive npub");
// Verify the npub matches the keys
assert_eq!(npub, keys.public_key().to_bech32().unwrap());
assert!(npub.starts_with("npub1"));
src/http/nip11.rs
+2 -2
View File
@@ -102,12 +102,12 @@ mod tests {
assert_eq!(doc.name, "Test Relay");
assert_eq!(doc.description, "A test relay");
// Verify pubkey is present and is a valid npub
assert!(doc.pubkey.is_some());
let pubkey = doc.pubkey.unwrap();
assert!(pubkey.starts_with("npub1"));
assert!(doc.supported_nips.contains(&1));
assert!(doc.supported_nips.contains(&11));
assert!(doc.supported_nips.contains(&34));
src/nostr/builder.rs
+12 -8
View File
@@ -102,11 +102,11 @@ impl Nip34WritePolicy {
}
tracing::debug!("Accepted repository announcement: {}", event_id_str);
// Check purgatory for state events that might now be authorized
self.check_purgatory_state_events_for_identifier(&announcement.identifier)
.await;
WritePolicyResult::Accept
}
Err(e) => {
@@ -130,11 +130,11 @@ impl Nip34WritePolicy {
announcement.identifier
);
// Don't create bare repository for external announcements
// Check purgatory for state events that might now be authorized
self.check_purgatory_state_events_for_identifier(&announcement.identifier)
.await;
WritePolicyResult::Accept
}
Err(e) => {
@@ -324,20 +324,24 @@ impl Nip34WritePolicy {
/// 4. Keeps unauthorized events in purgatory (will expire naturally)
async fn check_purgatory_state_events_for_identifier(&self, identifier: &str) {
let state_events = self.ctx.purgatory.find_state(identifier);
if state_events.is_empty() {
return;
}
tracing::debug!(
identifier = %identifier,
count = state_events.len(),
"Checking purgatory state events after announcement acceptance"
);
for entry in state_events {
// Re-evaluate authorization with the new announcement
match self.state_policy.process_state_event(&entry.event, false).await {
match self
.state_policy
.process_state_event(&entry.event, false)
.await
{
Ok(WritePolicyResult::Accept) => {
tracing::info!(
event_id = %entry.event.id,
src/nostr/policy/state.rs
+5 -3
View File
@@ -93,9 +93,11 @@ impl StatePolicy {
});
}
let authorized_owners =
crate::git::authorization::pubkey_authorised_for_repo_owners(&event.pubkey, &db_repo_data);
let authorized_owners = crate::git::authorization::pubkey_authorised_for_repo_owners(
&event.pubkey,
&db_repo_data,
);
if authorized_owners.is_empty() {
tracing::warn!(
event_id = %event.id,
src/sync/metrics.rs
+26 -11
View File
@@ -156,19 +156,25 @@ impl SyncMetrics {
"ngit_sync_rejected_announcements_hot_cache_hits_total",
"Total hot cache hits (events re-processed from cache)",
))?;
registry.register(Box::new(rejected_announcements_hot_cache_hits_total.clone()))?;
registry.register(Box::new(
rejected_announcements_hot_cache_hits_total.clone(),
))?;
let rejected_announcements_hot_cache_misses_total = IntCounter::with_opts(Opts::new(
"ngit_sync_rejected_announcements_hot_cache_misses_total",
"Total hot cache misses (events not in cache when invalidated)",
))?;
registry.register(Box::new(rejected_announcements_hot_cache_misses_total.clone()))?;
registry.register(Box::new(
rejected_announcements_hot_cache_misses_total.clone(),
))?;
let rejected_announcements_hot_cache_expired_total = IntCounter::with_opts(Opts::new(
"ngit_sync_rejected_announcements_hot_cache_expired_total",
"Total expired entries removed from hot cache",
))?;
registry.register(Box::new(rejected_announcements_hot_cache_expired_total.clone()))?;
registry.register(Box::new(
rejected_announcements_hot_cache_expired_total.clone(),
))?;
let rejected_announcements_cold_index_current = IntGauge::with_opts(Opts::new(
"ngit_sync_rejected_announcements_cold_index_current",
@@ -180,7 +186,9 @@ impl SyncMetrics {
"ngit_sync_rejected_announcements_cold_index_expired_total",
"Total expired entries removed from cold index",
))?;
registry.register(Box::new(rejected_announcements_cold_index_expired_total.clone()))?;
registry.register(Box::new(
rejected_announcements_cold_index_expired_total.clone(),
))?;
let rejected_announcements_invalidated_total = IntCounter::with_opts(Opts::new(
"ngit_sync_rejected_announcements_invalidated_total",
@@ -430,7 +438,8 @@ impl SyncMetrics {
/// Update hot cache current size gauge.
pub fn update_hot_cache_size(&self, size: usize) {
self.rejected_announcements_hot_cache_current.set(size as i64);
self.rejected_announcements_hot_cache_current
.set(size as i64);
}
/// Record hot cache hit (event re-processed from cache).
@@ -445,22 +454,26 @@ impl SyncMetrics {
/// Record hot cache expired entries.
pub fn record_hot_cache_expired(&self, count: usize) {
self.rejected_announcements_hot_cache_expired_total.inc_by(count as u64);
self.rejected_announcements_hot_cache_expired_total
.inc_by(count as u64);
}
/// Update cold index current size gauge.
pub fn update_cold_index_size(&self, size: usize) {
self.rejected_announcements_cold_index_current.set(size as i64);
self.rejected_announcements_cold_index_current
.set(size as i64);
}
/// Record cold index expired entries.
pub fn record_cold_index_expired(&self, count: usize) {
self.rejected_announcements_cold_index_expired_total.inc_by(count as u64);
self.rejected_announcements_cold_index_expired_total
.inc_by(count as u64);
}
/// Record invalidation (maintainer announcement invalidated).
pub fn record_invalidation(&self, count: usize) {
self.rejected_announcements_invalidated_total.inc_by(count as u64);
self.rejected_announcements_invalidated_total
.inc_by(count as u64);
}
// === Rejected States Recording Methods ===
@@ -482,7 +495,8 @@ impl SyncMetrics {
/// Record state event hot cache expired entries.
pub fn record_states_hot_cache_expired(&self, count: usize) {
self.rejected_states_hot_cache_expired_total.inc_by(count as u64);
self.rejected_states_hot_cache_expired_total
.inc_by(count as u64);
}
/// Update state events cold index current size gauge.
@@ -492,7 +506,8 @@ impl SyncMetrics {
/// Record state event cold index expired entries.
pub fn record_states_cold_index_expired(&self, count: usize) {
self.rejected_states_cold_index_expired_total.inc_by(count as u64);
self.rejected_states_cold_index_expired_total
.inc_by(count as u64);
}
/// Record state event invalidation.
src/sync/mod.rs
+91 -94
View File
@@ -27,7 +27,7 @@ pub use algorithms::{AddFilters, RelaySyncNeeds};
pub use metrics::SyncMetrics;
// Re-export rejected index types
pub use rejected_index::{RejectionReason};
pub use rejected_index::RejectionReason;
// Note: RejectedEventsIndex struct exists in rejected_index.rs but not yet used
// Current code still uses the simple HashSet type alias below
@@ -73,7 +73,7 @@ pub type PendingSyncIndex = Arc<RwLock<HashMap<String, Vec<PendingBatch>>>>;
/// Tracks EventIds of announcement events (30617/30618) that were rejected during sync.
/// These events are excluded from negentropy sync and skipped during REQ+EOSE processing
/// to avoid repeatedly fetching and rejecting the same events.
///
///
/// Uses the two-tier RejectedEventsIndex from rejected_index.rs:
/// - Hot cache: Full events for 2 minutes (enables immediate re-processing)
/// - Cold index: Metadata for 7 days (prevents repeated downloads)
@@ -113,7 +113,9 @@ impl ConnectionStatus {
pub fn is_live_sync_active(&self) -> bool {
matches!(
self,
ConnectionStatus::Syncing | ConnectionStatus::Connected | ConnectionStatus::ConnectedHistoricSyncFailures
ConnectionStatus::Syncing
| ConnectionStatus::Connected
| ConnectionStatus::ConnectedHistoricSyncFailures
)
}
}
@@ -384,9 +386,7 @@ async fn run_rejected_index_cleanup(
let hot_cache_interval = Duration::from_secs(60);
let cold_index_interval = Duration::from_secs(86400); // 24 hours
tracing::info!(
"Rejected index cleanup started (hot cache: 60s, cold index: daily)"
);
tracing::info!("Rejected index cleanup started (hot cache: 60s, cold index: daily)");
let mut hot_cache_timer = tokio::time::interval(hot_cache_interval);
let mut cold_index_timer = tokio::time::interval(cold_index_interval);
@@ -399,7 +399,7 @@ async fn run_rejected_index_cleanup(
tokio::select! {
_ = hot_cache_timer.tick() => {
let manager = sync_manager.lock().await;
// Clean up announcements index
let (hot_expired, _) = manager.rejected_events_index.cleanup_expired();
if hot_expired > 0 {
@@ -408,7 +408,7 @@ async fn run_rejected_index_cleanup(
hot_expired
);
}
// Clean up states index
let (states_hot_expired, _) = manager.rejected_states_index.cleanup_states_expired();
if states_hot_expired > 0 {
@@ -420,7 +420,7 @@ async fn run_rejected_index_cleanup(
}
_ = cold_index_timer.tick() => {
let manager = sync_manager.lock().await;
// Clean up announcements index
let (_, cold_expired) = manager.rejected_events_index.cleanup_expired();
if cold_expired > 0 {
@@ -429,7 +429,7 @@ async fn run_rejected_index_cleanup(
cold_expired
);
}
// Clean up states index
let (_, states_cold_expired) = manager.rejected_states_index.cleanup_states_expired();
if states_cold_expired > 0 {
@@ -799,8 +799,7 @@ impl SyncManager {
if let (Some(requested), Some(received)) =
(&batch.requested_event_ids, &batch.received_event_ids)
{
let missing: Vec<EventId> =
requested.difference(received).cloned().collect();
let missing: Vec<EventId> = requested.difference(received).cloned().collect();
if !missing.is_empty() {
let requested_count = requested.len();
@@ -884,13 +883,11 @@ impl SyncManager {
// Re-acquire lock and update batch with new subscriptions
let mut pending = self.pending_sync_index.write().await;
if let Some(batches) = pending.get_mut(&relay_url_for_retry) {
if let Some(batch) =
batches.iter_mut().find(|b| b.batch_id == batch_id)
if let Some(batch) = batches.iter_mut().find(|b| b.batch_id == batch_id)
{
batch.outstanding_subs.extend(new_sub_ids.clone());
// Update requested_event_ids to only include missing ones
batch.requested_event_ids =
Some(missing.iter().cloned().collect());
batch.requested_event_ids = Some(missing.iter().cloned().collect());
// Clear received_event_ids for fresh tracking
batch.received_event_ids = Some(HashSet::new());
// Increment retry counter
@@ -921,14 +918,14 @@ impl SyncManager {
// Re-acquire lock to extract the batch
let mut pending = self.pending_sync_index.write().await;
if let Some(batches) = pending.get_mut(&relay_url_for_retry) {
if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id)
{
if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id) {
let completed_batch = batches.remove(idx);
if batches.is_empty() {
pending.remove(&relay_url_for_retry);
}
drop(pending);
self.confirm_batch(&relay_url_for_retry, completed_batch).await;
self.confirm_batch(&relay_url_for_retry, completed_batch)
.await;
}
}
return;
@@ -1023,17 +1020,17 @@ impl SyncManager {
"Batch completed but no RelayState found for relay"
);
}
// Release lock before checking if historic sync is complete
drop(relay_index);
// Spawn background task to check if historic sync is complete
// This avoids blocking the confirm_batch flow for 6 seconds
let relay_url = relay_url.to_string();
let pending_index = self.pending_sync_index.clone();
let relay_index = self.relay_sync_index.clone();
let metrics = self.metrics.clone();
tokio::spawn(async move {
Self::check_and_complete_historic_sync_impl(
&relay_url,
@@ -1073,29 +1070,33 @@ impl SyncManager {
// First check: Are there any pending batches?
let has_pending = {
let pending = pending_index.read().await;
pending.get(relay_url).is_some_and(|batches| !batches.is_empty())
pending
.get(relay_url)
.is_some_and(|batches| !batches.is_empty())
};
if has_pending {
// Still syncing, don't transition yet
return;
}
// Wait for self-subscriber batch window + buffer to catch any in-flight events
// that might create new Layer 2/3 filters
tokio::time::sleep(Duration::from_millis(6000)).await;
// Second check: Are there still no pending batches?
let has_pending = {
let pending = pending_index.read().await;
pending.get(relay_url).is_some_and(|batches| !batches.is_empty())
pending
.get(relay_url)
.is_some_and(|batches| !batches.is_empty())
};
if has_pending {
// New batches appeared during the wait - still syncing
return;
}
// No pending batches after waiting - safe to transition to Connected or ConnectedDegraded
let mut relay_index_guard = relay_index.write().await;
if let Some(state) = relay_index_guard.get_mut(relay_url) {
@@ -1106,11 +1107,11 @@ impl SyncManager {
} else {
ConnectionStatus::Connected
};
state.connection_status = new_status;
state.historic_sync_completed = true;
state.historic_sync_completed_at = Some(Timestamp::now());
tracing::info!(
relay = %relay_url,
repos_synced = state.repos.len(),
@@ -1120,7 +1121,7 @@ impl SyncManager {
"Historic sync complete - transitioned to {} status",
if state.historic_sync_had_failures { "ConnectedHistoricSyncFailures" } else { "Connected" }
);
// Update metrics
if let Some(ref metrics) = metrics {
metrics.record_connection_status(relay_url, new_status);
@@ -1362,8 +1363,8 @@ impl SyncManager {
);
return;
}
Some(ConnectionStatus::Syncing)
| Some(ConnectionStatus::Connected)
Some(ConnectionStatus::Syncing)
| Some(ConnectionStatus::Connected)
| Some(ConnectionStatus::ConnectedHistoricSyncFailures) => {
// Continue to subscribe - live sync is active, can accept new filters
}
@@ -1468,7 +1469,8 @@ impl SyncManager {
match relay_event {
RelayEvent::Event(event, subscription_id) => {
// Skip events we've already rejected (announcements only)
if (event.kind == Kind::GitRepoAnnouncement || event.kind == Kind::RepoState)
if (event.kind == Kind::GitRepoAnnouncement
|| event.kind == Kind::RepoState)
&& rejected_events_index.contains(&event.id)
{
tracing::trace!(
@@ -1479,7 +1481,7 @@ impl SyncManager {
);
continue;
}
let result = Self::process_event_static(
&event,
&relay_url_clone,
@@ -1863,11 +1865,16 @@ impl SyncManager {
// Create RelayConnection if not exists
if !self.connections.contains_key(&relay_url) {
// Get relay owner keys for NIP-42 authentication
let keys = self.config.relay_owner_keys()
let keys = self
.config
.relay_owner_keys()
.expect("relay_owner_keys should be available");
let connection =
RelayConnection::new_with_database(relay_url.clone(), Arc::clone(&self.database), keys);
let connection = RelayConnection::new_with_database(
relay_url.clone(),
Arc::clone(&self.database),
keys,
);
self.connections.insert(relay_url.clone(), connection);
tracing::debug!(relay = %relay_url, "Registered new relay connection");
}
@@ -1919,7 +1926,7 @@ impl SyncManager {
state.connection_status = ConnectionStatus::Connecting;
}
}
// Update metrics to show connecting status
if let Some(ref metrics) = self.metrics {
metrics.record_connection_status(relay_url, ConnectionStatus::Connecting);
@@ -1974,7 +1981,8 @@ impl SyncManager {
if let Some(ref metrics) = self.metrics {
metrics.record_connection_attempt(relay_url, false);
metrics.record_connection_status(relay_url, ConnectionStatus::Disconnected);
metrics.record_health_state(relay_url, self.health_tracker.get_state(relay_url));
metrics
.record_health_state(relay_url, self.health_tracker.get_state(relay_url));
}
}
}
@@ -2172,7 +2180,7 @@ impl SyncManager {
// immediately and should now pass validation.
if event.kind == Kind::GitRepoAnnouncement {
use crate::nostr::events::RepositoryAnnouncement;
match RepositoryAnnouncement::from_event(event.clone()) {
Ok(announcement) => {
if !announcement.maintainers.is_empty() {
@@ -2219,15 +2227,16 @@ impl SyncManager {
// 2. Second attempt uses maintainer exception (different code path)
// 3. If second attempt fails, stays in cold index only (no third attempt)
// Use Box::pin to avoid infinitely sized future
let reprocess_result = Box::pin(Self::process_event_static(
&maintainer_event,
relay_url,
database,
write_policy,
local_relay,
rejected_events_index,
))
.await;
let reprocess_result =
Box::pin(Self::process_event_static(
&maintainer_event,
relay_url,
database,
write_policy,
local_relay,
rejected_events_index,
))
.await;
match reprocess_result {
ProcessResult::Saved => {
@@ -2275,7 +2284,7 @@ impl SyncManager {
);
}
}
// When a repository announcement is accepted, re-process any state events
// that were previously rejected because no announcement existed.
// This handles the race condition where state events arrive before their
@@ -2284,7 +2293,10 @@ impl SyncManager {
Ok(announcement) => {
// Get the announcement author's state events that were rejected
let (removed, hot_events) = rejected_events_index
.invalidate_and_get_state_events(&event.pubkey, &announcement.identifier);
.invalidate_and_get_state_events(
&event.pubkey,
&announcement.identifier,
);
if removed > 0 {
tracing::info!(
@@ -2357,7 +2369,7 @@ impl SyncManager {
}
}
}
// When a state event is accepted (git data arrived), re-process any other
// rejected state events for the same repository. This handles the case where
// multiple state events arrive but only one has git data initially.
@@ -2454,7 +2466,7 @@ impl SyncManager {
reason = %message,
"Event rejected by write policy"
);
// Track rejected announcement and state events to avoid re-fetching them
if event.kind == Kind::GitRepoAnnouncement || event.kind == Kind::RepoState {
// Extract identifier from 'd' tag
@@ -2465,12 +2477,14 @@ impl SyncManager {
.and_then(|t| t.content())
{
// Determine rejection reason based on message
let reason = if message.contains("doesn't list this service")
|| message.contains("Announcement must list service") {
let reason = if message.contains("doesn't list this service")
|| message.contains("Announcement must list service")
{
rejected_index::RejectionReason::DoesNotListService
} else if message.contains("maintainer")
} else if message.contains("maintainer")
|| message.contains("no announcement exists")
|| message.contains("not authorized") {
|| message.contains("not authorized")
{
rejected_index::RejectionReason::MaintainerNotYetValid
} else {
rejected_index::RejectionReason::Other
@@ -2512,7 +2526,7 @@ impl SyncManager {
);
}
}
ProcessResult::Rejected
}
}
@@ -3042,7 +3056,8 @@ impl SyncManager {
// Get event IDs to exclude: purgatory + rejected announcements
let purgatory_ids = self.purgatory.event_ids();
let rejected_ids = self.rejected_events_index.get_all_event_ids();
let excluded_ids: HashSet<EventId> = purgatory_ids.union(&rejected_ids).cloned().collect();
let excluded_ids: HashSet<EventId> =
purgatory_ids.union(&rejected_ids).cloned().collect();
for (idx, result) in diff_results {
match result {
@@ -3166,8 +3181,7 @@ impl SyncManager {
if let Some(batch) = relay_batches.iter_mut().find(|b| b.batch_id == batch_id) {
batch.outstanding_subs.extend(subscription_ids.clone());
// Store requested event IDs for validation after EOSE
batch.requested_event_ids =
Some(all_remote_ids.iter().cloned().collect());
batch.requested_event_ids = Some(all_remote_ids.iter().cloned().collect());
batch.received_event_ids = Some(HashSet::new());
}
}
@@ -3359,20 +3373,16 @@ mod tests {
async fn test_rejected_events_excluded_from_negentropy() {
// Create indices
let purgatory_ids: HashSet<EventId> = HashSet::new();
let rejected_index = RejectedEventsIndex::new(
Duration::from_secs(120),
Duration::from_secs(604800),
);
let rejected_index =
RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
// Create test event IDs
let _rejected_id = EventId::from_hex(
"0000000000000000000000000000000000000000000000000000000000000001",
)
.unwrap();
let valid_id = EventId::from_hex(
"0000000000000000000000000000000000000000000000000000000000000002",
)
.unwrap();
let _rejected_id =
EventId::from_hex("0000000000000000000000000000000000000000000000000000000000000001")
.unwrap();
let valid_id =
EventId::from_hex("0000000000000000000000000000000000000000000000000000000000000002")
.unwrap();
// Add rejected event to index
let keys = Keys::generate();
@@ -3383,7 +3393,7 @@ mod tests {
))
.sign_with_keys(&keys)
.unwrap();
// Override the event ID for testing (we need a specific ID)
// Since we can't override the ID, let's use the actual event ID
let rejected_id = rejected_event.id;
@@ -3403,8 +3413,7 @@ mod tests {
remote_ids.insert(valid_id);
// Exclude rejected and purgatory events
let excluded_ids: HashSet<EventId> =
purgatory_ids.union(&rejected_ids).cloned().collect();
let excluded_ids: HashSet<EventId> = purgatory_ids.union(&rejected_ids).cloned().collect();
let filtered_ids: HashSet<EventId> =
remote_ids.difference(&excluded_ids).cloned().collect();
@@ -3422,22 +3431,14 @@ mod tests {
// Requested 5 events from negentropy diff
let mut requested: HashSet<EventId> = HashSet::new();
for i in 1u8..=5 {
let id = EventId::from_hex(&format!(
"{:0>64}",
format!("{:x}", i)
))
.unwrap();
let id = EventId::from_hex(&format!("{:0>64}", format!("{:x}", i))).unwrap();
requested.insert(id);
}
// Only received 3 events (simulating relay limit)
let mut received: HashSet<EventId> = HashSet::new();
for i in 1u8..=3 {
let id = EventId::from_hex(&format!(
"{:0>64}",
format!("{:x}", i)
))
.unwrap();
let id = EventId::from_hex(&format!("{:0>64}", format!("{:x}", i))).unwrap();
received.insert(id);
}
@@ -3461,11 +3462,7 @@ mod tests {
// Simulate scenario where all requested events are received
let mut requested: HashSet<EventId> = HashSet::new();
for i in 1u8..=3 {
let id = EventId::from_hex(&format!(
"{:0>64}",
format!("{:x}", i)
))
.unwrap();
let id = EventId::from_hex(&format!("{:0>64}", format!("{:x}", i))).unwrap();
requested.insert(id);
}
src/sync/rejected_index.rs
+15 -42
View File
@@ -190,9 +190,7 @@ impl HotCache {
let now = Instant::now();
let initial_count = entries.len();
entries.retain(|_, entry| {
now.duration_since(entry.cached_at) < self.expiry_duration
});
entries.retain(|_, entry| now.duration_since(entry.cached_at) < self.expiry_duration);
initial_count - entries.len()
}
@@ -284,9 +282,7 @@ impl ColdIndex {
let now = Instant::now();
let initial_count = entries.len();
entries.retain(|_, entry| {
now.duration_since(entry.rejected_at) < self.expiry_duration
});
entries.retain(|_, entry| now.duration_since(entry.rejected_at) < self.expiry_duration);
initial_count - entries.len()
}
@@ -389,12 +385,8 @@ impl RejectedEventsIndex {
reason: RejectionReason,
) {
// Add to hot cache (full event)
self.hot_cache.add(
event.clone(),
pubkey,
identifier.clone(),
reason,
);
self.hot_cache
.add(event.clone(), pubkey, identifier.clone(), reason);
// Add to cold index (metadata only)
self.cold_index.add(event.id, pubkey, identifier, reason);
@@ -419,12 +411,8 @@ impl RejectedEventsIndex {
reason: RejectionReason,
) {
// Add to hot cache (full event)
self.hot_cache.add(
event.clone(),
pubkey,
identifier.clone(),
reason,
);
self.hot_cache
.add(event.clone(), pubkey, identifier.clone(), reason);
// Add to cold index (metadata only)
self.cold_index.add(event.id, pubkey, identifier, reason);
@@ -608,8 +596,7 @@ mod tests {
async fn create_test_event() -> Event {
let keys = Keys::generate();
let unsigned = nostr_sdk::EventBuilder::text_note("test")
.build(keys.public_key());
let unsigned = nostr_sdk::EventBuilder::text_note("test").build(keys.public_key());
keys.sign_event(unsigned).await.unwrap()
}
@@ -695,10 +682,7 @@ mod tests {
#[tokio::test]
async fn test_two_tier_index_add_and_contains() {
let index = RejectedEventsIndex::new(
Duration::from_secs(120),
Duration::from_secs(604800),
);
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let event = create_test_event().await;
index.add_announcement(
@@ -715,10 +699,7 @@ mod tests {
#[tokio::test]
async fn test_invalidate_and_get_events() {
let index = RejectedEventsIndex::new(
Duration::from_secs(120),
Duration::from_secs(604800),
);
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();
@@ -773,10 +754,8 @@ mod tests {
#[tokio::test]
async fn test_hot_cache_miss_after_expiry() {
let index = RejectedEventsIndex::new(
Duration::from_millis(50),
Duration::from_secs(604800),
);
let index =
RejectedEventsIndex::new(Duration::from_millis(50), Duration::from_secs(604800));
let event = create_test_event().await;
let pubkey = event.pubkey;
let identifier = "test-repo".to_string();
@@ -801,20 +780,15 @@ mod tests {
#[tokio::test]
async fn test_multiple_maintainer_repos() {
let index = RejectedEventsIndex::new(
Duration::from_secs(120),
Duration::from_secs(604800),
);
let index = RejectedEventsIndex::new(Duration::from_secs(120), Duration::from_secs(604800));
let keys1 = Keys::generate();
let keys2 = Keys::generate();
let unsigned1 = nostr_sdk::EventBuilder::text_note("test1")
.build(keys1.public_key());
let unsigned1 = nostr_sdk::EventBuilder::text_note("test1").build(keys1.public_key());
let event1 = keys1.sign_event(unsigned1).await.unwrap();
let unsigned2 = nostr_sdk::EventBuilder::text_note("test2")
.build(keys2.public_key());
let unsigned2 = nostr_sdk::EventBuilder::text_note("test2").build(keys2.public_key());
let event2 = keys2.sign_event(unsigned2).await.unwrap();
// Add two different maintainer repos
@@ -836,8 +810,7 @@ mod tests {
assert_eq!(index.cold_index_len(), 2);
// Invalidate only first maintainer
let (removed, hot_events) =
index.invalidate_and_get_events(&event1.pubkey, "repo1");
let (removed, hot_events) = index.invalidate_and_get_events(&event1.pubkey, "repo1");
assert_eq!(removed, 1);
assert_eq!(hot_events.len(), 1);
src/sync/relay_connection.rs
+24 -6
View File
@@ -526,31 +526,46 @@ mod tests {
#[test]
fn test_normalize_url_with_wss_scheme() {
let url = "wss://relay.example.com";
assert_eq!(RelayConnection::normalize_url(url), "wss://relay.example.com");
assert_eq!(
RelayConnection::normalize_url(url),
"wss://relay.example.com"
);
}
#[test]
fn test_normalize_url_with_ws_scheme() {
let url = "ws://relay.example.com";
assert_eq!(RelayConnection::normalize_url(url), "ws://relay.example.com");
assert_eq!(
RelayConnection::normalize_url(url),
"ws://relay.example.com"
);
}
#[test]
fn test_normalize_url_without_scheme() {
let url = "relay.example.com";
assert_eq!(RelayConnection::normalize_url(url), "wss://relay.example.com");
assert_eq!(
RelayConnection::normalize_url(url),
"wss://relay.example.com"
);
}
#[test]
fn test_normalize_url_without_scheme_with_port() {
let url = "relay.example.com:8080";
assert_eq!(RelayConnection::normalize_url(url), "wss://relay.example.com:8080");
assert_eq!(
RelayConnection::normalize_url(url),
"wss://relay.example.com:8080"
);
}
#[test]
fn test_normalize_url_with_path() {
let url = "relay.example.com/nostr";
assert_eq!(RelayConnection::normalize_url(url), "wss://relay.example.com/nostr");
assert_eq!(
RelayConnection::normalize_url(url),
"wss://relay.example.com/nostr"
);
}
#[test]
@@ -587,6 +602,9 @@ mod tests {
fn test_normalize_url_real_world_example() {
// Test the exact case from the bug report
let url = "git.shakespeare.diy";
assert_eq!(RelayConnection::normalize_url(url), "wss://git.shakespeare.diy");
assert_eq!(
RelayConnection::normalize_url(url),
"wss://git.shakespeare.diy"
);
}
}
tests/common/sync_helpers.rs
+1 -1
View File
@@ -708,7 +708,7 @@ impl ParsedMetrics {
/// Check if a specific relay is connected
pub fn relay_connected(&self, relay: &str) -> Option<bool> {
self.gauge("ngit_sync_relay_connected", &[("relay", relay)])
.map(|v| v >= 2) // Syncing (2), Connected (3), or ConnectedHistoricSyncFailures (4)
.map(|v| v >= 2) // Syncing (2), Connected (3), or ConnectedHistoricSyncFailures (4)
}
/// Get total number of connected relays
tests/state_authorization.rs
+119 -123
View File
@@ -13,30 +13,27 @@ use nostr_sdk::prelude::*;
async fn test_reject_state_without_announcement() {
// Start test relay
let relay = TestRelay::start().await;
// Create test keypair
let keys = Keys::generate();
// Create a state event without any announcement
let state_event = EventBuilder::new(
Kind::RepoState,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&keys)
.unwrap();
let state_event = EventBuilder::new(Kind::RepoState, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&keys)
.unwrap();
// Connect to relay
let client = Client::default();
client.add_relay(relay.url()).await.unwrap();
client.connect().await;
// Try to send state event
let result = client.send_event(&state_event).await;
// Should be rejected
match result {
Ok(output) => {
@@ -45,22 +42,26 @@ async fn test_reject_state_without_announcement() {
"Event should be processed"
);
// Check if any relay rejected it
let rejected = output.failed.values().any(|err| {
err.to_string().contains("no announcement exists")
});
assert!(rejected, "Event should be rejected due to missing announcement");
let rejected = output
.failed
.values()
.any(|err| err.to_string().contains("no announcement exists"));
assert!(
rejected,
"Event should be rejected due to missing announcement"
);
}
Err(e) => {
// Also acceptable - relay rejected the event
assert!(
e.to_string().contains("no announcement exists") ||
e.to_string().contains("rejected"),
e.to_string().contains("no announcement exists")
|| e.to_string().contains("rejected"),
"Error should indicate missing announcement: {}",
e
);
}
}
relay.stop().await;
}
@@ -68,67 +69,67 @@ async fn test_reject_state_without_announcement() {
async fn test_reject_state_from_unauthorized_author() {
// Start test relay
let relay = TestRelay::start().await;
// Create two keypairs: one for announcement, one for unauthorized state
let announcement_keys = Keys::generate();
let unauthorized_keys = Keys::generate();
// Create announcement
let announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("clone"), [format!("https://{}/test.git", relay.domain())]),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
])
.sign_with_keys(&announcement_keys)
.unwrap();
let announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(
TagKind::custom("clone"),
[format!("https://{}/test.git", relay.domain())],
),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
])
.sign_with_keys(&announcement_keys)
.unwrap();
// Connect to relay
let client = Client::default();
client.add_relay(relay.url()).await.unwrap();
client.connect().await;
// Send announcement
client.send_event(&announcement).await.unwrap();
// Wait for announcement to be processed
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Try to send state event from unauthorized author
let state_event = EventBuilder::new(
Kind::RepoState,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&unauthorized_keys)
.unwrap();
let state_event = EventBuilder::new(Kind::RepoState, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&unauthorized_keys)
.unwrap();
let result = client.send_event(&state_event).await;
// Should be rejected
match result {
Ok(output) => {
let rejected = output.failed.values().any(|err| {
err.to_string().contains("not authorized")
});
assert!(rejected, "Event should be rejected due to unauthorized author");
let rejected = output
.failed
.values()
.any(|err| err.to_string().contains("not authorized"));
assert!(
rejected,
"Event should be rejected due to unauthorized author"
);
}
Err(e) => {
assert!(
e.to_string().contains("not authorized") ||
e.to_string().contains("rejected"),
e.to_string().contains("not authorized") || e.to_string().contains("rejected"),
"Error should indicate unauthorized author: {}",
e
);
}
}
relay.stop().await;
}
@@ -136,48 +137,45 @@ async fn test_reject_state_from_unauthorized_author() {
async fn test_accept_state_from_announcement_author() {
// Start test relay
let relay = TestRelay::start().await;
// Create keypair
let keys = Keys::generate();
// Create announcement
let announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("clone"), [format!("https://{}/test.git", relay.domain())]),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
])
.sign_with_keys(&keys)
.unwrap();
let announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(
TagKind::custom("clone"),
[format!("https://{}/test.git", relay.domain())],
),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
])
.sign_with_keys(&keys)
.unwrap();
// Connect to relay
let client = Client::default();
client.add_relay(relay.url()).await.unwrap();
client.connect().await;
// Send announcement
client.send_event(&announcement).await.unwrap();
// Wait for announcement to be processed
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Send state event from same author (should be accepted or go to purgatory)
let state_event = EventBuilder::new(
Kind::RepoState,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&keys)
.unwrap();
let state_event = EventBuilder::new(Kind::RepoState, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&keys)
.unwrap();
let result = client.send_event(&state_event).await;
// Should be accepted or go to purgatory (not permanently rejected)
match result {
Ok(output) => {
@@ -194,14 +192,13 @@ async fn test_accept_state_from_announcement_author() {
Err(e) => {
// Purgatory is acceptable
assert!(
e.to_string().contains("purgatory") ||
e.to_string().contains("waiting for git"),
e.to_string().contains("purgatory") || e.to_string().contains("waiting for git"),
"Error should be about purgatory, not authorization: {}",
e
);
}
}
relay.stop().await;
}
@@ -209,50 +206,50 @@ async fn test_accept_state_from_announcement_author() {
async fn test_accept_state_from_maintainer() {
// Start test relay
let relay = TestRelay::start().await;
// Create two keypairs: owner and maintainer
let owner_keys = Keys::generate();
let maintainer_keys = Keys::generate();
// Create announcement with maintainer
let announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("clone"), [format!("https://{}/test.git", relay.domain())]),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
Tag::custom(TagKind::custom("maintainers"), [maintainer_keys.public_key().to_hex()]),
])
.sign_with_keys(&owner_keys)
.unwrap();
let announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(
TagKind::custom("clone"),
[format!("https://{}/test.git", relay.domain())],
),
Tag::custom(TagKind::custom("relays"), [relay.url()]),
Tag::custom(
TagKind::custom("maintainers"),
[maintainer_keys.public_key().to_hex()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
// Connect to relay
let client = Client::default();
client.add_relay(relay.url()).await.unwrap();
client.connect().await;
// Send announcement
client.send_event(&announcement).await.unwrap();
// Wait for announcement to be processed
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Send state event from maintainer
let state_event = EventBuilder::new(
Kind::RepoState,
"",
)
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
let state_event = EventBuilder::new(Kind::RepoState, "")
.tags([
Tag::custom(TagKind::custom("d"), ["test-repo"]),
Tag::custom(TagKind::custom("refs/heads/main"), ["abc123"]),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
let result = client.send_event(&state_event).await;
// Should be accepted or go to purgatory (not permanently rejected)
match result {
Ok(output) => {
@@ -268,13 +265,12 @@ async fn test_accept_state_from_maintainer() {
Err(e) => {
// Purgatory is acceptable
assert!(
e.to_string().contains("purgatory") ||
e.to_string().contains("waiting for git"),
e.to_string().contains("purgatory") || e.to_string().contains("waiting for git"),
"Error should be about purgatory, not authorization: {}",
e
);
}
}
relay.stop().await;
}
tests/sync/maintainer_reprocessing.rs
+146 -165
View File
@@ -42,23 +42,18 @@ async fn test_maintainer_announcement_reprocessed_immediately() {
.await
.expect("Failed to connect to relay_a");
let maintainer_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Maintainer's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_a.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string()],
),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
let maintainer_announcement =
EventBuilder::new(Kind::GitRepoAnnouncement, "Maintainer's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_a.domain(), identifier)],
),
Tag::custom(TagKind::custom("relays"), vec![relay_a.url().to_string()]),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
client_a.send_event(&maintainer_announcement).await.unwrap();
println!("✓ Maintainer announcement sent to relay_a");
@@ -68,27 +63,24 @@ async fn test_maintainer_announcement_reprocessed_immediately() {
.await
.expect("Failed to connect to relay_b");
let owner_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Owner's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_b.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string(), relay_b.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec![maintainer_keys.public_key().to_hex()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
let owner_announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "Owner's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_b.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string(), relay_b.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec![maintainer_keys.public_key().to_hex()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
client_b.send_event(&owner_announcement).await.unwrap();
println!("✓ Owner announcement sent to relay_b");
@@ -104,7 +96,8 @@ async fn test_maintainer_announcement_reprocessed_immediately() {
.author(owner_keys.public_key())
.identifier(identifier);
let owner_found = wait_for_event_on_relay(relay_b.url(), owner_filter, Duration::from_secs(2)).await;
let owner_found =
wait_for_event_on_relay(relay_b.url(), owner_filter, Duration::from_secs(2)).await;
assert!(owner_found, "Owner announcement should be in relay_b");
let maintainer_filter = Filter::new()
@@ -112,8 +105,12 @@ async fn test_maintainer_announcement_reprocessed_immediately() {
.author(maintainer_keys.public_key())
.identifier(identifier);
let maintainer_found = wait_for_event_on_relay(relay_b.url(), maintainer_filter, Duration::from_secs(2)).await;
assert!(maintainer_found, "Maintainer announcement should be re-processed and accepted in relay_b");
let maintainer_found =
wait_for_event_on_relay(relay_b.url(), maintainer_filter, Duration::from_secs(2)).await;
assert!(
maintainer_found,
"Maintainer announcement should be re-processed and accepted in relay_b"
);
// Step 5: Verify it happened quickly (not 24 hours!)
assert!(
@@ -145,36 +142,34 @@ async fn test_maintainer_announcement_reprocessed_immediately() {
#[ignore] // Skip by default due to 2+ minute duration
async fn test_maintainer_announcement_cold_index_prevents_refetch() {
let relay = TestRelay::start().await;
// Create keys
let owner_keys = Keys::generate();
let maintainer_keys = Keys::generate();
let identifier = "test-repo-cold";
// Create client using TestClient helper
let client = TestClient::new(relay.url(), maintainer_keys.clone())
.await
.expect("Failed to connect to relay");
// Step 1: Send maintainer announcement (will be rejected - doesn't list our relay)
let maintainer_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Maintainer's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://example.com/{}.git", identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec!["wss://example.com".to_string()],
),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
let maintainer_announcement =
EventBuilder::new(Kind::GitRepoAnnouncement, "Maintainer's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://example.com/{}.git", identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec!["wss://example.com".to_string()],
),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
// Send maintainer announcement - expect it to be rejected
let _ = client.send_event(&maintainer_announcement).await;
@@ -185,27 +180,21 @@ async fn test_maintainer_announcement_cold_index_prevents_refetch() {
tokio::time::sleep(Duration::from_secs(125)).await;
// Step 3: Send owner announcement (lists maintainer)
let owner_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Owner's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec![maintainer_keys.public_key().to_hex()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
let owner_announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "Owner's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay.domain(), identifier)],
),
Tag::custom(TagKind::custom("relays"), vec![relay.url().to_string()]),
Tag::custom(
TagKind::custom("maintainers"),
vec![maintainer_keys.public_key().to_hex()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
client.send_event(&owner_announcement).await.unwrap();
tokio::time::sleep(Duration::from_millis(500)).await;
@@ -215,16 +204,18 @@ async fn test_maintainer_announcement_cold_index_prevents_refetch() {
.kind(Kind::GitRepoAnnouncement)
.author(owner_keys.public_key())
.identifier(identifier);
let owner_found = wait_for_event_on_relay(relay.url(), owner_filter, Duration::from_secs(2)).await;
let owner_found =
wait_for_event_on_relay(relay.url(), owner_filter, Duration::from_secs(2)).await;
assert!(owner_found, "Owner announcement should be accepted");
let maintainer_filter = Filter::new()
.kind(Kind::GitRepoAnnouncement)
.author(maintainer_keys.public_key())
.identifier(identifier);
let maintainer_found = wait_for_event_on_relay(relay.url(), maintainer_filter, Duration::from_millis(500)).await;
let maintainer_found =
wait_for_event_on_relay(relay.url(), maintainer_filter, Duration::from_millis(500)).await;
assert!(
!maintainer_found,
"Maintainer announcement should NOT be re-processed (hot cache expired)"
@@ -267,7 +258,10 @@ async fn test_multiple_maintainers_all_reprocessed() {
.await
.expect("Failed to connect to relay_a");
for (idx, maintainer_keys) in [&maintainer1_keys, &maintainer2_keys, &maintainer3_keys].iter().enumerate() {
for (idx, maintainer_keys) in [&maintainer1_keys, &maintainer2_keys, &maintainer3_keys]
.iter()
.enumerate()
{
let announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
format!("Maintainer {} repository", idx + 1),
@@ -278,10 +272,7 @@ async fn test_multiple_maintainers_all_reprocessed() {
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_a.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string()],
),
Tag::custom(TagKind::custom("relays"), vec![relay_a.url().to_string()]),
])
.sign_with_keys(maintainer_keys)
.unwrap();
@@ -295,31 +286,28 @@ async fn test_multiple_maintainers_all_reprocessed() {
.await
.expect("Failed to connect to relay_b");
let owner_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Owner's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_b.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string(), relay_b.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec![
maintainer1_keys.public_key().to_hex(),
maintainer2_keys.public_key().to_hex(),
maintainer3_keys.public_key().to_hex(),
],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
let owner_announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "Owner's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay_b.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay_a.url().to_string(), relay_b.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec![
maintainer1_keys.public_key().to_hex(),
maintainer2_keys.public_key().to_hex(),
maintainer3_keys.public_key().to_hex(),
],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
client_b.send_event(&owner_announcement).await.unwrap();
println!("✓ Owner announcement sent to relay_b");
@@ -340,11 +328,7 @@ async fn test_multiple_maintainers_all_reprocessed() {
.identifier(identifier);
let found = wait_for_event_on_relay(relay_b.url(), filter, Duration::from_secs(2)).await;
assert!(
found,
"{} announcement should be in relay_b",
name
);
assert!(found, "{} announcement should be in relay_b", name);
}
println!("✅ All three maintainer announcements re-processed successfully");
@@ -365,63 +349,55 @@ async fn test_multiple_maintainers_all_reprocessed() {
#[tokio::test]
async fn test_invalid_maintainer_pubkey_handled_gracefully() {
let relay = TestRelay::start().await;
// Create keys
let owner_keys = Keys::generate();
let maintainer_keys = Keys::generate();
let identifier = "invalid-maintainer-repo";
// Create client using TestClient helper
let client = TestClient::new(relay.url(), owner_keys.clone())
.await
.expect("Failed to connect to relay");
// Step 1: Send maintainer announcement (will be rejected - doesn't list our relay)
let maintainer_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Maintainer's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://example.com/{}.git", identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec!["wss://example.com".to_string()],
),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
let maintainer_announcement =
EventBuilder::new(Kind::GitRepoAnnouncement, "Maintainer's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://example.com/{}.git", identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec!["wss://example.com".to_string()],
),
])
.sign_with_keys(&maintainer_keys)
.unwrap();
// Send maintainer announcement - expect it to be rejected
let _ = client.send_event(&maintainer_announcement).await;
tokio::time::sleep(Duration::from_millis(200)).await;
// Step 2: Send owner announcement with INVALID maintainer hex
let owner_announcement = EventBuilder::new(
Kind::GitRepoAnnouncement,
"Owner's repository",
)
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay.domain(), identifier)],
),
Tag::custom(
TagKind::custom("relays"),
vec![relay.url().to_string()],
),
Tag::custom(
TagKind::custom("maintainers"),
vec!["invalid-hex-not-a-pubkey".to_string()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
let owner_announcement = EventBuilder::new(Kind::GitRepoAnnouncement, "Owner's repository")
.tags(vec![
Tag::identifier(identifier),
Tag::custom(
TagKind::custom("clone"),
vec![format!("https://{}/{}.git", relay.domain(), identifier)],
),
Tag::custom(TagKind::custom("relays"), vec![relay.url().to_string()]),
Tag::custom(
TagKind::custom("maintainers"),
vec!["invalid-hex-not-a-pubkey".to_string()],
),
])
.sign_with_keys(&owner_keys)
.unwrap();
client.send_event(&owner_announcement).await.unwrap();
tokio::time::sleep(Duration::from_millis(500)).await;
@@ -431,16 +407,21 @@ async fn test_invalid_maintainer_pubkey_handled_gracefully() {
.kind(Kind::GitRepoAnnouncement)
.author(owner_keys.public_key())
.identifier(identifier);
let owner_found = wait_for_event_on_relay(relay.url(), owner_filter, Duration::from_secs(2)).await;
assert!(owner_found, "Owner announcement should be accepted despite invalid maintainer");
let owner_found =
wait_for_event_on_relay(relay.url(), owner_filter, Duration::from_secs(2)).await;
assert!(
owner_found,
"Owner announcement should be accepted despite invalid maintainer"
);
let maintainer_filter = Filter::new()
.kind(Kind::GitRepoAnnouncement)
.author(maintainer_keys.public_key())
.identifier(identifier);
let maintainer_found = wait_for_event_on_relay(relay.url(), maintainer_filter, Duration::from_millis(500)).await;
let maintainer_found =
wait_for_event_on_relay(relay.url(), maintainer_filter, Duration::from_millis(500)).await;
assert!(
!maintainer_found,
"Maintainer announcement should NOT be re-processed (invalid pubkey)"