diff --git a/crates/chatd/src/adapters/tiny.rs b/crates/chatd/src/adapters/tiny.rs index b64faa5a..63243be7 100644 --- a/crates/chatd/src/adapters/tiny.rs +++ b/crates/chatd/src/adapters/tiny.rs @@ -32,6 +32,13 @@ pub fn prove_tiny_open_backfill() -> Result<(), String> { } mod direct_prove; +mod send_prove; + +/// The lab's tiny-send-identity proof. A send names the person at once and settles without the +/// members list. See `send_prove`. +pub fn prove_tiny_send_identity() -> Result<(), String> { + send_prove::run() +} /// Headless daemon scenario for a newly received DM, including its first message and history. pub fn prove_tiny_direct_created() -> Result<(), String> { @@ -169,6 +176,7 @@ struct Me { /// The session tests read this. The account is named by the server's host. #[cfg_attr(not(test), allow(dead_code))] display_name: String, + #[cfg_attr(not(test), allow(dead_code))] email: String, } @@ -401,10 +409,28 @@ fn saved_for(session: &str, user_id: &str) -> Result { .ok_or_else(|| "the Tiny server session is for another account".to_owned()) } +/// The one agent every ordinary request goes through, so a connection to the server (the TCP +/// connect and the TLS handshake, slow on Windows) is made once and kept for the next request. +/// The pool is per host inside it, and an idle connection is let go after [`IDLE`]. fn agent() -> ureq::Agent { - agent_with_timeout(TIMEOUT) + static SHARED: std::sync::OnceLock = std::sync::OnceLock::new(); + SHARED + .get_or_init(|| { + ureq::Agent::config_builder() + .tls_config(crate::tls::config()) + .timeout_global(Some(TIMEOUT)) + .max_idle_age(IDLE) + .http_status_as_error(false) + .max_redirects(0) + .max_redirects_will_error(false) + .user_agent("tinyclient") + .build() + .into() + }) + .clone() } +/// A bespoke agent for a request with its own long timeout (an upload). It keeps no pool. fn agent_with_timeout(timeout: Duration) -> ureq::Agent { ureq::Agent::config_builder() .tls_config(crate::tls::config()) @@ -417,6 +443,11 @@ fn agent_with_timeout(timeout: Duration) -> ureq::Agent { .into() } +/// How long an idle connection is kept for the next request. Longer than ureq's 15 s, so a +/// person typing at their own pace still finds a warm connection; short enough that a server +/// or middlebox that closes idle connections first is rarely met. +const IDLE: Duration = Duration::from_secs(30); + /// One request. A 3xx is a failure: a sign-in code and the bearer stay on the host that was asked. fn call( method: &str, @@ -434,19 +465,67 @@ fn call_with_timeout( body: Option, timeout: Duration, ) -> Result<(u16, Value), Fault> { - let agent = agent_with_timeout(timeout); + // A pooled connection the server closed while it sat idle fails on first use, before any + // answer. Asking again on a fresh one is safe for a read, and for a message post, which the + // server answers once per `client_id` (a retry returns the same message). Nothing else is + // sent twice. + let again = method == "GET" || (method == "POST" && body.as_ref().is_some_and(|body| body.get("client_id").is_some_and(Value::is_string))); + match call_once(method, url, token, body.clone(), timeout) { + Err(Attempt::Dropped) if again => { + tracing::debug!("chatd: Tiny server: a kept connection was closed; asking again on a new one"); + call_once(method, url, token, body, timeout).map_err(Attempt::fault) + } + other => other.map_err(Attempt::fault), + } +} + +/// Why one attempt failed: the connection was dropped before an answer, or for any other reason. +enum Attempt { + Dropped, + Failed(Fault), +} + +impl Attempt { + fn fault(self) -> Fault { + match self { + Attempt::Dropped => Fault::Unreachable, + Attempt::Failed(fault) => fault, + } + } +} + +/// Whether a request error is the connection having been closed under us. +fn dropped(err: &ureq::Error) -> bool { + matches!(err, ureq::Error::Io(io) if matches!( + io.kind(), + std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::UnexpectedEof + )) +} + +fn call_once( + method: &str, + url: &str, + token: Option<&str>, + body: Option, + timeout: Duration, +) -> Result<(u16, Value), Attempt> { + let agent = agent(); let bearer = token.map(|token| format!("Bearer {token}")); + let bearer = bearer.as_deref(); let sent = match method { "GET" => { - let mut request = agent.get(url); - if let Some(bearer) = bearer.as_deref() { + let mut request = agent.get(url).config().timeout_global(Some(timeout)).build(); + if let Some(bearer) = bearer { request = request.header("Authorization", bearer); } request.call() } "DELETE" => { - let mut request = agent.delete(url); - if let Some(bearer) = bearer.as_deref() { + let mut request = agent.delete(url).config().timeout_global(Some(timeout)).build(); + if let Some(bearer) = bearer { request = request.header("Authorization", bearer); } match body { @@ -455,8 +534,8 @@ fn call_with_timeout( } } "POST" => { - let mut request = agent.post(url); - if let Some(bearer) = bearer.as_deref() { + let mut request = agent.post(url).config().timeout_global(Some(timeout)).build(); + if let Some(bearer) = bearer { request = request.header("Authorization", bearer); } match body { @@ -465,8 +544,8 @@ fn call_with_timeout( } } "PATCH" => { - let mut request = agent.patch(url); - if let Some(bearer) = bearer.as_deref() { + let mut request = agent.patch(url).config().timeout_global(Some(timeout)).build(); + if let Some(bearer) = bearer { request = request.header("Authorization", bearer); } match body { @@ -475,8 +554,8 @@ fn call_with_timeout( } } "PUT" => { - let mut request = agent.put(url); - if let Some(bearer) = bearer.as_deref() { + let mut request = agent.put(url).config().timeout_global(Some(timeout)).build(); + if let Some(bearer) = bearer { request = request.header("Authorization", bearer); } match body { @@ -484,12 +563,12 @@ fn call_with_timeout( None => request.send_empty(), } } - _ => return Err(Fault::Unreachable), + _ => return Err(Attempt::Failed(Fault::Unreachable)), }; - let mut response = sent.map_err(|_| Fault::Unreachable)?; + let mut response = sent.map_err(|err| if dropped(&err) { Attempt::Dropped } else { Attempt::Failed(Fault::Unreachable) })?; let status = response.status().as_u16(); if (300..400).contains(&status) { - return Err(Fault::Unreachable); + return Err(Attempt::Failed(Fault::Unreachable)); } if status == 204 { return Ok((status, Value::Null)); diff --git a/crates/chatd/src/adapters/tiny/e2ee.rs b/crates/chatd/src/adapters/tiny/e2ee.rs index fb81eeca..164cea64 100644 --- a/crates/chatd/src/adapters/tiny/e2ee.rs +++ b/crates/chatd/src/adapters/tiny/e2ee.rs @@ -1865,16 +1865,24 @@ impl Device { } pub(crate) fn sync(&mut self, conversation_id: &str) -> Result<(), String> { + self.sync_settled(conversation_id).map(|_| ()) + } + + /// [`Device::sync`], saying whether it ended on a pass that moved nothing: the feed was + /// empty and nothing was committed or started, so this device and the server agree as of that + /// pass. False when it gave up after [`PASSES`] or the conversation is not encrypted. + pub(crate) fn sync_settled(&mut self, conversation_id: &str) -> Result { if !self.ready() { return Err("This device isn't set up for encrypted messages yet.".to_owned()); } + let mut settled = false; for _ in 0..PASSES { let (active, info) = self.group_info(conversation_id)?; let conv = active.group_id; if info.get("e2ee").and_then(Value::as_bool) != Some(true) { self.note_encrypted(conversation_id, false); self.joining.remove(conversation_id); - return Ok(()); + return Ok(false); } self.note_encrypted(conversation_id, true); if joining(&info) { @@ -1952,11 +1960,12 @@ impl Device { let moved = self.feed(conversation_id, conv, &server_key)?; let committed = self.commit(conversation_id, conv)?; if !moved && !committed && !started { + settled = true; break; } } let _ = self.engine()?.flush(); - Ok(()) + Ok(settled) } /// Applies the handshake feed from where this conversation was last read. True when anything @@ -2500,6 +2509,23 @@ impl Device { target_id: Option<&str>, envelope: &Envelope, attachment_ids: &[String], + ) -> Result { + self.post_after(conversation_id, client_id, kind, target_id, envelope, attachment_ids, false) + } + + /// [`Device::post`]. `synced` says the caller synced this conversation to a settled pass on + /// this very lock, so the first pass does not sync again. A later pass always does: a server + /// that moved on since answers `409` and the retry reads it. + #[allow(clippy::too_many_arguments)] + pub(crate) fn post_after( + &mut self, + conversation_id: &str, + client_id: &str, + kind: Kind, + target_id: Option<&str>, + envelope: &Envelope, + attachment_ids: &[String], + synced: bool, ) -> Result { if self.notes.key_changed.contains(conversation_id) { return Err(words(&Error::ServerKeyChanged)); @@ -2520,9 +2546,11 @@ impl Device { .map_or(0, |since| since.as_secs() as i64), imported_from: None, }; - for _ in 0..PASSES { + for pass in 0..PASSES { // Removals and joins first: a message is never sent to someone already leaving. - self.sync(conversation_id)?; + if !(synced && pass == 0) { + self.sync(conversation_id)?; + } let active = self.active(conversation_id)?; let conv = active.group_id; let sealed = match self.engine()?.encrypt(conv, kind, target_id, envelope) { @@ -2548,11 +2576,13 @@ impl Device { if !attachment_ids.is_empty() { body["attachment_ids"] = json!(attachment_ids); } + crate::sendtime::stage("seal"); let (status, answer) = self.ask( "POST", &format!("/v1/conversations/{conversation_id}/messages"), Some(body), )?; + crate::sendtime::stage("post"); match status { 200 | 201 => { let id = answer @@ -3177,29 +3207,35 @@ pub(crate) fn send( attachment_ids: &[String], ) -> Result { let until = Instant::now() + JOIN_WAIT; - loop { - let mut device = lock(shared); - device.sync(conversation_id)?; + let mut device = lock(shared); + // The lock stays held from the last sync to the encrypt, so what the sync settled on is what + // the post starts from and does not read the server again. + let settled = loop { + let settled = device.sync_settled(conversation_id)?; if !device.joining.contains(conversation_id) { - break; + break settled; } if Instant::now() >= until { tracing::warn!( "chatd: Tiny e2ee: {conversation_id}: sent while someone is still joining" ); - break; + break settled; } drop(device); std::thread::sleep(join_poll()); - } - lock(shared).post( + device = lock(shared); + }; + crate::sendtime::stage("sync"); + let sent = device.post_after( conversation_id, client_id, kind, target_id, envelope, attachment_ids, - ) + settled, + ); + sent } /// In-process tinyserver for the first-message prove and its unit tests. Compiled into chatd so diff --git a/crates/chatd/src/adapters/tiny/live.rs b/crates/chatd/src/adapters/tiny/live.rs index 64236f93..3aa7e05a 100644 --- a/crates/chatd/src/adapters/tiny/live.rs +++ b/crates/chatd/src/adapters/tiny/live.rs @@ -21,6 +21,7 @@ use super::super::discord::gateway::{wait_sockets, Pollable, WakePair, POLL_IN, use super::super::{net, Context, ReconnectWait, Refusal, Reply, PANIC_PAUSE, RECONNECTING}; use super::{members_of, Saved}; use crate::hub::lock; +use crate::store::PersonRow; use protocol::{ join_host, CallFailure, CallState, Cue, Link, MediaRequest, Occupant, Participant, PictureKind, Presence, TinyE2ee, TinyOffer, TinyPicture, TinySeat, TinySlot, @@ -692,12 +693,23 @@ fn hold(ctx: &Context, live: &Live) -> bool { ctx.stopped() } +/// Names the account by its server and says who we are on it. `me` is our person id, not the +/// email: the pending row, presence and the shell all look us up by it. When no person of ours is +/// cached yet, the session's display name stands in for it, so a pending message is not "You". +fn describe_self(ctx: &Context, saved: &Saved) { + let _ = ctx.describe(&join_host(&saved.base_url), &saved.user_id); + let named = ctx.hub.store().person(&ctx.key(&saved.user_id)).ok().flatten().is_some_and(|person| !person.name.is_empty()); + if !named && !saved.display_name.is_empty() { + let _ = ctx.people(&[PersonRow { id: saved.user_id.clone(), name: saved.display_name.clone(), ..PersonRow::default() }]); + } +} + /// A server with no live socket: connected if its session still works. False once it does not. fn absent(ctx: &Context, saved: &Saved, seat: &mut Seat) -> bool { match super::get_me(saved) { - Ok(me) => { + Ok(_) => { if !seat.described { - let _ = ctx.describe(&join_host(&saved.base_url), &me.email); + describe_self(ctx, saved); seat.described = true; } ctx.link(Link::Connected); @@ -1737,13 +1749,7 @@ fn ready_heard(ctx: &Context, saved: &Saved, live: &Live, seat: &mut Seat, heard .map(|member| (member.user_id.clone(), member.display_name.clone())) .collect(); if !seat.described { - let email = members - .iter() - .find(|member| member.user_id == seat.me) - .map_or(saved.email.as_deref().unwrap_or(""), |member| { - member.email.as_str() - }); - let _ = ctx.describe(&join_host(&saved.base_url), email); + describe_self(ctx, saved); seat.described = true; } } diff --git a/crates/chatd/src/adapters/tiny/live/tests.rs b/crates/chatd/src/adapters/tiny/live/tests.rs index b656580d..16b67c6f 100644 --- a/crates/chatd/src/adapters/tiny/live/tests.rs +++ b/crates/chatd/src/adapters/tiny/live/tests.rs @@ -2324,3 +2324,22 @@ fn the_command_inbox_accepts_an_incoming_direct_before_starting_media() { assert_eq!(frame["direct_id"], DIRECT); assert!(!media_started, "media waits for the server's acceptance event"); } + +#[test] +fn the_account_is_described_by_the_person_id_and_a_pending_send_can_name_them() { + let hub = Arc::new(Hub::new(Store::open(std::path::Path::new(":memory:")).expect("store"))); + let account = AccountKey { network: Network::Tiny, account: ME.to_owned() }; + let (_adapter, ctx, _inbox) = local(account.clone(), hub.clone()); + let mut saved = saved_for("http://127.0.0.1:9"); + saved.email = Some("ada@example.com".to_owned()); + describe_self(&ctx, &saved); + let row = hub.store().accounts().unwrap().into_iter().find(|row| row.key == account).expect("account row"); + assert_eq!(row.me, ME, "our id on the account is the person id, never the email"); + let person = hub.store().person(&Key::new(&account, ME)).unwrap().expect("self person"); + assert_eq!(person.name, "Ada", "the session's display name stands in until the people list names us"); + + // A name the server gave is not replaced by the session's. + hub.store().upsert_people(&account, &[PersonRow { id: ME.to_owned(), name: "Ada Vale".to_owned(), ..PersonRow::default() }]).unwrap(); + describe_self(&ctx, &saved); + assert_eq!(hub.store().person(&Key::new(&account, ME)).unwrap().unwrap().name, "Ada Vale"); +} diff --git a/crates/chatd/src/adapters/tiny/messages.rs b/crates/chatd/src/adapters/tiny/messages.rs index 48333909..f35a2742 100644 --- a/crates/chatd/src/adapters/tiny/messages.rs +++ b/crates/chatd/src/adapters/tiny/messages.rs @@ -1363,29 +1363,38 @@ fn fetch_page(ctx: &Context, saved: &Saved, conversation_id: &str, before: Optio } } -/// A send of text. One to an encrypted conversation can wait for the people of a new group to -/// join it (`e2ee::send`), so it goes to this account's sealed-send thread, in order, and the -/// commands go on meanwhile. +/// A send of text. It never runs on the commands thread: that one serves history, previews and +/// everything else the person is waiting on, and a send there waited behind all of it. A send +/// goes to this account's send thread, in order, and the commands go on meanwhile. One to an +/// encrypted conversation can wait for the people of a new group to join it (`e2ee::send`), so +/// it has a thread of its own, and does not hold up the plain ones. fn send(ctx: &Context, saved: &Saved, conversation_id: String, client_id: String, text: String, reply_to: Option) { - if super::safe_id(&conversation_id) && sealed_here(ctx, saved, &conversation_id) == Ok(true) { - let account = (saved.base_url.clone(), saved.user_id.clone()); - let (posting, saved) = (ctx.clone(), saved.clone()); - in_order(account, Box::new(move || { - let outcome = post_message(&posting, &saved, &conversation_id, &client_id, &text, reply_to.as_deref(), &[]); - posting.sent(&conversation_id, &client_id, outcome); - })); + if !super::safe_id(&conversation_id) { + let outcome = Err(super::CHECK.to_owned()); + ctx.sent(&conversation_id, &client_id, outcome); return; } - let outcome = post_message(ctx, saved, &conversation_id, &client_id, &text, reply_to.as_deref(), &[]); - ctx.sent(&conversation_id, &client_id, outcome); + let labelled = ctx.known(&conversation_id).ok().flatten().is_some_and(|row| row.encryption.is_sealed()); + let user = if labelled { saved.user_id.clone() } else { format!("{}#plain", saved.user_id) }; + let account = (saved.base_url.clone(), user); + let (posting, saved) = (ctx.clone(), saved.clone()); + in_order(account, Box::new(move || { + let _ = crate::sendtime::run(&client_id, || { + let outcome = post_message(&posting, &saved, &conversation_id, &client_id, &text, reply_to.as_deref(), &[]); + let sent = outcome.is_ok(); + posting.sent(&conversation_id, &client_id, outcome); + crate::sendtime::stage("store"); + if sent { Ok(()) } else { Err(()) } + }); + })); } type Job = Box; -/// The sealed-send thread of each account (server and user), by what takes its jobs. +/// The send thread of each account (server and user, and plain or sealed), by what takes its jobs. type Queues = HashMap<(String, String), std::sync::mpsc::Sender>; -/// Runs `job` on the account's sealed-send thread, after the ones before it. Made when first +/// Runs `job` on the account's send thread, after the ones before it. Made when first /// needed, and again if it ended. fn in_order(account: (String, String), job: Job) { static QUEUES: std::sync::OnceLock> = std::sync::OnceLock::new(); @@ -1398,7 +1407,7 @@ fn in_order(account: (String, String), job: Job) { None => job, }; let (queue, jobs) = std::sync::mpsc::channel::(); - let started = std::thread::Builder::new().name("tiny-sealed-send".into()).spawn(move || { + let started = std::thread::Builder::new().name("tiny-send".into()).spawn(move || { for job in jobs { job(); } @@ -1433,7 +1442,9 @@ fn post_message(ctx: &Context, saved: &Saved, conversation_id: &str, client_id: body.insert("attachment_ids".to_owned(), json!(attachment_ids)); } let path = format!("/v1/conversations/{conversation_id}/messages"); - match super::asked("POST", saved, &path, Some(Value::Object(body))) { + let answer = super::asked("POST", saved, &path, Some(Value::Object(body))); + crate::sendtime::stage("post"); + match answer { Ok((200 | 201, value)) => accept_message(ctx, saved, conversation_id, &value), Ok((status, value)) => Err(super::explain(status, &value, super::Missing::Plain).to_owned()), Err(detail) => Err(detail.to_owned()), @@ -1473,8 +1484,7 @@ fn post_sealed(ctx: &Context, saved: &Saved, conversation_id: &str, client_id: & fn send_with(ctx: &Context, saved: &Saved, conversation_id: String, client_id: String, text: String, reply_to: Option, outgoing: Outgoing) { if outgoing.files.is_empty() { - let outcome = post_message(ctx, saved, &conversation_id, &client_id, &text, reply_to.as_deref(), &[]); - ctx.sent(&conversation_id, &client_id, outcome); + send(ctx, saved, conversation_id, client_id, text, reply_to); return; } if outgoing.files.len() > protocol::ATTACHMENTS_LIMIT { @@ -1742,7 +1752,15 @@ fn accept_message(ctx: &Context, saved: &Saved, conversation: &str, value: &Valu let wire = one_wire(value).ok_or(super::UNREADABLE)?; let mut media = Vec::new(); let opened = open_rows(ctx, saved, conversation, &[&wire]); - let message = map_message(saved, &member_names(saved), &wire, &opened, &mut media).ok_or(super::UNREADABLE)?; + crate::sendtime::stage("open"); + // Our own message is named from the session. The people list (a full request to the server) + // is for the rest, and nothing waits on it to settle a row of ours. + let own = wire.author_id == saved.user_id && !saved.display_name.is_empty(); + let names = if own { HashMap::new() } else { member_names(saved) }; + if !own { + crate::sendtime::stage("members"); + } + let message = map_message(saved, &names, &wire, &opened, &mut media).ok_or(super::UNREADABLE)?; if !media.is_empty() { let _ = ctx.media(&media); } diff --git a/crates/chatd/src/adapters/tiny/send_prove.rs b/crates/chatd/src/adapters/tiny/send_prove.rs new file mode 100644 index 00000000..d58741e4 --- /dev/null +++ b/crates/chatd/src/adapters/tiny/send_prove.rs @@ -0,0 +1,208 @@ +//! Headless sender scenario: the real daemon, its IPC and the Tiny adapter, against a loopback +//! server with invented people. The person sends a message and the pending row must already +//! name them, and must settle as soon as the post is answered, even while the server is slow to +//! list its members. This does not certify native window behavior. + +use protocol::{ + read_frame, write_frame, AccountKey, ChatStream, Key, Network, Outgoing, Page, Request, Response, SendState, +}; +use serde_json::{json, Value}; +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +const ME: &str = "11111111-1111-1111-1111-111111111111"; +const PEER: &str = "22222222-2222-2222-2222-222222222222"; +const DM: &str = "33333333-3333-3333-3333-333333333333"; +const WHEN: &str = "2024-01-02T03:04:05Z"; +const EMAIL: &str = "ada@example.test"; +const TEXT: &str = "on my way"; +/// How long the invented server takes to list its members once the person is connected. +const SLOW_MEMBERS: Duration = Duration::from_millis(3000); +/// How long a send may take to settle. Well under [`SLOW_MEMBERS`]: a send that waits for the +/// members list cannot meet it. +const SETTLES_WITHIN: Duration = Duration::from_millis(1500); + +pub(super) fn run() -> Result<(), String> { + // Never read a real session or open the installed profile, even if called outside the driver. + let dir = tempfile::tempdir().map_err(|err| err.to_string())?; + std::env::set_var("TINYCLIENT_DIR", dir.path()); + std::env::set_var("TINYCLIENT_NOTIFY", "off"); + let listener = TcpListener::bind("127.0.0.1:0").map_err(|err| err.to_string())?; + let base = format!("http://{}", listener.local_addr().map_err(|err| err.to_string())?); + let slow = Arc::new(AtomicBool::new(false)); + let slow_here = slow.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let slow = slow_here.clone(); + std::thread::spawn(move || { + let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); + let mut peek = [0; 16]; + if stream.peek(&mut peek).is_err() { + return; + } + if peek.starts_with(b"GET /v1/live") { + let Ok(mut ws) = tungstenite::accept(stream) else { return }; + let ready = json!({"type":"ready", "seq":1, "me":{"user_id":ME,"role":"member"}, "online":[ME,PEER], "calls":[]}); + if ws.send(tungstenite::Message::text(ready.to_string())).is_err() { + return; + } + // Hold the socket open; nothing else is said. + std::thread::sleep(Duration::from_secs(60)); + return; + } + let mut request = Vec::new(); + let mut buf = [0; 2048]; + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + match stream.read(&mut buf) { + Ok(0) | Err(_) => return, + Ok(n) => request.extend_from_slice(&buf[..n]), + } + } + let request = String::from_utf8_lossy(&request).into_owned(); + let (status, body) = if request.starts_with("GET /v1/members ") { + if slow.load(Ordering::Relaxed) { + std::thread::sleep(SLOW_MEMBERS); + } + ( + 200, + json!([ + {"user_id":ME,"display_name":"Ada","email":EMAIL,"role":"member","created_at":WHEN}, + {"user_id":PEER,"display_name":"Bea","email":"bea@example.test","role":"member","created_at":WHEN}]), + ) + } else if request.starts_with("GET /v1/me ") { + (200, json!({"user_id":ME,"display_name":"Ada","email":EMAIL,"role":"member"})) + } else if request.starts_with("GET /v1/directs ") { + ( + 200, + json!({"directs":[{"id":DM,"peer":{"user_id":PEER,"display_name":"Bea"},"unread":0}]}), + ) + } else if request.starts_with("GET /v1/channels ") { + (200, json!({"channels":[]})) + } else if request.starts_with("GET /v1/rooms ") { + (200, json!([])) + } else if request.starts_with(&format!("GET /v1/conversations/{DM}/messages?")) { + (200, json!({"messages":[]})) + } else if request.starts_with(&format!("POST /v1/conversations/{DM}/messages ")) { + // The client_id the person's app made is read back out of the body. + let client_id = body_of(&mut stream, &request) + .and_then(|body| body.get("client_id").and_then(Value::as_str).map(str::to_owned)) + .unwrap_or_default(); + ( + 201, + json!({"id":"44444444-4444-4444-4444-444444444444","author_id":ME,"body":TEXT,"client_id":client_id,"created_at":WHEN}), + ) + } else { + (404, json!({})) + }; + let body = body.to_string(); + let answer = format!( + "HTTP/1.1 {status} -\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = stream.write_all(answer.as_bytes()); + }); + } + }); + for network in Network::ALL { + std::env::remove_var(format!("TINYCLIENT_SESSION_{}", network.label().to_uppercase())); + } + std::env::set_var( + format!("TINYCLIENT_SESSION_{}", Network::Tiny.label().to_uppercase()), + json!({"base_url":base,"token":"invented-local-token","user_id":ME, + "display_name":"Ada","email":EMAIL,"role":"member","plain_http_ok":false}) + .to_string(), + ); + crate::spawn_core(dir.path().to_owned(), crate::Host::InProcess, |_| Ok(())).map_err(|err| err.to_string())?; + let socket = protocol::chat_sock_path_in(dir.path()); + let ask = |request: Request| -> Result { + let mut stream = ChatStream::connect(&socket).map_err(|err| err.to_string())?; + stream.set_read_timeout(Some(Duration::from_secs(8))).map_err(|err| err.to_string())?; + write_frame(&mut stream, &request).map_err(|err| err.to_string())?; + read_frame(&mut stream).map_err(|err| err.to_string()) + }; + let account = AccountKey { network: Network::Tiny, account: ME.to_owned() }; + let conversation = Key::new(&account, DM); + let until = |what: &str, wait: Duration, check: &dyn Fn() -> Result| -> Result<(), String> { + let deadline = Instant::now() + wait; + loop { + if check()? { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(format!("{what} did not happen")); + } + std::thread::sleep(Duration::from_millis(10)); + } + }; + let eight = Duration::from_secs(8); + until("Tiny connected and told who we are", eight, &|| { + Ok(matches!(ask(Request::Status)?, Response::Status { accounts, .. } + if accounts.iter().any(|own| own.key == account && own.link == protocol::Link::Connected && !own.me.is_empty()))) + })?; + // Who we are on the account is our person id. The email is not an id anyone is looked up by. + let Response::Status { accounts, .. } = ask(Request::Status)? else { return Err("no status".into()) }; + let own = accounts.iter().find(|own| own.key == account).ok_or("no account row")?; + if own.me != ME { + return Err(format!("the account's own id was {:?}, not the person's id", own.me)); + } + until("the conversation listed", eight, &|| { + Ok(matches!(ask(Request::Conversation { conversation: conversation.clone() })?, Response::Conversation { .. })) + })?; + eprintln!("send-prove: connected, own id is the person id"); + + // From here the server is slow to list members. + slow.store(true, Ordering::Relaxed); + let client_id = "55555555-5555-4555-8555-555555555555".to_owned(); + let began = Instant::now(); + let sent = ask(Request::Send { + conversation: conversation.clone(), + client_id: client_id.clone(), + text: TEXT.to_owned(), + reply_to: None, + with: Outgoing::default(), + })?; + let Response::Sent { message } = sent else { return Err(format!("send was answered {sent:?}")) }; + if message.sender.id != ME || message.sender.name != "Ada" || message.shown_sender() != "Ada" { + return Err(format!( + "the pending row names {:?} (id {:?}), not the person", + message.shown_sender(), + message.sender.id + )); + } + eprintln!("send-prove: the pending row names the person"); + until("the pending row settled", SETTLES_WITHIN, &|| { + let Response::Messages { messages, .. } = + ask(Request::Messages { conversation: conversation.clone(), page: Page::Newest, limit: 10 })? + else { + return Ok(false); + }; + Ok(messages.iter().any(|row| row.client_id.as_deref() == Some(client_id.as_str()) && row.state == SendState::Sent)) + }) + .map_err(|why| format!("{why} within {} ms", SETTLES_WITHIN.as_millis()))?; + eprintln!("send-prove: settled after {} ms", began.elapsed().as_millis()); + Ok(()) +} + +/// The JSON body of the request, read past the head the caller already has. +fn body_of(stream: &mut std::net::TcpStream, head: &str) -> Option { + let (head, already) = head.split_once("\r\n\r\n")?; + let length: usize = head + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length").then(|| value.trim().parse().ok())? + })?; + let mut body = already.as_bytes().to_vec(); + let mut buf = [0; 2048]; + while body.len() < length { + match stream.read(&mut buf) { + Ok(0) | Err(_) => break, + Ok(n) => body.extend_from_slice(&buf[..n]), + } + } + serde_json::from_slice(&body).ok() +} diff --git a/crates/chatd/src/lib.rs b/crates/chatd/src/lib.rs index 8c8a5323..fcd7fabf 100644 --- a/crates/chatd/src/lib.rs +++ b/crates/chatd/src/lib.rs @@ -12,6 +12,7 @@ pub mod notify; pub mod playback; pub mod recorder; pub mod server; +pub mod sendtime; pub mod session; pub mod store; pub mod tls; diff --git a/crates/chatd/src/main.rs b/crates/chatd/src/main.rs index 68d52c6e..fa1f5604 100644 --- a/crates/chatd/src/main.rs +++ b/crates/chatd/src/main.rs @@ -23,6 +23,15 @@ fn main() -> io::Result<()> { eprintln!("prove-exit:0"); return Ok(()); } + // The lab's tiny-send-identity driver. A build without this returns to the server below and stays up. + if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("tiny-send-identity") { + if let Err(err) = chatd::adapters::tiny::prove_tiny_send_identity() { + eprintln!("tiny-send-identity: {err}"); + std::process::exit(1); + } + eprintln!("prove-tiny-send-identity:0"); + return Ok(()); + } // The lab's leave-chime driver. A build without this returns to the server below and stays up. if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("leave-chime") { if let Err(err) = chatd::leave_prove::run() { diff --git a/crates/chatd/src/sendtime.rs b/crates/chatd/src/sendtime.rs new file mode 100644 index 00000000..0592e627 --- /dev/null +++ b/crates/chatd/src/sendtime.rs @@ -0,0 +1,148 @@ +//! Where the time of one send goes. +//! +//! A send is cheap to the person who presses Enter and slow to the one who waits for the row to +//! settle, and nothing used to say which part was slow. [`submitted`] marks the moment chatd +//! took the request; the adapter that carries it out wraps the work in [`run`], and each stage of +//! that work calls [`stage`]. Every stage logs at `info` with its `elapsed_ms`, the whole send +//! logs once when it ends, and a send over [`SLOW_MS`] adds one `warn` with the breakdown. +//! +//! Only the opaque `client_id` is logged. Never the text, a name, an address or a person's id. + +use std::cell::RefCell; +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, Instant}; + +/// A send that takes longer than this is logged as a `warn` with its stages. +pub const SLOW_MS: u64 = 1000; +/// A submit nobody picked up (a network that does not time sends) is forgotten after this. +const FORGET_AFTER: Duration = Duration::from_secs(300); +/// The most submits remembered at once. +const MOST: usize = 256; + +fn submits() -> &'static Mutex> { + static SUBMITS: OnceLock>> = OnceLock::new(); + SUBMITS.get_or_init(Default::default) +} + +/// The request for `client_id` was taken. Call it for a network whose adapter calls [`run`]. +pub fn submitted(client_id: &str) { + let mut seen = submits().lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if seen.len() >= MOST { + seen.retain(|_, at| at.elapsed() < FORGET_AFTER); + if seen.len() >= MOST { + return; + } + } + seen.insert(client_id.to_owned(), Instant::now()); +} + +/// The stages of one send, in order, each with the time it took. +#[derive(Debug)] +pub struct Stopwatch { + client_id: String, + origin: Instant, + last: Instant, + stages: Vec<(&'static str, u64)>, +} + +impl Stopwatch { + fn new(client_id: &str) -> Self { + let now = Instant::now(); + let origin = submits().lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(client_id).unwrap_or(now); + let mut watch = Stopwatch { client_id: client_id.to_owned(), origin, last: origin, stages: Vec::new() }; + watch.mark("queue"); + watch + } + + fn mark(&mut self, name: &'static str) { + let now = Instant::now(); + let took = millis(now.duration_since(self.last)); + self.last = now; + self.stages.push((name, took)); + tracing::info!(client_id = %self.client_id, stage = name, elapsed_ms = took, "chatd: send stage"); + } + + fn finish(self, ok: bool) { + let total = millis(self.origin.elapsed()); + let outcome = if ok { "sent" } else { "failed" }; + tracing::info!(client_id = %self.client_id, outcome, elapsed_ms = total, "chatd: send finished"); + if total > SLOW_MS { + tracing::warn!( + client_id = %self.client_id, + outcome, + elapsed_ms = total, + stages = %breakdown(&self.stages), + "chatd: slow send" + ); + } + } +} + +/// `queue=3 seal=410 post=220`, for the slow-send warning. +pub fn breakdown(stages: &[(&'static str, u64)]) -> String { + stages.iter().map(|(name, took)| format!("{name}={took}")).collect::>().join(" ") +} + +fn millis(took: Duration) -> u64 { + took.as_millis().min(u128::from(u64::MAX)) as u64 +} + +thread_local! { + static CURRENT: RefCell> = const { RefCell::new(None) }; +} + +/// Runs one send on this thread, timed from its submit. `work` returns whether it succeeded. +pub fn run(client_id: &str, work: impl FnOnce() -> Result) -> Result { + let previous = CURRENT.with(|current| current.borrow_mut().replace(Stopwatch::new(client_id))); + let result = work(); + let watch = CURRENT.with(|current| std::mem::replace(&mut *current.borrow_mut(), previous)); + if let Some(watch) = watch { + watch.finish(result.is_ok()); + } + result +} + +/// The named stage of the send running on this thread just ended. Nothing when none is running. +pub fn stage(name: &'static str) { + CURRENT.with(|current| { + if let Some(watch) = current.borrow_mut().as_mut() { + watch.mark(name); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_run_records_its_stages_in_order_and_no_stage_outside_one() { + stage("ignored"); + let mut seen = Vec::new(); + let result: Result = run("client-1", || { + stage("seal"); + stage("post"); + CURRENT.with(|current| { + seen = current.borrow().as_ref().map(|watch| watch.stages.iter().map(|(name, _)| *name).collect()).unwrap_or_default(); + }); + Ok(1) + }); + assert_eq!(result, Ok(1)); + assert_eq!(seen, ["queue", "seal", "post"]); + assert!(CURRENT.with(|current| current.borrow().is_none())); + } + + #[test] + fn the_breakdown_names_each_stage_and_never_anything_else() { + assert_eq!(breakdown(&[("queue", 3), ("post", 1200)]), "queue=3 post=1200"); + } + + #[test] + fn a_submit_is_taken_once() { + submitted("client-2"); + assert!(submits().lock().unwrap().contains_key("client-2")); + let _ = Stopwatch::new("client-2"); + assert!(!submits().lock().unwrap().contains_key("client-2")); + } +} diff --git a/crates/chatd/src/server.rs b/crates/chatd/src/server.rs index 5bf73b6e..e75b5017 100644 --- a/crates/chatd/src/server.rs +++ b/crates/chatd/src/server.rs @@ -2229,8 +2229,17 @@ impl Chat { Request::MarkAllRead { network } => self.mark_all_read(network), Request::Forget { conversation } => self.forget_conversation(conversation), Request::Send { conversation, client_id, text, reply_to, with } => { + let began = std::time::Instant::now(); + if conversation.network == Network::Tiny { + crate::sendtime::submitted(&client_id); + } + let logged = client_id.clone(); let message = self.new_message(&conversation, client_id, text, reply_to.as_deref(), &with); - self.send(conversation, message, reply_to, with) + let pending = began.elapsed(); + let answer = self.send(conversation, message, reply_to, with); + let (took, row) = (began.elapsed().as_millis(), pending.as_millis()); + tracing::info!(client_id = %logged, elapsed_ms = took as u64, pending_row_ms = row as u64, "chatd: send submitted"); + answer } Request::CancelSend { conversation, client_id } => { if let Err(refused) = self.unsent(&conversation, &client_id) { @@ -3221,6 +3230,28 @@ mod tests { assert_ne!(pending.sender.name, "Night Shift"); } + #[test] + fn a_tiny_pending_sender_is_found_by_the_user_id_and_never_by_the_email() { + let mut store = Store::open(std::path::Path::new(":memory:")).unwrap(); + // A Tiny account is keyed by the user id, and its own rows name that id as the sender. + let account = AccountKey { network: Network::Tiny, account: "user-1".into() }; + store.set_account(&account, "tiny.example", "user-1").unwrap(); + store + .upsert_conversation(&account, &ConversationRow::new("direct", ConversationKind::Direct, "Bea")) + .unwrap(); + let direct = Key::new(&account, "direct"); + let mut earlier = Message::plain("m1", "Ada Vale", "hi", 10, true); + earlier.sender.id = "user-1".into(); + store.ingest(&direct, &earlier, false).unwrap(); + let found = super::pending_self_sender(&store, &direct); + assert_eq!((found.id.as_str(), found.name.as_str()), ("user-1", "Ada Vale")); + + // With the email stored as the account's own id nothing matches: the row read "You". + store.set_account(&account, "tiny.example", "ada@example.com").unwrap(); + let lost = super::pending_self_sender(&store, &direct); + assert_eq!(lost.name, "You"); + } + #[test] fn a_pending_context_avatar_has_its_accounts_landed_path_before_ack() { let mut store = Store::open(std::path::Path::new(":memory:")).unwrap(); diff --git a/e2e/scenarios/tiny-send-speed/README.md b/e2e/scenarios/tiny-send-speed/README.md new file mode 100644 index 00000000..6e9c6a1c --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/README.md @@ -0,0 +1,4 @@ +Component proof for a Tiny send. The real chatd runs against an invented loopback server with a +slow member list: the pending row names the person (the account's own id is their person id, not +their email), and the row settles from the post's answer without waiting on the member list. +release:false. No native GUI, real-server timing, or mobile certification is claimed. diff --git a/e2e/scenarios/tiny-send-speed/drive.linux.sh b/e2e/scenarios/tiny-send-speed/drive.linux.sh new file mode 100755 index 00000000..b22ba647 --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/drive.linux.sh @@ -0,0 +1,9 @@ +#!/bin/bash +# tiny-send-speed on Linux. The lab runs this with bash while ffmpeg films the display. +set -u +HERE=$(cd "$(dirname "$0")" && pwd) +if command -v python3 >/dev/null 2>&1; then + exec python3 "$HERE/drive.py" +fi +echo "python3 is not on PATH" >&2 +exit 1 diff --git a/e2e/scenarios/tiny-send-speed/drive.macos.sh b/e2e/scenarios/tiny-send-speed/drive.macos.sh new file mode 100755 index 00000000..eba5f6cb --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/drive.macos.sh @@ -0,0 +1,9 @@ +#!/bin/bash +# tiny-send-speed on macOS. The lab runs this with bash while the recorder films. +set -u +HERE=$(cd "$(dirname "$0")" && pwd) +if command -v python3 >/dev/null 2>&1; then + exec python3 "$HERE/drive.py" +fi +echo "python3 is not on PATH" >&2 +exit 1 diff --git a/e2e/scenarios/tiny-send-speed/drive.py b/e2e/scenarios/tiny-send-speed/drive.py new file mode 100755 index 00000000..9fb42ad1 --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/drive.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""A Tiny send names the person at once and settles without waiting for the members list. + +Starts this build's chatd with TINYCLIENT_PROVE=tiny-send-identity. That build runs the real +daemon against an invented loopback Tiny server whose member list takes three seconds, signs the +invented person in, and sends one message over the daemon's IPC. It exits 0 only when the account's +own id is the person's id (not their email), the pending row names them rather than "You", and the +row settles within 1.5 s of the post being answered. A build from before the fix ignores the +variable and keeps serving, so this driver fails there. + +This is a daemon proof against a fake server, not native GUI certification and not a +measurement against a real server. +Env from the lab: E2E_OUT, E2E_T0, E2E_APP_DIR. +""" + +import os +import subprocess +import sys +import time + +OUT = os.environ.get("E2E_OUT", "") +T0 = float(os.environ.get("E2E_T0") or time.time()) +APP = os.environ.get("E2E_APP_DIR", "") + +FAILED = False +STEPS = "" + + +def now(): + return time.time() - T0 + + +def step(words, ok): + global FAILED + if not ok: + FAILED = True + line = '{"t": %s, "step": "%s", "ok": %s}' % (round(now(), 1), words.replace('"', "'"), "true" if ok else "false") + print(("[ok] " if ok else "[FAILED] ") + words, flush=True) + if STEPS: + with open(STEPS, "a", encoding="utf-8") as handle: + handle.write(line + "\n") + + +def chatd_bin(app_dir): + if sys.platform == "win32": + return os.path.join(app_dir, "chatd.exe") + if sys.platform == "darwin" and app_dir.endswith(".app"): + return os.path.join(app_dir, "Contents", "MacOS", "chatd") + return os.path.join(app_dir, "chatd") + + +def main(): + global STEPS + if not OUT or not APP: + print("E2E_OUT and E2E_APP_DIR are required", file=sys.stderr) + return 1 + os.makedirs(OUT, exist_ok=True) + STEPS = os.path.join(OUT, "steps.jsonl") + open(STEPS, "w", encoding="utf-8").close() + binary = chatd_bin(APP) + if not os.path.isfile(binary): + step("chatd is in E2E_APP_DIR", False) + return 1 + + data = os.path.join(OUT, "data") + os.makedirs(data, exist_ok=True) + env = os.environ.copy() + env["TINYCLIENT_DIR"] = data + env["TINYCLIENT_PROVE"] = "tiny-send-identity" + env["TINYCLIENT_LOG_LEVEL"] = "info" + kwargs = {"cwd": data, "env": env, "stdin": subprocess.DEVNULL} + if sys.platform == "win32": + kwargs["creationflags"] = 0x08000000 + with open(os.path.join(OUT, "chatd.out.log"), "w", encoding="utf-8") as out, open(os.path.join(OUT, "chatd.err.log"), "w", encoding="utf-8") as err: + kwargs["stdout"] = out + kwargs["stderr"] = err + proc = subprocess.Popen([binary], **kwargs) + try: + code = proc.wait(timeout=45) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + code = None + with open(os.path.join(OUT, "chatd.err.log"), encoding="utf-8") as handle: + marked = "prove-tiny-send-identity:0" in handle.read() + step("a Tiny send names the person at once and settles without the members list", code == 0 and marked) + return 1 if FAILED else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/e2e/scenarios/tiny-send-speed/drive.windows.ps1 b/e2e/scenarios/tiny-send-speed/drive.windows.ps1 new file mode 100644 index 00000000..ed7b2d90 --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/drive.windows.ps1 @@ -0,0 +1,17 @@ +# tiny-send-speed on Windows. The lab runs this in the console session it is already filming. +$ErrorActionPreference = 'Stop' +$drive = Join-Path $PSScriptRoot 'drive.py' +if (Get-Command py -ErrorAction SilentlyContinue) { + & py -3 $drive + exit $LASTEXITCODE +} +if (Get-Command python -ErrorAction SilentlyContinue) { + & python $drive + exit $LASTEXITCODE +} +if (Get-Command python3 -ErrorAction SilentlyContinue) { + & python3 $drive + exit $LASTEXITCODE +} +Write-Host 'python is not on PATH (tried py -3, python, python3)' +exit 1 diff --git a/e2e/scenarios/tiny-send-speed/scenario.json b/e2e/scenarios/tiny-send-speed/scenario.json new file mode 100644 index 00000000..1232ae13 --- /dev/null +++ b/e2e/scenarios/tiny-send-speed/scenario.json @@ -0,0 +1,12 @@ +{ + "id": "tiny-send-speed", + "title": "A Tiny message names its sender at once and settles without waiting on the member list", + "platforms": ["windows", "linux", "macos"], + "not_applicable": { + "android": "This driver runs desktop chatd against a loopback server; no native mobile GUI proof is claimed.", + "ios": "This driver runs desktop chatd against a loopback server; no native mobile GUI proof is claimed." + }, + "needs": [], + "release": false, + "max_still_seconds": 60 +}