From 38738122b331e76b2f119c486c060057a99fc0c5 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 06:59:13 +0800 Subject: [PATCH] feat(messaging): deliver durable outbox events with publisher confirms --- backend/README.md | 12 +- .../src/kf_oj/infra/messaging/publisher.py | 180 ++++++++++++++ backend/tests/test_outbox.py | 219 ++++++++++++++++++ 3 files changed, 410 insertions(+), 1 deletion(-) create mode 100644 backend/src/kf_oj/infra/messaging/publisher.py create mode 100644 backend/tests/test_outbox.py diff --git a/backend/README.md b/backend/README.md index f1f8d7d..1f75f74 100644 --- a/backend/README.md +++ b/backend/README.md @@ -18,7 +18,17 @@ make worker-check API 提供存活、真实登录/退出/当前用户、公开题目与有效 Worker 租约的清单/工件读取。 配置数据库后在 lifespan 内组合模块，配置缺失时业务端口明确不可用，存活接口仍可用。 Worker 目前仅支持 `--check`，默认拒绝进入尚未实现的任务循环。 -提交创建、判题投递/成绩和模型接入尚未实现。账号/题目/工件命令及限制见对应模块 README。 +提交创建 API 在持久化源码后，同一事务写入初始评测和 outbox；按用户与幂等键 +防止重复创建，待评测总量和个人待评测数有固定上限。 + +待投递事件使用独立进程 `uv run --locked python -m kf_oj.infra.messaging.publisher`； +配置 `KF_OJ_DATABASE_URL` 为专用平台角色，`KF_OJ_AMQP_URL` 为内部 broker。 +发布器一次只锁定一个到期事件；使用持久队列、消息及 mandatory publisher confirm， +确认后才提交投递标记。锁过期允许重投，消息不是执行权，Coordinator claim 才是。 +队列最多 64 条、32 KiB，拒绝溢出，消息有效期 10 分钟；broker 不可用采用有界退避， +不阻塞提交事务。日志仅固定错误代码，不打印源码、答案或连接口令。 +后台每次投递最多 12 秒，关闭连接最多 2 秒，SIGTERM 在当前有界操作结束后退出。 +账号/题目/工件命令及限制见对应模块 README，前端成绩读取与模型接入尚未实现。 Python 虚拟环境位于 `backend/.venv`，可在 `backend/` 直接使用 `uv sync --locked` 和 `uv run --locked`。 数据库由 Alembic 唯一管理；业务模型在 `migrations/env.py` 中显式登记元数据。 diff --git a/backend/src/kf_oj/infra/messaging/publisher.py b/backend/src/kf_oj/infra/messaging/publisher.py new file mode 100644 index 0000000..558dab3 --- /dev/null +++ b/backend/src/kf_oj/infra/messaging/publisher.py @@ -0,0 +1,180 @@ +"""Finite transactional outbox delivery; confirmations are notifications, not grades.""" + +import asyncio +import logging +import signal +from uuid import uuid4 + +import aio_pika +from sqlalchemy import text + +from kf_oj.contracts.judge import JudgeTaskNotification +from kf_oj.core.settings import Settings +from kf_oj.infra.db.session import build_engine, build_session_factory + +QUEUE = "kf.judge.evaluations.v1" +QUEUE_ARGUMENTS = { + "x-max-length": 64, + "x-max-length-bytes": 32768, + "x-overflow": "reject-publish", + "x-message-ttl": 600000, +} + + +class OutboxRepository: + def __init__(self, sessions): + self.sessions = sessions + + async def claim(self): + owner = uuid4() + async with self.sessions.begin() as s: + row = ( + await s.execute( + text("""WITH candidate AS ( + SELECT id FROM outbox + WHERE delivered_at IS NULL AND next_attempt_at<=clock_timestamp() + AND (locked_until IS NULL OR locked_until<=clock_timestamp()) + ORDER BY next_attempt_at,id FOR UPDATE SKIP LOCKED LIMIT 1) + UPDATE outbox o SET lock_owner=:owner, + locked_until=clock_timestamp()+interval '15 seconds' + FROM candidate c WHERE o.id=c.id RETURNING o.id,o.evaluation_id"""), + {"owner": owner}, + ) + ).first() + return None if row is None else (row.id, row.evaluation_id, owner) + + async def confirmed(self, item): + async with self.sessions.begin() as s: + result = await s.execute( + text("""UPDATE outbox SET delivered_at=clock_timestamp(), + locked_until=NULL,lock_owner=NULL,last_error_code=NULL + WHERE id=:id AND lock_owner=:owner AND locked_until>clock_timestamp()"""), + {"id": item[0], "owner": item[2]}, + ) + return result.rowcount == 1 + + async def failed(self, item): + async with self.sessions.begin() as s: + await s.execute( + text("""UPDATE outbox SET failures=least(failures+1,100), + next_attempt_at=clock_timestamp()+ + least(power(2,least(failures+1,6)),60)*interval '1 second', + locked_until=NULL,lock_owner=NULL,last_error_code='broker_unavailable' + WHERE id=:id AND lock_owner=:owner AND locked_until>clock_timestamp()"""), + {"id": item[0], "owner": item[2]}, + ) + + +class ConfirmedBroker: + def __init__(self, url): + self.url = url + self.connection = None + self.channel = None + + async def open(self): + async with asyncio.timeout(10): + await self._open() + + async def _open(self): + # Use a finite connect, not a reconnect loop holding an outbox lock. + self.connection = await aio_pika.connect(self.url, timeout=5, heartbeat=30) + self.channel = await self.connection.channel(publisher_confirms=True, on_return_raises=True) + await self.channel.declare_queue(QUEUE, durable=True, arguments=QUEUE_ARGUMENTS, timeout=5) + + async def publish(self, item): + # The complete operation ends before the database claim expires. + async with asyncio.timeout(12): + await self._publish(item) + + async def _publish(self, item): + if self.channel is None or self.channel.is_closed: + await self.open() + body = ( + JudgeTaskNotification(schema_version=1, kind="judge.evaluation", evaluation_id=item[1]) + .model_dump_json() + .encode() + ) + assert len(body) <= 512 + confirmation = await self.channel.default_exchange.publish( + aio_pika.Message( + body, + content_type="application/json", + delivery_mode=aio_pika.DeliveryMode.PERSISTENT, + message_id=str(item[0]), + ), + routing_key=QUEUE, + mandatory=True, + timeout=5, + ) + from pamqp.commands import Basic + + if not isinstance(confirmation, Basic.Ack): + raise RuntimeError("notification not confirmed") + + async def close(self): + if self.connection is not None and not self.connection.is_closed: + try: + await asyncio.wait_for(self.connection.close(), timeout=2) + except Exception: + pass + self.connection = self.channel = None + + +async def run(settings: Settings, stop: asyncio.Event): + if settings.amqp_url is None: + raise ValueError("AMQP configuration required") + engine = build_engine(settings) + repository = OutboxRepository(build_session_factory(engine)) + broker = ConfirmedBroker(settings.amqp_url.get_secret_value()) + logger = logging.getLogger("kf_oj.outbox") + # Client exceptions may include connection URIs; suppress transport-library logging. + for name in ("aio_pika", "aiormq", "pamqp"): + logging.getLogger(name).disabled = True + try: + async with engine.connect() as c: + allowed = await c.scalar( + text("""SELECT NOT r.rolsuper AND NOT r.rolbypassrls + AND pg_has_role(current_user,'kf_oj_platform','MEMBER') + AND NOT has_column_privilege(current_user,'evaluations','state','UPDATE') + FROM pg_roles r WHERE r.rolname=current_user""") + ) + if not allowed: + raise ValueError("dedicated platform database role required") + while not stop.is_set(): + try: + item = await repository.claim() + if item is not None: + try: + await broker.publish(item) + await repository.confirmed(item) + except Exception: + await repository.failed(item) + await broker.close() + logger.warning("outbox broker unavailable") + except Exception: + logger.warning("outbox database unavailable") + try: + await asyncio.wait_for(stop.wait(), timeout=1) + except TimeoutError: + pass + finally: + await broker.close() + await engine.dispose() + + +def main(): + async def entry(): + stop = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stop.set) + await run(Settings(), stop) + + try: + asyncio.run(entry()) + except Exception: + raise SystemExit("outbox unavailable") from None + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_outbox.py b/backend/tests/test_outbox.py new file mode 100644 index 0000000..5ad1b68 --- /dev/null +++ b/backend/tests/test_outbox.py @@ -0,0 +1,219 @@ +"""Real private PostgreSQL/RabbitMQ control plane, no user program execution.""" + +import asyncio +import json +import os +import secrets +import subprocess +import sys +from pathlib import Path +from urllib.parse import urlsplit +from uuid import uuid4 + +import aio_pika +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine +from test_problems import import_fixture, platform_role + +from kf_oj.api.app import create_app +from kf_oj.contracts.submissions import CreateSubmission +from kf_oj.core.settings import Settings +from kf_oj.infra.messaging.publisher import QUEUE, ConfirmedBroker, OutboxRepository + + +def test_actual_confirm_duplicate_lock_recovery_and_broker_failure( + postgres_url, tmp_path, monkeypatch +): + asyncio.run(outbox_acceptance(postgres_url, tmp_path, monkeypatch)) + + +async def outbox_acceptance(postgres_url, tmp_path, monkeypatch): + config = os.environ.get("KF_OJ_TEST_AMQP_CONFIG") + if config is None: + pytest.skip("private RabbitMQ control acceptance requires KF_OJ_TEST_AMQP_CONFIG") + url = json.loads(Path(config).read_text())["amqp_url"] + parsed = urlsplit(url) + assert parsed.hostname == "127.0.0.1" and parsed.port == 26572 + _, policy = import_fixture(tmp_path / "import") + engine = create_async_engine(postgres_url) + platform_role(engine) + monkeypatch.setattr("kf_oj.api.app.build_engine", lambda _: engine) + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + broker = ConfirmedBroker(url) + connection = await aio_pika.connect(url, timeout=5) + try: + async with app.router.lifespan_context(app): + owner = await app.state.identity.create_account( + "outbox_test", secrets.token_urlsafe(24) + ) + version = await app.state.problems.import_directory(tmp_path / "import") + created = await app.state.submissions.create( + owner, + CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="synthetic control event; not compiled", + ), + ) + repo = OutboxRepository(app.state.sessions) + claimed = await asyncio.gather(repo.claim(), repo.claim()) + assert sum(v is not None for v in claimed) == 1 + item = next(v for v in claimed if v is not None) + assert item[1] == created.evaluation_id + # Kill a separate real publisher process after broker confirm, before DB mark. + code = """import asyncio,json,sys,signal +from uuid import UUID +from pathlib import Path +from kf_oj.infra.messaging.publisher import ConfirmedBroker +async def main(): + item=tuple(UUID(x) for x in json.loads(sys.stdin.readline())) + broker=ConfirmedBroker(json.loads(Path(sys.argv[1]).read_text())["amqp_url"]) + await broker.publish(item) + print("CONFIRMED",flush=True) + await asyncio.Event().wait() +asyncio.run(main()) +""" + child = subprocess.Popen( + [sys.executable, "-c", code, config], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + child.stdin.write(json.dumps([str(x) for x in item]) + "\n") + child.stdin.flush() + assert ( + await asyncio.wait_for(asyncio.to_thread(child.stdout.readline), 10) + == "CONFIRMED\n" + ) + finally: + child.kill() + await asyncio.to_thread(child.wait, 5) + child.stdin.close() + child.stdout.close() + child.stderr.close() + assert child.returncode == -9 + channel = await connection.channel() + queue = await channel.get_queue(QUEUE) + first = await queue.get(timeout=5) + assert json.loads(first.body) == { + "schema_version": 1, + "kind": "judge.evaluation", + "evaluation_id": str(created.evaluation_id), + } + assert len(first.body) <= 512 and first.delivery_mode == 2 + # A publisher lost after confirm cannot mark DB; actual DB clock lock expires. + assert await repo.claim() is None + await asyncio.sleep(15.1) + retry = await repo.claim() + assert retry[:2] == item[:2] and retry[2] != item[2] + assert not await repo.confirmed(item) + await broker.publish(retry) + assert await repo.confirmed(retry) + second = await queue.get(timeout=5) + assert second.body == first.body and second.message_id == first.message_id + await first.ack() + await second.ack() + assert await repo.claim() is None + async with app.state.sessions() as s: + assert ( + await s.scalar( + text("SELECT count(*) FROM outbox WHERE delivered_at IS NOT NULL") + ) + == 1 + ) + # A second transaction succeeds even when a separate broker is unavailable. + await app.state.submissions.create( + owner, + CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="broker failure synthetic event", + ), + ) + failed = await repo.claim() + unavailable = ConfirmedBroker(url.replace(":26572/", ":26579/")) + try: + with pytest.raises((ConnectionError, OSError)): + await unavailable.publish(failed) + finally: + await unavailable.close() + await repo.failed(failed) + assert await repo.claim() is None + async with app.state.sessions() as s: + row = ( + await s.execute( + text("""SELECT failures,last_error_code, + next_attempt_at>clock_timestamp() deferred,delivered_at IS NULL pending, + lock_owner IS NULL unlocked FROM outbox WHERE id=:id"""), + {"id": failed[0]}, + ) + ).one() + assert tuple(row) == (1, "broker_unavailable", True, True, True) + # Preserve the broker-failure row for inspection while testing another crash window. + async with app.state.sessions.begin() as s: + await s.execute( + text( + "UPDATE outbox SET next_attempt_at=clock_timestamp()+interval '1 hour' " + "WHERE id=:id" + ), + {"id": failed[0]}, + ) + before = await app.state.submissions.create( + owner, + CreateSubmission( + problem_version_id=version, + language="c17", + idempotency_key=uuid4(), + source="before-confirm synthetic event", + ), + ) + pending = await repo.claim() + assert pending[1] == before.evaluation_id + preconfirm = code.replace("await broker.publish(item)", "await broker.open()") + child = subprocess.Popen( + [sys.executable, "-c", preconfirm, config], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + child.stdin.write(json.dumps([str(x) for x in pending]) + "\n") + child.stdin.flush() + assert ( + await asyncio.wait_for(asyncio.to_thread(child.stdout.readline), 10) + == "CONFIRMED\n" + ) + finally: + child.kill() + await asyncio.to_thread(child.wait, 5) + child.stdin.close() + child.stdout.close() + child.stderr.close() + assert child.returncode == -9 + assert await queue.get(fail=False, timeout=5) is None + await asyncio.sleep(15.1) + retry = await repo.claim() + assert retry[:2] == pending[:2] + await broker.publish(retry) + assert await repo.confirmed(retry) + delivered = await queue.get(timeout=5) + assert json.loads(delivered.body)["evaluation_id"] == str(before.evaluation_id) + await delivered.ack() + + finally: + await broker.close() + await connection.close()