perf: batch monitoring record upserts
This commit is contained in:
xiangning
parent
65566dc112
commit
d371ca59db
@ -1,3 +1,6 @@
|
||||
using System.Data;
|
||||
using System.Data.Common;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
@ -27,6 +30,8 @@ public class SyncProcessor(
|
||||
IServiceProvider services,
|
||||
ILogger<SyncProcessor> logger)
|
||||
{
|
||||
private const int RecordsPerUpsert = 500;
|
||||
|
||||
/// <summary>对单个外部系统执行一次同步。返回本次 SyncJobEntity 记录（已落库）。</summary>
|
||||
public async Task<SyncJobEntity> ExecuteAsync(ExternalSystemEntity system, string triggerSource, CancellationToken ct)
|
||||
{
|
||||
@ -55,31 +60,41 @@ public class SyncProcessor(
|
||||
try
|
||||
{
|
||||
var result = await connector.FetchAsync(system, ct);
|
||||
var records = NormalizeRecords(system.Id, job.Id, result.Records);
|
||||
|
||||
// 4. 幂等落库 SyncedRecordEntity
|
||||
await PersistRecordsAsync(system.Id, job.Id, result.Records, ct);
|
||||
|
||||
// 5. 更新 SyncJobEntity 为 Succeeded（汇总指标写入 json_data）
|
||||
job.Status = SyncJobStatusEnum.Succeeded;
|
||||
job.FinishedAt = DateTime.UtcNow;
|
||||
job.DurationMs = (long)(job.FinishedAt.Value - job.StartedAt!.Value).TotalMilliseconds;
|
||||
job.RecordCount = result.RecordCount;
|
||||
job.Data = new JobData
|
||||
await using (var transaction = await db.Database.BeginTransactionAsync(ct))
|
||||
{
|
||||
TriggerSource = triggerSource,
|
||||
SummaryMetrics = result.SummaryMetrics,
|
||||
ErrorMessage = null,
|
||||
};
|
||||
await db.SaveChangesAsync(ct);
|
||||
try
|
||||
{
|
||||
await PersistRecordsAsync(records, ct);
|
||||
|
||||
// 6. 更新 ExternalSystemEntity 的最近同步信息（写入 json_data）
|
||||
var sysData = system.Data;
|
||||
sysData.LastSyncedAt = new DateTimeOffset(job.FinishedAt.Value).ToUnixTimeMilliseconds();
|
||||
sysData.LastSyncJobId = job.Id;
|
||||
system.Data = sysData; // 触发序列化回 JsonData
|
||||
await db.SaveChangesAsync(ct);
|
||||
job.Status = SyncJobStatusEnum.Succeeded;
|
||||
job.FinishedAt = DateTime.UtcNow;
|
||||
job.DurationMs = (long)(job.FinishedAt.Value - job.StartedAt!.Value).TotalMilliseconds;
|
||||
job.RecordCount = result.RecordCount;
|
||||
job.Data = new JobData
|
||||
{
|
||||
TriggerSource = triggerSource,
|
||||
SummaryMetrics = result.SummaryMetrics,
|
||||
ErrorMessage = null,
|
||||
};
|
||||
|
||||
// 7. 触发告警评估（失败不影响主流程）
|
||||
var sysData = system.Data;
|
||||
sysData.LastSyncedAt = new DateTimeOffset(job.FinishedAt.Value).ToUnixTimeMilliseconds();
|
||||
sysData.LastSyncJobId = job.Id;
|
||||
system.Data = sysData;
|
||||
|
||||
await db.SaveChangesAsync(ct);
|
||||
await transaction.CommitAsync(ct);
|
||||
}
|
||||
catch
|
||||
{
|
||||
await transaction.RollbackAsync(CancellationToken.None);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
// 成功数据提交后再评估告警，避免告警读取到未提交的同步结果。
|
||||
try
|
||||
{
|
||||
await EvaluateAlertsAsync(system.Id, result.SummaryMetrics, ct);
|
||||
@ -95,63 +110,135 @@ public class SyncProcessor(
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await ReloadAfterFailureAsync(system, job, ct);
|
||||
return await RecordFailureAsync(system, triggerSource, ex.Message, job, ct);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>UPSERT 同步记录：同一 (ExternalSystemId, ExternalId, RecordType) 只保留最新。</summary>
|
||||
/// <summary>按固定上限执行参数化 UPSERT，同一业务唯一键只保留最新快照。</summary>
|
||||
private async Task PersistRecordsAsync(
|
||||
Guid systemId, Guid jobId, IReadOnlyList<SyncedRecordData> records, CancellationToken ct)
|
||||
IReadOnlyList<SyncedRecordEntity> records,
|
||||
CancellationToken ct)
|
||||
{
|
||||
if (records.Count == 0) return;
|
||||
|
||||
foreach (var data in records)
|
||||
foreach (var batch in records.Chunk(RecordsPerUpsert))
|
||||
{
|
||||
var payload = ParsePayload(data.RawJson);
|
||||
var existing = await db.SyncedRecords.FirstOrDefaultAsync(
|
||||
r => r.ExternalSystemId == systemId
|
||||
&& r.ExternalId == data.ExternalId
|
||||
&& r.RecordType == data.RecordType, ct);
|
||||
|
||||
if (existing is not null)
|
||||
{
|
||||
existing.SyncJobId = jobId;
|
||||
existing.OccurredAt = data.OccurredAt;
|
||||
existing.Data = new RecordData
|
||||
{
|
||||
ExternalId = data.ExternalId,
|
||||
RecordType = data.RecordType,
|
||||
Payload = payload,
|
||||
};
|
||||
existing.UpdatedAt = DateTime.UtcNow;
|
||||
}
|
||||
else
|
||||
{
|
||||
db.SyncedRecords.Add(new SyncedRecordEntity
|
||||
{
|
||||
ExternalSystemId = systemId,
|
||||
SyncJobId = jobId,
|
||||
ExternalId = data.ExternalId,
|
||||
RecordType = data.RecordType,
|
||||
OccurredAt = data.OccurredAt,
|
||||
Data = new RecordData
|
||||
{
|
||||
ExternalId = data.ExternalId,
|
||||
RecordType = data.RecordType,
|
||||
Payload = payload,
|
||||
},
|
||||
});
|
||||
}
|
||||
await ExecuteUpsertBatchAsync(batch, ct);
|
||||
}
|
||||
await db.SaveChangesAsync(ct);
|
||||
}
|
||||
|
||||
/// <summary>把原始 JSON 解析为 payload 字典（异常返回原始字符串包裹的单键字典）。</summary>
|
||||
private static Dictionary<string, object?> ParsePayload(string rawJson)
|
||||
private IReadOnlyList<SyncedRecordEntity> NormalizeRecords(
|
||||
Guid systemId,
|
||||
Guid jobId,
|
||||
IReadOnlyList<SyncedRecordData> records)
|
||||
{
|
||||
if (records.Count == 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
var normalized = new List<SyncedRecordEntity>(records.Count);
|
||||
var now = DateTime.UtcNow;
|
||||
for (var index = 0; index < records.Count; index++)
|
||||
{
|
||||
var data = records[index];
|
||||
normalized.Add(new SyncedRecordEntity
|
||||
{
|
||||
ExternalSystemId = systemId,
|
||||
SyncJobId = jobId,
|
||||
ExternalId = data.ExternalId,
|
||||
RecordType = data.RecordType,
|
||||
OccurredAt = data.OccurredAt,
|
||||
Data = new RecordData
|
||||
{
|
||||
ExternalId = data.ExternalId,
|
||||
RecordType = data.RecordType,
|
||||
Payload = ParsePayload(systemId, data),
|
||||
},
|
||||
CreatedBy = "system",
|
||||
CreatedAt = now,
|
||||
UpdatedBy = "system",
|
||||
UpdatedAt = now,
|
||||
IsDeleted = false,
|
||||
OperatorIP = "system",
|
||||
});
|
||||
}
|
||||
|
||||
return normalized;
|
||||
}
|
||||
|
||||
private async Task ExecuteUpsertBatchAsync(
|
||||
IReadOnlyList<SyncedRecordEntity> records,
|
||||
CancellationToken ct)
|
||||
{
|
||||
await using var command = db.Database.GetDbConnection().CreateCommand();
|
||||
var sql = new StringBuilder(
|
||||
"""
|
||||
INSERT INTO b_synced_records (
|
||||
id, external_system_id, sync_job_id, external_id, record_type,
|
||||
occurred_at, json_data, created_by, created_at, updated_by,
|
||||
updated_at, is_deleted, operator_ip)
|
||||
VALUES
|
||||
""");
|
||||
|
||||
for (var index = 0; index < records.Count; index++)
|
||||
{
|
||||
if (index > 0)
|
||||
{
|
||||
sql.Append(',');
|
||||
}
|
||||
|
||||
sql.AppendLine();
|
||||
sql.Append($"(@id_{index}, @system_{index}, @job_{index}, @external_{index}, @type_{index}, ");
|
||||
sql.Append($"@occurred_{index}, CAST(@json_{index} AS jsonb), @created_by_{index}, ");
|
||||
sql.Append($"@created_at_{index}, @updated_by_{index}, @updated_at_{index}, ");
|
||||
sql.Append($"@deleted_{index}, @operator_ip_{index})");
|
||||
|
||||
var record = records[index];
|
||||
AddParameter(command, $"id_{index}", record.Id, DbType.Guid);
|
||||
AddParameter(command, $"system_{index}", record.ExternalSystemId, DbType.Guid);
|
||||
AddParameter(command, $"job_{index}", record.SyncJobId, DbType.Guid);
|
||||
AddParameter(command, $"external_{index}", record.ExternalId, DbType.String);
|
||||
AddParameter(command, $"type_{index}", record.RecordType, DbType.String);
|
||||
AddParameter(command, $"occurred_{index}", record.OccurredAt, DbType.DateTime);
|
||||
AddParameter(command, $"json_{index}", record.JsonData, DbType.String);
|
||||
AddParameter(command, $"created_by_{index}", record.CreatedBy, DbType.String);
|
||||
AddParameter(command, $"created_at_{index}", record.CreatedAt, DbType.DateTime);
|
||||
AddParameter(command, $"updated_by_{index}", record.UpdatedBy, DbType.String);
|
||||
AddParameter(command, $"updated_at_{index}", record.UpdatedAt, DbType.DateTime);
|
||||
AddParameter(command, $"deleted_{index}", record.IsDeleted, DbType.Boolean);
|
||||
AddParameter(command, $"operator_ip_{index}", record.OperatorIP, DbType.String);
|
||||
}
|
||||
|
||||
sql.AppendLine();
|
||||
sql.Append(
|
||||
"""
|
||||
ON CONFLICT (external_system_id, external_id, record_type)
|
||||
DO UPDATE SET
|
||||
sync_job_id = EXCLUDED.sync_job_id,
|
||||
occurred_at = EXCLUDED.occurred_at,
|
||||
json_data = EXCLUDED.json_data,
|
||||
updated_by = EXCLUDED.updated_by,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
operator_ip = EXCLUDED.operator_ip
|
||||
""");
|
||||
command.CommandText = sql.ToString();
|
||||
await db.Database.ExecuteSqlRawAsync(
|
||||
command.CommandText,
|
||||
command.Parameters.Cast<object>(),
|
||||
ct);
|
||||
}
|
||||
|
||||
/// <summary>把原始 JSON 解析为 payload 字典；非法内容保留原文但日志只记录标识符。</summary>
|
||||
private Dictionary<string, object?> ParsePayload(Guid systemId, SyncedRecordData data)
|
||||
{
|
||||
try
|
||||
{
|
||||
var doc = JsonDocument.Parse(rawJson);
|
||||
using var doc = JsonDocument.Parse(data.RawJson);
|
||||
if (doc.RootElement.ValueKind != JsonValueKind.Object)
|
||||
{
|
||||
throw new JsonException("同步记录 JSON 根节点必须是对象");
|
||||
}
|
||||
|
||||
var dict = new Dictionary<string, object?>();
|
||||
foreach (var prop in doc.RootElement.EnumerateObject())
|
||||
{
|
||||
@ -167,12 +254,39 @@ public class SyncProcessor(
|
||||
}
|
||||
return dict;
|
||||
}
|
||||
catch
|
||||
catch (JsonException)
|
||||
{
|
||||
return new Dictionary<string, object?> { ["_raw"] = rawJson };
|
||||
logger.LogWarning(
|
||||
"同步记录 JSON 无效，保留原始内容。系统 {SystemId}，外部记录 {ExternalId}，类型 {RecordType}",
|
||||
systemId,
|
||||
data.ExternalId,
|
||||
data.RecordType);
|
||||
return new Dictionary<string, object?> { ["_raw"] = data.RawJson };
|
||||
}
|
||||
}
|
||||
|
||||
private static void AddParameter(
|
||||
DbCommand command,
|
||||
string name,
|
||||
object? value,
|
||||
DbType dbType)
|
||||
{
|
||||
var parameter = command.CreateParameter();
|
||||
parameter.ParameterName = name;
|
||||
parameter.DbType = dbType;
|
||||
parameter.Value = value ?? DBNull.Value;
|
||||
command.Parameters.Add(parameter);
|
||||
}
|
||||
|
||||
private async Task ReloadAfterFailureAsync(
|
||||
ExternalSystemEntity system,
|
||||
SyncJobEntity job,
|
||||
CancellationToken ct)
|
||||
{
|
||||
await db.Entry(job).ReloadAsync(ct);
|
||||
await db.Entry(system).ReloadAsync(ct);
|
||||
}
|
||||
|
||||
/// <summary>记录失败任务（job 已存在则更新，否则新建）。</summary>
|
||||
private async Task<SyncJobEntity> RecordFailureAsync(
|
||||
ExternalSystemEntity system, string triggerSource, string error, SyncJobEntity? job, CancellationToken ct)
|
||||
|
||||
@ -0,0 +1,125 @@
|
||||
using System.Data.Common;
|
||||
using Microsoft.EntityFrameworkCore.Diagnostics;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// 统计同步记录查询和 UPSERT 命令，并可让两个旧式先查后写查询在同一点汇合以稳定复现竞争。
|
||||
/// </summary>
|
||||
public sealed class CountingUpsertInterceptor(int concurrentSelectParticipants = 0) : DbCommandInterceptor
|
||||
{
|
||||
private readonly TaskCompletionSource _selectBarrier =
|
||||
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private int _recordSelectCount;
|
||||
private int _upsertCommandCount;
|
||||
private int _parameterlessUpsertCount;
|
||||
private int _selectBarrierArrivals;
|
||||
|
||||
/// <summary>访问同步记录表的 SELECT 命令数。</summary>
|
||||
public int RecordSelectCount => Volatile.Read(ref _recordSelectCount);
|
||||
|
||||
/// <summary>包含 ON CONFLICT 的同步记录 INSERT 命令数。</summary>
|
||||
public int UpsertCommandCount => Volatile.Read(ref _upsertCommandCount);
|
||||
|
||||
/// <summary>所有 UPSERT 是否都通过数据库参数传值。</summary>
|
||||
public bool AllUpsertsParameterized => Volatile.Read(ref _parameterlessUpsertCount) == 0;
|
||||
|
||||
public override InterceptionResult<DbDataReader> ReaderExecuting(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<DbDataReader> result)
|
||||
{
|
||||
Count(command);
|
||||
return result;
|
||||
}
|
||||
|
||||
public override async ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<DbDataReader> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
Count(command);
|
||||
if (concurrentSelectParticipants > 0 && IsRecordSelect(command))
|
||||
{
|
||||
var arrivals = Interlocked.Increment(ref _selectBarrierArrivals);
|
||||
if (arrivals == concurrentSelectParticipants)
|
||||
{
|
||||
_selectBarrier.TrySetResult();
|
||||
}
|
||||
|
||||
if (arrivals <= concurrentSelectParticipants)
|
||||
{
|
||||
await _selectBarrier.Task.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
public override InterceptionResult<int> NonQueryExecuting(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<int> result)
|
||||
{
|
||||
Count(command);
|
||||
return result;
|
||||
}
|
||||
|
||||
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<int> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
Count(command);
|
||||
return ValueTask.FromResult(result);
|
||||
}
|
||||
|
||||
public override InterceptionResult<object> ScalarExecuting(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<object> result)
|
||||
{
|
||||
Count(command);
|
||||
return result;
|
||||
}
|
||||
|
||||
public override ValueTask<InterceptionResult<object>> ScalarExecutingAsync(
|
||||
DbCommand command,
|
||||
CommandEventData eventData,
|
||||
InterceptionResult<object> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
Count(command);
|
||||
return ValueTask.FromResult(result);
|
||||
}
|
||||
|
||||
private void Count(DbCommand command)
|
||||
{
|
||||
if (IsRecordSelect(command))
|
||||
{
|
||||
Interlocked.Increment(ref _recordSelectCount);
|
||||
}
|
||||
|
||||
if (!IsUpsert(command))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Interlocked.Increment(ref _upsertCommandCount);
|
||||
if (command.Parameters.Count == 0)
|
||||
{
|
||||
Interlocked.Increment(ref _parameterlessUpsertCount);
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsRecordSelect(DbCommand command)
|
||||
=> command.CommandText.TrimStart().StartsWith("SELECT", StringComparison.OrdinalIgnoreCase)
|
||||
&& command.CommandText.Contains("b_synced_records", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsUpsert(DbCommand command)
|
||||
=> command.CommandText.TrimStart().StartsWith("INSERT", StringComparison.OrdinalIgnoreCase)
|
||||
&& command.CommandText.Contains("b_synced_records", StringComparison.OrdinalIgnoreCase)
|
||||
&& command.CommandText.Contains("ON CONFLICT", StringComparison.OrdinalIgnoreCase);
|
||||
}
|
||||
184
tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs
Normal file
184
tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs
Normal file
@ -0,0 +1,184 @@
|
||||
using FluentAssertions;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Monitoring.Application.Sync;
|
||||
using Monitoring.Domain.Entities.ExternalSystem;
|
||||
using Monitoring.Domain.Enums;
|
||||
using Monitoring.Domain.Services;
|
||||
using Monitoring.Domain.Values;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>Monitoring 同步记录批量 UPSERT、并发幂等和日志安全集成测试。</summary>
|
||||
[Collection(PostgresMonitoringCollection.Name)]
|
||||
public sealed class MonitoringUpsertTests(PostgresMonitoringFixture fixture)
|
||||
{
|
||||
[Fact]
|
||||
public async Task Persists1001RecordsInThreeParameterizedUpsertsWithoutRecordSelects()
|
||||
{
|
||||
var interceptor = new CountingUpsertInterceptor();
|
||||
await using var db = fixture.CreateDbContext(interceptor);
|
||||
var system = CreateSystem("batch-upsert");
|
||||
db.ExternalSystems.Add(system);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var records = Enumerable.Range(0, 1001)
|
||||
.Select(index => new SyncedRecordData(
|
||||
$"order-{index}",
|
||||
"order",
|
||||
$$"""{"value":{{index}}}""",
|
||||
DateTime.UtcNow.AddMinutes(index)))
|
||||
.ToArray();
|
||||
using var services = CreateServices(new StaticExternalConnector(CreateResult(records)));
|
||||
var processor = new SyncProcessor(db, services, new RecordingLogger<SyncProcessor>());
|
||||
|
||||
var job = await processor.ExecuteAsync(system, "test", CancellationToken.None);
|
||||
var upsertCount = interceptor.UpsertCommandCount;
|
||||
var selectCount = interceptor.RecordSelectCount;
|
||||
var allParameterized = interceptor.AllUpsertsParameterized;
|
||||
db.ChangeTracker.Clear();
|
||||
var persistedCount = await db.SyncedRecords.CountAsync(x => x.ExternalSystemId == system.Id);
|
||||
|
||||
job.Status.Should().Be(SyncJobStatusEnum.Succeeded);
|
||||
persistedCount.Should().Be(1001);
|
||||
upsertCount.Should().Be(3);
|
||||
selectCount.Should().Be(0);
|
||||
allParameterized.Should().BeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ConcurrentSynchronizationsUpsertOneCoherentLatestRecord()
|
||||
{
|
||||
var systemId = Guid.NewGuid();
|
||||
await using (var seedDb = fixture.CreateDbContext())
|
||||
{
|
||||
seedDb.ExternalSystems.Add(CreateSystem("concurrent", systemId));
|
||||
await seedDb.SaveChangesAsync();
|
||||
}
|
||||
|
||||
var occurredAtA = new DateTime(2026, 7, 23, 1, 0, 0, DateTimeKind.Utc);
|
||||
var occurredAtB = occurredAtA.AddMinutes(1);
|
||||
var interceptor = new CountingUpsertInterceptor(concurrentSelectParticipants: 2);
|
||||
await using var dbA = fixture.CreateDbContext(interceptor);
|
||||
await using var dbB = fixture.CreateDbContext(interceptor);
|
||||
var systemA = await dbA.ExternalSystems.SingleAsync(x => x.Id == systemId);
|
||||
var systemB = await dbB.ExternalSystems.SingleAsync(x => x.Id == systemId);
|
||||
using var servicesA = CreateServices(new StaticExternalConnector(CreateResult(
|
||||
[new SyncedRecordData("shared-order", "order", """{"source":"A"}""", occurredAtA)])));
|
||||
using var servicesB = CreateServices(new StaticExternalConnector(CreateResult(
|
||||
[new SyncedRecordData("shared-order", "order", """{"source":"B"}""", occurredAtB)])));
|
||||
var processorA = new SyncProcessor(dbA, servicesA, new RecordingLogger<SyncProcessor>());
|
||||
var processorB = new SyncProcessor(dbB, servicesB, new RecordingLogger<SyncProcessor>());
|
||||
|
||||
var jobs = await Task.WhenAll(
|
||||
processorA.ExecuteAsync(systemA, "concurrent-a", CancellationToken.None),
|
||||
processorB.ExecuteAsync(systemB, "concurrent-b", CancellationToken.None));
|
||||
|
||||
jobs.Should().OnlyContain(job => job.Status == SyncJobStatusEnum.Succeeded);
|
||||
await using var verifyDb = fixture.CreateDbContext();
|
||||
var records = await verifyDb.SyncedRecords
|
||||
.Where(x => x.ExternalSystemId == systemId)
|
||||
.ToListAsync();
|
||||
records.Should().ContainSingle();
|
||||
var record = records.Single();
|
||||
new[] { jobs[0].Id, jobs[1].Id }.Should().Contain(record.SyncJobId);
|
||||
if (record.SyncJobId == jobs[0].Id)
|
||||
{
|
||||
record.OccurredAt.Should().Be(occurredAtA);
|
||||
record.Data.Payload["source"]?.ToString().Should().Be("A");
|
||||
}
|
||||
else
|
||||
{
|
||||
record.OccurredAt.Should().Be(occurredAtB);
|
||||
record.Data.Payload["source"]?.ToString().Should().Be("B");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SequentialSynchronizationReplacesJobTimeAndPayload()
|
||||
{
|
||||
await using var db = fixture.CreateDbContext();
|
||||
var system = CreateSystem("sequential");
|
||||
db.ExternalSystems.Add(system);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var firstTime = new DateTime(2026, 7, 23, 2, 0, 0, DateTimeKind.Utc);
|
||||
var secondTime = firstTime.AddHours(1);
|
||||
using var firstServices = CreateServices(new StaticExternalConnector(CreateResult(
|
||||
[new SyncedRecordData("updated-order", "order", """{"version":"first"}""", firstTime)])));
|
||||
var firstJob = await new SyncProcessor(db, firstServices, new RecordingLogger<SyncProcessor>())
|
||||
.ExecuteAsync(system, "first", CancellationToken.None);
|
||||
using var secondServices = CreateServices(new StaticExternalConnector(CreateResult(
|
||||
[new SyncedRecordData("updated-order", "order", """{"version":"second"}""", secondTime)])));
|
||||
|
||||
var secondJob = await new SyncProcessor(db, secondServices, new RecordingLogger<SyncProcessor>())
|
||||
.ExecuteAsync(system, "second", CancellationToken.None);
|
||||
|
||||
db.ChangeTracker.Clear();
|
||||
var record = await db.SyncedRecords.SingleAsync(x =>
|
||||
x.ExternalSystemId == system.Id && x.ExternalId == "updated-order");
|
||||
firstJob.Status.Should().Be(SyncJobStatusEnum.Succeeded);
|
||||
secondJob.Status.Should().Be(SyncJobStatusEnum.Succeeded);
|
||||
record.SyncJobId.Should().Be(secondJob.Id);
|
||||
record.OccurredAt.Should().Be(secondTime);
|
||||
record.Data.Payload["version"]?.ToString().Should().Be("second");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task InvalidJsonPreservesRawPayloadAndLogsIdentifiersOnly()
|
||||
{
|
||||
const string rawPayload = "invalid-json-sensitive-payload";
|
||||
const string externalId = "invalid-order";
|
||||
const string recordType = "order";
|
||||
await using var db = fixture.CreateDbContext();
|
||||
var system = CreateSystem("invalid-json");
|
||||
db.ExternalSystems.Add(system);
|
||||
await db.SaveChangesAsync();
|
||||
using var services = CreateServices(new StaticExternalConnector(CreateResult(
|
||||
[new SyncedRecordData(externalId, recordType, rawPayload, DateTime.UtcNow)])));
|
||||
var logger = new RecordingLogger<SyncProcessor>();
|
||||
var processor = new SyncProcessor(db, services, logger);
|
||||
|
||||
var job = await processor.ExecuteAsync(system, "invalid-json", CancellationToken.None);
|
||||
|
||||
db.ChangeTracker.Clear();
|
||||
var record = await db.SyncedRecords.SingleAsync(x => x.ExternalSystemId == system.Id);
|
||||
var warnings = logger.Entries.Where(entry => entry.Level == LogLevel.Warning).ToArray();
|
||||
job.Status.Should().Be(SyncJobStatusEnum.Succeeded);
|
||||
record.Data.Payload["_raw"]?.ToString().Should().Be(rawPayload);
|
||||
warnings.Should().ContainSingle();
|
||||
warnings[0].Message.Should().Contain(system.Id.ToString());
|
||||
warnings[0].Message.Should().Contain(externalId);
|
||||
warnings[0].Message.Should().Contain(recordType);
|
||||
warnings[0].Message.Should().NotContain(rawPayload);
|
||||
}
|
||||
|
||||
private static ExternalSystemEntity CreateSystem(string name, Guid? id = null)
|
||||
{
|
||||
var system = new ExternalSystemEntity
|
||||
{
|
||||
Data = new SystemData
|
||||
{
|
||||
Name = name,
|
||||
SystemType = ExternalSystemTypeEnum.GenericHttp,
|
||||
BaseUrl = "https://example.test",
|
||||
},
|
||||
};
|
||||
if (id.HasValue)
|
||||
{
|
||||
system.Id = id.Value;
|
||||
}
|
||||
|
||||
return system;
|
||||
}
|
||||
|
||||
private static FetchResult CreateResult(IReadOnlyList<SyncedRecordData> records)
|
||||
=> new(records.Count, new Dictionary<string, object?>(), records);
|
||||
|
||||
private static ServiceProvider CreateServices(IExternalConnector connector)
|
||||
=> new ServiceCollection()
|
||||
.AddSingleton(connector)
|
||||
.AddSingleton<IExternalConnector>(connector)
|
||||
.BuildServiceProvider();
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
using Xunit;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// 串行执行 Monitoring PostgreSQL 集成测试，避免数据库创建和并发用例互相干扰。
|
||||
/// </summary>
|
||||
[CollectionDefinition(Name, DisableParallelization = true)]
|
||||
public sealed class PostgresMonitoringCollection : ICollectionFixture<PostgresMonitoringFixture>
|
||||
{
|
||||
/// <summary>集合名称。</summary>
|
||||
public const string Name = "monitoring-postgres";
|
||||
}
|
||||
@ -0,0 +1,118 @@
|
||||
using BuildFlow.Shared.Domain.Common;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.EntityFrameworkCore.Diagnostics;
|
||||
using Monitoring.Infrastructure.Persistence;
|
||||
using Npgsql;
|
||||
using Xunit;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// 为 Monitoring 集成测试创建并销毁独立 PostgreSQL 数据库。
|
||||
/// </summary>
|
||||
public sealed class PostgresMonitoringFixture : IAsyncLifetime
|
||||
{
|
||||
private const string DefaultConnection =
|
||||
"Host=127.0.0.1;Port=55432;Username=buildflow;Password=buildflow;Database=postgres";
|
||||
|
||||
private string _databaseName = default!;
|
||||
private string _adminConnectionString = default!;
|
||||
|
||||
/// <summary>当前测试数据库连接串。</summary>
|
||||
public string ConnectionString { get; private set; } = default!;
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
var configured = Environment.GetEnvironmentVariable("BUILDFLOW_MONITORING_TEST_CONNECTION")
|
||||
?? DefaultConnection;
|
||||
var adminBuilder = new NpgsqlConnectionStringBuilder(configured)
|
||||
{
|
||||
Database = "postgres",
|
||||
Pooling = false,
|
||||
};
|
||||
_adminConnectionString = adminBuilder.ConnectionString;
|
||||
_databaseName = $"monitoring_upsert_{Guid.NewGuid():N}";
|
||||
|
||||
await using (var connection = new NpgsqlConnection(_adminConnectionString))
|
||||
{
|
||||
await connection.OpenAsync();
|
||||
await using var command = connection.CreateCommand();
|
||||
command.CommandText = $"CREATE DATABASE \"{_databaseName}\"";
|
||||
await command.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
var databaseBuilder = new NpgsqlConnectionStringBuilder(configured)
|
||||
{
|
||||
Database = _databaseName,
|
||||
Pooling = true,
|
||||
};
|
||||
ConnectionString = databaseBuilder.ConnectionString;
|
||||
|
||||
await using var db = CreateDbContext();
|
||||
await db.Database.EnsureCreatedAsync();
|
||||
}
|
||||
|
||||
/// <summary>创建指向独立测试数据库的 MonitoringDbContext。</summary>
|
||||
public MonitoringDbContext CreateDbContext(params IInterceptor[] interceptors)
|
||||
{
|
||||
var optionsBuilder = new DbContextOptionsBuilder<MonitoringDbContext>()
|
||||
.UseNpgsql(ConnectionString)
|
||||
.UseSnakeCaseNamingConvention();
|
||||
if (interceptors.Length > 0)
|
||||
{
|
||||
optionsBuilder.AddInterceptors(interceptors);
|
||||
}
|
||||
|
||||
var db = new MonitoringDbContext(optionsBuilder.Options);
|
||||
db.SavingChanges += FillAuditFields;
|
||||
return db;
|
||||
}
|
||||
|
||||
public async Task DisposeAsync()
|
||||
{
|
||||
NpgsqlConnection.ClearAllPools();
|
||||
await using var connection = new NpgsqlConnection(_adminConnectionString);
|
||||
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 void FillAuditFields(object? sender, EventArgs _)
|
||||
{
|
||||
if (sender is not DbContext db)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
foreach (var entry in db.ChangeTracker.Entries())
|
||||
{
|
||||
if (entry.Entity is IAuditable auditable)
|
||||
{
|
||||
auditable.CreatedBy ??= "test";
|
||||
auditable.UpdatedBy ??= "test";
|
||||
if (auditable.CreatedAt == default)
|
||||
{
|
||||
auditable.CreatedAt = now;
|
||||
}
|
||||
|
||||
auditable.UpdatedAt = now;
|
||||
}
|
||||
|
||||
if (entry.Entity is IHasOperatorIP operatorEntity)
|
||||
{
|
||||
operatorEntity.OperatorIP ??= "127.0.0.1";
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
25
tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs
Normal file
25
tests/BuildFlow.Monitoring.Tests/Sync/RecordingLogger.cs
Normal file
@ -0,0 +1,25 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>记录结构化日志格式化后的级别和文本，供安全断言使用。</summary>
|
||||
public sealed class RecordingLogger<T> : ILogger<T>
|
||||
{
|
||||
private readonly ConcurrentQueue<(LogLevel Level, string Message)> _entries = new();
|
||||
|
||||
/// <summary>当前已记录的所有日志。</summary>
|
||||
public IReadOnlyCollection<(LogLevel Level, string Message)> Entries => _entries.ToArray();
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(
|
||||
LogLevel logLevel,
|
||||
EventId eventId,
|
||||
TState state,
|
||||
Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
=> _entries.Enqueue((logLevel, formatter(state, exception)));
|
||||
}
|
||||
@ -0,0 +1,14 @@
|
||||
using Monitoring.Domain.Entities.ExternalSystem;
|
||||
using Monitoring.Domain.Enums;
|
||||
using Monitoring.Domain.Services;
|
||||
|
||||
namespace BuildFlow.Monitoring.Tests.Sync;
|
||||
|
||||
/// <summary>返回固定拉取结果的 Monitoring 集成测试连接器。</summary>
|
||||
public sealed class StaticExternalConnector(FetchResult result) : IExternalConnector
|
||||
{
|
||||
public ExternalSystemTypeEnum SystemType => ExternalSystemTypeEnum.GenericHttp;
|
||||
|
||||
public Task<FetchResult> FetchAsync(ExternalSystemEntity system, CancellationToken ct)
|
||||
=> Task.FromResult(result);
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user