commit 09caf2c6dfab061b1f7d63b30d0d883b7cce4b96
parent bd5b29f24a91084fde2aeddc1739e68cbb98a41a
Author: triesap <tyson@radroots.org>
Date: Mon, 27 Jul 2026 08:31:28 +0000
transport: seal relay event receipts
- make per-event receipt state private and observable through accessors
- validate event identity disposition and semantic state coherence
- bound admission codes and request diagnostic contributions
- reject oversized mutated and unknown strict wire representations
Diffstat:
2 files changed, 469 insertions(+), 120 deletions(-)
diff --git a/crates/transport_nostr/src/fetch.rs b/crates/transport_nostr/src/fetch.rs
@@ -5,7 +5,7 @@ use crate::{RadrootsRelayOutcome, RadrootsRelayTargetSet, RadrootsRelayTransport
use core::time::Duration;
use futures::{StreamExt, future::BoxFuture};
use nostr::{JsonUtil, filter::MatchEventOptions};
-use radroots_event::wire::v1::DEFAULT_RAW_JSON_MAX_BYTES;
+use radroots_event::{ids::RadrootsEventId, wire::v1::DEFAULT_RAW_JSON_MAX_BYTES};
use radroots_event_store::{
RadrootsEventAdmissionStatus, RadrootsEventIngest, RadrootsEventPersistence,
RadrootsEventStore, RadrootsEventVisibility, RadrootsTransportObservation,
@@ -619,22 +619,242 @@ pub enum RadrootsRelayFetchEventVisibility {
Suppressed,
}
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct RadrootsRelayFetchEventReceipt {
- pub relay_url: String,
- pub event_id: Option<String>,
- pub inserted: bool,
- pub duplicate: bool,
- pub not_persisted: bool,
- pub malformed: bool,
- pub out_of_filter: bool,
- pub skipped_over_limit: bool,
- pub verification: RadrootsRelayFetchEventVerification,
- pub admission: RadrootsRelayFetchEventAdmission,
- pub admission_code: Option<String>,
- pub valid_stream: RadrootsRelayFetchEventValidStream,
- pub visibility: RadrootsRelayFetchEventVisibility,
- pub message: Option<String>,
+ relay_url: String,
+ event_id: Option<String>,
+ inserted: bool,
+ duplicate: bool,
+ not_persisted: bool,
+ malformed: bool,
+ out_of_filter: bool,
+ skipped_over_limit: bool,
+ verification: RadrootsRelayFetchEventVerification,
+ admission: RadrootsRelayFetchEventAdmission,
+ admission_code: Option<String>,
+ valid_stream: RadrootsRelayFetchEventValidStream,
+ visibility: RadrootsRelayFetchEventVisibility,
+ message: Option<String>,
+}
+
+impl RadrootsRelayFetchEventReceipt {
+ fn checked(mut self) -> Result<Self, RadrootsRelayTransportError> {
+ self.relay_url = canonical_fetch_receipt_relay_url(self.relay_url.as_str())?;
+ if let Some(event_id) = self.event_id.as_deref() {
+ RadrootsEventId::parse(event_id)
+ .map_err(|error| invalid_fetch_receipt("event_id", error.to_string()))?;
+ }
+ if let Some(admission_code) = self.admission_code.as_deref()
+ && admission_code.len() > radroots_transport::RADROOTS_TRANSPORT_IDENTIFIER_MAX_BYTES
+ {
+ return Err(RadrootsRelayTransportError::FetchLimitTooLarge {
+ field: "admission_code_bytes",
+ max: radroots_transport::RADROOTS_TRANSPORT_IDENTIFIER_MAX_BYTES,
+ actual: admission_code.len(),
+ });
+ }
+ if let Some(message) = self.message.as_deref() {
+ validate_fetch_receipt_diagnostic("event_receipt_message", message)?;
+ }
+
+ let disposition_count = usize::from(self.inserted)
+ + usize::from(self.duplicate)
+ + usize::from(self.not_persisted)
+ + usize::from(self.malformed)
+ + usize::from(self.out_of_filter)
+ + usize::from(self.skipped_over_limit);
+ if disposition_count > 1 {
+ return Err(invalid_fetch_receipt(
+ "event_disposition",
+ "event receipt dispositions are mutually exclusive",
+ ));
+ }
+ if (self.inserted
+ || self.duplicate
+ || self.not_persisted
+ || self.out_of_filter
+ || self.skipped_over_limit)
+ && self.event_id.is_none()
+ {
+ return Err(invalid_fetch_receipt(
+ "event_id",
+ "this event receipt disposition requires an event id",
+ ));
+ }
+ if self.malformed
+ && (self.event_id.is_some()
+ || self.verification != RadrootsRelayFetchEventVerification::NotEvaluated)
+ {
+ return Err(invalid_fetch_receipt(
+ "malformed",
+ "malformed receipts cannot identify or verify an event",
+ ));
+ }
+ if (self.inserted
+ || self.duplicate
+ || self.not_persisted
+ || self.out_of_filter
+ || self.skipped_over_limit)
+ && self.verification != RadrootsRelayFetchEventVerification::Verified
+ {
+ return Err(invalid_fetch_receipt(
+ "verification",
+ "this event receipt disposition requires verified event bytes",
+ ));
+ }
+ if self.verification != RadrootsRelayFetchEventVerification::Verified
+ && (self.admission != RadrootsRelayFetchEventAdmission::NotEvaluated
+ || self.admission_code.is_some()
+ || self.valid_stream != RadrootsRelayFetchEventValidStream::NotEvaluated
+ || self.visibility != RadrootsRelayFetchEventVisibility::NotEvaluated)
+ {
+ return Err(invalid_fetch_receipt(
+ "verification",
+ "unverified event receipts cannot carry semantic outcomes",
+ ));
+ }
+ if self.admission == RadrootsRelayFetchEventAdmission::NotEvaluated
+ && (self.admission_code.is_some()
+ || self.valid_stream != RadrootsRelayFetchEventValidStream::NotEvaluated)
+ {
+ return Err(invalid_fetch_receipt(
+ "admission",
+ "unevaluated admission cannot carry a code or valid-stream result",
+ ));
+ }
+ if self.admission != RadrootsRelayFetchEventAdmission::NotEvaluated
+ && self.valid_stream == RadrootsRelayFetchEventValidStream::NotEvaluated
+ {
+ return Err(invalid_fetch_receipt(
+ "valid_stream",
+ "evaluated admission requires a valid-stream result",
+ ));
+ }
+ if self.not_persisted
+ != (self.visibility == RadrootsRelayFetchEventVisibility::NotPersisted)
+ {
+ return Err(invalid_fetch_receipt(
+ "visibility",
+ "not-persisted disposition and visibility must agree",
+ ));
+ }
+ if matches!(
+ self.visibility,
+ RadrootsRelayFetchEventVisibility::Visible
+ | RadrootsRelayFetchEventVisibility::NotAdmitted
+ | RadrootsRelayFetchEventVisibility::NotCurrent
+ | RadrootsRelayFetchEventVisibility::Suppressed
+ ) && !(self.inserted || self.duplicate)
+ {
+ return Err(invalid_fetch_receipt(
+ "visibility",
+ "stored visibility requires an inserted or duplicate receipt",
+ ));
+ }
+ Ok(self)
+ }
+
+ pub fn relay_url(&self) -> &str {
+ self.relay_url.as_str()
+ }
+
+ pub fn event_id(&self) -> Option<&str> {
+ self.event_id.as_deref()
+ }
+
+ pub fn was_inserted(&self) -> bool {
+ self.inserted
+ }
+
+ pub fn was_duplicate(&self) -> bool {
+ self.duplicate
+ }
+
+ pub fn was_not_persisted(&self) -> bool {
+ self.not_persisted
+ }
+
+ pub fn is_malformed(&self) -> bool {
+ self.malformed
+ }
+
+ pub fn is_out_of_filter(&self) -> bool {
+ self.out_of_filter
+ }
+
+ pub fn was_skipped_over_limit(&self) -> bool {
+ self.skipped_over_limit
+ }
+
+ pub fn verification(&self) -> RadrootsRelayFetchEventVerification {
+ self.verification
+ }
+
+ pub fn admission(&self) -> RadrootsRelayFetchEventAdmission {
+ self.admission
+ }
+
+ pub fn admission_code(&self) -> Option<&str> {
+ self.admission_code.as_deref()
+ }
+
+ pub fn valid_stream(&self) -> RadrootsRelayFetchEventValidStream {
+ self.valid_stream
+ }
+
+ pub fn visibility(&self) -> RadrootsRelayFetchEventVisibility {
+ self.visibility
+ }
+
+ pub fn message(&self) -> Option<&str> {
+ self.message.as_deref()
+ }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RadrootsRelayFetchEventReceiptWire {
+ relay_url: String,
+ event_id: Option<String>,
+ inserted: bool,
+ duplicate: bool,
+ not_persisted: bool,
+ malformed: bool,
+ out_of_filter: bool,
+ skipped_over_limit: bool,
+ verification: RadrootsRelayFetchEventVerification,
+ admission: RadrootsRelayFetchEventAdmission,
+ admission_code: Option<String>,
+ valid_stream: RadrootsRelayFetchEventValidStream,
+ visibility: RadrootsRelayFetchEventVisibility,
+ message: Option<String>,
+}
+
+impl<'de> Deserialize<'de> for RadrootsRelayFetchEventReceipt {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ let wire = RadrootsRelayFetchEventReceiptWire::deserialize(deserializer)?;
+ Self {
+ relay_url: wire.relay_url,
+ event_id: wire.event_id,
+ inserted: wire.inserted,
+ duplicate: wire.duplicate,
+ not_persisted: wire.not_persisted,
+ malformed: wire.malformed,
+ out_of_filter: wire.out_of_filter,
+ skipped_over_limit: wire.skipped_over_limit,
+ verification: wire.verification,
+ admission: wire.admission,
+ admission_code: wire.admission_code,
+ valid_stream: wire.valid_stream,
+ visibility: wire.visibility,
+ message: wire.message,
+ }
+ .checked()
+ .map_err(de::Error::custom)
+ }
}
#[derive(Clone, Debug)]
@@ -785,7 +1005,7 @@ where
let max_raw_json_bytes = request.max_raw_json_bytes;
let filters = request.filters.as_slice().to_vec();
let items = adapter.fetch(request).await?;
- Ok(process_relay_fetch_items(
+ process_relay_fetch_items(
target_relays,
filters,
observed_at_ms,
@@ -794,7 +1014,7 @@ where
max_raw_json_bytes,
items,
)?
- .into_fetched_events_receipt())
+ .into_fetched_events_receipt()
}
#[cfg(feature = "runtime-tokio")]
@@ -872,22 +1092,25 @@ where
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()),
- });
+ 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;
}
};
@@ -932,12 +1155,16 @@ where
valid_stream,
visibility,
message: None,
- };
+ }
+ .checked()?;
receipt.events.push(event_receipt);
}
}
}
receipt.refresh_final_semantic_outcomes(event_store).await?;
+ for event in &receipt.events {
+ event.clone().checked()?;
+ }
Ok(receipt)
}
@@ -1002,17 +1229,19 @@ struct RadrootsRelayProcessedFetch {
}
impl RadrootsRelayProcessedFetch {
- fn into_fetched_events_receipt(self) -> RadrootsRelayFetchedEventsReceipt {
+ fn into_fetched_events_receipt(
+ self,
+ ) -> Result<RadrootsRelayFetchedEventsReceipt, RadrootsRelayTransportError> {
let mut events = Vec::new();
let mut event_receipts = Vec::new();
for item in self.items {
match item {
RadrootsRelayProcessedFetchItem::Accepted(event) => {
- event_receipts.push(accepted_fetch_event_receipt(&event));
+ event_receipts.push(accepted_fetch_event_receipt(&event)?);
events.push(event);
}
RadrootsRelayProcessedFetchItem::Duplicate(event) => {
- event_receipts.push(duplicate_fetch_event_receipt(&event));
+ event_receipts.push(duplicate_fetch_event_receipt(&event)?);
}
RadrootsRelayProcessedFetchItem::Receipt(receipt) => event_receipts.push(receipt),
}
@@ -1032,7 +1261,7 @@ impl RadrootsRelayProcessedFetch {
reason: outcome.message.clone().unwrap_or_default(),
})
.collect();
- RadrootsRelayFetchedEventsReceipt {
+ Ok(RadrootsRelayFetchedEventsReceipt {
target_relays: self.target_relays,
connected_relays,
failed_relays,
@@ -1048,7 +1277,7 @@ impl RadrootsRelayProcessedFetch {
closed_count: self.closed_count,
notice_count: self.notice_count,
relay_outcomes: self.relay_outcomes,
- }
+ })
}
}
@@ -1305,7 +1534,8 @@ fn process_relay_fetch_items(
message: Some(format!(
"event raw JSON exceeds {DEFAULT_RAW_JSON_MAX_BYTES} byte limit"
)),
- },
+ }
+ .checked()?,
));
continue;
}
@@ -1330,7 +1560,8 @@ fn process_relay_fetch_items(
valid_stream: RadrootsRelayFetchEventValidStream::NotEvaluated,
visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
message: Some("event JSON parse failed".to_owned()),
- },
+ }
+ .checked()?,
));
continue;
};
@@ -1356,7 +1587,8 @@ fn process_relay_fetch_items(
valid_stream: RadrootsRelayFetchEventValidStream::NotEvaluated,
visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
message: Some(error.to_string()),
- },
+ }
+ .checked()?,
));
continue;
}
@@ -1380,7 +1612,8 @@ fn process_relay_fetch_items(
valid_stream: RadrootsRelayFetchEventValidStream::NotEvaluated,
visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
message: Some("event did not match relay fetch filters".to_owned()),
- },
+ }
+ .checked()?,
));
continue;
}
@@ -1421,7 +1654,8 @@ fn process_relay_fetch_items(
message: Some(
"accepted relay fetch event limit reached".to_owned(),
),
- },
+ }
+ .checked()?,
));
continue;
}
@@ -1489,7 +1723,7 @@ fn canonical_requested_fetch_relay(
fn accepted_fetch_event_receipt(
event: &RadrootsRelayFetchedEvent,
-) -> RadrootsRelayFetchEventReceipt {
+) -> Result<RadrootsRelayFetchEventReceipt, RadrootsRelayTransportError> {
RadrootsRelayFetchEventReceipt {
relay_url: event.relay_url.clone(),
event_id: Some(event.event.id.to_hex()),
@@ -1506,11 +1740,12 @@ fn accepted_fetch_event_receipt(
visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
message: Some("event accepted by relay fetch filters".to_owned()),
}
+ .checked()
}
fn duplicate_fetch_event_receipt(
event: &RadrootsRelayFetchedEvent,
-) -> RadrootsRelayFetchEventReceipt {
+) -> Result<RadrootsRelayFetchEventReceipt, RadrootsRelayTransportError> {
RadrootsRelayFetchEventReceipt {
relay_url: event.relay_url.clone(),
event_id: Some(event.event.id.to_hex()),
@@ -1527,6 +1762,7 @@ fn duplicate_fetch_event_receipt(
visibility: RadrootsRelayFetchEventVisibility::NotEvaluated,
message: Some("event ID was already observed in this relay fetch".to_owned()),
}
+ .checked()
}
#[cfg_attr(coverage_nightly, coverage(off))]
diff --git a/crates/transport_nostr/tests/transport.rs b/crates/transport_nostr/tests/transport.rs
@@ -2439,11 +2439,11 @@ async fn fetch_verifies_events_before_acceptance_budgeting() {
assert_eq!(receipt.verification_failed_count, 1);
assert_eq!(receipt.skipped_over_limit_count, 0);
assert_eq!(
- receipt.event_receipts[0].verification,
+ receipt.event_receipts[0].verification(),
RadrootsRelayFetchEventVerification::Failed
);
assert_eq!(
- receipt.event_receipts[1].verification,
+ receipt.event_receipts[1].verification(),
RadrootsRelayFetchEventVerification::Verified
);
}
@@ -2476,10 +2476,10 @@ async fn fetch_deduplicates_event_ids_without_starving_unique_events() {
assert_eq!(receipt.skipped_over_limit_count, 0);
assert_eq!(receipt.event_receipts.len(), 3);
assert_eq!(
- receipt.event_receipts[1].event_id.as_deref(),
+ receipt.event_receipts[1].event_id(),
Some(first_id.as_str())
);
- assert!(receipt.event_receipts[1].duplicate);
+ assert!(receipt.event_receipts[1].was_duplicate());
}
#[tokio::test]
@@ -2616,14 +2616,18 @@ async fn fetch_ingests_events_and_records_transport_observations() {
assert_eq!(receipt.notice_count, 1);
assert_eq!(
receipt.inserted_count,
- receipt.events.iter().filter(|event| event.inserted).count()
+ receipt
+ .events
+ .iter()
+ .filter(|event| event.was_inserted())
+ .count()
);
assert_eq!(
receipt.duplicate_count,
receipt
.events
.iter()
- .filter(|event| event.duplicate)
+ .filter(|event| event.was_duplicate())
.count()
);
assert_eq!(
@@ -2631,7 +2635,7 @@ async fn fetch_ingests_events_and_records_transport_observations() {
receipt
.events
.iter()
- .filter(|event| event.not_persisted)
+ .filter(|event| event.was_not_persisted())
.count()
);
assert_eq!(
@@ -2639,7 +2643,7 @@ async fn fetch_ingests_events_and_records_transport_observations() {
receipt
.events
.iter()
- .filter(|event| event.admission == RadrootsRelayFetchEventAdmission::Unsupported)
+ .filter(|event| event.admission() == RadrootsRelayFetchEventAdmission::Unsupported)
.count()
);
assert_eq!(
@@ -2647,7 +2651,7 @@ async fn fetch_ingests_events_and_records_transport_observations() {
receipt
.events
.iter()
- .filter(|event| event.admission == RadrootsRelayFetchEventAdmission::Invalid)
+ .filter(|event| event.admission() == RadrootsRelayFetchEventAdmission::Invalid)
.count()
);
assert_eq!(
@@ -2655,7 +2659,7 @@ async fn fetch_ingests_events_and_records_transport_observations() {
receipt
.events
.iter()
- .filter(|event| event.verification == RadrootsRelayFetchEventVerification::Failed)
+ .filter(|event| event.verification() == RadrootsRelayFetchEventVerification::Failed)
.count()
);
assert_eq!(
@@ -2663,16 +2667,16 @@ async fn fetch_ingests_events_and_records_transport_observations() {
receipt
.events
.iter()
- .filter(|event| event.malformed)
+ .filter(|event| event.is_malformed())
.count()
);
assert!(receipt.events.iter().all(|event| {
- usize::from(event.inserted)
- + usize::from(event.duplicate)
- + usize::from(event.not_persisted)
+ usize::from(event.was_inserted())
+ + usize::from(event.was_duplicate())
+ + usize::from(event.was_not_persisted())
<= 1
- && (!event.malformed
- || event.verification == RadrootsRelayFetchEventVerification::NotEvaluated)
+ && (!event.is_malformed()
+ || event.verification() == RadrootsRelayFetchEventVerification::NotEvaluated)
}));
assert_eq!(receipt.relay_outcomes.len(), 4);
assert_eq!(receipt.relay_outcomes[0].relay_url(), RELAY_PRIMARY_WSS);
@@ -2703,75 +2707,72 @@ async fn fetch_ingests_events_and_records_transport_observations() {
);
assert!(receipt.relay_outcomes[3].relay_outcome().is_none());
assert_eq!(
- receipt.events[0].admission,
+ receipt.events[0].admission(),
RadrootsRelayFetchEventAdmission::Admitted
);
assert_eq!(
- receipt.events[0].valid_stream,
+ receipt.events[0].valid_stream(),
RadrootsRelayFetchEventValidStream::Eligible
);
assert_eq!(
- receipt.events[0].visibility,
+ receipt.events[0].visibility(),
RadrootsRelayFetchEventVisibility::Visible
);
assert_eq!(
- receipt.events[1].admission,
+ receipt.events[1].admission(),
RadrootsRelayFetchEventAdmission::Admitted
);
assert_eq!(
- receipt.events[1].valid_stream,
+ receipt.events[1].valid_stream(),
RadrootsRelayFetchEventValidStream::Eligible
);
assert_eq!(
- receipt.events[1].visibility,
+ receipt.events[1].visibility(),
RadrootsRelayFetchEventVisibility::Visible
);
assert_eq!(
- receipt.events[2].admission,
+ receipt.events[2].admission(),
RadrootsRelayFetchEventAdmission::Unsupported
);
+ assert_eq!(receipt.events[2].admission_code(), Some("unsupported_kind"));
assert_eq!(
- receipt.events[2].admission_code.as_deref(),
- Some("unsupported_kind")
- );
- assert_eq!(
- receipt.events[2].valid_stream,
+ receipt.events[2].valid_stream(),
RadrootsRelayFetchEventValidStream::Ineligible
);
assert_eq!(
- receipt.events[2].visibility,
+ receipt.events[2].visibility(),
RadrootsRelayFetchEventVisibility::NotAdmitted
);
assert_eq!(
- receipt.events[3].admission,
+ receipt.events[3].admission(),
RadrootsRelayFetchEventAdmission::Invalid
);
assert_eq!(
- receipt.events[3].admission_code.as_deref(),
+ receipt.events[3].admission_code(),
Some("reply_event_id_invalid")
);
assert_eq!(
- receipt.events[3].valid_stream,
+ receipt.events[3].valid_stream(),
RadrootsRelayFetchEventValidStream::Ineligible
);
assert_eq!(
- receipt.events[3].visibility,
+ receipt.events[3].visibility(),
RadrootsRelayFetchEventVisibility::NotAdmitted
);
assert_eq!(
- receipt.events[4].verification,
+ receipt.events[4].verification(),
RadrootsRelayFetchEventVerification::Failed
);
assert_eq!(
- receipt.events[4].admission,
+ receipt.events[4].admission(),
RadrootsRelayFetchEventAdmission::NotEvaluated
);
assert_eq!(
- receipt.events[5].verification,
+ receipt.events[5].verification(),
RadrootsRelayFetchEventVerification::NotEvaluated
);
assert_eq!(
- receipt.events[5].admission,
+ receipt.events[5].admission(),
RadrootsRelayFetchEventAdmission::NotEvaluated
);
@@ -2855,19 +2856,19 @@ async fn fetch_reports_final_replaceable_visibility_when_newer_arrives_first() {
assert_eq!(receipt.visible_count, 1);
assert_eq!(receipt.not_current_count, 1);
assert_eq!(
- receipt.events[0].visibility,
+ receipt.events[0].visibility(),
RadrootsRelayFetchEventVisibility::Visible
);
assert_eq!(
- receipt.events[1].admission,
+ receipt.events[1].admission(),
RadrootsRelayFetchEventAdmission::Admitted
);
assert_eq!(
- receipt.events[1].valid_stream,
+ receipt.events[1].valid_stream(),
RadrootsRelayFetchEventValidStream::Eligible
);
assert_eq!(
- receipt.events[1].visibility,
+ receipt.events[1].visibility(),
RadrootsRelayFetchEventVisibility::NotCurrent
);
}
@@ -2904,11 +2905,11 @@ async fn fetch_reports_final_replaceable_visibility_when_older_arrives_first() {
assert_eq!(receipt.visible_count, 1);
assert_eq!(receipt.not_current_count, 1);
assert_eq!(
- receipt.events[0].visibility,
+ receipt.events[0].visibility(),
RadrootsRelayFetchEventVisibility::NotCurrent
);
assert_eq!(
- receipt.events[1].visibility,
+ receipt.events[1].visibility(),
RadrootsRelayFetchEventVisibility::Visible
);
}
@@ -2958,7 +2959,7 @@ async fn fetch_maps_one_final_visibility_snapshot_back_to_duplicate_receipts() {
receipt
.events
.iter()
- .map(|event| event.event_id.as_deref())
+ .map(|event| event.event_id())
.collect::<Vec<_>>(),
vec![
Some(older_id.as_str()),
@@ -2970,7 +2971,7 @@ async fn fetch_maps_one_final_visibility_snapshot_back_to_duplicate_receipts() {
receipt
.events
.iter()
- .map(|event| event.visibility)
+ .map(|event| event.visibility())
.collect::<Vec<_>>(),
vec![
RadrootsRelayFetchEventVisibility::NotCurrent,
@@ -2978,7 +2979,7 @@ async fn fetch_maps_one_final_visibility_snapshot_back_to_duplicate_receipts() {
RadrootsRelayFetchEventVisibility::NotCurrent,
]
);
- assert!(receipt.events[2].duplicate);
+ assert!(receipt.events[2].was_duplicate());
}
#[tokio::test]
@@ -3024,23 +3025,23 @@ async fn fetch_reports_store_suppression_when_deletion_precedes_target_replay()
assert_eq!(receipt.suppressed_count, 1);
assert_eq!(receipt.not_current_count, 0);
assert_eq!(
- receipt.events[0].visibility,
+ receipt.events[0].visibility(),
RadrootsRelayFetchEventVisibility::Visible
);
assert_eq!(
- receipt.events[1].event_id.as_deref(),
+ receipt.events[1].event_id(),
Some(target.id.to_hex().as_str())
);
assert_eq!(
- receipt.events[1].admission,
+ receipt.events[1].admission(),
RadrootsRelayFetchEventAdmission::Admitted
);
assert_eq!(
- receipt.events[1].valid_stream,
+ receipt.events[1].valid_stream(),
RadrootsRelayFetchEventValidStream::Eligible
);
assert_eq!(
- receipt.events[1].visibility,
+ receipt.events[1].visibility(),
RadrootsRelayFetchEventVisibility::Suppressed
);
}
@@ -3076,14 +3077,14 @@ async fn fetch_reports_ephemeral_events_as_not_persisted_without_duplicate_or_st
assert_eq!(receipt.valid_stream_eligible_count, 0);
assert_eq!(receipt.events.len(), 2);
assert!(receipt.events.iter().all(|event| {
- !event.inserted
- && !event.duplicate
- && event.not_persisted
- && event.verification == RadrootsRelayFetchEventVerification::Verified
- && event.admission == RadrootsRelayFetchEventAdmission::Admitted
- && event.admission_code.is_none()
- && event.valid_stream == RadrootsRelayFetchEventValidStream::Ineligible
- && event.visibility == RadrootsRelayFetchEventVisibility::NotPersisted
+ !event.was_inserted()
+ && !event.was_duplicate()
+ && event.was_not_persisted()
+ && event.verification() == RadrootsRelayFetchEventVerification::Verified
+ && event.admission() == RadrootsRelayFetchEventAdmission::Admitted
+ && event.admission_code().is_none()
+ && event.valid_stream() == RadrootsRelayFetchEventValidStream::Ineligible
+ && event.visibility() == RadrootsRelayFetchEventVisibility::NotPersisted
}));
assert!(
store
@@ -3146,9 +3147,9 @@ async fn fetch_rejects_out_of_filter_events_before_store_mutation() {
assert_eq!(receipt.malformed_count, 0);
assert_eq!(receipt.admission_unsupported_count, 0);
assert_eq!(receipt.events.len(), 3);
- assert!(receipt.events[0].out_of_filter);
- assert!(!receipt.events[1].out_of_filter);
- assert!(receipt.events[2].out_of_filter);
+ assert!(receipt.events[0].is_out_of_filter());
+ assert!(!receipt.events[1].is_out_of_filter());
+ assert!(receipt.events[2].is_out_of_filter());
assert!(
store
.raw_event(accepted.id_str())
@@ -3212,10 +3213,10 @@ async fn fetch_event_cap_counts_accepted_in_filter_events_and_preserves_later_co
assert_eq!(receipt.out_of_filter_count, 1);
assert_eq!(receipt.skipped_over_limit_count, 1);
assert_eq!(receipt.events.len(), 4);
- assert!(receipt.events[0].malformed);
- assert!(receipt.events[1].out_of_filter);
- assert!(receipt.events[2].inserted);
- assert!(receipt.events[3].skipped_over_limit);
+ assert!(receipt.events[0].is_malformed());
+ assert!(receipt.events[1].is_out_of_filter());
+ assert!(receipt.events[2].was_inserted());
+ assert!(receipt.events[3].was_skipped_over_limit());
assert_eq!(receipt.eose_count, 1);
assert_eq!(receipt.closed_count, 1);
assert_eq!(receipt.notice_count, 1);
@@ -3318,10 +3319,10 @@ async fn fetch_relay_events_applies_shared_filter_limit_and_outcome_evidence() {
assert_eq!(receipt.closed_count, 1);
assert_eq!(receipt.notice_count, 1);
assert_eq!(receipt.event_receipts.len(), 4);
- assert!(receipt.event_receipts[0].malformed);
- assert!(receipt.event_receipts[1].out_of_filter);
- assert!(!receipt.event_receipts[2].malformed);
- assert!(receipt.event_receipts[3].skipped_over_limit);
+ assert!(receipt.event_receipts[0].is_malformed());
+ assert!(receipt.event_receipts[1].is_out_of_filter());
+ assert!(!receipt.event_receipts[2].is_malformed());
+ assert!(receipt.event_receipts[3].was_skipped_over_limit());
}
#[tokio::test]
@@ -3468,7 +3469,7 @@ async fn fetch_raw_json_budget_charges_every_preparse_event_class() {
assert_eq!(receipt.duplicate_count, 1);
assert_eq!(receipt.skipped_over_limit_count, 2);
assert_eq!(receipt.events.len(), 5);
- assert!(receipt.events[4].skipped_over_limit);
+ assert!(receipt.events[4].was_skipped_over_limit());
assert_eq!(receipt.eose_count, 2);
}
@@ -3721,6 +3722,118 @@ fn fetch_relay_outcomes_and_failures_seal_state_and_strict_wire() {
assert!(serde_json::from_value::<RadrootsRelayFetchFailure>(unknown_failure).is_err());
}
+#[test]
+fn fetch_event_receipts_reject_oversized_and_incoherent_wire_state() {
+ let event_id = "a".repeat(64);
+ let exact_message = "m".repeat(RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES);
+ let accepted_wire = serde_json::json!({
+ "relay_url": RELAY_PRIMARY_WSS,
+ "event_id": event_id,
+ "inserted": false,
+ "duplicate": false,
+ "not_persisted": false,
+ "malformed": false,
+ "out_of_filter": false,
+ "skipped_over_limit": false,
+ "verification": "verified",
+ "admission": "not_evaluated",
+ "admission_code": null,
+ "valid_stream": "not_evaluated",
+ "visibility": "not_evaluated",
+ "message": exact_message,
+ });
+ let accepted =
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ accepted_wire.clone(),
+ )
+ .expect("exact-limit accepted event receipt");
+ assert_eq!(accepted.relay_url(), RELAY_PRIMARY_WSS);
+ assert_eq!(accepted.event_id(), Some(event_id.as_str()));
+ assert_eq!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ serde_json::to_value(&accepted).expect("event receipt wire"),
+ )
+ .expect("strict event receipt reload"),
+ accepted
+ );
+
+ let mut oversized_message = accepted_wire.clone();
+ oversized_message["message"] =
+ serde_json::json!("m".repeat(RADROOTS_TRANSPORT_DIAGNOSTIC_MAX_BYTES + 1));
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ oversized_message,
+ )
+ .is_err()
+ );
+
+ let mut exact_code = accepted_wire.clone();
+ exact_code["inserted"] = serde_json::json!(true);
+ exact_code["admission"] = serde_json::json!("admitted");
+ exact_code["admission_code"] =
+ serde_json::json!("c".repeat(radroots_transport::RADROOTS_TRANSPORT_IDENTIFIER_MAX_BYTES));
+ exact_code["valid_stream"] = serde_json::json!("eligible");
+ exact_code["visibility"] = serde_json::json!("visible");
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ exact_code.clone(),
+ )
+ .is_ok()
+ );
+ exact_code["admission_code"] = serde_json::json!(
+ "c".repeat(radroots_transport::RADROOTS_TRANSPORT_IDENTIFIER_MAX_BYTES + 1)
+ );
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ exact_code,
+ )
+ .is_err()
+ );
+
+ for mutation in [
+ ("duplicate", serde_json::json!(true)),
+ ("event_id", serde_json::json!("not-an-event-id")),
+ ("extra", serde_json::json!(true)),
+ ] {
+ let mut wire = accepted_wire.clone();
+ wire["inserted"] = serde_json::json!(true);
+ wire[mutation.0] = mutation.1;
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ wire
+ )
+ .is_err(),
+ "mutation {} must be rejected",
+ mutation.0
+ );
+ }
+
+ let mut malformed_verified = accepted_wire.clone();
+ malformed_verified["malformed"] = serde_json::json!(true);
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ malformed_verified,
+ )
+ .is_err()
+ );
+ let mut unevaluated_with_semantics = accepted_wire.clone();
+ unevaluated_with_semantics["admission_code"] = serde_json::json!("code");
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ unevaluated_with_semantics,
+ )
+ .is_err()
+ );
+ let mut false_not_persisted_visibility = accepted_wire;
+ false_not_persisted_visibility["visibility"] = serde_json::json!("not_persisted");
+ assert!(
+ serde_json::from_value::<radroots_transport_nostr::RadrootsRelayFetchEventReceipt>(
+ false_not_persisted_visibility,
+ )
+ .is_err()
+ );
+}
+
#[tokio::test]
async fn outbox_publish_persists_partial_success_and_skips_accepted_retry() {
let outbox = RadrootsOutbox::open_memory().await.expect("outbox");
@@ -5556,9 +5669,9 @@ async fn smoke_relay_fetch_processes_one_thousand_event_receipts() {
assert_eq!(receipt.valid_stream_eligible_count, 1_000);
assert_eq!(receipt.visible_count, 1_000);
assert_eq!(receipt.events.len(), 1_000);
- assert!(receipt.events.iter().all(|event| event.valid_stream
+ assert!(receipt.events.iter().all(|event| event.valid_stream()
== RadrootsRelayFetchEventValidStream::Eligible
- && event.visibility == RadrootsRelayFetchEventVisibility::Visible));
+ && event.visibility() == RadrootsRelayFetchEventVisibility::Visible));
let replay = store.valid_stream_after(0, 1_000).await.expect("replay");
assert_eq!(replay.len(), 1_000);
}
