problems = new ArrayList<>();
+ VariableContract.validateDefinition(model, false);
if (model.nodes == null || model.nodes.isEmpty()) {
throw new BizException(ErrorCode.PARAM_ERROR, "流程定义缺少节点");
@@ -248,7 +249,7 @@ public final class DefValidator {
* (命名/唯一/同分支上下文由度约束段复用 {@link #validateOutlets} 口径)。
*
* EXPRESSION 模式(2026-10-10 四议题)分层校验:语法错/出口缺条件 → 发布期硬拦
- * (表单变量是子应用运行时传的,发布期无从核对 schema,一律放行、缺失按 null 注入求值)。
+ * 新契约的变量引用/类型/值域由 VariableContract 统一检查;历史无声明版本保持兼容。
*/
private static void validateCondition(FlowDefModel model, FlowDefModel.Node n, Set enabledRouterKeys,
List problems) {
diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java b/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
index 08f14ca..ef0e950 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
@@ -1,81 +1,201 @@
package com.yunhai.flowcenter.engine;
-import com.yunhai.flowcenter.entity.FlwDef;
-import com.yunhai.flowcenter.entity.FlwInstance;
-import com.yunhai.flowcenter.entity.FlwTask;
+import com.yunhai.flowcenter.entity.*;
import com.yunhai.flowcenter.enums.FlwEnums;
-import com.yunhai.flowcenter.repository.FlwDefRepository;
-import com.yunhai.flowcenter.repository.FlwInstanceRepository;
-import com.yunhai.flowcenter.repository.FlwTaskRepository;
+import com.yunhai.flowcenter.repository.*;
import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Propagation;
-import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.annotation.*;
-/**
- * 节点放行推进(AFTER_COMMIT 监听器经代理调用本 bean,REQUIRES_NEW 生效——
- * 事务已提交后的同步监听阶段没有活动事务,@Modifying 批量语句必须有新事务承载):
- * 该节点待办清零且办件在途才推进;并发冲突由 flw_instance.version 乐观锁暴露,监听器层负责重试。
- */
-@Slf4j
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+import java.time.LocalDateTime;
+import java.util.*;
+
+/** D09: durable activation membership, serialized last-handler payload, and revision-scoped settlement. */
@Service
@RequiredArgsConstructor
public class FlowAdvanceService {
-
private final FlwInstanceRepository instanceRepository;
private final FlwDefRepository defRepository;
private final FlwTaskRepository taskRepository;
private final FlowEngine flowEngine;
+ private final FlwAdvanceIntentRepository intentRepository;
+ @PersistenceContext private EntityManager entityManager;
+
+ /** The listener groups the existing totalTasks TASK_CREATED events within their creation transaction. */
+ @Transactional(propagation = Propagation.MANDATORY)
+ public Long registerTask(FlowEvent event, Long activationIntentId) {
+ String nodeId = (String) event.getData().get("nodeId");
+ FlwAdvanceIntent intent;
+ if (activationIntentId == null) {
+ FlwInstance instance = instanceRepository.findById(event.getInstanceId()).orElseThrow(IllegalStateException::new);
+ intent = new FlwAdvanceIntent();
+ intent.setInstanceId(instance.getId());intent.setNodeId(nodeId);
+ intent.setActivationTaskId(event.getTaskId());
+ intent.setActivationDefId(instance.getDefId());intent.setActivationDefVersion(instance.getDefVersion());
+ intent.setActivationApplyRound(instance.getApplyRound());
+ intent.setStatus(FlwAdvanceIntent.STATE_DONE); // Registration only; no advance work before handling.
+ intent.setNextRetryAt(LocalDateTime.now());
+ } else {
+ intent = intentRepository.findById(activationIntentId).orElseThrow(IllegalStateException::new);
+ if (!Objects.equals(intent.getInstanceId(), event.getInstanceId()) || !Objects.equals(intent.getNodeId(), nodeId)) {
+ throw new IllegalStateException("Activation registration identity mismatch");
+ }
+ }
+ intent.addTask(event.getTaskId());
+ return intentRepository.save(intent).getId();
+ }
+
+ /** TRANSFERRED lacks the parent task ID. Accept only a uniquely proven parent activation, never guess by max ID. */
+ @Transactional(propagation = Propagation.MANDATORY)
+ public void registerTransfer(FlowEvent event) {
+ String nodeId = (String) event.getData().get("fromNodeId");
+ List generations = intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(event.getInstanceId(), nodeId);
+ Set parents = new HashSet<>();
+ List candidates = taskRepository.findTransferParentCandidates(event.getTaskId(), event.getOperatorId());
+ for (Long taskId : candidates) {
+ FlwAdvanceIntent parent = memberOf(generations, taskId);
+ if (parent == null) throw new IllegalStateException("D09 transfer parent requires audited activation backfill");
+ parents.add(parent.getId());
+ }
+ if (parents.size() != 1) throw new IllegalStateException("D09 ambiguous transfer activation; parent identity is required");
+ FlwAdvanceIntent parent = lockIntent(parents.iterator().next());
+ parent.addTask(event.getTaskId());
+ intentRepository.save(parent);
+ }
+
+ /** Called synchronously in the task transaction, after claimAndFinish. The head lock orders commits, not clocks/IDs. */
+ @Transactional(propagation = Propagation.MANDATORY)
+ public void recordHandled(FlowEvent event) {
+ String nodeId = (String) event.getData().get("nodeId");
+ FlwAdvanceIntent candidate = memberOf(intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(event.getInstanceId(), nodeId), event.getTaskId());
+ if (candidate == null) throw new IllegalStateException("D09 task has no proven activation; audited backfill required");
+ FlwAdvanceIntent intent = lockIntent(candidate.getId());
+ if (intent.handledTask(event.getTaskId())) return; // Duplicate/late publication cannot rewind the last handler.
+ intent.addHandledTask(event.getTaskId());intent.setTriggerTaskId(event.getTaskId());
+ intent.setEdgeName((String) event.getData().get("edgeName"));
+ intent.setAssigneeId((Integer) event.getData().get("assigneeId"));
+ intent.setAssigneeName((String) event.getData().get("assigneeName"));
+ intent.setStatus(FlwAdvanceIntent.STATE_PENDING);intent.setAttempts(0);
+ intent.setNextRetryAt(LocalDateTime.now());intent.setLastError(null);
+ intentRepository.save(intent);
+ }
+
+ public static final class IntentRef {
+ public final Long id;
+ public final Long revision;
+ public IntentRef(Long id, Long revision) { this.id = id; this.revision = revision; }
+ }
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
+ public IntentRef pendingForTask(Long instanceId, String nodeId, Long taskId) {
+ FlwAdvanceIntent intent = memberOf(intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(instanceId, nodeId), taskId);
+ return intent == null || !FlwAdvanceIntent.STATE_PENDING.equals(intent.getStatus()) ? null
+ : new IntentRef(intent.getId(), intent.getRevision());
+ }
/**
- * @param nodeId 待放行节点
- * @param edgeName 多出口节点所选出口(办理事件携带);可空——兜底从该节点
- * 最近一条已同意任务的 edge_name 列重放(推进在 AFTER_COMMIT 新事务重查,
- * 事件传参只是快路径,落库的值才是权威)
- * @param operator 本节点办理人(TASK_HANDLED 事件 data 穿透;推进产生的
- * TASK_CREATED/COMPLETED 操作人信封字段来源),可空
- * @return 是否实际推进了
+ * One transaction owns validation, engine effects and settlement. RC sees the last committed task status;
+ * the head lock serializes against last-handler refresh, and instance @Version fences reactivation writes.
*/
- @Transactional(propagation = Propagation.REQUIRES_NEW)
- public boolean tryAdvance(Long instanceId, String nodeId, String edgeName, FlowOperator operator) {
- FlwInstance instance = instanceRepository.findById(instanceId).orElse(null);
- if (instance == null
- || instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()) {
+ @Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.READ_COMMITTED)
+ public boolean tryAdvance(Long intentId, Long observedRevision) {
+ FlwAdvanceIntent intent = lockIntent(intentId);
+ if (!ownsRevision(intent, observedRevision)) return false;
+ FlwAdvanceIntent latest = intentRepository.findFirstByInstanceIdAndNodeIdAndActivationTaskIdIsNotNullOrderByIdDesc(
+ intent.getInstanceId(), intent.getNodeId()).orElse(null);
+ if (latest != null && !Objects.equals(latest.getId(), intent.getId())) {
+ done(intent, "D09_STALE_ACTIVATION: successor=" + latest.getId());
+ handoff(latest.getId());
return false;
}
- // 已推进过的节点不再放行(currentNodes 由引擎在放行时移除,乐观锁保证互斥)
- String current = instance.getCurrentNodes();
- boolean active = current != null && java.util.Arrays.stream(current.split(","))
- .anyMatch(nodeId::equals);
- if (!active) {
+ if (intent.getActivationTaskId() == null || !intent.containsTask(intent.getActivationTaskId())
+ || intent.getTriggerTaskId() == null || !intent.handledTask(intent.getTriggerTaskId())) {
+ hold(intent, "D09_UNANCHORED: audited cohort/last-handler backfill required; not executed");
return false;
}
- long pending = taskRepository.findByInstanceIdAndStatus(instanceId,
- FlwEnums.TaskStatus.PENDING.getCode()).stream()
- .filter(t -> nodeId.equals(t.getNodeId()))
- .count();
- if (pending > 0) {
- return false;
+ FlwInstance instance = instanceRepository.findById(intent.getInstanceId()).orElse(null);
+ if (instance == null || instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()
+ || !active(instance, intent.getNodeId())) {
+ done(intent, null);return false;
}
- FlwDef def = defRepository.findById(instance.getDefId()).orElse(null);
- if (def == null) {
- log.error("[flow-center] 办件定义缺失,推进中止:instance={} defId={}",
- instanceId, instance.getDefId());
- return false;
+ if (!Objects.equals(instance.getDefId(), intent.getActivationDefId())
+ || !Objects.equals(instance.getDefVersion(), intent.getActivationDefVersion())
+ || !Objects.equals(instance.getApplyRound(), intent.getActivationApplyRound())) {
+ hold(intent, "D09_IDENTITY_MISMATCH: no proven successor; audited backfill required");return false;
}
- String chosen = edgeName;
- if (chosen == null) {
- chosen = taskRepository.findByInstanceIdAndStatus(instanceId,
- FlwEnums.TaskStatus.APPROVED.getCode()).stream()
- .filter(t -> nodeId.equals(t.getNodeId()) && t.getEdgeName() != null)
- .max(java.util.Comparator.comparing(
- FlwTask::getHandleTime, java.util.Comparator.nullsFirst(java.util.Comparator.naturalOrder())))
- .map(FlwTask::getEdgeName)
- .orElse(null);
+ boolean pending = taskRepository.findByInstanceIdAndStatus(instance.getId(), FlwEnums.TaskStatus.PENDING.getCode())
+ .stream().anyMatch(t -> intent.getNodeId().equals(t.getNodeId()));
+ if (pending) {
+ // A future handler holds/obtains this same head lock and must re-arm it in its task transaction.
+ done(intent, null);return false;
}
- flowEngine.continueAdvance(instance, def, nodeId, chosen, operator);
- instanceRepository.save(instance);
+ FlwTask trigger = taskRepository.findById(intent.getTriggerTaskId()).orElse(null);
+ if (trigger == null || !intent.containsTask(trigger.getId())
+ || !Objects.equals(trigger.getInstanceId(), instance.getId())
+ || !Objects.equals(trigger.getNodeId(), intent.getNodeId())
+ || !Integer.valueOf(FlwEnums.TaskStatus.APPROVED.getCode()).equals(trigger.getStatus())) {
+ hold(intent, "D09_INVALID_LAST_HANDLER: not executed");return false;
+ }
+ FlwDef def = defRepository.findById(instance.getDefId()).orElseThrow(() -> new IllegalStateException("Missing activation definition"));
+ flowEngine.continueAdvance(instance, def, intent.getNodeId(), trigger.getEdgeName(),
+ FlowOperator.of(trigger.getAssigneeId(), trigger.getAssigneeName()));
+ if (active(instance, intent.getNodeId())) throw new IllegalStateException("Activation did not advance; retaining recoverable intent");
+ instanceRepository.saveAndFlush(instance); // An optimistic conflict rolls back engine writes AND settlement.
+ done(intent, null);
return true;
}
+
+ /** Never merge a worker's detached row. A refreshed revision owns its own retry budget. */
+ @Transactional(propagation = Propagation.REQUIRES_NEW)
+ public void defer(Long intentId, Long observedRevision, String error) {
+ FlwAdvanceIntent intent = lockIntent(intentId);
+ if (!ownsRevision(intent, observedRevision)) return;
+ intent.setAttempts(intent.getAttempts() + 1);intent.setLastError(truncate(error));
+ if (intent.getAttempts() >= 6) intent.setStatus(FlwAdvanceIntent.STATE_EXHAUSTED);
+ else intent.setNextRetryAt(LocalDateTime.now().plusSeconds(Math.min(1800, 30L * (1L << (intent.getAttempts() - 1)))));
+ intentRepository.save(intent);
+ }
+
+ private FlwAdvanceIntent lockIntent(Long id) {
+ // Queries in a handling transaction may already have loaded a stale RR snapshot into the identity map.
+ // A lock upgrade on that managed row checks its obsolete @Version before refresh can run.
+ // Flush our own changes, detach ONLY this row, then load a new managed row with a locking current read.
+ entityManager.flush();
+ FlwAdvanceIntent cached = entityManager.find(FlwAdvanceIntent.class, id);
+ if (cached != null) entityManager.detach(cached);
+ return intentRepository.findByIdForUpdate(id).orElse(null);
+ }
+
+ private void handoff(Long successorId) {
+ FlwAdvanceIntent successor = lockIntent(successorId);
+ // A stale worker must not consume its successor. Re-arm a previously waiting head, retaining its own payload.
+ if (successor != null && successor.getTriggerTaskId() != null && FlwAdvanceIntent.STATE_DONE.equals(successor.getStatus())) {
+ successor.setStatus(FlwAdvanceIntent.STATE_PENDING);successor.setAttempts(0);
+ successor.setNextRetryAt(LocalDateTime.now());successor.setLastError(null);intentRepository.save(successor);
+ }
+ }
+
+ private static FlwAdvanceIntent memberOf(List intents, Long taskId) {
+ FlwAdvanceIntent found = null;
+ for (FlwAdvanceIntent intent : intents) if (intent.containsTask(taskId)) {
+ if (found != null && !Objects.equals(found.getId(), intent.getId())) throw new IllegalStateException("Task belongs to multiple activations");
+ found = intent;
+ }
+ return found;
+ }
+ private static boolean ownsRevision(FlwAdvanceIntent intent, Long revision) {
+ return intent != null && FlwAdvanceIntent.STATE_PENDING.equals(intent.getStatus()) && Objects.equals(intent.getRevision(), revision);
+ }
+ private static boolean active(FlwInstance instance, String nodeId) {
+ return instance.getCurrentNodes() != null && Arrays.asList(instance.getCurrentNodes().split(",")).contains(nodeId);
+ }
+ private void done(FlwAdvanceIntent intent, String reason) {
+ intent.setStatus(FlwAdvanceIntent.STATE_DONE);intent.setLastError(reason);intentRepository.save(intent);
+ }
+ private void hold(FlwAdvanceIntent intent, String reason) {
+ intent.setStatus(FlwAdvanceIntent.STATE_EXHAUSTED);intent.setLastError(reason);intentRepository.save(intent);
+ }
+ private static String truncate(String s) { return s == null || s.length() <= 255 ? s : s.substring(0,255); }
}
diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java b/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
index db0076c..64c2f67 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
@@ -56,6 +56,18 @@ public class FlowDefModel {
public List nodes;
public List edges;
public Settings settings;
+ /** null = 历史无契约版本;新发布必须显式声明(无业务变量时为 [])。 */
+ public List variables;
+
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class Variable {
+ public String key;
+ /** string/number/boolean/date/datetime/money/stringArray/fileArray/userArray */
+ public String type;
+ public Boolean required;
+ @com.fasterxml.jackson.annotation.JsonProperty("enum")
+ public List enumValues;
+ }
/**
* 画布坐标(nodeId → 左上角),仅前端设计器/查看器复用;引擎不消费。
* 缺失时前端自行做自动布局,故可选、可空。
diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java b/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
index 6c9bb7f..d1b7543 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
@@ -45,6 +45,9 @@ public class FlowEngine {
/** 发起后首推:从 START 激活(instance 已建好、status=10)。operator=null:TASK_CREATED/COMPLETED 的操作人回退发起人 */
public void startAdvance(FlwInstance instance, FlwDef def) {
FlowDefModel model = parseModel(def);
+ // D11:整份定义的固定 USERS 先整体预检,非法人不得等到后续节点才暴露(会留下卡单)。
+ // 校验失败抛业务异常,由发起事务整体回滚(零实例零任务)。
+ assigneeResolver.requireEligibleFixedAssignees(instance.getSchId(), model);
setCurrentNodes(instance, null);
advance(instance, model, model.start, null, null);
}
diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java b/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
index f500a59..84b7efc 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
@@ -69,6 +69,55 @@ public class FlowEvent {
private final String operatorName;
/** 事件明细(result/opinion/rejectNode/task.created 的 n/m 等),可为空 */
private final Map data;
+ private Map roundIdentity;
+
+ // ===== T05 稳定事件身份(2026-10-25) =====
+ /** 事实行 event_id:发布线程同步监听器在状态事务内落事实行时一次性赋值,全订阅方/重试复用 */
+ private String stableEventId;
+ /** 事实行 event_seq(实例内单调序号) */
+ private Integer stableEventSeq;
+ /**
+ * R2(Codex D04)冻结信封:事实落库时一次性构建的完整投递载荷 JSON(含 eventId/eventSeq/
+ * flowGuid 等全部信封字段)。正常分发与崩溃恢复分发共用同一份冻结体——恢复消息体不再缺
+ * eventId/eventSeq,两条路径字节级同构。事实行缺席(存量路径)为 null,由分发器现场构建。
+ */
+ private String stablePayloadJson;
+
+ public void assignStableIdentity(String eventId, Integer eventSeq) {
+ if (this.stableEventId != null) {
+ return; // 只赋一次:幂等重入(同事件多监听器)不换身份
+ }
+ this.stableEventId = eventId;
+ this.stableEventSeq = eventSeq;
+ }
+
+ public String getStableEventId() {
+ return stableEventId;
+ }
+
+ public Integer getStableEventSeq() {
+ return stableEventSeq;
+ }
+
+ /** 随稳定身份一并冻结完整投递信封(只赋一次;身份与信封同源同事务) */
+ public void assignStablePayload(String payloadJson) {
+ if (this.stablePayloadJson != null) {
+ return;
+ }
+ this.stablePayloadJson = payloadJson;
+ }
+
+ public String getStablePayloadJson() {
+ return stablePayloadJson;
+ }
+
+ public void captureRoundIdentity(Integer round, String requestToken) {
+ if (roundIdentity != null) return;
+ Map identity = new HashMap<>();
+ identity.put("applyRound", round);
+ identity.put("requestToken", requestToken);
+ roundIdentity = java.util.Collections.unmodifiableMap(identity);
+ }
public static FlowEvent of(Type type, Long instanceId) {
return new FlowEvent(type, instanceId, null, null, null, new HashMap<>());
diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java b/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
index 1ac00c3..d0e8a93 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
@@ -1,50 +1,102 @@
package com.yunhai.flowcenter.engine;
+import com.yunhai.flowcenter.entity.FlwAdvanceIntent;
+import com.yunhai.flowcenter.repository.FlwAdvanceIntentRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
import org.springframework.orm.ObjectOptimisticLockingFailureException;
+import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
-/**
- * 节点放行推进监听器:TASK_HANDLED 事件 AFTER_COMMIT 后在新事务里重查——
- * 该节点待办是否清零、办件是否仍在途,全部成立才继续推进(下一节点/JOIN 合并/END)。
- * 会签并发办理时各方提交的快照互相看不到对方未提交的办理行,事务内判断
- * "节点是否全部办结"必假,提交后重查才是唯一可靠时机;推进串行化由办件乐观锁兜底
- * (重试循环在监听器侧,跨 bean 调用 {@link FlowAdvanceService#tryAdvance} 保证 REQUIRES_NEW 生效)。
- */
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+/** TASK_CREATED establishes membership; TASK_HANDLED refreshes that same activation in the handling transaction. */
@Slf4j
@Component
@RequiredArgsConstructor
public class FlowNodeAdvanceListener {
-
private final FlowAdvanceService advanceService;
+ private final FlwAdvanceIntentRepository intentRepository;
+
+ @EventListener
+ @Order(Ordered.HIGHEST_PRECEDENCE + 2)
+ public void persistIntent(FlowEvent event) {
+ if (event.getData() == null) return;
+ if (event.getType() == FlowEvent.Type.TASK_CREATED) {
+ ActivationBatch batch = batch();
+ if (!batch.tasks.add(event.getTaskId())) return;
+ String key = event.getInstanceId() + ":" + event.getData().get("nodeId");
+ int total = ((Number) event.getData().get("totalTasks")).intValue();
+ if (total <= 0) throw new IllegalStateException("Missing activation batch size");
+ Registration registration = batch.nodes.get(key);
+ if (registration == null) { registration = new Registration(total);batch.nodes.put(key, registration); }
+ if (registration.total != total) throw new IllegalStateException("Activation batch size changed");
+ registration.intentId = advanceService.registerTask(event, registration.intentId);
+ if (--registration.remaining == 0) batch.nodes.remove(key);
+ } else if (event.getType() == FlowEvent.Type.TRANSFERRED) {
+ advanceService.registerTransfer(event);
+ } else if (event.getType() == FlowEvent.Type.TASK_HANDLED && event.getData().get("nodeId") != null) {
+ advanceService.recordHandled(event);
+ }
+ }
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
public void onTaskHandled(FlowEvent event) {
- if (event.getType() != FlowEvent.Type.TASK_HANDLED || event.getData() == null) {
- return;
- }
+ if (event.getType() != FlowEvent.Type.TASK_HANDLED || event.getData() == null) return;
String nodeId = (String) event.getData().get("nodeId");
- if (nodeId == null) {
- return;
- }
- String edgeName = (String) event.getData().get("edgeName");
- // 办理人随事件 data 穿透进推进链:新 TASK_CREATED/COMPLETED 的操作人信封字段(2026-09-20)
- FlowOperator operator = FlowOperator.of(
- (Integer) event.getData().get("assigneeId"),
- (String) event.getData().get("assigneeName"));
+ if (nodeId == null) return;
+ FlowAdvanceService.IntentRef ref = advanceService.pendingForTask(event.getInstanceId(), nodeId, event.getTaskId());
+ if (ref == null) return;
for (int attempt = 1; attempt <= 3; attempt++) {
- try {
- advanceService.tryAdvance(event.getInstanceId(), nodeId, edgeName, operator);
- return;
- } catch (ObjectOptimisticLockingFailureException e) {
- log.warn("[flow-center] 推进冲突重试 {}/3:instance={} node={}",
- attempt, event.getInstanceId(), nodeId);
+ try { advanceService.tryAdvance(ref.id, ref.revision);return; }
+ catch (ObjectOptimisticLockingFailureException conflict) {
+ log.warn("[flow-center] D09 advance conflict {}/3: intent={} revision={}", attempt, ref.id, ref.revision);
+ } catch (Exception failure) {
+ advanceService.defer(ref.id, ref.revision, failure.toString());return;
}
}
- log.error("[flow-center] 推进重试耗尽(请人工检查):instance={} node={}",
- event.getInstanceId(), nodeId);
+ advanceService.defer(ref.id, ref.revision, "AFTER_COMMIT retries exhausted; sweep will resume");
+ }
+
+ @Scheduled(fixedDelay = 30_000)
+ public void sweepIntents() {
+ for (FlwAdvanceIntent detached : intentRepository.findTop50ByStatusAndNextRetryAtLessThanEqualOrderByCreateTimeAsc(
+ FlwAdvanceIntent.STATE_PENDING, LocalDateTime.now())) {
+ Long id = detached.getId(), revision = detached.getRevision();
+ try { advanceService.tryAdvance(id, revision); }
+ catch (Exception failure) { advanceService.defer(id, revision, failure.toString()); }
+ }
+ }
+
+ /** Synchronizations are suspended with REQUIRES_NEW; an outer activation batch cannot leak into its child. */
+ private static ActivationBatch batch() {
+ if (!TransactionSynchronizationManager.isSynchronizationActive()) throw new IllegalStateException("Activation requires its creation transaction");
+ for (TransactionSynchronization sync : TransactionSynchronizationManager.getSynchronizations()) {
+ if (sync instanceof ActivationBatch) return (ActivationBatch) sync;
+ }
+ ActivationBatch created = new ActivationBatch();
+ TransactionSynchronizationManager.registerSynchronization(created);return created;
+ }
+ private static final class ActivationBatch implements TransactionSynchronization {
+ final Map nodes = new HashMap<>();
+ final java.util.Set tasks = new java.util.HashSet<>();
+ @Override public void beforeCommit(boolean readOnly) {
+ if (!nodes.isEmpty()) throw new IllegalStateException("Incomplete activation membership; creation transaction must roll back");
+ }
+ }
+ private static final class Registration {
+ final int total;
+ int remaining;
+ Long intentId;
+ Registration(int total) { this.total = total;this.remaining = total; }
}
}
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
index a2cf36c..85d38c0 100644
--- a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
@@ -12,14 +12,16 @@ import javax.persistence.UniqueConstraint;
import java.time.LocalDateTime;
/**
- * 回调推送日志 = 重试队列(概览 §4.9):eventId 唯一(重试不变,业务方按它幂等去重);
+ * 回调推送日志 = 重试队列(概览 §4.9;2026-10-25 R2 幂等键改复合):eventId 全订阅方复用
+ * (重试不变,业务方按它幂等去重),(event_id, reg_id) 复合唯一——同一事件对同一订阅方只建
+ * 一行,多订阅方各一行;并发 exists→insert 竞态由复合 uk 兜底(DuplicateKeyException=已建行)。
* payload_json 完整快照可重放;索引 (status, next_retry_time) 是扫描主路径。
*/
@Getter
@Setter
@Entity
@Table(name = "flw_callback_log",
- uniqueConstraints = @UniqueConstraint(name = "uk_flw_log_event", columnNames = "event_id"),
+ uniqueConstraints = @UniqueConstraint(name = "uk_flw_log_event_reg", columnNames = {"event_id", "reg_id"}),
indexes = {
@Index(name = "idx_flw_log_scan", columnList = "status, next_retry_time"),
@Index(name = "idx_flw_log_inst", columnList = "instance_id")})
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
index 5e04140..7f83a56 100644
--- a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
@@ -21,6 +21,9 @@ import javax.persistence.UniqueConstraint;
columnNames = {"sch_id", "app_key", "flow_guid"}),
indexes = @Index(name = "idx_flw_reg_flow", columnList = "sch_id, flow_guid, status"))
public class FlwCallbackReg extends BaseEntity {
+ @Column(name="source_principal_id",length=36)
+ private String sourcePrincipalId;
+
/** 租户键 */
@Column(name = "sch_id", nullable = false)
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java b/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
index edc7724..9204b9a 100644
--- a/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
@@ -74,6 +74,15 @@ public class FlwDef extends BaseEntity {
@Column(name = "form_version")
private Integer formVersion;
+ /**
+ * 线级停发标志(2026-10-25 T03):1=该 flowGuid 整线停发——阻止新发起/预览,不影响在途实例
+ * (在途按实例固定 defId 原版推进)。与版本级 status(DISABLED) 正交:线级停发后 enable()
+ * 恢复的只是版本状态,线仍不可新发起;enableLine 才恢复线可用性,且不改写任何发布内容字节。
+ * 批量 UPDATE 全版本行同值,读任一行即读线状态。
+ */
+ @Column(name = "line_offline", nullable = false, columnDefinition = "INT NOT NULL DEFAULT 0")
+ private Integer lineOffline = 0;
+
/** §5.1 流程定义 JSON */
@Column(name = "content_json", nullable = false, columnDefinition = "LONGTEXT")
private String contentJson;
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java b/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
index 42be69b..b8e8143 100644
--- a/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
@@ -14,8 +14,9 @@ import java.time.LocalDateTime;
/**
* 办件实例(概览 §4.3 + V2.0 Part D):托管模式下业务对象留在业务方,本表只存
- * biz_title/biz_url/biz_snapshot 快照与关联键;uk(sch_id, biz_type, biz_id) 幂等——
- * 同一业务对象一条办件,重复代发起返回原办件;驳回/撤回后重提复用同一行(R4 重启语义)。
+ * biz_title/biz_url/biz_snapshot 快照与关联键。正式命令每轮独立实例,唯一域为
+ * (school, sourcePrincipalId, bizType, bizId, applyRound),单在途由坐标表事务指针保护。
+ * sourcePrincipalId=NULL / legacySlot=1 仅保留迁移前旧入口的一业务一行语义;正式模式关闭旧写入口。
*
* version 是乐观锁(推进串行化)兼回调 payload 的 instanceVersion(业务方按它丢弃乱序旧事件),
* 由 JPA @Version 在 flush 时自增。
@@ -24,12 +25,19 @@ import java.time.LocalDateTime;
@Setter
@Entity
@Table(name = "flw_instance",
- uniqueConstraints = @UniqueConstraint(name = "uk_flw_instance_biz",
- columnNames = {"sch_id", "biz_type", "biz_id"}),
+ uniqueConstraints = {
+ @UniqueConstraint(name = "uk_flw_instance_legacy", columnNames = {"sch_id", "biz_type", "biz_id", "legacy_slot"}),
+ @UniqueConstraint(name = "uk_flw_instance_round", columnNames = {"sch_id", "source_principal_id", "biz_type", "biz_id", "apply_round"})},
indexes = {
@Index(name = "idx_flw_inst_submitter", columnList = "sch_id, submitter_id, status"),
@Index(name = "idx_flw_inst_status_time", columnList = "sch_id, status, submit_time")})
public class FlwInstance extends BaseEntity {
+ @Column(name="source_principal_id",length=36,updatable=false)
+ private String sourcePrincipalId;
+ /** 1 for legacy writers, NULL for formal commands. Migration must drop the old three-column UK. */
+ @Column(name="legacy_slot")
+ private Integer legacySlot = 1;
+
/** 租户键 */
@Column(name = "sch_id", nullable = false)
@@ -141,7 +149,7 @@ public class FlwInstance extends BaseEntity {
private String applyRequestToken;
/**
- * 当前发起轮次(1 起;随新一轮请求标识进位)。
+ * 当前发起轮次(1 起)。正式命令由服务端坐标锁内分配,不接受客户端 attempt 分轮。
*
*
轮次守卫是单调的:请求带来的 attempt 低于行上的 applyRound 就是陈旧轮次的迟到重投,
* 只回读现状。两侧(消费方命令行 + 本行)都持久化,崩溃恢复后仍能判出"哪一轮才是当前轮"。
@@ -149,6 +157,22 @@ public class FlwInstance extends BaseEntity {
@Column(name = "apply_round")
private Integer applyRound;
+ /**
+ * 受理时规范载荷指纹(2026-10-25 T04,SHA-256 十六进制):同 requestToken 必须同指纹——
+ * 同 ID 不同载荷是两份命令,撞在同一个 token 上按冲突拒绝,不伪装成幂等成功。
+ * 存量行(指纹列为 NULL)跳过该核对(守卫缺省是关,与 token/round 存量纪律一致)。
+ */
+ @Column(name = "apply_payload_hash", length = 64)
+ private String applyPayloadHash;
+
+ /**
+ * 事件序号计数器(2026-10-25 T05):实例内事件事实 eventSeq 的发号器。发号时在事务内
+ * 悲观锁住本行(findByIdForUpdate 当前读)再 +1——REPEATABLE READ 下对聚合做快照读会
+ * 拿到陈旧 max(FlowEngineIT 会签并发实证),行上计数器+行锁才是串行安全的。
+ */
+ @Column(name = "event_seq_counter")
+ private Integer eventSeqCounter;
+
/**
* 台账展示位(F11-4):该办件是否「路由挂起」(审批中且最新 attempt 的路由判断行无出口)。
* 非持久列——由管理端办件分页(FlowAdminController.instancePage)按页查询后填充,
diff --git a/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java b/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
index 946caf3..d31fc8e 100644
--- a/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
+++ b/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
@@ -39,6 +39,11 @@ public final class FlowApplyRetryModels {
* 这样"网络重投"与"用户授权重提"在中心侧也是两个不同的事实。
*/
private Boolean retryOnly;
+ /**
+ * 期望的前一轮次(T04 服务端轮次,可选):显式重提必须声明基于第 N 轮;
+ * 服务端在坐标锁内核对并只分配 N+1,不信客户端 attempt。null=既有语义。
+ */
+ private Integer expectedPrevRound;
}
/** v2 响应:在旧 {instanceId,duplicated} 之上把"中心判出来的是哪一种命中"说清楚 */
@@ -60,6 +65,9 @@ public final class FlowApplyRetryModels {
private String branch;
/** 行上的当前轮次(消费方据此校正自己的命令表认知) */
private Integer applyRound;
+ /** 同一响应的真实状态/实例版本;分支判据保持不变。 */
+ private Integer status;
+ private Integer version;
}
/** 分支码常量(与 {@link ApplyResultV2#branch} 同值域) */
@@ -71,6 +79,11 @@ public final class FlowApplyRetryModels {
public static final String REPLAY_ONLY = "REPLAY_ONLY";
public static final String RESTARTED = "RESTARTED";
public static final String CONCURRENT_CREATED = "CONCURRENT_CREATED";
+ /**
+ * 冲突(2026-10-25 T04):同 token 异载荷、或带 token 撞上他人/其他轮次的在途与通过件——
+ * 调用方(OapiFlowService)把该分支翻译成业务异常,不伪装幂等成功。
+ */
+ public static final String CONFLICT = "CONFLICT";
private Branch() {
}
diff --git a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
index 4760a6c..611203f 100644
--- a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
+++ b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
@@ -107,6 +107,13 @@ public final class OapiModels {
@Data
public static class InstanceVo {
private Long instanceId;
+ private Integer applyRound;
+ private String requestToken;
+ /**
+ * 来源应用快照(2026-10-25 T04):UNKNOWN 恢复链按"稳定来源+requestToken"核验命令归属的
+ * 依据——查得实例但 applyAppKey 与本应用不符=他人办件,不得认领(P0-4 关闭点)。
+ */
+ private String applyAppKey;
private Integer status;
private Integer version;
private String nodeSummary;
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
index 03839d6..ee0a176 100644
--- a/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
@@ -13,6 +13,9 @@ import java.util.List;
public interface FlwCallbackLogRepository extends JpaRepository {
+ /** T05 投递幂等:同一事件事实对同一订阅方只建一行(补分发与正常分发竞争时去重) */
+ boolean existsByEventIdAndRegId(String eventId, Long regId);
+
/** 重推扫描主路径:待推/重试中且到点 */
List findTop50ByStatusAndNextRetryTimeLessThanEqualOrderByNextRetryTimeAsc(
Integer status, LocalDateTime now);
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
index 72c2488..cf8552a 100644
--- a/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
@@ -14,7 +14,7 @@ public interface FlwCcRepository extends JpaRepository {
/** 抄送我的(用户面页签,走 user+sent_time 索引;appKey 可空=不按应用过滤,口径=定义归属 flw_def.app_key) */
@Query("select c from FlwCc c join FlwInstance i on i.id = c.instanceId join FlwDef d on d.id = i.defId "
+ "where c.schId = :schId and c.userId = :userId "
- + "and (:appKey is null or d.appKey = :appKey) order by c.sentTime desc")
+ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) order by c.sentTime desc")
Page pageCc(@Param("schId") Long schId, @Param("userId") Integer userId,
@Param("appKey") String appKey, Pageable pageable);
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
index 83f11b7..c582182 100644
--- a/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
@@ -32,12 +32,22 @@ public interface FlwDefRepository extends JpaRepository {
List findBySchIdOrderByUpdateTimeDesc(Long schId);
- /** 已发布流程列表(oapi list 数据源):每条流程线取最高已发布版本 */
- @Query("select d from FlwDef d where d.schId = :schId and d.status = 1 "
+ /**
+ * 已发布流程列表(oapi list 数据源):每条流程线取最高已发布版本。
+ * 2026-10-25 T03:外层追加 {@code lineOffline=0}——线级停发的整线不出现在新发起预览源,
+ * 且不存在"停发后回退旧版"的形态(内层仍是该线已发布最高版本,停发不改变版本事实)。
+ */
+ @Query("select d from FlwDef d where d.schId = :schId and d.status = 1 and d.lineOffline = 0 "
+ "and d.version = (select max(d2.version) from FlwDef d2 where d2.schId = d.schId "
+ "and d2.flowGuid = d.flowGuid and d2.status = 1) order by d.updateTime desc")
List findPublishedLatestPerGuid(@Param("schId") Long schId);
+ /** 线级停发/恢复(T03):批量 UPDATE 全版本行同值;返回受影响行数 */
+ @org.springframework.data.jpa.repository.Modifying
+ @Query("update FlwDef d set d.lineOffline = :offline where d.schId = :schId and d.flowGuid = :flowGuid")
+ int updateLineOffline(@Param("schId") Long schId, @Param("flowGuid") String flowGuid,
+ @Param("offline") Integer offline);
+
/** 在途统计(议题二④):一次取回多条流程线的全部版本行 */
List findBySchIdAndFlowGuidIn(Long schId, Collection flowGuids);
}
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
index 1cc57d6..05e6cd0 100644
--- a/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
@@ -12,8 +12,35 @@ import java.util.List;
import java.util.Optional;
public interface FlwInstanceRepository extends JpaRepository {
+ @org.springframework.data.jpa.repository.Lock(javax.persistence.LockModeType.PESSIMISTIC_WRITE)
+ @Query("select i from FlwInstance i where i.id=:id")
+ Optional findFormalPreviousForUpdate(@Param("id") Long id);
+
+
+ /**
+ * T05 事件序号发号(2026-10-25 FlowEngineIT 会签并发实证三轮修正,终形态):
+ * ①FOR UPDATE 标量当前读发号(取行 X 锁,任何隔离级别都看到已提交最新值)→
+ * ②按读到的值回写计数器(值已在我锁内计算,无竞态)→③事实行按该值落库。
+ * 纯快照读 max+1 会撞 uk;实体 merge 会触发 @Version 冲突;自增 UPDATE 后普通 SELECT
+ * 的读回在并发窗口同样不可靠——FOR UPDATE 标量读是唯一两全形态。
+ */
+ @org.springframework.data.jpa.repository.Query(
+ value = "select coalesce(event_seq_counter, 0) + 1 from flw_instance where id = :id for update",
+ nativeQuery = true)
+ Integer nextEventSeqForUpdate(@Param("id") Long id);
+
+ @org.springframework.data.jpa.repository.Modifying
+ @org.springframework.data.jpa.repository.Query(
+ value = "update flw_instance set event_seq_counter = :v where id = :id", nativeQuery = true)
+ int setEventSeqCounter(@Param("id") Long id, @Param("v") Integer v);
+
+ Optional findFirstBySchIdAndSourcePrincipalIdAndBizTypeAndBizIdOrderByApplyRoundDesc(Long school, String source, String type, String biz);
+
+ @Query("select i from FlwInstance i where i.schId=:school and i.sourcePrincipalId=:source and (:bizType is null or i.bizType=:bizType) and (:status is null or i.status=:status) order by i.id desc")
+ Page pageSelf(@Param("school") Long school,@Param("source") String source,@Param("bizType") String bizType,@Param("status") Integer status, Pageable pageable);
/** biz uk 幂等主键 */
+ @Query("select i from FlwInstance i where i.schId=?1 and i.bizType=?2 and i.bizId=?3 and i.sourcePrincipalId is null")
Optional findBySchIdAndBizTypeAndBizId(Long schId, String bizType, String bizId);
List findBySchIdAndSubmitterIdAndStatusOrderByIdDesc(Long schId, Integer submitterId, Integer status);
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
index dc21dfc..f196414 100644
--- a/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
@@ -13,6 +13,12 @@ import java.util.List;
public interface FlwTaskRepository extends JpaRepository {
+ /** Additive D09 query: the real transfer parent is in this set. Ambiguous generations must be rejected. */
+ @Query(value="select t.id from flw_task t join flw_task n on n.id=:newTaskId "
+ + "where t.instance_id=n.instance_id and t.node_id=n.node_id and t.status=5 "
+ + "and t.assignee_id=:operatorId and t.handle_time=n.arrive_time", nativeQuery=true)
+ List findTransferParentCandidates(@Param("newTaskId") Long newTaskId, @Param("operatorId") Integer operatorId);
+
List findByInstanceIdOrderByIdAsc(Long instanceId);
List findByInstanceIdAndStatus(Long instanceId, Integer status);
@@ -52,10 +58,13 @@ public interface FlwTaskRepository extends JpaRepository {
@Query("update FlwTask t set t.status = 3 where t.instanceId = :instanceId and t.status = 0 and t.id <> :exceptTaskId")
int invalidatePendingExcept(@Param("instanceId") Long instanceId, @Param("exceptTaskId") Long exceptTaskId);
+ @Query("select t from FlwTask t join FlwInstance i on i.id=t.instanceId where t.schId=:school and i.sourcePrincipalId=:source and t.assigneeId=:user and t.status=0 and (:type is null or i.bizType=:type) order by t.id asc")
+ Page pageSelfTodo(@Param("school") Long school,@Param("source") String source,@Param("user") Integer user,@Param("type") String type, Pageable pageable);
+
/** 我的待办(主路径,走 assignee 索引;appKey 可空=不按应用过滤,口径=办件定义归属应用 flw_def.app_key) */
@Query("select t from FlwTask t join FlwInstance i on i.id = t.instanceId join FlwDef d on d.id = i.defId "
+ "where t.schId = :schId and t.assigneeId = :assigneeId and t.status = 0 "
- + "and (:appKey is null or d.appKey = :appKey) "
+ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) "
+ "order by t.arriveTime asc, t.id asc")
Page pageTodo(@Param("schId") Long schId, @Param("assigneeId") Integer assigneeId,
@Param("appKey") String appKey, Pageable pageable);
@@ -84,7 +93,7 @@ public interface FlwTaskRepository extends JpaRepository {
@Query("select t from FlwTask t join FlwInstance i on i.id = t.instanceId join FlwDef d on d.id = i.defId "
+ "where t.schId = :schId and t.assigneeId = :assigneeId "
+ "and t.status in :statuses "
- + "and (:appKey is null or d.appKey = :appKey) "
+ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) "
+ "and t.id = (select max(t2.id) from FlwTask t2 where t2.instanceId = t.instanceId "
+ " and t2.assigneeId = :assigneeId and t2.status in :statuses) "
+ "order by t.handleTime desc, t.id desc")
diff --git a/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java b/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
index e8d717d..1725612 100644
--- a/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
+++ b/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
@@ -33,6 +33,7 @@ public class FlowCallbackAdminService {
private final FlwCallbackLogRepository logRepository;
private final CallbackPushService pushService;
private final CallbackProperties callbackProperties;
+ private final com.yunhai.flowcenter.callback.CallbackLogStore logStore;
public List listRegs(Long schId) {
return regRepository.findBySchIdOrderByUpdateTimeDesc(schId);
@@ -104,7 +105,6 @@ public class FlowCallbackAdminService {
* 管理端手工标记死信(持续重试语义下死信的唯一人工入口:人工判定事件无意义时止血)。
* 标记后可经既有 rePush 复活。
*/
- @Transactional
public void markDead(Long schId, Long logId) {
FlwCallbackLog callbackLog = logRepository.findById(logId)
.orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "推送日志不存在:" + logId));
@@ -119,7 +119,6 @@ public class FlowCallbackAdminService {
}
/** 手工重推(死信/失败日志):置回待推并立即触发一次推送 */
- @Transactional
public void rePush(Long schId, List logIds) {
if (logIds == null || logIds.isEmpty()) {
throw new BizException(ErrorCode.PARAM_ERROR, "请选择要重推的日志");
@@ -128,20 +127,9 @@ public class FlowCallbackAdminService {
if (logs.size() != logIds.size()) {
throw new BizException(ErrorCode.PARAM_ERROR, "存在不属于该校或不存在 dead/pending 日志");
}
- for (FlwCallbackLog callbackLog : logs) {
- if (callbackLog.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()) {
- continue;
- }
- callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
- callbackLog.setRetryCount(0);
- callbackLog.setNextRetryTime(LocalDateTime.now());
- logRepository.save(callbackLog);
- }
- logRepository.flush();
- for (FlwCallbackLog callbackLog : logs) {
- if (callbackLog.getStatus() != null) {
- pushService.pushAsync(callbackLog.getId());
- }
+ // 短事务提交并释放会话锁后才执行 HTTP;不能把管理端旧实体全量回写覆盖刚成功的状态。
+ for (Long id : logStore.resetForReplay(schId, logIds)) {
+ pushService.pushAsync(id);
}
}
diff --git a/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java b/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
index 001e9d1..da109e3 100644
--- a/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
+++ b/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
@@ -3,10 +3,12 @@ package com.yunhai.flowcenter.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhai.flowcenter.engine.DefValidator;
import com.yunhai.flowcenter.engine.FlowDefModel;
+import com.yunhai.flowcenter.engine.VariableContract;
import com.yunhai.flowcenter.engine.FlowEngine;
import com.yunhai.flowcenter.entity.FlwAssigneeMap;
import com.yunhai.flowcenter.entity.FlwDef;
import com.yunhai.flowcenter.enums.FlwEnums;
+import com.yunhai.flowcenter.enums.FormBindingErrorCode;
import com.yunhai.flowcenter.repository.FlwAssigneeMapRepository;
import com.yunhai.flowcenter.repository.FlwDefRepository;
import com.yunhai.platform.apistd.BizException;
@@ -192,12 +194,13 @@ public class FlowDefAdminService {
boolean pairing = formUuid != null && !clearing && formVersion == null;
boolean versionOnly = formUuid == null && formVersion != null;
if (pairing || versionOnly) {
- throw new BizException(ErrorCode.PARAM_ERROR,
+ throw new BizException(formVersion == null ? FormBindingErrorCode.VERSION_REQUIRED
+ : FormBindingErrorCode.INVALID_COORDINATE,
"关联表单必须成对提交 formUuid 与 formVersion(只传一个拒):当前 formUuid="
+ (formUuid == null ? "未传" : formUuid.trim()) + ",formVersion=" + formVersion);
}
if (formUuid != null && !clearing) {
- checkedSchema(schId, formUuid.trim(), formVersion); // Validate before the first save.
+ checkedSchema(schId, formUuid.trim(), formVersion, "保存"); // Validate before the first save.
}
FlwDef saved = save(schId, id, code, name, contentJson, baseGuid, appKey, changeLog);
if (formUuid != null) {
@@ -216,7 +219,7 @@ public class FlowDefAdminService {
* 2026-10-20 C1 第一刀:def 关联了表单(form_uuid 非空)时,发布期拉 form-center schema 快照
* 嵌入版本行 content_json.formContract(与 defId 快照纪律同构:在途办件按发起时快照校验),
* 并做表达式键查①红牌(变量 ⊆ 快照字段 ∪ 内置变量)。拉取失败=发布阻断(可重试);
- * 运行期对 form-center 零依赖。form_uuid 为空(存量线/Mode C)行为逐字不变。 */
+ * 运行期对 form-center 零依赖;新发布的无绑定版本必须显式声明 variables。 */
@Transactional
public FlwDef publish(Long schId, Long id) {
FlwDef def = requireDef(schId, id);
@@ -227,86 +230,33 @@ public class FlowDefAdminService {
throw new BizException(ErrorCode.PARAM_ERROR, "停用版本请走启用;修改表单关联须保存为新版本草稿");
}
FlowDefModel model = FlowDefModel.parse(objectMapper, def.getContentJson());
- DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
if (def.getFormUuid() != null && !def.getFormUuid().trim().isEmpty()) {
- Map snapshot;
- try {
- // 管理链始终使用确切版本;缺版本须重新选择,不回落当前发布版。
- snapshot = checkedSchema(schId, def.getFormUuid().trim(), def.getFormVersion());
- } catch (FormCenterSchemaClient.SchemaUnavailableException e) {
- throw new BizException(ErrorCode.PARAM_ERROR,
- "关联表单快照拉取失败,发布已阻断(可重试):" + e.getMessage());
- } catch (FormCenterSchemaClient.SchemaContractViolationException e) {
- throw new BizException(ErrorCode.PARAM_ERROR,
- "关联表单版本核验失败,发布已阻断(重试无意义,请重新选择表单与版本):" + e.getMessage());
- }
- List problems = new ArrayList<>();
- validateExpressionKeysAgainstSnapshot(model, snapshot, problems);
- validateExpressionTypesAndDomains(model, snapshot, problems);
- if (!problems.isEmpty()) {
- throw new BizException(ErrorCode.PARAM_ERROR,
- "发布校验失败(表达式三查):\n" + String.join("\n", problems));
- }
- def.setContentJson(embedFormContract(def.getContentJson(), snapshot));
+ Map snapshot = checkedSchema(schId, def.getFormUuid().trim(),
+ def.getFormVersion(), "发布");
+ model.variables = VariableContract.fromSnapshot(snapshot);
+ VariableContract.validateDefinition(model, true);
+ DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
+ def.setContentJson(embedFormContract(def.getContentJson(), snapshot, model.variables));
+ } else if (def.getFormVersion() != null) {
+ throw new BizException(FormBindingErrorCode.INVALID_COORDINATE, "关联表单版本必须同时绑定 formUuid");
+ }
+ if (def.getFormUuid() == null || def.getFormUuid().trim().isEmpty()) {
+ VariableContract.validateDefinition(model, true);
+ DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
}
def.setStatus(FlwEnums.DefStatus.PUBLISHED.getCode());
def.setPublishTime(LocalDateTime.now());
return defRepository.save(def);
}
- /**
- * 表达式键查①(C1 三查之一,红牌):对每条 EXPRESSION 判断的出边表达式,
- * 变量集 ⊆ 快照字段 keys ∪ 内置变量({@link DefValidator#EXPRESSION_BUILTINS})。
- * 不执行表达式、不改沙箱(口径与 v2 方案 §4.7 一致)。
- */
- private void validateExpressionKeysAgainstSnapshot(FlowDefModel model, Map snapshot,
- List problems) {
- @SuppressWarnings("unchecked")
- List