实现 Ticket14 区间完整性检查与检测完成

yeg

2026-08-31 9bd8e752b5d7ba7177fa16d74565b741699f5c22

 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)
    {