commit 89fbbd82b03f32dcc226053f6f412e6ac2abb4c9
parent 62fd018c1d76d4d894af02e0d3edc9d6fd242ee3
Author: triesap <tyson@radroots.org>
Date: Sun, 23 Aug 2026 05:45:08 +0000
transport-nostr: split delivery preparation from execution
- seal exact prepared delivery behind inert adapter validation
- consume configuration-bound authority only at relay execution
- preserve lazy EventSink compatibility and existing outcome behavior
- verify adapter, contract, API, Clippy, doctest, and Rustdoc gates
Diffstat:
5 files changed, 258 insertions(+), 75 deletions(-)
diff --git a/contracts/api_baselines/radroots_transport_nostr.txt b/contracts/api_baselines/radroots_transport_nostr.txt
@@ -98,6 +98,9 @@ impl radroots_transport_nostr::NostrTransport
pub const fn radroots_transport_nostr::NostrTransport::config(&self) -> &radroots_transport_nostr::Config
pub fn radroots_transport_nostr::NostrTransport::new(radroots_transport_nostr::Config) -> Self
pub fn radroots_transport_nostr::NostrTransport::relay_status(&self) -> radroots_transport_nostr::RelayStatusReport
+impl radroots_transport_nostr::NostrTransport
+pub fn radroots_transport_nostr::NostrTransport::execute_prepared_delivery(&self, radroots_transport_nostr::PreparedDelivery) -> radroots_transport::source::BoxFuture<'_, core::result::Result<radroots_transport::sink::DeliveryReceipt, radroots_transport::sink::SinkFailure>>
+pub fn radroots_transport_nostr::NostrTransport::prepare_delivery(&self, radroots_transport::sink::DeliveryRequest) -> core::result::Result<radroots_transport_nostr::PreparedDelivery, alloc::boxed::Box<radroots_transport::sink::SinkFailure>>
impl core::fmt::Debug for radroots_transport_nostr::NostrTransport
pub fn radroots_transport_nostr::NostrTransport::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl radroots_transport::sink::EventSink for radroots_transport_nostr::NostrTransport
@@ -108,6 +111,11 @@ pub fn radroots_transport_nostr::NostrTransport::fetch(&self, radroots_transport
pub fn radroots_transport_nostr::NostrTransport::status(&self) -> radroots_transport::source::BoxFuture<'_, core::result::Result<radroots_transport::status::SourceStatus, radroots_transport::error::Error>>
impl radroots_transport::source::EventSubscriber for radroots_transport_nostr::NostrTransport
pub fn radroots_transport_nostr::NostrTransport::subscribe(&self, radroots_transport::source::SubscriptionRequest) -> radroots_transport::source::BoxFuture<'_, core::result::Result<radroots_transport::source::BoxSubscription, radroots_transport::error::Error>>
+pub struct radroots_transport_nostr::PreparedDelivery
+impl radroots_transport_nostr::PreparedDelivery
+pub const fn radroots_transport_nostr::PreparedDelivery::request(&self) -> &radroots_transport::sink::DeliveryRequest
+impl core::fmt::Debug for radroots_transport_nostr::PreparedDelivery
+pub fn radroots_transport_nostr::PreparedDelivery::fmt(&self, &mut core::fmt::Formatter<'_>) -> core::fmt::Result
pub struct radroots_transport_nostr::ReconnectBackoff
impl radroots_transport_nostr::ReconnectBackoff
pub const fn radroots_transport_nostr::ReconnectBackoff::initial_delay_ms(self) -> u64
diff --git a/crates/transport_nostr/README.md b/crates/transport_nostr/README.md
@@ -55,6 +55,31 @@ and `DeliveryRequest` values from `radroots_transport`, polls the returned
futures on its executor, and applies any retry or scheduling policy outside
this crate.
+## Prepared delivery boundary
+
+[`NostrTransport::prepare_delivery`] validates the exact request, writable
+relay bindings, and signed-event conversion without reading a clock, polling
+status, or performing relay I/O. It returns a sealed [`PreparedDelivery`]
+whose ordinary `Debug` is redacted. The composing host may bind the retained
+request to durable Submitted state and then pass the capability to
+[`NostrTransport::execute_prepared_delivery`], which consumes it and is the
+only half of this boundary that may contact relays. Executing a capability
+through a differently configured transport fails closed.
+
+Callers cannot forge or mutate prepared authority:
+
+```compile_fail
+use radroots_transport_nostr::PreparedDelivery;
+
+let _forged = PreparedDelivery {
+ request: panic!(),
+ config: panic!(),
+ event: panic!(),
+ authorized: panic!(),
+ skipped: panic!(),
+};
+```
+
## Public surface
- [`RelayProfile`] defines public, loopback-simulator, and physical-device
@@ -69,6 +94,8 @@ this crate.
used by scoped fetch continuation cursors.
- [`NostrTransport`] implements all three transport SPIs, exposes passive typed
per-relay evidence, and provides explicit NIP-42 challenge lifecycle methods.
+- [`PreparedDelivery`] is the non-forgeable, consuming boundary between inert
+ adapter validation and relay execution.
- [`Error`] contains only package-owned validation and authentication errors;
upstream failures are normalized before crossing the public boundary.
diff --git a/crates/transport_nostr/src/lib.rs b/crates/transport_nostr/src/lib.rs
@@ -18,6 +18,7 @@ pub use cursor::RelayCursor;
pub use error::Error;
pub use profile::{RelayAccess, RelayEndpoint, RelayProfile, RelayProfileKind};
pub use relay::{RelayUrl, RelayUrlPolicy};
+pub use sink::PreparedDelivery;
pub use status::{
RelayAggregateState, RelayCapabilityEvidence, RelayEvidenceState, RelayStatus,
RelayStatusReport,
diff --git a/crates/transport_nostr/src/sink.rs b/crates/transport_nostr/src/sink.rs
@@ -1,11 +1,11 @@
//! Nostr implementation of the transport event sink.
use crate::{NostrTransport, RelayUrl, status};
-use core::time::Duration;
+use core::{fmt, time::Duration};
use futures::{StreamExt, stream};
use radroots_nostr::event::Event;
use radroots_transport::{
- BoxFuture, DeliveryReceipt, DeliveryRequest, EventSink, SinkFailure,
+ BoxFuture, DeliveryReceipt, DeliveryRequest, EventSink, SinkFailure, Target,
outcome::DeliveryOutcome,
sink::{DeliveryTargetReceipt, SinkStatus},
};
@@ -17,6 +17,35 @@ pub(crate) struct RelayPublishResult {
outcome: DeliveryOutcome,
}
+/// Sealed, no-I/O result of validating one delivery against this adapter.
+///
+/// The value retains the exact request and converted signed event. It is
+/// constructed only by [`NostrTransport::prepare_delivery`] and is consumed by
+/// [`NostrTransport::execute_prepared_delivery`]. Ordinary `Debug` never
+/// exposes event bytes, request identities, or relay destinations.
+#[must_use = "prepared delivery must be durably bound before execution or deliberately discarded"]
+pub struct PreparedDelivery {
+ request: DeliveryRequest,
+ config: crate::Config,
+ event: Event,
+ authorized: Vec<(RelayUrl, Target)>,
+ skipped: Vec<DeliveryTargetReceipt>,
+}
+
+impl PreparedDelivery {
+ /// Returns the exact validated request retained for persistence binding.
+ #[must_use]
+ pub const fn request(&self) -> &DeliveryRequest {
+ &self.request
+ }
+}
+
+impl fmt::Debug for PreparedDelivery {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("PreparedDelivery([redacted])")
+ }
+}
+
pub(crate) trait RelayClient: Send + Sync {
fn publish<'a>(
&'a self,
@@ -96,60 +125,77 @@ impl RelayClient for LiveRelayClient {
}
}
-impl EventSink for NostrTransport {
- fn status(&self) -> BoxFuture<'_, Result<SinkStatus, radroots_transport::Error>> {
- Box::pin(async move { Ok(status::sink_status(&self.status)) })
+impl NostrTransport {
+ /// Validates and converts one delivery without reading a clock or performing relay I/O.
+ pub fn prepare_delivery(
+ &self,
+ request: DeliveryRequest,
+ ) -> Result<PreparedDelivery, Box<SinkFailure>> {
+ let mut authorized = Vec::new();
+ let mut skipped = Vec::new();
+ for target in request.target_set().targets() {
+ match self.config().endpoint_for_target(target) {
+ Some(endpoint) if endpoint.access().can_write() => {
+ authorized.push((endpoint.url().clone(), target.clone()));
+ }
+ None | Some(_) => skipped.push(
+ DeliveryTargetReceipt::skipped(
+ target.clone(),
+ DeliveryOutcome::rejected()
+ .with_detail("target_denied", "target is not configured for this sink")
+ .map_err(|_| Box::new(SinkFailure::invalid_contract(&request)))?,
+ )
+ .map_err(|_| Box::new(SinkFailure::invalid_contract(&request)))?,
+ ),
+ }
+ }
+ let event = radroots_nostr::event::to_nostr(request.payload().event().envelope())
+ .map_err(|_| Box::new(SinkFailure::invalid_contract(&request)))?;
+ Ok(PreparedDelivery {
+ request,
+ config: self.config().clone(),
+ event,
+ authorized,
+ skipped,
+ })
}
- fn deliver(
+ /// Performs relay I/O for one exact prepared delivery and consumes its authority.
+ pub fn execute_prepared_delivery(
&self,
- request: DeliveryRequest,
+ prepared: PreparedDelivery,
) -> BoxFuture<'_, Result<DeliveryReceipt, SinkFailure>> {
Box::pin(async move {
+ let PreparedDelivery {
+ request,
+ config,
+ event,
+ authorized,
+ mut skipped,
+ } = prepared;
+ if config != *self.config() {
+ return Err(SinkFailure::invalid_contract(&request));
+ }
let now_unix_ms = unix_time_ms();
let mut requested = Vec::new();
- let mut skipped = Vec::new();
- for target in request.target_set().targets() {
- match self.config().endpoint_for_target(target) {
- Some(endpoint) if endpoint.access().can_write() => {
- let relay = endpoint.url().clone();
- if self.status.may_write(&relay, now_unix_ms) {
- requested.push((relay, target.clone()));
- } else {
- skipped.push(
- DeliveryTargetReceipt::skipped(
- target.clone(),
- DeliveryOutcome::unavailable()
- .with_detail(
- "reconnect_backoff",
- "relay reconnect backoff is active",
- )
- .map_err(|_| SinkFailure::invalid_contract(&request))?,
- )
- .map_err(|_| SinkFailure::invalid_contract(&request))?,
- );
- }
- }
- None | Some(_) => skipped.push(
+ for (relay, target) in authorized {
+ if self.status.may_write(&relay, now_unix_ms) {
+ requested.push((relay, target));
+ } else {
+ skipped.push(
DeliveryTargetReceipt::skipped(
- target.clone(),
- DeliveryOutcome::rejected()
+ target,
+ DeliveryOutcome::unavailable()
.with_detail(
- "target_denied",
- "target is not configured for this sink",
+ "reconnect_backoff",
+ "relay reconnect backoff is active",
)
.map_err(|_| SinkFailure::invalid_contract(&request))?,
)
.map_err(|_| SinkFailure::invalid_contract(&request))?,
- ),
+ );
}
}
-
- let event = match radroots_nostr::event::to_nostr(request.payload().event().envelope())
- {
- Ok(event) => event,
- Err(_) => return Err(SinkFailure::invalid_contract(&request)),
- };
let remaining_ms = request.deadline_unix_ms().saturating_sub(now_unix_ms);
let operation_timeout_ms = remaining_ms.min(self.config().request_timeout_ms());
if operation_timeout_ms == 0 {
@@ -211,6 +257,22 @@ impl EventSink for NostrTransport {
}
}
+impl EventSink for NostrTransport {
+ fn status(&self) -> BoxFuture<'_, Result<SinkStatus, radroots_transport::Error>> {
+ Box::pin(async move { Ok(status::sink_status(&self.status)) })
+ }
+
+ fn deliver(
+ &self,
+ request: DeliveryRequest,
+ ) -> BoxFuture<'_, Result<DeliveryReceipt, SinkFailure>> {
+ Box::pin(async move {
+ let prepared = self.prepare_delivery(request).map_err(|failure| *failure)?;
+ self.execute_prepared_delivery(prepared).await
+ })
+ }
+}
+
#[cfg_attr(coverage_nightly, coverage(off))]
fn unix_time_ms() -> u64 {
std::time::SystemTime::now()
@@ -264,6 +326,23 @@ mod tests {
}
}
+ #[derive(Debug)]
+ struct CountingRelayClient(Arc<AtomicUsize>);
+
+ impl RelayClient for CountingRelayClient {
+ fn publish<'a>(
+ &'a self,
+ _relays: Vec<RelayUrl>,
+ _event: Event,
+ _max_connections: usize,
+ _connect_timeout: Duration,
+ _operation_timeout: Duration,
+ ) -> BoxFuture<'a, Vec<RelayPublishResult>> {
+ self.0.fetch_add(1, Ordering::SeqCst);
+ Box::pin(async { Vec::new() })
+ }
+ }
+
fn payload() -> DeliveryPayload {
let raw = r#"{"id":"56bfc78223bb2221bad82b539efdec1ade0f56d0eb0e1f592fd387df4b2ceee0","pubkey":"585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df","created_at":1700000001,"kind":0,"tags":[],"content":"{}","sig":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"}"#;
DeliveryPayload::new(radroots_event_codec::decode::signed_event(raw).expect("signed event"))
@@ -336,23 +415,6 @@ mod tests {
#[test]
fn dropping_an_unpolled_delivery_performs_no_relay_work() {
- #[derive(Debug)]
- struct CountingRelayClient(Arc<AtomicUsize>);
-
- impl RelayClient for CountingRelayClient {
- fn publish<'a>(
- &'a self,
- _relays: Vec<RelayUrl>,
- _event: Event,
- _max_connections: usize,
- _connect_timeout: Duration,
- _operation_timeout: Duration,
- ) -> BoxFuture<'a, Vec<RelayPublishResult>> {
- self.0.fetch_add(1, Ordering::SeqCst);
- Box::pin(async { Vec::new() })
- }
- }
-
let calls = Arc::new(AtomicUsize::new(0));
let config = Config::from_profile(
crate::profile::test_profile(
@@ -370,24 +432,52 @@ mod tests {
}
#[test]
- fn expired_delivery_deadline_performs_no_relay_work() {
- #[derive(Debug)]
- struct CountingRelayClient(Arc<AtomicUsize>);
-
- impl RelayClient for CountingRelayClient {
- fn publish<'a>(
- &'a self,
- _relays: Vec<RelayUrl>,
- _event: Event,
- _max_connections: usize,
- _connect_timeout: Duration,
- _operation_timeout: Duration,
- ) -> BoxFuture<'a, Vec<RelayPublishResult>> {
- self.0.fetch_add(1, Ordering::SeqCst);
- Box::pin(async { Vec::new() })
- }
- }
+ fn preparation_is_no_io_redacted_consuming_and_bound_to_exact_config() {
+ let calls = Arc::new(AtomicUsize::new(0));
+ let profile = crate::profile::test_profile(
+ crate::RelayProfileKind::Public,
+ RelayUrlPolicy::Public,
+ ["wss://one.example", "wss://two.example"],
+ )
+ .expect("profile");
+ let config = Config::from_profile(profile.clone());
+ let transport =
+ NostrTransport::with_client(config, Arc::new(CountingRelayClient(Arc::clone(&calls))));
+ let request = request();
+
+ let prepared = transport
+ .prepare_delivery(request.clone())
+ .expect("prepared delivery");
+ assert_eq!(prepared.request(), &request);
+ assert_eq!(format!("{prepared:?}"), "PreparedDelivery([redacted])");
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+ let receipt = futures::executor::block_on(transport.execute_prepared_delivery(prepared))
+ .expect("executed delivery");
+ assert_eq!(receipt.request_id(), request.request_id());
+ assert_eq!(calls.load(Ordering::SeqCst), 1);
+
+ let mismatch_calls = Arc::new(AtomicUsize::new(0));
+ let mismatched = NostrTransport::with_client(
+ Config::from_profile(profile)
+ .with_timeouts(5_000, 20_000, 2_000)
+ .expect("different bounded config"),
+ Arc::new(CountingRelayClient(Arc::clone(&mismatch_calls))),
+ );
+ let prepared = transport
+ .prepare_delivery(request)
+ .expect("second prepared delivery");
+ assert_eq!(
+ futures::executor::block_on(mismatched.execute_prepared_delivery(prepared))
+ .expect_err("prepared authority is config-bound")
+ .code(),
+ "invalid_transport_contract"
+ );
+ assert_eq!(mismatch_calls.load(Ordering::SeqCst), 0);
+ }
+
+ #[test]
+ fn expired_delivery_deadline_performs_no_relay_work() {
let calls = Arc::new(AtomicUsize::new(0));
let config = Config::from_profile(
crate::profile::test_profile(
diff --git a/crates/transport_nostr/tests/package_boundary.rs b/crates/transport_nostr/tests/package_boundary.rs
@@ -9,6 +9,7 @@ const PUBLIC_API: &str =
include_str!("../../../contracts/api_baselines/radroots_transport_nostr.txt");
const ROOT: &str = include_str!("../src/lib.rs");
const PROFILE: &str = include_str!("../src/profile.rs");
+const SINK: &str = include_str!("../src/sink.rs");
const SUBSCRIPTION: &str = include_str!("../src/subscription.rs");
#[test]
@@ -55,6 +56,7 @@ fn manifest_and_root_match_the_governed_transport_boundary() {
"pub use error::Error;",
"pub use profile::{",
"pub use relay::{RelayUrl, RelayUrlPolicy};",
+ "pub use sink::PreparedDelivery;",
] {
assert!(ROOT.contains(export), "crate root is missing `{export}`");
}
@@ -67,6 +69,7 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
"## Public surface",
"## Relay and network security",
"## Fetch, live subscription, delivery, and outcome behavior",
+ "## Prepared delivery boundary",
"## Deadlines, cancellation, and commit points",
"## Serialization and diagnostics",
"## Features and runtime requirements",
@@ -79,6 +82,11 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
"event-ID tie breaker",
"upstream auto-close deadline",
"adapter-owned worker",
+ "validates the exact request, writable\nrelay bindings, and signed-event conversion without reading a clock, polling\nstatus, or performing relay I/O",
+ "Executing a capability\nthrough a differently configured transport fails closed",
+ "let _forged = PreparedDelivery {",
+ "request: panic!(),",
+ "skipped: panic!(),",
] {
assert!(README.contains(required), "README is missing `{required}`");
}
@@ -102,6 +110,7 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
for required in [
"pub struct radroots_transport_nostr::Config",
"pub struct radroots_transport_nostr::NostrTransport",
+ "pub struct radroots_transport_nostr::PreparedDelivery",
"pub enum radroots_transport_nostr::RelayAggregateState",
"pub struct radroots_transport_nostr::RelayProfile",
"pub fn radroots_transport_nostr::RelayEndpoint::new(",
@@ -116,6 +125,8 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
"NostrTransport::begin_authentication",
"NostrTransport::complete_authentication",
"NostrTransport::reject_authentication",
+ "NostrTransport::prepare_delivery",
+ "NostrTransport::execute_prepared_delivery",
] {
assert!(
PUBLIC_API.contains(required),
@@ -129,6 +140,8 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
"radroots_storage",
"radroots_outbox",
"pub trait radroots_transport_nostr",
+ "impl core::clone::Clone for radroots_transport_nostr::PreparedDelivery",
+ "impl serde_core::ser::Serialize for radroots_transport_nostr::PreparedDelivery",
] {
assert!(
!PUBLIC_API.contains(forbidden),
@@ -138,6 +151,50 @@ fn documentation_example_and_reviewed_api_baseline_are_complete() {
}
#[test]
+fn preparation_is_sealed_and_separated_from_execution_io() {
+ let prepare = SINK
+ .split_once("pub fn prepare_delivery(")
+ .expect("prepared delivery function")
+ .1
+ .split_once("pub fn execute_prepared_delivery(")
+ .expect("execution boundary")
+ .0;
+ for forbidden in [".await", "unix_time_ms", "self.status", ".publish("] {
+ assert!(
+ !prepare.contains(forbidden),
+ "delivery preparation contains I/O authority `{forbidden}`"
+ );
+ }
+ for required in [
+ "pub struct PreparedDelivery",
+ "#[must_use = \"prepared delivery must be durably bound before execution or deliberately discarded\"]",
+ "request: DeliveryRequest",
+ "event: Event",
+ "config: crate::Config",
+ "formatter.write_str(\"PreparedDelivery([redacted])\")",
+ "pub const fn request(&self) -> &DeliveryRequest",
+ ] {
+ assert!(
+ SINK.contains(required),
+ "prepared boundary is missing `{required}`"
+ );
+ }
+ for forbidden in [
+ "impl Clone for PreparedDelivery",
+ "derive(Clone",
+ "pub fn new(",
+ "pub request:",
+ "pub event:",
+ "pub config:",
+ ] {
+ assert!(
+ !prepare.contains(forbidden),
+ "prepared boundary exposes `{forbidden}`"
+ );
+ }
+}
+
+#[test]
fn relay_profiles_have_no_implicit_destination_or_policy_constructor() {
for forbidden in [
"DEFAULT_PUBLIC_RELAY",
