Tiny sends name the sender at once and settle faster

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<Saved, String> {
         .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<ureq::Agent> = 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<Value>,
     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<Value>,
+    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,21 +554,21 @@ 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 {
                 Some(body) => request.send_json(body),
                 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));

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<bool, String> {
         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<Value, String> {
+        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<Value, String> {
         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<Value, String> {
     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

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;
             }
         }

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");
+}