From 0952fd6d0db962d898a2b80df2e61eb386be7ac4 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Mon, 9 Nov 2026 16:24:03 +0200 Subject: [PATCH 1/2] Recover pending encrypted chat membership after reconnect Revisit known encrypted conversations after list publication and socket resume so missed handshake events do not leave device additions uncommitted. Coalesce membership catch-up workers and backfill messages after successful sync. Add a regression covering offline admission and document the required recorded app scenario. Verified that the regression fails with the original list-refresh code; the Tiny adapter suite passes with 246 tests passed and 9 ignored. Full recorded platform proof remains pending. --- crates/chatd/src/adapters/tiny/e2ee.rs | 11 ++++ crates/chatd/src/adapters/tiny/e2ee/fake.rs | 54 +++++++++++++++++++ .../chatd/src/adapters/tiny/e2ee/linking.rs | 20 ++++++- crates/chatd/src/adapters/tiny/live.rs | 2 + crates/chatd/src/adapters/tiny/messages.rs | 3 ++ docs/e2ee-text-client.md | 7 ++- .../encrypted-membership-reconnect/README.md | 36 +++++++++++++ .../scenario.json | 7 +++ 8 files changed, 137 insertions(+), 3 deletions(-) create mode 100644 e2e/scenarios/encrypted-membership-reconnect/README.md create mode 100644 e2e/scenarios/encrypted-membership-reconnect/scenario.json 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 Date: Wed, 11 Nov 2026 16:59:24 +0200 Subject: [PATCH 2/2] Address reconnect membership review feedback --- crates/chatd/src/adapters/tiny.rs | 2 ++ crates/chatd/src/adapters/tiny/e2ee/fake.rs | 22 ++++++++++++---- .../chatd/src/adapters/tiny/e2ee/linking.rs | 3 ++- crates/chatd/src/adapters/tiny/live.rs | 5 ++-- crates/chatd/src/adapters/tiny/messages.rs | 3 --- .../encrypted-membership-reconnect/README.md | 25 +++++++++++++++---- 6 files changed, 44 insertions(+), 16 deletions(-) diff --git a/crates/chatd/src/adapters/tiny.rs b/crates/chatd/src/adapters/tiny.rs index 71f6092a..05ae1e2d 100644 --- a/crates/chatd/src/adapters/tiny.rs +++ b/crates/chatd/src/adapters/tiny.rs @@ -1222,6 +1222,8 @@ pub(crate) fn start_with(ctx: Context, session: &str, inbox: Receiver, // made, registered, key packages up. Never stops the account. e2ee::begin(&saved, ctx.commands.clone()); messages::refresh(&ctx, &saved); + // Catch missed membership hints even when this server has no live socket. + e2ee::resume(&saved); serve_inbox(&ctx, &saved, &live, inbox); // The socket thread sleeps on the socket; this is how it hears the adapter stopped. live.wake(); diff --git a/crates/chatd/src/adapters/tiny/e2ee/fake.rs b/crates/chatd/src/adapters/tiny/e2ee/fake.rs index fd1c4b46..2ae98a99 100644 --- a/crates/chatd/src/adapters/tiny/e2ee/fake.rs +++ b/crates/chatd/src/adapters/tiny/e2ee/fake.rs @@ -428,7 +428,7 @@ fn the_first_message_of_a_new_conversation_opens_for_the_other_person() { } #[test] -fn publishing_conversations_settles_an_add_missed_while_offline() { +fn resuming_after_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(); @@ -445,6 +445,16 @@ fn publishing_conversations_settles_an_add_missed_while_offline() { 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()); + struct Registered(Vec<(String, String)>); + impl Drop for Registered { + fn drop(&mut self) { + let mut registry = lock(devices()); + for key in &self.0 { + registry.remove(key); + } + } + } + let _registered = Registered(vec![registry_key(&creator_saved), registry_key(&recipient_saved)]); lock(devices()).insert(registry_key(&creator_saved), creator.clone()); lock(devices()).insert(registry_key(&recipient_saved), recipient.clone()); let context = |saved: &Saved| { @@ -452,9 +462,12 @@ fn publishing_conversations_settles_an_add_missed_while_offline() { 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. + // Navigation refreshes the list too, but must not start membership workers. super::super::messages::publish(&creator_ctx, &creator_saved, &HashMap::new()); + assert!(!lock(&creator).membership_nudge); + assert_eq!(lock(&fake.server).epoch, Some(0)); + // Startup and socket ready/resume explicitly nudge after list publication. + resume(&creator_saved); 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)); @@ -463,13 +476,12 @@ fn publishing_conversations_settles_an_add_missed_while_offline() { if committed { let (_adapter, ctx, _inbox) = context(&recipient_saved); super::super::messages::publish(&ctx, &recipient_saved, &HashMap::new()); + resume(&recipient_saved); 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(); diff --git a/crates/chatd/src/adapters/tiny/e2ee/linking.rs b/crates/chatd/src/adapters/tiny/e2ee/linking.rs index 4e7a3368..f4f2c029 100644 --- a/crates/chatd/src/adapters/tiny/e2ee/linking.rs +++ b/crates/chatd/src/adapters/tiny/e2ee/linking.rs @@ -315,7 +315,8 @@ pub(super) fn nudge(device: Shared) { } match device.sync(&conversation) { Ok(()) => device.reload(&BTreeSet::from([conversation])), - Err(why) => tracing::warn!("chatd: Tiny e2ee: {conversation}: {why}"), + // Cached groups can belong to rooms we have left or that were deleted. + Err(why) => tracing::debug!("chatd: Tiny e2ee: {conversation}: {why}"), } } } diff --git a/crates/chatd/src/adapters/tiny/live.rs b/crates/chatd/src/adapters/tiny/live.rs index 71248b76..52a5316f 100644 --- a/crates/chatd/src/adapters/tiny/live.rs +++ b/crates/chatd/src/adapters/tiny/live.rs @@ -1809,9 +1809,10 @@ 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); } + // Both fresh ready and resume can follow a missed membership hint. UI list + // refreshes also call publish, so catch-up belongs here rather than there. + super::e2ee::resume(saved); } /// Who is in one room's call, with their names, as a voice channel's occupants. A call chatd diff --git a/crates/chatd/src/adapters/tiny/messages.rs b/crates/chatd/src/adapters/tiny/messages.rs index eac16c0b..83ae665f 100644 --- a/crates/chatd/src/adapters/tiny/messages.rs +++ b/crates/chatd/src/adapters/tiny/messages.rs @@ -212,9 +212,6 @@ pub(crate) fn publish(ctx: &Context, saved: &Saved, names: &HashMap