# bpm-engine
**Rust 语言实现的正确性优先工作流执行内核,专为确定性重放和 crash-safe 长流程设计。**
本项目专注于**执行语义、持久化正确性和 crash 安全**,而非 UI 或 low-code 特性。它被设计为一个** token 驱动、持久化优先的 BPM 引擎**,具有形式化定义的不变式。
---
## 这是什么?
`bpm-engine` 是一个用 Rust 实现的**工作流 / BPM 执行引擎**。
其核心思想是将流程执行为**持久化的 token 状态机**,其中:
- 每个执行步骤都由数据库状态驱动
- 每个状态转换都记录为历史
- 每个执行都可以被重放和验证
- 并发、重试和崩溃是首要考虑因素
这使得引擎适用于**长运行、分布式和易故障的工作流**。
---
## 为什么要另一个 BPM 引擎?
大多数 BPM 引擎以**特性和建模体验为优化目标**。
本引擎是一个**正确性优先的工作流执行内核**:以**正确性为优化目标**。
具体而言:
- Token 状态是**显式且持久化的**
- 执行**天然是 crash-safe 的**
- 外部任务使用**基于租约的执行**
- 定时器**完全持久化**
- 所有执行都是**可审计和可重放的**
- 核心行为受**形式化不变式保护**
如果你关心流程为什么到达某个状态——而不仅仅是它到达了——这个引擎适合你。
---
## 何时不使用 bpm-engine
本引擎以**正确性和可审计性为首要目标**。在以下情况下考虑替代方案:
- **你需要 low-code BPMN 建模和表单设计器**——使用 Camunda 或类似平台,它们开箱即提供可视化建模和任务 UI。
- **你严重依赖复杂的人工工作流和审批 UI**——本引擎专注于执行和语义,不提供内置的任务列表或表单。
- **执行语义不重要,只需要"快"或简单的 DAG**——更轻量的选项(如 AWS Step Functions)可能更简单。
如果你的优先级是**正确性、重放和清晰的执行语义**,本引擎是一个好选择。
---
## 核心概念
### 流程与实例
- **流程定义**是一张不可变的执行图
- **流程实例**是运行时 token 的容器
### Token
Token 代表执行单元。
- 每个 token 都有清晰的生命周期
- 状态转换被持久化
- 并行性通过 token 分叉和汇合来建模
### 外部任务
外部任务允许将工作交由外部 worker 执行:
- Worker 按主题获取任务
- 任务受**租约**保护
- 重试、超时和崩溃由引擎处理
- **引擎**保证 exactly-once token 完成;**worker** 至少一次且**必须**实现幂等处理器
### 定时器
定时器是持久的,由调度器驱动:
- 无内存中定时器
- 跨重启安全
- 自然可扩展
### 历史与重放
- 每个状态变更都会发出历史事件
- 执行可以确定性重放
- 历史可用于调试、审计和验证
**可观测性 API:**
**历史 API 语义:** 事件是追加式的;序列在每个实例内全局有序;重放会重现相同的 token 状态;schema 一旦发布向后兼容。API 稳定性和历史/追踪语义保证:见 [api-spec.md](docs/api-spec.md)(§ API & Semantic Stability,§ History & Trace Semantic Guarantees)。
**Crash 恢复验证:** 要验证 kill → 重启后的正确性(无重复完成、有序历史),请按照 [deploy/README.md](deploy/README.md) 操作,并从仓库根目录运行 `./deploy/verify-recovery.sh`。关于事故驱动的叙述(支付超时 → worker 重启 → 幂等完成,以及为何不变式成立),见 [docs/accident-scenarios.md](docs/accident-scenarios.md)。
### 不变式
引擎强制形式化不变式,例如:
- Token 只能到达最终状态一次
- Join 节点只有在所有分支都完成时才完成
- 外部任务同时只有一个拥有者
- 重试是单调的
详见 [docs/invariants.md](docs/invariants.md)。关于与 Camunda、Temporal 和 AWS Step Functions 的语义比较,见 [docs/why-correctness.md](docs/why-correctness.md)。
---
## 架构概览
```
+-------------------+
| Scheduler |
| Token Executor |
| Invariants |
+-------------------+
|
v
+-------------------+
| Runtime Tables |
| History / Timers |
+-------------------+
External Workers (fetch / lock / complete via API)
```
持久化层是**单一真相来源**。默认后端是**内存**(快速启动无需数据库)。引擎可以通过重新运行调度器来恢复。关于使用 PostgreSQL 的持久化导向部署,见 [docs/recovery.md](docs/recovery.md) 和 [docs/database-schema.md](docs/database-schema.md)。
### 代码阅读入口
- **引擎入口**:`bpm-engine-runtime::BpmEngine::run_async`
- **Token 转换**:`crates/runtime/src/handler/*`(及相关处理器)
- **持久化边界**:`bpm-engine-storage` traits(process, token, history, external task, timer)
- **历史发出**:`EngineEvent` 和 `Engine` 中的 `HistoryHandler`;REST 中的 `GET .../history`
- **不变式**:[docs/invariants.md](docs/invariants.md) 和 `tests/invariant_*.rs`
---
## 快速开始(5 分钟)
**要求:** Rust(stable)。默认内存后端无需 Docker。
```bash
git clone https://github.com/fanjia1024/bpm-engine.git
cd bpm-engine
cargo build
```
**1. 启动引擎**
```bash
cargo run -p bpm-server-rest
```
服务器监听 **http://127.0.0.1:3000**。内置流程定义:`minimal`(Start → End)、`payment-flow`(Start → ExternalTask `payment` → End)。
**2. 运行最小流程(Start → End)**
在另一个终端,启动实例并轮询直到完成:
```bash
cargo run --example simple_process
```
或使用 curl:
```bash
curl -X POST http://127.0.0.1:3000/api/v1/process-instances \
-H "Content-Type: application/json" \
-d '{"process_def_id":"minimal"}'
# 然后 GET /api/v1/process-instances/:id 直到状态为 COMPLETED
```
**3. 运行带有外部任务的流程(payment)**
启动流程实例:
```bash
curl -X POST http://127.0.0.1:3000/api/v1/process-instances \
-H "Content-Type: application/json" \
-d '{"process_def_id":"payment-flow","variables":{"amount":"100"}}'
```
运行 payment worker(在第三个终端):
```bash
cargo run -p bpm-worker-sdk --example payment
```
Worker 轮询引擎,锁定 `payment` 任务,运行处理器,然后完成;流程继续到 End。
---
## 示例:外部任务 Worker
```rust
use bpm_worker_sdk::{EngineClient, ExternalTask, TaskContext, TaskHandler, TaskResult, Worker, WorkerConfig};
struct PaymentHandler;
#[async_trait::async_trait]
impl TaskHandler for PaymentHandler {
fn task_type(&self) -> &str { "payment" }
async fn handle(&self, task: ExternalTask, _ctx: TaskContext) -> TaskResult {
// 业务逻辑
let mut variables = std::collections::HashMap::new();
variables.insert("status".to_string(), "PAID".to_string());
TaskResult::Complete { variables }
}
}
// Worker:无状态、crash-safe、水平可扩展
let worker = Worker::builder()
.client(EngineClient::new("http://127.0.0.1:3000"))
.handler(PaymentHandler)
.config(WorkerConfig::new("worker-1").poll_interval(std::time::Duration::from_secs(1)))
.build();
worker.start().await;
```
完整示例见 [crates/worker-sdk/examples/payment.rs](crates/worker-sdk/examples/payment.rs)。
---
## 保证
本引擎提供以下保证:
- **Exactly-once token 完成**
- **Crash-safe 执行**
- **确定性重放**
- **持久化定时器**
- **形式化不变式**在测试中验证
这些保证是**设计目标**,而非尽力而为的行为。
---
## 使用与 API
### REST API(基础路径 `/api/v1`)
| POST | `/process-instances` | 启动实例。Body: `{ "process_def_id", "variables"?: {} }` |
| GET | `/process-instances/:id` | 获取实例状态和当前节点 |
| POST | `/process-definitions/deploy` | 从 BPMN 2.0 XML 部署流程(body: raw XML) |
| GET | `/tasks?type=user\|external` | 列出等待中的任务 |
| POST | `/tasks/:task_id/complete` | 完成用户任务 |
| POST | `/external-tasks/fetch-and-lock` | Worker:获取并锁定任务 |
| POST | `/external-tasks/:task_id/complete` | Worker:完成任务 |
| POST | `/external-tasks/:task_id/fail` | Worker:失败任务 |
可选 header:`x-tenant-id` 用于租户隔离。
### 工作区 crates
- **bpm-core**:ProcessDefinition、NodeType(Start、End、UserTask、ExternalTask、gateways)、Token、ProcessInstance、EngineEvent
- **bpm-storage**:异步 traits(ProcessInstanceStore、TokenStore、ExternalTaskStore 等)
- **bpm-runtime**:BpmEngine、处理器、转换辅助
- **bpm-adapter-memory**:MemoryRepo;内存定义的 ProcessDefStore
- **bpm-bpmn**:BPMN 2.0 XML 解析器和 ProcessDefinition 编译器
- **bpm-server-rest**:HTTP API 服务器
- **bpm-worker-sdk**:EngineClient、Worker、TaskHandler;worker 代码无需 BPM 知识
将引擎作为库使用:按路径依赖上述 crate,构建带有 repo 的 [EngineContext](crates/runtime/src/handler.rs),然后运行 `BpmEngine::run_async(initial_event, &mut ctx)`。关于接线方式见 [crates/server/rest](crates/server/rest)。
---
## 文档
- [架构概览](docs/architecture.md)
- [执行模型(Token 与并发)](docs/execution-model.md)
- [不变式](docs/invariants.md)
- [持久化与恢复](docs/recovery.md)
- [恢复演示(kill → 重启)](docs/recovery-demo.md)
- [事故级场景](docs/accident-scenarios.md)(payment+retry、fork fail、worker reclaim)
- [数据库 Schema](docs/database-schema.md)
- [Saga 与补偿](docs/docs_saga.md)
- [测试策略](docs/docs_testing_strategy.md)
- [BPMN 映射](docs/bpmn-spec-mapping.md)
- [API 规范](docs/api-spec.md)
- [常见问题](docs/faq.md)
- [速查表](docs/cheat-sheet.md)
- [Rust Worker SDK](docs/sdk-rust.md)
- [Python SDK(计划中)](docs/sdk-python.md)
---
## 项目状态
本项目处于**积极开发中**。
- 核心执行语义已稳定
- API 可能演进
- 尚不推荐用于任务关键的生产环境
也就是说,引擎已适用于:
- 研究
- 原型开发
- 内部系统
- 正确性优先的实验
---
## 路线图
- Worker SDK 稳定化(Rust / Python)
- 只读执行检查器(Cockpit 类 UI)
- 更多不变式覆盖
- 文档与示例
- 历史/重放文档(待定)
---
## 贡献
欢迎贡献。
特别有价值的方向:
- 测试与不变式用例
- 文档
- Worker SDK 人体工程学
- 可视化工具
请见 [CONTRIBUTING.md](CONTRIBUTING.md)。
---
## 许可证
MIT