简化

src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs

@@ -8,7 +8,8 @@ using Oms.Application.Services.Orders.Batch;
 namespace Oms.Application.Messaging;
 
 /// <summary>
-/// 订单批量集成事件处理器。消费通道/Ack/Nack/DLX/重试由 Fake.EventBus.RabbitMQ 负责。
+/// 订单批量集成事件处理器。
+/// 正常返回 Ack；抛异常由 Fake.EventBus.RabbitMQ 重试后 Nack 进 DLX。
 /// </summary>
 public sealed class OrderBatchJobEventHandler(
     OrderBatchJobProcessor processor,
@@ -29,15 +30,7 @@ public sealed class OrderBatchJobEventHandler(
         }
 
         using var unitOfWork = unitOfWorkManager.Begin();
-        var message = @event.ToMessage();
-        var result = await processor.ProcessAsync(message, cancellationToken);
+        await processor.ProcessAsync(@event.ToMessage(), cancellationToken);
         await unitOfWork.CompleteAsync(cancellationToken);
-
-        // Fake EventBus：成功返回则 Ack；抛异常则框架内重试后 Nack(requeue=false) 进 DLX。
-        if (result.Kind == OrderBatchJobProcessor.ProcessResultKind.Reject && result.Requeue)
-        {
-            throw new InvalidOperationException(
-                $"Order batch system failure, allow Fake EventBus retry. action={message.Action} jobItemId={message.JobItemId} messageId={message.MessageId}");
-        }
     }
 }

src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs

@@ -1,7 +1,7 @@
 namespace Oms.Application.Messaging;
 
 /// <summary>
-/// 订单批量消息业务开关。连接/消费/QoS/DLX 由 Fake.EventBus.RabbitMQ（RabbitMqEventBus）配置。
+/// 订单批量消息业务开关。连接/消费/QoS/DLX/重试由 Fake.EventBus.RabbitMQ 配置。
 /// </summary>
 public class OrderBatchRabbitMqOptions
 {
@@ -11,9 +11,4 @@ public class OrderBatchRabbitMqOptions
     /// 是否启用订单批量消息发布与处理。
     /// </summary>
     public bool Enabled { get; set; }
-
-    /// <summary>
-    /// 系统失败时任务项最大重试次数（业务侧），与 Fake EventBus 进程内重试相互独立。
-    /// </summary>
-    public int MaxRetryCount { get; set; } = 3;
 }

src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs

@@ -1,7 +1,5 @@
-using Fake.Application;
 using Fake.DependencyInjection;
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Options;
 using Newtonsoft.Json;
 using Oms.Application.Dtos.Orders;
 using Oms.Application.Messaging;
@@ -13,15 +11,15 @@ namespace Oms.Application.Services.Orders.Batch;
 
 /// <summary>
 /// 消费 RabbitMQ 消息后更新任务状态并执行业务。
+/// 正常返回由 Fake EventBus Ack；抛异常则框架重试后进 DLX。
 /// </summary>
 public sealed class OrderBatchJobProcessor(
     IOrderBatchJobRepository batchJobRepository,
     IOrderTaskRepository orderTaskRepository,
     IOrderBatchActionExecutor actionExecutor,
-    IOptions<OrderBatchRabbitMqOptions> rabbitMqOptions,
     ILogger<OrderBatchJobProcessor> logger) : IScopedDependency
 {
-    public async Task<ProcessResult> ProcessAsync(
+    public async Task ProcessAsync(
         OrderBatchJobMessage message,
         CancellationToken cancellationToken = default)
     {
@@ -32,18 +30,18 @@ public sealed class OrderBatchJobProcessor(
             if (item == null)
             {
                 logger.LogWarning("Job item {JobItemId} not found, ack message {MessageId}", message.JobItemId, message.MessageId);
-                return ProcessResult.Ack;
+                return;
             }
 
             if (item.Status == OrderBatchJobItemStatusCodes.Success)
             {
-                return ProcessResult.Ack;
+                return;
             }
 
             if (item.Status == OrderBatchJobItemStatusCodes.Cancelled)
             {
                 logger.LogInformation("Skip cancelled job item {JobItemId}", item.Id);
-                return ProcessResult.Ack;
+                return;
             }
 
             item.Status = OrderBatchJobItemStatusCodes.Running;
@@ -56,7 +54,7 @@ public sealed class OrderBatchJobProcessor(
 
         if (message.JobItemId <= 0)
         {
-            // fire-and-forget（如轨迹同步）：失败也不 requeue，避免空转打爆 DB
+            // fire-and-forget（如轨迹同步）：失败只记日志，不抛、不进 DLX
             if (!result.Success)
             {
                 logger.LogWarning(
@@ -66,13 +64,13 @@ public sealed class OrderBatchJobProcessor(
                     result.FailReason);
             }
 
-            return ProcessResult.Ack;
+            return;
         }
 
         item = await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken);
         if (item == null)
         {
-            return ProcessResult.Ack;
+            return;
         }
 
         item.FinishedAt = DateTime.Now;
@@ -98,7 +96,7 @@ public sealed class OrderBatchJobProcessor(
 
             await batchJobRepository.UpdateItemAsync(item, cancellationToken);
             await RefreshJobSummaryAsync(item.JobId, cancellationToken);
-            return ProcessResult.Ack;
+            return;
         }
 
         if (result.IsBusinessFailure)
@@ -112,39 +110,7 @@ public sealed class OrderBatchJobProcessor(
             }
 
             await RefreshJobSummaryAsync(item.JobId, cancellationToken);
-            return ProcessResult.Ack;
         }
-
-        item.RetryCount++;
-        var maxRetryCount = Math.Max(1, rabbitMqOptions.Value.MaxRetryCount);
-        if (item.RetryCount >= maxRetryCount)
-        {
-            item.Status = OrderBatchJobItemStatusCodes.Failed;
-            item.FailReason = string.IsNullOrWhiteSpace(result.FailReason)
-                ? "系统重试次数已耗尽"
-                : result.FailReason;
-            item.FinishedAt = DateTime.Now;
-            item.UpdateTime = DateTime.Now;
-            await batchJobRepository.UpdateItemAsync(item, cancellationToken);
-            if (message.Action == OrderBatchActionCodes.ImportRow)
-            {
-                await RecordImportTaskFailAsync(message, item.FailReason, cancellationToken);
-            }
-
-            await RefreshJobSummaryAsync(item.JobId, cancellationToken);
-            logger.LogWarning(
-                "Job item {JobItemId} retry exhausted after {RetryCount} attempts",
-                item.Id,
-                item.RetryCount);
-            return ProcessResult.Ack;
-        }
-
-        item.Status = OrderBatchJobItemStatusCodes.Pending;
-        item.FailReason = null;
-        item.FinishedAt = null;
-        item.UpdateTime = DateTime.Now;
-        await batchJobRepository.UpdateItemAsync(item, cancellationToken);
-        return ProcessResult.Reject(isBusinessFailure: false);
     }
 
     private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken)
@@ -251,22 +217,4 @@ public sealed class OrderBatchJobProcessor(
             }
         ], cancellationToken);
     }
-
-    public enum ProcessResultKind
-    {
-        Ack,
-        Reject,
-    }
-
-    public readonly struct ProcessResult
-    {
-        public ProcessResultKind Kind { get; init; }
-
-        public bool Requeue { get; init; }
-
-        public static ProcessResult Ack => new() { Kind = ProcessResultKind.Ack };
-
-        public static ProcessResult Reject(bool isBusinessFailure) =>
-            new() { Kind = ProcessResultKind.Reject, Requeue = !isBusinessFailure };
-    }
 }

src/Oms.Tests/OrderBatchJobProcessorTests.cs

@@ -1,6 +1,5 @@
 using FluentAssertions;
 using Microsoft.Extensions.Logging.Abstractions;
-using Microsoft.Extensions.Options;
 using Moq;
 using Oms.Application.Messaging;
 using Oms.Application.Services.Orders.Batch;
@@ -23,12 +22,11 @@ public class OrderBatchJobProcessorTests
             _batchJobRepository.Object,
             _orderTaskRepository.Object,
             _actionExecutor.Object,
-            Options.Create(new OrderBatchRabbitMqOptions { MaxRetryCount = 3 }),
             NullLogger<OrderBatchJobProcessor>.Instance);
     }
 
     [Fact]
-    public async Task ProcessAsync_BusinessFailure_ShouldMarkFailedAndAck()
+    public async Task ProcessAsync_BusinessFailure_ShouldMarkFailed()
     {
         var item = CreateItem(id: 10, status: OrderBatchJobItemStatusCodes.Pending);
         var message = CreateMessage(item.Id);
@@ -38,54 +36,30 @@ public class OrderBatchJobProcessorTests
             .ReturnsAsync(OrderBatchActionExecuteResult.BusinessFail("订单不存在"));
         SetupJobRefresh(item.JobId, [item]);
 
-        var result = await _processor.ProcessAsync(message);
+        await _processor.ProcessAsync(message);
 
-        result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack);
         item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
         item.FailReason.Should().Be("订单不存在");
     }
 
     [Fact]
-    public async Task ProcessAsync_SystemFailure_WithRetriesRemaining_ShouldResetPendingAndReject()
+    public async Task ProcessAsync_SystemFailure_ShouldThrow()
     {
-        var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending, retryCount: 1);
+        var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending);
         var message = CreateMessage(item.Id);
         SetupItemLookup(item);
         _actionExecutor
             .Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
             .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
 
-        var result = await _processor.ProcessAsync(message);
+        var act = () => _processor.ProcessAsync(message);
 
-        result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Reject);
-        result.Requeue.Should().BeTrue();
-        item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
-        item.RetryCount.Should().Be(2);
-        item.FailReason.Should().BeNull();
-        item.FinishedAt.Should().BeNull();
+        await act.Should().ThrowAsync<InvalidOperationException>()
+            .WithMessage("*下游超时*");
     }
 
     [Fact]
-    public async Task ProcessAsync_SystemFailure_WhenRetryExhausted_ShouldMarkFailedAndAck()
-    {
-        var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending, retryCount: 2);
-        var message = CreateMessage(item.Id);
-        SetupItemLookup(item);
-        _actionExecutor
-            .Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
-            .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
-        SetupJobRefresh(item.JobId, [item]);
-
-        var result = await _processor.ProcessAsync(message);
-
-        result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack);
-        item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
-        item.RetryCount.Should().Be(3);
-        item.FailReason.Should().Be("下游超时");
-    }
-
-    [Fact]
-    public async Task ProcessAsync_CancelledItem_ShouldAckWithoutExecuting()
+    public async Task ProcessAsync_CancelledItem_ShouldSkipWithoutExecuting()
     {
         var item = CreateItem(id: 13, status: OrderBatchJobItemStatusCodes.Cancelled);
         var message = CreateMessage(item.Id);
@@ -93,16 +67,15 @@ public class OrderBatchJobProcessorTests
             .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny<CancellationToken>()))
             .ReturnsAsync(item);
 
-        var result = await _processor.ProcessAsync(message);
+        await _processor.ProcessAsync(message);
 
-        result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack);
         _actionExecutor.Verify(
             x => x.ExecuteAsync(It.IsAny<OrderBatchJobMessage>(), It.IsAny<CancellationToken>()),
             Times.Never);
     }
 
     [Fact]
-    public async Task ProcessAsync_FireAndForgetFailure_ShouldAckWithoutRequeue()
+    public async Task ProcessAsync_FireAndForgetFailure_ShouldNotThrow()
     {
         var message = new OrderBatchJobMessage
         {
@@ -116,10 +89,8 @@ public class OrderBatchJobProcessorTests
             .Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
             .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
 
-        var result = await _processor.ProcessAsync(message);
+        await _processor.ProcessAsync(message);
 
-        result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack);
-        result.Requeue.Should().BeFalse();
         _batchJobRepository.Verify(
             x => x.GetItemByIdAsync(It.IsAny<int>(), It.IsAny<CancellationToken>()),
             Times.Never);

src/SimpleAdmin.Api/appsettings.Development.json

@@ -31,10 +31,6 @@
     "RefreshExpiryInDays": "7",
     "ClockSkewMinutes": "5"
   },
-  "FeiShu": {
-    "Webhook": "https://example.com/webhook",
-    "EnableFeiShuExceptionSubscribe": false
-  },
   "FeiShuNotice": {
     "Webhook": "https://example.com/webhook",
     "EnableFeiShuExceptionSubscribe": false
@@ -112,8 +108,7 @@
   },
   "OrderBatch": {
     "RabbitMq": {
-      "Enabled": false,
-      "MaxRetryCount": 3
+      "Enabled": false
     }
   }
 }