From 61b6f9a9b9504222e0a8d94b9a1d85a6281d7c32 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Wed, 14 Oct 2026 19:37:56 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=88=9B=E5=8D=95=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E9=87=8A=E6=94=BE=E5=90=8E=E5=86=8D=E6=8F=90=E4=BA=A4=209610?= =?UTF-8?q?=E3=80=81=E8=BD=A8=E8=BF=B9=E5=92=8C=E6=8F=BD=E6=94=B6=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 避免在创单事务里写 Outbox 征用不到连接，导致这三类事件被丢掉。 Co-authored-by: Cursor --- .../Messaging/OmsAfterCommit.cs | 16 ++++-- .../Messaging/OmsIntegrationEventPublisher.cs | 16 ++++-- .../Orders/OrderService.ParcelBilling.cs | 51 +++++++------------ 3 files changed, 40 insertions(+), 43 deletions(-) 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)