diff --git a/app/api/v1/endpoints/operational_tasks.py b/app/api/v1/endpoints/operational_tasks.py index 68cdabb..c9d6e5a 100644 --- a/app/api/v1/endpoints/operational_tasks.py +++ b/app/api/v1/endpoints/operational_tasks.py @@ -122,12 +122,17 @@ async def list_operational_tasks( task_status: TaskStatus | None = Query(default=None, alias="status"), limit: int = Query(default=50, ge=1, le=100), offset: int = Query(default=0, ge=0), + peripheral_work_order_id: UUID | None = Query( + default=None, description="Only tasks sourced from or linked to this peripheral work order", + ), ctx: ProjectContext = Depends(require_permission(OPERATIONAL_TASK_VIEW)), conn: AsyncConnection = Depends(get_project_pg_connection), ): try: items, next_offset = await _service(ctx, conn).list( - task_type=task_type, status=task_status, limit=limit, offset=offset + task_type=task_type, status=task_status, limit=limit, offset=offset, + **({"peripheral_work_order_id": peripheral_work_order_id} + if peripheral_work_order_id is not None else {}), ) return OperationalTaskPage( items=items, limit=limit, offset=offset, next_offset=next_offset diff --git a/app/api/v1/endpoints/workbench_events.py b/app/api/v1/endpoints/workbench_events.py index 23755f4..71df44d 100644 --- a/app/api/v1/endpoints/workbench_events.py +++ b/app/api/v1/endpoints/workbench_events.py @@ -131,9 +131,9 @@ async def workbench_events( if now - last_keepalive >= 15: yield f": keepalive {int(now)}\n\n" last_keepalive = now - # LISTEN is only a wake-up path. The timeout also reconciles - # against the durable cursor if a notification was missed. - wake_generation = await hub.wait_for_wake(wake_generation, timeout=1) + # The Hub invalidates shared reads on NOTIFY and every 30s. + # Per-client timeout only keeps the transport alive, not SQL. + wake_generation = await hub.wait_for_wake(wake_generation, timeout=15) yield "event: reconnect\ndata: {}\n\n" return StreamingResponse( diff --git a/app/infra/db/postgresql/analysis.py b/app/infra/db/postgresql/analysis.py index 99be323..6268df5 100644 --- a/app/infra/db/postgresql/analysis.py +++ b/app/infra/db/postgresql/analysis.py @@ -4,6 +4,7 @@ from uuid import UUID, uuid4 from psycopg import AsyncConnection, Connection from psycopg.types.json import Jsonb +from app.infra.db.postgresql.workbench_events import lock_event_transaction_sync class AnalysisRepository: @@ -19,6 +20,7 @@ class AnalysisRepository: parameters: dict[str, Any], run_id: UUID | None = None, ) -> dict[str, Any]: + lock_event_transaction_sync(conn) execution_id = run_id or uuid4() with conn.cursor() as cur: cur.execute( @@ -54,6 +56,7 @@ class AnalysisRepository: created_by: str, parameters: dict[str, Any], ) -> int: + lock_event_transaction_sync(conn) with conn.cursor() as cur: cur.execute( """ diff --git a/app/infra/db/postgresql/operational_tasks.py b/app/infra/db/postgresql/operational_tasks.py index d2d0548..d274e3c 100644 --- a/app/infra/db/postgresql/operational_tasks.py +++ b/app/infra/db/postgresql/operational_tasks.py @@ -10,6 +10,7 @@ from psycopg.types.json import Jsonb from uuid import uuid4 from app.domain.domain_errors import DomainError +from app.infra.db.postgresql.workbench_events import lock_event_transaction class OperationalTaskRepository: @@ -21,6 +22,7 @@ class OperationalTaskRepository: async def transaction(self): try: async with self.connection.transaction(): + await lock_event_transaction(self.connection) yield self except UndefinedTable as exc: raise DomainError( @@ -239,6 +241,7 @@ class OperationalTaskRepository: status: str | None, limit: int, offset: int, + peripheral_work_order_id: str | None = None, ) -> list[dict]: cursor = await self.connection.execute( """SELECT task.document, @@ -255,6 +258,16 @@ class OperationalTaskRepository: AND (task.document#>>'{source,deleted_at}') IS NULL AND (%s::text IS NULL OR task.task_type=%s::text) AND (%s::text IS NULL OR task.status=%s::text) + AND (%s::text IS NULL + OR (task.document#>>'{source,kind}'='external_system' + AND task.document#>>'{source,external_id}' IN (%s,%s)) + OR EXISTS ( + SELECT 1 FROM workbench.operational_task_actions AS action + WHERE action.project_id=task.project_id AND action.task_id=task.task_id + AND action.document->>'kind'='create_work_order' + AND action.document#>>'{receipt,adapter}'='peripheral_sample' + AND action.document#>>'{receipt,receipt_id}'=%s + )) ORDER BY task.updated_at DESC,task.task_id LIMIT %s OFFSET %s""", ( @@ -263,6 +276,10 @@ class OperationalTaskRepository: task_type, status, status, + peripheral_work_order_id, + peripheral_work_order_id, + f"peripheral_sample:{peripheral_work_order_id}" if peripheral_work_order_id else None, + peripheral_work_order_id, limit, offset, ), diff --git a/app/infra/db/postgresql/task_alerts.py b/app/infra/db/postgresql/task_alerts.py index 90ecbf2..23603af 100644 --- a/app/infra/db/postgresql/task_alerts.py +++ b/app/infra/db/postgresql/task_alerts.py @@ -9,6 +9,7 @@ from psycopg.types.json import Jsonb from uuid import uuid4 from app.domain.domain_errors import DomainError +from app.infra.db.postgresql.workbench_events import lock_event_transaction class TaskAlertRepository: @@ -20,6 +21,7 @@ class TaskAlertRepository: async def transaction(self): try: async with self.connection.transaction(): + await lock_event_transaction(self.connection) yield self except UndefinedTable as exc: raise DomainError( diff --git a/app/infra/db/postgresql/workbench_event_hub.py b/app/infra/db/postgresql/workbench_event_hub.py index 152159b..33b35aa 100644 --- a/app/infra/db/postgresql/workbench_event_hub.py +++ b/app/infra/db/postgresql/workbench_event_hub.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import contextlib import logging +from collections import OrderedDict from contextlib import asynccontextmanager from uuid import UUID @@ -27,6 +28,8 @@ class WorkbenchEventHub: self._connection: AsyncConnection | None = None self._pool: AsyncConnectionPool | None = None self._connection_lock = asyncio.Lock() + self._read_lock = asyncio.Lock() + self._reads: OrderedDict[tuple, tuple[int, object]] = OrderedDict() self._condition = asyncio.Condition() self._generation = 0 self._ready = asyncio.Event() @@ -59,6 +62,7 @@ class WorkbenchEventHub: async def _listen_loop(self) -> None: retry_seconds = 1 + next_reconcile = asyncio.get_running_loop().time() + 30 while not self._closed: try: if self._connection is None or self._connection.closed: @@ -76,8 +80,11 @@ class WorkbenchEventHub: if notification.payload.startswith(f"{self.project_id}:"): notified = True break - if notified: + now = asyncio.get_running_loop().time() + if notified or now >= next_reconcile: await self._wake() + if now >= next_reconcile: + next_reconcile = now + 30 await asyncio.sleep(0) except asyncio.CancelledError: raise @@ -109,22 +116,35 @@ class WorkbenchEventHub: return self._generation async def oldest_cursor(self) -> int | None: - return await self._read(lambda repository: repository.oldest_cursor()) + return await self._read(("oldest",), lambda repository: repository.oldest_cursor()) async def current_cursor(self) -> int: - return await self._read(lambda repository: repository.current_cursor()) + return await self._read(("current",), lambda repository: repository.current_cursor()) async def read_after(self, after: int) -> list[dict]: - return await self._read(lambda repository: repository.read_after(after)) + return await self._read(("after", after), lambda repository: repository.read_after(after)) - async def _read(self, operation): + async def _read(self, key: tuple, operation): await self.start() - async with self._connection_lock: - if self._connection is None or self._connection.closed: - self._connection = await self._connect() - self._ready.set() - repository = WorkbenchEventRepository(self._connection, self.project_id) - return await operation(repository) + async with self._read_lock: + cached = self._reads.get(key) + if cached is not None and cached[0] == self.generation: + self._reads.move_to_end(key) + return cached[1] + # Record the generation before the query: a concurrent NOTIFY must + # invalidate even an empty result from an older database snapshot. + generation = self.generation + async with self._connection_lock: + if self._connection is None or self._connection.closed: + self._connection = await self._connect() + self._ready.set() + repository = WorkbenchEventRepository(self._connection, self.project_id) + result = await operation(repository) + self._reads[key] = (generation, result) + self._reads.move_to_end(key) + while len(self._reads) > 32: + self._reads.popitem(last=False) + return result async def _drop_connection(self) -> None: async with self._connection_lock: @@ -135,6 +155,7 @@ class WorkbenchEventHub: async def close(self) -> None: self._closed = True + self._reads.clear() if self._idle_task is not None: self._idle_task.cancel() self._idle_task = None diff --git a/app/infra/db/postgresql/workbench_events.py b/app/infra/db/postgresql/workbench_events.py index b7337e6..c873258 100644 --- a/app/infra/db/postgresql/workbench_events.py +++ b/app/infra/db/postgresql/workbench_events.py @@ -1,7 +1,31 @@ """Cursor reads for the durable project-scoped workbench event log.""" from __future__ import annotations -from psycopg import AsyncConnection +from psycopg import AsyncConnection, Connection + + +EVENT_ORDER_LOCK = "tjwater-workbench:event-order" + + +async def lock_event_transaction(connection: AsyncConnection) -> None: + """Serialize v2 event-producing transactions before business locks/writes. + + PostgreSQL sequence allocation is not commit order. All producers must use + this same transaction lock so a committed higher cursor cannot hide an + earlier, still-uncommitted event. Other BizDBs remain rollout opt-out. + """ + if getattr(getattr(connection, "info", None), "dbname", None) == "tjwater_v2": + await connection.execute( + "SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,) + ) + + +def lock_event_transaction_sync(connection: Connection) -> None: + """Synchronous counterpart using the identical PostgreSQL lock key.""" + if getattr(getattr(connection, "info", None), "dbname", None) == "tjwater_v2": + connection.execute( + "SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,) + ) class WorkbenchEventRepository: diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 62a975a..48344c2 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -200,10 +200,13 @@ class OperationalTaskService: return await self.detail(task_id) async def list( - self, *, task_type: str | None, status: str | None, limit: int, offset: int + self, *, task_type: str | None, status: str | None, limit: int, offset: int, + peripheral_work_order_id: UUID | None = None, ) -> tuple[list[OperationalTaskSummary], int | None]: rows = await self.repo.list_tasks( - task_type=task_type, status=status, limit=limit + 1, offset=offset + task_type=task_type, status=status, limit=limit + 1, offset=offset, + **({"peripheral_work_order_id": str(peripheral_work_order_id)} + if peripheral_work_order_id is not None else {}), ) items: list[OperationalTaskSummary] = [] for row in rows[:limit]: @@ -922,9 +925,13 @@ class OperationalTaskService: require(OPERATIONAL_ACTION_WORK_ORDER_APPROVE in self.permissions and OPERATIONAL_ACTION_WORK_ORDER_EXECUTE in self.permissions, "PERMISSION_DENIED", "Current user cannot confirm and dispatch work orders", status=403) - current = await self._action(task_id, action_id) - require(current.kind == "create_work_order", "INVALID_ACTION_KIND", - "Only work orders support one-step confirmation", status=422) + # A SELECT on the default non-autocommit connection starts an outer + # transaction. End it explicitly, or approval/submit become savepoints + # whose event lock survives into the peripheral HTTP call. + async with self.repo.transaction(): + current = await self._action(task_id, action_id) + require(current.kind == "create_work_order", "INVALID_ACTION_KIND", + "Only work orders support one-step confirmation", status=422) approval = await self.decide_action(task_id, action_id, ActionDecisionCommand( request_id=uuid5(NAMESPACE_URL, f"confirm:{command.request_id}:approve"), expected_version=command.expected_version, decision="approved", @@ -1024,7 +1031,8 @@ class OperationalTaskService: scope, command.request_id, body, {"action": _json(action)} ) - task = await self.detail(task_id) + async with self.repo.transaction(): + task = await self.detail(task_id) try: outcome = await dispatcher.dispatch( context=self.context, task=task, action=action, credential=credential @@ -1099,19 +1107,21 @@ class OperationalTaskService: ) -> HumanApprovalAction: body = _json(command) scope = f"task:{task_id}:action:{action_id}:reconcile" - replay = await self.repo.get_request(scope, str(command.request_id)) - if replay: - self._require_same_request(replay, body) - return HumanApprovalAction.model_validate(replay["response"]["action"]) - action = await self._action(task_id, action_id) - require(action.version == command.expected_version, "VERSION_CONFLICT", - "Action version changed", status=409) - require(action.execution_status in {"submitted", "unknown"}, "INVALID_TRANSITION", - "Only submitted or unknown actions can be reconciled", status=409) - require(action.required_execution_permission in self.permissions, - "PERMISSION_DENIED", "Current user cannot reconcile this action", status=403) + async with self.repo.transaction(): + replay = await self.repo.get_request(scope, str(command.request_id)) + if replay: + self._require_same_request(replay, body) + return HumanApprovalAction.model_validate(replay["response"]["action"]) + action = await self._action(task_id, action_id) + require(action.version == command.expected_version, "VERSION_CONFLICT", + "Action version changed", status=409) + require(action.execution_status in {"submitted", "unknown"}, "INVALID_TRANSITION", + "Only submitted or unknown actions can be reconciled", status=409) + require(action.required_execution_permission in self.permissions, + "PERMISSION_DENIED", "Current user cannot reconcile this action", status=403) + task = await self.detail(task_id) outcome = await dispatcher.reconcile( - context=self.context, task=await self.detail(task_id), action=action, + context=self.context, task=task, action=action, credential=credential, ) self._validate_dispatch_outcome(outcome) diff --git a/contracts/manifest.json b/contracts/manifest.json index 948ca65..b956256 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "f684880cf2c17f9643899bbf52af31e877e127cd9a35f73bfcd00a2263bec835" + "sha256": "0669dd2bec261c6ab9320df5eb506b85945829cb5582504483f89b482d826631" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 83e2a80..7928ebe 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -23716,6 +23716,25 @@ "type": "integer" } }, + { + "description": "Only tasks sourced from or linked to this peripheral work order", + "in": "query", + "name": "peripheral_work_order_id", + "required": false, + "schema": { + "anyOf": [ + { + "format": "uuid", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Only tasks sourced from or linked to this peripheral work order", + "title": "Peripheral Work Order Id" + } + }, { "in": "header", "name": "X-Project-Id", diff --git a/docs/peripheral-work-order-events.md b/docs/peripheral-work-order-events.md index 52aa9d1..c2ded1b 100644 --- a/docs/peripheral-work-order-events.md +++ b/docs/peripheral-work-order-events.md @@ -26,12 +26,18 @@ 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` 源事件唤醒，不被自己写出的任务投影事件再次唤醒。 +继续复用 `OPERATIONAL_WORKER_*` OAuth 工作身份及 `OPERATIONAL_ACTION_PERIPHERAL_URL`，不需要新账号或密钥。启动和游标过期时先读取事件游标，再进行全量对账；扫描期间的事件从该游标继续补读。相同源版本仍由原幂等键去重。源事件按工单 ID 唤醒同步器；动作回执事件只补查该任务关联的工单，避免源创建早于回执保存时漏同步。任务投影事件不再触发全量扫描。 -正常状态收到源事件即扫描；无事件时默认每 300 秒对账一次，处理漏通知、断线、旧批量脚本直接写库及重启。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前清除唤醒标记，扫描期间新到达的事件会触发下一轮，失败后的重试也不依赖新消息。 +正常状态收到源事件即处理对应 ID：`GET /api/v1/operational-tasks?peripheral_work_order_id=` 只返回同项目的来源任务和动作回执关联任务；外围接口只读取该工单，不再分页遍历全部工单和全部调查详情。重复 ID 合并；目标工单的明确删除只撤回该工单投影，不影响其他任务。 + +内存待处理 ID 最多 200 个，超过上限合并为一次全量对账，不持续扩张队列。这是突发积压保护，不增加公共配置项。 + +默认每 300 秒仍进行一次全量对账，处理漏通知、断线、旧批量脚本直接写库及重启。持续收到事件不会推迟全量对账期限。可用 `--reconcile-seconds` 调整这一兜底周期（最小 1 秒），不再复用工况 worker 的 30 秒租约检查周期。同步失败按 5、10、20、40、60 秒退避重试；事件连接断开每 5 秒重试，保留最后游标。扫描开始前取出待处理 ID，扫描期间新到达的 ID 留给下一轮；失败的 ID 会放回队列，重试不依赖新消息。 浏览器收到源事件时不会提前刷新业务快照，而是等待任务投影事件后读取对应任务。工况手动刷新仍只读 Server 权威快照。工单事件不等于 Header 通知；只有需要处理的业务待办或告警才进入通知列表。 +游标顺序、共享 Hub 查询与兼容边界见 [事件链路收敛](workbench-event-refinement.md)。新增查询参数已从 Server OpenAPI 同步生成 Agent 类型，不另建手写契约。 + ## 验证与启用 ```bash diff --git a/docs/workbench-database-rollout.md b/docs/workbench-database-rollout.md index 9b74e85..13b0225 100644 --- a/docs/workbench-database-rollout.md +++ b/docs/workbench-database-rollout.md @@ -71,4 +71,6 @@ SQL 文件包含自己的 `BEGIN` 和 `COMMIT` 时，不能在外层简单包一 ## 失败与回滚 +2026-10-08 的 [事件链路收敛](workbench-event-refinement.md) 没有 DDL 或持久数据迁移。新增事务排序保护按实际连接的物理数据库名仅启用 `tjwater_v2`；并发及源事件验证使用强制回滚事务。其他数据库未被写入或迁移，扩展仍需明确确认。 + 任何数据库出现迁移失败、接口 5xx、跨项目记录、Run 代次错乱或游标异常时，立即停止后续项目。未提交的事务直接回滚；已经提交且不能安全前向修复时，按该项目的迁移前备份恢复。不要为保持各项目版本一致而继续扩大故障范围。 diff --git a/docs/workbench-event-refinement.md b/docs/workbench-event-refinement.md new file mode 100644 index 0000000..7aa180b --- /dev/null +++ b/docs/workbench-event-refinement.md @@ -0,0 +1,34 @@ +# 工作台事件链路收敛（2026-10-08） + +本次保留事务事件日志、任务/Run 分离、通知收件箱和低频对账，不增加表、字段、迁移、账号、消息中继或服务。数据库验证和新增事件排序保护只作用于物理数据库 `tjwater_v2`；未向其他 BizDB 应用变更。 + +## 修复与成本边界 + +- 自增序号不等于提交顺序。任务、告警、分析和外围工单写入在业务锁和写入之前取得同一个 PostgreSQL 事务 advisory lock；后续写者必须等前一个提交或回滚后再分配事件序号。锁范围是单个 BizDB，数据库自动在事务结束时释放，不需要新表。事务回滚仍可能留下序号空洞，这是正常情况，不能要求游标连续。 +- 所有事件生产者必须遵守同一锁约定；旧进程和绕开约定的直接 SQL 写入不受保护。激活时必须一起更新 Server 和外围源服务；混合版本不能声称提交顺序已保证。分析计算、Agent 执行和外部 HTTP 调用不得放进持锁事务。确认派发的前置查询、派发前详情读取和回执核对读取均使用显式短事务，避免默认非 autocommit 连接的隐式事务使写阶段退化为 savepoint、锁一直跨 HTTP 存活。不改其他业务模块的连接池配置。 +- 这个锁会串行化事件写事务，任务仓储的短命令事务（包括心跳）也使用它。这是当前低频业务规模下的明确取舍，不适用于高频遥测或长事务。扩大业务事件量前应实际测量写入等待时间，不把增加锁误称为写吞吐优化。 +- 每个进程/项目仍只有一条专用 LISTEN 连接，不长期占用普通请求池。Hub 按唤醒代次共享同游标读取；最多保留 32 个游标批次，每批最多 200 条，慢客户端不会形成无界缓存。事件通知和每 30 秒一次的共享对账使缓存失效；每个浏览器的 15 秒 keepalive 不再触发独立 SQL 查询。无订阅者 60 秒后仍关闭 Hub。 +- `reset_required` 允许权威快照把客户端游标从较大值重置为 0；普通快照仍禁止回退，过时并发响应仍不能覆盖新响应。重置失败会在重连时重试。只有 SSE 请求成功打开后才显示连接正常，HTTP 失败按原退避策略重连；离开或重置时释放流读取器。 +- 工单源事件按 ID 定向同步。任务投影的 `operational.task.changed` 负责更新详情，其伴随的 `workbench.work_order.changed/operational_task` 不再重复拉取整份快照。告警和分析事件仍合并后刷新运行态；SCADA/流量采样通道不变。全量对账见 [外围工单主动同步](peripheral-work-order-events.md)。 +- 项目查询按同一身份提供器共享，缓存 5 分钟。每个调用仍重新获取 token，各调用可独立取消；共享查询有 15 秒超时，失败后清除缓存。此缓存不授予权限，Server 每次请求和每次 SSE 重建仍鉴权。 +- 事件接收游标与数据读取成功分开处理：增量详情、通知或事件触发的运行态快照失败时，只对失败读取按 1、2、4 秒递增至 30 秒重试，成功后才确认失效事件。被并发游标守卫拒绝的快照不能当作刷新成功；仅在更新的成功快照已覆盖失效游标时停止重试。正常读取成功后不安排定时请求；依赖变化、重置或离开页面会取消旧请求及重试定时器。这不是新增正常轮询通道，也不扩大初始化列表的裁剪/重试语义。 + +20 个相同游标订阅者的单元用例中，数据库读取从 20 次降为 1 次；这是查询次数验证，不是生产 CPU、连接预算或服务器费用压测结论。 + +## 保留但不扩展的兼容项 + +任务表的旧执行字段、旧 `status` 和 JSON/标量重复投影先保留；新执行流程以 Run 为准。空的 schedule 表/API 仍不接入前端 mock 定时任务。未经前向迁移、备份及调用方验收，不删除这些兼容结构。现有表的数量并不是本次优化目标；应减少重复计算、扫描和通信，而不是合并独立生命周期的业务对象。 + +## 验证与启用 + +```bash +conda run -n server python -m pytest tests/unit tests/auth -q +WORKBENCH_EVENT_ORDER_TEST=1 PERIPHERAL_SAMPLE_LIVE_TEST=1 conda run -n server python -m pytest tests/integration/test_workbench_event_order.py tests/integration/test_peripheral_work_order_events.py -q +conda run -n server python scripts/check_openapi.py +``` + +并发测试强制连接 `tjwater_v2`，观察后写者等待事务锁；所有随机项目事件均回滚。测试会消耗序号，但不会保留测试业务记录或事件，也不变更 schema。外围原子性用例验证源记录、历史及事件同事务回滚。前端浏览器用例验证从 41 重置到 0 后重连，以及源事件/投影事件只触发一次详情读取。 + +默认非 autocommit 连接的三条动作回归（确认派发、直接派发、核对回执）运行真实读取/事务锁和实际服务编排，以内存动作夹具替代业务写入。外部适配器替身检查调用时连接为 IDLE，并用第二条真实连接取得同一事件锁，确认没有跨调用持锁；这不是实际外围 HTTP 或设备执行联调。 + +代码提交不代表部署。本轮不重启现有进程，不运行全库迁移，不推送。启用新同步器需要先上线含新增查询参数的 Server，再按维护者授权的发布流程更新外围服务和同步器，最后更新前端。回退旧代码可恢复低频全量对账，但也会重新暴露旧游标排序问题，不能作为已解决并发问题的版本。 diff --git a/samples/peripheral/store.py b/samples/peripheral/store.py index 6722051..9328a04 100644 --- a/samples/peripheral/store.py +++ b/samples/peripheral/store.py @@ -12,6 +12,7 @@ from psycopg.types.json import Jsonb from app.domain.schemas.shared import TimeRange from app.services.content_digest import digest +from app.infra.db.postgresql.workbench_events import lock_event_transaction_sync from .errors import SampleFailure, ensure from .auth import Identity @@ -113,6 +114,8 @@ class PostgresStore: lost = False try: with self.connection() as conn: + if "work_order" in key: + lock_event_transaction_sync(conn) conn.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", ("peripheral-sample:" + who.project_id,)) cached, request_digest = self._cached(conn, who, key, command) diff --git a/scripts/sync_peripheral_work_orders.py b/scripts/sync_peripheral_work_orders.py index c5a160d..dcdaa20 100644 --- a/scripts/sync_peripheral_work_orders.py +++ b/scripts/sync_peripheral_work_orders.py @@ -28,6 +28,7 @@ logger = logging.getLogger("peripheral-task-sync") STATUS = {"draft": "scheduled", "open": "assigned", "in_progress": "in_progress", "awaiting_reply": "awaiting_feedback", "completed": "completed", "cancelled": "cancelled"} EVENT_WORK_ORDER_TYPES = frozenset({"water_quality_complaint", "burst_report"}) +MAX_TARGETED_ORDERS = 200 def peripheral_order_id(external_id: str | None) -> str | None: @@ -69,31 +70,66 @@ def work_order_snapshot(order: dict) -> dict: return snapshot +async def work_order_pages(client, url, headers, work_order_ids, deleted_ids): + if work_order_ids is not None: + for order_id in sorted(work_order_ids): + response = await client.get(f"{url}/{order_id}", headers=headers) + if response.status_code == 404: + if response.json().get("code") != "WORK_ORDER_NOT_FOUND": + raise ValueError("peripheral work order lookup did not confirm deletion") + deleted_ids.add(order_id) + continue + response.raise_for_status() + order = response.json() + if order["work_order_id"] != order_id: + raise ValueError("peripheral work order identity mismatch") + yield {"items": [order]} + return + cursor = None + while True: + params = {"limit": 100} + if cursor: + params["cursor"] = cursor + response = await client.get(url, headers=headers, params=params) + response.raise_for_status() + page = response.json() + yield page + cursor = page.get("next_cursor") + if not cursor: + return + + async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, peripheral_url: str, - credentials: CredentialProvider) -> int: + credentials: CredentialProvider, *, work_order_ids: set[str] | None = None) -> int: + if work_order_ids is not None: + work_order_ids = {str(UUID(order_id)) for order_id in work_order_ids} headers = await credentials.headers(client) tasks_url = f"{settings.server_url.rstrip('/')}/api/v1/operational-tasks" existing: dict[str, dict] = {} event_tasks: dict[str | None, dict] = {} - investigations: list[dict] = [] - offset = 0 - while True: - response = await client.get(tasks_url, headers=headers, params={"limit": 100, "offset": offset}) - response.raise_for_status() - page = response.json() - for task in page["items"]: - if task["task_type"] == "external_work_order": - existing[task["source"].get("external_id")] = task - elif task["task_type"] == "event_investigation": - event_tasks[task.get("source", {}).get("external_id")] = task - if task["status"] in {"queued", "waiting_human", "running"}: - investigations.append(task) - if page.get("next_offset") is None: - break - offset = page["next_offset"] + investigations: dict[str, dict] = {} + for order_id in sorted(work_order_ids) if work_order_ids is not None else [None]: + offset = 0 + while True: + params = {"limit": 100, "offset": offset} + if order_id is not None: + params["peripheral_work_order_id"] = order_id + response = await client.get(tasks_url, headers=headers, params=params) + response.raise_for_status() + page = response.json() + for task in page["items"]: + if task["task_type"] == "external_work_order": + existing[task["source"].get("external_id")] = task + elif task["task_type"] == "event_investigation": + event_tasks[task.get("source", {}).get("external_id")] = task + if task["status"] in {"queued", "waiting_human", "running"}: + investigations[task["task_id"]] = task + if page.get("next_offset") is None: + break + offset = page["next_offset"] linked: dict[str, list[str]] = {} - for task in investigations: + for task in investigations.values(): response = await client.get(f"{tasks_url}/{task['task_id']}", headers=headers) response.raise_for_status() for action in response.json().get("actions", []): @@ -102,16 +138,13 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe linked.setdefault(receipt["receipt_id"], []).append(task["task_id"]) linked_order_ids = set(linked) - cursor, synchronized = None, 0 + synchronized = 0 + deleted_ids: set[str] = set() seen_order_ids: set[str] = set() - while True: - params = {"limit": 100} - if cursor: - params["cursor"] = cursor - response = await client.get(f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders", - headers=headers, params=params) - response.raise_for_status() - page = response.json() + async for page in work_order_pages( + client, f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders", headers, + work_order_ids, deleted_ids, + ): for order in page["items"]: if order["project_id"] != settings.project_id or order["origin"] != "synthetic": raise ValueError("peripheral work order scope or origin mismatch") @@ -178,9 +211,6 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe result = await client.post(f"{tasks_url}/{target}/external-feedback", headers=headers, json=payload) result.raise_for_status() synchronized += 1 - cursor = page.get("next_cursor") - if not cursor: - break for external_id, task in existing.items(): order_id = peripheral_order_id(external_id) @@ -188,18 +218,21 @@ async def sync_once(client: httpx.AsyncClient, settings: WorkerSettings, periphe continue if order_id in seen_order_ids: continue + if work_order_ids is not None and order_id not in work_order_ids: + continue # The list may change during pagination. Only a scoped 404 proves that # the source order is now absent; transient errors must stop this tick. - response = await client.get( - f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders/{order_id}", - headers=headers, - ) - if response.status_code == 200: - continue - if response.status_code != 404: - response.raise_for_status() - if response.json().get("code") != "WORK_ORDER_NOT_FOUND": - raise ValueError("peripheral work order lookup did not confirm deletion") + if order_id not in deleted_ids: + response = await client.get( + f"{peripheral_url.rstrip('/')}/api/sample/v1/work-orders/{order_id}", + headers=headers, + ) + if response.status_code == 200: + continue + if response.status_code != 404: + response.raise_for_status() + if response.json().get("code") != "WORK_ORDER_NOT_FOUND": + raise ValueError("peripheral work order lookup did not confirm deletion") result = await client.post( f"{tasks_url}/{task['task_id']}/withdraw-external-source", headers=headers, @@ -224,6 +257,16 @@ class PeripheralWorkOrderSynchronizer: self.credentials = credentials self.wake = asyncio.Event() self.cursor: int | None = None + self.pending_order_ids: set[str] = set() + self.full_reconcile = False + + def _queue_order(self, order_id: str) -> None: + if not self.full_reconcile: + self.pending_order_ids.add(order_id) + if len(self.pending_order_ids) > MAX_TARGETED_ORDERS: + self.pending_order_ids.clear() + self.full_reconcile = True + self.wake.set() async def _reset_cursor(self, headers: dict[str, str]) -> None: response = await self.client.get( @@ -235,6 +278,7 @@ class PeripheralWorkOrderSynchronizer: if not isinstance(cursor, int) or isinstance(cursor, bool) or cursor < 0: raise ValueError("invalid workbench event cursor") self.cursor = cursor + self.full_reconcile = True # Capture the cursor before scanning sources, so updates during the # initial/reset scan are replayed by the following SSE connection. self.wake.set() @@ -264,11 +308,25 @@ class PeripheralWorkOrderSynchronizer: 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() + self._queue_order(str(UUID(event.aggregate_id))) + elif event.topic == "operational.action.changed": + # The source create can commit before its action + # receipt is saved. Revisit just those receipt IDs. + task_id = str(UUID(event.metadata["task_id"])) + task = await self.client.get( + f"{self.settings.server_url.rstrip('/')}/api/v1/operational-tasks/{task_id}", + headers=headers, + ) + task.raise_for_status() + for action in task.json().get("actions", []): + receipt = action.get("receipt") + if (action["kind"] == "create_work_order" and receipt + and receipt["adapter"] == "peripheral_sample"): + self._queue_order(str(UUID(receipt["receipt_id"]))) + self.cursor = event.event_seq elif not line: event_name = "" await asyncio.sleep(1) @@ -280,26 +338,37 @@ class PeripheralWorkOrderSynchronizer: 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 + next_reconcile = asyncio.get_running_loop().time() + reconcile_seconds try: while True: try: - await asyncio.wait_for(self.wake.wait(), timeout=wait_seconds) + remaining = max(0, next_reconcile - asyncio.get_running_loop().time()) + await asyncio.wait_for(self.wake.wait(), timeout=remaining) except TimeoutError: - pass + self.full_reconcile = True # 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() + full_reconcile = self.full_reconcile + self.full_reconcile = False + order_ids, self.pending_order_ids = self.pending_order_ids, set() try: count = await sync_once( self.client, self.settings, self.peripheral_url, self.credentials, + **({"work_order_ids": order_ids} if not full_reconcile else {}), ) logger.info("Peripheral progress synchronization completed: %d revisions", count) - wait_seconds, retry_seconds = reconcile_seconds, 5 + retry_seconds = 5 + if full_reconcile: + next_reconcile = asyncio.get_running_loop().time() + reconcile_seconds except (httpx.HTTPError, ValueError, KeyError): logger.error("Peripheral synchronization failed; persisted revisions will be retried") - wait_seconds = retry_seconds + self.full_reconcile |= full_reconcile + for order_id in order_ids: + self._queue_order(order_id) + await asyncio.sleep(retry_seconds) + self.wake.set() retry_seconds = min(retry_seconds * 2, 60) finally: listener.cancel() diff --git a/tests/integration/test_workbench_event_order.py b/tests/integration/test_workbench_event_order.py new file mode 100644 index 0000000..7f83568 --- /dev/null +++ b/tests/integration/test_workbench_event_order.py @@ -0,0 +1,232 @@ +"""Opt-in v2 concurrent writer regression; every event is rolled back.""" +import asyncio +import os +from datetime import datetime, timezone +from uuid import uuid4 +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import psycopg +import pytest +from psycopg.conninfo import make_conninfo +from psycopg.rows import dict_row +from psycopg.pq import TransactionStatus + +from app.core.config import get_pgconn_string +from app.auth.permissions import OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE +from app.auth.project_dependencies import ProjectContext +from app.domain.schemas.operational_task import ( + ActionExecuteCommand, ActionReconcileCommand, ActionReceipt, HumanApprovalAction, + WorkOrderConfirmDispatchCommand, + OperationalTaskCreate, +) +from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository +from app.infra.db.postgresql.task_alerts import TaskAlertRepository +from app.infra.db.postgresql.workbench_events import lock_event_transaction +from app.infra.integrations.operational_actions import ActionDispatchOutcome +from app.services.operational_tasks import OperationalTaskService + + +pytestmark = pytest.mark.skipif( + os.getenv("WORKBENCH_EVENT_ORDER_TEST") != "1", reason="explicit v2 concurrency opt-in required", +) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.anyio +async def test_later_producer_cannot_allocate_a_cursor_before_first_transaction_finishes(): + dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2") + first_ready, release_first, second_ready = asyncio.Event(), asyncio.Event(), asyncio.Event() + project_id = uuid4() + async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as first: + async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as second: + async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as observer: + assert first.info.dbname == second.info.dbname == observer.info.dbname == "tjwater_v2" + + async def write_first(): + async with first.transaction(force_rollback=True): + repo = OperationalTaskRepository(first, project_id) + async with repo.transaction(): + await repo.append_event( + topic="operational.task.changed", aggregate_type="operational_task", + aggregate_id=str(uuid4()), aggregate_revision=1, + occurred_at=datetime.now(timezone.utc), metadata={}, + ) + first_ready.set() + await release_first.wait() + + async def write_second(): + async with second.transaction(force_rollback=True): + async with TaskAlertRepository(second, project_id).transaction(): + second_ready.set() + + first_task = asyncio.create_task(write_first()) + second_task = None + try: + await asyncio.wait_for(first_ready.wait(), 3) + second_task = asyncio.create_task(write_second()) + async with asyncio.timeout(3): + while True: + row = await (await observer.execute( + "SELECT wait_event_type FROM pg_stat_activity WHERE pid=%s", + (second.info.backend_pid,), + )).fetchone() + if row and row["wait_event_type"] == "Lock": + break + assert not second_ready.is_set(), "later producer bypassed the event ordering lock" + await asyncio.sleep(0.01) + assert not second_ready.is_set() + release_first.set() + await asyncio.wait_for(asyncio.gather(first_task, second_task), 3) + assert second_ready.is_set() + row = await (await observer.execute( + "SELECT count(*) AS n FROM workbench.domain_event_log WHERE project_id=%s", + (project_id,), + )).fetchone() + assert row["n"] == 0 + finally: + release_first.set() + pending = [task for task in (first_task, second_task) if task is not None] + for task in pending: + if not task.done(): + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + +@pytest.mark.anyio +@pytest.mark.parametrize("operation", ["confirm", "submit", "reconcile"]) +async def test_default_connection_ends_transactions_before_external_dispatch(monkeypatch, operation): + """Real psycopg reads/locks and service orchestration, without fixture writes.""" + dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2") + task_id, action_id, order_id = uuid4(), uuid4(), uuid4() + now = datetime.now(timezone.utc) + context = ProjectContext(project_id=uuid4(), project_code="event-order-test", + user_id=uuid4(), project_role="member") + action = HumanApprovalAction( + action_id=action_id, task_id=task_id, version=1, kind="create_work_order", + reason="AUTOTEST transaction boundary", evidence_refs=[], + payload={"title": "AUTOTEST", "description": "not persisted", "work_order_type": "general"}, + required_approval_permission=OPERATIONAL_ACTION_WORK_ORDER_APPROVE, + required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, + approval_status="pending" if operation == "confirm" else "approved", + execution_status="submitted" if operation == "reconcile" else "not_submitted", + proposed_by="AUTOTEST", proposed_at=now, + ) + async with await psycopg.AsyncConnection.connect(dsn, row_factory=dict_row) as conn: + async with await psycopg.AsyncConnection.connect(dsn, autocommit=True) as external: + assert conn.info.dbname == external.info.dbname == "tjwater_v2" + repo = OperationalTaskRepository(conn, context.project_id) + service = OperationalTaskService(repo, context, { + OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, + }) + + async def get_action(*_args): + await conn.execute("SELECT 1") # Same implicit BEGIN as a product read. + return action.model_dump(mode="json") + + async def update_action(_task_id, document): + nonlocal action + await conn.execute("SELECT 1") + action = HumanApprovalAction.model_validate(document) + + async def get_task(*_args): + await conn.execute("SELECT 1") + return {"title": "AUTOTEST", "lifecycle_status": "open", "source": {}} + + async def get_request(*_args): + await conn.execute("SELECT 1") + return None + + async def detail(*_args): + await conn.execute("SELECT 1") + return SimpleNamespace(title="AUTOTEST", task_id=task_id, project_id=context.project_id) + + monkeypatch.setattr(repo, "get_action", get_action) + monkeypatch.setattr(repo, "get_task", get_task) + monkeypatch.setattr(repo, "get_request", get_request) + monkeypatch.setattr(repo, "update_action", update_action) + monkeypatch.setattr(repo, "save_request", AsyncMock()) + monkeypatch.setattr(service, "detail", detail) + monkeypatch.setattr(service, "_action_event", AsyncMock()) + monkeypatch.setattr(service, "_resolve_notifications", AsyncMock()) + monkeypatch.setattr(service, "_notify", AsyncMock()) + + async def dispatch(**_kwargs): + assert conn.info.transaction_status == TransactionStatus.IDLE, \ + "implicit outer transaction survives into the external call" + async with external.transaction(): + await external.execute("SET LOCAL lock_timeout='1s'") + await lock_event_transaction(external) + return ActionDispatchOutcome("submitted", "AUTOTEST receipt", ActionReceipt( + adapter="peripheral_sample", receipt_id=str(order_id), submitted_at=now, + external_status="open", detail="not persisted", + )) + + dispatcher = SimpleNamespace(dispatch=dispatch, reconcile=dispatch) + command = dict(request_id=uuid4(), expected_version=1) + if operation == "confirm": + result = await service.confirm_work_order_dispatch( + task_id, action_id, WorkOrderConfirmDispatchCommand(**command), dispatcher, None, + ) + elif operation == "submit": + result = await service.submit_action( + task_id, action_id, ActionExecuteCommand(**command), dispatcher, None, + ) + else: + result = await service.reconcile_action( + task_id, action_id, ActionReconcileCommand(**command), dispatcher, None, + ) + assert result.execution_status == "submitted" + + +@pytest.mark.anyio +async def test_work_order_filter_matches_source_and_receipts_without_cross_project_rows(): + dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2") + project_id, order_id = uuid4(), uuid4() + context = ProjectContext(project_id=project_id, project_code="event-filter-test", + user_id=uuid4(), project_role="member") + now = datetime.now(timezone.utc) + async with await psycopg.AsyncConnection.connect(dsn, row_factory=dict_row) as conn: + assert conn.info.dbname == "tjwater_v2" + async with conn.transaction(force_rollback=True): + repo = OperationalTaskRepository(conn, project_id) + service = OperationalTaskService(repo, context, set()) + + async def create(source, *, task_type="external_work_order"): + return await service.create(OperationalTaskCreate( + request_id=uuid4(), task_type=task_type, workflow="external-work-order", + title="AUTOTEST filtered work order", summary="isolated rollback", source=source, + trigger={"kind": "manual", "occurred_at": now, "detail": "filter regression"}, + )) + + projection = await create({"kind": "external_system", "external_id": f"peripheral_sample:{order_id}"}) + investigation = await create({"kind": "external_system", "external_id": str(order_id)}, + task_type="event_investigation") + linked = await create({"kind": "operator"}, task_type="event_investigation") + await create({"kind": "external_system", "external_id": f"peripheral_sample:{uuid4()}"}) + action = HumanApprovalAction( + action_id=uuid4(), task_id=linked.task_id, version=1, kind="create_work_order", + reason="AUTOTEST receipt", evidence_refs=[], + payload={"title": "AUTOTEST", "description": "rollback", "work_order_type": "general"}, + required_approval_permission=OPERATIONAL_ACTION_WORK_ORDER_APPROVE, + required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, + approval_status="approved", execution_status="submitted", proposed_by="AUTOTEST", proposed_at=now, + receipt=ActionReceipt(adapter="peripheral_sample", receipt_id=str(order_id), submitted_at=now, + external_status="open", detail="rollback"), + ) + await repo.insert_action(str(linked.task_id), action.model_dump(mode="json")) + items, cursor = await service.list(task_type=None, status=None, limit=100, offset=0, + peripheral_work_order_id=order_id) + assert {item.task_id for item in items} == {projection.task_id, investigation.task_id, linked.task_id} + assert cursor is None + foreign = OperationalTaskRepository(conn, uuid4()) + assert await foreign.list_tasks(task_type=None, status=None, limit=100, offset=0, + peripheral_work_order_id=str(order_id)) == [] + row = await (await conn.execute( + "SELECT count(*) AS n FROM workbench.operational_tasks WHERE project_id=%s", (project_id,), + )).fetchone() + assert row["n"] == 0 diff --git a/tests/unit/test_peripheral_push_sync.py b/tests/unit/test_peripheral_push_sync.py index 4518b6f..5b561bf 100644 --- a/tests/unit/test_peripheral_push_sync.py +++ b/tests/unit/test_peripheral_push_sync.py @@ -42,12 +42,25 @@ class PushStream(httpx.AsyncByteStream): ) +def test_large_source_burst_coalesces_to_full_reconciliation_with_bounded_ids(): + settings = WorkerSettings("http://server.test", "http://agent.test", str(uuid4()), bearer="TEST_ONLY") + synchronizer = PeripheralWorkOrderSynchronizer(None, settings, "http://sample.test", CredentialProvider(settings)) + order_id = str(uuid4()) + for _ in range(1000): + synchronizer._queue_order(order_id) + assert synchronizer.pending_order_ids == {order_id} + for _ in range(1000): + synchronizer._queue_order(str(uuid4())) + assert synchronizer.full_reconcile + assert synchronizer.pending_order_ids == set() + + @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 = [], [], [] + orders, tasks, cursors, task_queries = [], [], [], [] initial_scan, projected = asyncio.Event(), asyncio.Event() scan_count = 0 @@ -63,8 +76,11 @@ async def test_source_push_synchronizes_new_and_updated_orders_without_waiting_f if request.url.host == "sample.test": scan_count += 1 initial_scan.set() + if request.url.path.endswith(f"/work-orders/{order_id}"): + return httpx.Response(200, json=orders[0]) return httpx.Response(200, json={"items": orders, "next_cursor": None}) if request.method == "GET": + task_queries.append(dict(request.url.params)) return httpx.Response(200, json={"items": tasks, "next_offset": None}) body = json.loads(request.content) if request.url.path.endswith("external-feedback"): @@ -100,6 +116,8 @@ async def test_source_push_synchronizes_new_and_updated_orders_without_waiting_f assert tasks[0]["source"]["external_revision"] == 2 assert tasks[0]["status"] == "in_progress" assert scan_count == 3 # Startup plus two pushed source changes. + assert task_queries[0] == {"limit": "100", "offset": "0"} + assert all(query["peripheral_work_order_id"] == order_id for query in task_queries[1:]) assert cursors == ["20"] finally: runner.cancel() @@ -221,7 +239,7 @@ async def test_failed_scan_retries_without_requiring_another_source_message(monk return httpx.Response(200, json={"event_cursor": 20}) return httpx.Response(200, stream=stream) - async def sync_once(*_args): + async def sync_once(*_args, **_kwargs): nonlocal attempts attempts += 1 if attempts == 1: @@ -256,13 +274,14 @@ async def test_source_event_during_scan_schedules_another_pass(monkeypatch): return httpx.Response(200, json={"event_cursor": 20}) return httpx.Response(200, stream=stream) - async def sync_once(*_args): + async def sync_once(*_args, **kwargs): nonlocal attempts attempts += 1 if attempts == 1: scanning.set() await release.wait() else: + assert kwargs["work_order_ids"] == {order_id} caught_up.set() return 0 diff --git a/tests/unit/test_peripheral_task_sync.py b/tests/unit/test_peripheral_task_sync.py index 8617ad5..3123206 100644 --- a/tests/unit/test_peripheral_task_sync.py +++ b/tests/unit/test_peripheral_task_sync.py @@ -240,6 +240,7 @@ async def test_sample_sync_rejects_cross_project_data_before_writing(): @pytest.mark.anyio +@pytest.mark.parametrize("targeted", [False, True]) @pytest.mark.parametrize( ("investigation_status", "expected_feedback_count", "routes_to_investigation"), [ @@ -252,19 +253,24 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active investigation_status, expected_feedback_count, routes_to_investigation, + targeted, ): project_id, order_id, investigation_id, projection_id = map(str, (uuid4(), uuid4(), uuid4(), uuid4())) received = [] def handler(request): if request.url.host == "sample.test": - return httpx.Response(200, json={"items": [{ + order = { "project_id": project_id, "work_order_id": order_id, "origin": "synthetic", "created_by": str(uuid4()), "revision": 2, "summary": "采样已完成", "status": "completed", "updated_at": "2026-09-23T09:00:00Z", "title": "合成采样工单", "priority": "normal", "work_order_type": "general", "created_at": "2026-09-23T08:00:00Z", "scheduled_at": None, - }], "next_cursor": None}) + } + if targeted: + assert request.url.path.endswith(f"/work-orders/{order_id}") + return httpx.Response(200, json=order) + return httpx.Response(200, json={"items": [order], "next_cursor": None}) if request.method == "POST": assert request.url.path.endswith("external-feedback"), "Linked child orders must not start another investigation" received.append(request.url.path) @@ -276,6 +282,8 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active return httpx.Response(200, json={"actions": [{"kind": "create_work_order", "receipt": { "adapter": "peripheral_sample", "receipt_id": order_id, }}]}) + if targeted: + assert request.url.params["peripheral_work_order_id"] == order_id return httpx.Response(200, json={"items": [ {"task_id": investigation_id, "task_type": "event_investigation", "status": investigation_status}, {"task_id": projection_id, "task_type": "external_work_order", "source": { @@ -285,7 +293,8 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active settings = WorkerSettings("http://server.test", "http://agent.test", project_id, bearer="TEST_ONLY") async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: assert await sync_once( - client, settings, "http://sample.test", CredentialProvider(settings) + client, settings, "http://sample.test", CredentialProvider(settings), + **({"work_order_ids": {order_id}} if targeted else {}), ) == expected_feedback_count assert any(investigation_id in path for path in received) is routes_to_investigation assert any(projection_id in path for path in received) diff --git a/tests/unit/test_workbench_event_hub.py b/tests/unit/test_workbench_event_hub.py index 486d6ec..f633b89 100644 --- a/tests/unit/test_workbench_event_hub.py +++ b/tests/unit/test_workbench_event_hub.py @@ -1,10 +1,12 @@ import asyncio +from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from app.infra.db.postgresql.workbench_event_hub import WorkbenchEventHub, WorkbenchEventHubManager +from app.infra.db.postgresql import workbench_event_hub as hub_module @pytest.fixture @@ -30,3 +32,60 @@ async def test_project_subscribers_share_one_hub_and_idle_close(monkeypatch): await asyncio.sleep(0.03) assert len(manager._hubs) == 0 close.assert_awaited_once() + + +@pytest.mark.anyio +async def test_idle_subscribers_share_cursor_reads_and_notification_invalidates_the_cache(monkeypatch): + reads = 0 + + class Repository: + def __init__(self, *_args): + pass + + async def read_after(self, _after): + nonlocal reads + reads += 1 + return [] + + monkeypatch.setattr(hub_module, "WorkbenchEventRepository", Repository) + hub = WorkbenchEventHub(uuid4(), "postgresql://test/db") + hub._connection = SimpleNamespace(closed=False) + monkeypatch.setattr(hub, "start", AsyncMock()) + await asyncio.gather(*(hub.read_after(0) for _ in range(20))) + assert reads == 1 + await hub._wake() + await asyncio.gather(*(hub.read_after(0) for _ in range(20))) + assert reads == 2 + + +@pytest.mark.anyio +async def test_notification_during_read_does_not_cache_stale_results(monkeypatch): + entered, release = asyncio.Event(), asyncio.Event() + reads = 0 + + class Repository: + def __init__(self, *_args): + pass + + async def read_after(self, _after): + nonlocal reads + reads += 1 + if reads == 1: + entered.set() + await release.wait() + return [] + + monkeypatch.setattr(hub_module, "WorkbenchEventRepository", Repository) + hub = WorkbenchEventHub(uuid4(), "postgresql://test/db") + hub._connection = SimpleNamespace(closed=False) + monkeypatch.setattr(hub, "start", AsyncMock()) + first = asyncio.create_task(hub.read_after(0)) + await entered.wait() + await hub._wake() + release.set() + await first + await hub.read_after(0) + assert reads == 2 + for cursor in range(100): + await hub.read_after(cursor) + assert len(hub._reads) == 32 diff --git a/tests/unit/test_workbench_event_order.py b/tests/unit/test_workbench_event_order.py new file mode 100644 index 0000000..ed7a35b --- /dev/null +++ b/tests/unit/test_workbench_event_order.py @@ -0,0 +1,49 @@ +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock +from uuid import uuid4 + +import pytest + +from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository +from app.infra.db.postgresql.task_alerts import TaskAlertRepository +from app.infra.db.postgresql.workbench_events import ( + EVENT_ORDER_LOCK, lock_event_transaction_sync, +) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.anyio +@pytest.mark.parametrize("repository", [OperationalTaskRepository, TaskAlertRepository]) +@pytest.mark.parametrize("database", ["tjwater_v2", "other_project"]) +async def test_event_order_lock_precedes_business_writes_and_is_v2_only(repository, database): + @asynccontextmanager + async def transaction(): + yield + + connection = SimpleNamespace( + info=SimpleNamespace(dbname=database), execute=AsyncMock(), transaction=transaction, + ) + async with repository(connection, uuid4()).transaction(): + if database == "tjwater_v2": + connection.execute.assert_awaited_once_with( + "SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,), + ) + else: + connection.execute.assert_not_awaited() + + +def test_sync_producers_use_same_database_scoped_lock(): + connection = SimpleNamespace(info=SimpleNamespace(dbname="tjwater_v2"), execute=Mock()) + lock_event_transaction_sync(connection) + connection.execute.assert_called_once_with( + "SELECT pg_advisory_xact_lock(hashtextextended(%s,0))", (EVENT_ORDER_LOCK,), + ) + connection.info.dbname = "other_project" + connection.execute.reset_mock() + lock_event_transaction_sync(connection) + connection.execute.assert_not_called()