From 9bd8e752b5d7ba7177fa16d74565b741699f5c22 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期一, 31 八月 2026 00:43:06 +0800
Subject: [PATCH] 实现 Ticket14 区间完整性检查与检测完成
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java | 121 +++++++++++++++++++++++++++++++++++++++-
1 files changed, 117 insertions(+), 4 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java
index e0a5e17..b886001 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java
@@ -27,6 +27,7 @@
import com.ruoyi.system.domain.TTestChannelBinding;
import com.ruoyi.system.domain.TTestCollectorChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
+import com.ruoyi.system.domain.TTestIntegrityIssue;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestBackfillService;
import com.ruoyi.system.service.ITTestService;
@@ -36,6 +37,7 @@
public class TestBackfillServiceImpl implements ITestBackfillService
{
private static final int PAGE_SIZE = 1000;
+ private static final int MAX_INTEGRITY_RETRIES = 3;
private final TTestMapper mapper;
private final DeviceAdapterRegistry registry;
private final BtsSimulatorStateStore collectorHistoryStore;
@@ -119,7 +121,20 @@
? List.<TTestCollectorChannel>of() : battery.getCollectorChannels())
backfillCollector(test, battery, mapping);
}
- mapper.updateTestBackfillCompleted(test.getTestId(), new Date());
+ Date checkedAt = new Date();
+ boolean complete = checkIntegrity(test, checkedAt);
+ if (complete)
+ {
+ mapper.updateTestCompleted(test.getTestId(), checkedAt);
+ mapper.insertTestAudit(test.getTestId(), "TEST_COMPLETED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("backfillStatus", "COMPLETED", "integrity", "PASSED")));
+ }
+ else
+ {
+ mapper.updateTestBackfillWaiting(test.getTestId(), checkedAt, "区间完整性检查未通过");
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_CHECK_BLOCKED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("backfillStatus", "WAITING", "integrity", "OPEN")));
+ }
}
private void backfillCabinet(TTest test, TTestBattery battery, TTestChannelBinding binding)
@@ -206,7 +221,7 @@
for (CollectorMeasurementRecord record : page.records())
{
if (!mapping.getMeasurementCode().equalsIgnoreCase(record.metricCode()) || !within(record.deviceTime(), intervals)) continue;
- saveCollectorSample(test, battery, mapping, record);
+ saveCollectorSample(test, battery, mapping, intervals, record);
}
offset = page.nextOffset();
cursor.setNextPosition(offset + 1L);
@@ -218,7 +233,7 @@
}
private void saveCollectorSample(TTest test, TTestBattery battery, TTestCollectorChannel mapping,
- CollectorMeasurementRecord record)
+ List<TTestReceptionInterval> intervals, CollectorMeasurementRecord record)
{
byte[] hash = hash(record.sequenceId() + "|" + record.deviceTimeRaw() + "|" + record.metricCode() + "|"
+ record.value() + "|" + record.unit());
@@ -230,11 +245,109 @@
"采集器=" + record.collectorCode() + ", seqid=" + record.sequenceId());
return;
}
- mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(), battery.getCollectorId(),
+ mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(),
+ intervalId(intervals, record.deviceTime()), battery.getCollectorId(),
record.channelCode(), record.sequenceId(), Date.from(record.deviceTime()), record.deviceTimeRaw(),
Date.from(record.serverReceivedAt()), record.metricCode(), record.value(), record.unit(), hash);
}
+ private boolean checkIntegrity(TTest test, Date checkedAt)
+ {
+ boolean complete = true;
+ for (TTestReceptionInterval interval : mapper.selectTestReceptionIntervals(test.getTestId()))
+ {
+ String sourceType = interval.getSourceType();
+ List<Long> sequenceIds = "CABINET_CHANNEL".equals(sourceType)
+ ? mapper.selectBtsSequencesByInterval(interval.getIntervalId())
+ : mapper.selectCollectorSequencesByInterval(interval.getIntervalId());
+ List<IntervalIntegrityChecker.Gap> gaps = IntervalIntegrityChecker.gaps(sequenceIds);
+ if (sequenceIds == null || sequenceIds.isEmpty())
+ {
+ complete = false;
+ TTestIntegrityIssue old = mapper.selectOpenIntegrityIssue(interval.getIntervalId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode(), null, null);
+ if (old == null)
+ {
+ TTestIntegrityIssue issue = new TTestIntegrityIssue();
+ issue.setTestId(test.getTestId()); issue.setIntervalId(interval.getIntervalId());
+ issue.setBindingId(interval.getBindingId()); issue.setSourceType(sourceType);
+ issue.setSourceId(interval.getSourceId()); issue.setSourceChannelCode(interval.getSourceChannelCode());
+ issue.setIssueType("NO_SAMPLES"); issue.setRetryCount(1);
+ issue.setDetail("接收区间没有可确认的历史采样"); issue.setFirstDetectedAtUtc(checkedAt);
+ issue.setLastDetectedAtUtc(checkedAt); mapper.insertIntegrityIssue(issue);
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_DETECTED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId(), "sourceType", sourceType,
+ "issueType", "NO_SAMPLES")));
+ }
+ else
+ {
+ int retry = old.getRetryCount() == null ? 1 : old.getRetryCount() + 1;
+ int cappedRetry = Math.min(retry, MAX_INTEGRITY_RETRIES);
+ mapper.touchIntegrityIssue(old.getIssueId(), cappedRetry,
+ old.getDetail(), checkedAt);
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RETRIED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("issueId", old.getIssueId(), "retryCount", cappedRetry)));
+ if (cappedRetry < MAX_INTEGRITY_RETRIES)
+ mapper.rewindBackfillCursor(test.getTestId(), interval.getBindingId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode(), 1L);
+ }
+ continue;
+ }
+ if (gaps.isEmpty())
+ {
+ int resolved = mapper.resolveIntegrityIssues(test.getTestId(), interval.getIntervalId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode(), checkedAt);
+ if (resolved > 0) mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RESOLVED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId())));
+ continue;
+ }
+ complete = false;
+ for (TTestIntegrityIssue open : mapper.selectOpenIntegrityIssues(interval.getIntervalId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode()))
+ {
+ boolean stillOpen = gaps.stream().anyMatch(gap -> Objects.equals(gap.from(), open.getMissingFromSequence())
+ && Objects.equals(gap.to(), open.getMissingToSequence()));
+ if (!stillOpen && mapper.resolveIntegrityIssue(open.getIssueId(), checkedAt) > 0)
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RESOLVED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("issueId", open.getIssueId())));
+ }
+ for (IntervalIntegrityChecker.Gap gap : gaps)
+ {
+ TTestIntegrityIssue old = mapper.selectOpenIntegrityIssue(interval.getIntervalId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode(), gap.from(), gap.to());
+ if (old == null)
+ {
+ TTestIntegrityIssue issue = new TTestIntegrityIssue();
+ issue.setTestId(test.getTestId()); issue.setIntervalId(interval.getIntervalId());
+ issue.setBindingId(interval.getBindingId()); issue.setSourceType(sourceType);
+ issue.setSourceId(interval.getSourceId()); issue.setSourceChannelCode(interval.getSourceChannelCode());
+ issue.setIssueType("SEQUENCE_GAP"); issue.setMissingFromSequence(gap.from());
+ issue.setMissingToSequence(gap.to()); issue.setRetryCount(1);
+ issue.setDetail("接收区间内存在未解释序号缺口"); issue.setFirstDetectedAtUtc(checkedAt);
+ issue.setLastDetectedAtUtc(checkedAt); mapper.insertIntegrityIssue(issue);
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_DETECTED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId(), "sourceType", sourceType,
+ "missingFrom", gap.from(), "missingTo", gap.to())));
+ }
+ else
+ {
+ int retry = old.getRetryCount() == null ? 1 : old.getRetryCount() + 1;
+ int cappedRetry = Math.min(retry, MAX_INTEGRITY_RETRIES);
+ mapper.touchIntegrityIssue(old.getIssueId(), cappedRetry,
+ old.getDetail(), checkedAt);
+ mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RETRIED", null, "system", null,
+ JSON.toJSONString(java.util.Map.of("issueId", old.getIssueId(), "retryCount", cappedRetry)));
+ if (cappedRetry >= MAX_INTEGRITY_RETRIES) continue;
+ }
+ // next_position 是分页偏移而非 seqid；从安全起点重放，依靠幂等键避免覆盖原始事实。
+ mapper.rewindBackfillCursor(test.getTestId(), interval.getBindingId(), sourceType,
+ interval.getSourceId(), interval.getSourceChannelCode(), 1L);
+ }
+ }
+ return complete && mapper.countOpenIntegrityIssues(test.getTestId()) == 0
+ && mapper.countOpenExceptions(test.getTestId()) == 0;
+ }
+
private TTestBackfillCursor cursor(TTest test, Long bindingId, String sourceType, Long sourceId,
String channelCode, String protocolTestId)
{
--
Gitblit v1.7.1