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

6 files changed, 256 insertions, 117 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index c44888a..b903fcf 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -781,8 +781,9 @@ async def invite_notify(
 
     try:
         await mail.send_off_loop(
-            mail.send_invite_notification,
-            email, body.code, current_user.username, group.name)
+            db, mail.send_invite_notification,
+            email, body.code, current_user.username, group.name,
+            purpose="invite")
     except Exception:
         return {"status": "send_failed"}
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 702d687..7cebd91 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -14,7 +14,7 @@ from pydantic import BaseModel, field_validator
 from sqlalchemy import delete, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub import mail
+from meshbay_hub import hub_settings, mail
 from meshbay_hub.api.deps import get_current_user, require_user_scope
 from meshbay_hub.api.middleware import limiter
 from meshbay_hub.api.netutil import client_ip
@@ -130,10 +130,6 @@ class RefreshRequest(BaseModel):
 
 # ── Endpoints ─────────────────────────────────────────────────────────────────
 
-# Between two verification mails to one pending account.
-VERIFICATION_RESEND_COOLDOWN = 120
-
-
 @router.post("/register", status_code=201)
 @limiter.limit("5/minute")
 async def register(
@@ -159,13 +155,16 @@ async def register(
             # `mail.py` bounds the recipient regardless, and this stops the
             # door being hammered — the answer is the same either way, which
             # is the one this endpoint has always given.
+            resend_cooldown = await hub_settings.get_int(
+                db, "mail.verification_resend_cooldown",
+                hub_settings.mail_default("verification_resend_cooldown"))
             recent = await db.execute(
                 select(EmailVerification).where(
                     EmailVerification.user_id == found.id,
                     EmailVerification.purpose == "registration",
                     EmailVerification.created_at
                     > datetime.now(timezone.utc)
-                    - timedelta(seconds=VERIFICATION_RESEND_COOLDOWN),
+                    - timedelta(seconds=resend_cooldown),
                 ))
             if not recent.first():
                 await _create_and_send_verification(
@@ -245,7 +244,8 @@ async def _create_and_send_verification(
     ))
     await db.flush()
     await mail.send_off_loop(
-        mail.send_verification_code, email, code, recovery_key=recovery_key)
+        db, mail.send_verification_code, email, code,
+        purpose="registration", recovery_key=recovery_key)
 
 
 class VerifyEmailRequest(BaseModel):
@@ -657,13 +657,6 @@ class UpdateProfileRequest(BaseModel):
         return v
 
 
-# One change-of-address mail per account per this many seconds. The window is
-# the account's, not the caller's IP: the point is the mailbox on the receiving
-# end, and an IP is not what fills it.
-EMAIL_CHANGE_COOLDOWN = 60
-MAX_EMAIL_CHANGES_PER_DAY = 3
-
-
 @router.patch("/me")
 @limiter.limit("10/minute")
 async def update_profile(
@@ -688,40 +681,40 @@ async def update_profile(
         new_email = body.email.strip()
         eh = hash_email_blind(new_email)
 
-        # A per-account floor under the rate limit above, which counts by IP
-        # and so is not a bound on how much mail one account can cause.
-        recent = await db.execute(
+        # How often one account may point the hub at a *different* address.
+        # Long, because this is the only path where a signed-in account chooses
+        # who receives a message, and a short delay alone still allows one
+        # stranger per interval indefinitely.
+        #
+        # Asking again for the address already pending is exempt: it reaches no
+        # new recipient — and that recipient is bounded by the per-destination
+        # allowance anyway — while without the exemption a typo would lock the
+        # account out of correcting it for the whole window.
+        pending = (await db.execute(
             select(EmailVerification).where(
                 EmailVerification.user_id == current_user.id,
                 EmailVerification.purpose == "email_change",
-                EmailVerification.created_at
-                > datetime.now(timezone.utc) - timedelta(seconds=EMAIL_CHANGE_COOLDOWN),
-            ))
-        if recent.first():
-            raise HTTPException(
-                status_code=429,
-                detail="A code was just sent. Wait a minute before asking again.")
+                EmailVerification.verified_at.is_(None),
+            ).order_by(EmailVerification.created_at.desc()))).scalars().first()
+        same_address_again = pending is not None and pending.email_hash == eh
 
-        # And a ceiling per day, because the cooldown alone still allows one
-        # stranger a minute — 1440 a day, from one account, on a hub where
-        # registration is open. This is the only path where a signed-in
-        # account chooses the recipient, so it is the only one that needs it.
-        #
-        # Counted from the IP log, not from EmailVerification: the block below
-        # deletes this account's unverified rows before writing a new one, so
-        # counting those would have counted one, always. (Which is what the
-        # first version of this did.) The log entry is worth having for its
-        # own sake — this endpoint wrote none, alone among the ones that mail.
-        today = await db.execute(
-            select(func.count()).select_from(IPLog).where(
-                IPLog.user_id == current_user.id,
-                IPLog.event == "email_change_request",
-                IPLog.timestamp > datetime.now(timezone.utc) - timedelta(days=1),
-            ))
-        if (today.scalar() or 0) >= MAX_EMAIL_CHANGES_PER_DAY:
-            raise HTTPException(
-                status_code=429,
-                detail="Too many address changes today. Try again tomorrow.")
+        if not same_address_again:
+            cooldown = await hub_settings.get_int(
+                db, "mail.email_change_cooldown",
+                hub_settings.mail_default("email_change_cooldown"))
+            since = datetime.now(timezone.utc) - timedelta(seconds=cooldown)
+            recent = await db.execute(
+                select(IPLog).where(
+                    IPLog.user_id == current_user.id,
+                    IPLog.event == "email_change_request",
+                    IPLog.timestamp > since,
+                ))
+            if recent.first():
+                raise HTTPException(
+                    status_code=429,
+                    detail="This account changed its address recently. "
+                           "Try again later, or ask for a new code for the "
+                           "address already pending.")
 
         # Check that no other active/pending account uses this email
         dup = await db.execute(
@@ -755,7 +748,8 @@ async def update_profile(
         db.add(IPLog(user_id=current_user.id, event="email_change_request",
                      ip_address=client_ip(request)))
         await db.flush()
-        await mail.send_off_loop(mail.send_email_change_code, new_email, code)
+        await mail.send_off_loop(db, mail.send_email_change_code, new_email, code,
+                                 purpose="email_change")
         pending_email = new_email
 
     await db.commit()
@@ -930,12 +924,6 @@ class ResetPasswordRequest(BaseModel):
     new_auth_key: str
 
 
-# One reset mail per account per this many seconds, whoever asks and from
-# wherever. The rate limit above counts by IP, which bounds a caller, not an
-# inbox.
-RESET_MAIL_COOLDOWN = 60
-
-
 @router.post("/password/reset-request")
 @limiter.limit("5/minute")
 async def password_reset_request(
@@ -963,12 +951,15 @@ async def password_reset_request(
         # Per account, under the per-IP limit above. Knowing the pair is the
         # hard part and this endpoint is careful about it, but once someone
         # does, the cost of repeating lands in a mailbox that is not theirs.
+        reset_cooldown = await hub_settings.get_int(
+            db, "mail.reset_cooldown",
+            hub_settings.mail_default("reset_cooldown"))
         recent = await db.execute(
             select(EmailVerification).where(
                 EmailVerification.user_id == user.id,
                 EmailVerification.purpose == "password_reset",
                 EmailVerification.created_at
-                > datetime.now(timezone.utc) - timedelta(seconds=RESET_MAIL_COOLDOWN),
+                > datetime.now(timezone.utc) - timedelta(seconds=reset_cooldown),
             ))
         if recent.first():
             return {"status": "sent_if_exists"}
@@ -996,7 +987,8 @@ async def password_reset_request(
         await db.flush()
         try:
             await mail.send_off_loop(
-                mail.send_password_reset_code, decrypt_email(user.email), code)
+                db, mail.send_password_reset_code, decrypt_email(user.email), code,
+                purpose="password_reset")
         except Exception:
             log.exception("Failed to send passphrase reset code")
         await db.commit()
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py
new file mode 100644
index 0000000..8f2e4d2
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/e5f6a7b8c9d0_add_mail_quota.py
@@ -0,0 +1,36 @@
+"""add mail_quota
+
+The hub's outbound mail allowance, kept across restarts. It lived in two
+dicts in `mail.py`, so every deploy handed out a fresh budget — and the hub is
+deployed often. A bound a restart forgets is not a bound.
+
+Revision ID: e5f6a7b8c9d0
+Revises: d4e5f6a7b8c9
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "e5f6a7b8c9d0"
+down_revision: Union[str, Sequence[str], None] = "d4e5f6a7b8c9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "mail_quota",
+        sa.Column("key", sa.String(64), primary_key=True),
+        sa.Column("window_start", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("last_sent", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    # Dropping this returns the hub to sending with no recorded history, not to
+    # sending without a bound: the limits themselves live in `hub_settings` and
+    # in the configuration file.
+    op.drop_table("mail_quota")
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index e6c7d33..b052e00 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -308,6 +308,28 @@ class UserPreference(Base):
     value:   Mapped[str] = mapped_column(Text, nullable=False)
 
 
+class MailQuota(Base):
+    """How much mail has gone where, kept across restarts.
+
+    This was a pair of dicts in `mail.py`, which meant a restart handed out a
+    fresh allowance — and a hub restarts whenever it is deployed. A budget a
+    restart forgets is not a budget, for the same reason the denylist is
+    persisted rather than held in memory (S3).
+
+    One row per thing being counted:
+      `hour`             the instance's hourly total
+      `dest:<hash>`      one recipient, hashed — this table must not become a
+                         list of plaintext addresses (S2)
+    """
+
+    __tablename__ = "mail_quota"
+
+    key:          Mapped[str] = mapped_column(String(64), primary_key=True)
+    window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+    count:        Mapped[int] = mapped_column(Integer, default=0)
+    last_sent:    Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
 class HubSetting(Base):
     """
     Instance-wide settings an admin changes at runtime from the panel.
diff --git a/packages/meshbay-hub/src/meshbay_hub/mail.py b/packages/meshbay-hub/src/meshbay_hub/mail.py
index 5d13b8d..661a1f3 100644
--- a/packages/meshbay-hub/src/meshbay_hub/mail.py
+++ b/packages/meshbay-hub/src/meshbay_hub/mail.py
@@ -16,12 +16,22 @@ import asyncio
 import hashlib
 import logging
 import smtplib
-import time
+from datetime import datetime, timedelta, timezone
 from email.message import EmailMessage
 
 log = logging.getLogger(__name__)
 
 
+_hub_domain: str = "meshbay.org"
+_hub_url: str = "https://meshbay.org"
+
+
+def configure(hub_id: str) -> None:
+    global _hub_domain, _hub_url
+    _hub_domain = hub_id
+    _hub_url = f"https://{hub_id}"
+
+
 class MailRefused(Exception):
     """The gate below declined to send. Never carries the address."""
 
@@ -36,82 +46,136 @@ class MailRefused(Exception):
 #   email_change   confirm a new address before it replaces the old one
 #
 # Only `registration` and `email_change` can reach an address this hub has no
-# prior relationship with. Both are necessary; both are why the per-destination
-# bound below is keyed on the recipient rather than on who asked.
+# prior relationship with. Both are necessary; both are why the bound that
+# matters is keyed on the recipient rather than on who asked.
 ALLOWED_PURPOSES = frozenset({
     "registration", "password_reset", "invite", "email_change"})
 
-# Per recipient, across every purpose, every account and every IP. This is the
-# bound that matters: it is what a person being mail-bombed actually
-# experiences, and no combination of accounts, addresses or endpoints moves it.
-DESTINATION_COOLDOWN_SECONDS = 120
-DESTINATION_DAILY_CAP = 5
+# The two a person is actively waiting on. They may spend the whole hourly
+# budget; the other two may not spend the reserved share of it, so a flood of
+# sign-ups cannot lock out someone trying to recover their passphrase.
+RECOVERY_PURPOSES = frozenset({"password_reset", "invite"})
 
-# Instance-wide ceiling. Registration is open, so "per account" is a bound an
-# attacker buys more of; this one cannot be bought. Sized far above a real
-# hub's traffic — meshbay.org sends single-digit mails a day — and low enough
-# that being used as a relay is not worth the trouble.
-HOURLY_BUDGET = 200
 
-# Single-process state, like `_connected_nodes` and `_node_groups` next door:
-# the hub serves on one worker (`server.workers` defaults to 1) and the
-# signaling registries already require it. A restart clears these, which costs
-# at most one burst and is not a security decision — unlike the denylist,
-# which persists for exactly that reason (S3).
-_destinations: dict[str, tuple[float, int, float]] = {}   # key → (last, day count, day start)
-_hour: tuple[float, int] = (0.0, 0)                       # (window start, count)
+def destination_key(address: str) -> str:
+    """A stable handle for one recipient that is not the address itself.
 
+    Hashed because this table would otherwise be the one place in the hub
+    holding a list of plaintext addresses — the rest of the codebase goes to
+    the trouble of encrypting them at rest (S2).
+    """
+    return "dest:" + hashlib.sha256(
+        address.strip().lower().encode()).hexdigest()[:32]
 
-def _destination_key(address: str) -> str:
-    """A stable handle for one recipient that is not the address itself.
 
-    Hashed because this dict is the one place in the hub that would otherwise
-    hold a list of plaintext addresses in memory — the rest of the codebase
-    goes to the trouble of encrypting them at rest (S2).
+async def _take(db, key: str, window: timedelta, ceiling: int,
+                cooldown: timedelta | None) -> None:
+    """Charge one send against a counter, or raise MailRefused.
+
+    Rows live in `mail_quota` rather than in a module dict, so the allowance
+    survives the restart that a deploy is. The caller owns the commit — which
+    means a send that then fails to commit is not charged, and that is the
+    right way round: the alternative charges for mail nobody received.
     """
-    return hashlib.sha256(address.strip().lower().encode()).hexdigest()[:32]
+    from meshbay_hub.db.models import MailQuota
+
+    now = datetime.now(timezone.utc)
+    row = await db.get(MailQuota, key)
+    if row is None:
+        row = MailQuota(key=key, window_start=now, count=0, last_sent=None)
+        db.add(row)
 
+    started = row.window_start
+    if started.tzinfo is None:
+        started = started.replace(tzinfo=timezone.utc)
+    if now - started >= window:
+        row.window_start, row.count = now, 0
 
-def _budget_or_refuse(purpose: str, address: str) -> None:
-    """Raise MailRefused unless this hub may send this, to this person, now."""
-    global _hour
+    if cooldown is not None and row.last_sent is not None:
+        last = row.last_sent
+        if last.tzinfo is None:
+            last = last.replace(tzinfo=timezone.utc)
+        if now - last < cooldown:
+            raise MailRefused("too soon since the last message to this recipient")
+
+    if row.count >= ceiling:
+        raise MailRefused("allowance spent for this window")
+
+    row.count += 1
+    row.last_sent = now
+
+
+async def reserve(db, purpose: str, address: str) -> None:
+    """Charge one send, or raise MailRefused. The caller owns the commit."""
+    from meshbay_hub import hub_settings
 
     if purpose not in ALLOWED_PURPOSES:
         raise MailRefused(f"not a purpose this hub sends mail for: {purpose!r}")
 
-    now = time.monotonic()
-    start, count = _hour
-    if now - start >= 3600:
-        start, count = now, 0
-    if count >= HOURLY_BUDGET:
-        _hour = (start, count)
-        raise MailRefused("the hub's hourly mail budget is spent")
+    limits = await hub_settings.mail_limits(db)
 
-    key = _destination_key(address)
-    last, sent_today, day_start = _destinations.get(key, (0.0, 0, now))
-    if now - day_start >= 86400:
-        sent_today, day_start = 0, now
-    if last and now - last < DESTINATION_COOLDOWN_SECONDS:
-        raise MailRefused("too soon since the last mail to this recipient")
-    if sent_today >= DESTINATION_DAILY_CAP:
-        raise MailRefused("this recipient has had its daily allowance")
+    # The instance ceiling first, and the recovery share is subtracted for the
+    # purposes that are not one. Registration is open, so a limit counted per
+    # account or per IP is one an attacker buys more of; this one is not.
+    budget = limits["hourly_budget"]
+    if purpose not in RECOVERY_PURPOSES:
+        budget = max(0, budget - limits["hourly_reserved_for_recovery"])
+    await _take(db, "hour", timedelta(hours=1), budget, None)
 
-    if len(_destinations) > 10_000:
-        for k, (_, _, started) in list(_destinations.items()):
-            if now - started >= 86400:
-                _destinations.pop(k, None)
+    # Then the recipient: across every purpose, account and endpoint. This is
+    # what a person being mail-bombed actually experiences, and the only bound
+    # that describes it.
+    await _take(
+        db, destination_key(address), timedelta(days=1),
+        limits["destination_daily_cap"],
+        timedelta(seconds=limits["destination_cooldown_seconds"]))
 
-    _destinations[key] = (now, sent_today + 1, day_start)
-    _hour = (start, count + 1)
 
-_hub_domain: str = "meshbay.org"
-_hub_url: str = "https://meshbay.org"
+async def status(db) -> dict:
+    """What the operator sees in the panel: is the hub still sending?"""
+    from meshbay_hub import hub_settings
+    from meshbay_hub.db.models import MailQuota
+    from sqlalchemy import func, select
 
+    limits = await hub_settings.mail_limits(db)
+    row = await db.get(MailQuota, "hour")
+    used = 0
+    window_start = None
+    if row is not None:
+        started = row.window_start
+        if started.tzinfo is None:
+            started = started.replace(tzinfo=timezone.utc)
+        if datetime.now(timezone.utc) - started < timedelta(hours=1):
+            used, window_start = row.count, started.isoformat()
 
-def configure(hub_id: str) -> None:
-    global _hub_domain, _hub_url
-    _hub_domain = hub_id
-    _hub_url = f"https://{hub_id}"
+    recipients = await db.scalar(
+        select(func.count()).select_from(MailQuota)
+        .where(MailQuota.key.like("dest:%"))) or 0
+
+    general = max(0, limits["hourly_budget"] - limits["hourly_reserved_for_recovery"])
+    return {
+        "hourly_budget":      limits["hourly_budget"],
+        "hourly_used":        used,
+        "hour_started_at":    window_start,
+        "reserved_for_recovery": limits["hourly_reserved_for_recovery"],
+        # What a sign-up would meet right now, which is the number an operator
+        # is actually asking about when they open this panel.
+        "general_remaining":  max(0, general - used),
+        "recovery_remaining": max(0, limits["hourly_budget"] - used),
+        "recipients_tracked": recipients,
+    }
+
+
+async def purge_expired_quota(db) -> int:
+    """Drop counters whose window has passed. Returns how many went."""
+    from meshbay_hub.db.models import MailQuota
+    from sqlalchemy import delete
+
+    cutoff = datetime.now(timezone.utc) - timedelta(days=1)
+    result = await db.execute(
+        delete(MailQuota).where(MailQuota.window_start < cutoff))
+    await db.commit()
+    return result.rowcount or 0
 
 
 def _send(msg: EmailMessage, *, purpose: str) -> bool:
@@ -123,11 +187,14 @@ def _send(msg: EmailMessage, *, purpose: str) -> bool:
     keyword-required so forgetting it is a TypeError rather than an
     unrestricted send.
     """
-    try:
-        _budget_or_refuse(purpose, msg["To"] or "")
-    except MailRefused as refusal:
-        # Never the address: this line goes to the journal.
-        log.warning("Mail refused (%s): %s", purpose, refusal)
+    if purpose not in ALLOWED_PURPOSES:
+        # The structural half of the gate, and the half that needs no state:
+        # it is here rather than only in `reserve` so that nothing which puts a
+        # message on the wire — a new helper, a script, a test — can name a
+        # reason this hub does not send mail for. The counting half is
+        # `reserve`, which needs a database session and so cannot live here.
+        log.warning("Mail refused: %r is not a purpose this hub sends for",
+                    purpose)
         return False
     try:
         with smtplib.SMTP("localhost", 25, timeout=10) as s:
@@ -138,7 +205,7 @@ def _send(msg: EmailMessage, *, purpose: str) -> bool:
         return False
 
 
-async def send_off_loop(fn, *args, **kwargs) -> None:
+async def send_off_loop(db, fn, *args, purpose: str, **kwargs) -> bool:
     """Run one of the `send_*` functions below in a worker thread.
 
     `smtplib` is synchronous and this one waits up to ten seconds. Called
@@ -150,8 +217,22 @@ async def send_off_loop(fn, *args, **kwargs) -> None:
     reaches it (`PATCH /v1/users/me`) had no rate limit at all.
 
     So the cost of a slow MTA is one request now, not the instance.
+
+    It is also the one door: `reserve` charges the send against the recipient's
+    allowance and the instance's before anything is handed to the thread, and
+    the recipient is `args[0]` because that is the first parameter of every
+    `send_*` function below. Returns whether the message went. The caller owns
+    the commit, so a request that fails afterwards is not charged for mail
+    nobody received.
     """
+    try:
+        await reserve(db, purpose, args[0])
+    except MailRefused as refusal:
+        # Never the address: this line goes to the journal.
+        log.warning("Mail refused (%s): %s", purpose, refusal)
+        return False
     await asyncio.to_thread(fn, *args, **kwargs)
+    return True
 
 
 def send_verification_code(to: str, code: str, recovery_key: str | None = None) -> None:
diff --git a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
index 7f8a5f2..dfe7c78 100644
--- a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
+++ b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
@@ -57,6 +57,13 @@ async def cleanup_loop(get_session):
                     stale = await purge_stale_pending_users(db)
                     if stale:
                         log.info("Purged %d stale pending users", stale)
+                    # One row per recipient the hub has written to, and the
+                    # window is a day: without this the table grows for the
+                    # life of the instance and nothing ever reads the old rows.
+                    from meshbay_hub import mail
+                    quota = await mail.purge_expired_quota(db)
+                    if quota:
+                        log.info("Purged %d expired mail counters", quota)
             except asyncio.CancelledError:
                 raise
             except Exception as e: