太好了，**B2 是 BPMN 真正“难的地方”**。
你能把 **Gateway 的 fork / join 并发语义**设计清楚，你这个引擎就已经站在 **Camunda / Zeebe 同一技术层级**了。
下面我给你的是 **严格对齐 BPMN 语义、但完全贴合你现有 Token 模型的设计**。
不是照抄规范，而是**工程可实现版本**。
---
# 🎯 一、B2 的目标（先讲清楚）
我们要解决的只有三件事：
1. **ExclusiveGateway（XOR）**
- 条件选择
- default flow
2. **ParallelGateway（AND）Fork**
- 一进多出
3. **ParallelGateway（AND）Join**
- 多进一出
- 严格同步
❗**不处理事件、不处理子流程、不处理复杂 Merge**
只聚焦「Token 并发 & Join 正确性」
---
# 🧠 二、核心思想（你一定要用）
## 关键结论（非常重要）
> **Join 不是节点问题，是 Token 协调问题**
因此：
- Gateway **不保存状态**
- Join 状态 **不挂在 Node 上**
- Join 状态 **属于运行时实例（Execution Scope）**
---
# 🧩 三、你需要新增的 Runtime 概念
## 1️⃣ JoinKey（唯一性）
```rust
#[derive(Hash, Eq, PartialEq)]
pub struct JoinKey {
pub process_instance_id: ProcessInstanceId,
pub gateway_id: NodeId,
}
```
这个 key 定义了：
> **“哪些 Token 应该一起 join”**
---
## 2️⃣ JoinState（运行时）
```rust
pub struct JoinState {
pub expected: usize,
pub arrived: HashSet<TokenId>,
}
```
- expected = gateway.incoming.len()
- arrived = 到达的 token
👉 存在 DB / 内存 Store 中
---
# 🔀 四、ExclusiveGateway（XOR）设计
## 语义
```text
1 Token in
→ evaluate outgoing in order
→ first condition == true
→ 1 Token out
```
---
## Engine 执行伪代码
```rust
fn execute_exclusive_gateway(
token: Token,
node: &CompiledNode,
vars: &Variables,
) {
for flow in &node.outgoing {
if flow.condition.map(|c| eval(c, vars)).unwrap_or(true) {
move_token(token, flow.target);
return;
}
}
panic!("No matching condition and no default flow");
}
```
---
## 注意事项（规范 + 工程）
- condition == None → true
- default flow → condition == None 且标记为 default
- 只能有一个 default flow（Compiler 校验）
---
# 🔱 五、ParallelGateway Fork（AND-split）
## 语义
```text
1 Token in
→ spawn N Tokens
→ each goes to outgoing
→ original token consumed
```
---
## 执行逻辑
```rust
fn fork_parallel_gateway(token: Token, node: &CompiledNode) {
for flow in &node.outgoing {
spawn_token(
token.process_instance_id,
flow.target,
token.variables.clone(),
);
}
consume(token);
}
```
👉 **Fork 永远是无状态的**
---
# 🔱 六、ParallelGateway Join（AND-join）【重点】
## BPMN 语义（简化）
> 当 **所有 incoming 分支** 都到达后，才继续
---
## Engine 行为（一步一步）
### 1️⃣ Token 到达 Join Gateway
```text
Token arrives at gateway G
→ register arrival
→ check if arrived == expected
```
---
### 2️⃣ 到达记录（必须原子）
```rust
fn on_token_arrival(token: Token, gateway: Gateway) {
let key = JoinKey {
process_instance_id: token.process_instance_id,
gateway_id: gateway.id,
};
let mut state = join_store.load_or_create(key, gateway.incoming.len());
state.arrived.insert(token.id);
if state.arrived.len() == state.expected {
join_store.remove(key);
merge_and_continue(state.arrived, gateway);
} else {
suspend(token);
}
}
```
---
### 3️⃣ merge_and_continue（合并）
你有两种选择：
### ✔️ 方案 A（推荐，最简单）
- 删除所有 arrived tokens
- 创建 **一个新 token**
- 用其中一个 token 的 variables（或 merge）
```rust
fn merge_and_continue(tokens: HashSet<TokenId>, gateway: Gateway) {
let merged_vars = merge_vars(tokens);
spawn_token(process_id, gateway.outgoing[0], merged_vars);
}
```
---
### ⚠️ 变量合并策略（必须明确）
| last-write-wins | 简单 |
| strict | key 冲突 → error |
| namespace | 按 branch |
👉 **建议 Phase 1 用 last-write-wins**
---
# 🔒 七、并发 & 事务边界（非常重要）
Join 是**并发最危险的地方**。
## 必须保证：
> **同一 JoinKey 下，只能有一个线程触发 merge**
---
## 推荐 DB 方案（Postgres）
```sql
SELECT *
FROM join_states
WHERE process_instance_id = ?
AND gateway_id = ?
FOR UPDATE;
```
---
## join_states 表
```sql
CREATE TABLE join_states (
process_instance_id UUID,
gateway_id TEXT,
expected INT,
arrived_tokens UUID[],
PRIMARY KEY (process_instance_id, gateway_id)
);
```
---
# ⚠️ 八、必须防的坑（血泪经验）
### ❌ 1. Token 重复到达
- 重试 / crash / 网络问题
- arrived_tokens 用 HashSet
---
### ❌ 2. Deadlock Join
- incoming 设计错误
- Compiler 阶段校验：
- ParallelJoin incoming > 1
- outgoing == 1
---
### ❌ 3. 混合 Gateway
BPMN 规范禁止：
```text
同一 ParallelGateway
既当 fork 又当 join
```
👉 Compiler 必须强制拆分
---
# 🧪 九、测试场景（你一定要写）
### Case 1：最简单 AND
```
A
\
+--> G --> C
/
B
```
---
### Case 2：三路并发
```
A -> G1 -> B,C,D -> G2 -> E
```
---
### Case 3：并发 + External Task + Retry
- 一个分支失败 & 重试
- join 必须等它
---
# 🏁 十、你做到这一步，意味着什么？
✔ 你解决了 BPM 引擎 **最难的并发问题**
✔ 你 Token 模型已经是「企业级」
✔ BPMN 执行语义完整闭环
---
## 下一步我强烈建议你选：
- **B4** → BPMN 编译期错误 & 诊断（非常重要）
- 或
- **W** → Worker + Gateway 在真实分布式下的交互
直接回：
👉 **B4** 或 **W**
我会继续陪你把这个引擎打到 **可以被真实公司用的级别** 💪