commit 437789fe6181ce636ac223ccef143d819c7488b9
parent ecc79cd00cf47c0ccde8d9e5ed2da51645873926
Author: triesap <tyson@radroots.org>
Date: Mon, 27 Jul 2026 11:46:35 +0000
transport: bound relay request lifecycle
- Pin validated DNS answers through direct TLS connections and reject proxy and special-use destinations.
- Isolate Nostr relay pools per request and clean them on completion, timeout, cancellation, or drop.
- Stream fetch items under shared count, byte, diagnostic, and total-deadline budgets.
- Add adversarial security, lifecycle, and resource-boundary coverage.
Diffstat:
12 files changed, 1345 insertions(+), 347 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -5243,9 +5243,12 @@ dependencies = [
name = "radroots_transport_nostr"
version = "1.0.0-alpha.1"
dependencies = [
+ "async-wsocket",
"futures",
"hex",
"nostr",
+ "nostr-relay-pool",
+ "nostr-sdk",
"radroots_authority",
"radroots_blossom",
"radroots_event",
@@ -5261,6 +5264,7 @@ dependencies = [
"tempfile",
"thiserror 1.0.69",
"tokio",
+ "tokio-tungstenite",
"url",
]
diff --git a/Cargo.toml b/Cargo.toml
@@ -65,6 +65,7 @@ homepage = "https://radroots.org"
readme = "README"
[workspace.dependencies]
+async-wsocket = { version = "0.13.2" }
dto_bindgen = { version = "0.1.0" }
dto_bindgen_core = { version = "0.1.0" }
radroots_core = { path = "crates/core", version = "=1.0.0-alpha.1", default-features = false }
@@ -206,6 +207,10 @@ tempfile = { version = "3" }
tar = { version = "0.4" }
thiserror = { version = "1" }
tokio = { version = "1" }
+tokio-tungstenite = { version = "0.26.2", default-features = false, features = [
+ "connect",
+ "rustls-tls-webpki-roots",
+] }
toml = { version = "0.8" }
tracing = { version = "0.1", default-features = false }
tracing-appender = { version = "0.2" }
diff --git a/crates/nostr/src/client.rs b/crates/nostr/src/client.rs
@@ -182,6 +182,14 @@ impl RadrootsNostrClient {
self.inner.try_connect(timeout).await
}
+ pub async fn disconnect(&self) {
+ self.inner.disconnect().await;
+ }
+
+ pub async fn shutdown(&self) {
+ self.inner.shutdown().await;
+ }
+
pub async fn add_relay(&self, url: &str) -> Result<bool, RadrootsNostrError> {
Ok(self.inner.add_relay(url).await?)
}
@@ -199,6 +207,10 @@ impl RadrootsNostrClient {
Ok(())
}
+ pub async fn force_remove_all_relays(&self) {
+ self.inner.force_remove_all_relays().await;
+ }
+
pub async fn relays(&self) -> HashMap<RadrootsNostrRelayUrl, RadrootsNostrRelay> {
self.inner.relays().await
}
@@ -509,6 +521,29 @@ mod tests {
}
#[tokio::test]
+ async fn client_lifecycle_request_scoped_cleanup_prevents_relay_inheritance() {
+ let first = RadrootsNostrClient::new_signerless();
+ first
+ .add_write_relay("wss://relay-a.example")
+ .await
+ .expect("first request relay");
+ first
+ .add_read_relay("wss://relay-b.example")
+ .await
+ .expect("second request relay");
+ assert_eq!(first.relays().await.len(), 2);
+
+ first.disconnect().await;
+ first.force_remove_all_relays().await;
+ assert!(first.relays().await.is_empty());
+ first.shutdown().await;
+
+ let second = RadrootsNostrClient::new_signerless();
+ assert!(second.relays().await.is_empty());
+ second.shutdown().await;
+ }
+
+ #[tokio::test]
async fn generic_builder_rejects_all_typed_authoring_reservations_before_signer_access() {
let client = RadrootsNostrClient::new_signerless();
let raw_kind_one = RadrootsNostrKind::Custom(RadrootsNostrKind::TextNote.as_u16());
diff --git a/crates/transport_nostr/Cargo.toml b/crates/transport_nostr/Cargo.toml
@@ -15,7 +15,12 @@ readme = "README"
default = ["std", "client", "storage", "runtime-tokio"]
std = []
client = [
+ "dep:async-wsocket",
+ "dep:nostr-relay-pool",
+ "dep:nostr-sdk",
"dep:radroots_nostr",
+ "dep:tokio",
+ "dep:tokio-tungstenite",
"radroots_nostr/std",
"radroots_nostr/client",
"radroots_nostr/events",
@@ -35,6 +40,7 @@ runtime-tokio = [
]
[dependencies]
+async-wsocket = { workspace = true, optional = true }
radroots_event = { workspace = true, default-features = false, features = [
"serde",
"signature",
@@ -57,11 +63,14 @@ radroots_transport = { workspace = true, default-features = false, features = ["
futures = { workspace = true }
hex = { workspace = true }
nostr = { workspace = true }
+nostr-relay-pool = { workspace = true, optional = true }
+nostr-sdk = { workspace = true, optional = true }
serde = { workspace = true, features = ["derive", "std"] }
serde_json = { workspace = true, features = ["std"] }
sqlx = { workspace = true, optional = true }
thiserror = { workspace = true }
-tokio = { workspace = true, optional = true, features = ["rt"] }
+tokio = { workspace = true, optional = true, features = ["net", "rt", "time"] }
+tokio-tungstenite = { workspace = true, optional = true }
url = { workspace = true }
[dev-dependencies]
diff --git a/crates/transport_nostr/README b/crates/transport_nostr/README
@@ -15,7 +15,7 @@ Every fetch path verifies the NIP-01 id and signature before filter matching,
unique-event budgeting, or returning an event. Repeated event ids preserve
per-relay observation evidence without consuming the unique-event limit. The
unique-event limit is bounded at 1,000 so final stored-event visibility can be
-evaluated in one coherent event-store snapshot. A fetch scans at most 64,000
+evaluated in one coherent event-store snapshot. A fetch scans at most 4,096
raw events and 64 MiB of aggregate raw JSON, and rejects any individual raw
event over 256 KiB before Radroots parses adapter raw JSON. Count and byte
budgets are charged globally, in adapter order, before Radroots parsing,
@@ -43,9 +43,10 @@ separate flags and aggregate counts. Local event-store failures abort the
operation and remain typed transport errors, so callers can retry without
confusing storage failure with bad relay input.
-`RadrootsRelayUrlPolicy::Public` is for trusted relay configuration. It rejects
-non-canonical and known non-global literal destinations, but hostname checks do
-not pin DNS resolution in the SDK connector and are not an SSRF boundary for
-attacker-controlled relay hostnames. Validate resolved addresses at the network
-boundary or use a connector that pins approved resolutions before accepting
-untrusted relay configuration.
+`RadrootsRelayUrlPolicy::Public` rejects non-canonical and known non-global
+literal destinations. The official publish and fetch adapters additionally
+resolve every hostname at connect and reconnect, reject the complete answer if
+any address is forbidden, and pin that accepted answer through TCP connect
+while retaining the original hostname for TLS verification. Parsing a public
+relay URL by itself remains syntax validation rather than an SSRF boundary;
+custom adapters must enforce equivalent resolution and connection policy.
diff --git a/crates/transport_nostr/src/connector.rs b/crates/transport_nostr/src/connector.rs
@@ -0,0 +1,447 @@
+#![forbid(unsafe_code)]
+
+#[cfg(target_arch = "wasm32")]
+use radroots_nostr::prelude::RadrootsNostrClient;
+
+#[cfg(target_arch = "wasm32")]
+pub(crate) fn request_scoped_nostr_client() -> RadrootsNostrClient {
+ RadrootsNostrClient::new_signerless()
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+mod native {
+ use async_wsocket::{ConnectionMode, Message};
+ use futures::{Sink, StreamExt};
+ use nostr_relay_pool::transport::error::TransportError;
+ use nostr_relay_pool::transport::websocket::{
+ WebSocketSink, WebSocketStream, WebSocketTransport,
+ };
+ use radroots_nostr::prelude::RadrootsNostrClient;
+ use std::collections::BTreeSet;
+ use std::fmt;
+ use std::future::Future;
+ use std::net::{IpAddr, SocketAddr};
+ use std::pin::Pin;
+ use std::sync::Arc;
+ use std::task::{Context, Poll};
+ use std::time::Duration;
+ use tokio::net::TcpStream;
+ use tokio_tungstenite::tungstenite::Message as TungsteniteMessage;
+ use tokio_tungstenite::{MaybeTlsStream, WebSocketStream as TokioWebSocketStream};
+
+ use crate::{RadrootsRelayUrl, RadrootsRelayUrlPolicy};
+
+ type ResolveFuture<'a> =
+ Pin<Box<dyn Future<Output = Result<Vec<SocketAddr>, PinnedConnectError>> + Send + 'a>>;
+ type NativeWebSocket = TokioWebSocketStream<MaybeTlsStream<TcpStream>>;
+
+ trait RelayDnsResolver: fmt::Debug + Send + Sync {
+ fn resolve<'a>(&'a self, host: &'a str, port: u16) -> ResolveFuture<'a>;
+ }
+
+ #[derive(Clone, Copy, Debug, Default)]
+ struct SystemRelayDnsResolver;
+
+ impl RelayDnsResolver for SystemRelayDnsResolver {
+ fn resolve<'a>(&'a self, host: &'a str, port: u16) -> ResolveFuture<'a> {
+ Box::pin(async move {
+ tokio::net::lookup_host((host, port))
+ .await
+ .map(|addresses| addresses.collect())
+ .map_err(|_| PinnedConnectError::ResolutionFailed)
+ })
+ }
+ }
+
+ #[derive(Clone, Debug)]
+ struct PinnedWebsocketTransport<R> {
+ resolver: Arc<R>,
+ }
+
+ impl<R> PinnedWebsocketTransport<R> {
+ fn new(resolver: R) -> Self {
+ Self {
+ resolver: Arc::new(resolver),
+ }
+ }
+ }
+
+ impl<R> WebSocketTransport for PinnedWebsocketTransport<R>
+ where
+ R: RelayDnsResolver + 'static,
+ {
+ fn support_ping(&self) -> bool {
+ true
+ }
+
+ fn connect<'a>(
+ &'a self,
+ url: &'a nostr::Url,
+ mode: &'a ConnectionMode,
+ timeout: Duration,
+ ) -> nostr::util::BoxedFuture<'a, Result<(WebSocketSink, WebSocketStream), TransportError>>
+ {
+ Box::pin(async move {
+ if !matches!(mode, ConnectionMode::Direct) {
+ return Err(TransportError::backend(
+ PinnedConnectError::ProxyModeForbidden,
+ ));
+ }
+ let socket = tokio::time::timeout(timeout, self.connect_pinned(url))
+ .await
+ .map_err(|_| TransportError::backend(PinnedConnectError::DeadlineExceeded))?
+ .map_err(TransportError::backend)?;
+ let (sink, stream) = socket.split();
+ let sink: WebSocketSink = Box::new(PinnedSink(sink));
+ let stream: WebSocketStream = Box::pin(stream.map(|message| {
+ message
+ .map_err(TransportError::backend)
+ .and_then(native_message)
+ }));
+ Ok((sink, stream))
+ })
+ }
+ }
+
+ impl<R> PinnedWebsocketTransport<R>
+ where
+ R: RelayDnsResolver,
+ {
+ async fn connect_pinned(
+ &self,
+ url: &nostr::Url,
+ ) -> Result<NativeWebSocket, PinnedConnectError> {
+ let addresses = self.resolve_and_validate(url).await?;
+ let mut connected = false;
+ for address in addresses {
+ let stream = match TcpStream::connect(address).await {
+ Ok(stream) => stream,
+ Err(_) => continue,
+ };
+ connected = true;
+ if let Ok((socket, response)) =
+ tokio_tungstenite::client_async_tls(url.as_str(), stream).await
+ {
+ if response.status().as_u16() == 101 {
+ return Ok(socket);
+ }
+ return Err(PinnedConnectError::RedirectOrHandshakeRejected);
+ }
+ }
+ if connected {
+ Err(PinnedConnectError::TlsValidationFailed)
+ } else {
+ Err(PinnedConnectError::ConnectionFailed)
+ }
+ }
+
+ async fn resolve_and_validate(
+ &self,
+ url: &nostr::Url,
+ ) -> Result<Vec<SocketAddr>, PinnedConnectError> {
+ let host = url.host_str().ok_or(PinnedConnectError::InvalidRelayUrl)?;
+ let port = url
+ .port_or_known_default()
+ .ok_or(PinnedConnectError::InvalidRelayUrl)?;
+ let is_exact_loopback_host = matches!(host, "localhost" | "127.0.0.1" | "::1");
+ let policy = if is_exact_loopback_host {
+ RadrootsRelayUrlPolicy::Localhost
+ } else {
+ RadrootsRelayUrlPolicy::Public
+ };
+ let relay = RadrootsRelayUrl::parse(url.as_str(), policy)
+ .map_err(|_| PinnedConnectError::InvalidRelayUrl)?;
+ let addresses = self.resolver.resolve(host, port).await?;
+ if addresses.is_empty() {
+ return Err(PinnedConnectError::ResolutionReturnedNoAddresses);
+ }
+ if addresses.iter().any(|address| address.port() != port) {
+ return Err(PinnedConnectError::DestinationForbidden);
+ }
+ if is_exact_loopback_host {
+ if addresses
+ .iter()
+ .any(|address| !is_exact_loopback(address.ip()))
+ {
+ return Err(PinnedConnectError::DestinationForbidden);
+ }
+ } else {
+ relay
+ .validate_public_resolved_ip_addrs(addresses.iter().map(|address| address.ip()))
+ .map_err(|_| PinnedConnectError::DestinationForbidden)?;
+ }
+ Ok(addresses
+ .into_iter()
+ .collect::<BTreeSet<_>>()
+ .into_iter()
+ .collect())
+ }
+ }
+
+ fn is_exact_loopback(address: IpAddr) -> bool {
+ match address {
+ IpAddr::V4(address) => address.is_loopback(),
+ IpAddr::V6(address) => address.is_loopback(),
+ }
+ }
+
+ struct PinnedSink(futures::stream::SplitSink<NativeWebSocket, TungsteniteMessage>);
+
+ impl Sink<Message> for PinnedSink {
+ type Error = TransportError;
+
+ fn poll_ready(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<Result<(), Self::Error>> {
+ Pin::new(&mut self.0)
+ .poll_ready(context)
+ .map_err(TransportError::backend)
+ }
+
+ fn start_send(mut self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
+ Pin::new(&mut self.0)
+ .start_send(item.into())
+ .map_err(TransportError::backend)
+ }
+
+ fn poll_flush(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<Result<(), Self::Error>> {
+ Pin::new(&mut self.0)
+ .poll_flush(context)
+ .map_err(TransportError::backend)
+ }
+
+ fn poll_close(
+ mut self: Pin<&mut Self>,
+ context: &mut Context<'_>,
+ ) -> Poll<Result<(), Self::Error>> {
+ Pin::new(&mut self.0)
+ .poll_close(context)
+ .map_err(TransportError::backend)
+ }
+ }
+
+ fn native_message(message: TungsteniteMessage) -> Result<Message, TransportError> {
+ match message {
+ TungsteniteMessage::Text(value) => Ok(Message::Text(value.to_string())),
+ TungsteniteMessage::Binary(value) => Ok(Message::Binary(value.to_vec())),
+ TungsteniteMessage::Ping(value) => Ok(Message::Ping(value.to_vec())),
+ TungsteniteMessage::Pong(value) => Ok(Message::Pong(value.to_vec())),
+ TungsteniteMessage::Close(value) => Ok(Message::Close(value.map(|frame| {
+ async_wsocket::message::CloseFrame {
+ code: frame.code.into(),
+ reason: frame.reason.to_string(),
+ }
+ }))),
+ TungsteniteMessage::Frame(_) => Err(TransportError::backend(
+ PinnedConnectError::UnexpectedRawFrame,
+ )),
+ }
+ }
+
+ #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+ enum PinnedConnectError {
+ InvalidRelayUrl,
+ ProxyModeForbidden,
+ ResolutionFailed,
+ ResolutionReturnedNoAddresses,
+ DestinationForbidden,
+ ConnectionFailed,
+ TlsValidationFailed,
+ RedirectOrHandshakeRejected,
+ DeadlineExceeded,
+ UnexpectedRawFrame,
+ }
+
+ impl fmt::Display for PinnedConnectError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(match self {
+ Self::InvalidRelayUrl => "relay URL rejected",
+ Self::ProxyModeForbidden => "proxy connection mode rejected",
+ Self::ResolutionFailed => "relay DNS resolution failed",
+ Self::ResolutionReturnedNoAddresses => "relay DNS resolution returned no addresses",
+ Self::DestinationForbidden => "relay destination rejected",
+ Self::ConnectionFailed => "relay connection failed",
+ Self::TlsValidationFailed => "relay TLS validation failed",
+ Self::RedirectOrHandshakeRejected => "relay redirect or handshake rejected",
+ Self::DeadlineExceeded => "relay connection deadline exceeded",
+ Self::UnexpectedRawFrame => "relay returned an unexpected raw frame",
+ })
+ }
+ }
+
+ impl std::error::Error for PinnedConnectError {}
+
+ pub(crate) fn request_scoped_nostr_client() -> RadrootsNostrClient {
+ let inner = nostr_sdk::ClientBuilder::new()
+ .websocket_transport(PinnedWebsocketTransport::new(SystemRelayDnsResolver))
+ .build();
+ RadrootsNostrClient::from_inner(inner)
+ }
+
+ #[cfg(test)]
+ mod tests {
+ use super::{
+ PinnedConnectError, PinnedWebsocketTransport, RelayDnsResolver, ResolveFuture,
+ };
+ use async_wsocket::ConnectionMode;
+ use nostr_relay_pool::transport::websocket::WebSocketTransport;
+ use std::collections::VecDeque;
+ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+ use std::sync::Mutex;
+ use std::time::Duration;
+
+ #[derive(Debug)]
+ struct ScriptedResolver {
+ answers: Mutex<VecDeque<Vec<SocketAddr>>>,
+ }
+
+ impl ScriptedResolver {
+ fn new(answers: impl IntoIterator<Item = Vec<SocketAddr>>) -> Self {
+ Self {
+ answers: Mutex::new(answers.into_iter().collect()),
+ }
+ }
+ }
+
+ impl RelayDnsResolver for ScriptedResolver {
+ fn resolve<'a>(&'a self, _host: &'a str, _port: u16) -> ResolveFuture<'a> {
+ Box::pin(async move {
+ self.answers
+ .lock()
+ .expect("scripted resolver lock")
+ .pop_front()
+ .ok_or(PinnedConnectError::ResolutionFailed)
+ })
+ }
+ }
+
+ fn socket(address: IpAddr, port: u16) -> SocketAddr {
+ SocketAddr::new(address, port)
+ }
+
+ #[tokio::test]
+ async fn relay_security_resolution_rejects_mixed_special_and_rebinding_answers() {
+ let public = socket(IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34)), 443);
+ let private = socket(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 443);
+ let metadata = socket(IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)), 443);
+ let mapped_private =
+ socket(IpAddr::V6(Ipv4Addr::new(10, 0, 0, 1).to_ipv6_mapped()), 443);
+ let transport = PinnedWebsocketTransport::new(ScriptedResolver::new([
+ vec![public, public],
+ vec![public, private],
+ vec![metadata],
+ vec![mapped_private],
+ vec![private],
+ ]));
+ let url = nostr::Url::parse("wss://relay.example").unwrap();
+
+ assert_eq!(
+ transport.resolve_and_validate(&url).await.unwrap(),
+ vec![public]
+ );
+ for _ in 0..4 {
+ assert_eq!(
+ transport.resolve_and_validate(&url).await.unwrap_err(),
+ PinnedConnectError::DestinationForbidden
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn relay_security_local_policy_requires_exact_loopback_resolution() {
+ let loopback_v4 = socket(IpAddr::V4(Ipv4Addr::LOCALHOST), 80);
+ let loopback_v6 = socket(IpAddr::V6(Ipv6Addr::LOCALHOST), 80);
+ let public = socket(IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34)), 80);
+ let transport = PinnedWebsocketTransport::new(ScriptedResolver::new([
+ vec![loopback_v6, loopback_v4],
+ vec![loopback_v4, public],
+ Vec::new(),
+ ]));
+ let local = nostr::Url::parse("ws://localhost").unwrap();
+
+ assert_eq!(
+ transport.resolve_and_validate(&local).await.unwrap(),
+ vec![loopback_v4, loopback_v6]
+ );
+ assert_eq!(
+ transport.resolve_and_validate(&local).await.unwrap_err(),
+ PinnedConnectError::DestinationForbidden
+ );
+ assert_eq!(
+ transport.resolve_and_validate(&local).await.unwrap_err(),
+ PinnedConnectError::ResolutionReturnedNoAddresses
+ );
+ }
+
+ #[tokio::test]
+ async fn relay_security_rejects_wrong_port_and_every_special_use_address_class() {
+ let forbidden = [
+ IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
+ IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
+ IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
+ IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)),
+ IpAddr::V4(Ipv4Addr::new(224, 0, 0, 1)),
+ IpAddr::V4(Ipv4Addr::BROADCAST),
+ IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)),
+ IpAddr::V4(Ipv4Addr::new(100, 64, 0, 1)),
+ IpAddr::V4(Ipv4Addr::new(192, 0, 0, 8)),
+ IpAddr::V4(Ipv4Addr::new(192, 88, 99, 2)),
+ IpAddr::V4(Ipv4Addr::new(198, 18, 0, 1)),
+ IpAddr::V4(Ipv4Addr::new(240, 0, 0, 1)),
+ IpAddr::V6(Ipv6Addr::UNSPECIFIED),
+ IpAddr::V6(Ipv6Addr::LOCALHOST),
+ IpAddr::V6("ff02::1".parse().unwrap()),
+ IpAddr::V6("fc00::1".parse().unwrap()),
+ IpAddr::V6("fe80::1".parse().unwrap()),
+ IpAddr::V6("64:ff9b::a00:1".parse().unwrap()),
+ IpAddr::V6("2001:db8::1".parse().unwrap()),
+ IpAddr::V6("2001::1".parse().unwrap()),
+ IpAddr::V6("2002::1".parse().unwrap()),
+ IpAddr::V6("3fff::1".parse().unwrap()),
+ IpAddr::V6(Ipv4Addr::new(10, 0, 0, 1).to_ipv6_mapped()),
+ ];
+ let mut answers = Vec::with_capacity(forbidden.len() + 1);
+ answers.push(vec![socket(
+ IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34)),
+ 8443,
+ )]);
+ answers.extend(
+ forbidden
+ .into_iter()
+ .map(|address| vec![socket(address, 443)]),
+ );
+ let transport = PinnedWebsocketTransport::new(ScriptedResolver::new(answers));
+ let url = nostr::Url::parse("wss://relay.example").unwrap();
+
+ for _ in 0..24 {
+ assert_eq!(
+ transport.resolve_and_validate(&url).await.unwrap_err(),
+ PinnedConnectError::DestinationForbidden
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn relay_security_rejects_proxy_mode_before_resolution() {
+ let transport = PinnedWebsocketTransport::new(ScriptedResolver::new([]));
+ let url = nostr::Url::parse("wss://relay.example").unwrap();
+ let mode = ConnectionMode::proxy(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 9050));
+
+ let result =
+ WebSocketTransport::connect(&transport, &url, &mode, Duration::from_millis(10))
+ .await;
+ let error = match result {
+ Ok(_) => panic!("proxy mode must be rejected"),
+ Err(error) => error,
+ };
+ assert!(error.to_string().contains("proxy connection mode rejected"));
+ }
+ }
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) use native::request_scoped_nostr_client;
diff --git a/crates/transport_nostr/src/diagnostic.rs b/crates/transport_nostr/src/diagnostic.rs
@@ -0,0 +1,111 @@
+#![forbid(unsafe_code)]
+
+use radroots_transport::RadrootsTransportTarget;
+use std::collections::HashMap;
+use std::fmt;
+use std::hash::Hash;
+
+pub(crate) fn stable_connection_diagnostic(message: &str) -> &'static str {
+ let bounded = &message.as_bytes()[..message
+ .len()
+ .min(radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES)];
+ if contains_ascii_case_insensitive(bounded, b"proxy") {
+ "proxy-mode-rejected"
+ } else if contains_ascii_case_insensitive(bounded, b"timeout")
+ || contains_ascii_case_insensitive(bounded, b"deadline")
+ {
+ "connection-timeout"
+ } else if contains_ascii_case_insensitive(bounded, b"dns")
+ || contains_ascii_case_insensitive(bounded, b"resolv")
+ {
+ "dns-resolution-failed"
+ } else if contains_ascii_case_insensitive(bounded, b"tls")
+ || contains_ascii_case_insensitive(bounded, b"certificate")
+ || contains_ascii_case_insensitive(bounded, b"handshake")
+ {
+ "tls-or-handshake-failed"
+ } else if contains_ascii_case_insensitive(bounded, b"destination")
+ || contains_ascii_case_insensitive(bounded, b"forbidden")
+ || contains_ascii_case_insensitive(bounded, b"rejected")
+ {
+ "destination-rejected"
+ } else {
+ "connection-failed"
+ }
+}
+
+fn contains_ascii_case_insensitive(haystack: &[u8], needle: &[u8]) -> bool {
+ haystack
+ .windows(needle.len())
+ .any(|window| window.eq_ignore_ascii_case(needle))
+}
+
+pub(crate) fn stable_nostr_failure_summary<K, E>(failed: &HashMap<K, E>) -> String
+where
+ K: fmt::Display + Eq + Hash,
+ E: fmt::Display,
+{
+ if failed.is_empty() {
+ return "no-relay-acknowledged".to_owned();
+ }
+ let mut summaries = failed
+ .iter()
+ .map(|(relay, error)| {
+ let relay = relay.to_string();
+ let identity = RadrootsTransportTarget::nostr_relay(relay.as_str())
+ .map(|target| target.fingerprint().as_str().to_owned())
+ .unwrap_or_else(|_| "invalid-relay-identity".to_owned());
+ format!(
+ "{identity}={}",
+ stable_connection_diagnostic(&error.to_string())
+ )
+ })
+ .collect::<Vec<_>>();
+ summaries.sort_unstable();
+ summaries.dedup();
+ summaries.truncate(radroots_transport::RADROOTS_TRANSPORT_TARGET_MAX_COUNT);
+ summaries.join(";")
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{stable_connection_diagnostic, stable_nostr_failure_summary};
+ use std::collections::HashMap;
+
+ #[test]
+ fn relay_security_diagnostics_are_sorted_bounded_and_redacted() {
+ let mut failures = HashMap::new();
+ failures.insert(
+ "wss://relay-b.example",
+ "TLS certificate rejected for 10.0.0.1 with secret-token",
+ );
+ failures.insert(
+ "wss://relay-a.example",
+ "DNS resolution failed for 192.168.1.1",
+ );
+ let first = stable_nostr_failure_summary(&failures);
+ let second = stable_nostr_failure_summary(&failures);
+
+ assert_eq!(first, second);
+ assert!(first.len() <= radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES);
+ assert!(!first.contains("10.0.0.1"));
+ assert!(!first.contains("192.168.1.1"));
+ assert!(!first.contains("secret-token"));
+ assert_eq!(
+ stable_connection_diagnostic("operation deadline exceeded"),
+ "connection-timeout"
+ );
+ assert_eq!(
+ stable_connection_diagnostic("proxy connection mode rejected"),
+ "proxy-mode-rejected"
+ );
+ assert_eq!(
+ stable_connection_diagnostic("TLS handshake rejected"),
+ "tls-or-handshake-failed"
+ );
+ assert_eq!(
+ stable_nostr_failure_summary::<String, String>(&HashMap::new()),
+ "no-relay-acknowledged"
+ );
+ }
+}
diff --git a/crates/transport_nostr/src/error.rs b/crates/transport_nostr/src/error.rs
@@ -103,6 +103,9 @@ pub enum RadrootsRelayTransportError {
actual: usize,
},
+ #[error("Relay fetch exceeded its {timeout_ms} millisecond total deadline")]
+ FetchDeadlineExceeded { timeout_ms: u64 },
+
#[error("Relay transport {field} uses {actual} UTF-8 bytes; maximum is {max}")]
DiagnosticLimitExceeded {
field: &'static str,
diff --git a/crates/transport_nostr/src/fetch.rs b/crates/transport_nostr/src/fetch.rs
@@ -1,5 +1,7 @@
#![forbid(unsafe_code)]
+use crate::connector::request_scoped_nostr_client;
+use crate::diagnostic::{stable_connection_diagnostic, stable_nostr_failure_summary};
use crate::error::ensure_nonnegative_timestamp;
use crate::{RadrootsRelayOutcome, RadrootsRelayTargetSet, RadrootsRelayTransportError};
use core::time::Duration;
@@ -11,7 +13,7 @@ use radroots_event_store::{
RadrootsEventStore, RadrootsEventVisibility, RadrootsTransportObservation,
RadrootsTransportObservationType,
};
-use radroots_nostr::prelude::{RadrootsNostrClient, RadrootsNostrEvent, RadrootsNostrFilter};
+use radroots_nostr::prelude::{RadrootsNostrEvent, RadrootsNostrFilter};
use radroots_transport::{
RADROOTS_TRANSPORT_FETCH_ADMITTED_EVENT_MAX_COUNT, RADROOTS_TRANSPORT_FETCH_FILTER_MAX_BYTES,
RADROOTS_TRANSPORT_FETCH_FILTER_MAX_COUNT, RADROOTS_TRANSPORT_FETCH_FILTERS_MAX_BYTES,
@@ -1302,8 +1304,134 @@ pub struct RadrootsRelayFetchReceipt {
pub trait RadrootsRelayFetchAdapter: Send + Sync {
fn fetch<'a>(
&'a self,
- request: RadrootsRelayFetchRequest,
- ) -> BoxFuture<'a, Result<Vec<RadrootsRelayFetchItem>, RadrootsRelayTransportError>>;
+ request: &'a RadrootsRelayFetchRequest,
+ emitter: &'a mut RadrootsRelayFetchEmitter,
+ ) -> BoxFuture<'a, Result<(), RadrootsRelayTransportError>>;
+}
+
+#[derive(Debug)]
+pub struct RadrootsRelayFetchEmitter {
+ items: Vec<RadrootsRelayFetchItem>,
+ deadline: tokio::time::Instant,
+ timeout_ms: u64,
+ max_raw_events: usize,
+ max_raw_json_bytes: usize,
+ raw_events: usize,
+ raw_json_bytes: usize,
+ diagnostic_bytes: usize,
+}
+
+impl RadrootsRelayFetchEmitter {
+ fn new(deadline: tokio::time::Instant, timeout_ms: u64) -> Self {
+ Self {
+ items: Vec::new(),
+ deadline,
+ timeout_ms,
+ max_raw_events: RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX,
+ max_raw_json_bytes: RADROOTS_RELAY_FETCH_RAW_JSON_BYTE_LIMIT_MAX,
+ raw_events: 0,
+ raw_json_bytes: 0,
+ diagnostic_bytes: 0,
+ }
+ }
+
+ pub fn remaining_duration(&self) -> Result<Duration, RadrootsRelayTransportError> {
+ remaining_fetch_duration(self.deadline, self.timeout_ms)
+ }
+
+ pub fn emit(
+ &mut self,
+ item: RadrootsRelayFetchItem,
+ ) -> Result<(), RadrootsRelayTransportError> {
+ if self.items.len() == RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX {
+ return Err(RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "raw_item_count",
+ max: RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX,
+ actual: RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX + 1,
+ });
+ }
+ if let RadrootsRelayFetchItemBody::Event { raw_json, .. } = &item.body {
+ let raw_events = self.raw_events.checked_add(1).ok_or(
+ RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "max_raw_events",
+ max: self.max_raw_events,
+ actual: usize::MAX,
+ },
+ )?;
+ if raw_events > self.max_raw_events {
+ return Err(RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "max_raw_events",
+ max: self.max_raw_events,
+ actual: raw_events,
+ });
+ }
+ let raw_json_bytes = self.raw_json_bytes.checked_add(raw_json.len()).ok_or(
+ RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "aggregate_raw_json_bytes",
+ max: self.max_raw_json_bytes,
+ actual: usize::MAX,
+ },
+ )?;
+ if raw_json_bytes > self.max_raw_json_bytes {
+ return Err(RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "aggregate_raw_json_bytes",
+ max: self.max_raw_json_bytes,
+ actual: raw_json_bytes,
+ });
+ }
+ self.raw_events = raw_events;
+ self.raw_json_bytes = raw_json_bytes;
+ }
+ let message = match &item.body {
+ RadrootsRelayFetchItemBody::Truncated { message, .. }
+ | RadrootsRelayFetchItemBody::Closed { message, .. }
+ | RadrootsRelayFetchItemBody::Notice { message, .. } => Some(message.as_str()),
+ RadrootsRelayFetchItemBody::Event { .. } | RadrootsRelayFetchItemBody::Eose { .. } => {
+ None
+ }
+ };
+ if let Some(message) = message {
+ let diagnostic_bytes = self.diagnostic_bytes.checked_add(message.len()).ok_or(
+ RadrootsRelayTransportError::DiagnosticLimitExceeded {
+ field: "fetch_request_diagnostics",
+ max: radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES,
+ actual: usize::MAX,
+ },
+ )?;
+ if diagnostic_bytes > radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES {
+ return Err(RadrootsRelayTransportError::DiagnosticLimitExceeded {
+ field: "fetch_request_diagnostics",
+ max: radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES,
+ actual: diagnostic_bytes,
+ });
+ }
+ self.diagnostic_bytes = diagnostic_bytes;
+ }
+ self.items.push(item);
+ Ok(())
+ }
+
+ fn into_items(self) -> Vec<RadrootsRelayFetchItem> {
+ self.items
+ }
+}
+
+async fn emit_relay_fetch_items<A>(
+ adapter: &A,
+ request: &RadrootsRelayFetchRequest,
+ deadline: tokio::time::Instant,
+) -> Result<Vec<RadrootsRelayFetchItem>, RadrootsRelayTransportError>
+where
+ A: RadrootsRelayFetchAdapter,
+{
+ let mut emitter = RadrootsRelayFetchEmitter::new(deadline, request.timeout_ms);
+ tokio::time::timeout_at(deadline, adapter.fetch(request, &mut emitter))
+ .await
+ .map_err(|_| RadrootsRelayTransportError::FetchDeadlineExceeded {
+ timeout_ms: request.timeout_ms,
+ })??;
+ emitter.remaining_duration()?;
+ Ok(emitter.into_items())
}
pub async fn fetch_relay_events<A>(
@@ -1313,23 +1441,31 @@ pub async fn fetch_relay_events<A>(
where
A: RadrootsRelayFetchAdapter,
{
- let target_relays = request.relay_targets.relay_strings();
- let observed_at_ms = request.observed_at_ms;
- let max_events = request.max_events;
- let max_raw_events = request.max_raw_events;
- let max_raw_json_bytes = request.max_raw_json_bytes;
- let filters = request.filters.as_slice().to_vec();
- let items = adapter.fetch(request).await?;
- process_relay_fetch_items(
- target_relays,
- filters,
- observed_at_ms,
- max_events,
- max_raw_events,
- max_raw_json_bytes,
- items,
- )?
- .into_fetched_events_receipt()
+ let timeout_ms = request.timeout_ms;
+ let deadline = tokio::time::Instant::now() + Duration::from_millis(timeout_ms);
+ tokio::time::timeout_at(deadline, async move {
+ let target_relays = request.relay_targets.relay_strings();
+ let observed_at_ms = request.observed_at_ms;
+ let max_events = request.max_events;
+ let max_raw_events = request.max_raw_events;
+ let max_raw_json_bytes = request.max_raw_json_bytes;
+ let filters = request.filters.as_slice().to_vec();
+ let items = emit_relay_fetch_items(adapter, &request, deadline).await?;
+ let receipt = process_relay_fetch_items(
+ target_relays,
+ filters,
+ observed_at_ms,
+ max_events,
+ max_raw_events,
+ max_raw_json_bytes,
+ items,
+ )?
+ .into_fetched_events_receipt()?;
+ remaining_fetch_duration(deadline, timeout_ms)?;
+ Ok(receipt)
+ })
+ .await
+ .map_err(|_| RadrootsRelayTransportError::FetchDeadlineExceeded { timeout_ms })?
}
#[cfg(feature = "runtime-tokio")]
@@ -1356,122 +1492,131 @@ pub async fn fetch_and_ingest_relay_events<A>(
where
A: RadrootsRelayFetchAdapter,
{
- let mode = request.mode;
- let target_relays = request.relay_targets.relay_strings();
- let observed_at_ms = request.observed_at_ms;
- let max_events = request.max_events;
- let max_raw_events = request.max_raw_events;
- let max_raw_json_bytes = request.max_raw_json_bytes;
- let filters = request.filters.as_slice().to_vec();
- let items = adapter.fetch(request).await?;
- let processed = process_relay_fetch_items(
- target_relays,
- filters,
- observed_at_ms,
- max_events,
- max_raw_events,
- max_raw_json_bytes,
- items,
- )?;
- let mut receipt = RadrootsRelayFetchReceipt::from_processed_counts(&processed);
- for item in processed.items {
- match item {
- RadrootsRelayProcessedFetchItem::Receipt(event_receipt) => {
- receipt.events.push(event_receipt);
- }
- RadrootsRelayProcessedFetchItem::Accepted(event)
- | RadrootsRelayProcessedFetchItem::Duplicate(event) => {
- let (relay_url, raw_event, raw_json, observed_at_ms) = event.into_parts();
- let observation_type = match mode {
- RadrootsRelayFetchMode::Fetch => RadrootsTransportObservationType::Fetch,
- RadrootsRelayFetchMode::Subscription => {
- RadrootsTransportObservationType::Subscription
- }
- };
- let observation = RadrootsTransportObservation::new(
- RadrootsTransportKind::Nostr,
- relay_url.clone(),
- observation_type,
- observed_at_ms,
- )?;
- let ingest = match RadrootsEventIngest::from_raw_json(raw_json, observed_at_ms) {
- Ok(ingest) => ingest.with_observation(observation),
- Err(error) => {
- receipt.verification_failed_count += 1;
- receipt.events.push(
- RadrootsRelayFetchEventReceipt {
- relay_url,
- event_id: Some(raw_event.id.to_hex()),
- inserted: false,
- duplicate: false,
- not_persisted: false,
- malformed: false,
- out_of_filter: false,
- skipped_over_limit: false,
- verification: RadrootsRelayFetchEventVerification::Failed,
- admission: RadrootsRelayFetchEventAdmission::NotEvaluated,
- admission_code: None,
- valid_stream: RadrootsRelayFetchEventValidStream::NotEvaluated,
- visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
- message: Some(error.to_string()),
- }
- .checked()?,
- );
- continue;
- }
- };
- let store_receipt = event_store.ingest_event(ingest).await?;
- let admission = relay_fetch_admission(store_receipt.admission_status);
- let valid_stream = if store_receipt.valid_stream_eligible {
- RadrootsRelayFetchEventValidStream::Eligible
- } else {
- RadrootsRelayFetchEventValidStream::Ineligible
- };
- let (inserted, duplicate, not_persisted) = match store_receipt.persistence {
- RadrootsEventPersistence::Inserted { .. } => {
- receipt.inserted_count += 1;
- (true, false, false)
- }
- RadrootsEventPersistence::Duplicate { .. } => {
- receipt.duplicate_count += 1;
- (false, true, false)
- }
- RadrootsEventPersistence::NotPersisted => {
- receipt.not_persisted_count += 1;
- (false, false, true)
+ let timeout_ms = request.timeout_ms;
+ let deadline = tokio::time::Instant::now() + Duration::from_millis(timeout_ms);
+ tokio::time::timeout_at(deadline, async move {
+ let mode = request.mode;
+ let target_relays = request.relay_targets.relay_strings();
+ let observed_at_ms = request.observed_at_ms;
+ let max_events = request.max_events;
+ let max_raw_events = request.max_raw_events;
+ let max_raw_json_bytes = request.max_raw_json_bytes;
+ let filters = request.filters.as_slice().to_vec();
+ let items = emit_relay_fetch_items(adapter, &request, deadline).await?;
+ let processed = process_relay_fetch_items(
+ target_relays,
+ filters,
+ observed_at_ms,
+ max_events,
+ max_raw_events,
+ max_raw_json_bytes,
+ items,
+ )?;
+ let mut receipt = RadrootsRelayFetchReceipt::from_processed_counts(&processed);
+ for item in processed.items {
+ match item {
+ RadrootsRelayProcessedFetchItem::Receipt(event_receipt) => {
+ receipt.events.push(event_receipt);
+ }
+ RadrootsRelayProcessedFetchItem::Accepted(event)
+ | RadrootsRelayProcessedFetchItem::Duplicate(event) => {
+ let (relay_url, raw_event, raw_json, observed_at_ms) = event.into_parts();
+ let observation_type = match mode {
+ RadrootsRelayFetchMode::Fetch => RadrootsTransportObservationType::Fetch,
+ RadrootsRelayFetchMode::Subscription => {
+ RadrootsTransportObservationType::Subscription
+ }
+ };
+ let observation = RadrootsTransportObservation::new(
+ RadrootsTransportKind::Nostr,
+ relay_url.clone(),
+ observation_type,
+ observed_at_ms,
+ )?;
+ let ingest = match RadrootsEventIngest::from_raw_json(raw_json, observed_at_ms)
+ {
+ Ok(ingest) => ingest.with_observation(observation),
+ Err(error) => {
+ receipt.verification_failed_count += 1;
+ receipt.events.push(
+ RadrootsRelayFetchEventReceipt {
+ relay_url,
+ event_id: Some(raw_event.id.to_hex()),
+ inserted: false,
+ duplicate: false,
+ not_persisted: false,
+ malformed: false,
+ out_of_filter: false,
+ skipped_over_limit: false,
+ verification: RadrootsRelayFetchEventVerification::Failed,
+ admission: RadrootsRelayFetchEventAdmission::NotEvaluated,
+ admission_code: None,
+ valid_stream: RadrootsRelayFetchEventValidStream::NotEvaluated,
+ visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
+ message: Some(error.to_string()),
+ }
+ .checked()?,
+ );
+ continue;
+ }
+ };
+ let store_receipt = event_store.ingest_event(ingest).await?;
+ let admission = relay_fetch_admission(store_receipt.admission_status);
+ let valid_stream = if store_receipt.valid_stream_eligible {
+ RadrootsRelayFetchEventValidStream::Eligible
+ } else {
+ RadrootsRelayFetchEventValidStream::Ineligible
+ };
+ let (inserted, duplicate, not_persisted) = match store_receipt.persistence {
+ RadrootsEventPersistence::Inserted { .. } => {
+ receipt.inserted_count += 1;
+ (true, false, false)
+ }
+ RadrootsEventPersistence::Duplicate { .. } => {
+ receipt.duplicate_count += 1;
+ (false, true, false)
+ }
+ RadrootsEventPersistence::NotPersisted => {
+ receipt.not_persisted_count += 1;
+ (false, false, true)
+ }
+ };
+ let visibility = if not_persisted {
+ RadrootsRelayFetchEventVisibility::NotPersisted
+ } else {
+ RadrootsRelayFetchEventVisibility::NotEvaluated
+ };
+ let event_receipt = RadrootsRelayFetchEventReceipt {
+ relay_url,
+ event_id: Some(store_receipt.event_id),
+ inserted,
+ duplicate,
+ not_persisted,
+ malformed: false,
+ out_of_filter: false,
+ skipped_over_limit: false,
+ verification: RadrootsRelayFetchEventVerification::Verified,
+ admission,
+ admission_code: store_receipt.admission_code,
+ valid_stream,
+ visibility,
+ message: None,
}
- };
- let visibility = if not_persisted {
- RadrootsRelayFetchEventVisibility::NotPersisted
- } else {
- RadrootsRelayFetchEventVisibility::NotEvaluated
- };
- let event_receipt = RadrootsRelayFetchEventReceipt {
- relay_url,
- event_id: Some(store_receipt.event_id),
- inserted,
- duplicate,
- not_persisted,
- malformed: false,
- out_of_filter: false,
- skipped_over_limit: false,
- verification: RadrootsRelayFetchEventVerification::Verified,
- admission,
- admission_code: store_receipt.admission_code,
- valid_stream,
- visibility,
- message: None,
+ .checked()?;
+ receipt.events.push(event_receipt);
}
- .checked()?;
- receipt.events.push(event_receipt);
}
}
- }
- receipt.refresh_final_semantic_outcomes(event_store).await?;
- for event in &receipt.events {
- event.clone().checked()?;
- }
- receipt.checked()
+ receipt.refresh_final_semantic_outcomes(event_store).await?;
+ for event in &receipt.events {
+ event.clone().checked()?;
+ }
+ let receipt = receipt.checked()?;
+ remaining_fetch_duration(deadline, timeout_ms)?;
+ Ok(receipt)
+ })
+ .await
+ .map_err(|_| RadrootsRelayTransportError::FetchDeadlineExceeded { timeout_ms })?
}
fn relay_fetch_admission(
@@ -2543,49 +2688,51 @@ impl RadrootsRelayFetchAdapter for RadrootsNostrClientFetchAdapter {
#[cfg_attr(coverage_nightly, coverage(off))]
fn fetch<'a>(
&'a self,
- request: RadrootsRelayFetchRequest,
- ) -> BoxFuture<'a, Result<Vec<RadrootsRelayFetchItem>, RadrootsRelayTransportError>> {
- Box::pin(async move { fetch_from_nostr_relays(request).await })
+ request: &'a RadrootsRelayFetchRequest,
+ emitter: &'a mut RadrootsRelayFetchEmitter,
+ ) -> BoxFuture<'a, Result<(), RadrootsRelayTransportError>> {
+ Box::pin(async move { fetch_from_nostr_relays(request, emitter).await })
}
}
#[cfg_attr(coverage_nightly, coverage(off))]
async fn fetch_from_nostr_relays(
- request: RadrootsRelayFetchRequest,
-) -> Result<Vec<RadrootsRelayFetchItem>, RadrootsRelayTransportError> {
+ request: &RadrootsRelayFetchRequest,
+ emitter: &mut RadrootsRelayFetchEmitter,
+) -> Result<(), RadrootsRelayTransportError> {
if request.filters.as_slice().is_empty() {
return Err(RadrootsRelayTransportError::EmptyFetchFilters);
}
- let timeout = Duration::from_millis(request.timeout_ms);
let filters = request.filters.as_slice().to_vec();
let mut raw_budget =
RadrootsRelayFetchRawBudget::new(request.max_raw_events, request.max_raw_json_bytes);
- let mut items = Vec::new();
let relay_urls = request.relay_targets.relay_strings();
for (relay_index, relay_url) in relay_urls.iter().cloned().enumerate() {
if let Some(reason) = raw_budget.exhaustion_reason() {
for relay_url in relay_urls[relay_index..].iter().cloned() {
- items.push(RadrootsRelayFetchItem::truncated(
+ emitter.emit(RadrootsRelayFetchItem::truncated(
relay_url,
reason.unqueried_relay_message(),
- )?);
+ )?)?;
}
break;
}
- let client = RadrootsNostrClient::new_signerless();
+ let client = request_scoped_nostr_client();
if let Err(error) = client.add_read_relay(relay_url.as_str()).await {
- items.push(RadrootsRelayFetchItem::closed(
+ emitter.emit(RadrootsRelayFetchItem::closed(
relay_url,
- error.to_string(),
- )?);
+ stable_connection_diagnostic(&error.to_string()),
+ )?)?;
+ client.shutdown().await;
continue;
}
- let connection_output = client.try_connect(timeout).await;
+ let connection_output = client.try_connect(emitter.remaining_duration()?).await;
if connection_output.success.is_empty() {
- items.push(RadrootsRelayFetchItem::closed(
+ emitter.emit(RadrootsRelayFetchItem::closed(
relay_url,
- summarize_nostr_output_failures(&connection_output.failed),
- )?);
+ stable_nostr_failure_summary(&connection_output.failed),
+ )?)?;
+ client.shutdown().await;
continue;
}
let mut closed = false;
@@ -2600,7 +2747,7 @@ async fn fetch_from_nostr_relays(
.unwrap_or(raw_budget.remaining_events)
.min(raw_budget.remaining_events);
match client
- .stream_events(filter.limit(filter_limit), timeout)
+ .stream_events(filter.limit(filter_limit), emitter.remaining_duration()?)
.await
{
Ok(mut events) => {
@@ -2623,29 +2770,42 @@ async fn fetch_from_nostr_relays(
));
break;
}
- items.push(RadrootsRelayFetchItem::event(relay_url.clone(), raw_json)?);
+ emitter
+ .emit(RadrootsRelayFetchItem::event(relay_url.clone(), raw_json)?)?;
}
if truncated_message.is_some() {
break;
}
}
Err(error) => {
- items.push(RadrootsRelayFetchItem::closed(
+ emitter.emit(RadrootsRelayFetchItem::closed(
relay_url.clone(),
- error.to_string(),
- )?);
+ stable_connection_diagnostic(&error.to_string()),
+ )?)?;
closed = true;
break;
}
}
}
if let Some(message) = truncated_message {
- items.push(RadrootsRelayFetchItem::truncated(relay_url, message)?);
+ emitter.emit(RadrootsRelayFetchItem::truncated(relay_url, message)?)?;
} else if !closed {
- items.push(unproven_relay_stream_completion(relay_url)?);
+ emitter.emit(unproven_relay_stream_completion(relay_url)?)?;
}
+ client.shutdown().await;
}
- Ok(items)
+ Ok(())
+}
+
+fn remaining_fetch_duration(
+ deadline: tokio::time::Instant,
+ timeout_ms: u64,
+) -> Result<Duration, RadrootsRelayTransportError> {
+ let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+ if remaining.is_zero() {
+ return Err(RadrootsRelayTransportError::FetchDeadlineExceeded { timeout_ms });
+ }
+ Ok(remaining)
}
fn unproven_relay_stream_completion(
@@ -2657,21 +2817,6 @@ fn unproven_relay_stream_completion(
)
}
-fn summarize_nostr_output_failures<K, E>(failed: &std::collections::HashMap<K, E>) -> String
-where
- K: std::fmt::Display + Eq + std::hash::Hash,
- E: std::fmt::Display,
-{
- if failed.is_empty() {
- return "no relay acknowledged the operation".to_owned();
- }
- failed
- .iter()
- .map(|(relay, error)| format!("{relay}: {error}"))
- .collect::<Vec<_>>()
- .join("; ")
-}
-
#[derive(Clone, Default)]
pub struct RadrootsMockRelayFetchAdapter {
items: Arc<Mutex<Vec<RadrootsRelayFetchItem>>>,
@@ -2689,9 +2834,16 @@ impl RadrootsRelayFetchAdapter for RadrootsMockRelayFetchAdapter {
#[cfg_attr(coverage_nightly, coverage(off))]
fn fetch<'a>(
&'a self,
- _request: RadrootsRelayFetchRequest,
- ) -> BoxFuture<'a, Result<Vec<RadrootsRelayFetchItem>, RadrootsRelayTransportError>> {
- Box::pin(async move { Ok(self.items.lock().map_err(fetch_item_lock_error)?.clone()) })
+ _request: &'a RadrootsRelayFetchRequest,
+ emitter: &'a mut RadrootsRelayFetchEmitter,
+ ) -> BoxFuture<'a, Result<(), RadrootsRelayTransportError>> {
+ Box::pin(async move {
+ let items = self.items.lock().map_err(fetch_item_lock_error)?;
+ for item in items.iter().cloned() {
+ emitter.emit(item)?;
+ }
+ Ok(())
+ })
}
}
@@ -2703,19 +2855,24 @@ fn fetch_item_lock_error<T>(_error: PoisonError<T>) -> RadrootsRelayTransportErr
#[cfg(test)]
mod tests {
use super::{
- RadrootsNostrEvent, RadrootsRelayFetchEventAdmission, RadrootsRelayFetchEventValidStream,
- RadrootsRelayFetchEventVerification, RadrootsRelayFetchEventVisibility,
- RadrootsRelayFetchItem, RadrootsRelayFetchRawBudget, RadrootsRelayFetchRawBudgetExhaustion,
- RadrootsRelayTransportError, relay_fetch_event_matches_filters, relay_fetch_visibility,
- required_persisted_fetch_receipt_event_id, summarize_nostr_output_failures,
+ RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX, RadrootsNostrEvent, RadrootsRelayFetchAdapter,
+ RadrootsRelayFetchEmitter, RadrootsRelayFetchEventAdmission,
+ RadrootsRelayFetchEventValidStream, RadrootsRelayFetchEventVerification,
+ RadrootsRelayFetchEventVisibility, RadrootsRelayFetchItem, RadrootsRelayFetchRawBudget,
+ RadrootsRelayFetchRawBudgetExhaustion, RadrootsRelayFetchRequest, RadrootsRelayTargetSet,
+ RadrootsRelayTransportError, fetch_relay_events, relay_fetch_event_matches_filters,
+ relay_fetch_visibility, required_persisted_fetch_receipt_event_id,
unproven_relay_stream_completion,
};
+ use futures::future::{BoxFuture, pending};
use nostr::JsonUtil;
use radroots_event_store::{RadrootsEventVisibility, RadrootsNip09SuppressionReason};
use radroots_nostr::prelude::{
RadrootsNostrFilter, RadrootsNostrKeys, RadrootsNostrKind, RadrootsNostrSecretKey,
};
- use std::collections::HashMap;
+ use std::sync::Arc;
+ use std::sync::atomic::{AtomicBool, Ordering};
+ use std::time::Duration;
const FIXTURE_ALICE_SECRET_KEY_HEX: &str =
"10c5304d6c9ae3a1a16f7860f1cc8f5e3a76225a2663b3a989a0d775919b7df5";
@@ -2730,6 +2887,120 @@ mod tests {
RadrootsNostrEvent::from_json(event.as_json().as_str()).expect("raw event")
}
+ struct FetchCancellationWitness(Arc<AtomicBool>);
+
+ impl Drop for FetchCancellationWitness {
+ fn drop(&mut self) {
+ self.0.store(true, Ordering::SeqCst);
+ }
+ }
+
+ struct HangingFetchAdapter {
+ dropped: Arc<AtomicBool>,
+ }
+
+ impl RadrootsRelayFetchAdapter for HangingFetchAdapter {
+ fn fetch<'a>(
+ &'a self,
+ _request: &'a RadrootsRelayFetchRequest,
+ _emitter: &'a mut RadrootsRelayFetchEmitter,
+ ) -> BoxFuture<'a, Result<(), RadrootsRelayTransportError>> {
+ Box::pin(async move {
+ let _witness = FetchCancellationWitness(Arc::clone(&self.dropped));
+ pending::<()>().await;
+ Ok(())
+ })
+ }
+ }
+
+ struct BlockingFetchAdapter;
+
+ impl RadrootsRelayFetchAdapter for BlockingFetchAdapter {
+ fn fetch<'a>(
+ &'a self,
+ _request: &'a RadrootsRelayFetchRequest,
+ _emitter: &'a mut RadrootsRelayFetchEmitter,
+ ) -> BoxFuture<'a, Result<(), RadrootsRelayTransportError>> {
+ Box::pin(async move {
+ std::thread::sleep(Duration::from_millis(5));
+ Ok(())
+ })
+ }
+ }
+
+ fn fetch_budget_request(timeout_ms: u64) -> RadrootsRelayFetchRequest {
+ RadrootsRelayFetchRequest::subscription(
+ 1,
+ 1,
+ RadrootsRelayTargetSet::new(
+ ["wss://relay.example"],
+ crate::RadrootsRelayUrlPolicy::Public,
+ )
+ .expect("relay target"),
+ [RadrootsNostrFilter::new()],
+ )
+ .and_then(|request| request.with_timeout_ms(timeout_ms))
+ .expect("bounded fetch request")
+ }
+
+ #[tokio::test]
+ async fn fetch_budget_total_deadline_cancels_subscription_adapter() {
+ let dropped = Arc::new(AtomicBool::new(false));
+ let error = fetch_relay_events(
+ &HangingFetchAdapter {
+ dropped: Arc::clone(&dropped),
+ },
+ fetch_budget_request(1),
+ )
+ .await
+ .expect_err("hanging subscription must reach the total deadline");
+
+ assert!(matches!(
+ error,
+ RadrootsRelayTransportError::FetchDeadlineExceeded { timeout_ms: 1 }
+ ));
+ assert!(dropped.load(Ordering::SeqCst));
+ }
+
+ #[tokio::test]
+ async fn fetch_budget_total_deadline_rejects_noncooperative_adapter_completion() {
+ let error = fetch_relay_events(&BlockingFetchAdapter, fetch_budget_request(1))
+ .await
+ .expect_err("blocking adapter completion after the deadline must be rejected");
+
+ assert!(matches!(
+ error,
+ RadrootsRelayTransportError::FetchDeadlineExceeded { timeout_ms: 1 }
+ ));
+ }
+
+ #[test]
+ fn fetch_budget_emitter_rejects_one_over_before_retention() {
+ let mut emitter = RadrootsRelayFetchEmitter::new(
+ tokio::time::Instant::now() + Duration::from_secs(1),
+ 1_000,
+ );
+ for _ in 0..RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX {
+ emitter
+ .emit(
+ RadrootsRelayFetchItem::event("wss://relay.example", "{}")
+ .expect("bounded event item"),
+ )
+ .expect("exact item budget");
+ }
+ assert!(matches!(
+ emitter.emit(
+ RadrootsRelayFetchItem::event("wss://relay.example", "{}")
+ .expect("bounded event item")
+ ),
+ Err(RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "raw_item_count",
+ max: RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX,
+ actual,
+ }) if actual == RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX + 1
+ ));
+ }
+
#[test]
fn relay_fetch_filter_helper_rejects_empty_filter_set() {
let event = signed_raw_event();
@@ -2741,23 +3012,6 @@ mod tests {
}
#[test]
- fn nostr_output_failure_summary_covers_empty_and_reported_failures() {
- assert_eq!(
- summarize_nostr_output_failures::<String, String>(&HashMap::new()),
- "no relay acknowledged the operation"
- );
-
- let mut failures = HashMap::new();
- failures.insert("wss://relay.example.com".to_owned(), "timeout".to_owned());
- failures.insert("wss://relay-2.example.com".to_owned(), "denied".to_owned());
-
- let summary = summarize_nostr_output_failures(&failures);
- assert!(summary.contains("wss://relay.example.com: timeout"));
- assert!(summary.contains("wss://relay-2.example.com: denied"));
- assert!(summary.contains("; "));
- }
-
- #[test]
fn unproven_sdk_stream_completion_never_claims_eose() {
assert_eq!(
unproven_relay_stream_completion("wss://relay.example".to_owned())
diff --git a/crates/transport_nostr/src/lib.rs b/crates/transport_nostr/src/lib.rs
@@ -1,6 +1,10 @@
#![cfg_attr(coverage_nightly, feature(coverage_attribute))]
#![forbid(unsafe_code)]
+#[cfg(feature = "client")]
+mod connector;
+#[cfg(feature = "client")]
+mod diagnostic;
mod error;
#[cfg(feature = "storage")]
mod fetch;
@@ -19,13 +23,14 @@ pub use fetch::{
RADROOTS_RELAY_FETCH_FILTER_LIMIT_MAX, RADROOTS_RELAY_FETCH_FILTER_SET_JSON_BYTE_LIMIT_MAX,
RADROOTS_RELAY_FETCH_RAW_EVENT_LIMIT_MAX, RADROOTS_RELAY_FETCH_RAW_JSON_BYTE_LIMIT_MAX,
RADROOTS_RELAY_FETCH_TIMEOUT_MS_MAX, RadrootsMockRelayFetchAdapter,
- RadrootsNostrClientFetchAdapter, RadrootsRelayFetchAdapter, RadrootsRelayFetchEventAdmission,
- RadrootsRelayFetchEventReceipt, RadrootsRelayFetchEventValidStream,
- RadrootsRelayFetchEventVerification, RadrootsRelayFetchEventVisibility,
- RadrootsRelayFetchFailure, RadrootsRelayFetchFilters, RadrootsRelayFetchItem,
- RadrootsRelayFetchMode, RadrootsRelayFetchOutcomeKind, RadrootsRelayFetchReceipt,
- RadrootsRelayFetchRelayOutcome, RadrootsRelayFetchRequest, RadrootsRelayFetchedEvent,
- RadrootsRelayFetchedEventsReceipt, fetch_and_ingest_relay_events, fetch_relay_events,
+ RadrootsNostrClientFetchAdapter, RadrootsRelayFetchAdapter, RadrootsRelayFetchEmitter,
+ RadrootsRelayFetchEventAdmission, RadrootsRelayFetchEventReceipt,
+ RadrootsRelayFetchEventValidStream, RadrootsRelayFetchEventVerification,
+ RadrootsRelayFetchEventVisibility, RadrootsRelayFetchFailure, RadrootsRelayFetchFilters,
+ RadrootsRelayFetchItem, RadrootsRelayFetchMode, RadrootsRelayFetchOutcomeKind,
+ RadrootsRelayFetchReceipt, RadrootsRelayFetchRelayOutcome, RadrootsRelayFetchRequest,
+ RadrootsRelayFetchedEvent, RadrootsRelayFetchedEventsReceipt, fetch_and_ingest_relay_events,
+ fetch_relay_events,
};
#[cfg(feature = "storage")]
pub use outbox::{
diff --git a/crates/transport_nostr/src/outcome.rs b/crates/transport_nostr/src/outcome.rs
@@ -186,37 +186,18 @@ impl RadrootsRelayOutcome {
pub fn classify(message: impl AsRef<str>) -> Result<Self, crate::RadrootsRelayTransportError> {
let message = message.as_ref().trim();
- let lower = message.to_ascii_lowercase();
- let kind = if lower.starts_with("duplicate:") {
- RadrootsRelayOutcomeKind::DuplicateAccepted
- } else if lower.starts_with("blocked:") {
- RadrootsRelayOutcomeKind::Blocked
- } else if lower.starts_with("rate-limited:") {
- RadrootsRelayOutcomeKind::RateLimited
- } else if lower.starts_with("invalid:") {
- RadrootsRelayOutcomeKind::Invalid
- } else if lower.starts_with("pow:") {
- RadrootsRelayOutcomeKind::PowRequired
- } else if lower.starts_with("restricted:") {
- RadrootsRelayOutcomeKind::Restricted
- } else if lower.starts_with("auth-required:") {
- RadrootsRelayOutcomeKind::AuthRequired
- } else if lower.starts_with("mute:") {
- RadrootsRelayOutcomeKind::Muted
- } else if lower.starts_with("unsupported:") {
- RadrootsRelayOutcomeKind::Unsupported
- } else if lower.starts_with("payment-required:") {
- RadrootsRelayOutcomeKind::PaymentRequired
- } else if lower.starts_with("error:") {
- RadrootsRelayOutcomeKind::Error
- } else if lower.starts_with("timeout:") {
- RadrootsRelayOutcomeKind::Timeout
- } else {
- RadrootsRelayOutcomeKind::Unknown
- };
+ let kind = classify_relay_outcome_kind(message);
Self::try_new(kind, Some(message.to_owned()))
}
+ #[cfg(feature = "client")]
+ pub(crate) fn classify_redacted(
+ message: impl AsRef<str>,
+ ) -> Result<Self, crate::RadrootsRelayTransportError> {
+ let kind = classify_relay_outcome_kind(message.as_ref().trim());
+ Self::try_new(kind, Some(format!("relay-{}", kind.as_str())))
+ }
+
pub fn kind(&self) -> RadrootsRelayOutcomeKind {
self.kind
}
@@ -246,6 +227,43 @@ impl RadrootsRelayOutcome {
}
}
+fn classify_relay_outcome_kind(message: &str) -> RadrootsRelayOutcomeKind {
+ if starts_with_ascii_case_insensitive(message, "duplicate:") {
+ RadrootsRelayOutcomeKind::DuplicateAccepted
+ } else if starts_with_ascii_case_insensitive(message, "blocked:") {
+ RadrootsRelayOutcomeKind::Blocked
+ } else if starts_with_ascii_case_insensitive(message, "rate-limited:") {
+ RadrootsRelayOutcomeKind::RateLimited
+ } else if starts_with_ascii_case_insensitive(message, "invalid:") {
+ RadrootsRelayOutcomeKind::Invalid
+ } else if starts_with_ascii_case_insensitive(message, "pow:") {
+ RadrootsRelayOutcomeKind::PowRequired
+ } else if starts_with_ascii_case_insensitive(message, "restricted:") {
+ RadrootsRelayOutcomeKind::Restricted
+ } else if starts_with_ascii_case_insensitive(message, "auth-required:") {
+ RadrootsRelayOutcomeKind::AuthRequired
+ } else if starts_with_ascii_case_insensitive(message, "mute:") {
+ RadrootsRelayOutcomeKind::Muted
+ } else if starts_with_ascii_case_insensitive(message, "unsupported:") {
+ RadrootsRelayOutcomeKind::Unsupported
+ } else if starts_with_ascii_case_insensitive(message, "payment-required:") {
+ RadrootsRelayOutcomeKind::PaymentRequired
+ } else if starts_with_ascii_case_insensitive(message, "error:") {
+ RadrootsRelayOutcomeKind::Error
+ } else if starts_with_ascii_case_insensitive(message, "timeout:") {
+ RadrootsRelayOutcomeKind::Timeout
+ } else {
+ RadrootsRelayOutcomeKind::Unknown
+ }
+}
+
+fn starts_with_ascii_case_insensitive(value: &str, prefix: &str) -> bool {
+ value
+ .as_bytes()
+ .get(..prefix.len())
+ .is_some_and(|candidate| candidate.eq_ignore_ascii_case(prefix.as_bytes()))
+}
+
fn ensure_relay_outcome_message(message: &str) -> Result<(), crate::RadrootsRelayTransportError> {
if message.len() > RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES {
return Err(
diff --git a/crates/transport_nostr/src/publish.rs b/crates/transport_nostr/src/publish.rs
@@ -1,5 +1,9 @@
#![forbid(unsafe_code)]
+#[cfg(feature = "client")]
+use crate::connector::request_scoped_nostr_client;
+#[cfg(feature = "client")]
+use crate::diagnostic::stable_connection_diagnostic;
use crate::error::ensure_nonnegative_timestamp;
use crate::{RadrootsRelayOutcome, RadrootsRelayTargetSet, RadrootsRelayTransportError};
#[cfg(feature = "client")]
@@ -10,6 +14,8 @@ use radroots_event::{
ids::RadrootsEventId,
wire::RadrootsNip01EventWire,
};
+#[cfg(feature = "client")]
+use radroots_transport::RADROOTS_TRANSPORT_TOTAL_DEADLINE_MAX_MS;
use radroots_transport::{
RadrootsTransport, RadrootsTransportCapabilities, RadrootsTransportDeliveryReceipt,
RadrootsTransportDeliveryRequest, RadrootsTransportError, RadrootsTransportFetchReceipt,
@@ -31,6 +37,9 @@ use radroots_nostr::prelude::{RadrootsNostrClient, RadrootsNostrEvent};
#[cfg(feature = "client")]
const RELAY_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+#[cfg(feature = "client")]
+const RELAY_PUBLISH_TOTAL_TIMEOUT: Duration =
+ Duration::from_millis(RADROOTS_TRANSPORT_TOTAL_DEADLINE_MAX_MS);
pub const RADROOTS_RELAY_PUBLISH_IDEMPOTENCY_KEY_MAX_BYTES: usize = 256;
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -326,6 +335,7 @@ impl RadrootsRelayPublishReceipt {
),
});
}
+ validate_publish_diagnostic_budget(&relays)?;
let mut canonical_relays = Vec::with_capacity(relays.len());
for receipt in &relays {
let canonical = RadrootsTransportTarget::nostr_relay(receipt.relay_url())?
@@ -422,6 +432,32 @@ impl RadrootsRelayPublishReceipt {
}
}
+fn validate_publish_diagnostic_budget(
+ relays: &[RadrootsRelayPublishRelayReceipt],
+) -> Result<(), RadrootsRelayTransportError> {
+ let max = radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES;
+ let mut actual = 0usize;
+ for receipt in relays {
+ if let Some(message) = receipt.outcome().message() {
+ actual = actual.checked_add(message.len()).ok_or(
+ RadrootsRelayTransportError::DiagnosticLimitExceeded {
+ field: "publish_request_diagnostics",
+ max,
+ actual: usize::MAX,
+ },
+ )?;
+ if actual > max {
+ return Err(RadrootsRelayTransportError::DiagnosticLimitExceeded {
+ field: "publish_request_diagnostics",
+ max,
+ actual,
+ });
+ }
+ }
+ }
+ Ok(())
+}
+
struct BoundedPublishRelayReceipts;
impl<'de> de::Visitor<'de> for BoundedPublishRelayReceipts {
@@ -677,6 +713,7 @@ fn nostr_error_to_transport_error(error: RadrootsRelayTransportError) -> Radroot
}
RadrootsRelayTransportError::Transport(_) => RadrootsTransportError::InvalidTransportKind,
RadrootsRelayTransportError::EmptyFetchFilters
+ | RadrootsRelayTransportError::FetchDeadlineExceeded { .. }
| RadrootsRelayTransportError::InvalidFetchLimit { .. }
| RadrootsRelayTransportError::FetchLimitTooLarge { .. }
| RadrootsRelayTransportError::InvalidFetchReceipt { .. }
@@ -1199,16 +1236,20 @@ fn captured_raw_event_lock_error<T>(_error: PoisonError<T>) -> RadrootsRelayTran
}
#[cfg(feature = "client")]
-#[derive(Clone)]
-pub struct RadrootsNostrClientPublishAdapter {
- client: RadrootsNostrClient,
-}
+#[derive(Clone, Copy, Debug, Default)]
+pub struct RadrootsNostrClientPublishAdapter;
#[cfg(feature = "client")]
impl RadrootsNostrClientPublishAdapter {
#[cfg_attr(coverage_nightly, coverage(off))]
- pub fn new(client: RadrootsNostrClient) -> Self {
- Self { client }
+ /// Compatibility constructor that consumes, but never reuses, the supplied
+ /// client so each publication owns an isolated relay pool.
+ pub fn new(_client: RadrootsNostrClient) -> Self {
+ Self
+ }
+
+ pub const fn request_scoped() -> Self {
+ Self
}
}
@@ -1221,115 +1262,145 @@ impl RadrootsRelayPublishAdapter for RadrootsNostrClientPublishAdapter {
) -> BoxFuture<'a, Result<Vec<RadrootsRelayPublishRelayReceipt>, RadrootsRelayTransportError>>
{
Box::pin(async move {
- let event =
- RadrootsNostrEvent::from_json(request.signed_event.signed_event().raw_json())
+ let timeout_targets = request.targets.relay_strings();
+ let execution = async move {
+ let client = request_scoped_nostr_client();
+ let result = async {
+ let event = RadrootsNostrEvent::from_json(
+ request.signed_event.signed_event().raw_json(),
+ )
.map_err(|error| {
RadrootsRelayTransportError::NostrEventJson(error.to_string())
})?;
- ensure_raw_event_matches_signed_event(&event, request.signed_event.signed_event())?;
- let target_strings = request.targets.relay_strings();
- for relay_url in &target_strings {
- self.client
- .add_write_relay(relay_url.as_str())
- .await
- .map_err(|error| RadrootsRelayTransportError::Transport(error.to_string()))?;
- }
- let connection_output = self.client.try_connect(RELAY_CONNECT_TIMEOUT).await;
- let target_url_set = target_strings
- .iter()
- .map(|relay_url| relay_url.trim_end_matches('/').to_owned())
- .collect::<BTreeSet<_>>();
- let connected_strings = self
- .client
- .relays()
- .await
- .into_values()
- .filter(|relay| relay.is_connected())
- .map(|relay| relay.url().to_string())
- .filter(|relay_url| target_url_set.contains(relay_url.trim_end_matches('/')))
- .collect::<Vec<_>>();
- let connection_failures = connection_output
- .failed
- .iter()
- .map(|(relay, reason)| {
- (
- relay.to_string().trim_end_matches('/').to_owned(),
- reason.clone(),
- )
- })
- .collect::<BTreeMap<_, _>>();
- if connected_strings.is_empty() {
- return target_strings
+ ensure_raw_event_matches_signed_event(
+ &event,
+ request.signed_event.signed_event(),
+ )?;
+ let target_strings = request.targets.relay_strings();
+ for relay_url in &target_strings {
+ client
+ .add_write_relay(relay_url.as_str())
+ .await
+ .map_err(|error| {
+ RadrootsRelayTransportError::Transport(
+ stable_connection_diagnostic(&error.to_string()).to_owned(),
+ )
+ })?;
+ }
+ let connection_output = client.try_connect(RELAY_CONNECT_TIMEOUT).await;
+ let target_url_set = target_strings
+ .iter()
+ .map(|relay_url| relay_url.trim_end_matches('/').to_owned())
+ .collect::<BTreeSet<_>>();
+ let connected_strings = client
+ .relays()
+ .await
+ .into_values()
+ .filter(|relay| relay.is_connected())
+ .map(|relay| relay.url().to_string())
+ .filter(|relay_url| {
+ target_url_set.contains(relay_url.trim_end_matches('/'))
+ })
+ .collect::<Vec<_>>();
+ let connection_failures = connection_output
+ .failed
+ .iter()
+ .map(|(relay, reason)| {
+ (
+ relay.to_string().trim_end_matches('/').to_owned(),
+ stable_connection_diagnostic(reason).to_owned(),
+ )
+ })
+ .collect::<BTreeMap<_, _>>();
+ if connected_strings.is_empty() {
+ return target_strings
+ .into_iter()
+ .map(|relay_url| {
+ let target_url = relay_url.trim_end_matches('/');
+ let reason = connection_failures
+ .get(target_url)
+ .cloned()
+ .unwrap_or_else(|| "connection-failed".to_owned());
+ RadrootsRelayPublishRelayReceipt::attempted(
+ relay_url,
+ RadrootsRelayOutcome::connection_failed(reason)?,
+ )
+ })
+ .collect();
+ }
+ let output = match client.send_event_to(connected_strings, &event).await {
+ Ok(output) => output,
+ Err(error) => {
+ let message =
+ stable_connection_diagnostic(&error.to_string()).to_owned();
+ return target_strings
+ .into_iter()
+ .map(|relay_url| {
+ RadrootsRelayPublishRelayReceipt::attempted(
+ relay_url,
+ RadrootsRelayOutcome::connection_failed(message.clone())?,
+ )
+ })
+ .collect();
+ }
+ };
+ let mut receipts = Vec::new();
+ for relay_url in &target_strings {
+ let target_url = relay_url.trim_end_matches('/');
+ let success = output.success.iter().any(|success_url| {
+ success_url.to_string().trim_end_matches('/') == target_url
+ });
+ if success {
+ receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
+ relay_url,
+ RadrootsRelayOutcome::accepted_with_message(
+ "nostr-relay-pool-success-ok-message-unavailable",
+ )?,
+ )?);
+ continue;
+ }
+ if let Some(reason) = connection_failures.get(target_url) {
+ receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
+ relay_url,
+ RadrootsRelayOutcome::connection_failed(reason.clone())?,
+ )?);
+ continue;
+ }
+ let failed = output.failed.iter().find_map(|(failed_url, message)| {
+ if failed_url.to_string().trim_end_matches('/') == target_url {
+ Some(message.clone())
+ } else {
+ None
+ }
+ });
+ let outcome = failed
+ .map(RadrootsRelayOutcome::classify_redacted)
+ .transpose()?
+ .unwrap_or(RadrootsRelayOutcome::classify_redacted(
+ "error: relay output omitted target",
+ )?);
+ receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
+ relay_url, outcome,
+ )?);
+ }
+ Ok(receipts)
+ }
+ .await;
+ client.shutdown().await;
+ result
+ };
+ match tokio::time::timeout(RELAY_PUBLISH_TOTAL_TIMEOUT, execution).await {
+ Ok(result) => result,
+ Err(_) => timeout_targets
.into_iter()
.map(|relay_url| {
- let target_url = relay_url.trim_end_matches('/');
- let reason = connection_failures
- .get(target_url)
- .cloned()
- .unwrap_or_else(|| "relay did not connect".to_owned());
RadrootsRelayPublishRelayReceipt::attempted(
relay_url,
- RadrootsRelayOutcome::connection_failed(reason)?,
+ RadrootsRelayOutcome::timeout("publish-total-deadline-exceeded")?,
)
})
- .collect();
- }
- let output = match self.client.send_event_to(connected_strings, &event).await {
- Ok(output) => output,
- Err(error) => {
- let message = error.to_string();
- return target_strings
- .into_iter()
- .map(|relay_url| {
- RadrootsRelayPublishRelayReceipt::attempted(
- relay_url,
- RadrootsRelayOutcome::connection_failed(message.clone())?,
- )
- })
- .collect();
- }
- };
- let mut receipts = Vec::new();
- for relay_url in &target_strings {
- let target_url = relay_url.trim_end_matches('/');
- let success = output
- .success
- .iter()
- .any(|success_url| success_url.to_string().trim_end_matches('/') == target_url);
- if success {
- receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
- relay_url,
- RadrootsRelayOutcome::accepted_with_message(
- "nostr-relay-pool-success-ok-message-unavailable",
- )?,
- )?);
- continue;
- }
- if let Some(reason) = connection_failures.get(target_url) {
- receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
- relay_url,
- RadrootsRelayOutcome::connection_failed(reason.clone())?,
- )?);
- continue;
- }
- let failed = output.failed.iter().find_map(|(failed_url, message)| {
- if failed_url.to_string().trim_end_matches('/') == target_url {
- Some(message.clone())
- } else {
- None
- }
- });
- let outcome = failed
- .map(RadrootsRelayOutcome::classify)
- .transpose()?
- .unwrap_or(RadrootsRelayOutcome::classify(
- "error: relay output omitted target",
- )?);
- receipts.push(RadrootsRelayPublishRelayReceipt::attempted(
- relay_url, outcome,
- )?);
+ .collect(),
}
- Ok(receipts)
})
}
}
@@ -1389,7 +1460,11 @@ fn ensure_raw_event_matches_signed_event(
#[cfg(all(test, feature = "client"))]
mod tests {
- use super::{RadrootsNostrEvent, ensure_raw_event_matches_signed_event};
+ use super::{
+ RadrootsNostrEvent, RadrootsRelayOutcome, RadrootsRelayPublishReceipt,
+ RadrootsRelayPublishRelayReceipt, RadrootsRelayTransportError,
+ ensure_raw_event_matches_signed_event,
+ };
use nostr::JsonUtil;
use radroots_event::draft::{RadrootsEventDraft, RadrootsSignedEvent};
use radroots_event::kinds::KIND_GEOCHAT;
@@ -1463,4 +1538,35 @@ mod tests {
wire.tags.push(vec!["t".to_owned(), "compost".to_owned()]);
assert_mismatch(raw_event_from_wire(wire), &signed_event);
}
+
+ #[test]
+ fn relay_security_publish_diagnostics_share_one_complete_receipt_budget() {
+ let max = radroots_transport::RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES;
+ let receipt = |relay: &str, message: String| {
+ RadrootsRelayPublishRelayReceipt::attempted(
+ relay,
+ RadrootsRelayOutcome::unknown(message).expect("bounded outcome"),
+ )
+ .expect("relay receipt")
+ };
+ let exact = vec![
+ receipt("wss://relay-a.example", "a".repeat(max / 2)),
+ receipt("wss://relay-b.example", "b".repeat(max / 2)),
+ ];
+ RadrootsRelayPublishReceipt::new("00".repeat(32), 1, false, exact)
+ .expect("exact publish diagnostic budget");
+
+ let one_over = vec![
+ receipt("wss://relay-a.example", "a".repeat(max / 2)),
+ receipt("wss://relay-b.example", "b".repeat(max / 2 + 1)),
+ ];
+ assert!(matches!(
+ RadrootsRelayPublishReceipt::new("00".repeat(32), 1, false, one_over),
+ Err(RadrootsRelayTransportError::DiagnosticLimitExceeded {
+ field: "publish_request_diagnostics",
+ max: observed_max,
+ actual,
+ }) if observed_max == max && actual == max + 1
+ ));
+ }
}
