跳转至

外延服务

外延服务（plaita.server.services，需 server extra）是运行在流程之外的后台服务，负责监听外部触发源并把它们转成事件 publish 到 EventBus，从而唤醒挂起的流程。它们与扩展节点是"触发端"与"声明端"的配对。

ServiceManager

ServiceManager 统一管理所有外延服务的生命周期与任务分发：

from plaita.server.services.service_manager import ServiceManager

manager = ServiceManager(event_bus=bus)
manager.start_all_services(service_configs)
# ... 运行 ...
manager.stop_all_services()

它内置五类服务，按扩展节点产出的 service_config.type 分发任务：

type 服务类 配对节点
delay DelayService DelayNode
redis_queue RedisQueueService RedisQueueNode
kafka_queue KafkaQueueService KafkaQueueNode
http_callback HttpCallbackService HttpCallbackNode
approval ApprovalService ApprovalNode

可通过 register_service_class(service_type, service_class) 注册自定义服务。

闭环

flowchart LR
    N["扩展节点
(delay/approval/...)"] -->|挂起 + service_config| FW["FlowWorker"]
    FW -->|保存 context| ES["ExecutionStorage"]
    FW -->|注册订阅| EB["EventBus"]
    FW -->|RPUSH| Q["plaita:{type}:queue"]
    Q -->|消费| SC["外延服务"]
    SC -->|到点/收到消息/审批通过| EB
    EB -->|publish 事件| FW
    FW -->|resume| N
  1. 流程跑到扩展节点 → 生成 service_config → 挂起、存 context、订阅事件
  2. FlowWorker 把 service_config 当任务 RPUSH 到 plaita:{type}:queue（type 取节点产出的服务类型，如 plaita:approval:queue）
  3. 配对服务消费该队列并建立挂起所需的状态（审批记录 / 回调注册 / 排程），随后监听外部触发源（定时器/队列/HTTP/审批系统）
  4. 触发条件满足 → publish 一个 correlation_id=execution_id 的事件
  5. FlowWorker 收到事件 → run_distributed(resume_type="event", resume_data) 恢复

派发队列必须有消费方

消费 plaita:{type}:queue 是各服务自己的职责：DelayService 与 ApprovalService 起了常驻消费线程，http_callback / redis_queue / kafka_queue 的派发队列目前没有消费方（依赖它们的扩展节点在集群档上会一直挂起）。新增挂起类节点时务必同时确认队列有人读。

各服务职责

DelayService

按 service_config.delay_ms 设定定时器，到点发布 delay_trigger 事件。适合"X 分钟后继续"的场景。

RedisQueueService / KafkaQueueService

阻塞监听对应队列，消息到达后包装成事件发布。适合"等某条消息到达再继续"。

HttpCallbackService

注册回调路径（Redis 键 plaita:http_callback:registered:{path}，7 天 TTL，跨实例共享） 并在回调到达时先校验注册记录里的 auth_config、再原子认领，把回调 body 作为 event_data 发布。适合"等第三方异步通知"。

未授权请求返回 error 且不消费注册记录（认领是 GET+DEL 一体，先认领会把一次 未授权请求变成挂起执行永久失醒）；认证已配置时「凭据不对」与「路径未注册」是同一条 响应，不构成 path-existence oracle（免认证注册记录是已文档化的边界）；凭据载体见 扩展节点 § http_callback。

尚未暴露 HTTP 端点

唯一入口是进程内的 handle_callback_request(path, request_data, headers)； 集群档拉起的 python -m plaita.server.services http_callback_service 只是空转， 且其任务队列 plaita:http_callback:queue 无消费方（见上文派发队列警告）。

ApprovalService

对接审批系统：通知审批人、收集决策；满足 approval_strategy（any/all/majority）后发布 approval_decision 事件，event_data 含审批结果与意见。支持自动升级（auto_escalation）。

它常驻消费 plaita:approval:queue（worker 挂起时派发的审批任务），为每个审批落一条 plaita:approval:pending:{id} 记录（7 天 TTL，跨实例共享）；处理走完才出队，重放不会覆盖已落记录的审批决策。

启动入口

plaita.server.services.__main__ 提供命令行启动：

python -m plaita.server.services --help

也可在应用代码里手动构造 ServiceManager 嵌入自己的服务进程。

多实例与负载均衡

ServiceManager 支持多实例部署：事件总线（redis/sqlalchemy 后端）与状态存储共享，多个 worker 实例可分担挂起流程的恢复负载，EventProcessingTracker 保证事件不重复消费。

下一步