fix: 创单事务释放后再提交 9610、轨迹和揽收事件
All checks were successful
Build and Deploy / deploy (push) Successful in 59s
All checks were successful
Build and Deploy / deploy (push) Successful in 59s
避免在创单事务里写 Outbox 征用不到连接,导致这三类事件被丢掉。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
79fa20a9ca
commit
61b6f9a9b9
@ -9,16 +9,24 @@ namespace Oms.Application.Messaging;
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
internal static class OmsAfterCommit
|
internal static class OmsAfterCommit
|
||||||
{
|
{
|
||||||
public static async Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func<Task> action)
|
public static Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func<Task> action)
|
||||||
{
|
{
|
||||||
var current = unitOfWorkManager.Current;
|
var current = unitOfWorkManager.Current;
|
||||||
if (current != null && !current.IsCompleted)
|
if (current != null && !current.IsCompleted)
|
||||||
{
|
{
|
||||||
current.OnCompleted(action);
|
// OnCompleted 仍处在创单事务里,此时再写 Outbox 征用不到连接,事件会丢。
|
||||||
return;
|
// 等工作单元释放后再开独立事务。
|
||||||
|
EventHandler<UnitOfWorkEventArgs> handler = null!;
|
||||||
|
handler = (_, _) =>
|
||||||
|
{
|
||||||
|
current.Disposed -= handler;
|
||||||
|
_ = Task.Run(action);
|
||||||
|
};
|
||||||
|
current.Disposed += handler;
|
||||||
|
return Task.CompletedTask;
|
||||||
}
|
}
|
||||||
|
|
||||||
await action();
|
return action();
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@ -33,9 +33,9 @@ internal static class OmsIntegrationEventPublisher
|
|||||||
@event.Id);
|
@event.Id);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
catch (Exception ex) when (IsTxConnectionMismatch(ex))
|
catch (Exception ex) when (ShouldUseIndependentOutbox(ex))
|
||||||
{
|
{
|
||||||
// SqlSugar 已在另一条连接上开事务时,EF 无法征用该 DbTransaction。
|
// 环境事务已提交、连接对不上,或还没有可征用的数据库事务时,改走独立短事务。
|
||||||
logger.LogWarning(
|
logger.LogWarning(
|
||||||
ex,
|
ex,
|
||||||
"Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}",
|
"Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}",
|
||||||
@ -97,12 +97,18 @@ internal static class OmsIntegrationEventPublisher
|
|||||||
return dbProvider.GetDbContextAsync();
|
return dbProvider.GetDbContextAsync();
|
||||||
}
|
}
|
||||||
|
|
||||||
private static bool IsTxConnectionMismatch(Exception ex)
|
private static bool ShouldUseIndependentOutbox(Exception ex)
|
||||||
{
|
{
|
||||||
for (var e = ex; e != null; e = e.InnerException!)
|
for (var e = ex; e != null; e = e.InnerException!)
|
||||||
{
|
{
|
||||||
if (e.Message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase)
|
var message = e.Message;
|
||||||
|| e.Message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase))
|
if (message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase)
|
||||||
|
|| message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase)
|
||||||
|
|| message.Contains("可征用的数据库事务", StringComparison.Ordinal)
|
||||||
|
|| message.Contains("returned null", StringComparison.OrdinalIgnoreCase)
|
||||||
|
|| message.Contains("No ambient transaction", StringComparison.OrdinalIgnoreCase)
|
||||||
|
|| message.Contains("已经回滚", StringComparison.Ordinal)
|
||||||
|
|| message.Contains("已被调用", StringComparison.Ordinal))
|
||||||
{
|
{
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|||||||
@ -2866,49 +2866,32 @@ public partial class OrderService
|
|||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
var customsTask = OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp =>
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp =>
|
||||||
sp.GetRequiredService<OrderService>().PublishCustomsForOrderAsync(orderId, platformCode));
|
sp.GetRequiredService<OrderService>().PublishCustomsForOrderAsync(orderId, platformCode));
|
||||||
var forecastTask = includeForecast
|
|
||||||
? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp =>
|
var forecastQueued = true;
|
||||||
|
if (includeForecast)
|
||||||
|
{
|
||||||
|
forecastQueued = await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp =>
|
||||||
{
|
{
|
||||||
var service = sp.GetRequiredService<OrderService>();
|
var service = sp.GetRequiredService<OrderService>();
|
||||||
await service.EnqueueForecastBillingByIdAsync(orderId, couponInstanceId, operatorUserId);
|
await service.EnqueueForecastBillingByIdAsync(orderId, couponInstanceId, operatorUserId);
|
||||||
var latest = await sp.GetRequiredService<IOrderRepository>()
|
});
|
||||||
.FirstOrDefaultAsync(x => x.Id == orderId);
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp =>
|
||||||
if (latest != null)
|
sp.GetRequiredService<OrderService>().PublishForecastTrackAsync(orderId));
|
||||||
{
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
|
||||||
output.OrderStatus = latest.OrderStatus;
|
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId));
|
||||||
output.PredeclareFailReason = latest.PredeclareFailReason;
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
|
||||||
output.ForecastBillingQueued = string.Equals(
|
sp.GetRequiredService<OrderService>().PublishForecastAutoTriggerAsync(orderId, operatorUserId));
|
||||||
latest.PredeclareFailReason,
|
}
|
||||||
ForecastBillingProcessingReason,
|
|
||||||
StringComparison.Ordinal);
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp =>
|
||||||
}
|
|
||||||
})
|
|
||||||
: Task.FromResult(true);
|
|
||||||
var pickupTask = OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp =>
|
|
||||||
sp.GetRequiredService<OrderService>().TryAttachPickupAfterParcelCreateAsync(input, output));
|
sp.GetRequiredService<OrderService>().TryAttachPickupAfterParcelCreateAsync(input, output));
|
||||||
var trackTask = includeForecast
|
|
||||||
? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp =>
|
|
||||||
sp.GetRequiredService<OrderService>().PublishForecastTrackAsync(orderId))
|
|
||||||
: Task.FromResult(true);
|
|
||||||
var labelTask = includeForecast
|
|
||||||
? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
|
|
||||||
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId))
|
|
||||||
: Task.FromResult(true);
|
|
||||||
var triggerTask = includeForecast
|
|
||||||
? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
|
|
||||||
sp.GetRequiredService<OrderService>().PublishForecastAutoTriggerAsync(orderId, operatorUserId))
|
|
||||||
: Task.FromResult(true);
|
|
||||||
|
|
||||||
await Task.WhenAll(customsTask, forecastTask, pickupTask, trackTask, labelTask, triggerTask);
|
if (includeForecast && !forecastQueued)
|
||||||
|
|
||||||
if (includeForecast && !await forecastTask)
|
|
||||||
{
|
{
|
||||||
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp =>
|
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp =>
|
||||||
sp.GetRequiredService<OrderService>().MarkForecastEnqueueFailedIfStillProcessingAsync(orderId));
|
sp.GetRequiredService<OrderService>().MarkForecastEnqueueFailedIfStillProcessingAsync(orderId));
|
||||||
output.ForecastBillingQueued = false;
|
|
||||||
output.PredeclareFailReason = ForecastBillingEnqueueFailReason;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user