randogoth/pneumatube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: port pneumatube to Rust on fumi-core

Browse source
This commit is contained in:
randogoth 2026-10-05 11:25:11 +03:00
parent 5dda6567af
commit 65d52bd0e9
37 changed files with 4232 additions and 1912 deletions
Download patch file Download diff file Expand all files Collapse all files

317
scripts/stress_test.py
View file

@ -1,38 +1,51 @@
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["aiosmtpd>=1.4", "pynacl>=1.5", "noiseprotocol>=0.3.1"]
# ///
"""Concurrent load test for pneumatube.
Two targets:
direct Talks to a `pneumatube` subprocess directly over LMTP/SMTP,
bypassing Postfix. Isolates pneumatube's own concurrency
behaviour -- in particular, SmolWorker (worker.py) serializes
every smolmail operation onto one thread, so this is where that
design's throughput ceiling actually shows up.
behaviour: with the Rust port there is no serialization point
(fumi's store is Send + Sync and every dial is its own Noise
session), so the ceiling here belongs to the smolmail server.
postfix Drives the same load through the real docker-compose.yml rig
(Postfix -> LMTP -> pneumatube), to confirm nothing changes with
a real MTA in front. Needs podman (or docker) + compose.
Both targets spin up a throwaway smolmaild, register a gateway account plus
N recipient/sender accounts against it, generate concurrent load, and verify
every expected message actually arrived exactly once -- throughput numbers
are meaningless if messages were silently dropped or duplicated.
Both targets spin up a throwaway bunshin (the Rust smolmail server), a
gateway account plus N recipient/sender accounts against it, generate
concurrent load, and verify every expected message actually arrived
exactly once -- throughput numbers are meaningless if messages were
silently dropped or duplicated.
The sender and recipient accounts are driven in-process through the
reference smolmail.py client, which doubles as an interop check: the
Python reference client against the Rust gateway and Rust server.
Usage:
uv run python scripts/stress_test.py --target direct
uv run python scripts/stress_test.py --target postfix
uv run python scripts/stress_test.py --target both --concurrency 40
uv run scripts/stress_test.py --target direct
uv run scripts/stress_test.py --target postfix
uv run scripts/stress_test.py --target both --concurrency 40
Needs a `smolmail` checkout next to this repo (same default pneumatube
itself uses: PNEUMATUBE_SMOLMAIL_REPO, default "../smolmail").
Needs a `bunshin` and a `fumi` binary on PATH (or --server-bin/--fumi-bin),
and a `smolmail` checkout next to this repo for the load-driving client
(--smolmail-repo, default "../smolmail").
"""
from __future__ import annotations
import argparse
import base64
import contextlib
import importlib.util
import os
import shutil
import smtplib
import socket
import sqlite3
import subprocess
import sys
import tempfile
@ -44,10 +57,24 @@ from pathlib import Path
from aiosmtpd.controller import Controller
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
from pneumatube import vendor # noqa: E402
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_PORT = 1961
def load_smolmail(repo_path: Path):
"""Loads the sibling reference client as a library. smolmail.py is a
single-file `uv run --script` program, not an installed package, so it
is loaded by path from its checkout; this script's inline dependencies
(pynacl, noiseprotocol) are declared in the PEP 723 block above so the
exec'd module resolves its imports."""
module_path = repo_path / "smolmail.py"
if not module_path.is_file():
raise FileNotFoundError(f"no smolmail.py at {module_path}; pass --smolmail-repo")
spec = importlib.util.spec_from_file_location("smolmail", module_path)
module = importlib.util.module_from_spec(spec)
sys.modules["smolmail"] = module
spec.loader.exec_module(module)
return module
def free_port() -> int:
@ -111,42 +138,30 @@ class SmolServer:
process: subprocess.Popen | None = None
def parse_pubkey(text: str) -> str:
for line in text.splitlines():
if "public key" in line:
return line.strip().rsplit(" ", 1)[-1]
raise RuntimeError(f"no public key line in output:\n{text}")
def start_smolmaild(smolmail_repo: Path, tmp_dir: Path, rate_connections: int,
rate_sends: int) -> SmolServer:
def start_bunshin(binary: str, tmp_dir: Path, rate_connections: int,
rate_sends: int) -> SmolServer:
port = free_port()
key_path = tmp_dir / "server.key"
subprocess.run(
["uv", "run", str(smolmail_repo / "smolmaild.py"), "keygen", "--key", str(key_path)],
cwd=smolmail_repo, check=True, capture_output=True, text=True)
process = subprocess.Popen(
["uv", "run", str(smolmail_repo / "smolmaild.py"), "serve",
"--key", str(key_path), "--db", str(tmp_dir / "server.db"),
"--host", "127.0.0.1", "--port", str(port),
# smolmaild's connection/send rate limits are keyed by source IP
# (SPEC.md S10); every sender in this script shares one IP, so the
# defaults (120 conn/min, 60 sends/min) would measure the rate
# limiter, not pneumatube, unless raised for a many-sender test.
"--rate-connections", str(rate_connections), "--rate-sends", str(rate_sends)],
cwd=smolmail_repo, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
keygen = subprocess.run([binary, "keygen", "--key", str(key_path)],
capture_output=True, text=True, check=True)
pubkey = None
deadline = time.monotonic() + 15
while time.monotonic() < deadline:
line = process.stdout.readline()
if not line:
break
if "server public key" in line:
pubkey = line.strip().rsplit(" ", 1)[-1]
for line in keygen.stdout.splitlines():
# "public key: <b32>", not the trailing "Publish the public key..."
# sentence, which would otherwise match a looser contains-check.
if line.startswith("public key"):
pubkey = line.split()[-1]
break
if pubkey is None:
process.kill()
raise RuntimeError("smolmaild never printed its public key")
raise RuntimeError(f"no public key line in bunshin keygen output:\n{keygen.stdout}")
process = subprocess.Popen(
[binary, "serve", "--key", str(key_path), "--db", str(tmp_dir / "server.db"),
"--host", "127.0.0.1", "--port", str(port),
# bunshin's connection/send rate limits are keyed by source IP;
# every sender in this script shares one IP, so the defaults would
# measure the rate limiter, not pneumatube, unless raised for a
# many-sender test.
"--rate-connections", str(rate_connections), "--rate-sends", str(rate_sends)],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
wait_for_port("127.0.0.1", port)
return SmolServer("127.0.0.1", port, pubkey, process)
@ -160,14 +175,11 @@ class SmolAccount:
def make_account(smol, tmp_dir: Path, name: str, server: SmolServer) -> SmolAccount:
"""Keygen + trust + register, reusing smolmail.py's own CLI commands
directly (the same functions the gateway itself calls) rather than
reimplementing the protocol or shelling out per account.
A bare TCP connect (wait_for_port) can succeed slightly before a
containerized smolmaild can actually complete a Noise handshake, so
register is retried a few times rather than treating that race as fatal.
"""
"""Keygen + trust + register, reusing smolmail.py's own CLI functions
directly (the reference client) rather than reimplementing the
protocol or shelling out per account. Registered with retries, since a
bare TCP connect can succeed slightly before the server completes
Noise handshakes."""
key_path, db_path = tmp_dir / f"{name}.key", tmp_dir / f"{name}.db"
store = smol.Store(str(db_path))
smol.cmd_keygen(argparse.Namespace(key=str(key_path), force=True), store)
@ -187,14 +199,48 @@ def make_account(smol, tmp_dir: Path, name: str, server: SmolServer) -> SmolAcco
return SmolAccount(name, key_path, db_path, address)
def repoint_gateway_account(smol, gateway: SmolAccount, host: str, port: int,
pubkey_b32: str) -> None:
def make_gateway_account(fumi_bin: str, tmp_dir: Path, name: str, server: SmolServer) -> SmolAccount:
"""The gateway's own state is a fumi store, so it is bootstrapped with
the fumi CLI: keygen + trust + register, retried for the same
slow-to-handshake reason as make_account."""
key_path, db_path = tmp_dir / f"{name}.key", tmp_dir / f"{name}.db"
run = lambda *args: subprocess.run(
[fumi_bin, "--key", str(key_path), "--db", str(db_path), *args],
capture_output=True, text=True)
run("keygen")
trust = run("trust", f"{server.host}:{server.port}", server.pubkey_b32)
if trust.returncode != 0:
raise RuntimeError(f"fumi trust failed: {trust.stdout}{trust.stderr}")
address = f"{name}@{server.host}:{server.port}"
for attempt in range(20):
register = run("register", address)
if register.returncode == 0:
return SmolAccount(name, key_path, db_path, address)
if attempt == 19:
raise RuntimeError(f"fumi register failed: {register.stdout}{register.stderr}")
time.sleep(1.0)
raise RuntimeError("unreachable")
def repoint_gateway_account(gateway: SmolAccount, host: str, port: int,
pubkey_b32: str, fumi_bin: str) -> None:
"""Point the gateway's own remembered server connection at a different
host string (e.g. a compose service name) without a real round trip --
`trust` only ever stores a pin locally (smolmail.py's own cmd_trust)."""
store = smol.Store(str(gateway.db_path))
smol.cmd_trust(argparse.Namespace(host=host, key_b32=pubkey_b32, force=True), store)
store.set_account(smol.Address(gateway.username, host, port))
host string (e.g. a compose service name) without a real round trip:
`fumi trust` only ever stores a pin locally, and the account row is
rewritten directly in the store (fumi schema v2: one state row, pins
keyed by host[:port])."""
trust = subprocess.run(
[fumi_bin, "--key", str(gateway.key_path), "--db", str(gateway.db_path),
"trust", f"{host}:{port}", pubkey_b32],
capture_output=True, text=True)
if trust.returncode != 0:
raise RuntimeError(f"fumi trust failed: {trust.stdout}{trust.stderr}")
con = sqlite3.connect(gateway.db_path)
try:
con.execute("UPDATE state SET host = ?, port = ? WHERE id = 1", (host, port))
con.commit()
finally:
con.close()
# --- outbound SMTP sink (stands in for the real relay) -------------------------
@ -283,6 +329,49 @@ def run_concurrent(report: Report, concurrency: int, jobs: list) -> list[bool]:
return valid_flags
def gateway_env(gateway: SmolAccount, server: SmolServer, *,
require_accepted_sender: bool = False,
require_accepted_recipient: bool = False,
**extra: str) -> dict[str, str]:
env = {**os.environ,
"PNEUMATUBE_SMOLMAIL_KEY": str(gateway.key_path),
"PNEUMATUBE_SMOLMAIL_DB": str(gateway.db_path),
"PNEUMATUBE_SMOLMAIL_DOMAIN": f"{server.host}:{server.port}",
"PNEUMATUBE_EMAIL_DOMAIN": "gateway.example",
"PNEUMATUBE_FETCH_INTERVAL": "2",
# The accept-gate defaults live here, not in the docstring's
# "throughput, not gates" comment: the stress test relaxes them
# because its throwaway accounts never go through the accept
# dance, while the security test needs each mode explicitly.
"PNEUMATUBE_REQUIRE_ACCEPTED_SENDER": "1" if require_accepted_sender else "0",
"PNEUMATUBE_REQUIRE_ACCEPTED_RECIPIENT": "1" if require_accepted_recipient else "0",
"PNEUMATUBE_LISTEN_PROTOCOL": "lmtp",
"PNEUMATUBE_LISTEN_HOST": "127.0.0.1",
"PNEUMATUBE_SMTP_RELAY_STARTTLS": "0",
"PNEUMATUBE_GATEWAY_FROM_ADDRESS": "gateway@gateway.example"}
env.update(extra)
return env
def start_pneumatube(binary: str, env: dict[str, str], listen_port: int,
lookup_port: int, sink_port: int) -> subprocess.Popen:
env.update({"PNEUMATUBE_LISTEN_PORT": str(listen_port),
"PNEUMATUBE_LOOKUP_HOST": "127.0.0.1",
"PNEUMATUBE_LOOKUP_PORT": str(lookup_port),
"PNEUMATUBE_SMTP_RELAY_HOST": "127.0.0.1",
"PNEUMATUBE_SMTP_RELAY_PORT": str(sink_port)})
proc = subprocess.Popen([binary], cwd=REPO_ROOT, env=env,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
wait_for_port("127.0.0.1", listen_port)
return proc
def stop_proc(proc: subprocess.Popen) -> None:
proc.terminate()
with contextlib.suppress(Exception):
proc.wait(timeout=5)
# --- direct target ---------------------------------------------------------------
@ -290,12 +379,12 @@ def run_direct(args: argparse.Namespace) -> None:
smolmail_repo = Path(args.smolmail_repo).resolve()
with tempfile.TemporaryDirectory(prefix="pneumatube-stress-") as tmp:
tmp_dir = Path(tmp)
print(f"[direct] starting smolmaild in {tmp_dir}")
server = start_smolmaild(smolmail_repo, tmp_dir, args.server_rate_connections,
args.server_rate_sends)
smol = vendor.load(smolmail_repo)
print(f"[direct] starting bunshin in {tmp_dir}")
server = start_bunshin(args.server_bin, tmp_dir, args.server_rate_connections,
args.server_rate_sends)
smol = load_smolmail(smolmail_repo)
gateway = make_account(smol, tmp_dir, "gatewayuser", server)
gateway = make_gateway_account(args.fumi_bin, tmp_dir, "gatewayuser", server)
recipients = [make_account(smol, tmp_dir, f"recipient{i}", server)
for i in range(args.recipients)]
senders = [make_account(smol, tmp_dir, f"sender{i}", server)
@ -309,32 +398,11 @@ def run_direct(args: argparse.Namespace) -> None:
sink_controller.start()
listen_port, lookup_port = free_port(), free_port()
env = {**os.environ,
"PNEUMATUBE_SMOLMAIL_REPO": str(smolmail_repo),
"PNEUMATUBE_SMOLMAIL_KEY": str(gateway.key_path),
"PNEUMATUBE_SMOLMAIL_DB": str(gateway.db_path),
"PNEUMATUBE_SMOLMAIL_DOMAIN": f"{server.host}:{server.port}",
"PNEUMATUBE_EMAIL_DOMAIN": "gateway.example",
"PNEUMATUBE_FETCH_INTERVAL": "2",
# Throughput, not the accept-gate features, is this script's
# concern -- the throwaway recipient/sender accounts here
# never go through the accept dance with the gateway.
"PNEUMATUBE_REQUIRE_ACCEPTED_SENDER": "0",
"PNEUMATUBE_REQUIRE_ACCEPTED_RECIPIENT": "0",
"PNEUMATUBE_LISTEN_PROTOCOL": "lmtp",
"PNEUMATUBE_LISTEN_HOST": "127.0.0.1",
"PNEUMATUBE_LISTEN_PORT": str(listen_port),
"PNEUMATUBE_LOOKUP_HOST": "127.0.0.1",
"PNEUMATUBE_LOOKUP_PORT": str(lookup_port),
"PNEUMATUBE_SMTP_RELAY_HOST": "127.0.0.1",
"PNEUMATUBE_SMTP_RELAY_PORT": str(sink_port),
"PNEUMATUBE_SMTP_RELAY_STARTTLS": "0",
"PNEUMATUBE_GATEWAY_FROM_ADDRESS": "gateway@gateway.example"}
env = gateway_env(gateway, server, PNEUMATUBE_LOOKUP_HOST="127.0.0.1",
PNEUMATUBE_LOOKUP_PORT=str(lookup_port))
print("[direct] starting pneumatube")
proc = subprocess.Popen(["uv", "run", "pneumatube"], cwd=REPO_ROOT, env=env,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
proc = start_pneumatube(args.pneumatube_bin, env, listen_port, lookup_port, sink_port)
try:
wait_for_port("127.0.0.1", listen_port)
wait_for_port("127.0.0.1", lookup_port)
# Phase A: email -> smolmail, concurrent LMTP, mixed valid/invalid.
@ -362,20 +430,16 @@ def run_direct(args: argparse.Namespace) -> None:
report_b = Report("smolmail -> email (SEND + poll drain)")
run_concurrent(report_b, args.concurrency, jobs)
drain_deadline = time.monotonic() + max(30.0, total * 0.05)
drain_deadline = time.monotonic() + max(60.0, total * 0.25)
while sink.count() < report_b.succeeded and time.monotonic() < drain_deadline:
time.sleep(0.5)
drained = sink.count()
print(f" drained {drained}/{report_b.succeeded} into the outbound sink "
f"({'PASS' if drained == report_b.succeeded else 'FAIL: lost or stuck messages'})")
finally:
proc.terminate()
with contextlib.suppress(Exception):
proc.wait(timeout=5)
stop_proc(proc)
sink_controller.stop()
server.process.terminate()
with contextlib.suppress(Exception):
server.process.wait(timeout=5)
stop_proc(server.process)
# --- postfix target ---------------------------------------------------------------
@ -385,13 +449,18 @@ def podman(*cmd: str, **kwargs) -> subprocess.CompletedProcess:
return subprocess.run(["podman", *cmd], text=True, capture_output=True, **kwargs)
# The reference smolmaild runs in a container only for this rig; the uv
# image ships a python that can resolve its PEP 723 inline dependencies.
UV_IMAGE = "ghcr.io/astral-sh/uv:python3.13-bookworm-slim"
def run_postfix(args: argparse.Namespace) -> None:
if shutil.which("podman") is None:
print("[postfix] podman not found; skipping", file=sys.stderr)
return
smolmail_repo = Path(args.smolmail_repo).resolve()
smol = vendor.load(smolmail_repo)
smol = load_smolmail(smolmail_repo)
network = "pneumatube-stress-net"
env = {**os.environ,
"PNEUMATUBE_SMOLMAIL_DOMAIN": "smolmail-dev:1961",
@ -419,18 +488,23 @@ def run_postfix(args: argparse.Namespace) -> None:
server_dir = tmp_dir / "server"
server_dir.mkdir()
keygen = podman("run", "--rm", "-v", f"{smolmail_repo}:/smolmail:ro",
"-v", f"{server_dir}:/data",
"localhost/pneumatube_pneumatube:latest",
"uv", "run", "/smolmail/smolmaild.py", "keygen", "--key", "/data/server.key")
"-v", f"{server_dir}:/data", UV_IMAGE,
"uv", "run", "/smolmail/smolmaild.py", "keygen",
"--key", "/data/server.key")
keygen.check_returncode()
pubkey = parse_pubkey(keygen.stdout)
pubkey = None
for line in keygen.stdout.splitlines():
if "public key" in line and "Publish" not in line:
pubkey = line.strip().rsplit(" ", 1)[-1]
if pubkey is None:
raise RuntimeError(f"no public key in smolmaild keygen output:\n{keygen.stdout}")
host_port = free_port()
podman("run", "-d", "--name", "pneumatube-stress-smolmaild",
"--network", network, "--network-alias", "smolmail-dev",
"-p", f"{host_port}:1961",
"-v", f"{smolmail_repo}:/smolmail:ro", "-v", f"{server_dir}:/data",
"localhost/pneumatube_pneumatube:latest",
UV_IMAGE,
"uv", "run", "/smolmail/smolmaild.py", "serve",
"--key", "/data/server.key", "--db", "/data/server.db",
"--host", "0.0.0.0", "--port", "1961",
@ -439,12 +513,21 @@ def run_postfix(args: argparse.Namespace) -> None:
wait_for_port("127.0.0.1", host_port)
server = SmolServer("127.0.0.1", host_port, pubkey)
gateway = make_account(smol, tmp_dir, "gatewayuser", server)
gateway = make_gateway_account(args.fumi_bin, tmp_dir, "gatewayuser", server)
recipients = [make_account(smol, tmp_dir, f"recipient{i}", server)
for i in range(args.recipients)]
repoint_gateway_account(smol, gateway, "smolmail-dev", 1961, pubkey)
# The gateway db was created against 127.0.0.1:<host_port>, which
# only exists on the host; the pneumatube container resolves the
# compose-network alias instead.
repoint_gateway_account(gateway, "smolmail-dev", 1961, pubkey, args.fumi_bin)
(data_dir / "gateway.key").write_bytes(gateway.key_path.read_bytes())
(data_dir / "gateway.db").write_bytes(gateway.db_path.read_bytes())
# A leftover journal from an earlier rig run must not be applied
# to the fresh db copy (WAL salts would not match a foreign db,
# but the files must not leak into the image either).
for name in ("gateway.db-wal", "gateway.db-shm"):
with contextlib.suppress(FileNotFoundError):
(data_dir / name).unlink()
print("[postfix] starting pneumatube + postfix")
subprocess.run(["podman", "compose", "up", "-d"], cwd=REPO_ROOT, env=env, check=True)
@ -495,9 +578,12 @@ def run_postfix(args: argparse.Namespace) -> None:
subprocess.run(["podman", "compose", "down"], cwd=REPO_ROOT, env=env)
podman("rm", "-f", "pneumatube-stress-smolmaild")
podman("network", "rm", network)
with contextlib.suppress(Exception):
(data_dir / "gateway.key").unlink()
(data_dir / "gateway.db").unlink()
# The gateway db is a WAL-mode sqlite file; a stale -wal/-shm
# left behind would poison the next run's fresh db copy.
for name in ("gateway.key", "gateway.db", "gateway.db-wal", "gateway.db-shm"):
with contextlib.suppress(FileNotFoundError):
(data_dir / name).unlink()
with contextlib.suppress(OSError):
data_dir.rmdir()
@ -512,13 +598,18 @@ def main() -> None:
parser.add_argument("--recipients", type=int, default=10,
help="number of pre-registered valid recipient accounts")
parser.add_argument("--smolmail-repo", default="../smolmail")
parser.add_argument("--server-bin", default="bunshin",
help="bunshin binary for the throwaway test server")
parser.add_argument("--fumi-bin", default="fumi",
help="fumi binary used to bootstrap the gateway account")
parser.add_argument("--pneumatube-bin", default=str(REPO_ROOT / "target/release/pneumatube"))
parser.add_argument("--server-rate-connections", type=int, default=6000,
help="smolmaild --rate-connections for the throwaway test server "
"(default 120/min would rate-limit this whole test, since every "
"sender shares one source IP; pass 120 to test against that "
"realistic default instead)")
help="server --rate-connections for the throwaway test server "
"(default 120/min would rate-limit this whole test, since "
"every sender shares one source IP; pass 120 to test against "
"that realistic default instead)")
parser.add_argument("--server-rate-sends", type=int, default=6000,
help="smolmaild --rate-sends for the throwaway test server "
help="server --rate-sends for the throwaway test server "
"(see --server-rate-connections; realistic default is 60)")
args = parser.parse_args()