diff --git a/crates/chatd/src/adapters/tiny/e2ee.rs b/crates/chatd/src/adapters/tiny/e2ee.rs index fb81eeca..efdfe269 100644 --- a/crates/chatd/src/adapters/tiny/e2ee.rs +++ b/crates/chatd/src/adapters/tiny/e2ee.rs @@ -989,6 +989,8 @@ pub(crate) struct Device { /// Conversations where someone the server is adding has no device in the group yet, as the /// last sync saw them. joining: BTreeSet, + /// Coalesce reconnect and device-link membership catch-up workers. + membership_nudge: bool, /// Conversations with a row that did not open because this device was not in the group, or /// not at its epoch, yet. Read again when the group's handshake next comes. unopened: BTreeSet, @@ -1141,6 +1143,14 @@ pub(crate) fn begin(saved: &Saved, commands: std::sync::mpsc::Sender (200, json!([])), + "GET" if path == "/v1/directs" => (200, json!([{ + "id": CONVERSATION, "e2ee": true, + "peer": {"user_id": if caller == ADA { BEA } else { ADA }, "display_name": "Peer"} + }])), "GET" if path == "/v1/capabilities" => ( 200, json!({ "features": ["e2ee_text"], "e2ee_text": "optional" }), @@ -422,6 +427,55 @@ fn the_first_message_of_a_new_conversation_opens_for_the_other_person() { prove_first_encrypted_message().expect("first message opens for the other person"); } +#[test] +fn publishing_conversations_settles_an_add_missed_while_offline() { + let fake = Fake::start(0); + let mut creator = device(&fake, ADA, "offline-creator"); + creator.sync(CONVERSATION).unwrap(); + assert_eq!(lock(&fake.server).epoch, Some(0)); + // The creator is now offline. The new recipient can trigger a signed Add, + // but cannot commit it or join until the creator processes it. + let mut recipient = device(&fake, BEA, "waiting-recipient"); + recipient.sync(CONVERSATION).unwrap(); + assert!(!recipient.engine().unwrap().has_group(ConversationId::parse(CONVERSATION).unwrap())); + assert!(lock(&fake.server).pending.iter().any(|add| add.issued)); + + let creator_saved = creator.saved.clone(); + let recipient_saved = recipient.saved.clone(); + let creator = Arc::new(Mutex::new(creator)); + let recipient = Arc::new(Mutex::new(recipient)); + let registry_key = |saved: &Saved| (saved.base_url.clone(), saved.user_id.clone()); + lock(devices()).insert(registry_key(&creator_saved), creator.clone()); + lock(devices()).insert(registry_key(&recipient_saved), recipient.clone()); + let context = |saved: &Saved| { + let hub = Arc::new(crate::hub::Hub::new(crate::store::Store::open(Path::new(":memory:")).unwrap())); + crate::adapters::local(protocol::AccountKey { network: protocol::Network::Tiny, account: saved.user_id.clone() }, hub) + }; + let (_adapter, creator_ctx, _inbox) = context(&creator_saved); + // The normal list refresh after connection must drive membership even with + // no live handshake event and no messages sent or opened by the creator. + super::super::messages::publish(&creator_ctx, &creator_saved, &HashMap::new()); + let deadline = Instant::now() + Duration::from_secs(5); + while lock(&fake.server).epoch == Some(0) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + let committed = lock(&fake.server).epoch == Some(1); + if committed { + let (_adapter, ctx, _inbox) = context(&recipient_saved); + super::super::messages::publish(&ctx, &recipient_saved, &HashMap::new()); + while !lock(&recipient).engine().unwrap().has_group(ConversationId::parse(CONVERSATION).unwrap()) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + } + let joined = lock(&recipient).engine().unwrap().has_group(ConversationId::parse(CONVERSATION).unwrap()); + lock(devices()).remove(®istry_key(&creator_saved)); + lock(devices()).remove(®istry_key(&recipient_saved)); + assert!(committed, "returning creator must commit the missed Add without opening or sending a message"); + assert!(joined, "returning recipient must take its Welcome without a live hint"); + let answer = lock(&recipient).post(CONVERSATION, "after-resume", Kind::Message, None, &Envelope::message("joined after reconnect"), &[]).unwrap(); + assert_eq!(text(&fake.read(&mut lock(&creator)), &id_of(&answer)), "joined after reconnect"); +} + /// A row that did not open because this device was not in the group yet (here, the sync of the /// read failed) is read again when the group arrives, not only when another message does. #[test] diff --git a/crates/chatd/src/adapters/tiny/e2ee/linking.rs b/crates/chatd/src/adapters/tiny/e2ee/linking.rs index 9f021a4e..4e7a3368 100644 --- a/crates/chatd/src/adapters/tiny/e2ee/linking.rs +++ b/crates/chatd/src/adapters/tiny/e2ee/linking.rs @@ -286,9 +286,24 @@ const NUDGES: [Duration; 3] = [ /// Settle a device addition or removal immediately, so a linked device gets its Welcome before /// its first send. Older servers batch membership changes, so retain the later retries too. pub(super) fn nudge(device: Shared) { + { + let mut state = lock(&device); + if !state.ready() || state.membership_nudge { + return; + } + state.membership_nudge = true; + } + struct Running(Shared); + impl Drop for Running { + fn drop(&mut self) { + lock(&self.0).membership_nudge = false; + } + } + let running = Running(device.clone()); let _ = std::thread::Builder::new() .name("tiny-e2ee-nudge".into()) .spawn(move || { + let _running = running; for wait in NUDGES { std::thread::sleep(wait); let conversations = lock(&device).known_encrypted(); @@ -298,8 +313,9 @@ pub(super) fn nudge(device: Shared) { if !device.ready() { return; } - if let Err(why) = device.sync(&conversation) { - tracing::debug!("chatd: Tiny e2ee: {conversation}: {why}"); + match device.sync(&conversation) { + Ok(()) => device.reload(&BTreeSet::from([conversation])), + Err(why) => tracing::warn!("chatd: Tiny e2ee: {conversation}: {why}"), } } } diff --git a/crates/chatd/src/adapters/tiny/live.rs b/crates/chatd/src/adapters/tiny/live.rs index 0a2f60a2..71248b76 100644 --- a/crates/chatd/src/adapters/tiny/live.rs +++ b/crates/chatd/src/adapters/tiny/live.rs @@ -1809,6 +1809,8 @@ fn ready_heard(ctx: &Context, saved: &Saved, live: &Live, seat: &mut Seat, heard // without another members read: a resume is not fresh and reads nothing here. if fresh { super::messages::publish(ctx, saved, &seat.names); + } else { + super::e2ee::resume(saved); } } diff --git a/crates/chatd/src/adapters/tiny/messages.rs b/crates/chatd/src/adapters/tiny/messages.rs index 83ae665f..eac16c0b 100644 --- a/crates/chatd/src/adapters/tiny/messages.rs +++ b/crates/chatd/src/adapters/tiny/messages.rs @@ -212,6 +212,9 @@ pub(crate) fn publish(ctx: &Context, saved: &Saved, names: &HashMap