|
|
@@ -28,6 +28,7 @@ public class S8WatchSchedulerService : ITransient
|
|
|
private readonly S8OutOfRangeRuleEvaluator _outOfRangeEvaluator;
|
|
|
private readonly ILogger<S8WatchSchedulerService> _logger;
|
|
|
private readonly SqlSugarRepository<AdoS8DetectionLog> _detectionLogRep;
|
|
|
+ private readonly SqlSugarRepository<AdoS8RuleDetectionState> _detectionStateRep;
|
|
|
|
|
|
private const string DetectionTriggerSource = "WATCH_SCHEDULER";
|
|
|
private const string DetectResultCreated = "CREATED";
|
|
|
@@ -57,7 +58,8 @@ public class S8WatchSchedulerService : ITransient
|
|
|
S8ShortageRuleEvaluator shortageEvaluator,
|
|
|
S8OutOfRangeRuleEvaluator outOfRangeEvaluator,
|
|
|
ILogger<S8WatchSchedulerService> logger,
|
|
|
- SqlSugarRepository<AdoS8DetectionLog> detectionLogRep)
|
|
|
+ SqlSugarRepository<AdoS8DetectionLog> detectionLogRep,
|
|
|
+ SqlSugarRepository<AdoS8RuleDetectionState> detectionStateRep)
|
|
|
{
|
|
|
_ruleRep = ruleRep;
|
|
|
_alertRuleRep = alertRuleRep;
|
|
|
@@ -71,6 +73,7 @@ public class S8WatchSchedulerService : ITransient
|
|
|
_outOfRangeEvaluator = outOfRangeEvaluator;
|
|
|
_logger = logger;
|
|
|
_detectionLogRep = detectionLogRep;
|
|
|
+ _detectionStateRep = detectionStateRep;
|
|
|
}
|
|
|
|
|
|
public async Task<List<S8WatchExecutionRule>> LoadExecutionRulesAsync(long tenantId, long factoryId)
|
|
|
@@ -597,9 +600,54 @@ public class S8WatchSchedulerService : ITransient
|
|
|
continue;
|
|
|
}
|
|
|
|
|
|
+ // S8-SCHED-EXEC-1：建单前抗抖累计。
|
|
|
+ // 累计落 ado_s8_rule_detection_state；hitCount < trigger_count_required 时 pending、不建单、不写 CREATED 日志。
|
|
|
+ int hitCount;
|
|
|
+ AdoS8RuleDetectionState? state;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ (state, hitCount) = await UpsertDetectionStateOnHitAsync(tenantId, factoryId, rule, hit);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "antiflap_failed", ex.Message, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ var triggerRequired = NormalizeAntiflapCount(rule.TriggerCountRequired);
|
|
|
+ if (hitCount < triggerRequired)
|
|
|
+ {
|
|
|
+ _logger.LogInformation(
|
|
|
+ "antiflap_pending_hit ruleCode={RuleCode} dedupKey={DedupKey} hitCount={HitCount} trigger={Trigger}",
|
|
|
+ rule.RuleCode, hit.DedupKey, hitCount, triggerRequired);
|
|
|
+ results.Add(BuildSkipResult(rule, "antiflap_pending_hit", null, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
try
|
|
|
{
|
|
|
var entity = await _manualReportService.CreateFromHitAsync(hit);
|
|
|
+ // 新建异常的抗抖累计与触发计数对齐，便于后续 reconcile 解释累计来源。
|
|
|
+ await _exceptionRep.Context.Updateable<AdoS8Exception>()
|
|
|
+ .SetColumns(x => new AdoS8Exception
|
|
|
+ {
|
|
|
+ ConsecutiveHitCount = hitCount,
|
|
|
+ ConsecutiveMissCount = 0,
|
|
|
+ UpdatedAt = DateTime.Now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == entity.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ if (state != null)
|
|
|
+ {
|
|
|
+ await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
|
|
|
+ .SetColumns(x => new AdoS8RuleDetectionState
|
|
|
+ {
|
|
|
+ ActiveExceptionId = entity.Id,
|
|
|
+ UpdatedAt = DateTime.Now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == state.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ }
|
|
|
await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultCreated, entity.Id, runId));
|
|
|
results.Add(BuildCreatedResult(rule, hit, entity.Id));
|
|
|
}
|
|
|
@@ -613,6 +661,68 @@ public class S8WatchSchedulerService : ITransient
|
|
|
return results;
|
|
|
}
|
|
|
|
|
|
+ // S8-SCHED-EXEC-1：trigger / recover 抗抖计数兜底，null / <1 / >10 一律按 1，避免非法配置导致永远不建单 / 永远不恢复。
|
|
|
+ private static int NormalizeAntiflapCount(int raw)
|
|
|
+ {
|
|
|
+ if (raw < 1 || raw > 10) return 1;
|
|
|
+ return raw;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 命中时累加 detection_state.consecutive_hit_count；未存在则插入 hitCount=1。
|
|
|
+ /// 返回当前 state 行（含 Id）以及命中后的 hitCount。
|
|
|
+ /// 注意：本函数不消费 trigger_count_required；上游决定是否进入 CreateFromHitAsync。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<(AdoS8RuleDetectionState? state, int hitCount)> UpsertDetectionStateOnHitAsync(
|
|
|
+ long tenantId, long factoryId, AdoS8WatchRule rule, S8RuleHit hit)
|
|
|
+ {
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var existing = await _detectionStateRep.AsQueryable()
|
|
|
+ .Where(x => x.TenantId == tenantId
|
|
|
+ && x.FactoryId == factoryId
|
|
|
+ && x.RuleCode == rule.RuleCode
|
|
|
+ && x.DedupKey == hit.DedupKey)
|
|
|
+ .FirstAsync();
|
|
|
+
|
|
|
+ if (existing == null)
|
|
|
+ {
|
|
|
+ var fresh = new AdoS8RuleDetectionState
|
|
|
+ {
|
|
|
+ TenantId = tenantId,
|
|
|
+ FactoryId = factoryId,
|
|
|
+ RuleCode = rule.RuleCode,
|
|
|
+ DedupKey = hit.DedupKey,
|
|
|
+ SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? null : hit.SourceObjectType,
|
|
|
+ SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? null : hit.SourceObjectId,
|
|
|
+ ConsecutiveHitCount = 1,
|
|
|
+ ConsecutiveMissCount = 0,
|
|
|
+ LastSeenAt = now,
|
|
|
+ LastHitAt = now,
|
|
|
+ CreatedAt = now,
|
|
|
+ UpdatedAt = now
|
|
|
+ };
|
|
|
+ await _detectionStateRep.InsertAsync(fresh);
|
|
|
+ return (fresh, 1);
|
|
|
+ }
|
|
|
+
|
|
|
+ var newHitCount = existing.ConsecutiveHitCount + 1;
|
|
|
+ await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
|
|
|
+ .SetColumns(x => new AdoS8RuleDetectionState
|
|
|
+ {
|
|
|
+ ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
|
|
|
+ ConsecutiveMissCount = 0,
|
|
|
+ LastSeenAt = now,
|
|
|
+ LastHitAt = now,
|
|
|
+ SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? existing.SourceObjectType : hit.SourceObjectType,
|
|
|
+ SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? existing.SourceObjectId : hit.SourceObjectId,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == existing.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ existing.ConsecutiveHitCount = newHitCount;
|
|
|
+ return (existing, newHitCount);
|
|
|
+ }
|
|
|
+
|
|
|
private async Task<long> FindOpenExceptionByDedupKeyAsync(long tenantId, long factoryId, string dedupKey)
|
|
|
{
|
|
|
var ids = await _exceptionRep.AsQueryable()
|
|
|
@@ -692,15 +802,56 @@ public class S8WatchSchedulerService : ITransient
|
|
|
&& x.SourceRuleCode == rule.RuleCode
|
|
|
&& x.DedupKey != null
|
|
|
&& x.RecoveredAt == null)
|
|
|
- .Select(x => new { x.Id, x.DedupKey, x.SourceObjectType, x.SourceObjectId, x.RelatedObjectCode })
|
|
|
+ .Select(x => new { x.Id, x.DedupKey, x.SourceObjectType, x.SourceObjectId, x.RelatedObjectCode, x.ConsecutiveMissCount })
|
|
|
.ToListAsync();
|
|
|
if (candidates.Count == 0) return new List<long>();
|
|
|
|
|
|
var now = DateTime.Now;
|
|
|
+ var recoverRequired = NormalizeAntiflapCount(rule.RecoverCountRequired);
|
|
|
var recoveredIds = new List<long>();
|
|
|
foreach (var c in candidates)
|
|
|
{
|
|
|
if (hitDedupKeys.Contains(c.DedupKey!)) continue;
|
|
|
+
|
|
|
+ // S8-SCHED-EXEC-1：恢复抗抖累计。
|
|
|
+ // 1) 每次未命中：异常 ConsecutiveMissCount += 1，ConsecutiveHitCount 清零；
|
|
|
+ // 2) miss < recover_count_required：仅累计，不写 recovered_at、不写 RECOVERED；
|
|
|
+ // 3) miss >= recover_count_required：写 recovered_at、写 RECOVERED 日志。
|
|
|
+ var newMissCount = c.ConsecutiveMissCount + 1;
|
|
|
+ await _exceptionRep.Context.Updateable<AdoS8Exception>()
|
|
|
+ .SetColumns(x => new AdoS8Exception
|
|
|
+ {
|
|
|
+ ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
|
|
|
+ ConsecutiveHitCount = 0,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == c.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+
|
|
|
+ // detection_state 同步累计 miss（建单后 state.active_exception_id 仍指向 c.Id）。
|
|
|
+ await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
|
|
|
+ .SetColumns(x => new AdoS8RuleDetectionState
|
|
|
+ {
|
|
|
+ ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
|
|
|
+ ConsecutiveHitCount = 0,
|
|
|
+ LastSeenAt = now,
|
|
|
+ LastMissAt = now,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.TenantId == tenantId
|
|
|
+ && x.FactoryId == factoryId
|
|
|
+ && x.RuleCode == rule.RuleCode
|
|
|
+ && x.DedupKey == c.DedupKey)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+
|
|
|
+ if (newMissCount < recoverRequired)
|
|
|
+ {
|
|
|
+ _logger.LogInformation(
|
|
|
+ "antiflap_pending_recovery ruleCode={RuleCode} dedupKey={DedupKey} missCount={Miss} recoverRequired={Required}",
|
|
|
+ rule.RuleCode, c.DedupKey, newMissCount, recoverRequired);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
await _exceptionRep.Context.Updateable<AdoS8Exception>()
|
|
|
.SetColumns(x => new AdoS8Exception
|
|
|
{
|
|
|
@@ -720,7 +871,7 @@ public class S8WatchSchedulerService : ITransient
|
|
|
DetectResult = DetectResultRecovered,
|
|
|
ExceptionId = c.Id,
|
|
|
DetectedAt = now,
|
|
|
- PayloadSnapshot = JsonSerializer.Serialize(new { ruleId = rule.Id, ruleCode = rule.RuleCode, reason = "no_longer_hit" }),
|
|
|
+ PayloadSnapshot = JsonSerializer.Serialize(new { ruleId = rule.Id, ruleCode = rule.RuleCode, reason = "no_longer_hit", missCount = newMissCount, recoverRequired }),
|
|
|
RunId = runId, TriggerSource = DetectionTriggerSource,
|
|
|
Remark = "Rule no longer hit; recovered_at marked"
|
|
|
});
|
|
|
@@ -772,11 +923,17 @@ public class S8WatchSchedulerService : ITransient
|
|
|
|
|
|
private async Task RefreshDetectionAsync(long exceptionId, S8RuleHit hit)
|
|
|
{
|
|
|
+ // S8-SCHED-EXEC-1：刷新阶段抗抖累计 + 复发清空 recovered_at。
|
|
|
+ // ConsecutiveHitCount += 1（用 SetColumns 内表达式完成原子自增）；ConsecutiveMissCount 归零。
|
|
|
+ // RecoveredAt 不为 NULL 时（复发）一并清空，保持业务对"再次命中即视为活跃"的预期。
|
|
|
await _exceptionRep.Context.Updateable<AdoS8Exception>()
|
|
|
.SetColumns(x => new AdoS8Exception
|
|
|
{
|
|
|
LastDetectedAt = hit.DetectedAt,
|
|
|
SourcePayload = hit.SourcePayload,
|
|
|
+ ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
|
|
|
+ ConsecutiveMissCount = 0,
|
|
|
+ RecoveredAt = null,
|
|
|
UpdatedAt = DateTime.Now
|
|
|
})
|
|
|
.Where(x => x.Id == exceptionId)
|
|
|
@@ -985,6 +1142,538 @@ public class S8WatchSchedulerService : ITransient
|
|
|
|
|
|
return JsonSerializer.Serialize(payload);
|
|
|
}
|
|
|
+
|
|
|
+ // ============================================================
|
|
|
+ // S8-SCHED-EXEC-1：DB 驱动调度执行层
|
|
|
+ // ============================================================
|
|
|
+
|
|
|
+ private const int LeaseDurationMinutes = 5;
|
|
|
+ private const int AutoPauseFailureThreshold = 3;
|
|
|
+ private const int AutoPauseDurationHours = 1;
|
|
|
+ private const int DefaultPollIntervalSeconds = 300;
|
|
|
+ private const int MinPollIntervalSeconds = 60;
|
|
|
+ private const int MaxPollIntervalSeconds = 86400;
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：释放过期 lease（lock_until < NOW），不修改 last_status / last_error，仅清空 lock 三件套 + running_started_at。
|
|
|
+ /// 返回释放的行数。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<int> ResetExpiredLeasesAsync(long tenantId, long factoryId)
|
|
|
+ {
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
|
|
|
+ .SetColumns(x => new AdoS8WatchRule
|
|
|
+ {
|
|
|
+ LockToken = null,
|
|
|
+ LockedBy = null,
|
|
|
+ LockUntil = null,
|
|
|
+ RunningStartedAt = null,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.TenantId == tenantId
|
|
|
+ && x.FactoryId == factoryId
|
|
|
+ && x.LockUntil != null
|
|
|
+ && x.LockUntil < now)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ if (affected > 0)
|
|
|
+ {
|
|
|
+ _logger.LogWarning(
|
|
|
+ "lease_reset tenantId={Tenant} factoryId={Factory} releasedCount={Count}",
|
|
|
+ tenantId, factoryId, affected);
|
|
|
+ }
|
|
|
+ return affected;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：到期规则候选 + 乐观 UPDATE 抢锁，返回成功抢到的 lease 列表。
|
|
|
+ /// 抢锁条件：enabled=1 AND (paused_until IS NULL OR paused_until <= NOW)
|
|
|
+ /// AND (next_run_at IS NULL OR next_run_at <= NOW)
|
|
|
+ /// AND (lock_until IS NULL OR lock_until <= NOW)。
|
|
|
+ /// 抢锁回写：lock_token / locked_by / lock_until = NOW + 5min / running_started_at = NOW / last_run_id = runId。
|
|
|
+ /// affectedRows == 1 才算抢到；后续 OnRuleCompletedAsync 必须按 lockToken 回写，避免旧进程覆盖新 lease。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<List<S8RuleLease>> PickReadyRulesAsync(long tenantId, long factoryId, int batchSize, string lockedBy, string runId)
|
|
|
+ {
|
|
|
+ if (batchSize <= 0) batchSize = 16;
|
|
|
+ var now = DateTime.Now;
|
|
|
+
|
|
|
+ var candidates = await _ruleRep.AsQueryable()
|
|
|
+ .Where(x => x.TenantId == tenantId
|
|
|
+ && x.FactoryId == factoryId
|
|
|
+ && x.Enabled
|
|
|
+ && (x.PausedUntil == null || x.PausedUntil <= now)
|
|
|
+ && (x.NextRunAt == null || x.NextRunAt <= now)
|
|
|
+ && (x.LockUntil == null || x.LockUntil <= now))
|
|
|
+ .OrderBy(x => x.NextRunAt, OrderByType.Asc)
|
|
|
+ .OrderBy(x => x.Id, OrderByType.Asc)
|
|
|
+ .Take(batchSize)
|
|
|
+ .Select(x => new { x.Id, x.RuleCode, x.RuleType })
|
|
|
+ .ToListAsync();
|
|
|
+ if (candidates.Count == 0) return new();
|
|
|
+
|
|
|
+ var leases = new List<S8RuleLease>();
|
|
|
+ foreach (var c in candidates)
|
|
|
+ {
|
|
|
+ var token = Guid.NewGuid().ToString("N");
|
|
|
+ var lockUntil = DateTime.Now.AddMinutes(LeaseDurationMinutes);
|
|
|
+ var runningAt = DateTime.Now;
|
|
|
+ // 乐观 UPDATE：再校验一次条件，affectedRows=1 才算抢到。
|
|
|
+ var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
|
|
|
+ .SetColumns(x => new AdoS8WatchRule
|
|
|
+ {
|
|
|
+ LockToken = token,
|
|
|
+ LockedBy = lockedBy,
|
|
|
+ LockUntil = lockUntil,
|
|
|
+ RunningStartedAt = runningAt,
|
|
|
+ LastRunId = runId,
|
|
|
+ UpdatedAt = runningAt
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == c.Id
|
|
|
+ && x.Enabled
|
|
|
+ && (x.PausedUntil == null || x.PausedUntil <= runningAt)
|
|
|
+ && (x.NextRunAt == null || x.NextRunAt <= runningAt)
|
|
|
+ && (x.LockUntil == null || x.LockUntil <= runningAt))
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ if (affected == 1)
|
|
|
+ {
|
|
|
+ leases.Add(new S8RuleLease
|
|
|
+ {
|
|
|
+ RuleId = c.Id,
|
|
|
+ RuleCode = c.RuleCode,
|
|
|
+ RuleType = c.RuleType,
|
|
|
+ LockToken = token,
|
|
|
+ LockedBy = lockedBy,
|
|
|
+ LockUntil = lockUntil,
|
|
|
+ RunId = runId,
|
|
|
+ AcquiredAt = runningAt
|
|
|
+ });
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return leases;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：执行单条已抢锁规则的 evaluator → 抗抖去重 → 建单/刷新 → 恢复 reconcile。
|
|
|
+ /// 不释放 lease（OnRuleCompletedAsync 负责）；evaluator 抛异常时 Result.Success=false 并保留 ErrorMessage。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<S8RuleRunResult> RunSingleRuleAsync(long tenantId, long factoryId, S8RuleLease lease)
|
|
|
+ {
|
|
|
+ var rule = await _ruleRep.AsQueryable()
|
|
|
+ .Where(x => x.Id == lease.RuleId)
|
|
|
+ .FirstAsync();
|
|
|
+ if (rule == null)
|
|
|
+ {
|
|
|
+ return new S8RuleRunResult { Success = false, ErrorMessage = "rule_not_found", Stats = new() };
|
|
|
+ }
|
|
|
+
|
|
|
+ var ruleType = rule.RuleType;
|
|
|
+ if (string.IsNullOrWhiteSpace(ruleType))
|
|
|
+ {
|
|
|
+ // 未分类的旧规则不在新调度路径承载；标 SKIPPED 但不视为失败。
|
|
|
+ return new S8RuleRunResult { Success = true, ErrorMessage = "rule_type_empty_skipped", Stats = new() };
|
|
|
+ }
|
|
|
+
|
|
|
+ IS8RuleEvaluator? evaluator = ruleType switch
|
|
|
+ {
|
|
|
+ S8TimeoutRuleEvaluator.RuleTypeCode => _timeoutEvaluator,
|
|
|
+ S8ShortageRuleEvaluator.RuleTypeCode => _shortageEvaluator,
|
|
|
+ S8OutOfRangeRuleEvaluator.RuleTypeCode => _outOfRangeEvaluator,
|
|
|
+ _ => null
|
|
|
+ };
|
|
|
+ if (evaluator == null)
|
|
|
+ {
|
|
|
+ return new S8RuleRunResult { Success = false, ErrorMessage = $"unsupported_rule_type:{ruleType}", Stats = new() };
|
|
|
+ }
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ var alertRules = (await _alertRuleRep.AsQueryable()
|
|
|
+ .Where(x => x.TenantId == tenantId && x.FactoryId == factoryId)
|
|
|
+ .ToListAsync()).AsReadOnly();
|
|
|
+
|
|
|
+ var results = await ProcessSingleRuleAsync(tenantId, factoryId, rule, ruleType, evaluator, alertRules, lease.RunId);
|
|
|
+ var stats = new S8RuleRunStats
|
|
|
+ {
|
|
|
+ Hits = results.Count,
|
|
|
+ Created = results.Count(r => r.Created),
|
|
|
+ Refreshed = results.Count(r => r.Reason == "duplicate_pending"),
|
|
|
+ Pending = results.Count(r => r.Reason == "antiflap_pending_hit"),
|
|
|
+ Failed = results.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
|
|
|
+ };
|
|
|
+ return new S8RuleRunResult { Success = true, Stats = stats };
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ return new S8RuleRunResult
|
|
|
+ {
|
|
|
+ Success = false,
|
|
|
+ ErrorMessage = ex.Message,
|
|
|
+ Stats = new()
|
|
|
+ };
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：单规则处理（evaluator → reconcile → hit 循环）。
|
|
|
+ /// 与 ProcessRulesByTypeAsync 内单规则循环体语义一致；此处抽出便于新调度路径直接调用单条 rule。
|
|
|
+ /// </summary>
|
|
|
+ private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync(
|
|
|
+ long tenantId, long factoryId, AdoS8WatchRule rule, string ruleType,
|
|
|
+ IS8RuleEvaluator evaluator, IReadOnlyList<AdoS8AlertRule> alertRules, string runId)
|
|
|
+ {
|
|
|
+ var results = new List<S8WatchCreationResult>();
|
|
|
+ List<S8RuleHit> hits;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ hits = await evaluator.EvaluateAsync(tenantId, factoryId, rule, alertRules);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ var failureReason = ex is S8RuleEvaluatorException sre ? sre.Reason : ex.GetType().Name;
|
|
|
+ await WriteDetectionLogAsync(new AdoS8DetectionLog
|
|
|
+ {
|
|
|
+ TenantId = tenantId, FactoryId = factoryId,
|
|
|
+ RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
|
|
|
+ SourceObjectType = rule.SourceObjectType,
|
|
|
+ DetectResult = DetectResultEvaluateFailed,
|
|
|
+ DetectedAt = DateTime.Now,
|
|
|
+ FailureReason = failureReason,
|
|
|
+ FailureMessage = Truncate(ex.Message, 1000),
|
|
|
+ RunId = runId, TriggerSource = DetectionTriggerSource
|
|
|
+ });
|
|
|
+ results.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
|
|
|
+ throw;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<long> recoveredIds;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ recoveredIds = await ReconcileRecoveriesForRuleAsync(tenantId, factoryId, rule, ruleType, hits, runId);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _logger.LogWarning(ex, "recovery_reconcile_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
|
|
|
+ recoveredIds = new();
|
|
|
+ }
|
|
|
+
|
|
|
+ if (hits.Count == 0 && recoveredIds.Count == 0)
|
|
|
+ {
|
|
|
+ await WriteDetectionLogAsync(new AdoS8DetectionLog
|
|
|
+ {
|
|
|
+ TenantId = tenantId, FactoryId = factoryId,
|
|
|
+ RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
|
|
|
+ SourceObjectType = rule.SourceObjectType,
|
|
|
+ DetectResult = DetectResultNoHit,
|
|
|
+ DetectedAt = DateTime.Now,
|
|
|
+ RunId = runId, TriggerSource = DetectionTriggerSource
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ foreach (var hit in hits)
|
|
|
+ {
|
|
|
+ if (string.IsNullOrWhiteSpace(hit.DedupKey))
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "missing_dedup_key", null, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ long matchedId;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ matchedId = await FindOpenExceptionByDedupKeyAsync(tenantId, factoryId, hit.DedupKey);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (matchedId > 0)
|
|
|
+ {
|
|
|
+ try
|
|
|
+ {
|
|
|
+ await RefreshDetectionAsync(matchedId, hit);
|
|
|
+ await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultRefreshed, matchedId, runId));
|
|
|
+ results.Add(BuildSkippedDuplicate(rule, hit, matchedId));
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (string.Equals(ruleType, S8OutOfRangeRuleEvaluator.RuleTypeCode, StringComparison.OrdinalIgnoreCase))
|
|
|
+ {
|
|
|
+ long compatId;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ compatId = await FindLegacyOutOfRangeExceptionAsync(tenantId, factoryId, rule.Id, hit.RelatedObjectCode);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (compatId > 0)
|
|
|
+ {
|
|
|
+ try
|
|
|
+ {
|
|
|
+ await BackfillLegacyExceptionAsync(compatId, hit);
|
|
|
+ await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultRefreshed, compatId, runId));
|
|
|
+ results.Add(BuildSkippedDuplicate(rule, hit, compatId));
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ bool typeExists;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ typeExists = await _exceptionTypeRep.AsQueryable()
|
|
|
+ .Where(t => t.TypeCode == hit.ExceptionTypeCode
|
|
|
+ && (t.TenantId == 0 || t.TenantId == tenantId)
|
|
|
+ && (t.FactoryId == 0 || t.FactoryId == factoryId)
|
|
|
+ && t.Enabled)
|
|
|
+ .AnyAsync();
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (!typeExists)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "exception_type_missing", null, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ int hitCount;
|
|
|
+ AdoS8RuleDetectionState? state;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ (state, hitCount) = await UpsertDetectionStateOnHitAsync(tenantId, factoryId, rule, hit);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "antiflap_failed", ex.Message, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ var triggerRequired = NormalizeAntiflapCount(rule.TriggerCountRequired);
|
|
|
+ if (hitCount < triggerRequired)
|
|
|
+ {
|
|
|
+ _logger.LogInformation(
|
|
|
+ "antiflap_pending_hit ruleCode={RuleCode} dedupKey={DedupKey} hitCount={HitCount} trigger={Trigger}",
|
|
|
+ rule.RuleCode, hit.DedupKey, hitCount, triggerRequired);
|
|
|
+ results.Add(BuildSkipResult(rule, "antiflap_pending_hit", null, hit));
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ var entity = await _manualReportService.CreateFromHitAsync(hit);
|
|
|
+ await _exceptionRep.Context.Updateable<AdoS8Exception>()
|
|
|
+ .SetColumns(x => new AdoS8Exception
|
|
|
+ {
|
|
|
+ ConsecutiveHitCount = hitCount,
|
|
|
+ ConsecutiveMissCount = 0,
|
|
|
+ UpdatedAt = DateTime.Now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == entity.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ if (state != null)
|
|
|
+ {
|
|
|
+ await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
|
|
|
+ .SetColumns(x => new AdoS8RuleDetectionState
|
|
|
+ {
|
|
|
+ ActiveExceptionId = entity.Id,
|
|
|
+ UpdatedAt = DateTime.Now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == state.Id)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ }
|
|
|
+ await WriteDetectionLogAsync(BuildHitLog(tenantId, factoryId, rule, ruleType, hit, DetectResultCreated, entity.Id, runId));
|
|
|
+ results.Add(BuildCreatedResult(rule, hit, entity.Id));
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ results.Add(BuildSkipResult(rule, "create_failed", ex.Message, hit));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return results;
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：lease 执行完成回写。
|
|
|
+ /// 必须 WHERE id = lease.RuleId AND lock_token = lease.LockToken；affectedRows = 0 视为 lease 丢失，记录 Warning，不覆盖状态。
|
|
|
+ /// 失败 ≥ 阈值（默认 3）写 paused_until = NOW + 1h。
|
|
|
+ /// </summary>
|
|
|
+ public async Task OnRuleCompletedAsync(long tenantId, long factoryId, S8RuleLease lease, S8RuleRunResult result, int durationMs)
|
|
|
+ {
|
|
|
+ var rule = await _ruleRep.AsQueryable().Where(x => x.Id == lease.RuleId).FirstAsync();
|
|
|
+ if (rule == null)
|
|
|
+ {
|
|
|
+ _logger.LogWarning("lease_complete_rule_missing ruleId={RuleId}", lease.RuleId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ var now = DateTime.Now;
|
|
|
+ var effectiveInterval = NormalizePollInterval(rule.PollIntervalSeconds);
|
|
|
+ var nextRunAt = now.AddSeconds(effectiveInterval);
|
|
|
+ int affected;
|
|
|
+ if (result.Success)
|
|
|
+ {
|
|
|
+ affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
|
|
|
+ .SetColumns(x => new AdoS8WatchRule
|
|
|
+ {
|
|
|
+ LastRunAt = now,
|
|
|
+ NextRunAt = nextRunAt,
|
|
|
+ LastStatus = "SUCCESS",
|
|
|
+ LastError = null,
|
|
|
+ LastDurationMs = durationMs,
|
|
|
+ ConsecutiveFailureCount = 0,
|
|
|
+ LockToken = null,
|
|
|
+ LockedBy = null,
|
|
|
+ LockUntil = null,
|
|
|
+ RunningStartedAt = null,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ var errorTrunc = Truncate(result.ErrorMessage, 500);
|
|
|
+ var newFailures = rule.ConsecutiveFailureCount + 1;
|
|
|
+ DateTime? pausedUntil = rule.PausedUntil;
|
|
|
+ string? pauseReason = rule.PauseReason;
|
|
|
+ if (newFailures >= AutoPauseFailureThreshold)
|
|
|
+ {
|
|
|
+ pausedUntil = now.AddHours(AutoPauseDurationHours);
|
|
|
+ pauseReason = Truncate($"AUTO_PAUSED_AFTER_{AutoPauseFailureThreshold}_FAILURES: {errorTrunc}", 64);
|
|
|
+ }
|
|
|
+
|
|
|
+ affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
|
|
|
+ .SetColumns(x => new AdoS8WatchRule
|
|
|
+ {
|
|
|
+ LastRunAt = now,
|
|
|
+ NextRunAt = nextRunAt,
|
|
|
+ LastStatus = "FAILED",
|
|
|
+ LastError = errorTrunc,
|
|
|
+ LastDurationMs = durationMs,
|
|
|
+ ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
|
|
|
+ PausedUntil = pausedUntil,
|
|
|
+ PauseReason = pauseReason,
|
|
|
+ LockToken = null,
|
|
|
+ LockedBy = null,
|
|
|
+ LockUntil = null,
|
|
|
+ RunningStartedAt = null,
|
|
|
+ UpdatedAt = now
|
|
|
+ })
|
|
|
+ .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
|
|
|
+ .ExecuteCommandAsync();
|
|
|
+ }
|
|
|
+
|
|
|
+ if (affected == 0)
|
|
|
+ {
|
|
|
+ _logger.LogWarning(
|
|
|
+ "lease_lost_on_complete ruleId={RuleId} ruleCode={RuleCode} lockToken={LockToken}",
|
|
|
+ lease.RuleId, lease.RuleCode, lease.LockToken);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// S8-SCHED-EXEC-1：单 tick 完整流程。Job / debug 调度入口。
|
|
|
+ /// 1) ResetExpiredLeasesAsync
|
|
|
+ /// 2) PickReadyRulesAsync(batchSize)
|
|
|
+ /// 3) 每条 rule 独立 try/catch 调用 RunSingleRuleAsync + OnRuleCompletedAsync
|
|
|
+ /// 单条规则失败不影响其他规则；整 tick 不抛异常。
|
|
|
+ /// </summary>
|
|
|
+ public async Task<S8DispatchTickResult> RunDispatchTickAsync(long tenantId, long factoryId, int batchSize, string lockedBy)
|
|
|
+ {
|
|
|
+ var tickId = Guid.NewGuid().ToString("N").Substring(0, 8);
|
|
|
+ var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
|
|
|
+ var summary = new S8DispatchTickResult { TickId = tickId, RunId = runId };
|
|
|
+
|
|
|
+ try
|
|
|
+ {
|
|
|
+ summary.LeaseReleased = await ResetExpiredLeasesAsync(tenantId, factoryId);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _logger.LogError(ex, "tick_reset_lease_failed tickId={TickId}", tickId);
|
|
|
+ }
|
|
|
+
|
|
|
+ List<S8RuleLease> leases;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ leases = await PickReadyRulesAsync(tenantId, factoryId, batchSize, lockedBy, runId);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _logger.LogError(ex, "tick_pick_failed tickId={TickId}", tickId);
|
|
|
+ return summary;
|
|
|
+ }
|
|
|
+ summary.Picked = leases.Count;
|
|
|
+
|
|
|
+ foreach (var lease in leases)
|
|
|
+ {
|
|
|
+ var sw = System.Diagnostics.Stopwatch.StartNew();
|
|
|
+ S8RuleRunResult runResult;
|
|
|
+ try
|
|
|
+ {
|
|
|
+ runResult = await RunSingleRuleAsync(tenantId, factoryId, lease);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ runResult = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
|
|
|
+ }
|
|
|
+ sw.Stop();
|
|
|
+ try
|
|
|
+ {
|
|
|
+ await OnRuleCompletedAsync(tenantId, factoryId, lease, runResult, (int)sw.ElapsedMilliseconds);
|
|
|
+ }
|
|
|
+ catch (Exception ex)
|
|
|
+ {
|
|
|
+ _logger.LogError(ex, "tick_complete_failed tickId={TickId} ruleId={RuleId} ruleCode={RuleCode}", tickId, lease.RuleId, lease.RuleCode);
|
|
|
+ }
|
|
|
+
|
|
|
+ _logger.LogInformation(
|
|
|
+ "tick_rule_done tickId={TickId} runId={RunId} ruleId={RuleId} ruleCode={RuleCode} status={Status} durationMs={Dur} hits={Hits} created={Created} refreshed={Refreshed} pending={Pending} failed={Failed} error={Error}",
|
|
|
+ tickId, runId, lease.RuleId, lease.RuleCode,
|
|
|
+ runResult.Success ? "SUCCESS" : "FAILED",
|
|
|
+ sw.ElapsedMilliseconds,
|
|
|
+ runResult.Stats.Hits, runResult.Stats.Created, runResult.Stats.Refreshed, runResult.Stats.Pending, runResult.Stats.Failed,
|
|
|
+ runResult.ErrorMessage);
|
|
|
+
|
|
|
+ if (runResult.Success)
|
|
|
+ {
|
|
|
+ summary.Success++;
|
|
|
+ summary.Created += runResult.Stats.Created;
|
|
|
+ summary.Refreshed += runResult.Stats.Refreshed;
|
|
|
+ summary.Pending += runResult.Stats.Pending;
|
|
|
+ summary.PerRuleFailed += runResult.Stats.Failed;
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ summary.Failed++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return summary;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static int NormalizePollInterval(int raw)
|
|
|
+ {
|
|
|
+ if (raw < MinPollIntervalSeconds || raw > MaxPollIntervalSeconds) return DefaultPollIntervalSeconds;
|
|
|
+ return raw;
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
public sealed class S8WatchExecutionRule
|
|
|
@@ -1093,3 +1782,48 @@ public sealed class S8WatchHitResult
|
|
|
public long? ResponsibleDeptId { get; set; }
|
|
|
public string SourcePayload { get; set; } = string.Empty;
|
|
|
}
|
|
|
+
|
|
|
+/// <summary>S8-SCHED-EXEC-1：lease 抢占成功后传递的最小标识对象。</summary>
|
|
|
+public sealed class S8RuleLease
|
|
|
+{
|
|
|
+ public long RuleId { get; set; }
|
|
|
+ public string RuleCode { get; set; } = string.Empty;
|
|
|
+ public string? RuleType { get; set; }
|
|
|
+ public string LockToken { get; set; } = string.Empty;
|
|
|
+ public string LockedBy { get; set; } = string.Empty;
|
|
|
+ public DateTime LockUntil { get; set; }
|
|
|
+ public string RunId { get; set; } = string.Empty;
|
|
|
+ public DateTime AcquiredAt { get; set; }
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>S8-SCHED-EXEC-1：单条规则执行结果，OnRuleCompletedAsync 据此更新状态。</summary>
|
|
|
+public sealed class S8RuleRunResult
|
|
|
+{
|
|
|
+ public bool Success { get; set; }
|
|
|
+ public string? ErrorMessage { get; set; }
|
|
|
+ public S8RuleRunStats Stats { get; set; } = new();
|
|
|
+}
|
|
|
+
|
|
|
+public sealed class S8RuleRunStats
|
|
|
+{
|
|
|
+ public int Hits { get; set; }
|
|
|
+ public int Created { get; set; }
|
|
|
+ public int Refreshed { get; set; }
|
|
|
+ public int Pending { get; set; }
|
|
|
+ public int Failed { get; set; }
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>S8-SCHED-EXEC-1：单 tick 调度结果聚合。</summary>
|
|
|
+public sealed class S8DispatchTickResult
|
|
|
+{
|
|
|
+ public string TickId { get; set; } = string.Empty;
|
|
|
+ public string RunId { get; set; } = string.Empty;
|
|
|
+ public int LeaseReleased { get; set; }
|
|
|
+ public int Picked { get; set; }
|
|
|
+ public int Success { get; set; }
|
|
|
+ public int Failed { get; set; }
|
|
|
+ public int Created { get; set; }
|
|
|
+ public int Refreshed { get; set; }
|
|
|
+ public int Pending { get; set; }
|
|
|
+ public int PerRuleFailed { get; set; }
|
|
|
+}
|
YY968XX