commit 7acb2c09a64014786d635b0fbfd101460b73aaa3
parent 0fe3881f88753fcb72be17b51274dd932fc5714a
Author: triesap <tyson@radroots.org>
Date: Mon, 21 Sep 2026 07:56:49 +0000
transport_nostr: retain skipped target attempt evidence
- Carry entered publication state through private relay results
- Keep expired queued targets unattempted in delivery receipts
- Reject skipped outcomes that contradict acceptance evidence
- Verify loopback behavior, coverage and workspace contracts
Diffstat:
4 files changed, 93 insertions(+), 12 deletions(-)
diff --git a/contracts/architecture/decisions/nostr_exact_delivery.v1.json b/contracts/architecture/decisions/nostr_exact_delivery.v1.json
@@ -8,6 +8,7 @@
"connection": "SDK authentication, subscription requests and exact publication share one serialized writer on the existing DNS-pinned and TLS-verified connection. No second socket engine, executor, adapter task, event rewrite registry, signer or durable journal is introduced.",
"lifetime": "An instance-local registry contains only the configured relay keys and weak connection references. SDK sink destruction or write failure revokes raw send authority. Reconnection installs a distinct writer; an old handle cannot switch to the replacement connection.",
"receipt": "Subscribe to the same relay notification stream before sending. Only a matching event-ID OK may accept or reject this attempt. Other IDs cannot satisfy it. Timeout, disconnect, shutdown, notification loss and missing results cannot invent acceptance.",
+ "attempt_evidence": "A target that expires while queued or before its publication future enters the remote path remains skipped. Once the path starts, attempted remains true even if connection or acknowledgement fails; it does not prove that bytes reached the wire or a remote effect occurred. A skipped target cannot claim Accepted or Delivered. Missing internal results cannot establish absence of attempted work.",
"cancellation": "Futures are caller-polled. Cancelling a write releases its serialization guard but does not prove that no frame was buffered or published. Durable retry, settlement and late facts remain caller-owned.",
"bounds": "Preserve configured target access, bounded relay inventory, connection concurrency, frame bounds, timeouts and reconnect suppression. Queued relay attempts consume the same frozen operation deadline.",
"compatibility": "Public API, generic transport SPI, dependency versions, persisted event schemas, verification thresholds and release authority remain unchanged."
diff --git a/crates/transport_nostr/README.md b/crates/transport_nostr/README.md
@@ -190,10 +190,10 @@ exact relay provenance and that current checkpoint.
Event limits, absolute deadlines, explicit cancellation, source closure, and
stable repeated terminal results follow the generic subscription contract.
-Delivery validates an already signed Radroots event and sends its retained JSON, attempts
-each configured writable target once, and returns one normalized receipt entry per
-requested target. Relay rejection, authentication requirements, rate limits,
-timeouts, connection failures, missing results, and partial acceptance remain
+Delivery validates an already signed Radroots event and sends its retained JSON,
+admits at most one attempt per configured writable target, and returns one
+normalized receipt entry per requested target. Relay rejection, authentication
+requirements, rate limits, timeouts, connection failures, missing results, and partial acceptance remain
explicit; this crate never retries, falls back to another transport, or
rewrites an unknown result as success.
@@ -214,6 +214,10 @@ Queued relay batches consume that same frozen deadline; they never receive a
new timeout after an earlier relay stalls. Bounded local normalization retains
the events and distinct outcomes already collected when network work ends, so
one timed-out relay cannot erase another relay's earlier successful evidence.
+An expired queued target remains unattempted. The attempted flag records entry
+into connection/publication work, not proof that bytes reached the wire or that
+a remote effect occurred. A skipped target cannot report acceptance; missing
+results cannot prove that an attempt was absent.
Dropping an unpolled fetch, subscription-start, or delivery future performs no
I/O. Once polled, cancellation is best effort at the socket boundary. For
diff --git a/crates/transport_nostr/src/sink.rs b/crates/transport_nostr/src/sink.rs
@@ -14,6 +14,7 @@ use std::collections::{BTreeMap, BTreeSet};
#[derive(Clone, Debug)]
pub(crate) struct RelayPublishResult {
relay: RelayUrl,
+ attempted: bool,
outcome: DeliveryOutcome,
}
@@ -100,10 +101,12 @@ impl RelayClient for LiveRelayClient {
let event = event.clone();
async move {
let url = relay.as_str().to_owned();
+ let mut attempted = false;
let attempt = async {
if tokio::time::Instant::now() >= deadline {
return Err("timeout".to_owned());
}
+ attempted = true;
self.client
.add_relay(url.as_str())
.await
@@ -119,7 +122,11 @@ impl RelayClient for LiveRelayClient {
Ok(Err(error)) => status::delivery_failure(&error),
Ok(Ok(outcome)) => outcome,
};
- RelayPublishResult { relay, outcome }
+ RelayPublishResult {
+ relay,
+ attempted,
+ outcome,
+ }
}
}))
.buffered(max_connections)
@@ -231,7 +238,10 @@ impl NostrTransport {
let mut by_relay = BTreeMap::new();
let observed_at_unix_ms = unix_time_ms().max(now_unix_ms);
for result in results {
- if !expected.contains(&result.relay) || by_relay.contains_key(&result.relay) {
+ if !expected.contains(&result.relay)
+ || by_relay.contains_key(&result.relay)
+ || (!result.attempted && status::delivery_succeeded(&result.outcome))
+ {
return Err(SinkFailure::invalid_contract(&request));
}
let succeeded = status::delivery_succeeded(&result.outcome);
@@ -241,19 +251,27 @@ impl NostrTransport {
result.outcome.is_retryable(),
observed_at_unix_ms,
);
- by_relay.insert(result.relay, result.outcome);
+ by_relay.insert(result.relay, (result.outcome, result.attempted));
}
let mut receipts = skipped;
for (relay, target) in requested {
- let outcome = by_relay.remove(&relay).unwrap_or_else(|| {
+ let (outcome, attempted) = by_relay.remove(&relay).unwrap_or_else(|| {
self.status
.record_write(&relay, false, true, observed_at_unix_ms);
- DeliveryOutcome::unavailable()
- .with_detail("missing_result", "relay returned no result")
- .expect("static normalized outcome")
+ (
+ DeliveryOutcome::unavailable()
+ .with_detail("missing_result", "relay returned no result")
+ .expect("static normalized outcome"),
+ true,
+ )
+ });
+ receipts.push(if attempted {
+ DeliveryTargetReceipt::attempted(target, outcome)
+ } else {
+ DeliveryTargetReceipt::skipped(target, outcome)
+ .map_err(|_| SinkFailure::invalid_contract(&request))?
});
- receipts.push(DeliveryTargetReceipt::attempted(target, outcome));
}
DeliveryReceipt::for_request(&request, receipts)
.map_err(|_| SinkFailure::invalid_contract(&request))
@@ -318,6 +336,7 @@ mod tests {
relays
.into_iter()
.map(|relay| RelayPublishResult {
+ attempted: true,
outcome: self
.outcomes
.get(&relay)
@@ -538,6 +557,7 @@ mod tests {
let missing = futures::executor::block_on(
scripted(vec![RelayPublishResult {
relay: one.clone(),
+ attempted: true,
outcome: DeliveryOutcome::accepted(),
}])
.deliver(request()),
@@ -548,10 +568,12 @@ mod tests {
let duplicate = scripted(vec![
RelayPublishResult {
relay: one.clone(),
+ attempted: true,
outcome: DeliveryOutcome::accepted(),
},
RelayPublishResult {
relay: one,
+ attempted: true,
outcome: DeliveryOutcome::accepted(),
},
]);
@@ -565,6 +587,7 @@ mod tests {
let other = RelayUrl::parse("wss://other.example", RelayUrlPolicy::Public).expect("other");
let unexpected = scripted(vec![RelayPublishResult {
relay: other,
+ attempted: true,
outcome: DeliveryOutcome::accepted(),
}]);
assert_eq!(
@@ -592,6 +615,54 @@ mod tests {
}
#[test]
+ fn skipped_results_remain_skipped_and_cannot_claim_acceptance() {
+ let one = RelayUrl::parse("wss://one.example", RelayUrlPolicy::Public).unwrap();
+ let two = RelayUrl::parse("wss://two.example", RelayUrlPolicy::Public).unwrap();
+ let receipt = futures::executor::block_on(
+ scripted(vec![
+ RelayPublishResult {
+ relay: one.clone(),
+ attempted: false,
+ outcome: status::delivery_failure("timeout"),
+ },
+ RelayPublishResult {
+ relay: two,
+ attempted: true,
+ outcome: DeliveryOutcome::accepted(),
+ },
+ ])
+ .deliver(request()),
+ )
+ .unwrap();
+ assert!(!receipt.target_receipts()[0].was_attempted());
+ assert_eq!(
+ receipt.target_receipts()[0].outcome().code(),
+ Some("timeout")
+ );
+ assert!(receipt.target_receipts()[1].was_attempted());
+ assert_eq!(
+ receipt.target_receipts()[1].outcome(),
+ &DeliveryOutcome::accepted()
+ );
+ for outcome in [DeliveryOutcome::accepted(), DeliveryOutcome::delivered()] {
+ let transport = scripted(vec![RelayPublishResult {
+ relay: one.clone(),
+ attempted: false,
+ outcome,
+ }]);
+ let failure = futures::executor::block_on(transport.deliver(request())).unwrap_err();
+ assert_eq!(failure.code(), "invalid_transport_contract");
+ assert!(
+ transport
+ .relay_status()
+ .relays()
+ .iter()
+ .all(|relay| relay.write().last_success_unix_ms().is_none())
+ );
+ }
+ }
+
+ #[test]
fn live_relay_client_accepts_an_empty_batch_without_io() {
let client = LiveRelayClient::isolated();
let results = futures::executor::block_on(client.publish(
diff --git a/crates/transport_nostr/tests/exact_delivery.rs b/crates/transport_nostr/tests/exact_delivery.rs
@@ -250,6 +250,11 @@ async fn queued_delivery_targets_cannot_start_after_the_shared_deadline() {
.all(|target| !target.outcome().satisfies(SatisfactionClass::Accepted))
);
assert!(second.accept().now_or_never().is_none());
+ assert!(receipt.target_receipts()[0].was_attempted());
+ assert!(
+ !receipt.target_receipts()[1].was_attempted(),
+ "a target expired while queued never entered remote publication"
+ );
server.abort();
assert!(server.await.unwrap_err().is_cancelled());
}
