feat(engine): NOTIFY 抄送 + CONDITION 条件路由 + JOIN scope 修复(两批全栈)
build / build-push (push) Successful in 47s

- NOTIFY:TYPE_NOTIFY 瞬时抄送(flw_cc 逐人留痕 + CC_SENT 事件 + 站内通知不加待办角标)、/api/flow/cc/my、轨迹 kind=CC 与 oapi progress 抄送步、示意图青框卡片
- CONDITION:flw_router 判断服务注册 + flw_route_log 决策留痕(REQUIRES_NEW,reroute 回滚不丢失败行)+ 同步判断调用(服务目录寻址 + 平台签名,3s×3 退避重试)+ 失败挂起/管理端重跑(不配默认出口)+ 可选事件 flow.route.decided
- JOIN scope:branch_key 路径前缀编码 + scope 级到齐放行 + DefValidator SPLIT/JOIN 配对与键长守卫(存量无点号键零迁移,历史直连图回退实例级口径)
- 管理端:/api/flow/admin/router/* CRUD + POST /api/flow/admin/instance/{id}/reroute
- 生产 SQL:sql/20260923_create_flw_cc.sql(含 flw_task 时间列升 DATETIME(3))、20260924_create_flw_router_and_route_log.sql
- 测试:全量 125 绿(FlowEngineIT 24,含条件/嵌套并行/退回重走用例)
This commit is contained in:
chenxing
2026-09-24 08:58:26 +08:00
parent f2126cdf32
commit d2bef20e13
34 changed files with 2472 additions and 97 deletions
@@ -2,50 +2,45 @@ package com.yunhai.flowcenter.engine;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
/**
* 分支归属判定:节点属于哪个并行分支(branchKey=PARALLEL_SPLIT 出边目标节点 id),
* 主线节点与汇聚点之后属 null。运行时用于推导分支上下文({@link FlowEngine}),
* 分支归属判定(2026-09-24 §4.6 升级为路径前缀编码):节点属于哪个并行分支——
* 顶层分支键=分支头节点 id(无点号,与存量单层完全一致),内层分支键=「外层键.分支头id」
* (如 tb.tc,外层是内层的前缀),主线与汇聚链之后属 null。
* 运行时用于推导分支上下文({@link FlowEngine}:建行/到达标记/scope 级放行),
* 发布期用于拒绝跨分支连线({@link DefValidator})——两处必须同口径,故提取为静态工具。
*
* <p>嵌套口径(内层 JOIN 先于外层放行的关键):上溯遇 JOIN 视为透明块——跳到该 JOIN
* 匹配 SPLIT 的上游继续走(内层块整体属外层某分支,块的出口链沿用外层上下文)。
*/
public final class BranchScope {
/** 上溯递归深度上限:发布期已拦环,此处仅对未校验脏图防御性兜底 */
private static final int MAX_WALK_DEPTH = 64;
private BranchScope() {
}
/**
* 沿入边反向 BFS 走到最近的上游 PARALLEL_SPLIT,返回该 split 指向本链的 outEdge.to(即分支键);
* 遇 START/PARALLEL_JOIN 边界即停(join 之后属汇聚链,分支键已消耗,返回 null)。
* 节点的完整分支键(路径前缀编码)。沿入边反向 BFS 到最近的上游 PARALLEL_SPLIT,
* 返回「该 split 自身所在分支键.本链分支头id」;顶层 split 的分支头即键本身(存量兼容)。
* 遇 START 或(透明跳越后的)耗尽即停——主线节点返回 null。
*/
public static String keyOf(FlowDefModel model, String nodeId) {
if (model == null || nodeId == null) {
return null;
}
Set<String> visited = new HashSet<>();
Deque<String> queue = new ArrayDeque<>();
queue.add(nodeId);
visited.add(nodeId);
while (!queue.isEmpty()) {
String cur = queue.poll();
for (FlowDefModel.Edge e : model.inEdges(cur)) {
FlowDefModel.Node from = model.node(e.from);
if (from == null) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_SPLIT.equals(from.type)) {
return e.to;
}
if (FlowDefModel.TYPE_START.equals(from.type)
|| FlowDefModel.TYPE_PARALLEL_JOIN.equals(from.type)
|| !visited.add(e.from)) {
continue;
}
queue.add(e.from);
}
}
return null;
return keyOf(model, nodeId, new HashMap<String, String>(), new HashMap<String, FlowDefModel.Node>(), 0);
}
/**
* JOIN 的匹配分发节点:沿入边上溯(内层 JOIN 块透明跳越)首个 PARALLEL_SPLIT 即匹配者。
* 嵌套结构下内层 JOIN 必然先命中内层 split(内层先放行),外层 JOIN 透过内层块命中外层 split。
* 找不到(如主线节点直连 JOIN 的历史脏图)返回 null,由调用方决定兜底口径。
*/
public static FlowDefModel.Node matchedSplitOf(FlowDefModel model, String joinNodeId) {
return matchedSplitOf(model, joinNodeId, new HashMap<String, String>(), new HashMap<String, FlowDefModel.Node>(), 0);
}
/** 两节点是否同属一个分支上下文(含"都属主线"=都属于 null) */
@@ -54,4 +49,96 @@ public final class BranchScope {
String kb = keyOf(model, b);
return ka == null ? kb == null : ka.equals(kb);
}
private static String keyOf(FlowDefModel model, String nodeId,
Map<String, String> keyMemo, Map<String, FlowDefModel.Node> splitMemo, int depth) {
if (model == null || nodeId == null || depth > MAX_WALK_DEPTH) {
return null;
}
if (keyMemo.containsKey(nodeId)) {
return keyMemo.get(nodeId);
}
keyMemo.put(nodeId, null); // 占位防环:递归回访时按主线(null)处理并终止
String result = null;
Set<String> visited = new HashSet<>();
visited.add(nodeId);
Deque<String> queue = new ArrayDeque<>();
queue.add(nodeId);
while (!queue.isEmpty() && result == null) {
String cur = queue.poll();
for (FlowDefModel.Edge e : model.inEdges(cur)) {
FlowDefModel.Node from = model.node(e.from);
if (from == null) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_SPLIT.equals(from.type)) {
// 本链分支头确定:再取该 split 自身所在的外层上下文,拼成路径前缀
String outer = keyOf(model, from.id, keyMemo, splitMemo, depth + 1);
result = outer == null ? e.to : outer + "." + e.to;
break;
}
if (FlowDefModel.TYPE_START.equals(from.type)) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(from.type)) {
// 内层汇聚块透明:从其匹配 split 的上游继续上溯(块出口链沿用外层上下文)
FlowDefModel.Node innerSplit = matchedSplitOf(model, from.id, keyMemo, splitMemo, depth + 1);
if (innerSplit != null && visited.add(innerSplit.id)) {
queue.add(innerSplit.id);
}
continue;
}
if (visited.add(e.from)) {
queue.add(e.from);
}
}
}
keyMemo.put(nodeId, result);
return result;
}
private static FlowDefModel.Node matchedSplitOf(FlowDefModel model, String joinNodeId,
Map<String, String> keyMemo,
Map<String, FlowDefModel.Node> splitMemo, int depth) {
if (model == null || joinNodeId == null || depth > MAX_WALK_DEPTH) {
return null;
}
if (splitMemo.containsKey(joinNodeId)) {
return splitMemo.get(joinNodeId);
}
splitMemo.put(joinNodeId, null); // 占位防环
FlowDefModel.Node result = null;
Set<String> visited = new HashSet<>();
visited.add(joinNodeId);
Deque<String> queue = new ArrayDeque<>();
queue.add(joinNodeId);
while (!queue.isEmpty() && result == null) {
String cur = queue.poll();
for (FlowDefModel.Edge e : model.inEdges(cur)) {
FlowDefModel.Node from = model.node(e.from);
if (from == null) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_SPLIT.equals(from.type)) {
result = from;
break;
}
if (FlowDefModel.TYPE_START.equals(from.type)) {
continue;
}
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(from.type)) {
FlowDefModel.Node innerSplit = matchedSplitOf(model, from.id, keyMemo, splitMemo, depth + 1);
if (innerSplit != null && visited.add(innerSplit.id)) {
queue.add(innerSplit.id);
}
continue;
}
if (visited.add(e.from)) {
queue.add(e.from);
}
}
}
splitMemo.put(joinNodeId, result);
return result;
}
}