diff --git a/src/modules/workflow/Application/Engine/ProcessEngine.cs b/src/modules/workflow/Application/Engine/ProcessEngine.cs index 791f5ea..17cf70b 100644 --- a/src/modules/workflow/Application/Engine/ProcessEngine.cs +++ b/src/modules/workflow/Application/Engine/ProcessEngine.cs @@ -1003,8 +1003,8 @@ public class ProcessEngine /// /// Timer（定时/延时）节点：消费当前 token，按 config.delayMinutes 等待后自动放行。 - /// 实现方式：在 token 上记录 DueAt（= 现在 + delayMinutes），由 OverdueTaskProcessor 兼容的 - /// TimerSchedulerService 扫描到期后重新投放新 token。若 delayMinutes 未配置或 ≤ 0，立即放行。 + /// 未来 Timer 只持久化 WorkflowTimerJob，由 Host 调度器通过 Application processor 到期恢复； + /// 不创建进程内延时任务，确保多实例和 Host 重启后仍可继续。若 delayMinutes 未配置或 ≤ 0，立即放行。 /// config: /// delayMinutes: int - 延迟分钟数（默认 0 = 立即放行） /// targetTime: ISO8601 - 指定放行时刻（与 delayMinutes 二选一，targetTime 优先） @@ -1061,77 +1061,22 @@ public class ProcessEngine return; } - // 有延迟：token 保持 Active，记录到期时间。 - // TimerSchedulerService 会扫描到该 token 的 DueAt（复用 token 的元数据字段）， - // 到期后消费它并向下游投放新 token。 - // 由于 WorkflowToken 没有 DueAt 字段，用 instance.Variables 记录 timer 状态： - // key = timer:{nodeId}:{tokenId}，value = ISO8601 到期时间 - var timerKey = $"timer:{node.Id}:{token.Id}"; - var vars = ParseVariables(instance.Variables); - vars[timerKey] = fireAt.Value.ToString("O"); - instance.Variables = JsonSerializer.Serialize(vars); - - await _dbContext.SaveChangesAsync(); - - // 异步触发：在后台等待到期后放行（不阻塞当前请求） - _ = Task.Run(async () => + // 有延迟：token 保持 Active，同一 token 只持久化一个待领取任务。 + var existingJob = await _dbContext.WorkflowTimerJobs + .FirstOrDefaultAsync(x => x.TokenId == token.Id); + if (existingJob is null) { - var delay = fireAt.Value - DateTime.UtcNow; - if (delay > TimeSpan.Zero) + _dbContext.WorkflowTimerJobs.Add(new WorkflowTimerJob { - await Task.Delay(delay); - } - - using var scope = _serviceProvider.CreateScope(); - var db = scope.ServiceProvider.GetRequiredService(); - var engine = scope.ServiceProvider.GetRequiredService(); - - // 重新查 instance 和 token（可能已被人工干预终止） - var inst = await db.WorkflowInstances.FindAsync(instance.Id); - var tok = await db.WorkflowTokens.FindAsync(token.Id); - - if (inst is null || tok is null || tok.Status != TokenStatus.Active - || inst.Status != InstanceStatus.Running) - return; - - // 消费 timer token，向下传播 - tok.Status = TokenStatus.Consumed; - tok.CompletedAt = DateTime.UtcNow; - - // 清理 variables 中的 timer 标记 - var instVars = ParseVariables(inst.Variables); - instVars.Remove(timerKey); - inst.Variables = JsonSerializer.Serialize(instVars); - - // 后台 scope 的 engine 需重新加载该实例的图（X6 JSON 单一事实源） - await engine.EnsureGraphLoadedForAsync(inst); - - var edges = await engine.GetOutgoingEdgesPublicAsync(node.Id); - var edge = edges.FirstOrDefault(); - - if (edge is not null) - { - var newToken = new WorkflowToken - { - Id = Guid.NewGuid(), - InstanceId = inst.Id, - NodeId = edge.TargetNodeId, - Status = TokenStatus.Active, - }; - db.WorkflowTokens.Add(newToken); - await db.SaveChangesAsync(); - - var targetNode = await engine.FindNodePublicAsync(edge.TargetNodeId); - if (targetNode is not null) - { - await engine.ProcessNodeAsync(inst, newToken, targetNode); - } - } - else - { - await db.SaveChangesAsync(); - } - }); + Id = Guid.NewGuid(), + InstanceId = instance.Id, + TokenId = token.Id, + NodeId = node.Id, + DueAt = fireAt.Value, + Status = WorkflowTimerJobStatus.Pending, + }); + await _dbContext.SaveChangesAsync(); + } } /// diff --git a/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs b/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs new file mode 100644 index 0000000..46a1999 --- /dev/null +++ b/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs @@ -0,0 +1,37 @@ +namespace BuildFlow.Architecture.Tests; + +using Xunit; + +/// +/// 工作流持久化任务和 Host 调度边界的源码架构测试。 +/// +public class WorkflowJobArchitectureTests +{ + [Fact] + public void ProcessEngineShouldNotUseProcessLocalTimerTasks() + { + var source = ReadSource("src/modules/workflow/Application/Engine/ProcessEngine.cs"); + + Assert.DoesNotContain("Task.Run", source); + Assert.DoesNotContain("timer:{", source); + } + + private static string ReadSource(string relativePath) + { + return File.ReadAllText(Path.Combine( + FindRepoRoot().FullName, + relativePath.Replace('/', Path.DirectorySeparatorChar))); + } + + private static DirectoryInfo FindRepoRoot() + { + var current = new DirectoryInfo(AppContext.BaseDirectory); + while (current is not null) + { + if (File.Exists(Path.Combine(current.FullName, "BuildFlow.slnx"))) return current; + current = current.Parent; + } + + throw new InvalidOperationException("无法定位 BuildFlow.slnx，无法验证工作流任务架构边界。"); + } +} diff --git a/tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs b/tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs index 69d25e7..67351b9 100644 --- a/tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs +++ b/tests/BuildFlow.Workflow.Tests/Engine/ProcessEngineTests.cs @@ -156,6 +156,50 @@ public class ProcessEngineTests task.DueAt.Should().BeNull(); } + // ============================================================ + // Timer Node + // ============================================================ + + [Fact] + public async Task ProcessTimerNode_WithFutureDueAt_PersistsSinglePendingJob() + { + var timerNode = CreateNode(NodeType.Timer); + timerNode.Config = """{ "delayMinutes": 10 }"""; + var nextNode = CreateNode(NodeType.Approval); + var (_, instance) = PersistSetup( + [timerNode, nextNode], + [CreateEdge(timerNode, nextNode)]); + instance.Status = InstanceStatus.Running; + var token = PersistToken(instance, timerNode); + var variablesBefore = instance.Variables; + + await _engine.ProcessNodeAsync(instance, token, timerNode); + + var job = await _dbContext.WorkflowTimerJobs.SingleAsync(); + job.InstanceId.Should().Be(instance.Id); + job.TokenId.Should().Be(token.Id); + job.NodeId.Should().Be(timerNode.Id); + job.Status.Should().Be(WorkflowTimerJobStatus.Pending); + job.DueAt.Should().BeAfter(DateTime.UtcNow.AddMinutes(9)); + token.Status.Should().Be(TokenStatus.Active); + instance.Variables.Should().Be(variablesBefore); + } + + [Fact] + public async Task ProcessTimerNode_ReenteredForSameToken_DoesNotCreateDuplicateJob() + { + var timerNode = CreateNode(NodeType.Timer); + timerNode.Config = """{ "delayMinutes": 10 }"""; + var (_, instance) = PersistSetup([timerNode], []); + instance.Status = InstanceStatus.Running; + var token = PersistToken(instance, timerNode); + + await _engine.ProcessNodeAsync(instance, token, timerNode); + await _engine.ProcessNodeAsync(instance, token, timerNode); + + (await _dbContext.WorkflowTimerJobs.CountAsync()).Should().Be(1); + } + [Fact] public async Task ProcessApprovalNode_TriggersTaskArrivedNotificationWhenServiceRegistered() {