fix(hub): the mail allowance is written down, and recovery keeps a share

Two dicts in `mail.py` held the budget, so every deploy handed out a fresh
one — and this hub is deployed several times a day. A bound a restart forgets
is not a bound, for the reason the denylist is persisted rather than held in
memory (S3). It is a `mail_quota` table now, one row per counter, the
recipient hashed so the table does not become a list of plaintext addresses.

The counting moves with it, into an async `reserve` that has a session, and
`send_off_loop` is the one door it stands in. `_send` keeps the purpose
allow-list: that half needs no state, and it is what stops anything which
puts a message on the wire from naming a reason this hub does not send for.
The caller owns the commit, so a request that fails afterwards is not charged
for mail nobody received.

`hourly_reserved_for_recovery` is new. A flood of sign-ups used to be able to
spend the whole hour and lock out the person waiting on a passphrase reset;
registration and address changes may now spend only the unreserved share.

Values changed as agreed: 10 messages a day to one recipient, 300 s between
two reset codes. The address-change ceiling and its cooldown were two bounds
on one thing — 3 a day and 60 s apart — and collapse into one 48-hour delay.
Asking again for the address already pending is exempt: it reaches no new
recipient, that recipient is bounded anyway, and without the exemption a typo
locked the account out of correcting it for two days.

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

1 files changed, 303 insertions, 72 deletions

diff --git a/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py b/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
index a56ab17..7c1a9f4 100644
--- a/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
+++ b/packages/meshbay-hub/tests/test_mail_is_not_a_relay.py
@@ -55,13 +55,16 @@ def _real_mail_path(monkeypatch):
 
 
 @pytest.fixture(autouse=True)
-def _fresh_budget():
-    """Each test starts with the hub's allowance untouched."""
-    mail_mod._destinations.clear()
-    mail_mod._hour = (0.0, 0)
-    yield
-    mail_mod._destinations.clear()
-    mail_mod._hour = (0.0, 0)
+def _configured_limits():
+    """What `hub.toml` says, which is what a missing settings row falls back to.
+
+    Set at startup by the lifespan; pinned here so a test that never builds an
+    app still measures against the shipped defaults rather than against zero.
+    """
+    from meshbay_hub import hub_settings
+    from meshbay_hub.config import MailConfig
+
+    hub_settings.set_mail_defaults(MailConfig())
 
 
 @pytest.fixture
@@ -96,6 +99,17 @@ def _msg(to: str):
     return m
 
 
+async def _charge(db, purpose: str, address: str) -> bool:
+    """One send against the allowance. True if it was granted."""
+    try:
+        await mail_mod.reserve(db, purpose, address)
+    except mail_mod.MailRefused:
+        await db.rollback()
+        return False
+    await db.commit()
+    return True
+
+
 # ── The gate itself ──────────────────────────────────────────────────────────
 
 def test_a_send_must_name_a_purpose(wire):
@@ -105,50 +119,115 @@ def test_a_send_must_name_a_purpose(wire):
         _REAL_SEND(_msg("someone@example.test"))
 
 
-def test_only_the_four_purposes_send(wire):
-    assert _REAL_SEND(_msg("a@example.test"), purpose="registration")
+def test_send_refuses_a_purpose_this_hub_does_not_have(wire):
+    """The structural half of the gate, and the half that needs no state. It
+    is in `_send` so that nothing which puts a message on the wire can name a
+    reason this hub does not send mail for; the counting half needs a database
+    session and lives in `reserve`."""
     assert not _REAL_SEND(_msg("b@example.test"), purpose="marketing")
     assert not _REAL_SEND(_msg("c@example.test"), purpose="")
+    assert _REAL_SEND(_msg("a@example.test"), purpose="registration")
     assert [to for to, _ in wire] == ["a@example.test"]
 
 
-def test_one_recipient_has_a_cooldown_and_a_daily_allowance(wire):
-    """Across every purpose — the recipient is the thing being protected, not
-    the sender."""
+@pytest.mark.asyncio
+async def test_reserve_refuses_a_purpose_this_hub_does_not_have(db_session):
+    with pytest.raises(mail_mod.MailRefused):
+        await mail_mod.reserve(db_session, "marketing", "a@example.test")
+
+
+@pytest.mark.asyncio
+async def test_one_recipient_has_a_cooldown_and_a_daily_allowance(db_session):
+    """Across every purpose — the recipient is what is protected, not the
+    sender."""
+    from meshbay_hub.config import MailConfig
+    from meshbay_hub.db.models import MailQuota
+
     victim = "victim@example.test"
-    assert _REAL_SEND(_msg(victim), purpose="registration")
-    assert not _REAL_SEND(_msg(victim), purpose="password_reset"), (
+    assert await _charge(db_session, "registration", victim)
+    assert not await _charge(db_session, "password_reset", victim), (
         "a second purpose walked around the cooldown")
 
-    # Past the cooldown, up to the day's allowance, then no more.
-    for _ in range(mail_mod.DESTINATION_DAILY_CAP - 1):
-        mail_mod._destinations[mail_mod._destination_key(victim)] = (
-            0.0,
-            mail_mod._destinations[mail_mod._destination_key(victim)][1],
-            mail_mod._destinations[mail_mod._destination_key(victim)][2])
-        assert _REAL_SEND(_msg(victim), purpose="invite")
+    cap = MailConfig().destination_daily_cap
+    key = mail_mod.destination_key(victim)
+    for _ in range(cap - 1):
+        # Step past the cooldown without waiting it out. The daily count is
+        # deliberately left alone.
+        (await db_session.get(MailQuota, key)).last_sent = None
+        await db_session.commit()
+        assert await _charge(db_session, "invite", victim)
 
-    mail_mod._destinations[mail_mod._destination_key(victim)] = (
-        0.0, mail_mod.DESTINATION_DAILY_CAP,
-        mail_mod._destinations[mail_mod._destination_key(victim)][2])
-    assert not _REAL_SEND(_msg(victim), purpose="invite")
-    assert len(wire) == mail_mod.DESTINATION_DAILY_CAP
+    (await db_session.get(MailQuota, key)).last_sent = None
+    await db_session.commit()
+    assert not await _charge(db_session, "invite", victim), (
+        f"the {cap}-a-day allowance for one recipient was not enforced")
 
 
-def test_the_same_recipient_is_one_recipient_however_it_is_written(wire):
+@pytest.mark.asyncio
+async def test_the_same_recipient_is_one_recipient_however_it_is_written(db_session):
     """Case and surrounding space are not a new person."""
-    assert _REAL_SEND(_msg("Victim@Example.Test"), purpose="registration")
-    assert not _REAL_SEND(_msg(" victim@example.test "), purpose="registration")
-    assert len(wire) == 1
+    assert await _charge(db_session, "registration", "Victim@Example.Test")
+    assert not await _charge(db_session, "registration", " victim@example.test ")
 
 
-def test_the_instance_has_a_ceiling_no_account_can_buy_past(wire):
+@pytest.mark.asyncio
+async def test_the_instance_has_a_ceiling_no_account_can_buy_past(db_session):
     """Registration is open, so "per account" is a bound an attacker buys more
-    of. This one cannot be bought."""
-    for i in range(mail_mod.HOURLY_BUDGET):
-        assert _REAL_SEND(_msg(f"u{i}@example.test"), purpose="registration")
-    assert not _REAL_SEND(_msg("one-more@example.test"), purpose="registration")
-    assert len(wire) == mail_mod.HOURLY_BUDGET
+    of. This one is not."""
+    from meshbay_hub.config import MailConfig
+    cfg = MailConfig()
+    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
+
+    for i in range(general):
+        assert await _charge(db_session, "registration", f"u{i}@example.test")
+    assert not await _charge(db_session, "registration", "one-more@example.test")
+
+
+@pytest.mark.asyncio
+async def test_a_flood_of_sign_ups_cannot_lock_out_a_passphrase_reset(db_session):
+    """The reserved share: someone waiting on a reset code does not wait
+    because strangers were registering all hour."""
+    from meshbay_hub.config import MailConfig
+    cfg = MailConfig()
+    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
+
+    for i in range(general):
+        assert await _charge(db_session, "registration", f"flood{i}@example.test")
+    assert not await _charge(db_session, "registration", "another@example.test")
+
+    assert await _charge(db_session, "password_reset", "waiting@example.test"), (
+        "a sign-up flood spent the share kept for recovery")
+    assert await _charge(db_session, "invite", "invited@example.test")
+
+
+@pytest.mark.asyncio
+async def test_the_allowance_is_written_down_and_survives_a_restart(db_session):
+    """It was two module dicts, so every deploy handed out a fresh budget —
+    and the hub is deployed often."""
+    from meshbay_hub.db.models import MailQuota
+
+    assert await _charge(db_session, "registration", "kept@example.test")
+
+    # A restart loses everything that was not written down. These were.
+    assert await db_session.get(MailQuota, "hour") is not None
+    assert await db_session.get(
+        MailQuota, mail_mod.destination_key("kept@example.test")) is not None
+    assert not await _charge(db_session, "registration", "kept@example.test")
+
+
+@pytest.mark.asyncio
+async def test_a_bound_an_admin_changed_is_the_one_that_applies(db_session):
+    """These are settings, not constants: the hour a budget runs out is not
+    when anyone wants to edit a file and restart."""
+    from meshbay_hub import hub_settings
+
+    await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
+    await hub_settings.set_raw(db_session, "mail.destination_daily_cap", "2")
+    await db_session.commit()
+
+    assert await _charge(db_session, "registration", "tuned@example.test")
+    assert await _charge(db_session, "registration", "tuned@example.test")
+    assert not await _charge(db_session, "registration", "tuned@example.test")
 
 
 def test_a_refusal_never_names_the_address():
@@ -168,11 +247,6 @@ def test_a_refusal_never_names_the_address():
         def emit(self, record):
             records.append(record.getMessage())
 
-    # The refusal itself, which is what the log line is built from.
-    with pytest.raises(mail_mod.MailRefused) as refusal:
-        mail_mod._budget_or_refuse("marketing", "private@example.test")
-    assert "private@example.test" not in str(refusal.value)
-
     # And the line as emitted. Level and `logging.disable` are both forced:
     # attaching a handler is not enough when an earlier test has raised the
     # logger's level, which is why the first version of this saw nothing in
@@ -262,52 +336,209 @@ async def _signed_in(client, db_session, username: str, email: str) -> dict:
 
 
 @pytest.mark.asyncio
-async def test_a_signed_in_account_cannot_mail_a_stranger_repeatedly(
-        client, db_session, wire, monkeypatch):
+async def test_an_account_may_point_the_hub_at_one_stranger_then_wait(
+        client, db_session, wire):
     """
-    `PATCH /v1/users/me` is the one path where a signed-in account chooses the
-    recipient. A cooldown alone still allowed one stranger a minute — 1440 a
-    day — so there is a ceiling per day above it.
+    `PATCH /v1/users/me` is the only path where a signed-in account chooses the
+    recipient. A short delay alone still allows one stranger per interval,
+    indefinitely, so the delay is long — and a second, different address
+    inside it is refused.
     """
-    monkeypatch.setattr(
-        "meshbay_hub.api.users.EMAIL_CHANGE_COOLDOWN", 0)
-
     headers = await _signed_in(client, db_session, "relay_b",
                                "relay_b@example.test")
     before = len(wire)
 
-    accepted = 0
-    for i in range(8):
-        r = await client.patch("/v1/users/me", headers=headers,
-                               json={"email": f"stranger{i}@example.test"})
-        if r.status_code == 200:
-            accepted += 1
-        else:
-            assert r.status_code == 429, r.text
+    r = await client.patch("/v1/users/me", headers=headers,
+                           json={"email": "a-stranger@example.test"})
+    assert r.status_code == 200, r.text
+
+    r = await client.patch("/v1/users/me", headers=headers,
+                           json={"email": "another-stranger@example.test"})
+    assert r.status_code == 429, r.text
+    assert len(wire) - before == 1, "the hub mailed a second stranger on demand"
+
+
+@pytest.mark.asyncio
+async def test_the_address_already_pending_may_be_asked_for_again(
+        client, db_session, wire):
+    """
+    The exemption, and why it is not a hole: re-asking for a code for the
+    address already pending reaches no new recipient, and that recipient is
+    bounded by its own daily allowance regardless. Without it, a typo would
+    lock the account out of correcting it for the whole window.
+    """
+    from meshbay_hub import hub_settings
+    from meshbay_hub.db.models import MailQuota
+
+    headers = await _signed_in(client, db_session, "relay_d",
+                               "relay_d@example.test")
+    typo = "jean@gmial.test"
 
-    from meshbay_hub.api.users import MAX_EMAIL_CHANGES_PER_DAY
-    assert accepted == MAX_EMAIL_CHANGES_PER_DAY, (
-        f"{accepted} strangers mailed by one account in one day")
-    assert len(wire) - before == MAX_EMAIL_CHANGES_PER_DAY
+    r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
+    assert r.status_code == 200, r.text
+
+    # The recipient's own cooldown is not what is under test here.
+    await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
+    await db_session.commit()
+    row = await db_session.get(MailQuota, mail_mod.destination_key(typo))
+    row.last_sent = None
+    await db_session.commit()
+
+    before = len(wire)
+    r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
+    assert r.status_code == 200, (
+        "a typo locked the account out of correcting it: " + r.text)
+    assert len(wire) - before == 1
 
 
 @pytest.mark.asyncio
-async def test_the_daily_ceiling_survives_the_row_being_deleted(
-        client, db_session, wire, monkeypatch):
+async def test_the_delay_survives_the_verification_row_being_deleted(
+        client, db_session, wire):
     """
-    The change path deletes this account's unverified verification rows before
-    writing a new one, so a ceiling counted from that table counts one, always
-    — which is what the first version of this counted. It comes off the IP log.
+    The handler deletes this account's unverified rows before writing a new
+    one, so a window counted from that table would count one, always. Which is
+    what the first version of this counted. It comes off the IP log.
     """
-    monkeypatch.setattr("meshbay_hub.api.users.EMAIL_CHANGE_COOLDOWN", 0)
     headers = await _signed_in(client, db_session, "relay_c",
                                "relay_c@example.test")
 
-    for i in range(3):
-        await client.patch("/v1/users/me", headers=headers,
-                           json={"email": f"c-stranger{i}@example.test"})
+    r = await client.patch("/v1/users/me", headers=headers,
+                           json={"email": "c-first@example.test"})
+    assert r.status_code == 200, r.text
+
+    from sqlalchemy import delete
+
+    from meshbay_hub.db.models import EmailVerification
+    await db_session.execute(delete(EmailVerification))
+    await db_session.commit()
 
     r = await client.patch("/v1/users/me", headers=headers,
-                           json={"email": "c-one-more@example.test"})
+                           json={"email": "c-second@example.test"})
     assert r.status_code == 429, (
-        "the ceiling was counted from a table the handler empties")
+        "the delay was counted from a table the handler empties")
+
+
+# ── The operator's controls ──────────────────────────────────────────────────
+
+async def _admin(client, db_session, username: str) -> dict:
+    from sqlalchemy import update
+
+    from meshbay_hub.db.models import User
+
+    headers = await _signed_in(client, db_session, username,
+                               f"{username}@example.test")
+    await db_session.execute(
+        update(User).where(User.username == username).values(role="admin"))
+    await db_session.commit()
+    return headers
+
+
+@pytest.mark.asyncio
+async def test_the_panel_shows_the_bounds_their_defaults_and_the_range(
+        client, db_session):
+    """The panel draws the fields from this, so a value it cannot store is one
+    the operator is told about rather than one that is silently clamped."""
+    headers = await _admin(client, db_session, "mailadmin1")
+
+    r = await client.get("/v1/admin/settings", headers=headers)
+    assert r.status_code == 200, r.text
+    body = r.json()
+    from meshbay_hub import hub_settings
+    assert set(body["mail"]) == set(hub_settings.MAIL_KEYS)
+    assert set(body["mail_defaults"]) == set(hub_settings.MAIL_KEYS)
+    assert body["mail_bounds"]["hourly_budget"][0] >= 1, (
+        "a budget of zero would stop the hub sending anything at all")
+
+
+@pytest.mark.asyncio
+async def test_an_admin_changes_a_bound_and_it_takes_effect(client, db_session):
+    headers = await _admin(client, db_session, "mailadmin2")
+
+    r = await client.patch("/v1/admin/settings", headers=headers,
+                           json={"mail": {"destination_daily_cap": 3}})
+    assert r.status_code == 200, r.text
+    assert r.json()["mail"]["destination_daily_cap"] == 3
+
+    from meshbay_hub import hub_settings
+    assert (await hub_settings.mail_limits(db_session))[
+        "destination_daily_cap"] == 3
+
+
+@pytest.mark.asyncio
+async def test_a_value_outside_the_range_is_clamped_not_stored(client, db_session):
+    """A number typed into a web form is not a reason to send without a bound."""
+    headers = await _admin(client, db_session, "mailadmin3")
+
+    r = await client.patch("/v1/admin/settings", headers=headers,
+                           json={"mail": {"hourly_budget": 0,
+                                          "destination_daily_cap": 10 ** 9}})
+    assert r.status_code == 200, r.text
+    from meshbay_hub import hub_settings
+    low, _ = hub_settings.MAIL_BOUNDS["hourly_budget"]
+    _, high = hub_settings.MAIL_BOUNDS["destination_daily_cap"]
+    assert r.json()["mail"]["hourly_budget"] == low
+    assert r.json()["mail"]["destination_daily_cap"] == high
+
+
+@pytest.mark.asyncio
+async def test_an_unknown_mail_setting_is_refused(client, db_session):
+    headers = await _admin(client, db_session, "mailadmin4")
+    r = await client.patch("/v1/admin/settings", headers=headers,
+                           json={"mail": {"send_to_everyone": 1}})
+    assert r.status_code == 422, r.text
+
+
+@pytest.mark.asyncio
+async def test_a_moderator_may_read_the_bounds_but_not_change_them(
+        client, db_session):
+    from sqlalchemy import update
+
+    from meshbay_hub.db.models import User
+
+    headers = await _signed_in(client, db_session, "mailmod",
+                               "mailmod@example.test")
+    await db_session.execute(
+        update(User).where(User.username == "mailmod").values(role="moderator"))
+    await db_session.commit()
+
+    assert (await client.get("/v1/admin/settings", headers=headers)).status_code == 200
+    r = await client.patch("/v1/admin/settings", headers=headers,
+                           json={"mail": {"hourly_budget": 10}})
+    assert r.status_code == 403, r.text
+
+
+@pytest.mark.asyncio
+async def test_the_operator_can_see_whether_the_hub_is_still_sending(
+        client, db_session, wire):
+    """There was no way to see this at all: a refusal was a line in the
+    journal, so an instance that had stopped sending sign-up codes looked,
+    from the panel, exactly like one with no sign-ups."""
+    headers = await _admin(client, db_session, "mailadmin5")
+
+    r = await client.get("/v1/admin/mail", headers=headers)
+    assert r.status_code == 200, r.text
+    before = r.json()
+    # A delta, not an absolute: creating the admin above signed somebody up,
+    # which sent a verification code and spent one of the hour's allowance.
+    assert before["general_remaining"] < before["recovery_remaining"], (
+        "the reserved share must be visible as the difference between the two")
+
+    assert await _charge(db_session, "registration", "seen@example.test")
+
+    after = (await client.get("/v1/admin/mail", headers=headers)).json()
+    assert after["hourly_used"] == before["hourly_used"] + 1
+    assert after["general_remaining"] == before["general_remaining"] - 1
+    assert after["recipients_tracked"] > before["recipients_tracked"]
+
+
+def test_the_hub_refuses_to_start_with_more_than_one_worker():
+    """The node registry and the signaling futures are per-process dicts, so a
+    second worker makes a node intermittently unreachable for half its
+    members — a symptom that describes something else entirely."""
+    from meshbay_hub.daemon import single_worker_or_exit
+
+    single_worker_or_exit(1)
+    for workers in (0, 2, 8):
+        with pytest.raises(SystemExit) as exit_:
+            single_worker_or_exit(workers)
+        assert exit_.value.code == 2