diff --git a/src/Oms.Application/Messaging/OmsAfterCommit.cs b/src/Oms.Application/Messaging/OmsAfterCommit.cs index 66f07fb..9b31ee7 100644 --- a/src/Oms.Application/Messaging/OmsAfterCommit.cs +++ b/src/Oms.Application/Messaging/OmsAfterCommit.cs @@ -9,16 +9,24 @@ namespace Oms.Application.Messaging; /// internal static class OmsAfterCommit { - public static async Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func action) + public static Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func action) { var current = unitOfWorkManager.Current; if (current != null && !current.IsCompleted) { - current.OnCompleted(action); - return; + // OnCompleted 仍处在创单事务里,此时再写 Outbox 征用不到连接,事件会丢。 + // 等工作单元释放后再开独立事务。 + EventHandler handler = null!; + handler = (_, _) => + { + current.Disposed -= handler; + _ = Task.Run(action); + }; + current.Disposed += handler; + return Task.CompletedTask; } - await action(); + return action(); } /// diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs index 956bc1e..b572be4 100644 --- a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs +++ b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs @@ -33,9 +33,9 @@ internal static class OmsIntegrationEventPublisher @event.Id); return; } - catch (Exception ex) when (IsTxConnectionMismatch(ex)) + catch (Exception ex) when (ShouldUseIndependentOutbox(ex)) { - // SqlSugar 已在另一条连接上开事务时,EF 无法征用该 DbTransaction。 + // 环境事务已提交、连接对不上,或还没有可征用的数据库事务时,改走独立短事务。 logger.LogWarning( ex, "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(); } - private static bool IsTxConnectionMismatch(Exception ex) + private static bool ShouldUseIndependentOutbox(Exception ex) { for (var e = ex; e != null; e = e.InnerException!) { - if (e.Message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase) - || e.Message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase)) + var message = e.Message; + 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; } diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index afe1617..2d66718 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -2866,49 +2866,32 @@ public partial class OrderService { try { - var customsTask = OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp => + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp => sp.GetRequiredService().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(); await service.EnqueueForecastBillingByIdAsync(orderId, couponInstanceId, operatorUserId); - var latest = await sp.GetRequiredService() - .FirstOrDefaultAsync(x => x.Id == orderId); - if (latest != null) - { - output.OrderStatus = latest.OrderStatus; - output.PredeclareFailReason = latest.PredeclareFailReason; - output.ForecastBillingQueued = string.Equals( - latest.PredeclareFailReason, - ForecastBillingProcessingReason, - StringComparison.Ordinal); - } - }) - : Task.FromResult(true); - var pickupTask = OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp => + }); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp => + sp.GetRequiredService().PublishForecastTrackAsync(orderId)); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => + sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => + sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)); + } + + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp => sp.GetRequiredService().TryAttachPickupAfterParcelCreateAsync(input, output)); - var trackTask = includeForecast - ? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp => - sp.GetRequiredService().PublishForecastTrackAsync(orderId)) - : Task.FromResult(true); - var labelTask = includeForecast - ? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => - sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)) - : Task.FromResult(true); - var triggerTask = includeForecast - ? OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => - sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)) - : Task.FromResult(true); - await Task.WhenAll(customsTask, forecastTask, pickupTask, trackTask, labelTask, triggerTask); - - if (includeForecast && !await forecastTask) + if (includeForecast && !forecastQueued) { await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp => sp.GetRequiredService().MarkForecastEnqueueFailedIfStillProcessingAsync(orderId)); - output.ForecastBillingQueued = false; - output.PredeclareFailReason = ForecastBillingEnqueueFailReason; } } catch (Exception ex)