auth: offload IDENTIFY + SASL PLAIN verify off the engine lock

Both ran scram verify_plain (~1s PBKDF2 at 1.2M iters) inline under the engine
lock, freezing the whole daemon per login. Add NetAction::DeferAuthenticate +
AuthThen continuation + ctx.defer_authenticate, mirroring DeferPassword: the
module/SASL path fetches the verifier cheaply and defers; the link layer runs
verify_plain on spawn_blocking, then Engine::complete_authenticate finishes the
login (IDENTIFY reuses the same ctx helpers, so its login/AJOIN/vhost/memo
side-effects are unchanged). Tests resolve the defer inline (cfg(test)). The
gRPC web-login path was already fixed; no login path stalls services now.
(GHOST/DROP/CERT/GROUP still verify inline but are rare account ops, not logins.)

src/engine/db/store.rs

@@ -33,6 +33,9 @@ impl Store for Db {
     fn authenticate(&self, name: &str, password: &str) -> Option<&str> {
         Db::authenticate(self, name, password)
     }
+    fn scram_verifier(&self, name: &str) -> Option<(String, String)> {
+        Db::scram_lookup(self, name, "SCRAM-SHA-256").map(|(a, v)| (a.to_string(), v.to_string()))
+    }
     fn grouped_nicks(&self, account: &str) -> Vec<String> {
         Db::grouped_nicks(self, account)
     }

src/engine/mod.rs

@@ -9,7 +9,7 @@ use std::time::{Duration, Instant};
 use base64::{engine::general_purpose::STANDARD, Engine as _};
 use tokio::sync::mpsc;
 
-use crate::proto::{NetAction, NetEvent, RegReply};
+use crate::proto::{AuthThen, NetAction, NetEvent, RegReply};
 use db::{Db, LogEntry, RegError};
 use scram::Verifier;
 use echo_api::Privs;
@@ -1110,6 +1110,21 @@ impl Engine {
             NetEvent::Sasl { client, mode, data, .. } => self.sasl(client, mode, data),
             _ => Vec::new(),
         };
+        // In tests there is no link layer to run DeferAuthenticate off-thread, so
+        // resolve it inline (test iteration counts are cheap) — the login finish is
+        // exactly what the link layer produces, and it must happen before
+        // track_accounts so the login is recorded within this handle().
+        #[cfg(test)]
+        let evout: Vec<NetAction> = evout
+            .into_iter()
+            .flat_map(|a| match a {
+                NetAction::DeferAuthenticate { verifier, password, then } => {
+                    let ok = crate::engine::scram::verify_plain(crate::engine::scram::Hash::Sha256, &verifier, &password);
+                    self.complete_authenticate(ok, then)
+                }
+                other => vec![other],
+            })
+            .collect();
         self.track_accounts(&evout);
         out.extend(evout);
         // Give every user-removal a traceable incident id, stamped into its reason
@@ -1415,15 +1430,17 @@ fn ci_hash(text: &str) -> u64 {
     h.finish()
 }
 
-fn login_plain(b64: &str, db: &Db) -> Option<String> {
+// Decode a SASL PLAIN response into (authcid, password). The verify itself is
+// deferred off the lock, so this no longer touches the store.
+fn decode_plain(b64: &str) -> Option<(String, String)> {
     let raw = STANDARD.decode(b64).ok()?;
     let parts: Vec<&[u8]> = raw.split(|&b| b == 0).collect();
     if parts.len() != 3 {
         return None;
     }
-    let authcid = std::str::from_utf8(parts[1]).ok()?;
-    let passwd = std::str::from_utf8(parts[2]).ok()?;
-    db.authenticate(authcid, passwd).map(str::to_string)
+    let authcid = std::str::from_utf8(parts[1]).ok()?.to_string();
+    let passwd = std::str::from_utf8(parts[2]).ok()?.to_string();
+    Some((authcid, passwd))
 }
 
 // Report SASL failure to the ircd (drives 904).

src/engine/register.rs

@@ -268,4 +268,51 @@ impl Engine {
         };
         vec![NetAction::Notice { from: agent.to_string(), to: uid.to_string(), text }]
     }
+
+    /// Finish a deferred password verify once the off-thread `verify_plain` gave
+    /// `ok`. The cheap pre-checks (exists/suspended/lockout) already ran in the
+    /// caller; this is only the success/failure finish. IDENTIFY reuses the same
+    /// ctx helpers the inline path did, so its login side-effects (login, notice,
+    /// AJOIN, vhost, memo notice) stay identical.
+    pub fn complete_authenticate(&mut self, ok: bool, then: AuthThen) -> Vec<NetAction> {
+        match then {
+            AuthThen::Identify { uid, agent, name, account } => {
+                self.db.note_auth(&name, ok);
+                let mut ctx = ServiceCtx::default();
+                if !ok {
+                    ctx.count("nickserv.identify_fail");
+                    ctx.notice(&agent, &uid, "Invalid password. Please try again.");
+                } else {
+                    ctx.login(&uid, &account);
+                    ctx.count("nickserv.identify");
+                    ctx.notice(&agent, &uid, format!("You're now identified as \x02{account}\x02. Welcome back!"));
+                    for entry in self.db.ajoin_list(&account) {
+                        ctx.force_join(&uid, &entry.channel, &entry.key);
+                    }
+                    let now = self.now_secs();
+                    let vhost = self.db.account(&account).and_then(|a| {
+                        a.vhost.as_ref().filter(|v| v.expires.is_none_or(|e| e > now)).map(|v| v.host.clone())
+                    });
+                    if let Some(host) = vhost {
+                        ctx.apply_vhost(&uid, &host);
+                    }
+                    let unread = self.db.unread_memos(&account);
+                    if unread > 0 && self.db.memo_notify_on(&account) {
+                        ctx.notice(&agent, &uid, format!("You have \x02{unread}\x02 new memo(s). Read them with \x02/msg MemoServ READ NEW\x02."));
+                    }
+                }
+                for key in std::mem::take(&mut ctx.stats) {
+                    self.bump(&key);
+                }
+                ctx.actions
+            }
+            AuthThen::Sasl { agent, client, account } => {
+                if ok {
+                    self.sasl_login(&agent, &client, account)
+                } else {
+                    sasl_fail(&agent, &client)
+                }
+            }
+        }
+    }
 }

src/engine/sasl.rs

@@ -56,8 +56,17 @@ impl Engine {
                             self.stash_sasl(client.clone(), SaslSession::Plain { response });
                             return Vec::new(); // more chunks still to come
                         }
-                        match login_plain(&response, &self.db) {
-                            Some(account) => self.sasl_login(&agent, &client, account),
+                        // Decode, then defer the verify off the lock (the login
+                        // finish lands in Engine::complete_authenticate).
+                        match decode_plain(&response) {
+                            Some((authcid, passwd)) => match self.scram_verifier(&authcid) {
+                                Some((account, verifier)) => vec![NetAction::DeferAuthenticate {
+                                    verifier,
+                                    password: passwd,
+                                    then: AuthThen::Sasl { agent: agent.clone(), client: client.clone(), account },
+                                }],
+                                None => mk("D", vec!["F".to_string()]),
+                            },
                             None => mk("D", vec!["F".to_string()]),
                         }
                     }