“

系列：S5 智能决策 · 第 12 篇 | 难度：高级 | 阅读时间：20 分钟

Action 执行引擎：10 种执行器的统一调度

#TL;DR

Action 执行引擎是 coomia-dip 决策闭环中 Act（执行） 阶段的核心组件。它将决策结果转化为对 Ontology 的实际变更操作，通过统一的 Executor 抽象支持 10 种执行器类型——CreateObject、UpdateObject、DeleteObject、CreateRelation、DeleteRelation、InvokeFunction、Webhook、Notification、SimpleOp 和 CompositeOp。本文深入解析 ActionEngine 的调度架构、执行器生命周期、事务保证机制以及执行链编排模式，帮助你理解如何在企业级场景中实现可靠、可追溯的自动化操作执行。

#1. 为什么需要统一的 Action 引擎

#1.1 碎片化执行的困境

在传统系统中，不同类型的操作由不同的模块独立处理：

Code

传统模式：
┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│  CRUD 服务    │  │  消息推送服务  │  │  Webhook 服务 │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │                 │                 │
       ▼                 ▼                 ▼
   各自的事务          各自的重试          各自的日志
   各自的权限          各自的监控          各自的格式

这种碎片化带来的问题：

问题	具体表现	影响
事务不一致	部分操作成功、部分失败，无法回滚	数据损坏
审计困难	操作日志分散在多个系统中	合规风险
编排复杂	每增加一种操作类型需要新的集成代码	开发效率低
权限碎片	每个服务独立鉴权，策略不统一	安全漏洞

#1.2 统一调度的核心理念

coomia-dip 的 ActionEngine 采用 统一调度 + 多态执行 的架构：

Code

                 ┌───────────────────────────────────┐
                 │         ActionEngine               │
                 │                                   │
                 │  ┌─────────┐   ┌──────────────┐   │
                 │  │Scheduler│──→│ExecutorRouter │   │
                 │  └─────────┘   └──────┬───────┘   │
                 │                       │           │
                 │       ┌───────────────┼───────────┼───────┐
                 │       ▼               ▼           ▼       │
                 │  ┌─────────┐   ┌──────────┐  ┌────────┐  │
                 │  │CRUD Exec│   │Func Exec │  │Hook Exe│  │
                 │  └─────────┘   └──────────┘  └────────┘  │
                 │                                   │
                 │  ┌──────────────────────────────┐ │
                 │  │   Unified Audit Trail         │ │
                 │  └──────────────────────────────┘ │
                 └───────────────────────────────────┘

核心设计原则：

单一入口：所有操作通过 ActionEngine 统一调度
多态执行：10 种 Executor 实现统一接口
事务保证：支持本地事务和分布式 Saga
完整审计：每个 Action 都有完整的生命周期记录

#2. ActionEngine 核心架构

#2.1 整体架构图

Code

┌─────────────────────────────────────────────────────────────┐
│                      ActionEngine                           │
│                                                             │
│  ┌──────────┐    ┌────────────┐    ┌──────────────────┐     │
│  │ Action   │───→│ Validation │───→│ Permission Check │     │
│  │ Request  │    │   Layer    │    │    (ABAC/RBAC)   │     │
│  └──────────┘    └────────────┘    └────────┬─────────┘     │
│                                             │               │
│                                             ▼               │
│  ┌──────────────────────────────────────────────────┐       │
│  │              ExecutorRouter                       │       │
│  │                                                  │       │
│  │  executor_type ──→ resolve_executor()            │       │
│  │                        │                         │       │
│  │    ┌───────────────────┼───────────────────┐     │       │
│  │    ▼         ▼         ▼         ▼         ▼     │       │
│  │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐   │       │
│  │ │Create│ │Update│ │Delete│ │Invoke│ │Compo-│   │       │
│  │ │Object│ │Object│ │Object│ │ Func │ │ site │   │       │
│  │ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘   │       │
│  │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐   │       │
│  │ │Create│ │Delete│ │ Web- │ │Noti- │ │Simple│   │       │
│  │ │Relat.│ │Relat.│ │ hook │ │ficat.│ │  Op  │   │       │
│  │ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘   │       │
│  └──────────────────────────────────────────────────┘       │
│                         │                                   │
│                         ▼                                   │
│  ┌──────────────────────────────────────────────────┐       │
│  │              Result Collector                     │       │
│  │  ┌────────┐  ┌─────────┐  ┌────────────────┐    │       │
│  │  │ Audit  │  │ Metrics │  │ Event Emission │    │       │
│  │  │  Log   │  │ Report  │  │ (CDC Stream)   │    │       │
│  │  └────────┘  └─────────┘  └────────────────┘    │       │
│  └──────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────┘

#2.2 核心数据模型

Python

from enum import Enum
from pydantic import BaseModel, Field
from typing import Any
from datetime import datetime
import uuid


class ExecutorType(str, Enum):
    """10 种执行器类型"""
    CREATE_OBJECT = "CreateObject"
    UPDATE_OBJECT = "UpdateObject"
    DELETE_OBJECT = "DeleteObject"
    CREATE_RELATION = "CreateRelation"
    DELETE_RELATION = "DeleteRelation"
    INVOKE_FUNCTION = "InvokeFunction"
    WEBHOOK = "Webhook"
    NOTIFICATION = "Notification"
    SIMPLE_OP = "SimpleOp"
    COMPOSITE_OP = "CompositeOp"


class ActionStatus(str, Enum):
    PENDING = "pending"
    VALIDATING = "validating"
    EXECUTING = "executing"
    COMPENSATING = "compensating"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    ROLLED_BACK = "rolled_back"


class ActionRequest(BaseModel):
    """统一 Action 请求模型"""
    action_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    executor_type: ExecutorType
    target_object_type: str | None = None
    target_object_id: str | None = None
    parameters: dict[str, Any] = Field(default_factory=dict)
    idempotency_key: str | None = None
    triggered_by: str = "system"
    context: dict[str, Any] = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=datetime.utcnow)


class ActionResult(BaseModel):
    """统一 Action 结果模型"""
    action_id: str
    status: ActionStatus
    executor_type: ExecutorType
    result_data: dict[str, Any] = Field(default_factory=dict)
    error_message: str | None = None
    started_at: datetime | None = None
    completed_at: datetime | None = None
    duration_ms: int | None = None
    audit_trail: list[dict[str, Any]] = Field(default_factory=list)

#2.3 Executor 统一接口

Python

from abc import ABC, abstractmethod
from typing import AsyncContextManager


class BaseExecutor(ABC):
    """所有执行器的基类"""

    @property
    @abstractmethod
    def executor_type(self) -> ExecutorType:
        """返回执行器类型标识"""
        ...

    @abstractmethod
    async def validate(self, request: ActionRequest) -> list[str]:
        """
        验证请求参数是否合法。
        返回错误列表，空列表表示通过。
        """
        ...

    @abstractmethod
    async def execute(self, request: ActionRequest) -> ActionResult:
        """执行操作，返回结果"""
        ...

    @abstractmethod
    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """
        补偿操作：用于 Saga 回滚。
        根据已执行的结果，执行反向操作。
        """
        ...

    def supports_dry_run(self) -> bool:
        """是否支持 dry-run 模式"""
        return False

    async def dry_run(self, request: ActionRequest) -> ActionResult:
        """模拟执行，不实际修改数据"""
        raise NotImplementedError("Dry-run not supported")

#3. 十种执行器详解

#3.1 CRUD 执行器组（5 种）

CreateObjectExecutor

Python

class CreateObjectExecutor(BaseExecutor):
    """创建 Ontology 对象"""

    @property
    def executor_type(self) -> ExecutorType:
        return ExecutorType.CREATE_OBJECT

    async def validate(self, request: ActionRequest) -> list[str]:
        errors: list[str] = []
        if not request.target_object_type:
            errors.append("target_object_type is required")
        params = request.parameters
        if "properties" not in params:
            errors.append("properties dict is required")
        # 验证 ObjectType Schema
        schema = await self.ontology.get_object_type_schema(
            request.target_object_type
        )
        for field in schema.required_fields:
            if field not in params.get("properties", {}):
                errors.append(f"Missing required field: {field}")
        return errors

    async def execute(self, request: ActionRequest) -> ActionResult:
        started_at = datetime.utcnow()
        try:
            obj = await self.ontology.create_object(
                object_type=request.target_object_type,
                properties=request.parameters["properties"],
                created_by=request.triggered_by,
            )
            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.SUCCEEDED,
                executor_type=self.executor_type,
                result_data={"object_id": obj.id, "rid": obj.rid},
                started_at=started_at,
                completed_at=datetime.utcnow(),
            )
        except Exception as e:
            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.FAILED,
                executor_type=self.executor_type,
                error_message=str(e),
                started_at=started_at,
                completed_at=datetime.utcnow(),
            )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """补偿：删除已创建的对象"""
        if result.status == ActionStatus.SUCCEEDED:
            object_id = result.result_data["object_id"]
            await self.ontology.delete_object(
                object_type=request.target_object_type,
                object_id=object_id,
                deleted_by="saga_compensator",
            )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

UpdateObjectExecutor / DeleteObjectExecutor

执行器	操作	补偿策略	幂等键
CreateObject	创建 Ontology 对象	删除已创建对象	`{type}:{properties_hash}`
UpdateObject	更新对象属性	恢复为旧属性快照	`{type}:{id}:{version}`
DeleteObject	软/硬删除对象	重新创建对象（从快照恢复）	`{type}:{id}`

CreateRelationExecutor / DeleteRelationExecutor

Python

class CreateRelationExecutor(BaseExecutor):
    """创建对象之间的关联关系"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        relation = await self.ontology.create_relation(
            relation_type=params["relation_type"],
            source_object=params["source_rid"],
            target_object=params["target_rid"],
            properties=params.get("properties", {}),
        )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
            result_data={
                "relation_id": relation.id,
                "source_rid": params["source_rid"],
                "target_rid": params["target_rid"],
            },
        )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """补偿：删除已创建的关系"""
        await self.ontology.delete_relation(
            relation_id=result.result_data["relation_id"]
        )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

#3.2 函数调用执行器

Python

class InvokeFunctionExecutor(BaseExecutor):
    """调用用户自定义函数"""

    def __init__(self, function_runtime, timeout_seconds: int = 30):
        self.runtime = function_runtime
        self.timeout = timeout_seconds

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        function_ref = params["function_rid"]
        func_args = params.get("arguments", {})

        # 通过 FunctionRuntime 分发到对应沙箱
        execution = await self.runtime.invoke(
            function_rid=function_ref,
            arguments=func_args,
            timeout_seconds=self.timeout,
            caller_context=request.context,
        )

        return ActionResult(
            action_id=request.action_id,
            status=(ActionStatus.SUCCEEDED
                    if execution.success
                    else ActionStatus.FAILED),
            executor_type=self.executor_type,
            result_data={
                "return_value": execution.return_value,
                "execution_id": execution.id,
                "runtime_ms": execution.runtime_ms,
            },
            error_message=execution.error if not execution.success else None,
        )

    async def compensate(self, request: ActionRequest,
                         result: ActionResult) -> ActionResult:
        """调用函数对应的补偿函数（如果定义了）"""
        params = request.parameters
        compensate_ref = params.get("compensate_function_rid")
        if compensate_ref:
            await self.runtime.invoke(
                function_rid=compensate_ref,
                arguments={
                    "original_result": result.result_data,
                    "original_args": params.get("arguments", {}),
                },
            )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.ROLLED_BACK,
            executor_type=self.executor_type,
        )

#3.3 外部集成执行器

WebhookExecutor

Python

class WebhookExecutor(BaseExecutor):
    """调用外部 Webhook"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        url = params["url"]
        method = params.get("method", "POST")
        headers = params.get("headers", {})
        payload = params.get("payload", {})
        retry_config = params.get("retry", {"max_retries": 3, "backoff": "exponential"})

        # 签名验证
        if "signing_secret" in params:
            headers["X-Onto-Signature"] = self._compute_signature(
                payload, params["signing_secret"]
            )

        response = await self.http_client.request(
            method=method,
            url=url,
            json=payload,
            headers=headers,
            timeout=params.get("timeout_seconds", 30),
            retry_config=retry_config,
        )

        return ActionResult(
            action_id=request.action_id,
            status=(ActionStatus.SUCCEEDED
                    if 200 <= response.status_code < 300
                    else ActionStatus.FAILED),
            executor_type=self.executor_type,
            result_data={
                "status_code": response.status_code,
                "response_body": response.json(),
            },
        )

NotificationExecutor

Python

class NotificationExecutor(BaseExecutor):
    """发送通知到 9 种渠道"""

    SUPPORTED_CHANNELS = [
        "email", "sms", "webhook", "slack", "dingtalk",
        "wechat_work", "feishu", "in_app", "push",
    ]

    async def execute(self, request: ActionRequest) -> ActionResult:
        params = request.parameters
        channel = params["channel"]
        template_id = params.get("template_id")
        recipients = params["recipients"]

        # 渲染模板
        content = await self.template_engine.render(
            template_id=template_id,
            variables=params.get("variables", {}),
        )

        # 分发到对应渠道适配器
        adapter = self.channel_registry.get(channel)
        delivery_results = await adapter.send(
            recipients=recipients,
            content=content,
            metadata=params.get("metadata", {}),
        )

        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
            result_data={
                "channel": channel,
                "delivered": len([r for r in delivery_results if r.success]),
                "failed": len([r for r in delivery_results if not r.success]),
            },
        )

#3.4 编排执行器

SimpleOpExecutor

SimpleOp 是最轻量的执行器，用于执行不需要 Ontology 交互的简单操作：

Python

class SimpleOpExecutor(BaseExecutor):
    """执行简单的内联操作"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        op_type = request.parameters["op"]
        match op_type:
            case "set_variable":
                # 在执行上下文中设置变量
                key = request.parameters["key"]
                value = request.parameters["value"]
                request.context[key] = value
            case "log":
                level = request.parameters.get("level", "info")
                message = request.parameters["message"]
                self.logger.log(level, message)
            case "delay":
                seconds = request.parameters["seconds"]
                await asyncio.sleep(seconds)
            case "assert":
                condition = request.parameters["condition"]
                if not self._evaluate_condition(condition, request.context):
                    return ActionResult(
                        action_id=request.action_id,
                        status=ActionStatus.FAILED,
                        executor_type=self.executor_type,
                        error_message=f"Assertion failed: {condition}",
                    )
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED,
            executor_type=self.executor_type,
        )

CompositeOpExecutor

CompositeOp 是最强大的执行器，支持编排多个子 Action 的执行顺序：

Python

class CompositeOpExecutor(BaseExecutor):
    """编排多个子 Action 的复合执行器"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        steps = request.parameters["steps"]
        mode = request.parameters.get("mode", "sequential")
        # mode: sequential | parallel | conditional

        results: list[ActionResult] = []

        if mode == "sequential":
            for step in steps:
                sub_request = ActionRequest(**step)
                sub_result = await self.engine.dispatch(sub_request)
                results.append(sub_result)
                if sub_result.status == ActionStatus.FAILED:
                    if request.parameters.get("stop_on_failure", True):
                        break

        elif mode == "parallel":
            tasks = [
                self.engine.dispatch(ActionRequest(**step))
                for step in steps
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        elif mode == "conditional":
            for step in steps:
                condition = step.get("when")
                if condition and not self._evaluate(condition, request.context):
                    continue
                sub_request = ActionRequest(**step)
                sub_result = await self.engine.dispatch(sub_request)
                results.append(sub_result)

        all_succeeded = all(r.status == ActionStatus.SUCCEEDED for r in results)
        return ActionResult(
            action_id=request.action_id,
            status=ActionStatus.SUCCEEDED if all_succeeded else ActionStatus.FAILED,
            executor_type=self.executor_type,
            result_data={
                "mode": mode,
                "total_steps": len(steps),
                "completed": len([r for r in results if r.status == ActionStatus.SUCCEEDED]),
                "sub_results": [r.model_dump() for r in results],
            },
        )

#4. 调度器核心流程

#4.1 Action 生命周期

Code

┌──────────┐     ┌────────────┐     ┌───────────────┐     ┌───────────┐
│ PENDING  │────→│ VALIDATING │────→│  EXECUTING    │────→│ SUCCEEDED │
└──────────┘     └─────┬──────┘     └──────┬────────┘     └───────────┘
                       │                   │
                       │ validation        │ execution
                       │ failed            │ failed
                       ▼                   ▼
                 ┌──────────┐       ┌──────────────┐     ┌───────────┐
                 │  FAILED  │       │ COMPENSATING │────→│ROLLED_BACK│
                 └──────────┘       └──────────────┘     └───────────┘

#4.2 调度器实现

Python

class ActionScheduler:
    """Action 统一调度器"""

    def __init__(self, executor_registry: dict[ExecutorType, BaseExecutor]):
        self.executors = executor_registry
        self.audit_log = AuditLogger()
        self.metrics = MetricsCollector()
        self.idempotency_store = IdempotencyStore()

    async def dispatch(self, request: ActionRequest) -> ActionResult:
        """统一调度入口"""
        # 1. 幂等检查
        if request.idempotency_key:
            cached = await self.idempotency_store.get(request.idempotency_key)
            if cached:
                return cached

        # 2. 解析执行器
        executor = self.executors.get(request.executor_type)
        if not executor:
            raise ValueError(f"Unknown executor: {request.executor_type}")

        # 3. 权限检查
        await self._check_permissions(request)

        # 4. 参数验证
        errors = await executor.validate(request)
        if errors:
            result = ActionResult(
                action_id=request.action_id,
                status=ActionStatus.FAILED,
                executor_type=request.executor_type,
                error_message=f"Validation failed: {'; '.join(errors)}",
            )
            await self.audit_log.record(request, result, phase="validation")
            return result

        # 5. 执行
        await self.audit_log.record(request, phase="start")
        self.metrics.increment("action.dispatched", tags={
            "executor_type": request.executor_type.value
        })

        with self.metrics.timer("action.execution_time", tags={
            "executor_type": request.executor_type.value
        }):
            result = await executor.execute(request)

        # 6. 记录审计日志
        await self.audit_log.record(request, result, phase="complete")

        # 7. 缓存幂等结果
        if request.idempotency_key:
            await self.idempotency_store.set(
                request.idempotency_key, result, ttl=3600
            )

        # 8. 发出 CDC 事件
        await self._emit_cdc_event(request, result)

        return result

    async def _check_permissions(self, request: ActionRequest) -> None:
        """ABAC 权限检查"""
        decision = await self.policy_engine.evaluate(
            subject=request.triggered_by,
            action=request.executor_type.value,
            resource=f"{request.target_object_type}/{request.target_object_id}",
            context=request.context,
        )
        if not decision.allowed:
            raise PermissionError(
                f"Action denied: {decision.reason}"
            )

    async def _emit_cdc_event(self, request: ActionRequest,
                               result: ActionResult) -> None:
        """发出变更数据捕获事件"""
        if result.status == ActionStatus.SUCCEEDED:
            event = {
                "type": "action.completed",
                "action_id": request.action_id,
                "executor_type": request.executor_type.value,
                "target": {
                    "object_type": request.target_object_type,
                    "object_id": request.target_object_id,
                },
                "result": result.result_data,
                "timestamp": result.completed_at.isoformat(),
            }
            await self.event_bus.publish("ontology.actions", event)

#5. 执行器注册与发现

#5.1 自动注册机制

Python

from typing import Type


class ExecutorRegistry:
    """执行器注册表"""

    _executors: dict[ExecutorType, BaseExecutor] = {}

    @classmethod
    def register(cls, executor_cls: Type[BaseExecutor]) -> Type[BaseExecutor]:
        """装饰器：自动注册执行器"""
        instance = executor_cls()
        cls._executors[instance.executor_type] = instance
        return executor_cls

    @classmethod
    def get(cls, executor_type: ExecutorType) -> BaseExecutor:
        executor = cls._executors.get(executor_type)
        if not executor:
            raise KeyError(f"No executor registered for {executor_type}")
        return executor

    @classmethod
    def list_all(cls) -> dict[ExecutorType, BaseExecutor]:
        return dict(cls._executors)


# 使用装饰器注册
@ExecutorRegistry.register
class CreateObjectExecutor(BaseExecutor):
    ...

@ExecutorRegistry.register
class UpdateObjectExecutor(BaseExecutor):
    ...

#5.2 执行器能力矩阵

执行器	可补偿	幂等	Dry-run	超时控制	并行安全
CreateObject	Yes	Yes	Yes	No	Yes
UpdateObject	Yes	Yes	Yes	No	No*
DeleteObject	Yes	Yes	Yes	No	Yes
CreateRelation	Yes	Yes	Yes	No	Yes
DeleteRelation	Yes	Yes	Yes	No	Yes
InvokeFunction	Opt.	No	No	Yes	Yes
Webhook	No	Opt.	No	Yes	Yes
Notification	No	No	No	Yes	Yes
SimpleOp	N/A	Yes	No	Opt.	Yes
CompositeOp	Yes	Opt.	Yes	Yes	Dep.

“

*UpdateObject 对同一对象的并发更新需要乐观锁保护

#6. 事务保证与一致性

#6.1 本地事务（单个 Executor）

Python

class TransactionalExecutor(BaseExecutor):
    """带事务保证的执行器包装"""

    def __init__(self, inner: BaseExecutor, db_session_factory):
        self.inner = inner
        self.session_factory = db_session_factory

    async def execute(self, request: ActionRequest) -> ActionResult:
        async with self.session_factory() as session:
            try:
                result = await self.inner.execute(request)
                if result.status == ActionStatus.SUCCEEDED:
                    await session.commit()
                else:
                    await session.rollback()
                return result
            except Exception:
                await session.rollback()
                raise

#6.2 分布式事务（CompositeOp + Saga）

当 CompositeOp 包含多个跨服务的操作时，ActionEngine 与 Temporal 集成实现 Saga 模式：

Python

class SagaCompositeExecutor(CompositeOpExecutor):
    """基于 Saga 的复合执行器"""

    async def execute(self, request: ActionRequest) -> ActionResult:
        steps = request.parameters["steps"]
        completed: list[tuple[ActionRequest, ActionResult]] = []

        try:
            for step_data in steps:
                sub_request = ActionRequest(**step_data)
                sub_result = await self.engine.dispatch(sub_request)
                completed.append((sub_request, sub_result))

                if sub_result.status == ActionStatus.FAILED:
                    # 触发补偿
                    await self._compensate_all(completed)
                    return ActionResult(
                        action_id=request.action_id,
                        status=ActionStatus.ROLLED_BACK,
                        executor_type=self.executor_type,
                        error_message=f"Step failed: {sub_result.error_message}",
                    )

            return ActionResult(
                action_id=request.action_id,
                status=ActionStatus.SUCCEEDED,
                executor_type=self.executor_type,
            )

        except Exception as e:
            await self._compensate_all(completed)
            raise

    async def _compensate_all(
        self, completed: list[tuple[ActionRequest, ActionResult]]
    ) -> None:
        """反序补偿所有已完成的步骤"""
        for req, result in reversed(completed):
            executor = self.engine.executors.get(req.executor_type)
            if executor:
                try:
                    await executor.compensate(req, result)
                except Exception as e:
                    self.logger.error(
                        f"Compensation failed for {req.action_id}: {e}"
                    )
                    # 记录补偿失败，人工介入
                    await self.dead_letter_queue.enqueue(req, result, e)

#7. 监控与可观测性

#7.1 关键指标

Python

# ActionEngine 暴露的核心指标
METRICS = {
    "action_dispatched_total": Counter(
        "action_dispatched_total",
        "Total number of actions dispatched",
        labels=["executor_type", "status"],
    ),
    "action_execution_duration_seconds": Histogram(
        "action_execution_duration_seconds",
        "Action execution duration",
        labels=["executor_type"],
        buckets=[0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60],
    ),
    "action_compensation_total": Counter(
        "action_compensation_total",
        "Total compensations triggered",
        labels=["executor_type", "reason"],
    ),
    "action_dead_letter_total": Counter(
        "action_dead_letter_total",
        "Actions sent to dead letter queue",
        labels=["executor_type"],
    ),
}

#7.2 审计追踪

Code

┌────────────┬──────────────────────────────────────────────────────┐
│ 时间戳      │ 审计记录                                              │
├────────────┼──────────────────────────────────────────────────────┤
│ T+0ms      │ [DISPATCH] action_id=abc executor=CreateObject       │
│ T+2ms      │ [VALIDATE] passed, 0 errors                         │
│ T+5ms      │ [PERMISSION] allowed by policy: ontology.write       │
│ T+8ms      │ [EXECUTE]  started                                   │
│ T+45ms     │ [EXECUTE]  completed, object_id=obj-123              │
│ T+48ms     │ [CDC]      event emitted to ontology.actions          │
│ T+50ms     │ [AUDIT]    full record persisted                      │
└────────────┴──────────────────────────────────────────────────────┘

#8. 性能优化策略

#8.1 执行器连接池

Python

class ExecutorPool:
    """为高频执行器维护连接池"""

    def __init__(self, max_connections: int = 100):
        self.pool = asyncio.Semaphore(max_connections)
        self.active_count = 0

    async def execute_with_pool(
        self, executor: BaseExecutor, request: ActionRequest
    ) -> ActionResult:
        async with self.pool:
            self.active_count += 1
            try:
                return await executor.execute(request)
            finally:
                self.active_count -= 1

#8.2 批量操作优化

优化策略	适用场景	性能提升
批量 Create	一次创建多个同类型对象	5-10x
并行 Webhook	多个独立 Webhook 并发调用	3-5x
Pipeline 模式	顺序依赖但可流水线化	2-3x
缓存幂等结果	重复请求跳过执行	100x

#Key Takeaways

统一抽象：10 种执行器实现同一个 BaseExecutor 接口，调度器无需关心具体执行逻辑
可补偿设计：每个执行器定义 compensate() 方法，为 Saga 回滚提供标准化支持
幂等保证：通过 idempotency_key 和结果缓存避免重复执行
CompositeOp：支持 sequential/parallel/conditional 三种子 Action 编排模式
完整审计：每个 Action 从分发到完成的全生命周期均有审计记录
CDC 事件流：成功的 Action 自动发出变更事件，驱动下游响应

#Next Article

下一篇 S5-13 Saga 模式：长事务的补偿与回滚设计 将深入探讨当 CompositeOp 跨越多个微服务时，如何通过 Temporal 实现可靠的 Saga 编排、补偿策略设计以及死信队列处理。

tags: action-engine, executor, ontology, saga, composite-op, dispatch, coomia-dip