From b70283c2ff8c4e4997c826618be61596a5d55297 Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 8 Oct 2026 11:50:15 +0800 Subject: [PATCH] feat(work-orders): synchronize peripheral changes through events --- docs/operational-task-workers.md | 4 +- docs/peripheral-work-order-events.md | 44 +++ samples/peripheral/store.py | 26 +- scripts/sync_peripheral_work_orders.py | 115 ++++++- .../test_peripheral_sample_live.py | 6 +- .../test_peripheral_work_order_events.py | 84 ++++++ tests/unit/test_peripheral_push_sync.py | 284 ++++++++++++++++++ .../unit/test_peripheral_work_order_events.py | 50 +++ 8 files changed, 600 insertions(+), 13 deletions(-) create mode 100644 docs/peripheral-work-order-events.md create mode 100644 tests/integration/test_peripheral_work_order_events.py create mode 100644 tests/unit/test_peripheral_push_sync.py create mode 100644 tests/unit/test_peripheral_work_order_events.py diff --git a/docs/operational-task-workers.md b/docs/operational-task-workers.md index 4976fc3..5439545 100644 --- a/docs/operational-task-workers.md +++ b/docs/operational-task-workers.md @@ -79,9 +79,9 @@ conda run -n server python scripts/sync_peripheral_work_orders.py --once conda run -n server python scripts/sync_peripheral_work_orders.py ``` -使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器读取分页工单，将外围版本投影到 `external_work_order` 任务。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。 +使用相同工作身份，并配置 `OPERATIONAL_ACTION_PERIPHERAL_URL`。同步器订阅源工单变更事件，收到消息即读取分页工单，将外围版本投影到 `external_work_order` 任务；默认每 300 秒兜底对账。合成来源在摘要和证据中明确标注；工单完成只表示该外部任务完成，不代表其他调查自动完成。事件协议、`tjwater_v2` 范围与启用步骤见 [外围工单主动同步](peripheral-work-order-events.md)。 -工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单仍由独立 worker 使用服务身份周期同步，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与 worker 固定项目不一致。 +工况任务面板的刷新按钮只重新读取当前项目的工作台快照，不直接触发外围系统全量对账。外围工单由独立同步器使用服务身份响应后端事件，避免多标签页或普通用户刷新放大全量扫描，也避免浏览器项目与同步器固定项目不一致。 通用合成检查工单不属于预约任务。早期种子脚本创建的通信、压力和零流量检查工单已经退役，种子脚本不再恢复它们；同步器按不可编辑的专用 seed actor 跳过旧种子工单。投诉、爆管事件以及调查动作实际创建的现场工单继续同步。`018_remove_synthetic_check_task_projections.sql` 只清理专用 seed actor 创建的旧检查投影，不按用户可编辑的标题或状态删除任务。三张独立演示工单已迁入外围样板，由同一同步器维护。 diff --git a/docs/peripheral-work-order-events.md b/docs/peripheral-work-order-events.md new file mode 100644 index 0000000..52aa9d1 --- /dev/null +++ b/docs/peripheral-work-order-events.md @@ -0,0 +1,44 @@ +# 外围工单主动同步 + +工单的创建、编辑、现场反馈和删除改为由来源事件触发同步。浏览器继续使用已有工作台 SSE，不需要新增连接或定时工单轮询。 + +```text +外围工单写入 + → 同一 PostgreSQL 事务保存源记录、历史版本和变更事件 + → COMMIT 后 NOTIFY + → Server 工作台 SSE 唤醒外围同步器 + → 同步器通过授权 API 更新任务投影 + → task/action/notification 事件推送给浏览器 +``` + +## 数据库范围 + +本次仅启用 `tjwater_v2`。没有新增表、列或迁移；复用 `workbench.domain_event_log` 的 `workbench.work_order.changed` topic。源事件的聚合类型为 `peripheral_work_order`，元数据只包含 `source=peripheral_sample`、工单 ID 和 `created|updated|deleted`，不包含工单正文。事务失败时，源记录、历史、事件及通知一起回滚；同一请求的幂等重放不再产生事件。 + +外围服务只写源记录及源变更事实，不直接改业务任务。任务投影、调查创建和通知仍通过 Server 状态机和现有授权 API 处理。其他数据库不写入该源事件；扩展前遵循 [数据库变更门禁](workbench-database-rollout.md)。事件表缺失或无写权限时，`tjwater_v2` 的工单写入返回存储错误，不会假装成功。 + +## 同步器 + +从 Server 仓库运行已有命令： + +```bash +conda run -n server python scripts/sync_peripheral_work_orders.py +conda run -n server python scripts/sync_peripheral_work_orders.py --once +``` + +继续复用 `OPERATIONAL_WORKER_*` OAuth 工作身份及 `OPERATIONAL_ACTION_PERIPHERAL_URL`，不需要新账号或密钥。启动和游标过期时先读取事件游标，再进行全量对账；扫描期间的事件从该游标继续补读。相同源版本仍由原幂等键去重。同步器只被 `peripheral_work_order` 源事件唤醒，不被自己写出的任务投影事件再次唤醒。 + +正常状态收到源事件即扫描；无事件时默认每 300 秒对账一次，处理漏通知、断线、旧批量脚本直接写库及重启。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前清除唤醒标记，扫描期间新到达的事件会触发下一轮，失败后的重试也不依赖新消息。 + +浏览器收到源事件时不会提前刷新业务快照，而是等待任务投影事件后读取对应任务。工况手动刷新仍只读 Server 权威快照。工单事件不等于 Header 通知；只有需要处理的业务待办或告警才进入通知列表。 + +## 验证与启用 + +```bash +conda run -n server python -m pytest tests/unit/test_peripheral_push_sync.py tests/unit/test_peripheral_work_order_events.py tests/unit/test_peripheral_task_sync.py -q +PERIPHERAL_SAMPLE_LIVE_TEST=1 conda run -n server python -m pytest tests/integration/test_peripheral_work_order_events.py -q +``` + +数据库测试强制限定 `tjwater_v2`，使用随机项目键和回滚事务，不保留测试工单或事件。覆盖创建、修改、反馈、删除、幂等重放及回滚无事件。HTTP 隔离测试覆盖源事件即时投影、游标重置、断线续传、失败重试和监听取消。 + +上线时更新外围服务、Server 同步器及前端，再按既有发布流程重启对应进程。提交代码不会自动重启现有服务；旧同步器进程仍按旧的 30 秒周期运行。回退应用代码不涉及数据结构回滚，已有事件仍可正常补读；回退到旧同步器后恢复周期对账。 diff --git a/samples/peripheral/store.py b/samples/peripheral/store.py index 6e02397..6722051 100644 --- a/samples/peripheral/store.py +++ b/samples/peripheral/store.py @@ -1,4 +1,4 @@ -"""PostgreSQL persistence isolated to peripheral_sample, never existing business writes.""" +"""Peripheral persistence and transactional work-order change signals.""" from contextlib import contextmanager from copy import deepcopy from datetime import datetime, timedelta, timezone @@ -306,8 +306,31 @@ class PostgresStore: conn.execute("""INSERT INTO peripheral_sample.work_order_revisions (project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""", (who.project_id, document["work_order_id"], document["revision"], Jsonb(document))) + self._work_order_event(conn, who, document) return document + def _work_order_event(self, conn, who: Identity, document: dict[str, Any], *, deleted: bool = False) -> None: + # This rollout is restricted to the provisioned tjwater_v2 event log. + # Task projections remain owned by the Server's authenticated APIs. + if self.settings.db_name != "tjwater_v2": + return + metadata = { + "source": "peripheral_sample", + "work_order_id": document["work_order_id"], + "change": "deleted" if deleted else "created" if document["revision"] == 1 else "updated", + } + row = conn.execute( + """INSERT INTO workbench.domain_event_log + (project_id,event_id,topic,aggregate_type,aggregate_id, + aggregate_revision,occurred_at,metadata) + VALUES (%s,%s,'workbench.work_order.changed','peripheral_work_order',%s,%s,%s,%s) + RETURNING event_seq""", + (who.project_id, uuid4(), document["work_order_id"], document["revision"], + document["updated_at"], Jsonb(metadata)), + ).fetchone() + conn.execute("SELECT pg_notify('workbench_events', %s)", + (f"{who.project_id}:{row['event_seq']}",)) + def list_work_orders(self, who: Identity, limit: int, cursor: UUID | None = None) -> dict[str, Any]: return self._list_managed(who, "work_orders", "work_order_id", limit, cursor) @@ -385,6 +408,7 @@ class PostgresStore: conn.execute("""INSERT INTO peripheral_sample.work_order_revisions (project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""", (who.project_id, work_order_id, revision, Jsonb(tombstone))) + self._work_order_event(conn, who, tombstone, deleted=True) return result return self._mutate(who, f"delete_work_order:{work_order_id}", command, write) diff --git a/scripts/sync_peripheral_work_orders.py b/scripts/sync_peripheral_work_orders.py index 66405eb..c5a160d 100644 --- a/scripts/sync_peripheral_work_orders.py +++ b/scripts/sync_peripheral_work_orders.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse import asyncio +import json import logging import os import sys @@ -15,6 +16,7 @@ import httpx sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from samples.peripheral.seed_identity import SEED_ACTOR_ID +from app.domain.schemas.workbench_events import WorkbenchEvent try: from scripts.run_operational_worker import CredentialProvider, WorkerSettings @@ -211,30 +213,125 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe return synchronized -async def run(once: bool): +class PeripheralWorkOrderSynchronizer: + """Wake on committed source events; periodically reconcile missed changes.""" + + def __init__(self, client: httpx.AsyncClient, settings: WorkerSettings, + peripheral_url: str, credentials: CredentialProvider) -> None: + self.client = client + self.settings = settings + self.peripheral_url = peripheral_url + self.credentials = credentials + self.wake = asyncio.Event() + self.cursor: int | None = None + + async def _reset_cursor(self, headers: dict[str, str]) -> None: + response = await self.client.get( + f"{self.settings.server_url.rstrip('/')}/api/v1/workbench/sync-snapshot", + headers=headers, + ) + response.raise_for_status() + cursor = response.json()["event_cursor"] + if not isinstance(cursor, int) or isinstance(cursor, bool) or cursor < 0: + raise ValueError("invalid workbench event cursor") + self.cursor = cursor + # Capture the cursor before scanning sources, so updates during the + # initial/reset scan are replayed by the following SSE connection. + self.wake.set() + + async def listen(self) -> None: + while True: + try: + headers = await self.credentials.headers(self.client) + if self.cursor is None: + await self._reset_cursor(headers) + async with self.client.stream( + "GET", f"{self.settings.server_url.rstrip('/')}/api/v1/workbench/events", + headers={**headers, "Accept": "text/event-stream"}, + params={"after": self.cursor}, timeout=None, + ) as response: + response.raise_for_status() + event_name = "" + async for line in response.aiter_lines(): + if line.startswith("event:"): + event_name = line[6:].strip() + elif line.startswith("data:") and event_name == "reset_required": + await self._reset_cursor(headers) + break + elif line.startswith("data:") and event_name == "workbench_event": + event = WorkbenchEvent.model_validate(json.loads(line[5:].strip())) + if str(event.project_id) != self.settings.project_id: + raise ValueError("workbench event project mismatch") + if event.event_seq <= self.cursor: + continue + self.cursor = event.event_seq + if (event.topic == "workbench.work_order.changed" + and event.aggregate_type == "peripheral_work_order" + and event.metadata.get("source") == "peripheral_sample"): + self.wake.set() + elif not line: + event_name = "" + await asyncio.sleep(1) + except (httpx.HTTPError, ValueError, KeyError): + logger.warning("Work-order event stream disconnected; reconciliation remains active") + await asyncio.sleep(5) + + async def serve(self, reconcile_seconds: int = 300) -> None: + if reconcile_seconds < 1: + raise ValueError("reconciliation interval must be at least one second") + listener = asyncio.create_task(self.listen()) + wait_seconds = reconcile_seconds + retry_seconds = 5 + try: + while True: + try: + await asyncio.wait_for(self.wake.wait(), timeout=wait_seconds) + except TimeoutError: + pass + # Clear before work, so arrivals during a scan request another + # pass. A failed scan gets a retry even after its cursor advances. + self.wake.clear() + try: + count = await sync_once( + self.client, self.settings, self.peripheral_url, self.credentials, + ) + logger.info("Peripheral progress synchronization completed: %d revisions", count) + wait_seconds, retry_seconds = reconcile_seconds, 5 + except (httpx.HTTPError, ValueError, KeyError): + logger.error("Peripheral synchronization failed; persisted revisions will be retried") + wait_seconds = retry_seconds + retry_seconds = min(retry_seconds * 2, 60) + finally: + listener.cancel() + await asyncio.gather(listener, return_exceptions=True) + + +async def run(once: bool, reconcile_seconds: int = 300) -> None: settings = WorkerSettings.from_environment() peripheral_url = os.environ["OPERATIONAL_ACTION_PERIPHERAL_URL"] credentials = CredentialProvider(settings) async with httpx.AsyncClient(timeout=30, follow_redirects=False, trust_env=False) as client: - while True: + if once: try: count = await sync_once(client, settings, peripheral_url, credentials) logger.info("Peripheral progress synchronization completed: %d revisions", count) except (httpx.HTTPError, ValueError, KeyError): - logger.error("Peripheral synchronization failed; next tick will reconcile persisted revisions") - if once: - raise SystemExit(1) from None - if once: - return - await asyncio.sleep(settings.interval_seconds) + logger.error("Peripheral synchronization failed; persisted revisions remain available") + raise SystemExit(1) from None + return + await PeripheralWorkOrderSynchronizer(client, settings, peripheral_url, credentials).serve(reconcile_seconds) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--once", action="store_true") + parser.add_argument("--reconcile-seconds", type=int, default=300, + help="missed-event reconciliation interval (default: 300 seconds)") args = parser.parse_args() + if args.reconcile_seconds < 1: + parser.error("--reconcile-seconds must be at least 1") logging.basicConfig(level=logging.INFO) try: - asyncio.run(run(args.once)) + asyncio.run(run(args.once, args.reconcile_seconds)) except KeyboardInterrupt: pass diff --git a/tests/integration/test_peripheral_sample_live.py b/tests/integration/test_peripheral_sample_live.py index d8a4876..f9e4ae8 100644 --- a/tests/integration/test_peripheral_sample_live.py +++ b/tests/integration/test_peripheral_sample_live.py @@ -1,4 +1,4 @@ -"""Opt-in checks against provisioned sample tables, never business-table writes.""" +"""Opt-in checks against provisioned sample tables and transactional source events.""" from concurrent.futures import ThreadPoolExecutor import os import subprocess @@ -39,6 +39,10 @@ def sample(): yield store, who # Only this test's unguessable project key; tables created by the authorized setup. with store.connection() as conn: + if store.settings.db_name == "tjwater_v2" and conn.execute( + "SELECT to_regclass('workbench.domain_event_log') AS name" + ).fetchone()["name"] is not None: + conn.execute("DELETE FROM workbench.domain_event_log WHERE project_id=%s", (who.project_id,)) for table in ( "work_order_revisions", "test_data_record_revisions", "work_orders", "test_data_records", "feedback", "requests", "revisions", "tasks", "scenarios", diff --git a/tests/integration/test_peripheral_work_order_events.py b/tests/integration/test_peripheral_work_order_events.py new file mode 100644 index 0000000..43b9726 --- /dev/null +++ b/tests/integration/test_peripheral_work_order_events.py @@ -0,0 +1,84 @@ +"""tjwater_v2 source/event atomicity checks; every write is rolled back.""" +from contextlib import contextmanager +import os +from uuid import uuid4 + +import pytest + +from samples.peripheral.auth import Identity +from samples.peripheral.config import SampleSettings +from samples.peripheral.errors import SampleFailure +from samples.peripheral.schemas import CreateWorkOrder, DeleteResource, UpdateWorkOrder +from samples.peripheral.store import PostgresStore + + +pytestmark = pytest.mark.skipif( + os.getenv("PERIPHERAL_SAMPLE_LIVE_TEST") != "1", reason="explicit live sample opt-in required", +) + + +def test_work_order_mutations_and_events_are_atomic_idempotent_and_rollback_cleanly(monkeypatch): + settings = SampleSettings() + assert settings.db_name == "tjwater_v2", "event verification is restricted to tjwater_v2" + store = PostgresStore(settings) + who = Identity(str(uuid4()), str(uuid4()), True) + with store.connection() as conn: + @contextmanager + def connection(*, readonly=False, timeseries=False): + assert timeseries is False + with conn.transaction(): + yield conn + + monkeypatch.setattr(store, "connection", connection) + with conn.transaction(force_rollback=True): + command = CreateWorkOrder( + request_id=uuid4(), title="AUTOTEST push work order", summary="isolated transaction", + ) + created = store.create_work_order(who, command) + assert store.create_work_order(who, command) == created + order_id = created["work_order_id"] + store.update_work_order(who, order_id, UpdateWorkOrder( + request_id=uuid4(), expected_revision=1, status="in_progress", + )) + store.update_work_order(who, order_id, UpdateWorkOrder( + request_id=uuid4(), expected_revision=2, progress_note="AUTOTEST field feedback", + )) + with pytest.raises(SampleFailure) as stale: + store.update_work_order(who, order_id, UpdateWorkOrder( + request_id=uuid4(), expected_revision=1, status="completed", + )) + assert stale.value.code == "REVISION_CONFLICT" + store.delete_work_order(who, order_id, DeleteResource( + request_id=uuid4(), expected_revision=3, + )) + rows = conn.execute( + """SELECT project_id,topic,aggregate_type,aggregate_id,aggregate_revision,metadata + FROM workbench.domain_event_log WHERE project_id=%s ORDER BY event_seq""", + (who.project_id,), + ).fetchall() + assert len(rows) == 4 + assert [row["aggregate_revision"] for row in rows] == [1, 2, 3, 4] + assert [row["metadata"]["change"] for row in rows] == ["created", "updated", "updated", "deleted"] + assert all(row["topic"] == "workbench.work_order.changed" + and row["aggregate_type"] == "peripheral_work_order" + and row["aggregate_id"] == order_id for row in rows) + assert store.list_work_orders(who, 20)["items"] == [] + + original = store._work_order_event + + def fail_after_event(*args, **kwargs): + original(*args, **kwargs) + raise RuntimeError("AUTOTEST transaction failure") + + monkeypatch.setattr(store, "_work_order_event", fail_after_event) + with pytest.raises(RuntimeError, match="transaction failure"): + store.create_work_order(who, command.model_copy(update={"request_id": uuid4()})) + assert store.list_work_orders(who, 20)["items"] == [] + assert conn.execute( + "SELECT count(*) AS n FROM workbench.domain_event_log WHERE project_id=%s", (who.project_id,), + ).fetchone()["n"] == 4 + + for table in ("workbench.domain_event_log", "peripheral_sample.work_orders", + "peripheral_sample.work_order_revisions", "peripheral_sample.requests"): + assert conn.execute(f"SELECT count(*) AS n FROM {table} WHERE project_id=%s", + (who.project_id,)).fetchone()["n"] == 0 diff --git a/tests/unit/test_peripheral_push_sync.py b/tests/unit/test_peripheral_push_sync.py new file mode 100644 index 0000000..4518b6f --- /dev/null +++ b/tests/unit/test_peripheral_push_sync.py @@ -0,0 +1,284 @@ +import asyncio +import json +from uuid import uuid4 + +import httpx +import pytest + +from scripts.run_operational_worker import CredentialProvider, WorkerSettings +from scripts import sync_peripheral_work_orders as sync_module +from scripts.sync_peripheral_work_orders import PeripheralWorkOrderSynchronizer + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +class PushStream(httpx.AsyncByteStream): + def __init__(self): + self.messages = asyncio.Queue() + self.closed = False + + async def __aiter__(self): + while True: + yield await self.messages.get() + self.messages.task_done() + + async def aclose(self): + self.closed = True + + def publish(self, project_id, order_id, sequence, *, aggregate_type="peripheral_work_order"): + event = { + "schema": "workbench-event@1", "event_seq": sequence, + "event_id": str(uuid4()), "project_id": project_id, + "topic": "workbench.work_order.changed", "aggregate_type": aggregate_type, + "aggregate_id": order_id, "aggregate_revision": 1, + "occurred_at": "2026-10-08T00:00:00Z", + "metadata": {"source": "peripheral_sample", "work_order_id": order_id}, + } + self.messages.put_nowait( + f"id: {sequence}\nevent: workbench_event\ndata: {json.dumps(event)}\n\n".encode() + ) + + +@pytest.mark.anyio +async def test_source_push_synchronizes_new_and_updated_orders_without_waiting_for_reconciliation(): + project_id, order_id, task_id = map(str, (uuid4(), uuid4(), uuid4())) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + orders, tasks, cursors = [], [], [] + initial_scan, projected = asyncio.Event(), asyncio.Event() + scan_count = 0 + + def handler(request): + nonlocal scan_count + assert request.headers["X-Project-Id"] == project_id + assert request.headers["Authorization"] == "Bearer TEST_ONLY" + if request.url.path.endswith("sync-snapshot"): + return httpx.Response(200, json={"event_cursor": 20}) + if request.url.path.endswith("/events"): + cursors.append(request.url.params["after"]) + return httpx.Response(200, headers={"Content-Type": "text/event-stream"}, stream=stream) + if request.url.host == "sample.test": + scan_count += 1 + initial_scan.set() + return httpx.Response(200, json={"items": orders, "next_cursor": None}) + if request.method == "GET": + return httpx.Response(200, json={"items": tasks, "next_offset": None}) + body = json.loads(request.content) + if request.url.path.endswith("external-feedback"): + tasks[0]["source"]["external_revision"] = body["source_revision"] + tasks[0]["status"] = body["status"] + projected.set() + return httpx.Response(200, json={"task": tasks[0]}) + task = {**body, "task_id": task_id} + tasks.append(task) + return httpx.Response(201, json=task) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + runner = asyncio.create_task(synchronizer.serve()) + try: + await asyncio.wait_for(initial_scan.wait(), timeout=3) + orders.append({ + "project_id": project_id, "work_order_id": order_id, "origin": "synthetic", + "revision": 1, "title": "预约维修", "summary": "等待现场处理", "priority": "normal", + "status": "open", "created_at": "2026-10-08T00:00:00Z", + "updated_at": "2026-10-08T00:00:00Z", "scheduled_at": None, + "work_order_type": "general", + }) + stream.publish(project_id, order_id, 21) + await asyncio.wait_for(projected.wait(), timeout=3) + assert tasks[0]["source"]["external_revision"] == 1 + projected.clear() + orders[0].update(revision=2, status="in_progress") + stream.publish(project_id, order_id, 22) + await asyncio.wait_for(projected.wait(), timeout=3) + assert tasks[0]["source"]["external_revision"] == 2 + assert tasks[0]["status"] == "in_progress" + assert scan_count == 3 # Startup plus two pushed source changes. + assert cursors == ["20"] + finally: + runner.cancel() + await asyncio.gather(runner, return_exceptions=True) + assert stream.closed + + +@pytest.mark.anyio +async def test_expired_cursor_rebuilds_baseline_and_wakes_source_reconciliation(): + project_id = str(uuid4()) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + refreshed = asyncio.Event() + baselines = iter([10, 30]) + cursors = [] + + def handler(request): + if request.url.path.endswith("sync-snapshot"): + cursor = next(baselines) + if cursor == 30: + refreshed.set() + return httpx.Response(200, json={"event_cursor": cursor}) + cursors.append(request.url.params["after"]) + return httpx.Response(200, stream=stream) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + listener = asyncio.create_task(synchronizer.listen()) + try: + await asyncio.wait_for(synchronizer.wake.wait(), timeout=3) + synchronizer.wake.clear() + stream.messages.put_nowait(b"event: reset_required\ndata: {}\n\n") + await asyncio.wait_for(refreshed.wait(), timeout=3) + assert synchronizer.wake.is_set() + assert synchronizer.cursor == 30 + assert cursors == ["10"] + finally: + listener.cancel() + await asyncio.gather(listener, return_exceptions=True) + + +@pytest.mark.anyio +async def test_projection_events_and_duplicate_source_events_do_not_trigger_extra_scans(): + project_id, order_id = str(uuid4()), str(uuid4()) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + + def handler(request): + if request.url.path.endswith("sync-snapshot"): + return httpx.Response(200, json={"event_cursor": 20}) + return httpx.Response(200, stream=stream) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + listener = asyncio.create_task(synchronizer.listen()) + try: + await asyncio.wait_for(synchronizer.wake.wait(), timeout=3) + synchronizer.wake.clear() + stream.publish(project_id, order_id, 21, aggregate_type="operational_action") + await asyncio.wait_for(stream.messages.join(), timeout=3) + assert not synchronizer.wake.is_set() + stream.publish(project_id, order_id, 22) + await asyncio.wait_for(stream.messages.join(), timeout=3) + assert synchronizer.wake.is_set() + synchronizer.wake.clear() + stream.publish(project_id, order_id, 22) + await asyncio.wait_for(stream.messages.join(), timeout=3) + assert not synchronizer.wake.is_set() + finally: + listener.cancel() + await asyncio.gather(listener, return_exceptions=True) + + +@pytest.mark.anyio +async def test_disconnect_resumes_after_last_consumed_event(): + project_id, order_id = str(uuid4()), str(uuid4()) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + stream.publish(project_id, order_id, 21) + first_message = stream.messages.get_nowait() + stream.messages.task_done() + reconnected, cursors = asyncio.Event(), [] + + def handler(request): + if request.url.path.endswith("sync-snapshot"): + return httpx.Response(200, json={"event_cursor": 20}) + cursors.append(request.url.params["after"]) + if len(cursors) == 1: + return httpx.Response(200, content=first_message) + reconnected.set() + return httpx.Response(200, stream=stream) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + listener = asyncio.create_task(synchronizer.listen()) + try: + await asyncio.wait_for(reconnected.wait(), timeout=3) + assert cursors == ["20", "21"] + finally: + listener.cancel() + await asyncio.gather(listener, return_exceptions=True) + + +@pytest.mark.anyio +async def test_failed_scan_retries_without_requiring_another_source_message(monkeypatch): + project_id = str(uuid4()) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + recovered, attempts = asyncio.Event(), 0 + + def handler(request): + if request.url.path.endswith("sync-snapshot"): + return httpx.Response(200, json={"event_cursor": 20}) + return httpx.Response(200, stream=stream) + + async def sync_once(*_args): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise httpx.ConnectError("AUTOTEST temporary source outage") + recovered.set() + return 0 + + monkeypatch.setattr(sync_module, "sync_once", sync_once) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + runner = asyncio.create_task(synchronizer.serve()) + try: + await asyncio.wait_for(recovered.wait(), timeout=8) + assert attempts == 2 + finally: + runner.cancel() + await asyncio.gather(runner, return_exceptions=True) + + +@pytest.mark.anyio +async def test_source_event_during_scan_schedules_another_pass(monkeypatch): + project_id, order_id = str(uuid4()), str(uuid4()) + settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") + stream = PushStream() + scanning, release, caught_up = asyncio.Event(), asyncio.Event(), asyncio.Event() + attempts = 0 + + def handler(request): + if request.url.path.endswith("sync-snapshot"): + return httpx.Response(200, json={"event_cursor": 20}) + return httpx.Response(200, stream=stream) + + async def sync_once(*_args): + nonlocal attempts + attempts += 1 + if attempts == 1: + scanning.set() + await release.wait() + else: + caught_up.set() + return 0 + + monkeypatch.setattr(sync_module, "sync_once", sync_once) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + synchronizer = PeripheralWorkOrderSynchronizer( + client, settings, "http://sample.test", CredentialProvider(settings), + ) + runner = asyncio.create_task(synchronizer.serve()) + try: + await asyncio.wait_for(scanning.wait(), timeout=3) + stream.publish(project_id, order_id, 21) + await asyncio.wait_for(stream.messages.join(), timeout=3) + release.set() + await asyncio.wait_for(caught_up.wait(), timeout=3) + assert attempts == 2 + finally: + runner.cancel() + await asyncio.gather(runner, return_exceptions=True) diff --git a/tests/unit/test_peripheral_work_order_events.py b/tests/unit/test_peripheral_work_order_events.py new file mode 100644 index 0000000..5ccd12e --- /dev/null +++ b/tests/unit/test_peripheral_work_order_events.py @@ -0,0 +1,50 @@ +from uuid import uuid4 + +import pytest + +from samples.peripheral.auth import Identity +from samples.peripheral.config import SampleSettings +from samples.peripheral.store import PostgresStore + + +class EventConnection: + def __init__(self): + self.statements = [] + + def execute(self, statement, parameters): + self.statements.append((statement, parameters)) + return self + + def fetchone(self): + return {"event_seq": 42} + + +@pytest.mark.parametrize(("revision", "deleted", "change"), [ + (1, False, "created"), (2, False, "updated"), (3, True, "deleted"), +]) +def test_work_order_signals_use_same_connection_and_contain_only_identity(revision, deleted, change): + conn = EventConnection() + who = Identity(str(uuid4()), str(uuid4()), True) + order = {"work_order_id": str(uuid4()), "revision": revision, + "updated_at": "2026-10-08T00:00:00Z", "title": "private title"} + + PostgresStore(SampleSettings())._work_order_event(conn, who, order, deleted=deleted) + + statement, parameters = conn.statements[0] + assert "workbench.domain_event_log" in statement + assert "'workbench.work_order.changed','peripheral_work_order'" in statement + assert parameters[0] == who.project_id + assert parameters[2:5] == (order["work_order_id"], revision, order["updated_at"]) + assert parameters[5].obj == { + "source": "peripheral_sample", "work_order_id": order["work_order_id"], "change": change, + } + assert conn.statements[1] == ( + "SELECT pg_notify('workbench_events', %s)", (f"{who.project_id}:42",), + ) + + +def test_event_rollout_does_not_write_to_other_databases(): + conn = EventConnection() + store = PostgresStore(SampleSettings(db_name="another_project")) + store._work_order_event(conn, Identity(str(uuid4()), str(uuid4()), True), {}) + assert conn.statements == []