diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs
index 0f12f49..956bc1e 100644
--- a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs
+++ b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs
@@ -1,3 +1,4 @@
+using System.Data;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
@@ -9,6 +10,7 @@ namespace Oms.Application.Messaging;
///
/// 可靠发布 IntegrationEvent:事务 UoW 内写入 Outbox(由 Fake.EventBus.IntegrationEventLog 投递)。
+/// SqlSugar 与 EF 默认各持连接;若环境事务已由 SqlSugar 占用,则改用 RequiresNew 短事务只写 Outbox。
///
internal static class OmsIntegrationEventPublisher
{
@@ -21,24 +23,41 @@ internal static class OmsIntegrationEventPublisher
{
if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager))
{
- await eventBus.PublishAsync(@event, cancellationToken);
- logger.LogDebug(
- "Outbox enqueued {EventName} Id={EventId} (ambient TX)",
+ try
+ {
+ await EnsureOutboxDbContextAsync(unitOfWorkManager);
+ await eventBus.PublishAsync(@event, cancellationToken);
+ logger.LogDebug(
+ "Outbox enqueued {EventName} Id={EventId} (ambient TX)",
+ @event.GetType().Name,
+ @event.Id);
+ return;
+ }
+ catch (Exception ex) when (IsTxConnectionMismatch(ex))
+ {
+ // SqlSugar 已在另一条连接上开事务时,EF 无法征用该 DbTransaction。
+ logger.LogWarning(
+ ex,
+ "Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}",
+ @event.GetType().Name,
+ @event.Id);
+ }
+ }
+ else
+ {
+ logger.LogWarning(
+ "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX",
@event.GetType().Name,
@event.Id);
- return;
}
- // 无环境事务时开启短事务只写 Outbox,并告警:业务落库与事件非原子。
- logger.LogWarning(
- "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX",
- @event.GetType().Name,
- @event.Id);
-
- using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
- // 先打开 Outbox DbContext 以产生可征用的 DbTransaction
- var dbProvider = uow.ServiceProvider.GetRequiredService>();
- await dbProvider.GetDbContextAsync();
+ using var uow = unitOfWorkManager.Begin(new UnitOfWorkAttribute
+ {
+ IsTransactional = true,
+ RequiresNew = true,
+ IsolationLevel = IsolationLevel.ReadCommitted
+ });
+ await EnsureOutboxDbContextAsync(unitOfWorkManager);
await eventBus.PublishAsync(@event, cancellationToken);
await uow.CompleteAsync(cancellationToken);
}
@@ -65,4 +84,30 @@ internal static class OmsIntegrationEventPublisher
return false;
}
}
+
+ ///
+ /// 业务写库前先打开 Outbox DbContext,尽量让 EF 先占住环境事务连接,供 SqlSugar 征用。
+ ///
+ public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager)
+ {
+ var current = unitOfWorkManager.Current
+ ?? throw new InvalidOperationException("No ambient unit of work");
+ var dbProvider = current.ServiceProvider
+ .GetRequiredService>();
+ return dbProvider.GetDbContextAsync();
+ }
+
+ private static bool IsTxConnectionMismatch(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))
+ {
+ return true;
+ }
+ }
+
+ return false;
+ }
}
diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
index b73fa91..3cac21b 100644
--- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
+++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
@@ -1,5 +1,6 @@
using Fake.Application;
using Fake.DependencyInjection;
+using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
@@ -30,6 +31,7 @@ public sealed class OrderBatchJobCoordinator(
IObjectStorage fileStorage,
IOrderBatchJobPublisher publisher,
OrderBatchJobProcessor batchJobProcessor,
+ IUnitOfWorkManager unitOfWorkManager,
IOptions batchOptions,
ILogger logger) : IScopedDependency
{
@@ -76,6 +78,12 @@ public sealed class OrderBatchJobCoordinator(
}
}
+ // 先打开 EF Outbox 上下文,尽量让环境事务由 EF 占连接,避免 SqlSugar 先开事务后 Outbox 征用失败。
+ if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager))
+ {
+ await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager);
+ }
+
var now = DateTime.Now;
var job = new OrderBatchJob
{
diff --git a/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs b/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs
index 13dafbf..a467162 100644
--- a/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs
+++ b/src/Oms.Application/Services/Orders/OrderService.BatchJob.cs
@@ -1,3 +1,4 @@
+using System.Data;
using Fake.Application;
using Fake.Security.Claims;
using Fake.UnitOfWork;
@@ -22,7 +23,7 @@ public partial class OrderService
/// 创建订单批量异步任务并入队 RabbitMQ。
///
[HttpPost("/order/batch/jobs")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task CreateBatchJobAsync([FromBody] OrderBatchJobCreateInput input)
{
return BatchJobCoordinator.CreateAndEnqueueAsync(input, CurrentUser.GetId());
@@ -75,7 +76,7 @@ public partial class OrderService
/// 异步批量创建轨迹订单(替代前端逐条调用)。
///
[HttpPost("/order/batch/track-upsert")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchTrackUpsertAsync([FromBody] OrderBatchJobCreateInput input)
{
input.Action = OrderBatchActionCodes.TrackUpsertSmallBag;
@@ -97,7 +98,7 @@ public partial class OrderService
/// 异步批量预报下游订单。
///
[HttpPost("/order/batch/downstream-forward")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchDownstreamForwardAsync([FromBody] OrderBatchJobCreateInput input)
{
input.Action = OrderBatchActionCodes.DownstreamForwardCreateOne;
@@ -115,7 +116,7 @@ public partial class OrderService
string.IsNullOrWhiteSpace(authorization) ? null : authorization);
}
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
private Task EnqueueOrderBatchAsync(string action, OrderBatchJobCreateInput input)
{
input.Action = action;
@@ -126,7 +127,7 @@ public partial class OrderService
/// 异步批量支付订单。
///
[HttpPost("/order/batch/pay")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchPayAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Pay, input);
@@ -134,7 +135,7 @@ public partial class OrderService
/// 异步批量预报订单。
///
[HttpPost("/order/batch/pre-declare")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchPreDeclareAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.PreDeclare, input);
@@ -142,7 +143,7 @@ public partial class OrderService
/// 异步批量拦截订单。
///
[HttpPost("/order/batch/intercept")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchInterceptAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Intercept, input);
@@ -150,7 +151,7 @@ public partial class OrderService
/// 异步批量取消拦截。
///
[HttpPost("/order/batch/cancel-intercept")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchCancelInterceptAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.CancelIntercept, input);
@@ -158,7 +159,7 @@ public partial class OrderService
/// 异步批量删除订单。
///
[HttpPost("/order/batch/delete")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchDeleteAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Delete, input);
@@ -166,7 +167,7 @@ public partial class OrderService
/// 异步批量打印面单。
///
[HttpPost("/order/batch/print")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchPrintAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Print, input);
@@ -174,7 +175,7 @@ public partial class OrderService
/// 异步批量标记问题件。
///
[HttpPost("/order/batch/mark-problematic")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchMarkProblematicAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkProblematic, input);
@@ -182,7 +183,7 @@ public partial class OrderService
/// 异步批量重新排货。
///
[HttpPost("/order/batch/reallocate")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchReallocateAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.Reallocate, input);
@@ -190,7 +191,7 @@ public partial class OrderService
/// 异步批量标记已认领。
///
[HttpPost("/order/batch/mark-claimed")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchMarkClaimedAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.MarkClaimed, input);
@@ -198,7 +199,7 @@ public partial class OrderService
/// 异步批量更新处理状态。
///
[HttpPost("/order/batch/handle-status-update")]
- [UnitOfWork(IsTransactional = true)]
+ [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public Task BatchHandleStatusUpdateAsync([FromBody] OrderBatchJobCreateInput input) =>
EnqueueOrderBatchAsync(OrderBatchActionCodes.HandleStatusUpdate, input);
diff --git a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs
index 37ecb2f..c48e471 100644
--- a/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs
+++ b/src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs
@@ -1,6 +1,7 @@
using FluentAssertions;
using Fake.Application;
using Fake.ObjectStorage;
+using Fake.UnitOfWork;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Moq;
@@ -36,6 +37,8 @@ public class OrderBatchJobCoordinatorGovernanceTests
actionExecutor.Object,
batchOptions,
NullLogger.Instance);
+ var unitOfWorkManager = new Mock(MockBehavior.Loose);
+ unitOfWorkManager.SetupGet(x => x.Current).Returns((IUnitOfWork?)null);
_coordinator = new OrderBatchJobCoordinator(
_batchJobRepository.Object,
_orderRepository.Object,
@@ -44,6 +47,7 @@ public class OrderBatchJobCoordinatorGovernanceTests
_fileStorage.Object,
_publisher.Object,
processor,
+ unitOfWorkManager.Object,
batchOptions,
NullLogger.Instance);
}