fix(hub): Argon2 runs off the event loop, on exactly one worker

One derivation is 256 MB and a quarter to half a second of CPU (240 ms here,
485 ms on meshbay.org). All eleven call sites — sign-in, registration, the two
rehashes, passphrase change, reset and account deletion — ran it inline in an
async handler, so every one stopped the whole hub for that long: no request
served, no node socket read, no offer relayed. Measured on a local hub during
eight concurrent sign-ins, the worst `/v1/health` response went from 232 ms to
10 ms; the sign-ins themselves take the same time.

It could not simply go to a thread pool. Two concurrent `lanes=4` derivations
deadlock inside OpenSSL and never return, at no CPU — reproduced on
cryptography 50.0.x / OpenSSL 4.0.x both locally and on meshbay.org, while
`lanes=1` does not. `lanes` is part of every stored hash, so it is not ours to
change, and inline on the loop two derivations could never overlap, which is
the only reason production never hung.

So `auth.hash_password_off_loop` / `verify_password_off_loop` hand the work to
a dedicated executor with exactly one worker. Not a semaphore around
`to_thread`: a cancelled request would release its permit while its thread was
still deriving, and the next derivation would start beside it. One worker also
bounds Argon2's memory to one derivation whatever the number of callers.

`test_argon2_off_loop.py` reads every module for a direct call, pins the single
worker, runs four derivations and four sign-ins concurrently to completion, and
checks the loop keeps turning during a derivation; each fails with its guard
removed. CLAUDE.md and AV9 state the rule and the trap.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LcF3QKWii7uQ2kSyXErzCt

5 files changed, 150 insertions, 14 deletions

diff --git a/CLAUDE.md b/CLAUDE.md
index eba1ea2..c8abbfa 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -209,7 +209,15 @@ These are about working on the tree rather than about the design:
   catches — everything works, slowly, for everyone, whenever the mail server is
   having a bad day. `mail.send_off_loop` is the door, and
   `test_no_mail_is_sent_from_the_event_loop` reads the source for direct calls,
-  because there is nothing else to read
+  because there is nothing else to read. **Argon2 is the same, with a trap
+  under it**: one derivation is 256 MB and ~0.25–0.5 s, so it goes through
+  `auth.hash_password_off_loop` / `verify_password_off_loop` — but **two
+  concurrent `lanes=4` derivations deadlock inside OpenSSL** and never return,
+  at no CPU (cryptography 50.0.x, OpenSSL 4.0.x, reproduced locally and on
+  meshbay.org). Inline on the loop they could never overlap, which hid it. The
+  executor has **exactly one worker**, not a semaphore around `to_thread`: a
+  cancelled request releases a permit while its thread is still deriving.
+  `test_argon2_off_loop.py` holds both halves
 
 - **Before adding an endpoint or a message, ask who pays.** A participant
   supplies input; if anyone other than the sender bears the cost, there is a
diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md
index 6b1fd05..6615f66 100644
--- a/docs/MESHBAY_DESIGN.md
+++ b/docs/MESHBAY_DESIGN.md
@@ -2666,7 +2666,7 @@ had already been asked.
 | **AV6** | **A relay proves possession of its approved key.** A public key is not a password, and the register call is unauthenticated by design — it is not a user — so the proof is the only thing standing between a stranger and where nodes send relayed traffic |
 | **AV7** | **A node bounds how many peers it holds and how long an unproven one lasts.** The hub's cap is per calling account, which is a limit on each member and not on the machine, so without this an operator's exposure grew with the size of their groups |
 | **AV8** | **One account cannot make the hub mail another at will.** The invitation email's subject comes from the group row, never from the request, and the endpoint is metered |
-| **AV9** | **No mail is sent from the event loop.** `smtplib` is synchronous and waits up to ten seconds; called from an async handler that wait is the whole instance's, not one request's. Every send goes through `mail.send_off_loop` |
+| **AV9** | **No mail is sent from the event loop.** `smtplib` is synchronous and waits up to ten seconds; called from an async handler that wait is the whole instance's, not one request's. Every send goes through `mail.send_off_loop`. **Argon2 is held to the same rule**: every derivation runs on one dedicated worker thread (`auth.*_off_loop`), never on the loop and never two at a time, because two concurrent `lanes=4` derivations deadlock in OpenSSL |
 | **AV10** | **Every path that makes the hub send mail is metered, per account.** A rate limit that counts by IP bounds a caller, not an inbox. Changing one's address mails an arbitrary stranger, so it carries a cooldown *and* a daily ceiling; a reset request and a registration resend carry cooldowns |
 | **AV13** | **The mail server is not a relay, and `mail.py` is where that is decided.** Every message passes one function; `purpose` is keyword-required and checked against a closed list, so a helper that names anything else does not send and one that names nothing is a TypeError. Under it sit a bound per **recipient** — the thing a person being mail-bombed actually experiences, unmoved by which account, address or endpoint asks — and an instance-wide hourly ceiling, because registration is open and "per account" is a bound an attacker buys more of |
 | **AV11** | **A namespace a client writes into is closed, and its rows are capped.** The preference key space is an allow-list plus `default_tab:<group_id>` checked as a group id, the value is length-bounded, and the row count per account is bounded |
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 2a6baf0..05f58cd 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -24,11 +24,11 @@ from meshbay_hub.auth import (
     encrypt_email,
     generate_refresh_token,
     hash_email_blind,
-    hash_password,
+    hash_password_off_loop,
     hash_refresh_token,
     issue_access_token,
     pw_needs_rehash,
-    verify_password,
+    verify_password_off_loop,
 )
 from meshbay_hub.config import HubConfig
 from meshbay_hub.db.engine import get_db
@@ -192,7 +192,7 @@ async def register(
     if not credential:
         raise HTTPException(status_code=400, detail="auth_key or password required")
 
-    pw_hash, pw_salt = hash_password(credential)
+    pw_hash, pw_salt = await hash_password_off_loop(credential)
     pw_ver = current_pw_version() if body.auth_key else 2
     hub_id = _cfg.identity.id if _cfg else "meshbay.org"
     user = User(
@@ -346,7 +346,7 @@ async def login(
 
     if user.pw_version >= 3:
         # New scheme: verify auth_key
-        if not body.auth_key or not verify_password(
+        if not body.auth_key or not await verify_password_off_loop(
             body.auth_key, user.pw_hash, user.pw_salt, version=user.pw_version
         ):
             await _login_failed(db, body.username, ip, user.id)
@@ -357,19 +357,19 @@ async def login(
             await login_throttle.release(db, body.username)
             await db.commit()
             raise HTTPException(status_code=401, detail="auth_upgrade_required")
-        if not verify_password(
+        if not await verify_password_off_loop(
             body.password, user.pw_hash, user.pw_salt, version=user.pw_version
         ):
             await _login_failed(db, body.username, ip, user.id)
         # Migrate to new scheme if auth_key provided alongside password
         if body.auth_key:
-            new_hash, new_salt = hash_password(body.auth_key)
+            new_hash, new_salt = await hash_password_off_loop(body.auth_key)
             user.pw_hash = new_hash
             user.pw_salt = new_salt
             user.pw_version = current_pw_version()
         elif user.pw_version < 2:
             # Legacy rehash: upgrade Argon2 params within the password scheme (v1 -> v2)
-            new_hash, new_salt = hash_password(body.password)
+            new_hash, new_salt = await hash_password_off_loop(body.password)
             user.pw_hash = new_hash
             user.pw_salt = new_salt
             user.pw_version = 2
@@ -386,7 +386,7 @@ async def login(
 
     # Rehash within the auth_key scheme if Argon2 params upgraded beyond v3
     if user.pw_version >= 3 and pw_needs_rehash(user.pw_version):
-        new_hash, new_salt = hash_password(body.auth_key)
+        new_hash, new_salt = await hash_password_off_loop(body.auth_key)
         user.pw_hash = new_hash
         user.pw_salt = new_salt
         user.pw_version = current_pw_version()
@@ -885,7 +885,7 @@ async def change_password(
     db: AsyncSession = Depends(get_db),
 ):
     await _take_login_attempt(db, current_user.username)
-    if not verify_password(body.old_auth_key, current_user.pw_hash,
+    if not await verify_password_off_loop(body.old_auth_key, current_user.pw_hash,
                            current_user.pw_salt, current_user.pw_version):
         raise HTTPException(status_code=403,
                             detail="Current passphrase does not match")
@@ -894,7 +894,7 @@ async def change_password(
         raise HTTPException(status_code=400,
                             detail="New passphrase must differ from the current one")
 
-    new_hash, new_salt = hash_password(body.new_auth_key)
+    new_hash, new_salt = await hash_password_off_loop(body.new_auth_key)
     current_user.pw_hash = new_hash
     current_user.pw_salt = new_salt
     current_user.pw_version = current_pw_version()
@@ -1077,7 +1077,7 @@ async def password_reset(
         raise HTTPException(status_code=400, detail="Invalid code")
 
     verif.verified_at = now
-    new_hash, new_salt = hash_password(body.new_auth_key)
+    new_hash, new_salt = await hash_password_off_loop(body.new_auth_key)
     user.pw_hash = new_hash
     user.pw_salt = new_salt
     user.pw_version = current_pw_version()
@@ -1330,7 +1330,7 @@ async def delete_own_account(
     still never sees the passphrase itself.
     """
     await _take_login_attempt(db, current_user.username)
-    if not verify_password(body.auth_key, current_user.pw_hash, current_user.pw_salt,
+    if not await verify_password_off_loop(body.auth_key, current_user.pw_hash, current_user.pw_salt,
                            current_user.pw_version):
         raise HTTPException(status_code=403, detail="Passphrase does not match")
     await login_throttle.clear(db, current_user.username)
diff --git a/packages/meshbay-hub/src/meshbay_hub/auth.py b/packages/meshbay-hub/src/meshbay_hub/auth.py
index c5ea34d..58e2310 100644
--- a/packages/meshbay-hub/src/meshbay_hub/auth.py
+++ b/packages/meshbay-hub/src/meshbay_hub/auth.py
@@ -7,11 +7,13 @@ MeshBay Hub — authentication helpers.
   - Hub keypair: loaded from PEM file on startup
 """
 
+import asyncio
 import base64
 import hashlib
 import os
 import time
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 import blake3
@@ -139,6 +141,41 @@ def verify_password(password: str, pw_hash: bytes, salt: bytes, version: int = 2
         return False
 
 
+# ── Argon2 off the event loop, one at a time ─────────────────────────────────
+#
+# One derivation is 256 MB and a quarter to half a second of CPU. Called from an
+# async handler it stops the whole hub for that long — every request, every node
+# socket, every offer relayed — once per sign-in, passphrase change, reset and
+# registration.
+#
+# It cannot simply go to a thread pool: **two concurrent derivations with
+# `lanes=4` deadlock inside OpenSSL** and never return, at no CPU (measured
+# 2026-09-14 on cryptography 50.0.x / OpenSSL 4.0.x, here and on meshbay.org;
+# `lanes=1` does not, and `lanes` is part of every stored hash, so it is not
+# ours to change). Inline on the loop they could never overlap, which is the only
+# reason this never hung in production.
+#
+# So: a dedicated executor with exactly one worker. The loop is free while a
+# derivation runs, and derivations still never overlap — including when the
+# request that queued one is cancelled mid-way, which a semaphore around
+# `to_thread` would get wrong (the permit is released while the thread is still
+# deriving, and the next one starts beside it). It also bounds Argon2's memory
+# to one derivation, whatever the number of callers.
+_argon2_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="argon2")
+
+
+async def hash_password_off_loop(password: str) -> tuple[bytes, bytes]:
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(_argon2_executor, hash_password, password)
+
+
+async def verify_password_off_loop(password: str, pw_hash: bytes, salt: bytes,
+                                   version: int = 2) -> bool:
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(
+        _argon2_executor, verify_password, password, pw_hash, salt, version)
+
+
 def pw_needs_rehash(version: int) -> bool:
     return version < _ARGON2_CURRENT_VERSION
 
diff --git a/packages/meshbay-hub/tests/test_argon2_off_loop.py b/packages/meshbay-hub/tests/test_argon2_off_loop.py
new file mode 100644
index 0000000..b156ebc
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_argon2_off_loop.py
@@ -0,0 +1,91 @@
+"""
+Argon2 runs off the event loop, and never two at a time.
+
+One derivation is 256 MB and a quarter to half a second of CPU. On the loop it
+stopped the whole hub for that long at every sign-in. In a thread pool it would
+have been worse: two concurrent `lanes=4` derivations deadlock inside OpenSSL
+and never return (`auth._argon2_executor`). These pin both halves.
+"""
+
+import asyncio
+import base64
+import pathlib
+import re
+import time
+
+import pytest
+from meshbay_hub import auth
+
+SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "meshbay_hub"
+
+
+def test_nothing_derives_argon2_on_the_event_loop():
+    """Every line of every module, not the first match: a call added above an
+    existing one must not hide behind it."""
+    direct = re.compile(r"(?<![\w.])(hash_password|verify_password)\s*\(")
+    offenders = []
+    for path in SRC.rglob("*.py"):
+        if path.name == "auth.py":
+            continue
+        for n, line in enumerate(path.read_text().splitlines(), 1):
+            if direct.search(line):
+                offenders.append(f"{path.relative_to(SRC)}:{n}: {line.strip()}")
+    assert not offenders, (
+        "these derive Argon2 on the calling thread; use the *_off_loop versions:\n"
+        + "\n".join(offenders))
+
+
+def test_the_executor_has_exactly_one_worker():
+    """More than one lets two `lanes=4` derivations overlap, and they deadlock."""
+    assert auth._argon2_executor._max_workers == 1
+
+
+@pytest.mark.asyncio
+async def test_concurrent_derivations_all_return():
+    pw_hash, salt = auth.hash_password("k" * 44)
+    results = await asyncio.wait_for(asyncio.gather(*[
+        auth.verify_password_off_loop("k" * 44, pw_hash, salt, auth.current_pw_version())
+        for _ in range(4)]), timeout=30)
+    assert results == [True] * 4
+
+
+@pytest.mark.asyncio
+async def test_the_loop_keeps_turning_while_argon2_runs():
+    pw_hash, salt = auth.hash_password("k" * 44)
+    started = time.perf_counter()
+    auth.verify_password("k" * 44, pw_hash, salt, auth.current_pw_version())
+    inline = time.perf_counter() - started
+
+    gaps, done = [], asyncio.Event()
+
+    async def ticker():
+        last = time.perf_counter()
+        while not done.is_set():
+            await asyncio.sleep(0.005)
+            now = time.perf_counter()
+            gaps.append(now - last)
+            last = now
+
+    task = asyncio.create_task(ticker())
+    await asyncio.sleep(0.02)
+    assert await auth.verify_password_off_loop(
+        "k" * 44, pw_hash, salt, auth.current_pw_version())
+    done.set()
+    await task
+    # Inline, the loop stalls for the whole derivation; off it, for scheduling noise.
+    assert max(gaps) < inline / 2, (max(gaps), inline)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_sign_ins_all_complete(client):
+    key = base64.b64encode(b"k" * 32).decode()
+    names = [f"concurrent{i}" for i in range(4)]
+    for name in names:
+        r = await client.post("/v1/users/register", json={
+            "username": name, "email": f"{name}@example.test", "auth_key": key})
+        assert r.status_code == 201, r.text
+
+    responses = await asyncio.wait_for(asyncio.gather(*[
+        client.post("/v1/users/login", json={"username": n, "auth_key": key})
+        for n in names]), timeout=60)
+    assert [r.status_code for r in responses] == [200] * 4