Просмотр исходного кода

feat(停止栅栏): 用 owner 与 stop epoch 完成跨进程 CAS

重复 stop intent 直接复用当前 stop_epoch。新增接管 Owner 校验和 complete_stop CAS:只有 owner_instance、owner_epoch、stop_epoch 与 Trace 全部匹配时,才能把 stopping 收敛为 stopped,旧进程 token 无法覆盖。
SamLee 20 часов назад
Родитель
Сommit
4410806744
1 измененных файлов с 75 добавлено и 0 удалено
  1. 75 0
      script_build_host/src/script_build_host/infrastructure/ownership.py

+ 75 - 0
script_build_host/src/script_build_host/infrastructure/ownership.py

@@ -344,6 +344,8 @@ class FencedCommandGate:
                 raise BuildNotFound()
             if status in {BuildStatus.SUCCESS.value, BuildStatus.STOPPED.value}:
                 return int(binding["stop_epoch"] or 0)
+            if status == BuildStatus.STOPPING.value:
+                return int(binding["stop_epoch"] or 0)
             stop_epoch = int(binding["stop_epoch"] or 0) + 1
             await session.execute(
                 update(mission_binding_table)
@@ -357,5 +359,78 @@ class FencedCommandGate:
             )
             return stop_epoch
 
+    async def verify_stop_owner(self, token: MissionOwnerToken) -> None:
+        """Verify the takeover owner while a build is intentionally fenced."""
+
+        async with self._sessions() as session, session.begin():
+            binding = (
+                (
+                    await session.execute(
+                        select(mission_binding_table)
+                        .where(mission_binding_table.c.script_build_id == token.script_build_id)
+                        .with_for_update()
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if binding is None:
+                raise BuildNotFound()
+            if (
+                str(binding["root_trace_id"]) != token.root_trace_id
+                or int(binding["owner_epoch"] or 0) != token.owner_epoch
+                or int(binding["stop_epoch"] or 0) != token.stop_epoch
+                or binding["owner_instance_id"] != token.owner_instance_id
+            ):
+                raise MissionFencingTokenStale()
+            status = await session.scalar(
+                select(self._runtime_record.c.status)
+                .where(self._runtime_record.c.id == token.script_build_id)
+                .with_for_update()
+            )
+            if status not in {BuildStatus.STOPPING.value, BuildStatus.STOPPED.value}:
+                raise MissionFencingTokenStale()
+
+    async def complete_stop(self, token: MissionOwnerToken) -> None:
+        """CAS the durable build state after operations and Trace are stopped."""
+
+        async with self._sessions() as session, session.begin():
+            binding = (
+                (
+                    await session.execute(
+                        select(mission_binding_table)
+                        .where(mission_binding_table.c.script_build_id == token.script_build_id)
+                        .with_for_update()
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if binding is None or (
+                int(binding["owner_epoch"] or 0) != token.owner_epoch
+                or int(binding["stop_epoch"] or 0) != token.stop_epoch
+                or binding["owner_instance_id"] != token.owner_instance_id
+            ):
+                raise MissionFencingTokenStale()
+            result = cast(
+                CursorResult[Any],
+                await session.execute(
+                    update(self._runtime_record)
+                    .where(
+                        self._runtime_record.c.id == token.script_build_id,
+                        self._runtime_record.c.status == BuildStatus.STOPPING.value,
+                    )
+                    .values(status=BuildStatus.STOPPED.value, end_time=datetime.now(UTC))
+                ),
+            )
+            if result.rowcount != 1:
+                current = await session.scalar(
+                    select(self._runtime_record.c.status).where(
+                        self._runtime_record.c.id == token.script_build_id
+                    )
+                )
+                if current != BuildStatus.STOPPED.value:
+                    raise MissionFencingTokenStale()
+
 
 __all__ = ["FencedCommandGate", "OwnerLease"]