fix: persist workflow timer jobs
This commit is contained in:
xiangning
parent
e48f93aa04
commit
2eb6371c94
@ -1003,8 +1003,8 @@ public class ProcessEngine
|
||||
|
||||
/// <summary>
|
||||
/// 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<WorkflowDbContext>();
|
||||
var engine = scope.ServiceProvider.GetRequiredService<ProcessEngine>();
|
||||
|
||||
// 重新查 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();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@ -0,0 +1,37 @@
|
||||
namespace BuildFlow.Architecture.Tests;
|
||||
|
||||
using Xunit;
|
||||
|
||||
/// <summary>
|
||||
/// 工作流持久化任务和 Host 调度边界的源码架构测试。
|
||||
/// </summary>
|
||||
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，无法验证工作流任务架构边界。");
|
||||
}
|
||||
}
|
||||
@ -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()
|
||||
{
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user