build-flow/build-flow.api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: migrate workflow job reliability state

Browse Source
This commit is contained in:
xiangning 2026-07-24 03:03:57 +08:00
parent 42b9316f60
commit 4ba4769cc1
4 changed files with 2717 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2190
src/modules/workflow/Infrastructure/Persistence/Migrations/20260723110151_AddWorkflowJobReliability.Designer.cs generated Normal file
View File

File diff suppressed because it is too large Load Diff

153
src/modules/workflow/Infrastructure/Persistence/Migrations/20260723110151_AddWorkflowJobReliability.cs Normal file
View File

@ -0,0 +1,153 @@
using System;
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace Workflow.Infrastructure.Persistence.Migrations
{
/// <inheritdoc />
public partial class AddWorkflowJobReliability : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropIndex(
name: "ix_b_import_jobs_status",
table: "b_import_jobs");
migrationBuilder.AddColumn<int>(
name: "attempt_count",
table: "b_import_jobs",
type: "integer",
nullable: false,
defaultValue: 0,
comment: "累计领取次数");
migrationBuilder.AddColumn<DateTime>(
name: "last_heartbeat_at",
table: "b_import_jobs",
type: "timestamp with time zone",
nullable: true,
comment: "最近一次处理心跳时间（UTC）");
migrationBuilder.AddColumn<DateTime>(
name: "lease_expires_at",
table: "b_import_jobs",
type: "timestamp with time zone",
nullable: true,
comment: "租约到期时间（UTC）");
migrationBuilder.AddColumn<string>(
name: "lease_owner",
table: "b_import_jobs",
type: "character varying(200)",
maxLength: 200,
nullable: true,
comment: "当前处理租约的Host实例标识");
migrationBuilder.CreateTable(
name: "b_workflow_timer_jobs",
columns: table => new
{
id = table.Column<Guid>(type: "uuid", nullable: false, comment: "主键ID"),
instance_id = table.Column<Guid>(type: "uuid", nullable: false, comment: "所属流程实例ID"),
token_id = table.Column<Guid>(type: "uuid", nullable: false, comment: "等待到期的源令牌ID"),
node_id = table.Column<Guid>(type: "uuid", nullable: false, comment: "Timer节点ID"),
due_at = table.Column<DateTime>(type: "timestamp with time zone", nullable: false, comment: "计划触发时间（UTC）"),
status = table.Column<int>(type: "integer", nullable: false, comment: "任务状态（Pending/Processing/Completed/Failed/Cancelled）"),
lease_owner = table.Column<string>(type: "character varying(200)", maxLength: 200, nullable: true, comment: "当前租约持有者"),
lease_expires_at = table.Column<DateTime>(type: "timestamp with time zone", nullable: true, comment: "租约到期时间（UTC）"),
attempt_count = table.Column<int>(type: "integer", nullable: false, defaultValue: 0, comment: "累计领取次数"),
last_error = table.Column<string>(type: "character varying(2000)", maxLength: 2000, nullable: true, comment: "最近一次失败原因"),
completed_at = table.Column<DateTime>(type: "timestamp with time zone", nullable: true, comment: "进入终态的时间（UTC）"),
created_by = table.Column<string>(type: "character varying(100)", maxLength: 100, nullable: false, comment: "创建人ID"),
created_at = table.Column<DateTime>(type: "timestamp with time zone", nullable: false, comment: "创建时间"),
updated_by = table.Column<string>(type: "character varying(100)", maxLength: 100, nullable: false, comment: "更新人ID"),
updated_at = table.Column<DateTime>(type: "timestamp with time zone", nullable: false, comment: "更新时间")
},
constraints: table =>
{
table.PrimaryKey("pk_b_workflow_timer_jobs", x => x.id);
table.ForeignKey(
name: "fk_b_workflow_timer_jobs_b_workflow_instances_instance_id",
column: x => x.instance_id,
principalTable: "b_workflow_instances",
principalColumn: "id",
onDelete: ReferentialAction.Cascade);
table.ForeignKey(
name: "fk_b_workflow_timer_jobs_workflow_tokens_token_id",
column: x => x.token_id,
principalTable: "b_workflow_tokens",
principalColumn: "id",
onDelete: ReferentialAction.Cascade);
});
migrationBuilder.CreateIndex(
name: "ix_b_import_jobs_status_created_at",
table: "b_import_jobs",
columns: new[] { "status", "created_at" });
migrationBuilder.CreateIndex(
name: "ix_b_import_jobs_status_lease_expires_at",
table: "b_import_jobs",
columns: new[] { "status", "lease_expires_at" });
migrationBuilder.CreateIndex(
name: "ix_b_workflow_timer_jobs_instance_id",
table: "b_workflow_timer_jobs",
column: "instance_id");
migrationBuilder.CreateIndex(
name: "ix_b_workflow_timer_jobs_status_due_at",
table: "b_workflow_timer_jobs",
columns: new[] { "status", "due_at" });
migrationBuilder.CreateIndex(
name: "ix_b_workflow_timer_jobs_status_lease_expires_at",
table: "b_workflow_timer_jobs",
columns: new[] { "status", "lease_expires_at" });
migrationBuilder.CreateIndex(
name: "ix_b_workflow_timer_jobs_token_id",
table: "b_workflow_timer_jobs",
column: "token_id",
unique: true);
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropTable(
name: "b_workflow_timer_jobs");
migrationBuilder.DropIndex(
name: "ix_b_import_jobs_status_created_at",
table: "b_import_jobs");
migrationBuilder.DropIndex(
name: "ix_b_import_jobs_status_lease_expires_at",
table: "b_import_jobs");
migrationBuilder.DropColumn(
name: "attempt_count",
table: "b_import_jobs");
migrationBuilder.DropColumn(
name: "last_heartbeat_at",
table: "b_import_jobs");
migrationBuilder.DropColumn(
name: "lease_expires_at",
table: "b_import_jobs");
migrationBuilder.DropColumn(
name: "lease_owner",
table: "b_import_jobs");
migrationBuilder.CreateIndex(
name: "ix_b_import_jobs_status",
table: "b_import_jobs",
column: "status");
}
}
}

153
src/modules/workflow/Infrastructure/Persistence/Migrations/WorkflowDbContextModelSnapshot.cs
View File

@ -455,6 +455,13 @@ namespace BuildFlow.Workflow.Infrastructure.Persistence.Migrations
.HasColumnName("id")
.HasComment("主键ID");
b.Property<int>("AttemptCount")
.ValueGeneratedOnAdd()
.HasColumnType("integer")
.HasDefaultValue(0)
.HasColumnName("attempt_count")
.HasComment("累计领取次数");
b.Property<DateTime>("CreatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("created_at")
@ -507,6 +514,23 @@ namespace BuildFlow.Workflow.Infrastructure.Persistence.Migrations
.HasColumnName("is_deleted")
.HasComment("是否软删除");
b.Property<DateTime?>("LastHeartbeatAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("last_heartbeat_at")
.HasComment("最近一次处理心跳时间（UTC）");
b.Property<DateTime?>("LeaseExpiresAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("lease_expires_at")
.HasComment("租约到期时间（UTC）");
b.Property<string>("LeaseOwner")
.IsConcurrencyToken()
.HasMaxLength(200)
.HasColumnType("character varying(200)")
.HasColumnName("lease_owner")
.HasComment("当前处理租约的Host实例标识");
b.Property<string>("OperatorIP")
.IsRequired()
.HasMaxLength(500)
@ -564,8 +588,11 @@ namespace BuildFlow.Workflow.Infrastructure.Persistence.Migrations
b.HasIndex("FormDefinitionId")
.HasDatabaseName("ix_b_import_jobs_form_definition_id");
b.HasIndex("Status")
.HasDatabaseName("ix_b_import_jobs_status");
b.HasIndex("Status", "CreatedAt")
.HasDatabaseName("ix_b_import_jobs_status_created_at");
b.HasIndex("Status", "LeaseExpiresAt")
.HasDatabaseName("ix_b_import_jobs_status_lease_expires_at");
b.ToTable("b_import_jobs", (string)null);
});
@ -1900,6 +1927,111 @@ namespace BuildFlow.Workflow.Infrastructure.Persistence.Migrations
b.ToTable("b_workflow_tasks", (string)null);
});
modelBuilder.Entity("Workflow.Domain.Entities.WorkflowTimerJob", b =>
{
b.Property<Guid>("Id")
.HasColumnType("uuid")
.HasColumnName("id")
.HasComment("主键ID");
b.Property<int>("AttemptCount")
.ValueGeneratedOnAdd()
.HasColumnType("integer")
.HasDefaultValue(0)
.HasColumnName("attempt_count")
.HasComment("累计领取次数");
b.Property<DateTime?>("CompletedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("completed_at")
.HasComment("进入终态的时间（UTC）");
b.Property<DateTime>("CreatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("created_at")
.HasComment("创建时间");
b.Property<string>("CreatedBy")
.IsRequired()
.HasMaxLength(100)
.HasColumnType("character varying(100)")
.HasColumnName("created_by")
.HasComment("创建人ID");
b.Property<DateTime>("DueAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("due_at")
.HasComment("计划触发时间（UTC）");
b.Property<Guid>("InstanceId")
.HasColumnType("uuid")
.HasColumnName("instance_id")
.HasComment("所属流程实例ID");
b.Property<string>("LastError")
.HasMaxLength(2000)
.HasColumnType("character varying(2000)")
.HasColumnName("last_error")
.HasComment("最近一次失败原因");
b.Property<DateTime?>("LeaseExpiresAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("lease_expires_at")
.HasComment("租约到期时间（UTC）");
b.Property<string>("LeaseOwner")
.IsConcurrencyToken()
.HasMaxLength(200)
.HasColumnType("character varying(200)")
.HasColumnName("lease_owner")
.HasComment("当前租约持有者");
b.Property<Guid>("NodeId")
.HasColumnType("uuid")
.HasColumnName("node_id")
.HasComment("Timer节点ID");
b.Property<int>("Status")
.HasColumnType("integer")
.HasColumnName("status")
.HasComment("任务状态（Pending/Processing/Completed/Failed/Cancelled）");
b.Property<Guid>("TokenId")
.HasColumnType("uuid")
.HasColumnName("token_id")
.HasComment("等待到期的源令牌ID");
b.Property<DateTime>("UpdatedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("updated_at")
.HasComment("更新时间");
b.Property<string>("UpdatedBy")
.IsRequired()
.HasMaxLength(100)
.HasColumnType("character varying(100)")
.HasColumnName("updated_by")
.HasComment("更新人ID");
b.HasKey("Id")
.HasName("pk_b_workflow_timer_jobs");
b.HasIndex("InstanceId")
.HasDatabaseName("ix_b_workflow_timer_jobs_instance_id");
b.HasIndex("TokenId")
.IsUnique()
.HasDatabaseName("ix_b_workflow_timer_jobs_token_id");
b.HasIndex("Status", "DueAt")
.HasDatabaseName("ix_b_workflow_timer_jobs_status_due_at");
b.HasIndex("Status", "LeaseExpiresAt")
.HasDatabaseName("ix_b_workflow_timer_jobs_status_lease_expires_at");
b.ToTable("b_workflow_timer_jobs", (string)null);
});
modelBuilder.Entity("Workflow.Domain.Entities.WorkflowToken", b =>
{
b.Property<Guid>("Id")
@ -2009,6 +2141,23 @@ namespace BuildFlow.Workflow.Infrastructure.Persistence.Migrations
.HasConstraintName("fk_b_workflow_tasks_b_workflow_instances_instance_id");
});
modelBuilder.Entity("Workflow.Domain.Entities.WorkflowTimerJob", b =>
{
b.HasOne("Workflow.Domain.Entities.WorkflowInstance", null)
.WithMany()
.HasForeignKey("InstanceId")
.OnDelete(DeleteBehavior.Cascade)
.IsRequired()
.HasConstraintName("fk_b_workflow_timer_jobs_b_workflow_instances_instance_id");
b.HasOne("Workflow.Domain.Entities.WorkflowToken", null)
.WithOne()
.HasForeignKey("Workflow.Domain.Entities.WorkflowTimerJob", "TokenId")
.OnDelete(DeleteBehavior.Cascade)
.IsRequired()
.HasConstraintName("fk_b_workflow_timer_jobs_workflow_tokens_token_id");
});
modelBuilder.Entity("Workflow.Domain.Entities.WorkflowToken", b =>
{
b.HasOne("Workflow.Domain.Entities.WorkflowInstance", null)

223
tests/BuildFlow.Workflow.Tests/Reliability/WorkflowJobMigrationTests.cs Normal file
View File

@ -0,0 +1,223 @@
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Npgsql;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
using Xunit;
namespace BuildFlow.Workflow.Tests.Reliability;
/// <summary>
/// 工作流可靠性迁移的空库安装与旧结构升级测试。
/// </summary>
public class WorkflowJobMigrationTests
{
private const string PreviousMigration =
"20260719041812_FixWorkflowTaskAttachmentsColumn";
private const string DefaultConnection =
"Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres";
[Fact]
public async Task EmptyDatabaseMigratesToWorkflowJobReliabilitySchema()
{
var databaseName = $"workflow_migration_empty_{Guid.NewGuid():N}";
var connectionString = await CreateDatabaseAsync(databaseName);
try
{
await using var db = CreateDbContext(connectionString);
await db.Database.MigrateAsync();
await using var connection = new NpgsqlConnection(connectionString);
await connection.OpenAsync();
(await TableExistsAsync(connection, "b_workflow_timer_jobs")).Should().BeTrue();
var columns = await ReadColumnNamesAsync(connection, "b_import_jobs");
columns.Should().Contain(
"lease_owner",
"lease_expires_at",
"attempt_count",
"last_heartbeat_at");
}
finally
{
await DropDatabaseAsync(databaseName);
}
}
[Fact]
public async Task ExistingImportJobSurvivesReliabilityMigration()
{
var databaseName = $"workflow_migration_upgrade_{Guid.NewGuid():N}";
var connectionString = await CreateDatabaseAsync(databaseName);
var jobId = Guid.NewGuid();
var formDefinitionId = Guid.NewGuid();
const string filePath = "/tmp/existing-import.xlsx";
try
{
await using var db = CreateDbContext(connectionString);
var migrator = db.GetService<IMigrator>();
await migrator.MigrateAsync(PreviousMigration);
var now = DateTime.UtcNow;
await db.Database.ExecuteSqlInterpolatedAsync($"""
INSERT INTO b_import_jobs (
id,
form_definition_id,
status,
file_url,
total_rows,
success_rows,
failed_rows,
created_by,
created_at,
updated_by,
updated_at,
is_deleted,
operator_ip)
VALUES (
{jobId},
{formDefinitionId},
{(int)BuildFlowImportJobStatus.Pending},
{filePath},
{0},
{0},
{0},
{"migration-test"},
{now},
{"migration-test"},
{now},
{false},
{"127.0.0.1"})
""");
await db.Database.MigrateAsync();
await using var connection = new NpgsqlConnection(connectionString);
await connection.OpenAsync();
await using var command = connection.CreateCommand();
command.CommandText = """
SELECT id,
status,
file_url,
attempt_count,
lease_owner,
lease_expires_at,
last_heartbeat_at
FROM b_import_jobs
WHERE id = @job_id
""";
command.Parameters.AddWithValue("job_id", jobId);
await using var reader = await command.ExecuteReaderAsync();
(await reader.ReadAsync()).Should().BeTrue();
reader.GetGuid(0).Should().Be(jobId);
reader.GetInt32(1).Should().Be((int)BuildFlowImportJobStatus.Pending);
reader.GetString(2).Should().Be(filePath);
reader.GetInt32(3).Should().Be(0);
reader.IsDBNull(4).Should().BeTrue();
reader.IsDBNull(5).Should().BeTrue();
reader.IsDBNull(6).Should().BeTrue();
}
finally
{
await DropDatabaseAsync(databaseName);
}
}
private static WorkflowDbContext CreateDbContext(string connectionString)
{
var options = new DbContextOptionsBuilder<WorkflowDbContext>()
.UseNpgsql(connectionString)
.UseSnakeCaseNamingConvention()
.Options;
return new WorkflowDbContext(options);
}
private static async Task<string> CreateDatabaseAsync(string databaseName)
{
var configured = Environment.GetEnvironmentVariable("BUILDFLOW_WORKFLOW_TEST_CONNECTION")
?? DefaultConnection;
var adminBuilder = new NpgsqlConnectionStringBuilder(configured)
{
Database = "postgres",
Pooling = false,
};
await using (var connection = new NpgsqlConnection(adminBuilder.ConnectionString))
{
await connection.OpenAsync();
await using var command = connection.CreateCommand();
command.CommandText = $"CREATE DATABASE \"{databaseName}\"";
await command.ExecuteNonQueryAsync();
}
return new NpgsqlConnectionStringBuilder(configured)
{
Database = databaseName,
Pooling = true,
}.ConnectionString;
}
private static async Task DropDatabaseAsync(string databaseName)
{
NpgsqlConnection.ClearAllPools();
var configured = Environment.GetEnvironmentVariable("BUILDFLOW_WORKFLOW_TEST_CONNECTION")
?? DefaultConnection;
var adminBuilder = new NpgsqlConnectionStringBuilder(configured)
{
Database = "postgres",
Pooling = false,
};
await using var connection = new NpgsqlConnection(adminBuilder.ConnectionString);
await connection.OpenAsync();
await using (var terminate = connection.CreateCommand())
{
terminate.CommandText =
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = @database AND pid <> pg_backend_pid()";
terminate.Parameters.AddWithValue("database", databaseName);
await terminate.ExecuteNonQueryAsync();
}
await using var drop = connection.CreateCommand();
drop.CommandText = $"DROP DATABASE IF EXISTS \"{databaseName}\"";
await drop.ExecuteNonQueryAsync();
}
private static async Task<bool> TableExistsAsync(
NpgsqlConnection connection,
string tableName)
{
await using var command = connection.CreateCommand();
command.CommandText = """
SELECT EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = @table_name)
""";
command.Parameters.AddWithValue("table_name", tableName);
return (bool)(await command.ExecuteScalarAsync())!;
}
private static async Task<List<string>> ReadColumnNamesAsync(
NpgsqlConnection connection,
string tableName)
{
await using var command = connection.CreateCommand();
command.CommandText = """
SELECT column_name
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = @table_name
""";
command.Parameters.AddWithValue("table_name", tableName);
var columns = new List<string>();
await using var reader = await command.ExecuteReaderAsync();
while (await reader.ReadAsync())
{
columns.Add(reader.GetString(0));
}
return columns;
}
}