mod protocol_idle_replay {
use super::*;
use nostr::{EventId, RelayMessage};
async fn received_eose(
notifications: &mut tokio::sync::broadcast::Receiver<RelayPoolNotification>,
) -> Vec<EventId> {
tokio::time::timeout(Duration::from_secs(3), async {
let mut events = Vec::new();
loop {
match notifications.recv().await.unwrap() {
RelayPoolNotification::Message {
message: RelayMessage::Event { event, .. },
..
} => {
events.push(event.id);
}
RelayPoolNotification::Message {
message: RelayMessage::EndOfStoredEvents(id),
..
} if id.as_str() == "ndr-protocol" => return events,
_ => {}
}
}
})
.await
.expect("protocol subscription EOSE")
}
fn tick(core: &mut AppCore, incoming: &flume::Receiver<CoreMsg>) {
core.handle_protocol_subscription_liveness_check(core.protocol_liveness_token);
reconcile(core, incoming);
}
fn wait_until(mut ready: impl FnMut() -> bool) {
let deadline = Instant::now() + Duration::from_secs(3);
while !ready() {
assert!(Instant::now() < deadline, "relay readiness deadline");
std::thread::sleep(Duration::from_millis(10));
}
}
fn protocol_requests(relay: &crate::local_relay::TestRelay) -> usize {
relay
.wire()
.iter()
.filter(|(_, frame)| frame[0] == "REQ" && frame[1] == "ndr-protocol")
.count()
}
fn reconcile(core: &mut AppCore, incoming: &flume::Receiver<CoreMsg>) {
let deadline = Instant::now() + Duration::from_secs(5);
while core.protocol_subscription_runtime.refresh_in_flight {
assert!(
Instant::now() < deadline,
"subscription reconciliation stalled"
);
if let Ok(CoreMsg::Internal(event)) = incoming.recv_timeout(Duration::from_millis(20)) {
core.handle_internal(*event);
}
}
}
#[test]
fn populated_history_liveness_does_not_replay_healthy_subscriptions() {
let mut fixture = synthetic_history_fixture_with_devices(1, 54, 2, 3).0;
let core = &mut fixture.core;
let sender = Keys::generate();
let event = appcore_direct_message_event_for_test(
core.protocol_engine.as_mut().unwrap(),
&sender,
"Synthetic retained message",
200,
);
let owner = core.logged_in.as_ref().unwrap().owner_pubkey;
let device = core.logged_in.as_ref().unwrap().device_keys.public_key();
assert!(
core.protocol_engine
.as_ref()
.unwrap()
.has_pending_retry_work(),
"fixture must retain queued delivery work while idle"
);
let relay = crate::local_relay::TestRelay::start();
assert!(relay.wire().is_empty());
let urls = relay_urls_from_strings(&[relay.url().to_owned()]);
core.preferences.nostr_relay_urls = vec![relay.url().to_owned()];
core.logged_in.as_mut().unwrap().relay_urls = urls.clone();
let client = core.logged_in.as_ref().unwrap().client.clone();
core.runtime.block_on(async {
ensure_session_relays_configured(&client, &urls).await;
connect_client_with_timeout(&client, Duration::from_secs(2)).await;
let publisher = Client::new(Keys::generate());
ensure_session_relays_configured(&publisher, &urls).await;
connect_client_with_timeout(&publisher, Duration::from_secs(2)).await;
publisher.send_event(&event).await.unwrap();
publisher.disconnect().await;
});
let mut notifications = client.notifications();
let (tx, rx) = flume::unbounded();
core.core_sender = tx.clone();
core.priority_sender = tx;
core.protocol_subscription_runtime.desired_plan =
Some(protocol_plan_for_test(vec![event.pubkey], Vec::new()));
core.refresh_relay_connection_status();
core.reconcile_protocol_subscriptions("fixture_start", false);
reconcile(core, &rx);
let received = core.runtime.block_on(received_eose(&mut notifications));
assert!(
received.contains(&event.id),
"retained encrypted message must arrive"
);
assert_eq!(protocol_requests(&relay), 1);
for _ in 0..5 {
tick(core, &rx);
}
core.runtime.block_on(async {
client
.fetch_events(Filter::new().id(event.id), Duration::from_secs(2))
.await
.unwrap();
});
let wire = relay.wire();
for (_, frame) in &wire {
let serialized = frame.to_string();
assert!(!serialized.contains(&owner.to_hex()));
assert!(!serialized.contains(&device.to_hex()));
assert!(!serialized.contains(&sender.public_key().to_hex()));
}
assert_eq!(
protocol_requests(&relay),
1,
"idle liveness must not re-request retained history on healthy subscriptions"
);
assert_eq!(
wire.iter()
.filter(|(_, f)| f[0] == "CLOSE" && f[1] == "ndr-protocol")
.count(),
0
);
assert!(core
.protocol_engine
.as_ref()
.unwrap()
.has_pending_retry_work());
core.shutdown();
}
#[test]
fn liveness_repairs_closed_and_missing_subscriptions_without_repeated_replay() {
let owner = Keys::generate();
let device = Keys::generate();
let author = Keys::generate();
let mut core = logged_in_test_core("subscription-recovery", &owner, &device);
let relay = crate::local_relay::TestRelay::start();
assert!(relay.wire().is_empty());
let urls = relay_urls_from_strings(&[relay.url().to_owned()]);
core.preferences.nostr_relay_urls = vec![relay.url().to_owned()];
core.logged_in.as_mut().unwrap().relay_urls = urls.clone();
let client = core.logged_in.as_ref().unwrap().client.clone();
core.runtime.block_on(async {
client
.pool()
.add_relay(
urls[0].clone(),
nostr_sdk::RelayOptions::default()
.retry_interval(Duration::from_millis(10))
.adjust_retry_interval(false),
)
.await
.unwrap();
connect_client_with_timeout(&client, Duration::from_secs(2)).await;
});
let sdk_relay = core.runtime.block_on(client.relay(&urls[0])).unwrap();
let id = SubscriptionId::new("ndr-protocol");
let (tx, rx) = flume::unbounded();
core.core_sender = tx.clone();
core.priority_sender = tx;
core.refresh_relay_connection_status();
core.start_relay_status_watchers();
let CoreMsg::Internal(initial) = rx.recv_timeout(Duration::from_secs(2)).unwrap() else {
panic!("initial relay status");
};
core.handle_internal(*initial);
let mut notifications = client.notifications();
core.protocol_subscription_runtime.desired_plan = Some(protocol_plan_for_test(
vec![author.public_key()],
Vec::new(),
));
core.reconcile_protocol_subscriptions("fixture_start", false);
reconcile(&mut core, &rx);
core.runtime.block_on(received_eose(&mut notifications));
let mut requests = 1;
for reason in [
"error: try later",
"auth-required: authenticate",
"rate-limited: try later",
] {
assert_eq!(relay.close_subscription("ndr-protocol", reason), 1);
wait_until(|| core.runtime.block_on(sdk_relay.subscription(&id)).is_none());
tick(&mut core, &rx);
core.runtime.block_on(received_eose(&mut notifications));
requests += 1;
assert_eq!(protocol_requests(&relay), requests);
for _ in 0..3 {
tick(&mut core, &rx);
}
assert_eq!(protocol_requests(&relay), requests);
}
core.runtime.block_on(sdk_relay.unsubscribe(&id)).unwrap();
tick(&mut core, &rx);
core.runtime.block_on(received_eose(&mut notifications));
requests += 1;
assert_eq!(protocol_requests(&relay), requests);
let second = crate::local_relay::TestRelay::start();
assert!(second.wire().is_empty());
let both_urls = relay_urls_from_strings(&[relay.url().into(), second.url().into()]);
core.preferences.nostr_relay_urls = vec![relay.url().into(), second.url().into()];
core.logged_in.as_mut().unwrap().relay_urls = both_urls.clone();
core.runtime.block_on(async {
ensure_session_relays_configured(&client, &both_urls).await;
client.connect().await;
client
.relay(&both_urls[1])
.await
.unwrap()
.wait_for_connection(Duration::from_secs(2))
.await;
});
core.refresh_relay_connection_status();
tick(&mut core, &rx);
core.runtime.block_on(received_eose(&mut notifications));
assert_eq!(protocol_requests(&second), 1);
assert_eq!(protocol_requests(&relay), requests);
assert_eq!(relay.disconnect_clients(), 1);
wait_until(|| {
if let Ok(CoreMsg::Internal(event)) = rx.recv_timeout(Duration::from_millis(10)) {
core.handle_internal(*event);
}
protocol_requests(&relay) > requests
&& !core.protocol_subscription_runtime.refresh_in_flight
&& core.runtime.block_on(sdk_relay.subscription(&id)).is_some()
});
core.runtime.block_on(received_eose(&mut notifications));
requests = protocol_requests(&relay);
for _ in 0..3 {
tick(&mut core, &rx);
}
assert_eq!(protocol_requests(&relay), requests);
assert_eq!(protocol_requests(&second), 1);
let next_author = Keys::generate();
core.protocol_subscription_runtime.desired_plan = Some(protocol_plan_for_test(
vec![next_author.public_key()],
Vec::new(),
));
core.reconcile_protocol_subscriptions("fixture_changed", false);
reconcile(&mut core, &rx);
core.runtime.block_on(received_eose(&mut notifications));
requests += 1;
assert_eq!(protocol_requests(&relay), requests);
let last_request = relay
.wire()
.into_iter()
.rev()
.find(|(_, f)| f[0] == "REQ" && f[1] == "ndr-protocol")
.unwrap()
.1;
assert!(last_request
.to_string()
.contains(&next_author.public_key().to_hex()));
assert!(!last_request
.to_string()
.contains(&author.public_key().to_hex()));
for _ in 0..3 {
tick(&mut core, &rx);
}
let live = EventBuilder::new(
Kind::from(MESSAGE_EVENT_KIND as u16),
"synthetic transport payload",
)
.sign_with_keys(&next_author)
.unwrap();
core.runtime.block_on(async {
let publisher = Client::new(Keys::generate());
ensure_session_relays_configured(&publisher, &urls).await;
connect_client_with_timeout(&publisher, Duration::from_secs(2)).await;
publisher.send_event(&live).await.unwrap();
tokio::time::timeout(Duration::from_secs(3), async {
loop {
if let RelayPoolNotification::Event { event, .. } =
notifications.recv().await.unwrap()
{
if event.id == live.id {
break;
}
}
}
})
.await
.expect("live event after quiet liveness");
publisher.disconnect().await;
});
assert_eq!(protocol_requests(&relay), requests);
core.protocol_subscription_runtime.desired_plan = None;
core.reconcile_protocol_subscriptions("fixture_removed", false);
reconcile(&mut core, &rx);
assert!(core.runtime.block_on(sdk_relay.subscription(&id)).is_none());
let wire = relay.wire();
let receiving_connections = wire
.iter()
.filter(|(_, frame)| frame[0] == "REQ" && frame[1] == "ndr-protocol")
.map(|(connection, _)| *connection)
.collect::<HashSet<_>>();
for (_, frame) in wire
.iter()
.filter(|(connection, _)| receiving_connections.contains(connection))
{
let serialized = frame.to_string();
assert!(!serialized.contains(&owner.public_key().to_hex()));
assert!(!serialized.contains(&device.public_key().to_hex()));
}
core.shutdown();
}
}