mirror of
https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git
synced 2026-10-05 15:08:24 +00:00
DanConwayDev
ed8bdae231
Repeated cold syncs reported nearly identical saved totals because the accepted-event persistence facade discarded SaveEventStatus. A stale replaceable event rejected as Replaced was consequently counted as a new insert, broadcast, and allowed to trigger downstream dependency work.\n\nReturn a bounded saved-or-duplicate outcome from the central facade. Sync now reports Duplicate for database Duplicate/Replaced results and broadcasts or expands dependencies only after SaveEventStatus::Success. Hot-cache promotion likewise avoids broadcasting superseded events while retaining the dependency retry needed for an already-stored maintainer announcement.\n\nThis deliberately does not change write-policy admission, database replacement semantics, or the historical mailbox filters. A focused regression preloads a newer replaceable event and verifies that syncing its predecessor is classified as duplicate without displacing the stored event.\n\nValidation: git diff --check. The focused test and broader suite were not executed in this session per operator instruction.
1221 lines
51 KiB
Rust
1221 lines
51 KiB
Rust
/// Nostr Relay Builder Configuration
|
||
///
|
||
/// Wires nostr-relay-builder with NIP-34 admission/routing policy.
|
||
/// Deletion behaviour is owned by the `nostr::lifecycle` subsystem;
|
||
/// this module keeps top-level `admit_event`
|
||
/// dispatch and non-deletion admission/router policy logic.
|
||
use std::future::Future;
|
||
use std::net::SocketAddr;
|
||
use std::num::NonZeroUsize;
|
||
use std::path::Path;
|
||
use std::pin::Pin;
|
||
use std::sync::{Arc, RwLock};
|
||
use std::time::Duration;
|
||
|
||
use anyhow::Result;
|
||
use nostr::nips::nip19::ToBech32;
|
||
use nostr_lmdb::NostrLmdb;
|
||
use nostr_memory::MemoryDatabase;
|
||
use nostr_sdk::prelude::*;
|
||
|
||
use crate::config::{Config, DatabaseBackend};
|
||
use crate::nostr::events::RepositoryAnnouncement;
|
||
use crate::nostr::lifecycle::{
|
||
DeletionContext, DeletionRuntime, DeletionService, HoldingStore, ReplaceableHistoryStore,
|
||
RepositoryLifecycle, Tombstones,
|
||
};
|
||
use crate::nostr::persistence::{AcceptedEventSaveOutcome, EventPersistence, SaveContext};
|
||
use crate::nostr::policy::{
|
||
accepted_purgatory, duplicate, reject_error, reject_invalid, reject_restricted,
|
||
AnnouncementPolicy, AnnouncementResult, IdentityAdmission, PolicyContext, PrEventPolicy,
|
||
ReferenceResult, RelatedEventPolicy, SharedProactiveParticipantAuthorIndex, StatePolicy,
|
||
StateResult,
|
||
};
|
||
use crate::nostr::SharedDatabase;
|
||
use crate::private::PrivateAccess;
|
||
use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks;
|
||
use crate::sync::rejected_index::RejectedEventsIndex;
|
||
|
||
/// Connection-wide frame allowance layered above the operation-specific
|
||
/// write, query, and authentication quotas.
|
||
///
|
||
/// rust-nostr's 300/minute default closed production browser clients during
|
||
/// ordinary bursty subscription churn. One hundred frames per second preserves a
|
||
/// bounded catch-all for malformed/non-operation traffic while leaving the
|
||
/// tighter operation quotas in charge of valid protocol work.
|
||
const CLIENT_MESSAGES_PER_MINUTE: u32 = 6_000;
|
||
|
||
/// Temporary compatibility override for rust-nostr 0.45's newly introduced
|
||
/// 120-query-per-minute default.
|
||
///
|
||
/// A finite per-connection bound remains useful as one DoS layer, but 120 is
|
||
/// too restrictive for legitimate sync and NIP-77 currently charges every
|
||
/// SDK-managed `NEG-MSG` continuation against the same allowance. Re-evaluate
|
||
/// this 10× value after upstream separates or otherwise revises that accounting.
|
||
const CLIENT_QUERIES_PER_MINUTE: u32 = 1_200;
|
||
|
||
/// Cumulative serialized REQ state retained for one client connection.
|
||
///
|
||
/// Repository sync legitimately keeps multiple byte-budgeted live filters
|
||
/// open at once. Production's 34-filter coverage reached about 1.2 MiB and
|
||
/// was partially rejected by rust-nostr's 1 MiB default. Five MiB remains a
|
||
/// finite per-connection allocation boundary while matching the largest
|
||
/// individual WebSocket message we already admit and leaving roughly 4x room
|
||
/// above the observed working set.
|
||
const MAX_SUBSCRIPTION_STATE_BYTES: usize = 5 * 1024 * 1024;
|
||
|
||
/// NIP-34 Write Policy — admission and routing for GRASP-01 events
|
||
///
|
||
/// Acts as the top-level admission gate and router. Each incoming event is:
|
||
/// 1. Checked against the event blacklist (skipped for the trusted
|
||
/// relay-owner key).
|
||
/// 2. Passed through the deletion gate (`DeletionService::gate`), which
|
||
/// applies to every author — including the relay owner — so replaying a
|
||
/// retracted event cannot undo its NIP-09/NIP-62 tombstone.
|
||
/// 3. Dispatched to the appropriate sub-policy. Relay-owner trust is scoped:
|
||
/// owner-signed events of kinds with a dedicated sub-policy below (NIP-34
|
||
/// repository kinds, deletion/vanish requests) take the same validation
|
||
/// paths as any other author, while the relay's own kind 0/10002 identity
|
||
/// and owner-signed kinds with no dedicated policy (e.g. `ngit-ci`
|
||
/// coordinator advertisements) are accepted instead of falling through to
|
||
/// the generic related-event rejection.
|
||
/// - `AnnouncementPolicy` — repository announcement validation
|
||
/// - `StatePolicy` — state event validation + ref alignment
|
||
/// - `PrEventPolicy` — PR / PR-Update validation
|
||
/// - `RelatedEventPolicy` — forward/backward reference checking
|
||
/// - `DeletionService` — NIP-09 deletion and NIP-62 vanish handling
|
||
///
|
||
/// Deletion logic (cascade, holding, archive, recovery, startup passes) lives
|
||
/// entirely in `crate::nostr::lifecycle`. This struct exposes a `deletion()`
|
||
/// accessor for callers that need to reach those operations directly.
|
||
#[derive(Clone)]
|
||
pub struct Nip34WritePolicy {
|
||
ctx: PolicyContext,
|
||
relay_owner_pubkey: Option<PublicKey>,
|
||
announcement_policy: AnnouncementPolicy,
|
||
state_policy: StatePolicy,
|
||
pr_event_policy: PrEventPolicy,
|
||
related_event_policy: RelatedEventPolicy,
|
||
proactive_participant_authors: SharedProactiveParticipantAuthorIndex,
|
||
deletion: DeletionService,
|
||
rejected_events_index: Arc<RwLock<Option<Arc<RejectedEventsIndex>>>>,
|
||
}
|
||
|
||
impl std::fmt::Debug for Nip34WritePolicy {
|
||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||
f.debug_struct("Nip34WritePolicy")
|
||
.field("domain", &self.ctx.domain)
|
||
.field("git_data_path", &self.ctx.git_data_path)
|
||
.field("database", &"<database>")
|
||
.finish()
|
||
}
|
||
}
|
||
|
||
impl Nip34WritePolicy {
|
||
#[allow(clippy::too_many_arguments)]
|
||
pub fn new(
|
||
database: SharedDatabase,
|
||
tombstones: Tombstones,
|
||
holding: HoldingStore,
|
||
lifecycle: RepositoryLifecycle,
|
||
history: ReplaceableHistoryStore,
|
||
git_data_path: impl Into<std::path::PathBuf>,
|
||
purgatory: std::sync::Arc<crate::purgatory::Purgatory>,
|
||
config: crate::config::Config,
|
||
repo_init_locks: crate::grasp06::receive::RepoInitLocks,
|
||
private_access: Option<PrivateAccess>,
|
||
) -> Self {
|
||
let git_data_path = git_data_path.into();
|
||
let domain = config.service_address();
|
||
let deletion_ctx = DeletionContext::new(
|
||
domain.clone(),
|
||
database.clone(),
|
||
tombstones.clone(),
|
||
holding.clone(),
|
||
lifecycle,
|
||
history.clone(),
|
||
git_data_path.clone(),
|
||
purgatory.clone(),
|
||
config.clone(),
|
||
repo_init_locks.clone(),
|
||
);
|
||
|
||
let ctx = PolicyContext::new(domain, database, git_data_path, purgatory, config.clone());
|
||
Self {
|
||
// RelayServer validates and always supplies this key. Keeping it
|
||
// optional here preserves library/test callers that construct a
|
||
// standalone policy without running the server configuration
|
||
// loader first.
|
||
relay_owner_pubkey: config.relay_owner_keys().ok().map(|keys| keys.public_key()),
|
||
announcement_policy: AnnouncementPolicy::new(
|
||
ctx.clone(),
|
||
config.clone(),
|
||
private_access,
|
||
),
|
||
state_policy: StatePolicy::new(ctx.clone()),
|
||
pr_event_policy: PrEventPolicy::new(ctx.clone(), repo_init_locks),
|
||
related_event_policy: RelatedEventPolicy::new(ctx.clone()),
|
||
proactive_participant_authors:
|
||
crate::nostr::policy::ProactiveParticipantAuthorIndex::shared(),
|
||
deletion: DeletionService::new(deletion_ctx),
|
||
rejected_events_index: Arc::new(RwLock::new(None)),
|
||
ctx,
|
||
}
|
||
}
|
||
|
||
pub fn proactive_participant_authors(&self) -> SharedProactiveParticipantAuthorIndex {
|
||
self.proactive_participant_authors.clone()
|
||
}
|
||
|
||
async fn handle_proactive_identity(&self, event: &Event) -> WritePolicyResult {
|
||
match self.proactive_participant_authors.admit(event).await {
|
||
IdentityAdmission::Accept => WritePolicyResult::Accept,
|
||
IdentityAdmission::IrrelevantAuthor => reject_restricted(
|
||
"Kind 0/10002 author must participate in an accepted repository thread",
|
||
),
|
||
IdentityAdmission::NotIdentity => {
|
||
reject_invalid("Proactive identity policy received a non-identity event")
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Check if an event author is blacklisted
|
||
///
|
||
/// Returns Some(reason) if blacklisted, None if not blacklisted.
|
||
fn check_event_blacklist(&self, event: &Event) -> Option<String> {
|
||
let event_blacklist = self.ctx.config.event_blacklist_config();
|
||
if !event_blacklist.enabled() {
|
||
return None;
|
||
}
|
||
|
||
let npub = event.pubkey.to_bech32().ok()?;
|
||
event_blacklist.check(&npub)
|
||
}
|
||
|
||
/// Get a reference to the purgatory for read-only access
|
||
pub fn purgatory(&self) -> &std::sync::Arc<crate::purgatory::Purgatory> {
|
||
&self.ctx.purgatory
|
||
}
|
||
|
||
/// Get a reference to the holding store for retention cleanup orchestration.
|
||
pub fn holding(&self) -> &HoldingStore {
|
||
self.deletion.holding()
|
||
}
|
||
|
||
/// Get a reference to the replaceable-history store.
|
||
pub fn history(&self) -> &ReplaceableHistoryStore {
|
||
self.deletion.history()
|
||
}
|
||
|
||
pub fn tombstones(&self) -> &Tombstones {
|
||
self.deletion.tombstones()
|
||
}
|
||
|
||
/// Set the local relay for purgatory notifications.
|
||
///
|
||
/// This must be called after the relay is created since the relay depends
|
||
/// on this policy, but purgatory sync needs the relay to notify subscribers.
|
||
pub fn set_local_relay(&self, relay: nostr_sdk::local_relay::LocalRelay) {
|
||
self.ctx.set_local_relay(relay);
|
||
}
|
||
|
||
/// Make sync's rejected-event dependencies available to direct-write
|
||
/// purgatory promotions.
|
||
pub fn set_rejected_events_index(&self, index: Arc<RejectedEventsIndex>) {
|
||
*self
|
||
.rejected_events_index
|
||
.write()
|
||
.expect("rejected events index lock poisoned") = Some(index);
|
||
}
|
||
|
||
pub(crate) fn rejected_events_index(&self) -> Option<Arc<RejectedEventsIndex>> {
|
||
self.rejected_events_index
|
||
.read()
|
||
.expect("rejected events index lock poisoned")
|
||
.clone()
|
||
}
|
||
|
||
pub(crate) fn local_relay(&self) -> Option<nostr_sdk::local_relay::LocalRelay> {
|
||
self.ctx.get_local_relay()
|
||
}
|
||
|
||
/// Access the deletion subsystem directly.
|
||
///
|
||
/// Exposes startup passes and recovery without requiring delegation wrappers
|
||
/// on this struct. Repository lifecycle coordination is exposed separately
|
||
/// through [`RepositoryLifecycle`].
|
||
pub fn deletion(&self) -> &DeletionService {
|
||
&self.deletion
|
||
}
|
||
|
||
fn event_persistence(&self) -> EventPersistence<'_> {
|
||
EventPersistence::new(&self.ctx.database)
|
||
}
|
||
|
||
/// Save an already-admitted event through the central accepted-event path.
|
||
pub async fn save_accepted_event(
|
||
&self,
|
||
event: &Event,
|
||
context: SaveContext,
|
||
) -> anyhow::Result<AcceptedEventSaveOutcome> {
|
||
self.event_persistence()
|
||
.save_accepted_event(event, context)
|
||
.await
|
||
}
|
||
|
||
/// Extract repository identifier from event's 'd' tag.
|
||
///
|
||
/// Used for structured logging when parsing fails - we try to extract
|
||
/// the identifier even if full parsing failed.
|
||
fn extract_identifier_from_event(event: &Event) -> String {
|
||
event
|
||
.tags
|
||
.iter()
|
||
.find(|t| t.kind() == "d")
|
||
.and_then(|t| t.content())
|
||
.map(|s| s.to_string())
|
||
.unwrap_or_else(|| "unknown".to_string())
|
||
}
|
||
|
||
/// Extract ALL repository identifiers from PR event's 'a' tags.
|
||
///
|
||
/// PR events can reference multiple repositories via multiple 'a' tags
|
||
/// (e.g., when there are multiple maintainers). Each tag has format
|
||
/// `30617:<owner_pubkey>:<identifier>`.
|
||
///
|
||
/// Returns a vector of unique identifiers, or `["unknown"]` if none found.
|
||
fn extract_repos_from_pr_event(event: &Event) -> Vec<String> {
|
||
let repos: Vec<String> = event
|
||
.tags
|
||
.iter()
|
||
.filter_map(|tag| {
|
||
let tag_vec = tag.clone().to_vec();
|
||
if tag_vec.len() >= 2 && tag_vec[0] == "a" && tag_vec[1].starts_with("30617:") {
|
||
// Format: 30617:<owner_pubkey>:<identifier>
|
||
let parts: Vec<&str> = tag_vec[1].split(':').collect();
|
||
if parts.len() >= 3 {
|
||
Some(parts[2].to_string())
|
||
} else {
|
||
None
|
||
}
|
||
} else {
|
||
None
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
// Deduplicate while preserving order
|
||
let mut seen = std::collections::HashSet::new();
|
||
let unique_repos: Vec<String> = repos
|
||
.into_iter()
|
||
.filter(|r| seen.insert(r.clone()))
|
||
.collect();
|
||
|
||
if unique_repos.is_empty() {
|
||
vec!["unknown".to_string()]
|
||
} else {
|
||
unique_repos
|
||
}
|
||
}
|
||
|
||
/// Handle repository announcement event
|
||
async fn handle_announcement(&self, event: &Event) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
let admission = self.announcement_policy.validate(event).await;
|
||
let deletion_hooks = self.deletion.admission_hooks();
|
||
let deletion_hook_result = deletion_hooks
|
||
.on_announcement_admission(event, &admission)
|
||
.await;
|
||
|
||
match admission {
|
||
AnnouncementResult::Accept | AnnouncementResult::AcceptArchive => {
|
||
// Parse announcement to get repository details
|
||
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)
|
||
{
|
||
tracing::warn!(
|
||
"Failed to create bare repository for {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
// Note: We still accept the event even if repo creation fails
|
||
}
|
||
|
||
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) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::AcceptPurgatory => {
|
||
let announcement = match RepositoryAnnouncement::from_event(event.clone()) {
|
||
Ok(announcement) => announcement,
|
||
Err(e) => {
|
||
return reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
};
|
||
|
||
if deletion_hook_result.recovered_repository() {
|
||
tracing::info!(
|
||
identifier = %announcement.identifier,
|
||
event_id = %event_id_str,
|
||
"Accepted re-announcement with recovery"
|
||
);
|
||
if let Err(e) = self
|
||
.announcement_policy
|
||
.ensure_bare_repository(&announcement)
|
||
{
|
||
tracing::warn!(
|
||
"Failed to ensure bare repository after recovery for {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
}
|
||
return WritePolicyResult::Accept;
|
||
}
|
||
|
||
// New announcement - add to purgatory
|
||
match self.announcement_policy.add_to_purgatory(event) {
|
||
Ok(()) => {
|
||
// The server may already have the repository's owner-authored state
|
||
// and Git objects. Reapply that stored state now so a newly accepted
|
||
// maintainer announcement can be promoted without waiting for the
|
||
// same replaceable state event to arrive over sync again.
|
||
self.reconcile_stored_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
tracing::info!(
|
||
"Accepted announcement to purgatory: {} (waiting for git data)",
|
||
event_id_str
|
||
);
|
||
accepted_purgatory("won't be served until git data arrives")
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
"Failed to add announcement to purgatory {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
reject_error(e)
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::AcceptMaintainer => {
|
||
// Parse announcement to get details for logging
|
||
match RepositoryAnnouncement::from_event(event.clone()) {
|
||
Ok(announcement) => {
|
||
tracing::info!(
|
||
"Accepted maintainer announcement {} (author {} is listed as maintainer for {})",
|
||
event_id_str,
|
||
event.pubkey.to_hex(),
|
||
announcement.identifier
|
||
);
|
||
// Don't create bare repository for external announcements
|
||
|
||
// Persist the announcement before re-evaluating state:
|
||
// membership is computed from the database and this may
|
||
// be an invited maintainer's acceptance. The relay
|
||
// builder's later save is an idempotent duplicate.
|
||
if let Err(e) = self.ctx.database.save_event(event).await {
|
||
tracing::warn!(
|
||
event_id = %event_id_str,
|
||
error = %e,
|
||
"Failed to pre-save maintainer announcement before reconciliation"
|
||
);
|
||
}
|
||
|
||
// Check purgatory for state events that might now be authorized
|
||
self.check_purgatory_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
// An acceptance can make already-stored state events
|
||
// authoritative for additional owner repositories, so
|
||
// reapply stored states with the updated member set.
|
||
self.reconcile_stored_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
WritePolicyResult::Accept
|
||
}
|
||
Err(e) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::Reject(reason) => {
|
||
tracing::debug!(
|
||
event_id = %event_id_str,
|
||
reason = %reason,
|
||
"Rejected repository announcement"
|
||
);
|
||
reject_invalid(reason)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle repository state event
|
||
///
|
||
/// # Arguments
|
||
/// * `event` - The state event to validate
|
||
/// * `is_synced` - True if this event came from proactive sync (vs user-submitted)
|
||
async fn handle_state(&self, event: &Event, is_synced: bool) -> WritePolicyResult {
|
||
match self.state_policy.validate(event) {
|
||
StateResult::Accept => {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
|
||
// Process state alignment asynchronously
|
||
match self
|
||
.state_policy
|
||
.process_state_event(event, is_synced, false, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
Ok(policy_result) => {
|
||
self.deletion
|
||
.admission_hooks()
|
||
.on_state_admission(event, &policy_result)
|
||
.await;
|
||
policy_result
|
||
}
|
||
Err(e) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
// reject if processing failed
|
||
reject_error(format!("{e}"))
|
||
}
|
||
}
|
||
}
|
||
StateResult::Reject(reason) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
reason,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(reason)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle PR or PR Update event
|
||
///
|
||
/// # Arguments
|
||
/// * `event` - The PR event to validate
|
||
/// * `is_synced` - True if this event came from proactive sync (vs user-submitted)
|
||
async fn handle_pr_event(&self, event: &Event, is_synced: bool) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
// duplicate check in purgatory
|
||
let in_purgatory = self
|
||
.ctx
|
||
.purgatory
|
||
.find_pr(&event.id.to_hex())
|
||
.is_some_and(|e| e.event.is_some());
|
||
if in_purgatory {
|
||
tracing::debug!(
|
||
"processed PR event duplicate (already in purgatory): {}",
|
||
event.id,
|
||
);
|
||
return duplicate("in purgatory");
|
||
}
|
||
|
||
// duplicate check in db
|
||
//
|
||
// Note: the database runs with `process_nip09(false)`, so `check_id`
|
||
// only ever returns `Saved` / `NotExistent` here — re-submission of a
|
||
// deleted PR is already rejected upstream by `DeletionService::gate`.
|
||
match &self.ctx.database.check_id(&event.id).await {
|
||
Ok(DatabaseEventStatus::Saved) => {
|
||
return duplicate("already have this event");
|
||
}
|
||
Err(e) => {
|
||
return reject_error(format!("internal error: {e}"));
|
||
}
|
||
_ => {} // continue
|
||
}
|
||
|
||
// Reject PRs unrelated to stored repositories / events.
|
||
//
|
||
// GRASP-06 (06.md lines 19–22) relaxes this: a PR / PR-Update event
|
||
// whose `clone` tag names this relay's
|
||
// `/prs/<signer-npub>/<d>.git` endpoint (and which carries a
|
||
// matching `a` tag of the form `30617:<hex>:<d>`) MUST be accepted
|
||
// even when no announcement for the target coord is on this relay.
|
||
// Such events skip the related-event check and fall through to the
|
||
// git-data-check + purgatory branch below.
|
||
match self.handle_related_event(event, "PR").await {
|
||
WritePolicyResult::Accept => {} // continue
|
||
rejected => {
|
||
if !crate::grasp06::policy::event_qualifies_for_pr_relaxation(
|
||
event,
|
||
&self.ctx.config,
|
||
) {
|
||
return rejected;
|
||
}
|
||
tracing::info!(
|
||
"Accepted orphan PR event {} under GRASP-06: clone tag names this relay's /prs/ endpoint",
|
||
event_id_str,
|
||
);
|
||
}
|
||
}
|
||
|
||
// Check if git data exists (delete any incorrect commits at refs/nostr/<event-id>, copies correct data to relivant repositories)
|
||
match self.pr_event_policy.git_data_check(event).await {
|
||
Ok(false) => {
|
||
// Only reject expired events if they're from sync (not user-submitted)
|
||
// User-submitted events should be allowed to retry in case git data became available
|
||
if is_synced && self.ctx.purgatory.is_expired(&event.id) {
|
||
tracing::debug!(
|
||
event_id = %event_id_str,
|
||
"PR event previously expired from purgatory (synced), rejecting to prevent re-sync loop"
|
||
);
|
||
return reject_invalid("previously expired from purgatory without git data");
|
||
}
|
||
|
||
// No git data exists - add to purgatory
|
||
let commit = event
|
||
.tags
|
||
.iter()
|
||
.find_map(|tag| {
|
||
let tag_vec = tag.clone().to_vec();
|
||
if tag_vec.len() >= 2 && tag_vec[0] == "c" {
|
||
Some(tag_vec[1].clone())
|
||
} else {
|
||
None
|
||
}
|
||
})
|
||
.unwrap_or_else(|| "unknown".to_string());
|
||
|
||
tracing::info!(
|
||
"PR event {} added to purgatory: waiting for git push with commit {}",
|
||
event_id_str,
|
||
commit
|
||
);
|
||
|
||
// Add to purgatory
|
||
self.ctx.purgatory.add_pr(
|
||
event.clone(),
|
||
event.id.to_hex(),
|
||
commit.clone(),
|
||
is_synced,
|
||
);
|
||
|
||
accepted_purgatory(format!(
|
||
"PR event stored, waiting for git push with commit {}",
|
||
commit
|
||
))
|
||
}
|
||
Ok(true) => {
|
||
// Git data exists - proceed with normal validation
|
||
tracing::debug!("Git data exists for PR event {}", event_id_str);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Err(e) => {
|
||
// Error checking git data - reject event
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Extract ALL repo identifiers from 'a' tags for PR events
|
||
// (PR events can reference multiple repos when there are multiple maintainers)
|
||
let repos = Self::extract_repos_from_pr_event(event);
|
||
// Structured log for migration scripts - log once per repo
|
||
for repo in &repos {
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"git data check failed: {}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
}
|
||
reject_error(format!("Failed to check git data: {}", e))
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Check purgatory for state events that might now be authorized by a new announcement
|
||
///
|
||
/// When an announcement is accepted, state events in purgatory that were previously
|
||
/// rejected due to missing announcements might now be authorized. This method:
|
||
/// 1. Finds all state events in purgatory for the identifier
|
||
/// 2. Re-evaluates authorization for each event
|
||
/// 3. Processes authorized events (releases from purgatory)
|
||
/// 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 {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
|
||
// Re-evaluate authorization with the new announcement
|
||
match self
|
||
.state_policy
|
||
.process_state_event(&entry.event, false, false, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
Ok(WritePolicyResult::Accept) => {
|
||
tracing::info!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
"State event in purgatory now authorized, will be processed"
|
||
);
|
||
// Event will be automatically removed from purgatory by process_state_event
|
||
// and broadcast to subscribers
|
||
}
|
||
Ok(WritePolicyResult::Reject { message, .. }) => {
|
||
if message.contains("not authorized") {
|
||
tracing::debug!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
"State event in purgatory still not authorized, keeping in purgatory"
|
||
);
|
||
// Keep in purgatory - will expire naturally after 30 minutes
|
||
} else {
|
||
tracing::debug!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
reason = %message,
|
||
"State event in purgatory rejected for other reason"
|
||
);
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
error = %e,
|
||
"Error re-evaluating state event in purgatory"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Reapply stored state after a new maintainer repository enters purgatory.
|
||
///
|
||
/// A shared GRASP server can already hold the owner's latest state and Git
|
||
/// objects when an invitee publishes their reciprocal announcement. Since
|
||
/// the state is replaceable and already stored, sync may not deliver it
|
||
/// again. Processing stored candidates newest-first lets each maintainer
|
||
/// chain apply its preferred state before considering older candidates,
|
||
/// without requiring an invitee-authored state event or push.
|
||
async fn reconcile_stored_state_events_for_identifier(&self, identifier: &str) {
|
||
let filter = Filter::new()
|
||
.kind(Kind::RepoState)
|
||
.custom_tag(SingleLetterTag::LOWERCASE_D, identifier.to_string());
|
||
let mut states: Vec<Event> = match self.ctx.database.query(filter).await {
|
||
Ok(events) => events.into_iter().collect(),
|
||
Err(error) => {
|
||
tracing::warn!(
|
||
identifier = %identifier,
|
||
error = %error,
|
||
"Failed to query stored state for new maintainer repository"
|
||
);
|
||
return;
|
||
}
|
||
};
|
||
|
||
states.sort_by(|left, right| {
|
||
right
|
||
.created_at
|
||
.cmp(&left.created_at)
|
||
.then_with(|| left.id.cmp(&right.id))
|
||
});
|
||
|
||
for state in states {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
if let Err(error) = self
|
||
.state_policy
|
||
.process_state_event(&state, true, true, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
tracing::warn!(
|
||
identifier = %identifier,
|
||
event_id = %state.id,
|
||
error = %error,
|
||
"Failed to reapply stored state to new maintainer repository"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle events that must reference accepted repositories or events
|
||
async fn handle_related_event(&self, event: &Event, event_type: &str) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
match self.related_event_policy.check_references(event).await {
|
||
Ok(ReferenceResult::ReferencesRepository(addr_ref)) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: references accepted repository {}",
|
||
event_type,
|
||
event_id_str,
|
||
addr_ref
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::ReferencesEvent(event_ref)) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: references accepted event {}",
|
||
event_type,
|
||
event_id_str,
|
||
event_ref
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::ReferencedByAccepted) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: referenced by accepted event",
|
||
event_type,
|
||
event_id_str
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::Orphan) => {
|
||
let (addressable_refs, event_refs) =
|
||
RelatedEventPolicy::extract_reference_tags(event);
|
||
tracing::info!(
|
||
"Rejected orphan {} event {} (kind={}, pubkey={}): no references to accepted repos or events (checked {} addressable, {} event refs)",
|
||
event_type,
|
||
event.id.to_hex(),
|
||
event.kind.as_u16(),
|
||
event.pubkey.to_hex(),
|
||
addressable_refs.len(),
|
||
event_refs.len()
|
||
);
|
||
reject_restricted(format!(
|
||
"{} event must reference an accepted repository or accepted event",
|
||
event_type
|
||
))
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
"Database query failed for {} {}, rejecting (fail-secure): {}",
|
||
event_type,
|
||
event_id_str,
|
||
e
|
||
);
|
||
reject_error(format!("Database query failed: {}", e))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
impl WritePolicy for Nip34WritePolicy {
|
||
fn admit_event<'a>(
|
||
&'a self,
|
||
event: &'a nostr_sdk::prelude::Event,
|
||
addr: &'a SocketAddr,
|
||
) -> Pin<Box<dyn Future<Output = WritePolicyResult> + Send + 'a>> {
|
||
Box::pin(async move {
|
||
// The operator may reuse this service identity with trusted
|
||
// automation such as ngit-ci, so owner-signed events skip the
|
||
// blacklist — if the key is ever compromised, rotating it is the
|
||
// only remediation. They do NOT skip the deletion gate (signed
|
||
// events are public, so replaying a retracted CI status event
|
||
// must not undo its NIP-09/NIP-62 tombstone) and they do NOT
|
||
// skip the dedicated admission policies in the dispatch below:
|
||
// owner-signed NIP-34 repository events are validated, aligned,
|
||
// and routed through purgatory exactly like anyone else's.
|
||
let is_relay_owner = self.relay_owner_pubkey == Some(event.pubkey);
|
||
|
||
// For all non-owner authors, the event blacklist overrides every
|
||
// repository and event-kind policy below.
|
||
if !is_relay_owner {
|
||
if let Some(reason) = self.check_event_blacklist(event) {
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
author = %event.pubkey.to_hex(),
|
||
reason = %reason,
|
||
"Rejected event from blacklisted author"
|
||
);
|
||
return WritePolicyResult::reject(MachineReadablePrefix::Blocked, reason);
|
||
}
|
||
}
|
||
|
||
// Deletion / vanish gate.
|
||
//
|
||
// The relay-builder library has a `check_id` gate that runs *before*
|
||
// this policy and used to reject re-submitted deleted events with
|
||
// "this event is deleted" — but that gate is fed by the LMDB
|
||
// backend's internal NIP-09/NIP-62 tables, which go silent now that
|
||
// we run the backend with `process_nip09(false)` /
|
||
// `process_nip62(false)`. We reproduce that rejection here from our
|
||
// own persistent `Tombstones` store so deleted events (and events
|
||
// from vanished pubkeys) stay rejected across restarts.
|
||
if let Some(rejection) = self.deletion.gate(event).await {
|
||
return rejection;
|
||
}
|
||
|
||
// Detect if this is a synced event (from proactive sync) vs user-submitted
|
||
// Sync uses localhost:0 as a dummy address
|
||
let is_synced = addr.ip().is_loopback() && addr.port() == 0;
|
||
|
||
let result = match event.kind {
|
||
// The relay's own identity events are always welcome: the
|
||
// startup seeding path stores them directly, and the
|
||
// operator may customize them through an ordinary client.
|
||
// Everyone else's kind 0/10002 goes through the
|
||
// proactive-identity policy.
|
||
Kind::Metadata | Kind::RelayList if is_relay_owner => {
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
kind = event.kind.as_u16(),
|
||
"Accepted relay-owner identity event"
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Kind::Metadata | Kind::RelayList => self.handle_proactive_identity(event).await,
|
||
Kind::GitRepoAnnouncement => self.handle_announcement(event).await,
|
||
Kind::RepoState => self.handle_state(event, is_synced).await,
|
||
Kind::GitPullRequest | Kind::GitPullRequestUpdate => {
|
||
self.handle_pr_event(event, is_synced).await
|
||
}
|
||
Kind::GitUserGraspList => {
|
||
// Accept all kind 10317 (User Grasp List) events
|
||
// for better GRASP repository discovery
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
author = %event.pubkey.to_hex(),
|
||
"Accepted kind 10317 user grasp list"
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Kind::EventDeletion => self.deletion.handle_nip09(event).await,
|
||
Kind::RequestToVanish => self.deletion.handle_vanish(event).await,
|
||
// Owner trust is scoped to kinds with no dedicated policy
|
||
// above: valid owner-signed events that would otherwise fall
|
||
// through to the generic related-event rejection — such as
|
||
// `ngit-ci` coordinator advertisements with no repository
|
||
// root tag — are accepted. NIP-34 repository kinds never
|
||
// reach this arm, so owner events cannot bypass announcement
|
||
// validation, state ref alignment, or purgatory git-data
|
||
// handling.
|
||
_ if is_relay_owner => {
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
kind = event.kind.as_u16(),
|
||
"Accepted event from relay-owner identity"
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
_ => self.handle_related_event(event, "Event").await,
|
||
};
|
||
|
||
result
|
||
})
|
||
}
|
||
}
|
||
|
||
/// Storage handles shared by the relay runtime and HTTP/runtime services.
|
||
pub struct RelayStores {
|
||
/// Main relay database used for event queries and persistence.
|
||
pub database: SharedDatabase,
|
||
/// Persistent NIP-09/NIP-62 tombstone store.
|
||
pub tombstones: Tombstones,
|
||
/// Holding store used for deletion archival + expiry cleanup
|
||
pub holding: HoldingStore,
|
||
/// Replaceable-history store used for rollback history capture.
|
||
pub history: ReplaceableHistoryStore,
|
||
}
|
||
|
||
/// Runtime components produced when constructing the local relay.
|
||
pub struct RelayRuntime {
|
||
/// The local relay instance.
|
||
pub relay: LocalRelay,
|
||
/// The write policy used for event admission, routing, and persistence.
|
||
pub write_policy: Nip34WritePolicy,
|
||
/// Deletion service shared with write policy for NIP-09/NIP-62 operations.
|
||
pub deletion: DeletionService,
|
||
/// Runtime facade for deletion startup tasks and background cleanup.
|
||
pub deletion_runtime: DeletionRuntime,
|
||
/// Repository lifecycle locks shared by HTTP serving and deletion/recovery.
|
||
pub lifecycle: RepositoryLifecycle,
|
||
/// Storage handles used by the relay runtime.
|
||
pub stores: RelayStores,
|
||
}
|
||
|
||
/// Create a configured LocalRelay with full GRASP-01 validation
|
||
///
|
||
/// Returns a `RelayRuntime` struct containing the relay, write policy, deletion
|
||
/// runtime, lifecycle locks, and storage handles used by HTTP/runtime services.
|
||
pub async fn create_relay(
|
||
config: &Config,
|
||
purgatory: Arc<crate::purgatory::Purgatory>,
|
||
repo_init_locks: crate::grasp06::receive::RepoInitLocks,
|
||
private_access: Option<PrivateAccess>,
|
||
) -> Result<RelayRuntime> {
|
||
tracing::info!("Configuring nostr relay with GRASP-01 validation...");
|
||
|
||
// Determine database path
|
||
let db_path = Path::new(&config.relay_data_path);
|
||
|
||
// Create database based on configuration
|
||
//
|
||
// NIP-09 (deletion) and NIP-62 (request to vanish) auto-processing is
|
||
// explicitly disabled on the main database: ngit-grasp owns this handling
|
||
// (see the deletion subsystem facade + `Tombstones` store).
|
||
// Leaving the backend defaults (`true`) would double-process: the backend
|
||
// would silently hard-delete events and block re-submission via its own
|
||
// internal tables that we cannot inspect, conflicting with our purgatory /
|
||
// bare-repo bookkeeping.
|
||
let (database, tombstones, holding, history): (
|
||
SharedDatabase,
|
||
Tombstones,
|
||
HoldingStore,
|
||
ReplaceableHistoryStore,
|
||
) = match config.database_backend {
|
||
DatabaseBackend::Memory => {
|
||
tracing::info!("Using in-memory database (no persistence)");
|
||
// Disable the backend's built-in NIP-09 / NIP-62 auto-processing to
|
||
// match the LMDB path: ngit-grasp owns this handling (deletion
|
||
// subsystem + Tombstones store). Leaving the memory
|
||
// backend's defaults (`true`) would silently suppress targeted events
|
||
// at query time the moment a kind-5 is *stored* — which in particular
|
||
// defeats `deletion_request_disrespector` (archival) mode, where the
|
||
// kind-5 is intentionally stored but not acted upon.
|
||
let db = MemoryDatabase::builder()
|
||
.max_events(NonZeroUsize::new(100_000).unwrap())
|
||
.process_nip09(false)
|
||
.process_nip62(false)
|
||
.build();
|
||
(
|
||
Arc::new(db),
|
||
Tombstones::in_memory(),
|
||
HoldingStore::in_memory(),
|
||
ReplaceableHistoryStore::in_memory(),
|
||
)
|
||
}
|
||
DatabaseBackend::Lmdb => {
|
||
tracing::info!("Using LMDB backend at: {}", db_path.display());
|
||
// Ensure the database directory exists
|
||
std::fs::create_dir_all(db_path).map_err(|e| {
|
||
anyhow::anyhow!(
|
||
"Failed to create LMDB directory {}: {}",
|
||
db_path.display(),
|
||
e
|
||
)
|
||
})?;
|
||
let db = NostrLmdb::builder(db_path)
|
||
.process_nip09(false)
|
||
.process_nip62(false)
|
||
.build()
|
||
.await
|
||
.map_err(|e| {
|
||
anyhow::anyhow!(
|
||
"Failed to open LMDB database at {}: {}",
|
||
db_path.display(),
|
||
e
|
||
)
|
||
})?;
|
||
let tombstones = Tombstones::open_lmdb(db_path).await?;
|
||
let holding =
|
||
HoldingStore::open_lmdb(db_path, Path::new(&config.effective_git_data_path()))
|
||
.await?;
|
||
let history = ReplaceableHistoryStore::open_lmdb(db_path).await?;
|
||
(Arc::new(db), tombstones, holding, history)
|
||
}
|
||
};
|
||
|
||
// Build relay with GRASP-01 validation
|
||
// Clone Arc for the write policy so both relay and policy can access the database
|
||
let git_data_path = config.effective_git_data_path();
|
||
|
||
// Log archive configuration (config.validate() must be called at startup)
|
||
let archive_config = config.archive_config();
|
||
if archive_config.enabled() {
|
||
tracing::info!(
|
||
"GRASP-05 archive mode enabled: archive_all={}, whitelist_entries={}, read_only={}",
|
||
archive_config.archive_all,
|
||
archive_config.whitelist.len(),
|
||
archive_config.read_only
|
||
);
|
||
}
|
||
|
||
// Log repository configuration
|
||
let repository_config = config.repository_config();
|
||
if repository_config.enabled() {
|
||
tracing::info!(
|
||
"Repository whitelist enabled: whitelist_entries={}",
|
||
repository_config.whitelist.len()
|
||
);
|
||
}
|
||
|
||
let lifecycle = match config.database_backend {
|
||
DatabaseBackend::Memory => RepositoryLifecycle::in_memory(),
|
||
DatabaseBackend::Lmdb => RepositoryLifecycle::for_git_data_path(Path::new(&git_data_path)),
|
||
};
|
||
|
||
// Create write policy with purgatory integration
|
||
let write_policy = Nip34WritePolicy::new(
|
||
database.clone(),
|
||
tombstones.clone(),
|
||
holding.clone(),
|
||
lifecycle.clone(),
|
||
history.clone(),
|
||
&git_data_path,
|
||
purgatory,
|
||
config.clone(),
|
||
repo_init_locks,
|
||
private_access,
|
||
);
|
||
|
||
let mut builder = LocalRelayBuilder::default()
|
||
.database(database.clone())
|
||
.write_policy(write_policy.clone())
|
||
.rate_limit(RateLimit {
|
||
max_reqs: config.relay_max_subscriptions,
|
||
notes_per_minute: 60,
|
||
})
|
||
.queries_per_minute(CLIENT_QUERIES_PER_MINUTE)
|
||
.auth_events_per_minute(30)
|
||
.messages_per_minute(CLIENT_MESSAGES_PER_MINUTE)
|
||
.max_websocket_message_size(5 * 1024 * 1024)
|
||
.max_event_size(config.relay_max_event_size_bytes)
|
||
.websocket_handshake_timeout(Duration::from_secs(10))
|
||
.max_subid_length(250)
|
||
.max_filters_per_req(20)
|
||
.max_subscription_bytes(MAX_SUBSCRIPTION_STATE_BYTES)
|
||
.max_negentropy_subscriptions(10)
|
||
.max_negentropy_items(50_000)
|
||
.max_filter_limit(config.relay_filter_limit)
|
||
.max_query_results(config.relay_filter_limit)
|
||
.default_filter_limit(config.relay_filter_limit);
|
||
|
||
// `LocalRelayBuilder` otherwise applies rust-nostr's own finite default.
|
||
// In ngit-grasp, an unset limit deliberately delegates admission control to
|
||
// the OS and surrounding infrastructure, as documented by every config
|
||
// surface. Explicit operator limits remain exact.
|
||
builder = builder.max_connections(effective_max_connections(config.max_connections));
|
||
|
||
let relay = builder.build();
|
||
|
||
let deletion_runtime = DeletionRuntime::new(
|
||
write_policy.deletion().clone(),
|
||
holding.clone(),
|
||
config.holding_retention(),
|
||
config.holding_cleanup_interval(),
|
||
);
|
||
|
||
tracing::info!(
|
||
"Relay configured with GRASP-01 validation for domain: {}",
|
||
config.domain
|
||
);
|
||
|
||
let deletion = write_policy.deletion().clone();
|
||
|
||
Ok(RelayRuntime {
|
||
relay,
|
||
write_policy,
|
||
deletion,
|
||
deletion_runtime,
|
||
lifecycle,
|
||
stores: RelayStores {
|
||
database,
|
||
tombstones,
|
||
holding,
|
||
history,
|
||
},
|
||
})
|
||
}
|
||
|
||
fn effective_max_connections(configured: Option<usize>) -> usize {
|
||
configured.unwrap_or(tokio::sync::Semaphore::MAX_PERMITS)
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod connection_limit_tests {
|
||
use super::effective_max_connections;
|
||
|
||
#[test]
|
||
fn unset_connection_limit_overrides_dependency_default() {
|
||
assert_eq!(
|
||
effective_max_connections(None),
|
||
tokio::sync::Semaphore::MAX_PERMITS
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn configured_connection_limit_remains_exact() {
|
||
assert_eq!(effective_max_connections(Some(2)), 2);
|
||
}
|
||
}
|