|
|
@@ -5,11 +5,23 @@ namespace Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
|
|
|
|
|
|
/// <summary>
|
|
|
/// 出站分发：扫描 mdp_outbox，按 source_type 路由到 API / DB 执行器。
|
|
|
+/// WP10 S4b / D12：退避 1m/5m/15m/30m/60m/120m，MaxRetry=6；配置类失败不烧重试。
|
|
|
/// </summary>
|
|
|
public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
{
|
|
|
public const int DefaultTake = 500;
|
|
|
- private const int MaxRetry = 3;
|
|
|
+ private const int MaxRetry = 6;
|
|
|
+
|
|
|
+ /// <summary>D12：按 retry_count 下标取退避分钟数（clamp 到末项）。</summary>
|
|
|
+ private static readonly int[] BackoffMinutes = [1, 5, 15, 30, 60, 120];
|
|
|
+
|
|
|
+ private static readonly string[] PermanentErrorPrefixes =
|
|
|
+ [
|
|
|
+ "PAYLOAD_INVALID",
|
|
|
+ "CONSTRAINT_VIOLATION",
|
|
|
+ "SQL_CONSTRAINT",
|
|
|
+ "UNSUPPORTED_OP",
|
|
|
+ ];
|
|
|
|
|
|
private readonly ISqlSugarClient _db;
|
|
|
private readonly MdpApiPushExecutor _api;
|
|
|
@@ -31,8 +43,11 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
public async Task<(int success, int failed, int skipped)> PushPendingAsync(
|
|
|
int take = DefaultTake, CancellationToken cancellationToken = default)
|
|
|
{
|
|
|
+ var now = DateTime.Now;
|
|
|
var pending = await _db.Queryable<MdpOutbox>()
|
|
|
- .Where(x => x.Status == 0 && x.RetryCount < MaxRetry)
|
|
|
+ .Where(x => x.Status == 0
|
|
|
+ && x.RetryCount < MaxRetry
|
|
|
+ && (x.NextRetryTime == null || x.NextRetryTime <= now))
|
|
|
.OrderBy(x => x.Id)
|
|
|
.Take(take > 0 ? take : DefaultTake)
|
|
|
.ToListAsync(cancellationToken);
|
|
|
@@ -57,6 +72,7 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
// 配置类失败：保持 pending，不烧 retry_count，便于配置修复后被 60s 兜底作业捡起
|
|
|
await MarkConfigIssueAsync(
|
|
|
item,
|
|
|
+ "SOURCE_DISABLED",
|
|
|
$"目标源 {item.TargetSourceCode} 未启用",
|
|
|
cancellationToken);
|
|
|
skipped++;
|
|
|
@@ -65,7 +81,7 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
|
|
|
if (!TryResolve(source, out var executor, out var routeError))
|
|
|
{
|
|
|
- await MarkConfigIssueAsync(item, routeError!, cancellationToken);
|
|
|
+ await MarkConfigIssueAsync(item, "SOURCE_ROUTE_UNRESOLVED", routeError!, cancellationToken);
|
|
|
skipped++;
|
|
|
continue;
|
|
|
}
|
|
|
@@ -76,23 +92,19 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
var result = await executor.PushAsync(source, item, cancellationToken);
|
|
|
if (result.Success)
|
|
|
{
|
|
|
- await MarkAsync(item, 1, result.ResponseJson, null, cancellationToken);
|
|
|
+ await MarkAsync(item, 1, result.ResponseJson, null, null, null, cancellationToken);
|
|
|
success++;
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
- item.RetryCount++;
|
|
|
- var status = item.RetryCount >= MaxRetry ? 2 : 0;
|
|
|
- await MarkAsync(item, status, result.ResponseJson, Truncate(result.ErrorMessage, 900), cancellationToken);
|
|
|
- if (status == 2) failed++; else skipped++;
|
|
|
+ await ApplyFailureAsync(item, result.ResponseJson, result.ErrorMessage, cancellationToken);
|
|
|
+ if (item.Status == 2) failed++; else skipped++;
|
|
|
}
|
|
|
}
|
|
|
catch (Exception ex)
|
|
|
{
|
|
|
- item.RetryCount++;
|
|
|
- var status = item.RetryCount >= MaxRetry ? 2 : 0;
|
|
|
- await MarkAsync(item, status, null, Truncate(ex.Message, 900), cancellationToken);
|
|
|
- if (status == 2) failed++; else skipped++;
|
|
|
+ await ApplyFailureAsync(item, null, ex.Message, cancellationToken);
|
|
|
+ if (item.Status == 2) failed++; else skipped++;
|
|
|
_logger.LogWarning(ex, "[MdpTargetPushDispatcher] outbox id={Id} failed", item.Id);
|
|
|
}
|
|
|
}
|
|
|
@@ -100,6 +112,96 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
return (success, failed, skipped);
|
|
|
}
|
|
|
|
|
|
+ private async Task ApplyFailureAsync(
|
|
|
+ MdpOutbox item, string? responseJson, string? errorMessage, CancellationToken ct)
|
|
|
+ {
|
|
|
+ var msg = Truncate(errorMessage, 900);
|
|
|
+ var (errorCode, permanent) = ClassifyFailure(msg);
|
|
|
+
|
|
|
+ if (permanent)
|
|
|
+ {
|
|
|
+ // 立即死信，不递增 retry_count
|
|
|
+ item.NextRetryTime = null;
|
|
|
+ item.LastErrorCode = errorCode;
|
|
|
+ await MarkAsync(item, 2, responseJson, msg, null, errorCode, ct);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ item.RetryCount++;
|
|
|
+ item.LastErrorCode = errorCode;
|
|
|
+ if (item.RetryCount >= MaxRetry)
|
|
|
+ {
|
|
|
+ item.NextRetryTime = null;
|
|
|
+ await MarkAsync(item, 2, responseJson, msg, null, errorCode, ct);
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ item.NextRetryTime = DateTime.Now.AddMinutes(GetBackoffMinutes(item.RetryCount));
|
|
|
+ await MarkAsync(item, 0, responseJson, msg, item.NextRetryTime, errorCode, ct);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 可重试 vs 不可重试。配置类走 <see cref="MarkConfigIssueAsync"/>，不进本方法。
|
|
|
+ /// </summary>
|
|
|
+ private static (string? errorCode, bool permanent) ClassifyFailure(string? error)
|
|
|
+ {
|
|
|
+ if (string.IsNullOrWhiteSpace(error))
|
|
|
+ return ("UNKNOWN", false);
|
|
|
+
|
|
|
+ var trimmed = error.Trim();
|
|
|
+ foreach (var prefix in PermanentErrorPrefixes)
|
|
|
+ {
|
|
|
+ if (trimmed.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
|
|
|
+ return (prefix, true);
|
|
|
+ }
|
|
|
+
|
|
|
+ // payload 反序列化 / 解析失败 → 永久
|
|
|
+ if (trimmed.Contains("payload 解析失败", StringComparison.OrdinalIgnoreCase)
|
|
|
+ || trimmed.Contains("payload parse", StringComparison.OrdinalIgnoreCase)
|
|
|
+ || trimmed.Contains("JsonException", StringComparison.OrdinalIgnoreCase)
|
|
|
+ || trimmed.Contains("反序列化", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return ("PAYLOAD_INVALID", true);
|
|
|
+
|
|
|
+ // SQL 约束 / 唯一冲突 → 永久
|
|
|
+ if (ContainsAny(trimmed,
|
|
|
+ "unique constraint", "unique key", "duplicate key", "duplicate entry",
|
|
|
+ "primary key", "violation of unique", "violation of primary",
|
|
|
+ "cannot insert duplicate", "约束", "唯一索引", "主键冲突"))
|
|
|
+ return ("CONSTRAINT_VIOLATION", true);
|
|
|
+
|
|
|
+ if (trimmed.Contains("不支持的 op", StringComparison.OrdinalIgnoreCase))
|
|
|
+ return ("UNSUPPORTED_OP", true);
|
|
|
+
|
|
|
+ // 连接 / 超时 → 可重试
|
|
|
+ if (ContainsAny(trimmed,
|
|
|
+ "timeout", "timed out", "connection", "network", "could not open",
|
|
|
+ "unable to connect", "transport-level", "broken pipe", "socket",
|
|
|
+ "连接", "超时", "无法连接"))
|
|
|
+ return ("TRANSIENT", false);
|
|
|
+
|
|
|
+ return ("RETRYABLE", false);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static bool ContainsAny(string haystack, params string[] needles)
|
|
|
+ {
|
|
|
+ foreach (var n in needles)
|
|
|
+ {
|
|
|
+ if (haystack.Contains(n, StringComparison.OrdinalIgnoreCase))
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>retry_count 从 1 起对应 Backoff[0]；越界 clamp 到末项。</summary>
|
|
|
+ private static int GetBackoffMinutes(int retryCount)
|
|
|
+ {
|
|
|
+ var idx = Math.Max(0, retryCount - 1);
|
|
|
+ if (idx >= BackoffMinutes.Length)
|
|
|
+ idx = BackoffMinutes.Length - 1;
|
|
|
+ return BackoffMinutes[idx];
|
|
|
+ }
|
|
|
+
|
|
|
/// <summary>
|
|
|
/// 按 source_type / 连接信息解析执行器；无法解析时返回可诊断错误（不再静默落到 API）。
|
|
|
/// </summary>
|
|
|
@@ -135,24 +237,43 @@ public sealed class MdpTargetPushDispatcher : ITransient
|
|
|
return false;
|
|
|
}
|
|
|
|
|
|
- /// <summary>配置类问题：仅写 ErrorMsg/UpdateTime，status 保持 0，retry_count 不变。</summary>
|
|
|
- private async Task MarkConfigIssueAsync(MdpOutbox item, string error, CancellationToken ct)
|
|
|
+ /// <summary>配置类问题：仅写 ErrorMsg/LastErrorCode/UpdateTime，status 保持 0，retry_count 不变。</summary>
|
|
|
+ private async Task MarkConfigIssueAsync(MdpOutbox item, string errorCode, string error, CancellationToken ct)
|
|
|
{
|
|
|
item.ErrorMsg = Truncate(error, 900);
|
|
|
+ item.LastErrorCode = Truncate(errorCode, 64);
|
|
|
item.UpdateTime = DateTime.Now;
|
|
|
await _db.Updateable(item)
|
|
|
- .UpdateColumns(x => new { x.ErrorMsg, x.UpdateTime })
|
|
|
+ .UpdateColumns(x => new { x.ErrorMsg, x.LastErrorCode, x.UpdateTime })
|
|
|
.ExecuteCommandAsync(ct);
|
|
|
}
|
|
|
|
|
|
- private async Task MarkAsync(MdpOutbox item, int status, string? responseJson, string? error, CancellationToken ct)
|
|
|
+ private async Task MarkAsync(
|
|
|
+ MdpOutbox item,
|
|
|
+ int status,
|
|
|
+ string? responseJson,
|
|
|
+ string? error,
|
|
|
+ DateTime? nextRetryTime,
|
|
|
+ string? lastErrorCode,
|
|
|
+ CancellationToken ct)
|
|
|
{
|
|
|
item.Status = status;
|
|
|
item.ResponseJson = Truncate(responseJson, 4000);
|
|
|
item.ErrorMsg = error;
|
|
|
+ item.NextRetryTime = nextRetryTime;
|
|
|
+ item.LastErrorCode = Truncate(lastErrorCode, 64);
|
|
|
item.UpdateTime = DateTime.Now;
|
|
|
await _db.Updateable(item)
|
|
|
- .UpdateColumns(x => new { x.Status, x.RetryCount, x.ResponseJson, x.ErrorMsg, x.UpdateTime })
|
|
|
+ .UpdateColumns(x => new
|
|
|
+ {
|
|
|
+ x.Status,
|
|
|
+ x.RetryCount,
|
|
|
+ x.ResponseJson,
|
|
|
+ x.ErrorMsg,
|
|
|
+ x.NextRetryTime,
|
|
|
+ x.LastErrorCode,
|
|
|
+ x.UpdateTime
|
|
|
+ })
|
|
|
.ExecuteCommandAsync(ct);
|
|
|
}
|
|
|
|
YY968XX