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

fix: resolve all fmt and clippy warnings

Browse Source 
Main lib (src/):
- Add #[allow(dead_code)] for build_info field (stored to prevent Prometheus unregistration)
- Add #[allow(dead_code)] for first_seen field (reserved for future rate limiting)
- Replace .or_insert_with(RelaySyncNeeds::default) with .or_default()
- Replace manual div_ceil implementations with .div_ceil(100)

Test code (tests/):
- Replace .expect(&format!(...)) with .unwrap_or_else(|_| panic!(...))
- Remove needless borrows in fetch_metrics() calls
- Add #[allow(dead_code)] and #[allow(unused_imports)] to test helpers module

grasp-audit:
- Apply cargo fmt to fix formatting
This commit is contained in:
DanConwayDev
2025-12-11 16:53:03 +00:00
parent 52489d3b1a
commit 2a9160836b
37 changed files with 515 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build.rs
+1 -1
View File
@@ -17,4 +17,4 @@ fn main() {
// Re-run if HEAD changes (new commits)
println!("cargo:rerun-if-changed=.git/HEAD");
println!("cargo:rerun-if-changed=.git/refs/heads/");
}
}
grasp-audit/src/client.rs
+3 -1
View File
@@ -585,7 +585,9 @@ mod tests {
"Missing 'grasp-audit-test-event' tag"
);
assert!(
tag_contents.iter().any(|t| t.starts_with("audit-isolated-")),
tag_contents
.iter()
.any(|t| t.starts_with("audit-isolated-")),
"Missing 'audit-isolated-*' tag"
);
assert!(
grasp-audit/src/fixtures.rs
+3 -3
View File
@@ -1973,11 +1973,11 @@ mod tests {
#[test]
fn test_context_mode_from_audit_mode() {
assert_eq!(ContextMode::from(AuditMode::Isolated), ContextMode::Isolated);
assert_eq!(
ContextMode::from(AuditMode::Shared),
ContextMode::Shared
ContextMode::from(AuditMode::Isolated),
ContextMode::Isolated
);
assert_eq!(ContextMode::from(AuditMode::Shared), ContextMode::Shared);
}
#[test]
grasp-audit/src/result.rs
+32 -9
View File
@@ -38,7 +38,9 @@ fn parse_spec_lines(spec_ref: &str) -> Vec<u32> {
if line_part.contains('-') {
let range_parts: Vec<&str> = line_part.split('-').collect();
if range_parts.len() == 2 {
if let (Ok(start), Ok(end)) = (range_parts[0].parse::<u32>(), range_parts[1].parse::<u32>()) {
if let (Ok(start), Ok(end)) =
(range_parts[0].parse::<u32>(), range_parts[1].parse::<u32>())
{
return (start..=end).collect();
}
}
@@ -162,10 +164,19 @@ impl AuditResult {
/// Print a detailed report aligned to GRASP-01 specification
pub fn print_report(&self) {
println!();
println!("{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}", BOLD, RESET);
println!(
"{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}",
BOLD, RESET
);
println!("{}GRASP-01 Compliance Report{}", BOLD, RESET);
println!("Source: github.com/nostr-protocol/grasp (commit: {})", GRASP_COMMIT_ID);
println!("{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}", BOLD, RESET);
println!(
"Source: github.com/nostr-protocol/grasp (commit: {})",
GRASP_COMMIT_ID
);
println!(
"{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}",
BOLD, RESET
);
// Build a map of spec line -> tests that cover it
let mut tests_by_line: BTreeMap<u32, Vec<&TestResult>> = BTreeMap::new();
@@ -185,7 +196,10 @@ impl AuditResult {
println!();
println!("{}{}## {}{}", CYAN, BOLD, section, RESET);
for req in GRASP_01_REQUIREMENTS.iter().filter(|r| r.section == section) {
for req in GRASP_01_REQUIREMENTS
.iter()
.filter(|r| r.section == section)
{
println!();
// Print spec requirement in blue
println!("{}📘 Line {}: {}{}", BLUE, req.line, req.text, RESET);
@@ -218,7 +232,10 @@ impl AuditResult {
}
println!();
println!("{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}", BOLD, RESET);
println!(
"{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}",
BOLD, RESET
);
// Summary statistics
let passed = self.passed_count();
@@ -252,7 +269,10 @@ impl AuditResult {
"{}Test results: {}/{} tests passed ({:.1}%){}",
summary_color, passed, total_tests, pass_rate, RESET
);
println!("{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}", BOLD, RESET);
println!(
"{}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━{}",
BOLD, RESET
);
println!();
}
@@ -313,7 +333,10 @@ mod tests {
#[test]
fn test_parse_spec_lines_range() {
assert_eq!(parse_spec_lines("GRASP-01:nostr-relay:7-9"), vec![7, 8, 9]);
assert_eq!(parse_spec_lines("GRASP-01:cors:44-47"), vec![44, 45, 46, 47]);
assert_eq!(
parse_spec_lines("GRASP-01:cors:44-47"),
vec![44, 45, 46, 47]
);
}
#[test]
@@ -327,4 +350,4 @@ mod tests {
assert_eq!(parse_spec_lines("GRASP-01:invalid"), Vec::<u32>::new());
assert_eq!(parse_spec_lines("GRASP-01:test:abc"), Vec::<u32>::new());
}
}
}
grasp-audit/src/specs/grasp01/mod.rs
+2 -2
View File
@@ -29,6 +29,6 @@ pub use nip11_document::Nip11DocumentTests;
pub use push_authorization::PushAuthorizationTests;
pub use repository_creation::RepositoryCreationTests;
pub use spec_requirements::{
get_requirement, get_requirements_for_section, get_sections, RequirementLevel,
SpecRequirement, GRASP_01_REQUIREMENTS, GRASP_COMMIT_ID,
get_requirement, get_requirements_for_section, get_sections, RequirementLevel, SpecRequirement,
GRASP_01_REQUIREMENTS, GRASP_COMMIT_ID,
};
grasp-audit/src/specs/grasp01/nip01_smoke.rs
+18 -14
View File
@@ -163,23 +163,27 @@ impl Nip01SmokeTests {
/// Spec: NIP-01 CLOSE message
/// Requirement: Relay MUST support CLOSE to end subscriptions
pub async fn test_close_subscription(client: &AuditClient) -> TestResult {
TestResult::new("close_subscription", "GRASP-01:nostr-relay:7", "Can close subscriptions")
.run(|| async {
// For now, we just verify we can query events
// Full subscription management with CLOSE would require
// lower-level WebSocket access
TestResult::new(
"close_subscription",
"GRASP-01:nostr-relay:7",
"Can close subscriptions",
)
.run(|| async {
// For now, we just verify we can query events
// Full subscription management with CLOSE would require
// lower-level WebSocket access
let filter = Filter::new().kind(Kind::TextNote).limit(1);
let filter = Filter::new().kind(Kind::TextNote).limit(1);
let _events = client
.subscribe(vec![filter], Some(std::time::Duration::from_secs(2)))
.await
.map_err(|e| format!("Failed to subscribe: {}", e))?;
let _events = client
.subscribe(vec![filter], Some(std::time::Duration::from_secs(2)))
.await
.map_err(|e| format!("Failed to subscribe: {}", e))?;
// If we got here, subscription worked
Ok(())
})
.await
// If we got here, subscription worked
Ok(())
})
.await
}
/// Test 5: Rejects events with invalid signatures
grasp-audit/src/specs/grasp01/push_authorization.rs
+69 -28
View File
@@ -407,8 +407,12 @@ impl PushAuthorizationTests {
let repo = match ctx.get_fixture(FixtureKind::ValidRepo).await {
Ok(r) => r,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event")
.fail(format!("Failed to create repo: {}", e))
return TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.fail(format!("Failed to create repo: {}", e))
}
};
@@ -427,8 +431,12 @@ impl PushAuthorizationTests {
let clone_path = match clone_repo(relay_domain, &npub, &repo_id) {
Ok(p) => p,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event")
.fail(&e)
return TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.fail(&e)
}
};
let cleanup = || {
@@ -437,8 +445,12 @@ impl PushAuthorizationTests {
if let Err(e) = create_commit(&clone_path, "Unauthorized commit") {
cleanup();
return TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event")
.fail(&e);
return TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.fail(&e);
}
// Do NOT publish state event - push should be rejected
@@ -446,14 +458,24 @@ impl PushAuthorizationTests {
cleanup();
match push_result {
Ok(false) => {
TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event").pass()
}
Ok(true) => TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event")
.fail("Push accepted but should be rejected"),
Err(e) => {
TestResult::new(test_name, "GRASP-01:git-http:30", "Push rejected without state event").fail(&e)
}
Ok(false) => TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.pass(),
Ok(true) => TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.fail("Push accepted but should be rejected"),
Err(e) => TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push rejected without state event",
)
.fail(&e),
}
}
@@ -480,11 +502,18 @@ impl PushAuthorizationTests {
// The OwnerStateDataPushed fixture handles all stages:
// Generate → Send → Verify → DataPush
match ctx.get_fixture(FixtureKind::OwnerStateDataPushed).await {
Ok(_state_event) => {
TestResult::new(test_name, "GRASP-01:git-http:30", "Push authorized with matching state").pass()
}
Err(e) => TestResult::new(test_name, "GRASP-01:git-http:30", "Push authorized with matching state")
.fail(format!("{}", e)),
Ok(_state_event) => TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push authorized with matching state",
)
.pass(),
Err(e) => TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Push authorized with matching state",
)
.fail(format!("{}", e)),
}
}
@@ -868,8 +897,12 @@ impl PushAuthorizationTests {
// Send the rogue state event using the raw client to bypass AuditClient's key check
if let Err(e) = client.client().send_event(&rogue_state).await {
cleanup();
return TestResult::new(test_name, "GRASP-01:git-http:30", "Non-maintainer state events ignored")
.fail(format!("Failed to send rogue state event: {}", e));
return TestResult::new(
test_name,
"GRASP-01:git-http:30",
"Non-maintainer state events ignored",
)
.fail(format!("Failed to send rogue state event: {}", e));
}
// Wait for event to propagate
@@ -1036,7 +1069,9 @@ impl PushAuthorizationTests {
.await
{
Ok(_pr_event) => TestResult::new(test_name, "GRASP-01:git-http:34", desc).pass(),
Err(e) => TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e)),
Err(e) => {
TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e))
}
}
}
@@ -1062,7 +1097,8 @@ impl PushAuthorizationTests {
{
Ok(e) => e,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
@@ -1072,7 +1108,8 @@ impl PushAuthorizationTests {
let repo = match ctx.get_fixture(FixtureKind::ValidRepo).await {
Ok(r) => r,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
@@ -1146,7 +1183,8 @@ impl PushAuthorizationTests {
{
Ok(e) => e,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
@@ -1156,7 +1194,8 @@ impl PushAuthorizationTests {
let repo = match ctx.get_fixture(FixtureKind::ValidRepo).await {
Ok(r) => r,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
@@ -1233,7 +1272,8 @@ impl PushAuthorizationTests {
{
Ok(e) => e,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
@@ -1243,7 +1283,8 @@ impl PushAuthorizationTests {
let repo = match ctx.get_fixture(FixtureKind::ValidRepo).await {
Ok(r) => r,
Err(e) => {
return TestResult::new(test_name, "GRASP-01:git-http:34", desc).fail(format!("{}", e));
return TestResult::new(test_name, "GRASP-01:git-http:34", desc)
.fail(format!("{}", e));
}
};
src/config.rs
+15 -3
View File
@@ -77,11 +77,19 @@ pub struct Config {
pub metrics_enabled: bool,
/// Connections per IP before flagging as potential abuse in metrics (display only, no rate limiting)
#[arg(long = "metrics-connection-per-ip-abuse-threshold", env = "NGIT_METRICS_CONNECTION_PER_IP_ABUSE_THRESHOLD", default_value_t = 10)]
#[arg(
long = "metrics-connection-per-ip-abuse-threshold",
env = "NGIT_METRICS_CONNECTION_PER_IP_ABUSE_THRESHOLD",
default_value_t = 10
)]
pub metrics_connection_per_ip_abuse_threshold: u32,
/// Number of top bandwidth repos to track in metrics
#[arg(long = "metrics-top-n-repos", env = "NGIT_METRICS_TOP_N_REPOS", default_value_t = 10)]
#[arg(
long = "metrics-top-n-repos",
env = "NGIT_METRICS_TOP_N_REPOS",
default_value_t = 10
)]
pub metrics_top_n_repos: usize,
/// URL of bootstrap relay to sync from on startup (optional)
@@ -95,7 +103,11 @@ pub struct Config {
/// Interval in seconds for checking disconnected relays and attempting reconnection (default: 60)
/// Set to lower value for faster reconnection testing
#[arg(long, env = "NGIT_SYNC_DISCONNECT_CHECK_INTERVAL_SECS", default_value_t = 60)]
#[arg(
long,
env = "NGIT_SYNC_DISCONNECT_CHECK_INTERVAL_SECS",
default_value_t = 60
)]
pub sync_disconnect_check_interval_secs: u64,
/// Base backoff time in seconds for relay reconnection (default: 5)
src/http/landing.rs
+1 -4
View File
@@ -341,10 +341,7 @@ fn generate_hero_tags(nip11: &RelayInformationDocument) -> String {
// Add GRASP tags
for grasp in &nip11.supported_grasps {
html.push_str(&format!(
r#"<span class="tag tag-grasp">{}</span>"#,
grasp
));
html.push_str(&format!(r#"<span class="tag tag-grasp">{}</span>"#, grasp));
html.push('\n');
}
src/http/mod.rs
+7 -1
View File
@@ -509,7 +509,13 @@ pub async fn run_server(
loop {
let (socket, addr) = listener.accept().await?;
let io = TokioIo::new(socket);
let service = HttpService::new(relay.clone(), config.clone(), addr, database.clone(), metrics.clone());
let service = HttpService::new(
relay.clone(),
config.clone(),
addr,
database.clone(),
metrics.clone(),
);
tokio::spawn(async move {
if let Err(e) = http1::Builder::new()
src/main.rs
+7 -2
View File
@@ -37,7 +37,9 @@ async fn main() -> Result<()> {
// Initialize metrics if enabled
let metrics = if config.metrics_enabled {
info!("Metrics enabled on /metrics endpoint");
Some(Arc::new(Metrics::new(config.metrics_connection_per_ip_abuse_threshold)))
Some(Arc::new(Metrics::new(
config.metrics_connection_per_ip_abuse_threshold,
)))
} else {
info!("Metrics disabled");
None
@@ -65,7 +67,10 @@ async fn main() -> Result<()> {
);
if config.sync_bootstrap_relay_url.is_some() {
info!("Starting proactive sync with bootstrap relay: {:?}", config.sync_bootstrap_relay_url);
info!(
"Starting proactive sync with bootstrap relay: {:?}",
config.sync_bootstrap_relay_url
);
} else {
info!("Proactive sync enabled (will discover relays from stored announcements)");
}
src/metrics/bandwidth.rs
+10 -3
View File
@@ -80,7 +80,9 @@ impl BandwidthTracker {
&["repo"],
)
.unwrap();
registry.register(Box::new(top_repos_gauge.clone())).unwrap();
registry
.register(Box::new(top_repos_gauge.clone()))
.unwrap();
Self {
all_repos: DashMap::new(),
@@ -120,7 +122,12 @@ impl BandwidthTracker {
// Try to update the timestamp atomically to prevent concurrent refreshes
if self
.last_refresh_nanos
.compare_exchange(last_refresh, elapsed_nanos, Ordering::SeqCst, Ordering::Relaxed)
.compare_exchange(
last_refresh,
elapsed_nanos,
Ordering::SeqCst,
Ordering::Relaxed,
)
.is_ok()
{
self.refresh_top_n();
@@ -298,4 +305,4 @@ mod tests {
// Refresh should not panic on empty data
tracker.refresh_top_n();
}
}
}
src/metrics/connection.rs
+29 -27
View File
@@ -25,7 +25,8 @@ use tracing::warn;
struct ConnectionInfo {
/// Number of active connections from this IP
count: u32,
/// When the first connection from this IP was established
/// When the first connection from this IP was established (for future rate limiting)
#[allow(dead_code)]
first_seen: Instant,
/// Whether this IP has been flagged as potentially abusive
flagged_as_abuse: bool,
@@ -48,16 +49,16 @@ struct ConnectionInfo {
pub struct ConnectionTracker {
/// Active connections per IP (INTERNAL ONLY - never exposed to metrics)
connections: DashMap<IpAddr, ConnectionInfo>,
/// Threshold for abuse flagging (connections per IP)
abuse_threshold: u32,
/// Prometheus gauge: total active connections
active_connections: IntGauge,
/// Prometheus gauge: number of unique IPs connected
unique_ips: IntGauge,
/// Prometheus gauge: number of IPs flagged as potential abusers
flagged_abusers: IntGauge,
}
@@ -70,29 +71,30 @@ impl ConnectionTracker {
/// * `abuse_threshold` - Number of connections from a single IP before flagging
/// * `registry` - Prometheus registry to register metrics with
pub fn new(abuse_threshold: u32, registry: &Registry) -> Self {
let active_connections = IntGauge::with_opts(
Opts::new(
"ngit_websocket_connections_active",
"Current active WebSocket connections",
)
).unwrap();
registry.register(Box::new(active_connections.clone())).unwrap();
let active_connections = IntGauge::with_opts(Opts::new(
"ngit_websocket_connections_active",
"Current active WebSocket connections",
))
.unwrap();
registry
.register(Box::new(active_connections.clone()))
.unwrap();
let unique_ips = IntGauge::with_opts(
Opts::new(
"ngit_websocket_unique_ips",
"Number of unique IP addresses connected (NOT the IPs themselves)",
)
).unwrap();
let unique_ips = IntGauge::with_opts(Opts::new(
"ngit_websocket_unique_ips",
"Number of unique IP addresses connected (NOT the IPs themselves)",
))
.unwrap();
registry.register(Box::new(unique_ips.clone())).unwrap();
let flagged_abusers = IntGauge::with_opts(
Opts::new(
"ngit_websocket_flagged_abusers",
"Number of IPs exceeding connection threshold",
)
).unwrap();
registry.register(Box::new(flagged_abusers.clone())).unwrap();
let flagged_abusers = IntGauge::with_opts(Opts::new(
"ngit_websocket_flagged_abusers",
"Number of IPs exceeding connection threshold",
))
.unwrap();
registry
.register(Box::new(flagged_abusers.clone()))
.unwrap();
Self {
connections: DashMap::new(),
@@ -140,7 +142,7 @@ impl ConnectionTracker {
// Update Prometheus metrics (aggregate counts only)
self.active_connections.inc();
if is_new_ip {
self.unique_ips.inc();
}
@@ -334,4 +336,4 @@ mod tests {
assert_eq!(tracker.active_connections(), 0);
assert_eq!(tracker.unique_ip_count(), 0);
}
}
}
src/metrics/mod.rs
+82 -45
View File
@@ -87,7 +87,8 @@ struct MetricsInner {
// === System Health Metrics ===
/// Server start time for uptime calculation
pub start_time: Instant,
/// Build information gauge
/// Build information gauge (stored to prevent unregistration from Prometheus)
#[allow(dead_code)]
pub build_info: GaugeVec,
}
@@ -158,7 +159,10 @@ impl Metrics {
/// Start timing a git operation, returns a timer
pub fn start_git_operation_timer(&self, operation: &str) -> GitOperationTimer {
GitOperationTimer::new(self.inner.git_operation_duration.clone(), operation.to_string())
GitOperationTimer::new(
self.inner.git_operation_duration.clone(),
operation.to_string(),
)
}
/// Record bytes transferred for a git operation
@@ -266,13 +270,14 @@ impl MetricsInner {
}
// WebSocket metrics
let websocket_connections_total = Counter::with_opts(
Opts::new(
"ngit_websocket_connections_total",
"Total WebSocket connections since startup",
)
).unwrap();
REGISTRY.register(Box::new(websocket_connections_total.clone())).unwrap();
let websocket_connections_total = Counter::with_opts(Opts::new(
"ngit_websocket_connections_total",
"Total WebSocket connections since startup",
))
.unwrap();
REGISTRY
.register(Box::new(websocket_connections_total.clone()))
.unwrap();
let websocket_connection_duration = Histogram::with_opts(
HistogramOpts::new(
@@ -280,8 +285,11 @@ impl MetricsInner {
"Duration of WebSocket connections",
)
.buckets(vec![1.0, 5.0, 15.0, 30.0, 60.0, 300.0, 900.0, 3600.0]),
).unwrap();
REGISTRY.register(Box::new(websocket_connection_duration.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(websocket_connection_duration.clone()))
.unwrap();
let websocket_messages_received = CounterVec::new(
Opts::new(
@@ -289,8 +297,11 @@ impl MetricsInner {
"WebSocket messages received by type",
),
&["type"],
).unwrap();
REGISTRY.register(Box::new(websocket_messages_received.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(websocket_messages_received.clone()))
.unwrap();
let websocket_messages_sent = CounterVec::new(
Opts::new(
@@ -298,8 +309,11 @@ impl MetricsInner {
"WebSocket messages sent by type",
),
&["type"],
).unwrap();
REGISTRY.register(Box::new(websocket_messages_sent.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(websocket_messages_sent.clone()))
.unwrap();
// Git operation metrics
let git_operations_total = CounterVec::new(
@@ -308,8 +322,11 @@ impl MetricsInner {
"Git operations by type and status",
),
&["operation", "status"],
).unwrap();
REGISTRY.register(Box::new(git_operations_total.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(git_operations_total.clone()))
.unwrap();
let git_operation_duration = HistogramVec::new(
HistogramOpts::new(
@@ -318,8 +335,11 @@ impl MetricsInner {
)
.buckets(vec![0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0]),
&["operation"],
).unwrap();
REGISTRY.register(Box::new(git_operation_duration.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(git_operation_duration.clone()))
.unwrap();
let git_bytes_total = CounterVec::new(
Opts::new(
@@ -327,8 +347,11 @@ impl MetricsInner {
"Total bytes transferred for git operations",
),
&["direction"],
).unwrap();
REGISTRY.register(Box::new(git_bytes_total.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(git_bytes_total.clone()))
.unwrap();
let git_push_authorization = CounterVec::new(
Opts::new(
@@ -336,8 +359,11 @@ impl MetricsInner {
"Push authorization results",
),
&["result"],
).unwrap();
REGISTRY.register(Box::new(git_push_authorization.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(git_push_authorization.clone()))
.unwrap();
// Nostr event metrics
let events_received_total = CounterVec::new(
@@ -346,8 +372,11 @@ impl MetricsInner {
"Nostr events received by kind",
),
&["kind"],
).unwrap();
REGISTRY.register(Box::new(events_received_total.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(events_received_total.clone()))
.unwrap();
let events_stored_total = CounterVec::new(
Opts::new(
@@ -355,8 +384,11 @@ impl MetricsInner {
"Nostr events successfully stored by kind",
),
&["kind"],
).unwrap();
REGISTRY.register(Box::new(events_stored_total.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(events_stored_total.clone()))
.unwrap();
let events_rejected_total = CounterVec::new(
Opts::new(
@@ -364,31 +396,36 @@ impl MetricsInner {
"Nostr events rejected by kind and reason",
),
&["kind", "reason"],
).unwrap();
REGISTRY.register(Box::new(events_rejected_total.clone())).unwrap();
)
.unwrap();
REGISTRY
.register(Box::new(events_rejected_total.clone()))
.unwrap();
// Repository metrics
let repositories_total = Gauge::with_opts(
Opts::new(
"ngit_repositories_total",
"Total repositories hosted",
)
).unwrap();
REGISTRY.register(Box::new(repositories_total.clone())).unwrap();
let repositories_total = Gauge::with_opts(Opts::new(
"ngit_repositories_total",
"Total repositories hosted",
))
.unwrap();
REGISTRY
.register(Box::new(repositories_total.clone()))
.unwrap();
// Build info
let build_info = GaugeVec::new(
Opts::new(
"ngit_build_info",
"Build information",
),
Opts::new("ngit_build_info", "Build information"),
&["version", "commit"],
).unwrap();
)
.unwrap();
REGISTRY.register(Box::new(build_info.clone())).unwrap();
// Set build info gauge to 1 (it's just for labels)
build_info
.with_label_values(&[env!("CARGO_PKG_VERSION"), option_env!("GIT_HASH").unwrap_or("unknown")])
.with_label_values(&[
env!("CARGO_PKG_VERSION"),
option_env!("GIT_HASH").unwrap_or("unknown"),
])
.set(1.0);
Self {
@@ -472,7 +509,7 @@ mod tests {
// Note: This test may fail if run with other tests due to global registry
// In production, consider using a test-specific registry
let metrics = Metrics::new(10);
// Test that we can record metrics without panicking
metrics.record_websocket_connection();
metrics.record_message_received("REQ");
@@ -484,4 +521,4 @@ mod tests {
metrics.record_event_rejected(1, "invalid_signature");
metrics.set_repositories_total(5);
}
}
}
src/nostr/builder.rs
+8 -14
View File
@@ -16,8 +16,8 @@ use crate::nostr::events::{
KIND_REPOSITORY_STATE,
};
use crate::nostr::policy::{
AnnouncementPolicy, AnnouncementResult, PolicyContext, PrEventPolicy, RelatedEventPolicy,
ReferenceResult, StatePolicy, StateResult,
AnnouncementPolicy, AnnouncementResult, PolicyContext, PrEventPolicy, ReferenceResult,
RelatedEventPolicy, StatePolicy, StateResult,
};
/// Type alias for the shared database used by the relay
@@ -77,7 +77,9 @@ impl Nip34WritePolicy {
match RepositoryAnnouncement::from_event(event.clone()) {
Ok(announcement) => {
// Try to create bare repository if it doesn't exist
if let Err(e) = self.announcement_policy.ensure_bare_repository(&announcement)
if let Err(e) = self
.announcement_policy
.ensure_bare_repository(&announcement)
{
tracing::warn!(
"Failed to create bare repository for {}: {}",
@@ -145,22 +147,14 @@ impl Nip34WritePolicy {
Ok(_state) => {
// Process state alignment asynchronously
if let Err(e) = self.state_policy.process_state_event(event).await {
tracing::warn!(
"Failed to process state event {}: {}",
event_id_str,
e
);
tracing::warn!("Failed to process state event {}: {}", event_id_str, e);
}
tracing::debug!("Accepted repository state: {}", event_id_str);
PolicyResult::Accept
}
Err(e) => {
tracing::warn!(
"Failed to parse repository state {}: {}",
event_id_str,
e
);
tracing::warn!("Failed to parse repository state {}: {}", event_id_str, e);
// Still accept the event even if we can't parse it
// The validation passed, so it's structurally valid
PolicyResult::Accept
@@ -348,4 +342,4 @@ pub fn create_relay(config: &Config) -> Result<RelayWithDatabase> {
database,
write_policy,
})
}
}
src/nostr/policy/announcement.rs
+5 -2
View File
@@ -72,7 +72,10 @@ impl AnnouncementPolicy {
/// Create a bare git repository if it doesn't exist
/// Path format: <git_data_path>/<npub>/<identifier>.git
pub fn ensure_bare_repository(&self, announcement: &RepositoryAnnouncement) -> Result<(), String> {
pub fn ensure_bare_repository(
&self,
announcement: &RepositoryAnnouncement,
) -> Result<(), String> {
let repo_path = self.ctx.git_data_path.join(announcement.repo_path());
// Check if repository already exists
@@ -154,4 +157,4 @@ impl AnnouncementPolicy {
Ok(false)
}
}
}
src/nostr/policy/mod.rs
+1 -2
View File
@@ -5,7 +5,6 @@
/// - `StatePolicy` - State event validation + ref alignment
/// - `PrEventPolicy` - PR/PR Update validation
/// - `RelatedEventPolicy` - Forward/backward reference checking
mod announcement;
mod pr_event;
mod related;
@@ -38,4 +37,4 @@ impl PolicyContext {
git_data_path: git_data_path.into(),
}
}
}
}
src/nostr/policy/pr_event.rs
+1 -1
View File
@@ -195,4 +195,4 @@ impl PrEventPolicy {
Ok(None)
}
}
}
}
src/nostr/policy/related.rs
+2 -5
View File
@@ -169,10 +169,7 @@ impl RelatedEventPolicy {
/// Check if any events exist in database
/// Returns the first matching event ID found, or None if none match
async fn find_accepted_event(
&self,
event_ids: &[EventId],
) -> Result<Option<EventId>, String> {
async fn find_accepted_event(&self, event_ids: &[EventId]) -> Result<Option<EventId>, String> {
if event_ids.is_empty() {
return Ok(None);
}
@@ -273,4 +270,4 @@ impl RelatedEventPolicy {
Ok(false)
}
}
}
src/nostr/policy/state.rs
+5 -2
View File
@@ -239,7 +239,10 @@ impl StatePolicy {
}
// Build repository path: <git_data_path>/<owner_npub>/<identifier>.git
let repo_path = self.ctx.git_data_path.join(announcement.repo_path().clone());
let repo_path = self
.ctx
.git_data_path
.join(announcement.repo_path().clone());
owner_repos.push((announcement, repo_path));
}
@@ -416,4 +419,4 @@ impl StatePolicy {
result
}
}
}
src/sync/algorithms.rs
+2 -4
View File
@@ -65,9 +65,7 @@ pub fn derive_relay_targets(
for (repo_id, needs) in repo_index {
for relay_url in &needs.relays {
let entry = relay_targets
.entry(relay_url.clone())
.or_insert_with(RelaySyncNeeds::default);
let entry = relay_targets.entry(relay_url.clone()).or_default();
entry.repos.insert(repo_id.clone());
entry.root_events.extend(needs.root_events.iter().cloned());
@@ -586,4 +584,4 @@ mod tests {
);
assert_eq!(actions[0].relay_url, "wss://new-relay.com");
}
}
}
src/sync/filters.rs
+1 -1
View File
@@ -337,4 +337,4 @@ mod tests {
assert_eq!(filters.len(), 6);
}
}
}
src/sync/health.rs
+10 -9
View File
@@ -206,11 +206,7 @@ impl RelayHealthTracker {
health.next_retry_at = Some(now + backoff);
if old_state != HealthState::Degraded {
tracing::warn!(
"Relay {} degraded, backoff {:?}",
relay_url,
backoff
);
tracing::warn!("Relay {} degraded, backoff {:?}", relay_url, backoff);
} else {
tracing::debug!(
"Relay {} failure #{}, backoff {:?}",
@@ -308,12 +304,17 @@ impl RelayHealthTracker {
/// Get all tracked relay URLs
pub fn get_tracked_relays(&self) -> Vec<String> {
self.health.iter().map(|entry| entry.key().clone()).collect()
self.health
.iter()
.map(|entry| entry.key().clone())
.collect()
}
/// Get a clone of the health info for a relay
pub fn get_health(&self, relay_url: &str) -> Option<RelayHealth> {
self.health.get(relay_url).map(|entry| entry.value().clone())
self.health
.get(relay_url)
.map(|entry| entry.value().clone())
}
}
@@ -369,7 +370,7 @@ mod tests {
fn test_backoff_increases_exponentially() {
let base = DEFAULT_BASE_BACKOFF_SECS; // 5 seconds
let max = 3600u64;
// failure 1: 5s (base * 2^0 = 5)
assert_eq!(
RelayHealthTracker::get_backoff_duration(1, base, max),
@@ -498,4 +499,4 @@ mod tests {
let health = tracker.get_health("wss://nonexistent.example.com");
assert!(health.is_none());
}
}
}
src/sync/metrics.rs
+5 -5
View File
@@ -207,7 +207,9 @@ impl SyncMetrics {
HealthState::Degraded => 2,
HealthState::Dead => 3,
};
self.relay_status.with_label_values(&[relay]).set(state_value);
self.relay_status
.with_label_values(&[relay])
.set(state_value);
}
/// Record relay failure count.
@@ -259,9 +261,7 @@ impl SyncMetrics {
/// * `source` - The event source type (see [`record_event`](Self::record_event))
/// * `count` - Number of events to record
pub fn record_events(&self, source: &str, count: u64) {
self.events_total
.with_label_values(&[source])
.inc_by(count);
self.events_total.with_label_values(&[source]).inc_by(count);
}
/// Record a gap event filled during catchup.
@@ -451,4 +451,4 @@ mod tests {
let metrics2 = SyncMetrics::register(&registry);
assert!(metrics2.is_err());
}
}
}
src/sync/mod.rs
+4 -4
View File
@@ -512,8 +512,8 @@ impl SyncManager {
};
// Check if relay supports NIP-77 negentropy AND negentropy is not disabled
let use_negentropy = !self.config.sync_disable_negentropy
&& connection.supports_negentropy().await;
let use_negentropy =
!self.config.sync_disable_negentropy && connection.supports_negentropy().await;
// Unsubscribe all current subscriptions
connection.unsubscribe_all().await;
@@ -1657,12 +1657,12 @@ impl SyncManager {
let layer1_filters = 1;
let layer2_filters = if repo_count > 0 {
((repo_count + 99) / 100) * 3
repo_count.div_ceil(100) * 3
} else {
0
};
let layer3_filters = if event_count > 0 {
((event_count + 99) / 100) * 3
event_count.div_ceil(100) * 3
} else {
0
};
src/sync/relay_connection.rs
+16 -8
View File
@@ -150,17 +150,21 @@ impl RelayConnection {
//
// See: nostr-sdk-0.44 Client::try_connect_relay documentation
self.client
.try_connect_relay(&self.url, std::time::Duration::from_secs(connection_timeout_secs))
.try_connect_relay(
&self.url,
std::time::Duration::from_secs(connection_timeout_secs),
)
.await
.map_err(|e| format!("Failed to connect to relay {}: {}", self.url, e))?;
// Subscribe to Layer 1 (announcements)
let filter = build_announcement_filter(since);
let output = self
.client
.subscribe(filter, None)
.await
.map_err(|e| format!("Failed to subscribe to announcements on {}: {}", self.url, e))?;
let output = self.client.subscribe(filter, None).await.map_err(|e| {
format!(
"Failed to subscribe to announcements on {}: {}",
self.url, e
)
})?;
tracing::info!(url = %self.url, sub_id = %output.val, "Connected and subscribed to Layer 1 (announcements)");
Ok(output.val)
@@ -250,7 +254,8 @@ impl RelayConnection {
}
RelayMessage::Closed { message: msg, .. } => {
tracing::info!(relay = %url, message = %msg, "Relay closed subscription");
let _ = event_sender.send(RelayEvent::Closed(msg.to_string())).await;
let _ =
event_sender.send(RelayEvent::Closed(msg.to_string())).await;
break;
}
_ => {}
@@ -421,7 +426,10 @@ impl RelayConnection {
/// - Relay doesn't actually support NIP-77 (despite claiming to)
/// - Network errors during reconciliation
/// - Timeout during sync
pub async fn negentropy_sync_filter(&self, filter: Filter) -> Result<NegentropySyncResult, String> {
pub async fn negentropy_sync_filter(
&self,
filter: Filter,
) -> Result<NegentropySyncResult, String> {
// Use nostr-sdk's sync method which handles the NEG-OPEN/NEG-MSG exchange
let sync_opts = SyncOptions::default();
src/sync/self_subscriber.rs
+26 -16
View File
@@ -49,7 +49,12 @@ impl PendingUpdates {
}
/// Add or update a repo with its relays and root events
fn add_repo(&mut self, repo_id: String, relays: HashSet<String>, root_events: HashSet<EventId>) {
fn add_repo(
&mut self,
repo_id: String,
relays: HashSet<String>,
root_events: HashSet<EventId>,
) {
let entry = self.repos.entry(repo_id).or_insert_with(|| RepoSyncNeeds {
relays: HashSet::new(),
root_events: HashSet::new(),
@@ -251,9 +256,9 @@ impl SelfSubscriber {
///
/// Returns true if any extracted relay URL contains our domain
fn lists_our_relay(&self, event: &Event) -> bool {
Self::extract_relay_urls(event).iter().any(|url| {
url.contains(&self.relay_domain) || url == &self.own_relay_url
})
Self::extract_relay_urls(event)
.iter()
.any(|url| url.contains(&self.relay_domain) || url == &self.own_relay_url)
}
/// Main run loop
@@ -413,21 +418,21 @@ impl SelfSubscriber {
if let Some(repo_sync) = index.get_mut(&repo_ref) {
// Add event.id to root_events set in the index (immediate availability)
repo_sync.root_events.insert(event.id);
// Clone the relays before releasing the lock - Layer 3 filters need to be
// sent to the same relays as Layer 2 filters for this repo
let relays = repo_sync.relays.clone();
// Release lock before modifying pending
drop(index);
// Also add root event to pending - this ensures batch processing runs
// and creates Layer 3 filters for events referencing this root event.
// CRITICAL: Include relays so derive_relay_targets knows where to send filters!
let mut root_events = HashSet::new();
root_events.insert(event.id);
pending.add_repo(repo_ref.clone(), relays.clone(), root_events);
tracing::debug!(
event_id = %event.id,
repo_ref = %repo_ref,
@@ -475,10 +480,12 @@ impl SelfSubscriber {
for (repo_id, needs) in updates {
// Merge with existing entry or insert new
let entry = index.entry(repo_id.clone()).or_insert_with(|| RepoSyncNeeds {
relays: HashSet::new(),
root_events: HashSet::new(),
});
let entry = index
.entry(repo_id.clone())
.or_insert_with(|| RepoSyncNeeds {
relays: HashSet::new(),
root_events: HashSet::new(),
});
entry.relays.extend(needs.relays);
entry.root_events.extend(needs.root_events);
@@ -556,7 +563,7 @@ fn clone_url_to_relay_url(clone_url: &str) -> Option<String> {
} else {
return None;
};
// Extract just the host:port part (everything before the first /)
let host_port = rest.split('/').next()?;
Some(format!("{}{}", ws_scheme, host_port))
@@ -581,7 +588,7 @@ mod tests {
Some("ws://localhost:3000".to_string())
);
}
#[test]
fn test_clone_url_to_relay_url_with_port() {
assert_eq!(
@@ -593,6 +600,9 @@ mod tests {
#[test]
fn test_clone_url_to_relay_url_unsupported() {
assert_eq!(clone_url_to_relay_url("git://example.com/repo.git"), None);
assert_eq!(clone_url_to_relay_url("ssh://git@example.com/repo.git"), None);
assert_eq!(
clone_url_to_relay_url("ssh://git@example.com/repo.git"),
None
);
}
}
}
tests/common/mod.rs
+2
View File
@@ -1,4 +1,6 @@
//! Common test utilities
#![allow(dead_code)] // Test helpers may not be used in all test configurations
#![allow(unused_imports)] // Re-exports may not be used in all test configurations
pub mod relay;
pub mod sync_helpers;
tests/common/relay.rs
+5 -1
View File
@@ -104,7 +104,11 @@ impl TestRelay {
}
/// Start relay with full options
async fn start_with_full_options(port: u16, bootstrap_relay_url: Option<String>, disable_negentropy: bool) -> Self {
async fn start_with_full_options(
port: u16,
bootstrap_relay_url: Option<String>,
disable_negentropy: bool,
) -> Self {
let bind_address = format!("127.0.0.1:{}", port);
let url = format!("ws://127.0.0.1:{}", port);
tests/common/sync_helpers.rs
+25 -29
View File
@@ -173,7 +173,11 @@ impl TestClient {
/// # Returns
/// * `Ok(Event)` - Signed event ready to send
/// * `Err(String)` - If signing fails
pub fn build_layer2_issue_event(keys: &Keys, repo_coord: &str, title: &str) -> Result<Event, String> {
pub fn build_layer2_issue_event(
keys: &Keys,
repo_coord: &str,
title: &str,
) -> Result<Event, String> {
build_layer2_issue_with_tag(keys, repo_coord, title, TagVariant::LowercaseA)
}
@@ -256,10 +260,7 @@ pub fn build_layer3_comment_event(
// Choose tag based on kind (NIP-22 uses E, NIP-10 style uses e)
let tag = if kind_num == KIND_COMMENT {
// NIP-22 comment: uppercase 'E' tag
Tag::custom(
TagKind::custom("E"),
vec![parent_event_id.to_hex()],
)
Tag::custom(TagKind::custom("E"), vec![parent_event_id.to_hex()])
} else {
// Kind 1 reply: lowercase 'e' tag with root marker (NIP-10)
Tag::custom(
@@ -299,10 +300,7 @@ pub fn build_layer3_comment_with_uppercase_e_tag(
parent_event_id: &EventId,
content: &str,
) -> Result<Event, String> {
let tag = Tag::custom(
TagKind::custom("E"),
vec![parent_event_id.to_hex()],
);
let tag = Tag::custom(TagKind::custom("E"), vec![parent_event_id.to_hex()]);
EventBuilder::new(Kind::Custom(KIND_COMMENT), content)
.tags(vec![tag])
@@ -316,10 +314,7 @@ pub fn build_layer3_quote_with_q_tag(
parent_event_id: &EventId,
content: &str,
) -> Result<Event, String> {
let tag = Tag::custom(
TagKind::custom("q"),
vec![parent_event_id.to_hex()],
);
let tag = Tag::custom(TagKind::custom("q"), vec![parent_event_id.to_hex()]);
EventBuilder::new(Kind::Custom(1), content)
.tags(vec![tag])
@@ -587,10 +582,7 @@ pub fn repo_coord(keys: &Keys, identifier: &str) -> String {
/// ```
pub async fn fetch_metrics(relay_url: &str) -> Result<String, reqwest::Error> {
// Convert ws:// URL to http:// for metrics endpoint
let http_url = relay_url
.replace("ws://", "http://")
.replace("/", "")
+ "/metrics";
let http_url = relay_url.replace("ws://", "http://").replace("/", "") + "/metrics";
reqwest::get(&http_url).await?.text().await
}
@@ -888,8 +880,8 @@ mod tests {
let keys = Keys::generate();
let coord = repo_coord(&keys, "my-repo");
let event = build_layer2_issue_event(&keys, &coord, "Test Issue")
.expect("Should create event");
let event =
build_layer2_issue_event(&keys, &coord, "Test Issue").expect("Should create event");
// nostr-sdk 0.43: use field access
assert_eq!(event.kind.as_u16(), KIND_ISSUE);
@@ -937,8 +929,13 @@ mod tests {
let keys = Keys::generate();
let parent_id = EventId::all_zeros();
let event = build_layer3_comment_event(&keys, &parent_id, "Test comment", Kind::Custom(KIND_COMMENT))
.expect("Should create event");
let event = build_layer3_comment_event(
&keys,
&parent_id,
"Test comment",
Kind::Custom(KIND_COMMENT),
)
.expect("Should create event");
assert_eq!(event.kind.as_u16(), KIND_COMMENT);
@@ -980,8 +977,7 @@ mod tests {
let has_e_tag = event.tags.iter().any(|tag| {
let slice = tag.as_slice();
slice.first().is_some_and(|t| t == "e") &&
slice.get(3).is_some_and(|m| m == "root")
slice.first().is_some_and(|t| t == "e") && slice.get(3).is_some_and(|m| m == "root")
});
assert!(has_e_tag, "Should have 'e' tag with root marker");
}
@@ -1038,7 +1034,10 @@ mod tests {
fn test_parse_gauge_without_labels() {
let text = r#"ngit_sync_relays_tracked_total 3"#;
let metrics = ParsedMetrics::parse(text);
assert_eq!(metrics.gauge("ngit_sync_relays_tracked_total", &[]), Some(3));
assert_eq!(
metrics.gauge("ngit_sync_relays_tracked_total", &[]),
Some(3)
);
}
#[test]
@@ -1051,9 +1050,6 @@ mod tests {
fn test_parse_metric_with_relay_url_label() {
let text = r#"ngit_sync_relay_connected{relay="ws://127.0.0.1:12345"} 1"#;
let metrics = ParsedMetrics::parse(text);
assert_eq!(
metrics.relay_connected("ws://127.0.0.1:12345"),
Some(true)
);
assert_eq!(metrics.relay_connected("ws://127.0.0.1:12345"), Some(true));
}
}
}
tests/nip77_negentropy.rs
+16 -11
View File
@@ -45,13 +45,13 @@ async fn test_nip77_negentropy_sync_finds_events() {
let keys = Keys::generate();
// Create a repository announcement that will be accepted by the relay
let announcement = create_repo_announcement(
&keys,
&[&relay.domain()],
"test-repo-nip77",
);
let announcement = create_repo_announcement(&keys, &[&relay.domain()], "test-repo-nip77");
let event1_id = announcement.id;
println!("Created event 1: {} (kind {})", event1_id, announcement.kind.as_u16());
println!(
"Created event 1: {} (kind {})",
event1_id,
announcement.kind.as_u16()
);
// Create a second event (issue referencing the repo)
let repo_coord = format!(
@@ -63,7 +63,11 @@ async fn test_nip77_negentropy_sync_finds_events() {
let issue = build_layer2_issue_event(&keys, &repo_coord, "Test issue for NIP-77")
.expect("Failed to build issue event");
let event2_id = issue.id;
println!("Created event 2: {} (kind {})", event2_id, issue.kind.as_u16());
println!(
"Created event 2: {} (kind {})",
event2_id,
issue.kind.as_u16()
);
// 3. Send events to relay using TestClient
let publish_client = TestClient::new(relay.url(), keys.clone())
@@ -99,9 +103,10 @@ async fn test_nip77_negentropy_sync_finds_events() {
tokio::time::sleep(Duration::from_millis(500)).await;
// 6. Perform negentropy sync with filter matching our events
let filter = Filter::new()
.author(keys.public_key())
.kinds(vec![Kind::Custom(KIND_REPOSITORY_STATE), Kind::Custom(KIND_ISSUE)]);
let filter = Filter::new().author(keys.public_key()).kinds(vec![
Kind::Custom(KIND_REPOSITORY_STATE),
Kind::Custom(KIND_ISSUE),
]);
println!("Starting negentropy sync with filter: {:?}", filter);
@@ -177,7 +182,7 @@ async fn test_nip77_negentropy_sync_empty_result() {
// 3. Sync with filter that won't match anything
let filter = Filter::new()
.author(keys.public_key()) // Random new key, no events exist
.author(keys.public_key()) // Random new key, no events exist
.kind(Kind::Custom(KIND_REPOSITORY_STATE));
println!("Starting negentropy sync with empty filter");
tests/sync.rs
+1 -1
View File
@@ -37,4 +37,4 @@ mod sync {
pub mod live_sync;
pub mod metrics;
pub mod tag_variations;
}
}
tests/sync/bootstrap.rs
+6 -3
View File
@@ -167,7 +167,8 @@ async fn test_relay_replays_events_after_restart() {
.kind(Kind::Custom(KIND_REPOSITORY_STATE))
.author(keys.public_key());
let synced_first = wait_for_event_on_relay(relay_b.url(), filter.clone(), Duration::from_secs(5)).await;
let synced_first =
wait_for_event_on_relay(relay_b.url(), filter.clone(), Duration::from_secs(5)).await;
println!("First sync check: {}", synced_first);
// 8. Stop relay_b
@@ -193,7 +194,8 @@ async fn test_relay_replays_events_after_restart() {
// 12. Verify announcement is available on new relay_b
// The announcement listed the OLD relay_b domain, but since relay_a still
// has the event, new relay_b should be able to sync it via bootstrap
let synced_after_restart = wait_for_event_on_relay(relay_b_new.url(), filter, Duration::from_secs(5)).await;
let synced_after_restart =
wait_for_event_on_relay(relay_b_new.url(), filter, Duration::from_secs(5)).await;
// 13. Cleanup
relay_b_new.stop().await;
@@ -384,7 +386,8 @@ async fn test_history_sync_without_negentropy() {
relay_b_port,
Some(relay_a.url().into()),
true, // disable_negentropy = true
).await;
)
.await;
println!(
"relay_b started at {} (domain: {}) - negentropy DISABLED, will do HISTORY sync",
relay_b.url(),
tests/sync/discovery.rs
+10 -5
View File
@@ -88,7 +88,8 @@ async fn test_discovers_layer3_via_layer2() {
);
// 6. Create a patch event (Layer 2) that references the announcement
let patch = create_event_referencing_repo(&keys, &repo_coord, KIND_PATCH, "Test patch proposal");
let patch =
create_event_referencing_repo(&keys, &repo_coord, KIND_PATCH, "Test patch proposal");
let patch_id = patch.id;
println!("Created patch {} (kind {})", patch_id, patch.kind.as_u16());
@@ -252,7 +253,8 @@ async fn test_layer2_discovery_with_chain() {
let issue_filter = Filter::new()
.kind(Kind::Custom(KIND_ISSUE))
.author(keys.public_key());
let issue_synced = wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
let issue_synced =
wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
println!("Sync result:");
println!(" Issue {} synced: {}", issue_id, issue_synced);
@@ -296,7 +298,7 @@ async fn test_layer2_discovery_with_chain() {
#[tokio::test]
async fn test_recursive_relay_discovery_syncs_announcement() {
// 1. Start all three relays
// relay_b - will be the bootstrap relay, has announcement_x
let relay_b = TestRelay::start().await;
println!(
@@ -344,7 +346,10 @@ async fn test_recursive_relay_discovery_syncs_announcement() {
"repo-y-ac-only",
);
let announcement_y_id = announcement_y.id;
println!("Created announcement_y {} listing A+C only", announcement_y_id);
println!(
"Created announcement_y {} listing A+C only",
announcement_y_id
);
for tag in announcement_y.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -425,4 +430,4 @@ async fn test_recursive_relay_discovery_syncs_announcement() {
"announcement_y {} should have synced from discovered relay_c to relay_a (recursive discovery)",
announcement_y_id
);
}
}
tests/sync/live_sync.rs
+5 -2
View File
@@ -229,7 +229,10 @@ async fn test_live_sync_layer3_events() {
.send_event(&comment)
.await
.expect("Failed to send comment");
println!("Layer 3 comment {} sent to relay_a BEFORE Layer 3 subscription established", comment_id);
println!(
"Layer 3 comment {} sent to relay_a BEFORE Layer 3 subscription established",
comment_id
);
// 6. Now wait for issue to sync to relay_b (this triggers Layer 3 filter creation)
tokio::time::sleep(Duration::from_secs(2)).await;
@@ -394,7 +397,7 @@ async fn test_live_sync_event_ordering() {
client_a
.send_event(&issue)
.await
.expect(&format!("Failed to send issue {}", i));
.unwrap_or_else(|_| panic!("Failed to send issue {}", i));
// Delay between events to ensure different timestamps
tokio::time::sleep(Duration::from_millis(150)).await;
tests/sync/metrics.rs
+3 -3
View File
@@ -32,7 +32,7 @@ async fn test_prometheus_format_valid() {
let relay = TestRelay::start().await;
tokio::time::sleep(Duration::from_millis(500)).await;
let metrics = fetch_metrics(&relay.url())
let metrics = fetch_metrics(relay.url())
.await
.expect("Failed to fetch metrics");
@@ -67,7 +67,7 @@ async fn test_metrics_availability_during_sync() {
// Make multiple metrics requests while sync is active
for i in 0..3 {
let metrics = fetch_metrics(&sync_relay.url()).await;
let metrics = fetch_metrics(sync_relay.url()).await;
assert!(
metrics.is_ok(),
"Metrics request {} should succeed during sync",
@@ -135,7 +135,7 @@ async fn test_metric_values_are_numeric() {
let relay = TestRelay::start().await;
tokio::time::sleep(Duration::from_millis(500)).await;
let metrics = fetch_metrics(&relay.url())
let metrics = fetch_metrics(relay.url())
.await
.expect("Should fetch metrics");
tests/sync/tag_variations.rs
+77 -62
View File
@@ -57,11 +57,8 @@ async fn test_layer2_sync_with_lowercase_a_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-8a";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -88,11 +85,16 @@ async fn test_layer2_sync_with_lowercase_a_tag() {
// 4. Create and send Layer 2 issue with lowercase 'a' tag
let repo_coordinate = repo_coord(&keys, repo_id);
let issue = build_layer2_issue_event(&keys, &repo_coordinate, "Test Issue with lowercase a tag")
.expect("Failed to create issue event");
let issue =
build_layer2_issue_event(&keys, &repo_coordinate, "Test Issue with lowercase a tag")
.expect("Failed to create issue event");
let issue_id = issue.id;
println!("Created issue {} (kind {}) with lowercase 'a' tag", issue_id, issue.kind.as_u16());
println!(
"Created issue {} (kind {}) with lowercase 'a' tag",
issue_id,
issue.kind.as_u16()
);
for tag in issue.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -154,11 +156,8 @@ async fn test_layer2_sync_with_uppercase_a_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-8b";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -185,11 +184,19 @@ async fn test_layer2_sync_with_uppercase_a_tag() {
// 4. Create and send Layer 2 issue with uppercase 'A' tag
let repo_coordinate = repo_coord(&keys, repo_id);
let issue = build_layer2_issue_with_uppercase_a_tag(&keys, &repo_coordinate, "Test Issue with uppercase A tag")
.expect("Failed to create issue event");
let issue = build_layer2_issue_with_uppercase_a_tag(
&keys,
&repo_coordinate,
"Test Issue with uppercase A tag",
)
.expect("Failed to create issue event");
let issue_id = issue.id;
println!("Created issue {} (kind {}) with uppercase 'A' tag", issue_id, issue.kind.as_u16());
println!(
"Created issue {} (kind {}) with uppercase 'A' tag",
issue_id,
issue.kind.as_u16()
);
for tag in issue.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -250,11 +257,8 @@ async fn test_layer2_sync_with_q_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-8c";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -285,7 +289,11 @@ async fn test_layer2_sync_with_q_tag() {
.expect("Failed to create issue event");
let issue_id = issue.id;
println!("Created issue {} (kind {}) with 'q' tag", issue_id, issue.kind.as_u16());
println!(
"Created issue {} (kind {}) with 'q' tag",
issue_id,
issue.kind.as_u16()
);
for tag in issue.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -350,11 +358,8 @@ async fn test_layer3_sync_with_lowercase_e_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-9a";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -392,10 +397,9 @@ async fn test_layer3_sync_with_lowercase_e_tag() {
println!("Layer 2 issue {} sent to relay_a", issue_id);
// 5. Wait for issue to sync to relay_b
let issue_filter = Filter::new()
.kind(Kind::Custom(KIND_ISSUE))
.id(issue_id);
let issue_synced = wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
let issue_filter = Filter::new().kind(Kind::Custom(KIND_ISSUE)).id(issue_id);
let issue_synced =
wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
println!("Issue synced to relay_b: {}", issue_synced);
assert!(issue_synced, "Layer 2 issue should sync first");
@@ -412,7 +416,11 @@ async fn test_layer3_sync_with_lowercase_e_tag() {
.expect("Failed to create reply");
let reply_id = reply.id;
println!("Created reply {} (kind {}) with lowercase 'e' tag", reply_id, reply.kind.as_u16());
println!(
"Created reply {} (kind {}) with lowercase 'e' tag",
reply_id,
reply.kind.as_u16()
);
for tag in reply.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -428,11 +436,12 @@ async fn test_layer3_sync_with_lowercase_e_tag() {
// 7. Wait and verify reply syncs to relay_b
let reply_filter = Filter::new()
.kind(Kind::TextNote) // Kind 1
.kind(Kind::TextNote) // Kind 1
.author(keys.public_key())
.id(reply_id);
let reply_synced = wait_for_event_on_relay(relay_b.url(), reply_filter, Duration::from_secs(5)).await;
let reply_synced =
wait_for_event_on_relay(relay_b.url(), reply_filter, Duration::from_secs(5)).await;
println!("Reply {} synced to relay_b: {}", reply_id, reply_synced);
@@ -473,11 +482,8 @@ async fn test_layer3_sync_with_uppercase_e_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-9b";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -515,10 +521,9 @@ async fn test_layer3_sync_with_uppercase_e_tag() {
println!("Layer 2 issue {} sent to relay_a", issue_id);
// 5. Wait for issue to sync to relay_b
let issue_filter = Filter::new()
.kind(Kind::Custom(KIND_ISSUE))
.id(issue_id);
let issue_synced = wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
let issue_filter = Filter::new().kind(Kind::Custom(KIND_ISSUE)).id(issue_id);
let issue_synced =
wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
println!("Issue synced to relay_b: {}", issue_synced);
assert!(issue_synced, "Layer 2 issue should sync first");
@@ -531,11 +536,16 @@ async fn test_layer3_sync_with_uppercase_e_tag() {
tokio::time::sleep(Duration::from_millis(500)).await;
// 6. Create and send Layer 3 comment with uppercase 'E' tag (kind 1111)
let comment = build_layer3_comment_with_uppercase_e_tag(&keys, &issue_id, "Comment with uppercase E tag")
.expect("Failed to create comment");
let comment =
build_layer3_comment_with_uppercase_e_tag(&keys, &issue_id, "Comment with uppercase E tag")
.expect("Failed to create comment");
let comment_id = comment.id;
println!("Created comment {} (kind {}) with uppercase 'E' tag", comment_id, comment.kind.as_u16());
println!(
"Created comment {} (kind {}) with uppercase 'E' tag",
comment_id,
comment.kind.as_u16()
);
for tag in comment.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -551,13 +561,17 @@ async fn test_layer3_sync_with_uppercase_e_tag() {
// 7. Wait and verify comment syncs to relay_b
let comment_filter = Filter::new()
.kind(Kind::Custom(KIND_COMMENT)) // Kind 1111
.kind(Kind::Custom(KIND_COMMENT)) // Kind 1111
.author(keys.public_key())
.id(comment_id);
let comment_synced = wait_for_event_on_relay(relay_b.url(), comment_filter, Duration::from_secs(5)).await;
let comment_synced =
wait_for_event_on_relay(relay_b.url(), comment_filter, Duration::from_secs(5)).await;
println!("Comment {} synced to relay_b: {}", comment_id, comment_synced);
println!(
"Comment {} synced to relay_b: {}",
comment_id, comment_synced
);
// 8. Cleanup
relay_b.stop().await;
@@ -596,11 +610,8 @@ async fn test_layer3_sync_with_q_tag() {
// 2. Create and send repository announcement to both relays
let repo_id = "test-repo-tag-9c";
let announcement = create_repo_announcement(
&keys,
&[&relay_a.domain(), &relay_b.domain()],
repo_id,
);
let announcement =
create_repo_announcement(&keys, &[&relay_a.domain(), &relay_b.domain()], repo_id);
let client_a = TestClient::new(relay_a.url(), keys.clone())
.await
@@ -638,10 +649,9 @@ async fn test_layer3_sync_with_q_tag() {
println!("Layer 2 issue {} sent to relay_a", issue_id);
// 5. Wait for issue to sync to relay_b
let issue_filter = Filter::new()
.kind(Kind::Custom(KIND_ISSUE))
.id(issue_id);
let issue_synced = wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
let issue_filter = Filter::new().kind(Kind::Custom(KIND_ISSUE)).id(issue_id);
let issue_synced =
wait_for_event_on_relay(relay_b.url(), issue_filter, Duration::from_secs(5)).await;
println!("Issue synced to relay_b: {}", issue_synced);
assert!(issue_synced, "Layer 2 issue should sync first");
@@ -658,7 +668,11 @@ async fn test_layer3_sync_with_q_tag() {
.expect("Failed to create quote");
let quote_id = quote.id;
println!("Created quote {} (kind {}) with 'q' tag", quote_id, quote.kind.as_u16());
println!(
"Created quote {} (kind {}) with 'q' tag",
quote_id,
quote.kind.as_u16()
);
for tag in quote.tags.iter() {
println!(" Tag: {:?}", tag.as_slice());
}
@@ -674,11 +688,12 @@ async fn test_layer3_sync_with_q_tag() {
// 7. Wait and verify quote syncs to relay_b
let quote_filter = Filter::new()
.kind(Kind::TextNote) // Kind 1
.kind(Kind::TextNote) // Kind 1
.author(keys.public_key())
.id(quote_id);
let quote_synced = wait_for_event_on_relay(relay_b.url(), quote_filter, Duration::from_secs(5)).await;
let quote_synced =
wait_for_event_on_relay(relay_b.url(), quote_filter, Duration::from_secs(5)).await;
println!("Quote {} synced to relay_b: {}", quote_id, quote_synced);
@@ -690,4 +705,4 @@ async fn test_layer3_sync_with_q_tag() {
quote_synced,
"Layer 3 quote with 'q' tag should have synced to relay_b"
);
}
}