fields) {
chainStateMap.put(chainState.getInstanceId(), chainState);
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryLoopResultRepository.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryLoopResultRepository.java
new file mode 100644
index 0000000..55b0644
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryLoopResultRepository.java
@@ -0,0 +1,111 @@
+/**
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ *
+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl-3.0.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.easyagents.flow.core.chain.repository;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 进程内循环累计结果仓储，适用于单机运行和测试。
+ */
+public class InMemoryLoopResultRepository implements LoopResultRepository {
+
+ private final Map>> results = new ConcurrentHashMap<>();
+ private final Map> inputs = new ConcurrentHashMap<>();
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int storeInput(String resultId, Iterable items) {
+ List stored = new ArrayList<>();
+ for (Object item : items) {
+ stored.add(item);
+ }
+ List existing = inputs.putIfAbsent(resultId, stored);
+ return existing == null ? stored.size() : existing.size();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object loadInputItem(String resultId, int index) {
+ List stored = inputs.get(resultId);
+ if (stored == null) {
+ throw new IllegalStateException("Loop input not found: " + resultId);
+ }
+ return stored.get(index);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void removeInput(String resultId) {
+ inputs.remove(resultId);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void append(String resultId, int iterationIndex, Map outputValues) {
+ if (outputValues == null || outputValues.isEmpty()) {
+ return;
+ }
+ Map> result = results.computeIfAbsent(
+ resultId, ignored -> Collections.synchronizedMap(new LinkedHashMap<>()));
+ synchronized (result) {
+ for (Map.Entry entry : outputValues.entrySet()) {
+ List values = result.computeIfAbsent(entry.getKey(), ignored -> new ArrayList<>());
+ if (values.size() != iterationIndex) {
+ throw new IllegalStateException("Unexpected loop result index: " + iterationIndex);
+ }
+ values.add(entry.getValue());
+ }
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Map load(String resultId, int iterationCount, List outputNames) {
+ Map> result = results.get(resultId);
+ Map snapshot = new LinkedHashMap<>();
+ if (iterationCount == 0 || outputNames == null || outputNames.isEmpty()) {
+ return snapshot;
+ }
+ if (result == null) {
+ throw new IllegalStateException("Loop result not found: " + resultId);
+ }
+ synchronized (result) {
+ for (String outputName : outputNames) {
+ List values = result.get(outputName);
+ if (values == null || values.size() != iterationCount) {
+ throw new IllegalStateException("Incomplete loop result: " + outputName);
+ }
+ snapshot.put(outputName, new ArrayList<>(values));
+ }
+ }
+ return snapshot;
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryNodeStateRepository.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryNodeStateRepository.java
index 3bbfd38..231991b 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryNodeStateRepository.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/InMemoryNodeStateRepository.java
@@ -16,20 +16,33 @@
package com.easyagents.flow.core.chain.repository;
import com.easyagents.flow.core.chain.NodeState;
-import com.easyagents.flow.core.util.MapUtil;
import java.util.EnumSet;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+/**
+ * 进程内节点状态仓储。
+ */
public class InMemoryNodeStateRepository implements NodeStateRepository {
private static final Map chainStateMap = new ConcurrentHashMap<>();
+ /**
+ * {@inheritDoc}
+ */
@Override
public NodeState load(String instanceId, String nodeId) {
- String key = instanceId + "." + nodeId;
- return MapUtil.computeIfAbsent(chainStateMap, key, k -> {
+ // 保留进程内仓储原有的惰性初始化语义，避免改变既有直接读取行为。
+ return create(instanceId, nodeId, 0L);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public NodeState create(String instanceId, String nodeId, long chainStateVersion) {
+ return chainStateMap.computeIfAbsent(key(instanceId, nodeId), ignored -> {
NodeState nodeState = new NodeState();
nodeState.setChainInstanceId(instanceId);
nodeState.setNodeId(nodeId);
@@ -37,9 +50,23 @@ public class InMemoryNodeStateRepository implements NodeStateRepository {
});
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public boolean tryUpdate(NodeState newState, EnumSet fields, long version) {
- chainStateMap.put(newState.getChainInstanceId() + "." + newState.getNodeId(), newState);
+ chainStateMap.put(key(newState.getChainInstanceId(), newState.getNodeId()), newState);
return true;
}
+
+ /**
+ * 构建进程内节点状态键。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param nodeId 节点 ID
+ * @return 节点状态键
+ */
+ private String key(String instanceId, String nodeId) {
+ return instanceId + "." + nodeId;
+ }
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopInputReference.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopInputReference.java
new file mode 100644
index 0000000..73b921b
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopInputReference.java
@@ -0,0 +1,62 @@
+package com.easyagents.flow.core.chain.repository;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * 已分块保存的循环输入轻量引用。
+ *
+ * 循环节点按序读取分块；其他业务节点在参数读取边界会透明还原为与原输入等价的列表。
+ */
+public final class LoopInputReference implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final String REFERENCE_TYPE =
+ "easyflow.loop-input.v1";
+
+ private final String resultId;
+ private final int itemCount;
+
+ /**
+ * 创建循环输入引用。
+ *
+ * @param resultId 循环输入结果 ID
+ * @param itemCount 输入元素数量
+ */
+ public LoopInputReference(String resultId, int itemCount) {
+ this.resultId = Objects.requireNonNull(
+ resultId, "resultId must not be null");
+ if (itemCount < 0) {
+ throw new IllegalArgumentException(
+ "itemCount must not be negative");
+ }
+ this.itemCount = itemCount;
+ }
+
+ /**
+ * 获取循环输入结果 ID。
+ *
+ * @return 结果 ID
+ */
+ public String getResultId() {
+ return resultId;
+ }
+
+ /**
+ * 获取输入元素数量。
+ *
+ * @return 元素数量
+ */
+ public int getItemCount() {
+ return itemCount;
+ }
+
+ /**
+ * 获取跨异步审计边界使用的稳定引用类型。
+ *
+ * @return 引用类型
+ */
+ public String getReferenceType() {
+ return REFERENCE_TYPE;
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java
new file mode 100644
index 0000000..cd1a740
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultReference.java
@@ -0,0 +1,52 @@
+package com.easyagents.flow.core.chain.repository;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * 循环累计输出的轻量引用，避免完整列表回写到高频热状态。
+ */
+public final class LoopResultReference implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final String REFERENCE_TYPE =
+ "easyflow.loop-result.v1";
+
+ private final String resultId;
+ private final int iterationCount;
+ private final String outputName;
+
+ /**
+ * 创建循环结果引用。
+ *
+ * @param resultId 循环结果 ID
+ * @param iterationCount 迭代次数
+ * @param outputName 输出名称
+ */
+ public LoopResultReference(String resultId, int iterationCount, String outputName) {
+ this.resultId = Objects.requireNonNull(resultId, "resultId must not be null");
+ this.iterationCount = iterationCount;
+ this.outputName = Objects.requireNonNull(outputName, "outputName must not be null");
+ }
+
+ public String getResultId() {
+ return resultId;
+ }
+
+ public int getIterationCount() {
+ return iterationCount;
+ }
+
+ public String getOutputName() {
+ return outputName;
+ }
+
+ /**
+ * 获取跨异步审计边界使用的稳定引用类型。
+ *
+ * @return 引用类型
+ */
+ public String getReferenceType() {
+ return REFERENCE_TYPE;
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java
new file mode 100644
index 0000000..55cd40c
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/LoopResultRepository.java
@@ -0,0 +1,486 @@
+/**
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ *
+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl-3.0.txt
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.easyagents.flow.core.chain.repository;
+
+import com.easyagents.flow.core.chain.runtime.ExecutionBudgetExceededException;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+/**
+ * 循环节点累计结果仓储。
+ *
+ * 累计结果独立于高频更新的节点状态保存，避免每轮迭代重复序列化全部历史结果。
+ */
+public interface LoopResultRepository {
+
+ /**
+ * 流式保存不可随机访问的循环输入。
+ *
+ * @param resultId 循环结果 ID
+ * @param items 原始输入
+ * @return 输入元素数量
+ */
+ int storeInput(String resultId, Iterable items);
+
+ /**
+ * 在已启用的迭代预算内流式保存循环输入。
+ *
+ *
实现会在读取第 {@code maxItems + 1} 个元素前终止，避免超大或无限 Iterable
+ * 先产生无界 I/O。具体仓储应在下游写入异常时清理已落盘的部分分块。
+ *
+ * @param resultId 循环结果 ID
+ * @param items 原始输入
+ * @param maxItems 最大元素数；小于等于零表示不限制
+ * @return 输入元素数量
+ */
+ default int storeInput(String resultId, Iterable items, long maxItems) {
+ if (maxItems <= 0L) {
+ return storeInput(resultId, items);
+ }
+ Iterable bounded = () -> new Iterator() {
+ private final Iterator delegate = items.iterator();
+ private long count;
+
+ @Override
+ public boolean hasNext() {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public Object next() {
+ if (count >= maxItems) {
+ throw new ExecutionBudgetExceededException(
+ "Loop iteration budget exceeded while storing input "
+ + resultId
+ + ": more than "
+ + maxItems);
+ }
+ count++;
+ return delegate.next();
+ }
+ };
+ return storeInput(resultId, bounded);
+ }
+
+ /**
+ * 在实例锁和触发器认领均有效时流式保存循环输入。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param lockFencingToken 当前实例锁 token
+ * @param claimId 当前触发器 ID
+ * @param claimGeneration 当前触发器认领代际
+ * @param resultId 循环结果 ID
+ * @param items 原始输入
+ * @param maxItems 最大元素数；小于等于零表示不限制
+ * @return 输入元素数量
+ */
+ default int storeInput(
+ String instanceId,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration,
+ String resultId,
+ Iterable items,
+ long maxItems) {
+ return storeInput(resultId, items, maxItems);
+ }
+
+ /**
+ * 在生产者主动推送数据时流式保存循环输入。
+ *
+ * 缺省实现用于本地兼容仓储；分布式仓储应覆盖此方法并边接收边分块写入，
+ * 避免先构造完整列表。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param lockFencingToken 当前实例锁 token
+ * @param claimId 当前触发器 ID
+ * @param claimGeneration 当前触发器认领代际
+ * @param resultId 循环结果 ID
+ * @param producer 输入生产者
+ * @param maxItems 最大元素数；小于等于零表示不限制
+ * @return 输入元素数量
+ */
+ default int storeProducedInput(
+ String instanceId,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration,
+ String resultId,
+ InputProducer producer,
+ long maxItems) {
+ List items = new java.util.ArrayList<>();
+ producer.produce(item -> {
+ if (maxItems > 0L
+ && items.size() >= maxItems) {
+ throw new ExecutionBudgetExceededException(
+ "Loop iteration budget exceeded while storing input "
+ + resultId
+ + ": more than "
+ + maxItems);
+ }
+ items.add(item);
+ });
+ return storeInput(
+ instanceId,
+ lockFencingToken,
+ claimId,
+ claimGeneration,
+ resultId,
+ items,
+ 0L);
+ }
+
+ /**
+ * 主动向循环输入仓储推送元素的生产者。
+ */
+ @FunctionalInterface
+ interface InputProducer {
+
+ /**
+ * 生产并按原顺序推送输入元素。
+ *
+ * @param sink 单元素接收器
+ */
+ void produce(Consumer sink);
+ }
+
+ /**
+ * 按序号读取已保存的循环输入。
+ *
+ * @param resultId 循环结果 ID
+ * @param index 从零开始的序号
+ * @return 输入元素
+ */
+ Object loadInputItem(String resultId, int index);
+
+ /**
+ * 在业务参数读取边界透明还原完整循环输入。
+ *
+ * @param reference 循环输入引用
+ * @return 与原输入顺序一致的列表
+ */
+ default List loadInput(LoopInputReference reference) {
+ List items =
+ new java.util.ArrayList<>(
+ reference.getItemCount());
+ for (int index = 0;
+ index < reference.getItemCount();
+ index++) {
+ items.add(loadInputItem(
+ reference.getResultId(), index));
+ }
+ return items;
+ }
+
+ /**
+ * 清理循环输入。
+ *
+ * @param resultId 循环结果 ID
+ */
+ default void removeInput(String resultId) {
+ }
+
+ /**
+ * 释放指定循环结果的进程内活跃缓存。
+ *
+ * 该操作不得删除已经持久化的输入、输出分块或改变结果引用语义，仅用于在
+ * 循环完成后及时归还本机缓存空间。
+ *
+ * @param resultId 循环结果 ID
+ */
+ default void releaseActiveCache(String resultId) {
+ }
+
+ /**
+ * 在实例锁和触发器认领均有效时清理循环输入。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param lockFencingToken 当前实例锁 token
+ * @param claimId 当前触发器 ID
+ * @param claimGeneration 当前触发器认领代际
+ * @param resultId 循环结果 ID
+ */
+ default void removeInput(
+ String instanceId,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration,
+ String resultId) {
+ removeInput(resultId);
+ }
+
+ /**
+ * 追加一轮循环输出。
+ *
+ * @param resultId 循环结果 ID
+ * @param iterationIndex 从零开始的迭代序号
+ * @param outputValues 本轮输出
+ */
+ void append(String resultId, int iterationIndex, Map outputValues);
+
+ /**
+ * 在当前触发器 fencing token 仍有效时追加循环输出。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param fencingToken 当前触发器 token；非持久化执行为 {@code 0}
+ * @param resultId 循环结果 ID
+ * @param iterationIndex 迭代序号
+ * @param outputValues 本轮输出
+ */
+ default void append(
+ String instanceId,
+ long fencingToken,
+ String resultId,
+ int iterationIndex,
+ Map outputValues) {
+ append(resultId, iterationIndex, outputValues);
+ }
+
+ /**
+ * 在实例锁和当前触发器认领租约均有效时追加循环输出。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param lockFencingToken 当前实例锁 token；本地仓储调用为 {@code 0}
+ * @param claimId 当前触发器 ID；非持久化执行为 {@code null}
+ * @param claimGeneration 当前认领代际；非持久化执行为 {@code 0}
+ * @param resultId 循环结果 ID
+ * @param iterationIndex 迭代序号
+ * @param outputValues 本轮输出
+ */
+ default void append(
+ String instanceId,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration,
+ String resultId,
+ int iterationIndex,
+ Map outputValues) {
+ append(instanceId, lockFencingToken, resultId, iterationIndex, outputValues);
+ }
+
+ /**
+ * 加载完整循环累计结果。
+ *
+ * @param resultId 循环结果 ID
+ * @param iterationCount 已累计的迭代数
+ * @param outputNames 输出名称，顺序与工作流定义一致
+ * @return 按输出名称聚合的结果列表
+ */
+ Map load(String resultId, int iterationCount, List outputNames);
+
+ /**
+ * 为每个循环输出创建轻量引用。
+ *
+ * @param resultId 循环结果 ID
+ * @param iterationCount 已累计迭代数
+ * @param outputNames 输出名称
+ * @return 输出名称到轻量引用的映射
+ */
+ default Map references(
+ String resultId, int iterationCount, List outputNames) {
+ Map references = new LinkedHashMap<>();
+ if (outputNames != null) {
+ for (String outputName : outputNames) {
+ references.put(
+ outputName,
+ new LoopResultReference(resultId, iterationCount, outputName));
+ }
+ }
+ return references;
+ }
+
+ /**
+ * 解析单个循环输出引用。
+ *
+ * @param reference 循环输出引用
+ * @return 与旧实现相同的累计列表
+ */
+ default Object resolve(LoopResultReference reference) {
+ return load(
+ reference.getResultId(),
+ reference.getIterationCount(),
+ List.of(reference.getOutputName()))
+ .get(reference.getOutputName());
+ }
+
+ /**
+ * 递归解析业务输出中的循环引用，供参数读取和 API 边界透明还原。
+ *
+ * @param value 待解析值
+ * @return 不包含循环引用的业务值
+ */
+ default Object resolveReferences(Object value) {
+ return ReferenceResolver.resolve(this, value);
+ }
+
+ /**
+ * 单次递归解析中的批量读取器。
+ */
+ final class ReferenceResolver {
+
+ private ReferenceResolver() {
+ }
+
+ /**
+ * 收集同一循环结果的全部输出名称，并按结果组批量加载一次。
+ *
+ * @param repository 循环结果仓储
+ * @param value 待解析值
+ * @return 已透明还原的值
+ */
+ static Object resolve(LoopResultRepository repository, Object value) {
+ Map> outputNames =
+ new LinkedHashMap<>();
+ java.util.LinkedHashMap inputReferences =
+ new java.util.LinkedHashMap<>();
+ collect(value, outputNames, inputReferences);
+ Map> loaded = new LinkedHashMap<>();
+ outputNames.forEach((key, names) -> loaded.put(
+ key,
+ repository.load(
+ key.resultId,
+ key.iterationCount,
+ new java.util.ArrayList<>(names))));
+ Map> loadedInputs =
+ new LinkedHashMap<>();
+ inputReferences.forEach((resultId, reference) ->
+ loadedInputs.put(
+ resultId,
+ repository.loadInput(reference)));
+ return replace(value, loaded, loadedInputs);
+ }
+
+ /**
+ * 递归收集循环结果引用。
+ *
+ * @param value 当前值
+ * @param outputNames 分组后的输出名称
+ */
+ private static void collect(
+ Object value,
+ Map> outputNames,
+ Map inputReferences) {
+ if (value instanceof LoopResultReference) {
+ LoopResultReference reference = (LoopResultReference) value;
+ GroupKey key = new GroupKey(
+ reference.getResultId(), reference.getIterationCount());
+ outputNames.computeIfAbsent(
+ key, ignored -> new java.util.LinkedHashSet<>())
+ .add(reference.getOutputName());
+ return;
+ }
+ if (value instanceof LoopInputReference) {
+ LoopInputReference reference =
+ (LoopInputReference) value;
+ inputReferences.putIfAbsent(
+ reference.getResultId(), reference);
+ return;
+ }
+ if (value instanceof Map) {
+ ((Map) value).values().forEach(
+ item -> collect(
+ item, outputNames, inputReferences));
+ return;
+ }
+ if (value instanceof List) {
+ ((List) value).forEach(item -> collect(
+ item, outputNames, inputReferences));
+ }
+ }
+
+ /**
+ * 使用已批量加载的结果递归替换引用。
+ *
+ * @param value 当前值
+ * @param loaded 已加载结果
+ * @return 替换后的值
+ */
+ private static Object replace(
+ Object value,
+ Map> loaded,
+ Map> loadedInputs) {
+ if (value instanceof LoopResultReference) {
+ LoopResultReference reference = (LoopResultReference) value;
+ Map outputs = loaded.get(new GroupKey(
+ reference.getResultId(), reference.getIterationCount()));
+ return outputs == null ? null : outputs.get(reference.getOutputName());
+ }
+ if (value instanceof LoopInputReference) {
+ return loadedInputs.get(
+ ((LoopInputReference) value)
+ .getResultId());
+ }
+ if (value instanceof Map) {
+ Map resolved = new LinkedHashMap<>();
+ ((Map) value).forEach(
+ (key, item) -> resolved.put(
+ key,
+ replace(
+ item,
+ loaded,
+ loadedInputs)));
+ return resolved;
+ }
+ if (value instanceof List) {
+ List list = (List) value;
+ java.util.ArrayList resolved =
+ new java.util.ArrayList<>(list.size());
+ for (Object item : list) {
+ resolved.add(replace(
+ item, loaded, loadedInputs));
+ }
+ return resolved;
+ }
+ return value;
+ }
+
+ /**
+ * 循环结果批量读取分组键。
+ */
+ private static final class GroupKey {
+
+ private final String resultId;
+ private final int iterationCount;
+
+ private GroupKey(String resultId, int iterationCount) {
+ this.resultId = resultId;
+ this.iterationCount = iterationCount;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof GroupKey)) {
+ return false;
+ }
+ GroupKey that = (GroupKey) other;
+ return iterationCount == that.iterationCount
+ && java.util.Objects.equals(resultId, that.resultId);
+ }
+
+ @Override
+ public int hashCode() {
+ return java.util.Objects.hash(resultId, iterationCount);
+ }
+ }
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateField.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateField.java
index e5d902c..95359ee 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateField.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateField.java
@@ -27,5 +27,7 @@ public enum NodeStateField {
SUSPEND_NODE_IDS,
SUSPEND_FOR_PARAMETERS,
EXECUTE_RESULT,
- RETRY_COUNT, EXECUTE_COUNT, EXECUTE_EDGE_IDS, LOOP_COUNT, TRIGGER_COUNT, TRIGGER_EDGE_IDS, ENVIRONMENT
+ RETRY_COUNT, EXECUTE_COUNT, EXECUTE_EDGE_IDS, EXECUTION_ATTEMPT_KEY,
+ LOOP_COUNT, TRIGGER_COUNT, TRIGGER_EDGE_IDS, ENVIRONMENT,
+ VERSION
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateRepository.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateRepository.java
index 369395d..0f1d779 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateRepository.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/repository/NodeStateRepository.java
@@ -21,7 +21,123 @@ import java.util.EnumSet;
public interface NodeStateRepository {
+ /**
+ * 加载已存在的节点状态。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param nodeId 节点 ID
+ * @return 节点状态；纯读取实现可在状态缺失时返回 {@code null}，兼容实现可惰性创建
+ */
NodeState load(String instanceId, String nodeId);
+ /**
+ * 显式创建节点状态。
+ *
+ * 缺省实现兼容旧仓储中由 {@link #load(String, String)} 完成首次创建的行为。
+ * 支持持久化或分布式执行的实现应覆盖本方法并原子创建状态。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param nodeId 节点 ID
+ * @param chainStateVersion 创建时关联的工作流状态版本
+ * @return 已存在或新创建的节点状态
+ */
+ default NodeState create(String instanceId, String nodeId, long chainStateVersion) {
+ NodeState existing = load(instanceId, nodeId);
+ if (existing != null) {
+ return existing;
+ }
+ NodeState created = new NodeState();
+ created.setChainInstanceId(instanceId);
+ created.setNodeId(nodeId);
+ if (tryUpdate(
+ created,
+ EnumSet.noneOf(NodeStateField.class),
+ chainStateVersion)) {
+ return created;
+ }
+ return load(instanceId, nodeId);
+ }
+
+ /**
+ * 在当前触发器 fencing token 仍有效时显式创建节点状态。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param nodeId 节点 ID
+ * @param chainStateVersion 工作流状态版本
+ * @param fencingToken 当前触发器 token；非触发器调用为 {@code 0}
+ * @return 已存在或新创建的节点状态
+ */
+ default NodeState create(
+ String instanceId, String nodeId, long chainStateVersion, long fencingToken) {
+ return create(instanceId, nodeId, chainStateVersion);
+ }
+
+ /**
+ * 在实例锁和当前触发器认领租约均有效时显式创建节点状态。
+ *
+ * @param instanceId 工作流实例 ID
+ * @param nodeId 节点 ID
+ * @param chainStateVersion 工作流状态版本
+ * @param lockFencingToken 当前实例锁 token；本地仓储调用为 {@code 0}
+ * @param claimId 当前触发器 ID；非触发器调用为 {@code null}
+ * @param claimGeneration 当前认领代际；非触发器调用为 {@code 0}
+ * @return 已存在或新创建的节点状态
+ */
+ default NodeState create(
+ String instanceId,
+ String nodeId,
+ long chainStateVersion,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration) {
+ return create(instanceId, nodeId, chainStateVersion, lockFencingToken);
+ }
+
+ /**
+ * 按版本尝试提交节点状态。
+ *
+ * @param newState 待提交的新状态
+ * @param fields 本次变更字段
+ * @param chainStateVersion 本次提交依赖的工作流状态版本
+ * @return 提交成功时为 {@code true}，版本冲突时为 {@code false}
+ */
boolean tryUpdate(NodeState newState, EnumSet fields, long chainStateVersion);
+
+ /**
+ * 在工作流版本和 fencing token 同时有效时提交节点状态。
+ *
+ * @param newState 待提交节点状态
+ * @param fields 变化字段
+ * @param chainStateVersion 工作流状态版本
+ * @param fencingToken 当前触发器 token；非触发器调用为 {@code 0}
+ * @return 提交成功时为 {@code true}
+ */
+ default boolean tryUpdate(
+ NodeState newState,
+ EnumSet fields,
+ long chainStateVersion,
+ long fencingToken) {
+ return tryUpdate(newState, fields, chainStateVersion);
+ }
+
+ /**
+ * 在工作流版本、实例锁和当前触发器认领租约同时有效时提交节点状态。
+ *
+ * @param newState 待提交节点状态
+ * @param fields 变化字段
+ * @param chainStateVersion 工作流状态版本
+ * @param lockFencingToken 当前实例锁 token；本地仓储调用为 {@code 0}
+ * @param claimId 当前触发器 ID；非触发器调用为 {@code null}
+ * @param claimGeneration 当前认领代际；非触发器调用为 {@code 0}
+ * @return 提交成功时为 {@code true}
+ */
+ default boolean tryUpdate(
+ NodeState newState,
+ EnumSet fields,
+ long chainStateVersion,
+ long lockFencingToken,
+ String claimId,
+ long claimGeneration) {
+ return tryUpdate(newState, fields, chainStateVersion, lockFencingToken);
+ }
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
index a63bfa3..130d1a8 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
@@ -25,6 +25,7 @@ import com.easyagents.flow.core.chain.repository.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.Serializable;
import java.util.*;
import java.util.concurrent.*;
@@ -39,24 +40,61 @@ import java.util.concurrent.*;
public class ChainExecutor {
private static final Logger log = LoggerFactory.getLogger(ChainExecutor.class);
+ private static final String NESTED_DEPTH_MEMORY_KEY =
+ "__tinyflow.nesting.depth";
+ private static final String DEFINITION_CALL_PATH_MEMORY_KEY =
+ "__tinyflow.nesting.definitionPath";
+ /**
+ * 子工作流触发器使用的独立执行通道。
+ */
+ public static final String CHILD_WORKFLOW_EXECUTION_LANE_PREFIX =
+ "child-workflow:";
+ private volatile Semaphore rootChildExecutionPermits =
+ new Semaphore(32, true);
+ private volatile long persistentOutcomePollMillis = 500L;
+ private volatile int maxChildExecutionLaneDepth =
+ Integer.MAX_VALUE;
+ private static final String CHILD_EXECUTION_REFERENCE_KEY =
+ "__tinyflow.workflowNode.childExecution";
+ /**
+ * 进程内定义快照仅作为热点缓存；持久快照负责跨节点和跨进程恢复。
+ */
+ private static final int MAX_ACTIVE_DEFINITIONS = 1024;
private final ChainDefinitionRepository definitionRepository;
private final ChainStateRepository chainStateRepository;
private final NodeStateRepository nodeStateRepository;
+ private final LoopResultRepository loopResultRepository;
+ private final ChainDefinitionSnapshotRepository definitionSnapshotRepository;
private final TriggerScheduler triggerScheduler;
+ private final ExecutionBudget executionBudget;
private final EventManager eventManager = new EventManager();
/** 等待同步执行结果的任务，按工作流实例 ID 进行常量时间路由。 */
private final ConcurrentMap>> pendingExecutions =
new ConcurrentHashMap<>();
+ /**
+ * 活跃工作流实例使用的定义快照，避免每个节点触发都重新加载和解析定义。
+ */
+ private final Map activeDefinitions =
+ Collections.synchronizedMap(new LinkedHashMap<>(
+ MAX_ACTIVE_DEFINITIONS + 1, 0.75F, true) {
+ @Override
+ protected boolean removeEldestEntry(
+ Map.Entry eldest) {
+ return size() > MAX_ACTIVE_DEFINITIONS;
+ }
+ });
public ChainExecutor(ChainDefinitionRepository definitionRepository
, ChainStateRepository chainStateRepository
, NodeStateRepository nodeStateRepository
) {
- this.definitionRepository = definitionRepository;
- this.chainStateRepository = chainStateRepository;
- this.nodeStateRepository = nodeStateRepository;
- this.triggerScheduler = ChainRuntime.triggerScheduler();
- registerRuntimeCallbacks();
+ this(definitionRepository,
+ chainStateRepository,
+ nodeStateRepository,
+ new InMemoryLoopResultRepository(),
+ new InMemoryChainDefinitionSnapshotRepository(),
+ ChainRuntime.triggerScheduler(),
+ ExecutionBudget.defaults());
}
@@ -64,10 +102,92 @@ public class ChainExecutor {
, ChainStateRepository chainStateRepository
, NodeStateRepository nodeStateRepository
, TriggerScheduler triggerScheduler) {
+ this(definitionRepository,
+ chainStateRepository,
+ nodeStateRepository,
+ new InMemoryLoopResultRepository(),
+ new InMemoryChainDefinitionSnapshotRepository(),
+ triggerScheduler,
+ ExecutionBudget.defaults());
+ }
+
+ /**
+ * 创建使用指定调度器和资源预算的执行器。
+ *
+ * @param definitionRepository 工作流定义仓储
+ * @param chainStateRepository 工作流状态仓储
+ * @param nodeStateRepository 节点状态仓储
+ * @param triggerScheduler 触发调度器
+ * @param executionBudget 平台资源保护预算
+ */
+ public ChainExecutor(ChainDefinitionRepository definitionRepository
+ , ChainStateRepository chainStateRepository
+ , NodeStateRepository nodeStateRepository
+ , TriggerScheduler triggerScheduler
+ , ExecutionBudget executionBudget) {
+ this(definitionRepository,
+ chainStateRepository,
+ nodeStateRepository,
+ new InMemoryLoopResultRepository(),
+ new InMemoryChainDefinitionSnapshotRepository(),
+ triggerScheduler,
+ executionBudget);
+ }
+
+ /**
+ * 创建使用指定调度器、循环结果仓储和资源预算的执行器。
+ *
+ * @param definitionRepository 工作流定义仓储
+ * @param chainStateRepository 工作流状态仓储
+ * @param nodeStateRepository 节点状态仓储
+ * @param loopResultRepository 循环累计结果仓储
+ * @param triggerScheduler 触发调度器
+ * @param executionBudget 平台资源保护预算
+ */
+ public ChainExecutor(ChainDefinitionRepository definitionRepository
+ , ChainStateRepository chainStateRepository
+ , NodeStateRepository nodeStateRepository
+ , LoopResultRepository loopResultRepository
+ , TriggerScheduler triggerScheduler
+ , ExecutionBudget executionBudget) {
+ this(definitionRepository,
+ chainStateRepository,
+ nodeStateRepository,
+ loopResultRepository,
+ new InMemoryChainDefinitionSnapshotRepository(),
+ triggerScheduler,
+ executionBudget);
+ }
+
+ /**
+ * 创建使用持久定义快照的执行器。
+ *
+ * @param definitionRepository 工作流定义仓储
+ * @param chainStateRepository 工作流状态仓储
+ * @param nodeStateRepository 节点状态仓储
+ * @param loopResultRepository 循环累计结果仓储
+ * @param definitionSnapshotRepository 实例级定义快照仓储
+ * @param triggerScheduler 触发调度器
+ * @param executionBudget 平台资源保护预算
+ */
+ public ChainExecutor(ChainDefinitionRepository definitionRepository
+ , ChainStateRepository chainStateRepository
+ , NodeStateRepository nodeStateRepository
+ , LoopResultRepository loopResultRepository
+ , ChainDefinitionSnapshotRepository definitionSnapshotRepository
+ , TriggerScheduler triggerScheduler
+ , ExecutionBudget executionBudget) {
this.definitionRepository = definitionRepository;
this.chainStateRepository = chainStateRepository;
this.nodeStateRepository = nodeStateRepository;
+ this.loopResultRepository = loopResultRepository == null
+ ? new InMemoryLoopResultRepository()
+ : loopResultRepository;
+ this.definitionSnapshotRepository = definitionSnapshotRepository == null
+ ? new InMemoryChainDefinitionSnapshotRepository()
+ : definitionSnapshotRepository;
this.triggerScheduler = triggerScheduler;
+ this.executionBudget = executionBudget == null ? ExecutionBudget.defaults() : executionBudget;
registerRuntimeCallbacks();
}
@@ -80,33 +200,47 @@ public class ChainExecutor {
public Map execute(String definitionId, Map variables, long timeout, TimeUnit unit) {
Chain chain = createChain(definitionId);
String stateInstanceId = chain.getStateInstanceId();
- CompletableFuture> future = new CompletableFuture<>();
-
- CompletableFuture> existing = pendingExecutions.putIfAbsent(stateInstanceId, future);
- if (existing != null) {
- throw new IllegalStateException("Duplicate pending chain execution: " + stateInstanceId);
- }
try {
chain.start(variables);
- Map result = future.get(timeout, unit);
+ Map result = awaitPersistentOutcome(
+ stateInstanceId, timeout, unit, null);
clearDefaultStates(result);
return result;
} catch (TimeoutException e) {
- future.cancel(true);
+ cancel(stateInstanceId, "Execution timed out");
throw new RuntimeException("Execution timed out", e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- future.cancel(true);
+ cancel(stateInstanceId, "Execution interrupted");
throw new RuntimeException("Execution interrupted", e);
} catch (Throwable e) {
- future.cancel(true);
- throw new RuntimeException("Execution failed", e.getCause());
+ throw new RuntimeException("Execution failed", e);
} finally {
- pendingExecutions.remove(stateInstanceId, future);
+ activeDefinitions.remove(stateInstanceId);
}
}
+ /**
+ * 取消仍在运行的工作流实例。
+ *
+ * 取消状态写入后，未开始的触发器会在执行入口短路；已经完成的外部 I/O 也会在提交
+ * 结果前重新检查状态，避免继续推进下游。
+ *
+ * @param stateInstanceId 工作流实例 ID
+ * @param message 取消原因
+ * @return 本次是否完成了非终态到取消状态的转换
+ */
+ public boolean cancel(String stateInstanceId, String message) {
+ ChainState state = chainStateRepository.load(stateInstanceId);
+ if (state == null) {
+ return false;
+ }
+ ChainDefinition definition = getDefinitionForInstance(state);
+ Chain chain = configureChain(definition, stateInstanceId);
+ return chain.cancel(message);
+ }
+
/**
* 注册工作流调度和同步结果路由回调。
*/
@@ -114,6 +248,79 @@ public class ChainExecutor {
eventManager.addEventListener(ChainStatusChangeEvent.class, this::completePendingExecution);
eventManager.addChainErrorListener(this::failPendingExecution);
triggerScheduler.registerConsumer(this::accept);
+ triggerScheduler.registerFailureListener(
+ this::failDeadLetteredTrigger);
+ }
+
+ /**
+ * 配置同步子工作流的容量和持久状态轮询参数。
+ *
+ * 该方法应在执行器对外提供服务前调用。
+ *
+ * @param rootPermits 根级同步子流程最大并发
+ * @param pollMillis 持久终态轮询间隔毫秒数
+ * @param laneMaxDepth 已注册独立执行通道覆盖的最大深度
+ */
+ public void configureChildWorkflowRuntime(
+ int rootPermits,
+ long pollMillis,
+ int laneMaxDepth) {
+ if (rootPermits <= 0 || pollMillis <= 0L
+ || laneMaxDepth <= 0) {
+ throw new IllegalArgumentException(
+ "Child workflow runtime values must be positive");
+ }
+ this.rootChildExecutionPermits =
+ new Semaphore(rootPermits, true);
+ this.persistentOutcomePollMillis = pollMillis;
+ this.maxChildExecutionLaneDepth = laneMaxDepth;
+ }
+
+ /**
+ * 将已成功写入死信的触发器对应实例收敛为失败终态。
+ *
+ * 状态更新使用实例锁 fencing token，避免旧节点在锁失效后覆盖新 owner 的
+ * 业务终态；终态实例保持原状态。
+ *
+ * @param trigger 已死信触发器
+ * @param failure 最后一次失败
+ */
+ private boolean failDeadLetteredTrigger(
+ Trigger trigger, Throwable failure) {
+ if (trigger == null
+ || trigger.getStateInstanceId() == null) {
+ return true;
+ }
+ String instanceId = trigger.getStateInstanceId();
+ Throwable cause = failure == null
+ ? new ChainException(
+ "Workflow trigger delivery attempts exhausted: "
+ + trigger.getId())
+ : failure;
+ try {
+ ChainState state =
+ chainStateRepository.load(instanceId);
+ if (state == null
+ || (state.getStatus() != null
+ && state.getStatus().isTerminal())) {
+ return true;
+ }
+ ChainDefinition definition =
+ getDefinitionForInstance(state);
+ if (definition == null) {
+ definition = new ChainDefinition();
+ definition.setId(state.getChainDefinitionId());
+ }
+ Chain chain = configureChain(
+ definition, instanceId);
+ return chain.failTerminal(cause);
+ } catch (Throwable terminalError) {
+ log.error(
+ "Failed to mark dead-lettered workflow terminal, "
+ + "instanceId={}, triggerId={}",
+ instanceId, trigger.getId(), terminalError);
+ return false;
+ }
}
/**
@@ -130,19 +337,28 @@ public class ChainExecutor {
String stateInstanceId = chain.getStateInstanceId();
CompletableFuture> future = pendingExecutions.get(stateInstanceId);
- if (future == null) {
- return;
- }
-
try {
- ChainState state = chainStateRepository.load(stateInstanceId);
- if (state == null) {
- throw new ChainException("Chain state not found: " + stateInstanceId);
+ if (future != null) {
+ ChainState state = chainStateRepository.load(stateInstanceId);
+ if (state == null) {
+ throw new ChainException("Chain state not found: " + stateInstanceId);
+ }
+ @SuppressWarnings("unchecked")
+ Map execResult = (Map)
+ loopResultRepository.resolveReferences(state.getExecuteResult());
+ future.complete(execResult != null ? execResult : Collections.emptyMap());
}
- Map execResult = state.getExecuteResult();
- future.complete(execResult != null ? execResult : Collections.emptyMap());
} catch (Exception error) {
- future.completeExceptionally(error);
+ if (future != null) {
+ future.completeExceptionally(error);
+ }
+ log.error(
+ "Failed to complete workflow execution continuation, instanceId={}",
+ stateInstanceId,
+ error);
+ } finally {
+ activeDefinitions.remove(stateInstanceId);
+ definitionSnapshotRepository.remove(stateInstanceId);
}
}
@@ -153,10 +369,16 @@ public class ChainExecutor {
* @param chain 发生异常的工作流实例
*/
private void failPendingExecution(Throwable error, Chain chain) {
- CompletableFuture> future = pendingExecutions.get(chain.getStateInstanceId());
+ String stateInstanceId = chain.getStateInstanceId();
+ CompletableFuture> future = pendingExecutions.get(stateInstanceId);
if (future != null) {
future.completeExceptionally(error);
}
+ activeDefinitions.remove(stateInstanceId);
+ ChainState state = chainStateRepository.load(stateInstanceId);
+ if (state != null && state.getStatus().isTerminal()) {
+ definitionSnapshotRepository.remove(stateInstanceId);
+ }
}
/**
@@ -176,8 +398,416 @@ public class ChainExecutor {
public String executeAsync(String definitionId, Map variables) {
Chain chain = createChain(definitionId);
- chain.start(variables);
- return chain.getStateInstanceId();
+ try {
+ chain.start(variables);
+ return chain.getStateInstanceId();
+ } catch (RuntimeException | Error error) {
+ activeDefinitions.remove(chain.getStateInstanceId());
+ definitionSnapshotRepository.remove(chain.getStateInstanceId());
+ throw error;
+ }
+ }
+
+ /**
+ * 在独立触发执行通道中同步执行子工作流。
+ *
+ * 调用者继续获得与历史实现一致的同步结果；子流程自身的节点触发器在独立
+ * worker lane 执行，因此父节点等待不会占满子流程所需的普通节点工作线程。
+ * 根级子流程并发受宽松许可保护，防止异常调用一次创建过多等待线程。
+ *
+ * @param definitionId 子流程定义 ID
+ * @param variables 子流程输入
+ * @param parentChain 父流程
+ * @param parentNodeId 父工作流节点 ID
+ * @return 子流程输出
+ */
+ public Map executeChild(
+ String definitionId,
+ Map variables,
+ Chain parentChain,
+ String parentNodeId) {
+ Objects.requireNonNull(parentChain, "parentChain required");
+ if (parentNodeId == null || parentNodeId.isBlank()) {
+ throw new IllegalArgumentException("parentNodeId required");
+ }
+ ChainState parentState = chainStateRepository.load(
+ parentChain.getStateInstanceId());
+ int parentDepth = readNestedDepth(parentState);
+ int childDepth = parentDepth + 1;
+ executionBudget.checkNestedDepth(parentNodeId, childDepth);
+ if (childDepth > maxChildExecutionLaneDepth) {
+ throw new ExecutionBudgetExceededException(
+ "Child workflow depth "
+ + childDepth
+ + " exceeds registered execution lanes "
+ + maxChildExecutionLaneDepth);
+ }
+ List callPath = readDefinitionCallPath(
+ parentState, parentChain.getDefinition().getId());
+ String canonicalChildId = canonicalDefinitionId(definitionId);
+ if (callPath.contains(canonicalChildId)) {
+ throw new ExecutionBudgetExceededException(
+ "Recursive workflow call detected at "
+ + parentNodeId
+ + ": "
+ + String.join(" -> ", callPath)
+ + " -> "
+ + canonicalChildId);
+ }
+ List childCallPath = new ArrayList<>(callPath);
+ childCallPath.add(canonicalChildId);
+ Trigger currentTrigger = TriggerContext.getCurrentTrigger();
+ String invocationId = currentTrigger == null
+ ? "direct:" + UUID.randomUUID()
+ : (currentTrigger.getLogicalExecutionId() == null
+ || currentTrigger.getLogicalExecutionId().isBlank()
+ ? currentTrigger.getId()
+ : currentTrigger.getLogicalExecutionId());
+ boolean rootPermit = false;
+ try {
+ if (parentDepth == 0) {
+ if (!rootChildExecutionPermits.tryAcquire(
+ 1L, TimeUnit.SECONDS)) {
+ throw new RetryableTriggerException(
+ "子工作流并发繁忙，请稍后重试", null);
+ }
+ rootPermit = true;
+ }
+ ChildExecutionReference reference =
+ prepareChildExecution(
+ definitionId,
+ parentChain,
+ parentNodeId,
+ invocationId,
+ childDepth,
+ childCallPath);
+ startChildIfReady(reference, variables);
+ Map result = awaitPersistentOutcome(
+ reference.childInstanceId(),
+ Long.MAX_VALUE,
+ TimeUnit.SECONDS,
+ parentChain);
+ markChildExecutionCompleted(
+ parentChain, parentNodeId, reference);
+ clearDefaultStates(result);
+ return result;
+ } catch (InterruptedException error) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Child workflow execution interrupted", error);
+ } catch (TimeoutException impossible) {
+ throw new IllegalStateException(
+ "Unexpected child workflow timeout", impossible);
+ } finally {
+ if (rootPermit) {
+ rootChildExecutionPermits.release();
+ }
+ }
+ }
+
+ /**
+ * 在父实例短锁内复用或创建持久子流程关联。
+ *
+ * @param definitionId 子流程定义 ID
+ * @param parentChain 父流程
+ * @param parentNodeId 父节点 ID
+ * @param invocationId 父节点逻辑执行 ID
+ * @param childDepth 子流程综合深度
+ * @param childCallPath 子流程调用链
+ * @return 持久子流程关联
+ */
+ private ChildExecutionReference prepareChildExecution(
+ String definitionId,
+ Chain parentChain,
+ String parentNodeId,
+ String invocationId,
+ int childDepth,
+ List childCallPath) {
+ return parentChain.executeWithLock(
+ parentChain.getStateInstanceId(),
+ 10L,
+ TimeUnit.SECONDS,
+ () -> {
+ NodeState nodeState =
+ parentChain.updateNodeStateSafely(
+ parentNodeId, state -> null);
+ Object existingValue = nodeState.getMemory().get(
+ CHILD_EXECUTION_REFERENCE_KEY);
+ if (existingValue instanceof ChildExecutionReference) {
+ ChildExecutionReference existing =
+ (ChildExecutionReference) existingValue;
+ if (Objects.equals(
+ existing.invocationId(), invocationId)) {
+ return existing;
+ }
+ ChainState existingChild = chainStateRepository.load(
+ existing.childInstanceId());
+ if (existingChild != null
+ && (existingChild.getStatus() == null
+ || !existingChild.getStatus().isTerminal())) {
+ throw new RetryableTriggerException(
+ "前一次子工作流仍在执行", null);
+ }
+ }
+ Chain child = createChain(definitionId);
+ child.updateStateSafely(state -> {
+ state.getMemory().put(
+ NESTED_DEPTH_MEMORY_KEY, childDepth);
+ state.getMemory().put(
+ DEFINITION_CALL_PATH_MEMORY_KEY,
+ new ArrayList<>(childCallPath));
+ return EnumSet.of(ChainStateField.MEMORY);
+ });
+ ChildExecutionReference created =
+ new ChildExecutionReference(
+ invocationId,
+ child.getStateInstanceId(),
+ childDepth,
+ false);
+ parentChain.updateNodeStateSafely(
+ parentNodeId,
+ state -> {
+ state.getMemory().put(
+ CHILD_EXECUTION_REFERENCE_KEY,
+ created);
+ return EnumSet.of(NodeStateField.MEMORY);
+ });
+ return created;
+ });
+ }
+
+ /**
+ * 在子实例仍为 READY 时幂等启动。
+ *
+ * @param reference 子流程关联
+ * @param variables 子流程输入
+ */
+ private void startChildIfReady(
+ ChildExecutionReference reference,
+ Map variables) {
+ ChainState childState = chainStateRepository.load(
+ reference.childInstanceId());
+ if (childState == null) {
+ throw new ChainException(
+ "Child chain state not found: "
+ + reference.childInstanceId());
+ }
+ ChainDefinition definition =
+ getDefinitionForInstance(childState);
+ if (definition == null) {
+ throw new ChainException(
+ "Child chain definition not found: "
+ + reference.childInstanceId());
+ }
+ Chain child = configureChain(
+ definition, reference.childInstanceId());
+ child.setExecutionLane(childExecutionLane(reference.depth()));
+ child.setNestedDepthBase(reference.depth());
+ child.executeWithLock(
+ reference.childInstanceId(),
+ 10L,
+ TimeUnit.SECONDS,
+ () -> {
+ ChainState latest = chainStateRepository.load(
+ reference.childInstanceId());
+ if (latest != null
+ && (latest.getStatus() == ChainStatus.READY
+ || latest.getStatus()
+ == ChainStatus.RUNNING)) {
+ child.start(variables);
+ }
+ return null;
+ });
+ }
+
+ /**
+ * 标记当前父节点关联已观察到子流程终态。
+ *
+ * @param parentChain 父流程
+ * @param parentNodeId 父节点 ID
+ * @param reference 子流程关联
+ */
+ private void markChildExecutionCompleted(
+ Chain parentChain,
+ String parentNodeId,
+ ChildExecutionReference reference) {
+ parentChain.updateNodeStateSafely(parentNodeId, state -> {
+ Object current = state.getMemory().get(
+ CHILD_EXECUTION_REFERENCE_KEY);
+ if (!(current instanceof ChildExecutionReference)
+ || !Objects.equals(
+ ((ChildExecutionReference) current).invocationId(),
+ reference.invocationId())) {
+ return null;
+ }
+ state.getMemory().put(
+ CHILD_EXECUTION_REFERENCE_KEY,
+ new ChildExecutionReference(
+ reference.invocationId(),
+ reference.childInstanceId(),
+ reference.depth(),
+ true));
+ return EnumSet.of(NodeStateField.MEMORY);
+ });
+ }
+
+ /**
+ * 按综合嵌套深度生成独立子流程通道。
+ *
+ * @param depth 综合嵌套深度
+ * @return 通道名
+ */
+ public static String childExecutionLane(int depth) {
+ return CHILD_WORKFLOW_EXECUTION_LANE_PREFIX
+ + Math.max(1, depth);
+ }
+
+ /**
+ * 读取实例持久化的子工作流嵌套深度。
+ *
+ * @param state 工作流状态
+ * @return 非负嵌套深度
+ */
+ private int readNestedDepth(ChainState state) {
+ if (state == null) {
+ return 0;
+ }
+ Object value = state.getMemory().get(NESTED_DEPTH_MEMORY_KEY);
+ return value instanceof Number
+ ? Math.max(0, ((Number) value).intValue())
+ : 0;
+ }
+
+ /**
+ * 读取并规范化实例的工作流定义调用链。
+ *
+ * @param state 工作流状态
+ * @param currentDefinitionId 当前定义 ID
+ * @return 从根定义到当前定义的调用链
+ */
+ private List readDefinitionCallPath(
+ ChainState state, String currentDefinitionId) {
+ List path = new ArrayList<>();
+ Object value = state == null
+ ? null
+ : state.getMemory().get(DEFINITION_CALL_PATH_MEMORY_KEY);
+ if (value instanceof Collection) {
+ for (Object item : (Collection) value) {
+ if (item != null) {
+ path.add(canonicalDefinitionId(String.valueOf(item)));
+ }
+ }
+ }
+ if (path.isEmpty() && currentDefinitionId != null) {
+ path.add(canonicalDefinitionId(currentDefinitionId));
+ }
+ return path;
+ }
+
+ /**
+ * 将发布态和草稿态的同一工作流规范化为统一调用身份。
+ *
+ * @param definitionId 定义 ID
+ * @return 规范化定义 ID
+ */
+ private String canonicalDefinitionId(String definitionId) {
+ if (definitionId == null) {
+ return "";
+ }
+ String normalized = definitionId.trim();
+ return normalized.startsWith("published:")
+ ? normalized.substring("published:".length())
+ : normalized;
+ }
+
+ /**
+ * 轮询持久状态等待工作流终态，允许任意集群实例执行实际触发器。
+ *
+ * @param stateInstanceId 工作流实例 ID
+ * @param timeout 超时数值
+ * @param unit 超时单位
+ * @return 已解析业务结果
+ * @throws InterruptedException 等待线程被中断
+ * @throws TimeoutException 超时
+ */
+ @SuppressWarnings("unchecked")
+ private Map awaitPersistentOutcome(
+ String stateInstanceId,
+ long timeout,
+ TimeUnit unit,
+ Chain parentChain)
+ throws InterruptedException, TimeoutException {
+ Objects.requireNonNull(unit, "time unit required");
+ long timeoutNanos = timeout == Long.MAX_VALUE
+ ? Long.MAX_VALUE
+ : Math.max(0L, unit.toNanos(timeout));
+ long startedAt = System.nanoTime();
+ while (true) {
+ if (parentChain != null) {
+ ChainState parentState = chainStateRepository.load(
+ parentChain.getStateInstanceId());
+ if (parentState == null
+ || (parentState.getStatus() != null
+ && parentState.getStatus().isTerminal())) {
+ cancel(
+ stateInstanceId,
+ "Parent workflow is no longer running");
+ throw new ChainException(
+ "Parent workflow ended while child was running");
+ }
+ Trigger owner = TriggerContext.getCurrentTrigger();
+ if (owner != null) {
+ // 认领丢失只终止旧 owner 的等待；durable child 留给新 owner 复用。
+ triggerScheduler.assertClaimOwned(owner);
+ }
+ }
+ ChainState state = chainStateRepository.load(stateInstanceId);
+ if (state == null) {
+ throw new ChainException(
+ "Chain state not found: " + stateInstanceId);
+ }
+ ChainStatus status = state.getStatus();
+ if (status != null && status.isTerminal()) {
+ if (!status.isSuccess()) {
+ ExceptionSummary error = state.getError();
+ throw new ChainException(
+ error == null
+ ? "Workflow ended with status " + status
+ : error.getMessage());
+ }
+ Map result =
+ (Map)
+ loopResultRepository.resolveReferences(
+ state.getExecuteResult());
+ return result == null
+ ? Collections.emptyMap()
+ : result;
+ }
+ if (timeoutNanos != Long.MAX_VALUE
+ && System.nanoTime() - startedAt >= timeoutNanos) {
+ throw new TimeoutException(
+ "Workflow execution timed out: "
+ + stateInstanceId);
+ }
+ Thread.sleep(persistentOutcomePollMillis);
+ }
+ }
+
+ /**
+ * 父工作流节点与子实例之间的持久关联。
+ *
+ * @param invocationId 父节点逻辑执行 ID
+ * @param childInstanceId 子流程实例 ID
+ * @param depth 子流程综合嵌套深度
+ * @param completed 是否已观察到终态
+ */
+ private record ChildExecutionReference(
+ String invocationId,
+ String childInstanceId,
+ int depth,
+ boolean completed) implements Serializable {
+
+ private static final long serialVersionUID = 1L;
}
@@ -192,15 +822,21 @@ public class ChainExecutor {
public Map executeNode(String definitionId, String nodeId, Map variables) {
ChainDefinition chainDefinitionById = definitionRepository.getChainDefinitionById(definitionId);
Node node = chainDefinitionById.getNodeById(nodeId);
- Chain temp = createChain(definitionId);
- if (variables != null && !variables.isEmpty()) {
- temp.updateStateSafely(s -> {
- s.getMemory().putAll(variables);
- temp.applyStartParameterAliases(s.getMemory(), variables);
- return EnumSet.of(ChainStateField.MEMORY);
- });
+ Chain temp = createChain(chainDefinitionById);
+ try {
+ temp.initializeState();
+ if (variables != null && !variables.isEmpty()) {
+ temp.updateStateSafely(s -> {
+ s.getMemory().putAll(variables);
+ temp.applyStartParameterAliases(s.getMemory(), variables);
+ return EnumSet.of(ChainStateField.MEMORY);
+ });
+ }
+ return node.execute(temp);
+ } finally {
+ activeDefinitions.remove(temp.getStateInstanceId());
+ definitionSnapshotRepository.remove(temp.getStateInstanceId());
}
- return node.execute(temp);
}
@@ -229,17 +865,12 @@ public class ChainExecutor {
return;
}
- ChainDefinition definition = definitionRepository.getChainDefinitionById(state.getChainDefinitionId());
+ ChainDefinition definition = getDefinitionForInstance(state);
if (definition == null) {
return;
}
- Chain chain = new Chain(definition, state.getInstanceId());
- chain.setTriggerScheduler(triggerScheduler);
- chain.setChainStateRepository(chainStateRepository);
- chain.setNodeStateRepository(nodeStateRepository);
- chain.setEventManager(eventManager);
-
+ Chain chain = configureChain(definition, state.getInstanceId());
chain.resume(variables);
}
@@ -249,44 +880,132 @@ public class ChainExecutor {
if (definition == null) {
throw new RuntimeException("Chain definition not found");
}
+ return createChain(definition);
+ }
+ /**
+ * 使用已加载的定义创建工作流实例，避免同一次调用重复读取定义。
+ *
+ * @param definition 已加载的工作流定义
+ * @return 已完成运行时依赖配置的工作流实例
+ */
+ private Chain createChain(ChainDefinition definition) {
String stateInstanceId = UUID.randomUUID().toString();
+ activeDefinitions.put(stateInstanceId, definition);
+ try {
+ definitionSnapshotRepository.save(stateInstanceId, definition);
+ Chain chain = configureChain(definition, stateInstanceId);
+ chain.initializeState();
+ return chain;
+ } catch (RuntimeException | Error error) {
+ activeDefinitions.remove(stateInstanceId);
+ try {
+ definitionSnapshotRepository.remove(stateInstanceId);
+ } catch (RuntimeException cleanupError) {
+ error.addSuppressed(cleanupError);
+ }
+ throw error;
+ }
+ }
+
+ /**
+ * 为工作流实例配置共享运行时依赖。
+ *
+ * @param definition 工作流定义
+ * @param stateInstanceId 状态实例 ID
+ * @return 配置完成的工作流实例
+ */
+ private Chain configureChain(ChainDefinition definition, String stateInstanceId) {
+ return configureChain(definition, stateInstanceId, null);
+ }
+
+ /**
+ * 为工作流实例配置共享运行时依赖，并复用调用方已经加载的状态。
+ *
+ * @param definition 工作流定义
+ * @param stateInstanceId 状态实例 ID
+ * @param persistedState 已加载状态；为 {@code null} 时按需读取
+ * @return 配置完成的工作流实例
+ */
+ private Chain configureChain(
+ ChainDefinition definition,
+ String stateInstanceId,
+ ChainState persistedState) {
Chain chain = new Chain(definition, stateInstanceId);
chain.setTriggerScheduler(triggerScheduler);
chain.setChainStateRepository(chainStateRepository);
chain.setNodeStateRepository(nodeStateRepository);
+ chain.setLoopResultRepository(loopResultRepository);
chain.setEventManager(eventManager);
-
+ chain.setExecutionBudget(executionBudget);
+ ChainState state = persistedState == null
+ ? chainStateRepository.load(stateInstanceId)
+ : persistedState;
+ int nestedDepth = readNestedDepth(state);
+ chain.setNestedDepthBase(nestedDepth);
+ if (nestedDepth > 0) {
+ chain.setExecutionLane(childExecutionLane(nestedDepth));
+ }
return chain;
}
+ /**
+ * 获取工作流实例启动时使用的定义快照。
+ *
+ * @param state 工作流状态
+ * @return 活跃定义快照；当前实例首次由本节点接管时从仓储加载
+ */
+ private ChainDefinition getDefinitionForInstance(ChainState state) {
+ String stateInstanceId = state.getInstanceId();
+ ChainDefinition definition = activeDefinitions.get(stateInstanceId);
+ if (definition != null) {
+ return definition;
+ }
+ ChainDefinition loaded = definitionSnapshotRepository.load(stateInstanceId);
+ if (loaded == null) {
+ // 兼容升级前已经启动、尚未持久化定义快照的实例。
+ loaded = definitionRepository.getChainDefinitionById(state.getChainDefinitionId());
+ }
+ if (loaded == null) {
+ return null;
+ }
+ synchronized (activeDefinitions) {
+ ChainDefinition existing = activeDefinitions.get(stateInstanceId);
+ if (existing != null) {
+ return existing;
+ }
+ activeDefinitions.put(stateInstanceId, loaded);
+ return loaded;
+ }
+ }
+
private void accept(Trigger trigger, ExecutorService worker) {
ChainState state = chainStateRepository.load(trigger.getStateInstanceId());
if (state == null) {
- throw new ChainException("Chain state not found");
+ // 状态已过期或被清理时，该触发器已经失去业务目标，直接确认避免无限热重放。
+ return;
}
- ChainDefinition definition = definitionRepository.getChainDefinitionById(state.getChainDefinitionId());
+ ChainDefinition definition = getDefinitionForInstance(state);
if (definition == null) {
- throw new ChainException("Chain definition not found");
+ throw new NonRetryableTriggerException(
+ "Chain definition not found: " + state.getChainDefinitionId());
}
- Chain chain = new Chain(definition, trigger.getStateInstanceId());
- chain.setTriggerScheduler(triggerScheduler);
- chain.setChainStateRepository(chainStateRepository);
- chain.setNodeStateRepository(nodeStateRepository);
- chain.setEventManager(eventManager);
+ Chain chain = configureChain(
+ definition, trigger.getStateInstanceId(), state);
String nodeId = trigger.getNodeId();
if (nodeId == null) {
- throw new ChainException("Node ID not found in trigger.");
+ throw new NonRetryableTriggerException("Node ID not found in trigger.");
}
Node node = definition.getNodeById(nodeId);
if (node == null) {
- throw new ChainException("Node not found in definition(id: " + definition.getId() + ")");
+ throw new NonRetryableTriggerException(
+ "Node not found in definition(id: " + definition.getId() + ")");
}
chain.executeNode(node, trigger);
@@ -345,6 +1064,16 @@ public class ChainExecutor {
return triggerScheduler;
}
+ /**
+ * 在查询/API 边界透明还原循环结果引用。
+ *
+ * @param value 可能包含引用的值
+ * @return 业务可见值
+ */
+ public Object resolveResultReferences(Object value) {
+ return loopResultRepository.resolveReferences(value);
+ }
+
public EventManager getEventManager() {
return eventManager;
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudget.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudget.java
new file mode 100644
index 0000000..a48042e
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudget.java
@@ -0,0 +1,188 @@
+package com.easyagents.flow.core.chain.runtime;
+
+import java.io.Serializable;
+
+/**
+ * 工作流执行的全局资源保护预算。
+ *
+ * 所有默认值均为宽松的失控保护值。小于等于 {@code 0} 的配置表示关闭对应保护，
+ * 节点自身的循环次数、退出条件和重试配置仍按原有语义优先生效。
+ */
+public final class ExecutionBudget implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+
+ public static final long DEFAULT_MAX_ITERATIONS = 100_000L;
+ /**
+ * 缺省不限制墙钟时长，避免人工确认或长期挂起时间被误计为执行耗时。
+ */
+ public static final long DEFAULT_MAX_DURATION_MILLIS = 0L;
+ public static final long DEFAULT_MAX_CHILD_EXECUTIONS = 1_000_000L;
+ public static final long DEFAULT_MAX_ACCUMULATED_BYTES = 512L * 1024L * 1024L;
+ public static final int DEFAULT_MAX_NESTED_DEPTH = 32;
+ /**
+ * 热状态硬限制缺省关闭。循环历史已通过引用隔离，开启限制时由部署方按实际负载设置，
+ * 避免估算误差改变既有业务语义。
+ */
+ public static final long DEFAULT_MAX_HOT_STATE_BYTES = 0L;
+
+ private final long maxIterations;
+ private final long maxDurationMillis;
+ private final long maxChildExecutions;
+ private final long maxAccumulatedBytes;
+ private final int maxNestedDepth;
+ private final long maxHotStateBytes;
+
+ /**
+ * 创建执行预算。
+ *
+ * @param maxIterations 单循环最大迭代次数
+ * @param maxDurationMillis 单实例最大运行毫秒数
+ * @param maxChildExecutions 单实例最大节点执行次数
+ * @param maxAccumulatedBytes 单循环最大累计结果字节数
+ * @param maxNestedDepth 最大循环嵌套深度
+ * @param maxHotStateBytes 单实例热状态建议最大字节数
+ */
+ public ExecutionBudget(long maxIterations,
+ long maxDurationMillis,
+ long maxChildExecutions,
+ long maxAccumulatedBytes,
+ int maxNestedDepth,
+ long maxHotStateBytes) {
+ this.maxIterations = maxIterations;
+ this.maxDurationMillis = maxDurationMillis;
+ this.maxChildExecutions = maxChildExecutions;
+ this.maxAccumulatedBytes = maxAccumulatedBytes;
+ this.maxNestedDepth = maxNestedDepth;
+ this.maxHotStateBytes = maxHotStateBytes;
+ }
+
+ /**
+ * 创建使用宽松缺省值的执行预算。
+ *
+ * @return 默认执行预算
+ */
+ public static ExecutionBudget defaults() {
+ return new ExecutionBudget(
+ DEFAULT_MAX_ITERATIONS,
+ DEFAULT_MAX_DURATION_MILLIS,
+ DEFAULT_MAX_CHILD_EXECUTIONS,
+ DEFAULT_MAX_ACCUMULATED_BYTES,
+ DEFAULT_MAX_NESTED_DEPTH,
+ DEFAULT_MAX_HOT_STATE_BYTES);
+ }
+
+ public long getMaxIterations() {
+ return maxIterations;
+ }
+
+ public long getMaxDurationMillis() {
+ return maxDurationMillis;
+ }
+
+ public long getMaxChildExecutions() {
+ return maxChildExecutions;
+ }
+
+ public long getMaxAccumulatedBytes() {
+ return maxAccumulatedBytes;
+ }
+
+ public int getMaxNestedDepth() {
+ return maxNestedDepth;
+ }
+
+ public long getMaxHotStateBytes() {
+ return maxHotStateBytes;
+ }
+
+ /**
+ * 校验循环迭代总数。
+ *
+ * @param nodeId 循环节点 ID
+ * @param iterations 计划迭代次数
+ * @throws ExecutionBudgetExceededException 超过启用的迭代预算时抛出
+ */
+ public void checkIterations(String nodeId, long iterations) {
+ if (maxIterations > 0 && iterations > maxIterations) {
+ throw new ExecutionBudgetExceededException(
+ "Loop iteration budget exceeded for node " + nodeId
+ + ": " + iterations + " > " + maxIterations);
+ }
+ }
+
+ /**
+ * 校验循环嵌套深度。
+ *
+ * @param nodeId 节点 ID
+ * @param depth 当前深度
+ * @throws ExecutionBudgetExceededException 超过启用的深度预算时抛出
+ */
+ public void checkNestedDepth(String nodeId, int depth) {
+ if (maxNestedDepth > 0 && depth > maxNestedDepth) {
+ throw new ExecutionBudgetExceededException(
+ "Loop nested depth budget exceeded for node " + nodeId
+ + ": " + depth + " > " + maxNestedDepth);
+ }
+ }
+
+ /**
+ * 校验循环累计结果大小。
+ *
+ * @param nodeId 循环节点 ID
+ * @param accumulatedBytes 当前累计估算字节数
+ * @throws ExecutionBudgetExceededException 超过启用的累计结果预算时抛出
+ */
+ public void checkAccumulatedBytes(String nodeId, long accumulatedBytes) {
+ if (maxAccumulatedBytes > 0 && accumulatedBytes > maxAccumulatedBytes) {
+ throw new ExecutionBudgetExceededException(
+ "Loop accumulated result budget exceeded for node " + nodeId
+ + ": " + accumulatedBytes + " > " + maxAccumulatedBytes);
+ }
+ }
+
+ /**
+ * 校验单实例节点执行次数。
+ *
+ * @param executions 当前节点执行次数
+ * @throws ExecutionBudgetExceededException 超过启用的执行预算时抛出
+ */
+ public void checkChildExecutions(long executions) {
+ if (maxChildExecutions > 0 && executions > maxChildExecutions) {
+ throw new ExecutionBudgetExceededException(
+ "Workflow child execution budget exceeded: "
+ + executions + " > " + maxChildExecutions);
+ }
+ }
+
+ /**
+ * 校验单实例运行时长。
+ *
+ * @param startedAtMillis 实例开始时间
+ * @param nowMillis 当前时间
+ * @throws ExecutionBudgetExceededException 超过启用的时长预算时抛出
+ */
+ public void checkDuration(long startedAtMillis, long nowMillis) {
+ if (maxDurationMillis > 0
+ && startedAtMillis > 0
+ && nowMillis - startedAtMillis > maxDurationMillis) {
+ throw new ExecutionBudgetExceededException(
+ "Workflow duration budget exceeded: "
+ + (nowMillis - startedAtMillis) + "ms > " + maxDurationMillis + "ms");
+ }
+ }
+
+ /**
+ * 校验工作流热状态估算大小。
+ *
+ * @param estimatedBytes 当前热状态估算字节数
+ * @throws ExecutionBudgetExceededException 超过启用的热状态预算时抛出
+ */
+ public void checkHotStateBytes(long estimatedBytes) {
+ if (maxHotStateBytes > 0 && estimatedBytes > maxHotStateBytes) {
+ throw new ExecutionBudgetExceededException(
+ "Workflow hot state budget exceeded: "
+ + estimatedBytes + " > " + maxHotStateBytes);
+ }
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudgetExceededException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudgetExceededException.java
new file mode 100644
index 0000000..44d5ef7
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ExecutionBudgetExceededException.java
@@ -0,0 +1,18 @@
+package com.easyagents.flow.core.chain.runtime;
+
+import com.easyagents.flow.core.chain.ChainException;
+
+/**
+ * 工作流实例超过平台资源保护预算时抛出的异常。
+ */
+public class ExecutionBudgetExceededException extends ChainException {
+
+ /**
+ * 创建预算超限异常。
+ *
+ * @param message 可审计的超限原因
+ */
+ public ExecutionBudgetExceededException(String message) {
+ super(message);
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/InMemoryTriggerStore.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/InMemoryTriggerStore.java
index 550b4d6..c7634f9 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/InMemoryTriggerStore.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/InMemoryTriggerStore.java
@@ -17,13 +17,16 @@ package com.easyagents.flow.core.chain.runtime;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
public class InMemoryTriggerStore implements TriggerStore {
private final ConcurrentHashMap store = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap fencingTokens = new ConcurrentHashMap<>();
@Override
public Trigger save(Trigger trigger) {
@@ -34,6 +37,18 @@ public class InMemoryTriggerStore implements TriggerStore {
return trigger;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean saveIfAbsent(Trigger trigger) {
+ if (trigger.getId() == null || trigger.getId().isBlank()) {
+ throw new IllegalArgumentException("Stable trigger ID required");
+ }
+ return store.putIfAbsent(
+ trigger.getId(), trigger) == null;
+ }
+
@Override
public boolean remove(String triggerId) {
return store.remove(triggerId) != null;
@@ -46,12 +61,35 @@ public class InMemoryTriggerStore implements TriggerStore {
@Override
public List findDue(long uptoTimestamp) {
- return null;
+ List due = new ArrayList<>();
+ for (Trigger trigger : store.values()) {
+ if (trigger.getTriggerAt() <= uptoTimestamp) {
+ due.add(trigger);
+ }
+ }
+ due.sort(Comparator.comparingLong(Trigger::getTriggerAt));
+ return due;
}
@Override
public List findAllPending() {
return new ArrayList<>(store.values());
}
-}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Trigger claim(String triggerId, long leaseMillis) {
+ Trigger trigger = store.remove(triggerId);
+ if (trigger != null) {
+ String fencingScope = trigger.getStateInstanceId() == null
+ ? "__trigger__:" + trigger.getId()
+ : trigger.getStateInstanceId();
+ trigger.setFencingToken(fencingTokens
+ .computeIfAbsent(fencingScope, ignored -> new AtomicLong())
+ .incrementAndGet());
+ }
+ return trigger;
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/NonRetryableTriggerException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/NonRetryableTriggerException.java
new file mode 100644
index 0000000..2fab07d
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/NonRetryableTriggerException.java
@@ -0,0 +1,16 @@
+package com.easyagents.flow.core.chain.runtime;
+
+/**
+ * 表示触发器内容已经无法继续执行，应进入死信而非无限重放。
+ */
+public class NonRetryableTriggerException extends RuntimeException {
+
+ /**
+ * 创建不可重试触发器异常。
+ *
+ * @param message 异常说明
+ */
+ public NonRetryableTriggerException(String message) {
+ super(message);
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/RetryableTriggerException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/RetryableTriggerException.java
new file mode 100644
index 0000000..b792710
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/RetryableTriggerException.java
@@ -0,0 +1,17 @@
+package com.easyagents.flow.core.chain.runtime;
+
+/**
+ * 表示当前节点遇到短暂基础设施冲突，应重新投递同一触发器且不消耗业务重试次数。
+ */
+public class RetryableTriggerException extends RuntimeException {
+
+ /**
+ * 创建可重新投递异常。
+ *
+ * @param message 异常说明
+ * @param cause 原始异常
+ */
+ public RetryableTriggerException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/Trigger.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/Trigger.java
index 702e97c..dae7b08 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/Trigger.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/Trigger.java
@@ -16,14 +16,65 @@
package com.easyagents.flow.core.chain.runtime;
import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
public class Trigger implements Serializable {
+ private static final long serialVersionUID = 3165037658498721088L;
+
private String id;
private String stateInstanceId;
private String edgeId;
private String nodeId; // 可以为 null，代表触发整个 chain
private TriggerType type;
private long triggerAt; // epoch ms
+ /**
+ * 当前运行时分配的触发器认领代际。
+ *
+ * 字段名为兼容既有序列化数据保留。分布式仓储在触发器认领成功时分配，
+ * 并与该触发器租约共同续期和失效；该值不代表实例锁 fencing token。
+ */
+ private long fencingToken;
+ /**
+ * 创建派生触发器时必须仍然有效的父触发器 fencing token。
+ */
+ private long requiredFencingToken;
+ /**
+ * 创建派生触发器时必须仍然有效的父实例锁 fencing token。
+ */
+ private long requiredLockFencingToken;
+ /**
+ * 创建派生触发器时必须仍然有效的父触发器 claim ID。
+ */
+ private String requiredFencingClaimId;
+ /**
+ * 基础设施投递失败次数，不占用业务节点重试次数。
+ */
+ private int deliveryAttempt;
+ /**
+ * 已完成业务终态收敛、等待可靠写入死信的标记。
+ */
+ private boolean deadLetterPending;
+ /**
+ * 待写入死信的稳定失败原因。
+ */
+ private String deadLetterReason;
+ /**
+ * 跨重试保持不变的逻辑执行 ID，用于副作用幂等键。
+ */
+ private String logicalExecutionId;
+ /**
+ * 可选执行通道，用于把会同步等待的子工作流与普通节点工作线程隔离。
+ */
+ private String executionLane;
+ /**
+ * 首个稳定入口意图携带的初始变量。
+ *
+ * 仅用于实例仍处于 READY 时的崩溃恢复；正常启动提交后，运行时变量仍以
+ * {@code ChainState.memory} 为唯一业务数据源。
+ */
+ private Map startVariables;
+ private Map loopCursors;
public Trigger() {
}
@@ -77,6 +128,260 @@ public class Trigger implements Serializable {
this.triggerAt = triggerAt;
}
+ /**
+ * 获取当前节点逻辑执行 ID。
+ *
+ * @return 跨重试保持不变的逻辑执行 ID
+ */
+ public String getLogicalExecutionId() {
+ return logicalExecutionId;
+ }
+
+ /**
+ * 设置当前节点逻辑执行 ID。
+ *
+ * @param logicalExecutionId 跨重试保持不变的逻辑执行 ID
+ */
+ public void setLogicalExecutionId(String logicalExecutionId) {
+ this.logicalExecutionId = logicalExecutionId;
+ }
+
+ /**
+ * 获取执行通道。
+ *
+ * @return 通道名；{@code null} 表示默认通道
+ */
+ public String getExecutionLane() {
+ return executionLane;
+ }
+
+ /**
+ * 设置执行通道。
+ *
+ * @param executionLane 通道名
+ */
+ public void setExecutionLane(String executionLane) {
+ this.executionLane = executionLane;
+ }
+
+ /**
+ * 获取崩溃恢复所需的初始变量。
+ *
+ * @return 初始变量快照；未携带时为 {@code null}
+ */
+ public Map getStartVariables() {
+ return startVariables;
+ }
+
+ /**
+ * 设置崩溃恢复所需的初始变量。
+ *
+ * @param startVariables 初始变量；仅首个稳定入口意图需要携带
+ */
+ public void setStartVariables(Map startVariables) {
+ this.startVariables = startVariables == null
+ ? null
+ : new LinkedHashMap<>(startVariables);
+ }
+
+ /**
+ * 获取本次认领代际。
+ *
+ * @return 单触发器认领代际；未认领时为 {@code 0}
+ */
+ public long getFencingToken() {
+ return fencingToken;
+ }
+
+ /**
+ * 设置本次认领代际。
+ *
+ * @param fencingToken 单触发器认领代际
+ */
+ public void setFencingToken(long fencingToken) {
+ this.fencingToken = fencingToken;
+ }
+
+ /**
+ * 获取保存派生触发器所依赖的父实例锁 fencing token。
+ *
+ * @return 父实例锁 token；无锁约束时为 {@code 0}
+ */
+ public long getRequiredLockFencingToken() {
+ return requiredLockFencingToken;
+ }
+
+ /**
+ * 设置保存派生触发器所依赖的父实例锁 fencing token。
+ *
+ * @param requiredLockFencingToken 父实例锁 token
+ */
+ public void setRequiredLockFencingToken(long requiredLockFencingToken) {
+ this.requiredLockFencingToken = requiredLockFencingToken;
+ }
+
+ /**
+ * 获取保存派生触发器所依赖的父 fencing token。
+ *
+ * @return 父 fencing token；无父认领约束时为 {@code 0}
+ */
+ public long getRequiredFencingToken() {
+ return requiredFencingToken;
+ }
+
+ /**
+ * 设置保存派生触发器所依赖的父 fencing token。
+ *
+ * @param requiredFencingToken 父 fencing token
+ */
+ public void setRequiredFencingToken(long requiredFencingToken) {
+ this.requiredFencingToken = requiredFencingToken;
+ }
+
+ /**
+ * 获取保存派生触发器所依赖的父 claim ID。
+ *
+ * @return 父触发器 ID；无父认领约束时为 {@code null}
+ */
+ public String getRequiredFencingClaimId() {
+ return requiredFencingClaimId;
+ }
+
+ /**
+ * 设置保存派生触发器所依赖的父 claim ID。
+ *
+ * @param requiredFencingClaimId 父触发器 ID
+ */
+ public void setRequiredFencingClaimId(String requiredFencingClaimId) {
+ this.requiredFencingClaimId = requiredFencingClaimId;
+ }
+
+ /**
+ * 获取基础设施投递失败次数。
+ *
+ * @return 失败次数
+ */
+ public int getDeliveryAttempt() {
+ return deliveryAttempt;
+ }
+
+ /**
+ * 设置基础设施投递失败次数。
+ *
+ * @param deliveryAttempt 失败次数
+ */
+ public void setDeliveryAttempt(int deliveryAttempt) {
+ this.deliveryAttempt = deliveryAttempt;
+ }
+
+ /**
+ * 判断触发器是否正在补写死信终态。
+ *
+ * @return 等待死信持久化时为 {@code true}
+ */
+ public boolean isDeadLetterPending() {
+ return deadLetterPending;
+ }
+
+ /**
+ * 设置死信补写标记。
+ *
+ * @param deadLetterPending 是否等待死信持久化
+ */
+ public void setDeadLetterPending(boolean deadLetterPending) {
+ this.deadLetterPending = deadLetterPending;
+ }
+
+ /**
+ * 获取稳定死信原因。
+ *
+ * @return 死信原因
+ */
+ public String getDeadLetterReason() {
+ return deadLetterReason;
+ }
+
+ /**
+ * 设置稳定死信原因。
+ *
+ * @param deadLetterReason 死信原因
+ */
+ public void setDeadLetterReason(String deadLetterReason) {
+ this.deadLetterReason = deadLetterReason;
+ }
+
+ /**
+ * 获取触发器携带的循环代际游标。
+ *
+ * @return 循环节点 ID 到游标的映射
+ */
+ public Map getLoopCursors() {
+ if (loopCursors == null) {
+ loopCursors = new LinkedHashMap<>();
+ }
+ return loopCursors;
+ }
+
+ /**
+ * 设置循环代际游标。
+ *
+ * @param loopCursors 循环节点 ID 到游标的映射
+ */
+ public void setLoopCursors(Map loopCursors) {
+ this.loopCursors = loopCursors;
+ }
+
+ /**
+ * 循环分支代际游标，用于拒绝过期或重复的父节点回调。
+ */
+ public static class LoopCursor implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private String resultId;
+ private int iterationIndex;
+ private String branchId;
+
+ public LoopCursor() {
+ }
+
+ /**
+ * 创建循环游标。
+ *
+ * @param resultId 循环代际 ID
+ * @param iterationIndex 迭代序号
+ * @param branchId 直属分支 ID
+ */
+ public LoopCursor(String resultId, int iterationIndex, String branchId) {
+ this.resultId = resultId;
+ this.iterationIndex = iterationIndex;
+ this.branchId = branchId;
+ }
+
+ public String getResultId() {
+ return resultId;
+ }
+
+ public void setResultId(String resultId) {
+ this.resultId = resultId;
+ }
+
+ public int getIterationIndex() {
+ return iterationIndex;
+ }
+
+ public void setIterationIndex(int iterationIndex) {
+ this.iterationIndex = iterationIndex;
+ }
+
+ public String getBranchId() {
+ return branchId;
+ }
+
+ public void setBranchId(String branchId) {
+ this.branchId = branchId;
+ }
+ }
+
@Override
public String toString() {
return "Trigger{" +
@@ -89,4 +394,3 @@ public class Trigger implements Serializable {
'}';
}
}
-
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerClaimLostException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerClaimLostException.java
new file mode 100644
index 0000000..38c185c
--- /dev/null
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerClaimLostException.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ * Licensed under the GNU Lesser General Public License (LGPL), Version 3.0.
+ */
+package com.easyagents.flow.core.chain.runtime;
+
+/**
+ * 表示当前工作线程已经失去触发器租约，不再允许提交执行结果。
+ */
+public class TriggerClaimLostException extends RuntimeException {
+
+ /**
+ * 创建租约丢失异常。
+ *
+ * @param triggerId 已失去租约的触发器 ID
+ */
+ public TriggerClaimLostException(String triggerId) {
+ super("Workflow trigger claim ownership lost: " + triggerId);
+ }
+}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerScheduler.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerScheduler.java
index 7d07a03..82bff43 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerScheduler.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerScheduler.java
@@ -39,19 +39,49 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class TriggerScheduler {
private static final Logger log = LoggerFactory.getLogger(TriggerScheduler.class);
+ private static final long CLAIM_LEASE_MS = TimeUnit.MINUTES.toMillis(1);
+ private static final long CLAIM_RENEW_INTERVAL_MS = TimeUnit.SECONDS.toMillis(20);
+ private static final int MAX_DELIVERY_ATTEMPTS = 20;
+ private static final long MAX_REDELIVERY_DELAY_MS = TimeUnit.MINUTES.toMillis(1);
+ /**
+ * 本地仅缓存一批定时任务；容量溢出时持久仓储仍是恢复与补偿来源。
+ */
+ private static final int MAX_LOCAL_SCHEDULED = 1024;
+ /**
+ * 精确定时只预热宽松近期限窗口，远期任务继续由持久仓储保管。
+ */
+ private static final long MIN_LOCAL_SCHEDULE_HORIZON_MS =
+ TimeUnit.MINUTES.toMillis(1);
private final TriggerStore store;
private final ScheduledExecutorService scheduler;
private final ExecutorService worker;
+ private final Semaphore dispatchPermits;
+ private final ConcurrentMap laneWorkers =
+ new ConcurrentHashMap<>();
+ private final ConcurrentMap laneDispatchPermits =
+ new ConcurrentHashMap<>();
private final AtomicBoolean closed = new AtomicBoolean(false);
// map 用于管理取消：triggerId -> ScheduledFuture
private final ConcurrentMap> scheduledFutures = new ConcurrentHashMap<>();
+ /**
+ * 本地 Future 对应的绝对触发时间，用于容量满时保留更早到期任务。
+ */
+ private final ConcurrentMap scheduledTriggerTimes =
+ new ConcurrentHashMap<>();
+ private final ConcurrentMap> claimRenewals = new ConcurrentHashMap<>();
+ /**
+ * 串行化本地 Future 的容量检查与登记，确保并发调度时仍严格受容量上限约束。
+ */
+ private final Object localScheduleMonitor = new Object();
// consumer 来把 trigger 交给 ChainExecutor（或 ChainRuntime）去处理
private volatile TriggerConsumer consumer;
+ private volatile TriggerFailureListener failureListener;
// 周期扫查间隔（ms）
private final long scanIntervalMs;
+ private final long localScheduleHorizonMs;
// 扫描任务 future
private ScheduledFuture scanFuture;
@@ -60,13 +90,34 @@ public class TriggerScheduler {
void accept(Trigger trigger, ExecutorService worker);
}
+ /**
+ * 触发器不可恢复失败监听器。
+ */
+ public interface TriggerFailureListener {
+ /**
+ * 在触发器 claim 仍有效时通知业务运行时收敛实例终态。
+ *
+ * @param trigger 已死信触发器
+ * @param failure 最后一次执行失败
+ */
+ boolean onDeadLetter(Trigger trigger, Throwable failure);
+ }
+
public TriggerScheduler(TriggerStore store, ScheduledExecutorService scheduler, ExecutorService worker, long scanIntervalMs) {
this.store = Objects.requireNonNull(store, "TriggerStore required");
this.scheduler = Objects.requireNonNull(scheduler, "ScheduledExecutorService required");
this.worker = Objects.requireNonNull(worker, "ExecutorService required");
+ this.dispatchPermits = createDispatchPermits(worker);
this.scanIntervalMs = Math.max(1000, scanIntervalMs);
+ long scanHorizon = this.scanIntervalMs
+ > Long.MAX_VALUE / 3L
+ ? Long.MAX_VALUE
+ : this.scanIntervalMs * 3L;
+ this.localScheduleHorizonMs = Math.max(
+ MIN_LOCAL_SCHEDULE_HORIZON_MS,
+ scanHorizon);
- // 恢复并 schedule
+ // 启动时只恢复已经到期的任务，避免把全部远期任务复制到本机 DelayQueue。
recoverAndSchedulePending();
// 启动周期扫查 findDue
@@ -78,6 +129,40 @@ public class TriggerScheduler {
this.consumer = consumer;
}
+ /**
+ * 注册触发器不可恢复失败监听器。
+ *
+ * @param listener 失败监听器
+ */
+ public void registerFailureListener(TriggerFailureListener listener) {
+ this.failureListener = listener;
+ }
+
+ /**
+ * 注册独立执行通道。
+ *
+ * @param lane 通道名
+ * @param laneWorker 通道工作线程池
+ */
+ public void registerWorker(
+ String lane, ExecutorService laneWorker) {
+ if (lane == null || lane.isBlank()) {
+ throw new IllegalArgumentException("lane required");
+ }
+ ExecutorService workerToRegister =
+ Objects.requireNonNull(laneWorker, "laneWorker required");
+ ExecutorService previous =
+ laneWorkers.putIfAbsent(lane, workerToRegister);
+ if (previous != null && previous != workerToRegister) {
+ throw new IllegalStateException(
+ "Trigger worker lane already registered: " + lane);
+ }
+ Semaphore permits = createDispatchPermits(workerToRegister);
+ if (permits != null) {
+ laneDispatchPermits.putIfAbsent(lane, permits);
+ }
+ }
+
/**
* schedule a trigger: persist -> schedule (单机语义)
*/
@@ -86,20 +171,55 @@ public class TriggerScheduler {
if (trigger.getId() == null) {
trigger.setId(UUID.randomUUID().toString());
}
+ if (trigger.getLogicalExecutionId() == null || trigger.getLogicalExecutionId().isBlank()) {
+ trigger.setLogicalExecutionId(trigger.getId());
+ }
store.save(trigger);
scheduleInternal(trigger);
return trigger;
}
+ /**
+ * 仅在持久仓储中不存在同 ID 触发器时保存并调度。
+ *
+ * 调用方需持有工作流实例锁；该方法用于可重放启动协议，稳定 ID 可避免
+ * READY 到入口触发器持久化之间的崩溃窗口产生重复待执行任务。
+ *
+ * @param trigger 带稳定 ID 的触发器
+ * @return 已存在或新保存的触发器
+ */
+ public Trigger scheduleIfAbsent(Trigger trigger) {
+ if (closed.get()) {
+ throw new IllegalStateException("TriggerScheduler closed");
+ }
+ if (trigger == null || trigger.getId() == null
+ || trigger.getId().isBlank()) {
+ throw new IllegalArgumentException(
+ "Stable trigger ID required");
+ }
+ if (trigger.getLogicalExecutionId() == null
+ || trigger.getLogicalExecutionId().isBlank()) {
+ trigger.setLogicalExecutionId(
+ trigger.getId());
+ }
+ if (store.saveIfAbsent(trigger)) {
+ scheduleInternal(trigger);
+ return trigger;
+ }
+ Trigger existing = store.find(
+ trigger.getId());
+ // 已有触发器可能刚好被其他 owner 认领；稳定入口仍视为已成功登记。
+ return existing == null
+ ? trigger
+ : existing;
+ }
+
/**
* cancel trigger (从 store 删除并尝试取消已 schedule 的 future)
*/
public boolean cancel(String triggerId) {
boolean removed = store.remove(triggerId);
- ScheduledFuture f = scheduledFutures.remove(triggerId);
- if (f != null) {
- f.cancel(false);
- }
+ removeLocalSchedule(triggerId, true);
return removed;
}
@@ -107,83 +227,185 @@ public class TriggerScheduler {
* 主动触发（webhook/event/manual 场景）
*/
public boolean fire(String triggerId) {
- if (closed.get()) return false;
- Trigger t = store.find(triggerId);
- if (t == null) return false;
- if (consumer == null) {
- // 无 consumer，仍从 store 中移除
- store.remove(triggerId);
+ if (closed.get()) {
return false;
}
- // 在 worker 线程触发 consumer
- worker.submit(() -> {
- try {
- consumer.accept(t, worker);
- } catch (Exception e) {
- log.error(e.toString(), e);
- } finally {
- // 默认语义：触发后移除
- store.remove(triggerId);
- ScheduledFuture sf = scheduledFutures.remove(triggerId);
- if (sf != null) sf.cancel(false);
- }
- });
- return true;
+ Trigger candidate = store.find(triggerId);
+ if (candidate == null) {
+ return false;
+ }
+ removeLocalSchedule(triggerId, true);
+ return claimAndDispatch(candidate);
}
/**
* internal scheduling for a trigger (单机 scheduled semantics)
*/
private void scheduleInternal(Trigger trigger) {
- if (closed.get()) return;
-
- long delay = Math.max(0, trigger.getTriggerAt() - System.currentTimeMillis());
-
- // cancel any existing scheduled future for same id
- ScheduledFuture prev = scheduledFutures.remove(trigger.getId());
- if (prev != null) prev.cancel(false);
-
- ScheduledFuture future = scheduler.schedule(() -> {
- // double-check existence in store (可能已被 cancel)
- Trigger existing = store.find(trigger.getId());
- if (existing == null) {
- scheduledFutures.remove(trigger.getId());
+ if (closed.get()) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ if (trigger.getTriggerAt()
+ > scheduleHorizonTimestamp(now)) {
+ return;
+ }
+ synchronized (localScheduleMonitor) {
+ if (closed.get()) {
return;
}
-
- if (consumer != null) {
- worker.submit(() -> {
- try {
- TriggerContext.setCurrentTrigger(existing);
- consumer.accept(existing, worker);
- } catch (Throwable e) {
- log.error(e.toString(), e);
- } finally {
- TriggerContext.clearCurrentTrigger();
- store.remove(existing.getId());
- scheduledFutures.remove(existing.getId());
- }
- });
- } else {
- // 无 consumer，则移除
- store.remove(existing.getId());
- scheduledFutures.remove(existing.getId());
+ ScheduledFuture existing;
+ while ((existing = scheduledFutures.get(trigger.getId())) != null) {
+ if (!existing.isDone() && !existing.isCancelled()) {
+ return;
+ }
+ // 已完成或取消的占位必须先原子移除，否则 putIfAbsent 会永久阻断重新调度。
+ if (!scheduledFutures.remove(trigger.getId(), existing)) {
+ continue;
+ }
+ scheduledTriggerTimes.remove(
+ trigger.getId());
}
- }, delay, TimeUnit.MILLISECONDS);
+ if (!hasLocalScheduleCapacity(
+ trigger.getTriggerAt())) {
+ return;
+ }
+ long delayMillis = Math.max(
+ 0L,
+ trigger.getTriggerAt()
+ - System.currentTimeMillis());
+ ScheduledFuture future = scheduler.schedule(
+ () -> {
+ // 回调先在同一登记锁下同时移除两张索引，消除零延迟任务
+ // 在 Future 与时间索引分步登记之间完成所造成的孤儿记录。
+ removeLocalSchedule(
+ trigger.getId(), false);
+ claimAndDispatch(trigger);
+ },
+ delayMillis,
+ TimeUnit.MILLISECONDS);
+ ScheduledFuture concurrent =
+ scheduledFutures.putIfAbsent(
+ trigger.getId(), future);
+ if (concurrent != null) {
+ future.cancel(false);
+ return;
+ }
+ scheduledTriggerTimes.put(
+ trigger.getId(),
+ trigger.getTriggerAt());
+ // 零延迟任务可能在 put 前完成；完成态二次清理避免残留无效 Future。
+ if (future.isDone() || future.isCancelled()) {
+ scheduledFutures.remove(
+ trigger.getId(), future);
+ scheduledTriggerTimes.remove(
+ trigger.getId());
+ }
+ }
+ }
- scheduledFutures.put(trigger.getId(), future);
+ /**
+ * 检查本地调度容量，并在容量耗尽时清理已完成或已取消的占位。
+ *
+ * 正常路径只执行常量时间判断；达到上限时最多扫描
+ * {@link #MAX_LOCAL_SCHEDULED} 个条目，避免极窄竞态残留导致永久停摆。
+ *
+ * @param triggerAt 待登记任务的绝对触发时间
+ * @return 仍可接收本地到期任务时为 {@code true}
+ */
+ private boolean hasLocalScheduleCapacity(
+ long triggerAt) {
+ if (scheduledFutures.size() < MAX_LOCAL_SCHEDULED) {
+ return true;
+ }
+ for (Map.Entry> entry : scheduledFutures.entrySet()) {
+ ScheduledFuture future = entry.getValue();
+ if (future.isDone() || future.isCancelled()) {
+ if (scheduledFutures.remove(
+ entry.getKey(), future)) {
+ scheduledTriggerTimes.remove(
+ entry.getKey());
+ }
+ }
+ }
+ if (scheduledFutures.size()
+ < MAX_LOCAL_SCHEDULED) {
+ return true;
+ }
+ Map.Entry latest = null;
+ for (Map.Entry entry
+ : scheduledTriggerTimes.entrySet()) {
+ if (latest == null
+ || entry.getValue()
+ > latest.getValue()) {
+ latest = entry;
+ }
+ }
+ if (latest == null
+ || latest.getValue() <= triggerAt) {
+ return false;
+ }
+ removeLocalSchedule(
+ latest.getKey(), true);
+ return scheduledFutures.size()
+ < MAX_LOCAL_SCHEDULED;
+ }
+
+ /**
+ * 计算本轮应预热到本机的最远触发时间。
+ *
+ * @return 当前时间加本地近期限窗口；溢出时为最大时间戳
+ */
+ private long scheduleHorizonTimestamp() {
+ return scheduleHorizonTimestamp(
+ System.currentTimeMillis());
+ }
+
+ /**
+ * 基于给定时间计算本机预热边界。
+ *
+ * @param now 当前时间戳
+ * @return 当前时间加近期限窗口；溢出时为最大时间戳
+ */
+ private long scheduleHorizonTimestamp(
+ long now) {
+ return now > Long.MAX_VALUE
+ - localScheduleHorizonMs
+ ? Long.MAX_VALUE
+ : now + localScheduleHorizonMs;
+ }
+
+ /**
+ * 同时移除本地 Future 与其触发时间索引。
+ *
+ * @param triggerId 触发器 ID
+ * @param cancel 是否取消尚未完成的 Future
+ * @return 被移除的 Future；不存在时为 {@code null}
+ */
+ private ScheduledFuture removeLocalSchedule(
+ String triggerId,
+ boolean cancel) {
+ synchronized (localScheduleMonitor) {
+ ScheduledFuture scheduled =
+ scheduledFutures.remove(triggerId);
+ scheduledTriggerTimes.remove(triggerId);
+ if (cancel && scheduled != null) {
+ scheduled.cancel(false);
+ }
+ return scheduled;
+ }
}
private void recoverAndSchedulePending() {
try {
- List list = store.findAllPending();
+ List list = store.findDue(
+ scheduleHorizonTimestamp());
if (list == null || list.isEmpty()) return;
for (Trigger t : list) {
scheduleInternal(t);
}
} catch (Throwable t) {
- // 忽略单次恢复错误，继续运行
- t.printStackTrace();
+ log.error("Failed to recover pending workflow triggers", t);
}
}
@@ -191,7 +413,7 @@ public class TriggerScheduler {
if (closed.get()) return;
scanFuture = scheduler.scheduleAtFixedRate(() -> {
try {
- long upto = System.currentTimeMillis();
+ long upto = scheduleHorizonTimestamp();
List due = store.findDue(upto);
if (due == null || due.isEmpty()) return;
for (Trigger t : due) {
@@ -200,27 +422,363 @@ public class TriggerScheduler {
if (sf != null && !sf.isDone() && !sf.isCancelled()) {
continue;
}
- // 直接提交到 worker，让 consumer 处理；并从 store 中移除
- if (consumer != null) {
- worker.submit(() -> {
- try {
- consumer.accept(t, worker);
- } finally {
- store.remove(t.getId());
- ScheduledFuture f2 = scheduledFutures.remove(t.getId());
- if (f2 != null) f2.cancel(false);
- }
- });
- } else {
- store.remove(t.getId());
- }
+ scheduleInternal(t);
}
} catch (Throwable tt) {
- tt.printStackTrace();
+ log.error("Failed to scan due workflow triggers", tt);
}
}, scanIntervalMs, scanIntervalMs, TimeUnit.MILLISECONDS);
}
+ /**
+ * 原子认领触发器并提交给工作线程。
+ *
+ * @param candidate 已加载的候选触发器
+ * @return 成功认领并提交时为 true
+ */
+ private boolean claimAndDispatch(Trigger candidate) {
+ if (candidate == null || candidate.getId() == null) {
+ return false;
+ }
+ String triggerId = candidate.getId();
+ ExecutorService dispatchWorker = resolveWorker(candidate);
+ Semaphore selectedPermits = resolveDispatchPermits(candidate);
+ if (selectedPermits != null && !selectedPermits.tryAcquire()) {
+ // 保留持久化触发器并清理本地占位，让后续扫描能够重新调度。
+ removeLocalSchedule(triggerId, false);
+ return false;
+ }
+ Trigger claimed;
+ try {
+ claimed = store.claim(candidate, CLAIM_LEASE_MS);
+ } catch (RuntimeException | Error error) {
+ removeLocalSchedule(triggerId, false);
+ releaseDispatchPermit(selectedPermits);
+ throw error;
+ }
+ if (claimed == null) {
+ releaseDispatchPermit(selectedPermits);
+ removeLocalSchedule(triggerId, false);
+ return false;
+ }
+ TriggerConsumer currentConsumer = consumer;
+ if (currentConsumer == null) {
+ releaseClaimBestEffort(claimed, "consumer is unavailable");
+ releaseDispatchPermit(selectedPermits);
+ removeLocalSchedule(triggerId, false);
+ return false;
+ }
+
+ ScheduledFuture renewal;
+ try {
+ renewal = scheduler.scheduleAtFixedRate(
+ () -> renewClaim(claimed),
+ CLAIM_RENEW_INTERVAL_MS,
+ CLAIM_RENEW_INTERVAL_MS,
+ TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException error) {
+ releaseClaimBestEffort(claimed, "claim renewal scheduling was rejected");
+ releaseDispatchPermit(selectedPermits);
+ removeLocalSchedule(triggerId, false);
+ return false;
+ }
+ ScheduledFuture existingRenewal = claimRenewals.put(claimed, renewal);
+ if (existingRenewal != null) {
+ existingRenewal.cancel(false);
+ }
+
+ try {
+ dispatchWorker.submit(() -> consumeClaimedTrigger(
+ claimed,
+ currentConsumer,
+ dispatchWorker,
+ selectedPermits));
+ return true;
+ } catch (RejectedExecutionException error) {
+ cancelClaimRenewal(claimed);
+ releaseClaimBestEffort(claimed, "worker submission was rejected");
+ releaseDispatchPermit(selectedPermits);
+ removeLocalSchedule(triggerId, false);
+ log.error("Workflow trigger worker rejected task, triggerId={}", triggerId, error);
+ return false;
+ }
+ }
+
+ /**
+ * 执行已认领触发器，并按执行结果确认或释放。
+ *
+ * @param trigger 已认领触发器
+ * @param currentConsumer 本次执行使用的消费者快照
+ */
+ private void consumeClaimedTrigger(
+ Trigger trigger,
+ TriggerConsumer currentConsumer,
+ ExecutorService dispatchWorker,
+ Semaphore selectedPermits) {
+ boolean succeeded = false;
+ boolean deadLetter = false;
+ boolean retryWithoutDeadLetter = false;
+ boolean rescheduleReleasedTrigger = false;
+ Throwable failure = null;
+ try {
+ TriggerContext.setCurrentTrigger(trigger);
+ if (trigger.isDeadLetterPending()) {
+ deadLetter = true;
+ failure = new NonRetryableTriggerException(
+ trigger.getDeadLetterReason());
+ } else {
+ currentConsumer.accept(trigger, dispatchWorker);
+ succeeded = true;
+ }
+ } catch (NonRetryableTriggerException error) {
+ failure = error;
+ deadLetter = true;
+ log.error("Workflow trigger is not retryable, triggerId={}", trigger.getId(), error);
+ } catch (RetryableTriggerException error) {
+ failure = error;
+ retryWithoutDeadLetter = true;
+ log.warn(
+ "Workflow trigger hit transient infrastructure contention, triggerId={}",
+ trigger.getId());
+ } catch (Throwable error) {
+ failure = error;
+ log.error("Workflow trigger execution failed, triggerId={}", trigger.getId(), error);
+ } finally {
+ try {
+ if (succeeded) {
+ store.acknowledge(trigger);
+ } else {
+ int deliveryAttempt = trigger.isDeadLetterPending()
+ ? Math.max(1, trigger.getDeliveryAttempt())
+ : incrementDeliveryAttempt(trigger);
+ if (retryWithoutDeadLetter) {
+ trigger.setTriggerAt(
+ System.currentTimeMillis()
+ + redeliveryDelayMillis(
+ deliveryAttempt));
+ store.release(trigger);
+ rescheduleReleasedTrigger = true;
+ } else if (deadLetter
+ || deliveryAttempt
+ >= MAX_DELIVERY_ATTEMPTS) {
+ String reason = failure == null
+ ? "delivery attempts exhausted"
+ : failure.getClass().getName()
+ + ": "
+ + failure.getMessage();
+ if (trigger.isDeadLetterPending()
+ && trigger.getDeadLetterReason() != null) {
+ reason = trigger.getDeadLetterReason();
+ } else {
+ trigger.setDeadLetterPending(true);
+ trigger.setDeadLetterReason(reason);
+ store.markDeadLetterPending(
+ trigger);
+ }
+ if (!store.renewClaim(
+ trigger, CLAIM_LEASE_MS)) {
+ throw new TriggerClaimLostException(
+ trigger.getId());
+ }
+ TriggerFailureListener currentFailureListener =
+ failureListener;
+ if (currentFailureListener != null
+ && !currentFailureListener.onDeadLetter(
+ trigger, failure)) {
+ trigger.setTriggerAt(
+ System.currentTimeMillis()
+ + MAX_REDELIVERY_DELAY_MS);
+ store.release(trigger);
+ rescheduleReleasedTrigger = true;
+ return;
+ }
+ store.deadLetter(trigger, reason);
+ } else {
+ trigger.setTriggerAt(
+ System.currentTimeMillis()
+ + redeliveryDelayMillis(
+ deliveryAttempt));
+ store.release(trigger);
+ rescheduleReleasedTrigger = true;
+ }
+ }
+ } catch (Throwable terminalStoreError) {
+ if (trigger.isDeadLetterPending()) {
+ releaseDeadLetterPendingBestEffort(trigger);
+ }
+ log.error(
+ "Failed to finalize workflow trigger, triggerId={}",
+ trigger.getId(),
+ terminalStoreError);
+ } finally {
+ TriggerContext.clearCurrentTrigger();
+ cancelClaimRenewal(trigger);
+ removeLocalSchedule(
+ trigger.getId(), false);
+ releaseDispatchPermit(selectedPermits);
+ if (rescheduleReleasedTrigger) {
+ scheduleInternal(trigger);
+ }
+ }
+ }
+ }
+
+ /**
+ * 在仍持有 claim 时保存待补写死信标记，避免业务终态成功后被普通 ACK 吞掉。
+ *
+ * @param trigger 待补写死信的触发器
+ */
+ private void releaseDeadLetterPendingBestEffort(
+ Trigger trigger) {
+ try {
+ if (!store.renewClaim(trigger, CLAIM_LEASE_MS)) {
+ return;
+ }
+ trigger.setTriggerAt(
+ System.currentTimeMillis()
+ + MAX_REDELIVERY_DELAY_MS);
+ store.release(trigger);
+ } catch (Throwable releaseError) {
+ log.error(
+ "Failed to persist pending dead-letter finalization, triggerId={}",
+ trigger.getId(),
+ releaseError);
+ }
+ }
+
+ /**
+ * 尽力释放已认领触发器。
+ *
+ * 释放失败时保留 Redis 中的触发器和租约，等待租约自然过期恢复。调用方仍可继续
+ * 清理本地调度状态和归还容量许可。
+ *
+ * @param trigger 已认领触发器
+ * @param reason 释放原因
+ */
+ private void releaseClaimBestEffort(Trigger trigger, String reason) {
+ try {
+ store.release(trigger);
+ } catch (Throwable error) {
+ log.error(
+ "Failed to release workflow trigger, triggerId={}, reason={}",
+ trigger == null ? null : trigger.getId(),
+ reason,
+ error);
+ }
+ }
+
+ /**
+ * 增加基础设施投递失败次数。
+ *
+ * @param trigger 当前触发器
+ * @return 增加后的次数
+ */
+ private int incrementDeliveryAttempt(Trigger trigger) {
+ int attempt = Math.max(0, trigger.getDeliveryAttempt()) + 1;
+ trigger.setDeliveryAttempt(attempt);
+ return attempt;
+ }
+
+ /**
+ * 计算带上限的指数退避，避免缺失定义或短暂冲突形成热循环。
+ *
+ * @param attempt 已失败次数
+ * @return 下次投递延迟毫秒数
+ */
+ private long redeliveryDelayMillis(int attempt) {
+ int shift = Math.min(16, Math.max(0, attempt - 1));
+ long delay = 1_000L << shift;
+ return Math.min(MAX_REDELIVERY_DELAY_MS, delay);
+ }
+
+ /**
+ * 根据工作线程池的真实容量建立领取前背压。
+ *
+ * @param executor 工作线程池
+ * @return 容量信号量；无法识别容量时返回 null
+ */
+ private Semaphore createDispatchPermits(ExecutorService executor) {
+ if (!(executor instanceof ThreadPoolExecutor)) {
+ return null;
+ }
+ ThreadPoolExecutor pool = (ThreadPoolExecutor) executor;
+ long capacity = (long) pool.getMaximumPoolSize() + pool.getQueue().remainingCapacity();
+ return new Semaphore((int) Math.max(1L, Math.min(Integer.MAX_VALUE, capacity)));
+ }
+
+ /**
+ * 释放一个工作线程容量许可。
+ */
+ private void releaseDispatchPermit(Semaphore permits) {
+ if (permits != null) {
+ permits.release();
+ }
+ }
+
+ /**
+ * 解析触发器执行线程池。
+ *
+ * @param trigger 触发器
+ * @return 默认或独立通道线程池
+ */
+ private ExecutorService resolveWorker(Trigger trigger) {
+ String lane = trigger == null ? null : trigger.getExecutionLane();
+ return lane == null ? worker : laneWorkers.getOrDefault(lane, worker);
+ }
+
+ /**
+ * 解析所选线程池对应的领取前背压许可。
+ *
+ * @param trigger 触发器
+ * @return 容量许可；无法识别时为 {@code null}
+ */
+ private Semaphore resolveDispatchPermits(Trigger trigger) {
+ String lane = trigger == null ? null : trigger.getExecutionLane();
+ return lane == null
+ ? dispatchPermits
+ : laneDispatchPermits.getOrDefault(lane, dispatchPermits);
+ }
+
+ /**
+ * 续期触发器认领租约。
+ *
+ * @param trigger 已认领触发器
+ */
+ private void renewClaim(Trigger trigger) {
+ try {
+ if (!store.renewClaim(trigger, CLAIM_LEASE_MS)) {
+ log.warn("Workflow trigger claim renewal lost ownership, triggerId={}", trigger.getId());
+ cancelClaimRenewal(trigger);
+ }
+ } catch (Throwable error) {
+ log.error("Workflow trigger claim renewal failed, triggerId={}", trigger.getId(), error);
+ }
+ }
+
+ /**
+ * 在业务状态提交前验证具体触发器仍由当前工作线程持有。
+ *
+ * @param trigger 已认领触发器
+ * @throws TriggerClaimLostException 租约已失效或已转移
+ */
+ public void assertClaimOwned(Trigger trigger) {
+ if (trigger != null && !store.renewClaim(trigger, CLAIM_LEASE_MS)) {
+ throw new TriggerClaimLostException(trigger.getId());
+ }
+ }
+
+ /**
+ * 取消触发器租约续期任务。
+ *
+ * @param trigger 已认领触发器
+ */
+ private void cancelClaimRenewal(Trigger trigger) {
+ ScheduledFuture renewal = claimRenewals.remove(trigger);
+ if (renewal != null) {
+ renewal.cancel(false);
+ }
+ }
+
public void shutdown() {
if (closed.compareAndSet(false, true)) {
if (scanFuture != null) scanFuture.cancel(false);
@@ -232,6 +790,11 @@ public class TriggerScheduler {
}
}
scheduledFutures.clear();
+ scheduledTriggerTimes.clear();
+ for (ScheduledFuture renewal : claimRenewals.values()) {
+ renewal.cancel(false);
+ }
+ claimRenewals.clear();
try {
scheduler.shutdownNow();
@@ -241,6 +804,18 @@ public class TriggerScheduler {
worker.shutdownNow();
} catch (Throwable ignored) {
}
+ for (ExecutorService laneWorker :
+ new java.util.HashSet<>(laneWorkers.values())) {
+ if (laneWorker == worker) {
+ continue;
+ }
+ try {
+ laneWorker.shutdownNow();
+ } catch (Throwable ignored) {
+ }
+ }
+ laneWorkers.clear();
+ laneDispatchPermits.clear();
}
}
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerStore.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerStore.java
index fdb3dd9..02a2562 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerStore.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/TriggerStore.java
@@ -18,8 +18,27 @@ package com.easyagents.flow.core.chain.runtime;
import java.util.List;
public interface TriggerStore {
+
Trigger save(Trigger trigger);
+ /**
+ * 仅在同 ID 触发器尚不存在时原子保存。
+ *
+ * 缺省实现保证同一仓储实例内原子；分布式仓储必须覆盖为跨进程原子操作。
+ *
+ * @param trigger 带稳定 ID 的触发器
+ * @return 本次成功创建时为 {@code true}，已存在时为 {@code false}
+ */
+ default boolean saveIfAbsent(Trigger trigger) {
+ synchronized (this) {
+ if (find(trigger.getId()) != null) {
+ return false;
+ }
+ save(trigger);
+ return true;
+ }
+ }
+
boolean remove(String triggerId);
Trigger find(String triggerId);
@@ -27,4 +46,116 @@ public interface TriggerStore {
List findDue(long uptoTimestamp);
List findAllPending();
+
+ /**
+ * 原子认领待执行触发器。
+ *
+ * 缺省实现适用于单进程仓储：先读取再以删除结果作为认领成功标志。
+ *
+ * @param triggerId 触发器 ID
+ * @param leaseMillis 认领租约毫秒数
+ * @return 认领成功时返回触发器，否则返回 null
+ */
+ default Trigger claim(String triggerId, long leaseMillis) {
+ Trigger trigger = find(triggerId);
+ return trigger != null && remove(triggerId) ? trigger : null;
+ }
+
+ /**
+ * 原子认领已加载的待执行触发器。
+ *
+ *
分布式仓储可使用候选触发器中的实例 ID 构建与本次 claim 绑定的执行守卫，
+ * 避免认领前额外读取完整触发器负载。
+ *
+ * @param candidate 扫描或主动触发阶段已加载的候选触发器
+ * @param leaseMillis 认领租约毫秒数
+ * @return 认领成功时返回触发器，否则返回 {@code null}
+ */
+ default Trigger claim(Trigger candidate, long leaseMillis) {
+ return candidate == null ? null : claim(candidate.getId(), leaseMillis);
+ }
+
+ /**
+ * 仅按 ID 续期触发器租约。
+ *
+ * 分布式仓储无法仅凭 ID 验证 owner token。该兼容入口不应再用于运行时提交路径，
+ * 调用方必须保留 {@link Trigger} 认领对象并使用
+ * {@link #renewClaim(Trigger, long)}。
+ *
+ * @param triggerId 触发器 ID
+ * @param leaseMillis 新租约毫秒数
+ * @return 不适用
+ * @throws UnsupportedOperationException 始终抛出，防止无 owner token 的不安全续期
+ */
+ @Deprecated
+ default boolean renewClaim(String triggerId, long leaseMillis) {
+ throw new UnsupportedOperationException("Trigger claim token is required");
+ }
+
+ /**
+ * 续期调用方持有的具体触发器租约。
+ *
+ * @param trigger 已认领触发器对象
+ * @param leaseMillis 新租约毫秒数
+ * @return 续期成功时为 true
+ */
+ default boolean renewClaim(Trigger trigger, long leaseMillis) {
+ // 单进程缺省仓储在 claim 时已经移除触发器，不需要租约续期。
+ return true;
+ }
+
+ /**
+ * 仅按 ID 确认触发器。
+ *
+ * 分布式仓储无法仅凭 ID 验证 owner token，运行时必须使用
+ * {@link #acknowledge(Trigger)}。
+ *
+ * @param triggerId 触发器 ID
+ * @throws UnsupportedOperationException 始终抛出，防止旧 owner 删除新 owner 的任务
+ */
+ @Deprecated
+ default void acknowledge(String triggerId) {
+ throw new UnsupportedOperationException("Claimed trigger object is required");
+ }
+
+ /**
+ * 确认调用方持有的具体触发器执行成功。
+ *
+ * @param trigger 已认领触发器对象
+ */
+ default void acknowledge(Trigger trigger) {
+ // 单进程缺省认领已经移除触发器，无需再次处理。
+ }
+
+ /**
+ * 释放失败执行的触发器，使其可以再次被认领。
+ *
+ * @param trigger 执行失败的触发器
+ */
+ default void release(Trigger trigger) {
+ save(trigger);
+ }
+
+ /**
+ * 在保持当前 claim 的同时持久化待补写死信标记。
+ *
+ * 分布式仓储必须校验具体 owner token；进程崩溃后，新 owner 依靠该标记
+ * 跳过业务执行并继续终态协议。
+ *
+ * @param trigger 已认领且标记为待补写死信的触发器
+ */
+ default void markDeadLetterPending(
+ Trigger trigger) {
+ // 单进程仓储的 claimed trigger 仅存在于当前调用栈，无需额外持久化。
+ }
+
+ /**
+ * 将不可继续执行或超过投递上限的触发器移入死信。
+ *
+ * @param trigger 已认领触发器
+ * @param reason 死信原因
+ */
+ default void deadLetter(Trigger trigger, String reason) {
+ acknowledge(trigger);
+ }
}
diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/code/impl/JavascriptRuntimeEngine.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/code/impl/JavascriptRuntimeEngine.java
index 9d94608..040087e 100644
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/code/impl/JavascriptRuntimeEngine.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/code/impl/JavascriptRuntimeEngine.java
@@ -16,6 +16,8 @@
package com.easyagents.flow.core.code.impl;
import com.easyagents.flow.core.chain.Chain;
+import com.easyagents.flow.core.chain.ChainState;
+import com.easyagents.flow.core.chain.NodeState;
import com.easyagents.flow.core.code.CodeRuntimeEngine;
import com.easyagents.flow.core.node.CodeNode;
import com.easyagents.flow.core.util.graalvm.JsInteropUtils;
@@ -39,8 +41,13 @@ public class JavascriptRuntimeEngine implements CodeRuntimeEngine {
public Map execute(String code, CodeNode node, Chain chain) {
try (Context context = CONTEXT_BUILDER.build()) {
Value bindings = context.getBindings("js");
+ ChainState chainState =
+ chain.getExecutionState();
+ NodeState nodeState =
+ chain.getNodeState(node.getId());
- Map