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

3 files changed, 140 insertions, 12 deletions

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