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

yeg

2026-08-31 9bd8e752b5d7ba7177fa16d74565b741699f5c22

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

 .gitignore

@@ -15,5 +15,6 @@
!/ruoyi-admin/src/main/resources/db/migration/V9__move_discovery_candidates_to_redis.sql
!/ruoyi-admin/src/main/resources/db/migration/V19__collector_disconnect_open_dedupe.sql
!/ruoyi-admin/src/main/resources/db/migration/V21__natural_end_interval_backfill.sql
!/ruoyi-admin/src/main/resources/db/migration/V22__interval_integrity_and_completion.sql
/.mvn-review-repository/
/ruoyi-admin/logs/

 ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java

@@ -10,6 +10,7 @@
import com.ruoyi.common.core.controller.BaseController;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITestIntegrityService;

/** 测试控制台实时 REST/SSE 入口，仅负责鉴权和协议转发。 */
@RestController
@@ -17,13 +18,19 @@
public class TTestLiveController extends BaseController
{
    private final ITestLiveService liveService;
    public TTestLiveController(ITestLiveService liveService) { this.liveService = liveService; }
    private final ITestIntegrityService integrityService;
    public TTestLiveController(ITestLiveService liveService, ITestIntegrityService integrityService)
    { this.liveService = liveService; this.integrityService = integrityService; }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/live-state")
    public AjaxResult liveState(@PathVariable Long testId) { return success(liveService.current(testId)); }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/integrity")
    public AjaxResult integrity(@PathVariable Long testId) { return success(integrityService.selectIssues(testId)); }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping(value = "/{testId}/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter live(@PathVariable Long testId) { return liveService.stream(testId); }
}

 ruoyi-admin/src/main/resources/db/migration/V22__interval_integrity_and_completion.sql

New file
@@ -0,0 +1,34 @@
-- Ticket14：按接收区间保存序号完整性问题，并为采集器历史记录保留区间归属。
ALTER TABLE t_test_collector_sample
    ADD COLUMN reception_interval_id BIGINT NULL AFTER binding_id,
    ADD KEY idx_collector_sample_interval (reception_interval_id),
    ADD CONSTRAINT fk_collector_sample_reception_interval FOREIGN KEY (reception_interval_id)
        REFERENCES t_test_reception_interval (interval_id);

CREATE TABLE t_test_integrity_issue (
    issue_id BIGINT NOT NULL AUTO_INCREMENT,
    test_id BIGINT NOT NULL,
    interval_id BIGINT NOT NULL,
    binding_id BIGINT NOT NULL,
    source_type VARCHAR(32) NOT NULL,
    source_id BIGINT NOT NULL,
    source_channel_code VARCHAR(64) NULL,
    issue_type VARCHAR(64) NOT NULL,
    issue_status VARCHAR(16) NOT NULL DEFAULT 'OPEN',
    missing_from_sequence BIGINT NULL,
    missing_to_sequence BIGINT NULL,
    retry_count INT NOT NULL DEFAULT 0,
    detail VARCHAR(2000) NULL,
    first_detected_at_utc DATETIME(3) NOT NULL,
    last_detected_at_utc DATETIME(3) NOT NULL,
    resolved_at_utc DATETIME(3) NULL,
    PRIMARY KEY (issue_id),
    KEY idx_integrity_issue_test (test_id, issue_status, last_detected_at_utc),
    KEY idx_integrity_issue_interval (interval_id, source_type, source_id, source_channel_code),
    CONSTRAINT fk_integrity_issue_test FOREIGN KEY (test_id) REFERENCES t_test (test_id),
    CONSTRAINT fk_integrity_issue_interval FOREIGN KEY (interval_id) REFERENCES t_test_reception_interval (interval_id),
    CONSTRAINT fk_integrity_issue_binding FOREIGN KEY (binding_id) REFERENCES t_test_channel_binding (binding_id),
    CONSTRAINT chk_integrity_issue_source CHECK (source_type IN ('CABINET_CHANNEL', 'COLLECTOR_CHANNEL')),
    CONSTRAINT chk_integrity_issue_status CHECK (issue_status IN ('OPEN', 'RESOLVED')),
    CONSTRAINT chk_integrity_issue_range CHECK (missing_to_sequence IS NULL OR missing_to_sequence >= missing_from_sequence)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='测试接收区间完整性问题';

 ruoyi-admin/src/test/java/com/ruoyi/config/Ticket14MigrationContractTest.java

New file
@@ -0,0 +1,30 @@
package com.ruoyi.config;

import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import org.junit.jupiter.api.Test;

/** Ticket14 迁移契约：完整性事实追加存储且不改写历史迁移。 */
class Ticket14MigrationContractTest
{
    @Test
    void addsIntervalIntegrityFactsAndCollectorIntervalOwnership()
    {
        String sql;
        try (InputStream stream = getClass().getClassLoader().getResourceAsStream(
                "db/migration/V22__interval_integrity_and_completion.sql"))
        {
            if (stream == null) throw new AssertionError("缺少 Ticket14 迁移");
            sql = new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (Exception exception) { throw new AssertionError(exception); }
        assertTrue(sql.contains("t_test_integrity_issue"));
        assertTrue(sql.contains("reception_interval_id"));
        assertTrue(sql.contains("SEQUENCE_GAP") || sql.contains("issue_type"));
        assertTrue(sql.contains("issue_status"));
        assertFalse(sql.matches("(?s).*\\bDROP\\s+(TABLE|COLUMN)\\b.*"));
    }
}

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/IntervalIntegrityCheckerTest.java

New file
@@ -0,0 +1,64 @@
package com.ruoyi.system.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;

import java.util.Arrays;
import java.util.List;
import org.junit.jupiter.api.Test;

/** 区间完整性序号检查的外部行为测试。 */
class IntervalIntegrityCheckerTest
{
    @Test
    void ignoresDuplicatesAndOutOfOrderSamples()
    {
        assertEquals(List.of(new IntervalIntegrityChecker.Gap(3, 3), new IntervalIntegrityChecker.Gap(5, 5)),
                IntervalIntegrityChecker.gaps(List.of(4L, 1L, 2L, 4L, 6L)));
    }

    @Test
    void treatsEachReceptionIntervalIndependently()
    {
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(List.of(1L, 2L)));
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(List.of(8L, 9L)));
    }

    @Test
    void reportsMultipleOpenGaps()
    {
        assertEquals(List.of(new IntervalIntegrityChecker.Gap(3, 4), new IntervalIntegrityChecker.Gap(7, 8)),
                IntervalIntegrityChecker.gaps(List.of(1L, 2L, 5L, 6L, 9L)));
    }

    @Test
    void doesNotTreatPauseOrHandoverAsAnIntraIntervalGap()
    {
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(List.of(10L, 11L)));
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(List.of(20L, 21L)));
    }

    @Test
    void emptyIntervalHasNoSyntheticSequenceGap()
    {
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(List.of()));
    }

    @Test
    void filtersInvalidSequenceValuesWithoutInventingGaps()
    {
        assertEquals(List.of(), IntervalIntegrityChecker.gaps(Arrays.asList(null, 0L, -1L, 5L)));
    }

    @Test
    void remainsStableAcrossReplayPermutationsAndDuplicates()
    {
        List<IntervalIntegrityChecker.Gap> expected = List.of(
                new IntervalIntegrityChecker.Gap(3, 3), new IntervalIntegrityChecker.Gap(6, 7));
        List<List<Long>> replays = List.of(
                List.of(8L, 1L, 2L, 4L, 5L, 8L),
                List.of(1L, 8L, 5L, 4L, 2L, 5L),
                List.of(2L, 4L, 1L, 8L, 4L, 5L));
        for (List<Long> replay : replays) assertEquals(expected, IntervalIntegrityChecker.gaps(replay));
        assertEquals(expected, IntervalIntegrityChecker.gaps(replays.get(0)));
    }
}

 ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java

@@ -23,6 +23,15 @@
                .contains("system:test:query"));
    }

    @Test
    void integrityMappingIsAuthenticatedSurface()
    {
        Method integrity = method("integrity", Long.class);
        assertArrayEquals(new String[] { "/{testId}/integrity" }, integrity.getAnnotation(GetMapping.class).value());
        assertTrue(integrity.getAnnotation(org.springframework.security.access.prepost.PreAuthorize.class).value()
                .contains("system:test:query"));
    }

    private Method method(String name, Class<?>... types)
    {
        try { return TTestLiveController.class.getDeclaredMethod(name, types); }

 ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestIntegrityIssue.java

New file
@@ -0,0 +1,61 @@
package com.ruoyi.system.domain;

import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;

/** 测试来源区间完整性问题事实。 */
public class TTestIntegrityIssue
{
    private Long issueId;
    private Long testId;
    private Long intervalId;
    private Long bindingId;
    private String sourceType;
    private Long sourceId;
    private String sourceChannelCode;
    private String issueType;
    private String issueStatus;
    private Long missingFromSequence;
    private Long missingToSequence;
    private Integer retryCount;
    private String detail;
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX", timezone = "UTC")
    private Date firstDetectedAtUtc;
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX", timezone = "UTC")
    private Date lastDetectedAtUtc;
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX", timezone = "UTC")
    private Date resolvedAtUtc;

    public Long getIssueId() { return issueId; }
    public void setIssueId(Long value) { issueId = value; }
    public Long getTestId() { return testId; }
    public void setTestId(Long value) { testId = value; }
    public Long getIntervalId() { return intervalId; }
    public void setIntervalId(Long value) { intervalId = value; }
    public Long getBindingId() { return bindingId; }
    public void setBindingId(Long value) { bindingId = value; }
    public String getSourceType() { return sourceType; }
    public void setSourceType(String value) { sourceType = value; }
    public Long getSourceId() { return sourceId; }
    public void setSourceId(Long value) { sourceId = value; }
    public String getSourceChannelCode() { return sourceChannelCode; }
    public void setSourceChannelCode(String value) { sourceChannelCode = value; }
    public String getIssueType() { return issueType; }
    public void setIssueType(String value) { issueType = value; }
    public String getIssueStatus() { return issueStatus; }
    public void setIssueStatus(String value) { issueStatus = value; }
    public Long getMissingFromSequence() { return missingFromSequence; }
    public void setMissingFromSequence(Long value) { missingFromSequence = value; }
    public Long getMissingToSequence() { return missingToSequence; }
    public void setMissingToSequence(Long value) { missingToSequence = value; }
    public Integer getRetryCount() { return retryCount; }
    public void setRetryCount(Integer value) { retryCount = value; }
    public String getDetail() { return detail; }
    public void setDetail(String value) { detail = value; }
    public Date getFirstDetectedAtUtc() { return firstDetectedAtUtc; }
    public void setFirstDetectedAtUtc(Date value) { firstDetectedAtUtc = value; }
    public Date getLastDetectedAtUtc() { return lastDetectedAtUtc; }
    public void setLastDetectedAtUtc(Date value) { lastDetectedAtUtc = value; }
    public Date getResolvedAtUtc() { return resolvedAtUtc; }
    public void setResolvedAtUtc(Date value) { resolvedAtUtc = value; }
}

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java

@@ -15,6 +15,7 @@
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.TTestBackfillCursor;
import com.ruoyi.system.domain.TTestIntegrityIssue;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.TestBatteryOptionVo;
import com.ruoyi.system.domain.vo.TestChannelOptionVo;
@@ -127,6 +128,7 @@
    int updateTestBackfillWaiting(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc,
            @Param("error") String error);
    int updateTestBackfillCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestNaturallyEnded(@Param("testId") Long testId, @Param("endedAtUtc") Date endedAtUtc);
    String selectSampleContentHash(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
@@ -145,6 +147,7 @@
    Long selectBtsSampleId(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
    int insertCollectorSample(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("intervalId") Long intervalId,
            @Param("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("sequenceId") Long sequenceId, @Param("deviceTimeUtc") Date deviceTimeUtc,
            @Param("deviceTimeRaw") String deviceTimeRaw, @Param("receivedAtUtc") Date receivedAtUtc,
@@ -157,4 +160,27 @@
            @Param("metricCode") String metricCode);
    int insertBackfillConflict(@Param("testId") Long testId, @Param("channelId") Long channelId,
            @Param("detail") String detail);
    List<TTestIntegrityIssue> selectIntegrityIssues(Long testId);
    List<TTestIntegrityIssue> selectTestExceptionFacts(Long testId);
    List<Long> selectBtsSequencesByInterval(Long intervalId);
    List<Long> selectCollectorSequencesByInterval(Long intervalId);
    TTestIntegrityIssue selectOpenIntegrityIssue(@Param("intervalId") Long intervalId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode, @Param("missingFrom") Long missingFrom,
            @Param("missingTo") Long missingTo);
    List<TTestIntegrityIssue> selectOpenIntegrityIssues(@Param("intervalId") Long intervalId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode);
    int insertIntegrityIssue(TTestIntegrityIssue issue);
    int touchIntegrityIssue(@Param("issueId") Long issueId, @Param("retryCount") Integer retryCount,
            @Param("detail") String detail, @Param("detectedAtUtc") Date detectedAtUtc);
    int rewindBackfillCursor(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode, @Param("nextPosition") Long nextPosition);
    int resolveIntegrityIssues(@Param("testId") Long testId, @Param("intervalId") Long intervalId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode, @Param("resolvedAtUtc") Date resolvedAtUtc);
    int resolveIntegrityIssue(@Param("issueId") Long issueId, @Param("resolvedAtUtc") Date resolvedAtUtc);
    int countOpenIntegrityIssues(Long testId);
    int countOpenExceptions(Long testId);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestIntegrityService.java

New file
@@ -0,0 +1,10 @@
package com.ruoyi.system.service;

import java.util.List;
import com.ruoyi.system.domain.TTestIntegrityIssue;

/** 查询测试区间完整性事实。 */
public interface ITestIntegrityService
{
    List<TTestIntegrityIssue> selectIssues(Long testId);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java

@@ -3,6 +3,7 @@
import java.time.Instant;
import java.util.List;
import java.util.Map;
import com.ruoyi.system.domain.TTestIntegrityIssue;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

/** 提供测试控制台的双状态实时事实与鉴权 SSE。 */
@@ -14,7 +15,13 @@
    /** 打开实时 SSE；连接初次发送当前事实，后续只推送最新快照。 */
    SseEmitter stream(Long testId);
    record LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
            List<LiveSource> sources, List<Map<String, Object>> exceptions) { }
            List<LiveSource> sources, List<Map<String, Object>> exceptions,
            List<TTestIntegrityIssue> integrityIssues)
    {
        public LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
                List<LiveSource> sources, List<Map<String, Object>> exceptions)
        { this(testId, testStatus, receptionStatus, updatedAt, sources, exceptions, List.of()); }
    }
    record LiveSource(String sourceType, Long sourceId, String batteryNo, String deviceStatus,
            Map<String, Object> latestSnapshot) { }
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/IntervalIntegrityChecker.java

New file
@@ -0,0 +1,33 @@
package com.ruoyi.system.service.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

/** 按单个接收区间检查历史序号；暂停和区间边界由调用方分组后自然排除。 */
final class IntervalIntegrityChecker
{
    record Gap(long from, long to) { }

    private IntervalIntegrityChecker() { }

    static List<Gap> gaps(Collection<Long> sequenceIds)
    {
        if (sequenceIds == null || sequenceIds.isEmpty()) return List.of();
        Set<Long> unique = new LinkedHashSet<>();
        sequenceIds.stream().filter(value -> value != null && value > 0).forEach(unique::add);
        List<Long> ordered = new ArrayList<>(unique);
        ordered.sort(Comparator.naturalOrder());
        List<Gap> result = new ArrayList<>();
        for (int index = 1; index < ordered.size(); index++)
        {
            long previous = ordered.get(index - 1);
            long current = ordered.get(index);
            if (current > previous + 1) result.add(new Gap(previous + 1, current - 1));
        }
        return List.copyOf(result);
    }
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java

@@ -85,7 +85,13 @@
                .toList());
        boolean abnormal = persisted.stream().anyMatch(item -> "OPEN".equals(item.getIntervalStatus()));
        String testStatus = abnormal && "RUNNING".equals(test.getTestStatus()) ? "RUNNING_ABNORMAL" : test.getTestStatus();
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions);
        List<com.ruoyi.system.domain.TTestIntegrityIssue> integrity = new ArrayList<>();
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedIntegrity = testMapper.selectIntegrityIssues(testId);
        if (persistedIntegrity != null) integrity.addAll(persistedIntegrity);
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedExceptions = testMapper.selectTestExceptionFacts(testId);
        if (persistedExceptions != null) integrity.addAll(persistedExceptions);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions,
                integrity == null ? List.of() : integrity);
    }

    private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean freeze)
@@ -136,6 +142,7 @@
        emitter.send(SseEmitter.event().name("reception-status").data(Map.of("testId", testId, "receptionStatus", state.receptionStatus())));
        emitter.send(SseEmitter.event().name("test-status").data(Map.of("testId", testId, "testStatus", state.testStatus())));
        emitter.send(SseEmitter.event().name("exception").data(state.exceptions()));
        emitter.send(SseEmitter.event().name("integrity").data(state.integrityIssues()));
        emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt())));
    }


 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java

@@ -289,7 +289,7 @@
        testMapper.deleteChannelLeasesByTestId(testId);
        testMapper.deleteCollectorChannelLeasesByTestId(testId);
        testMapper.insertTestAudit(testId, "TEST_STOPPED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("testStatus", "COMPLETED", "receptionStatus", "ENDED",
                JSON.toJSONString(Map.of("testStatus", "COMPLETION_PENDING", "receptionStatus", "ENDED",
                        "polling", "STOPPED", "closedIntervals", closedIntervals, "release", "IMMEDIATE")));
        return 1;
    }

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

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestIntegrityServiceImpl.java

New file
@@ -0,0 +1,33 @@
package com.ruoyi.system.service.impl;

import java.util.List;
import org.springframework.stereotype.Service;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.system.domain.TTestIntegrityIssue;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ITestIntegrityService;

/** 完整性查询只返回持久化问题事实，不把实时快照当作历史结果。 */
@Service
public class TestIntegrityServiceImpl implements ITestIntegrityService
{
    private final TTestMapper mapper;
    private final ITTestService tests;

    public TestIntegrityServiceImpl(TTestMapper mapper, ITTestService tests)
    { this.mapper = mapper; this.tests = tests; }

    @Override
    public List<TTestIntegrityIssue> selectIssues(Long testId)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        tests.selectTestById(testId);
        List<TTestIntegrityIssue> issues = new java.util.ArrayList<>();
        List<TTestIntegrityIssue> integrity = mapper.selectIntegrityIssues(testId);
        if (integrity != null) issues.addAll(integrity);
        List<TTestIntegrityIssue> exceptions = mapper.selectTestExceptionFacts(testId);
        if (exceptions != null) issues.addAll(exceptions);
        return issues;
    }
}

 ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml

@@ -497,6 +497,11 @@
        update t_test set backfill_status = 'COMPLETED', backfill_completed_at_utc = #{updatedAtUtc},
            update_time = #{updatedAtUtc} where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestCompleted">
        update t_test set test_status = 'COMPLETED', backfill_status = 'COMPLETED',
            backfill_completed_at_utc = #{updatedAtUtc}, update_time = #{updatedAtUtc}
        where test_id = #{testId} and reception_status = 'ENDED' and test_status = 'COMPLETION_PENDING'
    </update>
    <update id="updateTestNaturallyEnded">
        update t_test set test_status = 'COMPLETION_PENDING', reception_status = 'ENDED',
            ended_at_utc = #{endedAtUtc}, backfill_status = 'NOT_STARTED', update_time = #{endedAtUtc}
@@ -526,11 +531,88 @@
    </select>
    <insert id="insertCollectorSample">
        insert ignore into t_test_collector_sample
        (test_id, binding_id, collector_id, collector_channel_code, sequence_id, device_time_utc,
        (test_id, binding_id, reception_interval_id, collector_id, collector_channel_code, sequence_id, device_time_utc,
         device_time_raw, server_received_at_utc, metric_code, metric_value, unit, content_sha256)
        values (#{testId}, #{bindingId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
        values (#{testId}, #{bindingId}, #{intervalId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{metricCode}, #{metricValue}, #{unit}, #{contentSha256})
    </insert>
    <resultMap id="TTestIntegrityIssueResultMap" type="com.ruoyi.system.domain.TTestIntegrityIssue">
        <id property="issueId" column="issue_id"/><result property="testId" column="test_id"/>
        <result property="intervalId" column="interval_id"/><result property="bindingId" column="binding_id"/>
        <result property="sourceType" column="source_type"/><result property="sourceId" column="source_id"/>
        <result property="sourceChannelCode" column="source_channel_code"/><result property="issueType" column="issue_type"/>
        <result property="issueStatus" column="issue_status"/><result property="missingFromSequence" column="missing_from_sequence"/>
        <result property="missingToSequence" column="missing_to_sequence"/><result property="retryCount" column="retry_count"/>
        <result property="detail" column="detail"/><result property="firstDetectedAtUtc" column="first_detected_at_utc"/>
        <result property="lastDetectedAtUtc" column="last_detected_at_utc"/><result property="resolvedAtUtc" column="resolved_at_utc"/>
    </resultMap>
    <select id="selectIntegrityIssues" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where test_id = #{testId}
        order by issue_status, last_detected_at_utc desc, issue_id desc
    </select>
    <select id="selectTestExceptionFacts" resultMap="TTestIntegrityIssueResultMap">
        select exception_id issue_id, test_id, null interval_id, null binding_id,
               case when channel_id is null then 'UNKNOWN_SOURCE' else 'CABINET_CHANNEL' end source_type,
               channel_id source_id, null source_channel_code, exception_type issue_type,
               exception_status issue_status, null missing_from_sequence, null missing_to_sequence,
               occurrence_count retry_count, detail, first_occurred_at_utc first_detected_at_utc,
               last_occurred_at_utc last_detected_at_utc, handled_at_utc resolved_at_utc
        from t_test_exception where test_id = #{testId}
        order by exception_status, last_occurred_at_utc desc, exception_id desc
    </select>
    <select id="selectBtsSequencesByInterval" resultType="long">
        select seq_id from t_test_sample where reception_interval_id = #{intervalId} order by seq_id
    </select>
    <select id="selectCollectorSequencesByInterval" resultType="long">
        select distinct sequence_id from t_test_collector_sample where reception_interval_id = #{intervalId} order by sequence_id
    </select>
    <select id="selectOpenIntegrityIssue" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
          and (missing_from_sequence = #{missingFrom} or (missing_from_sequence is null and #{missingFrom} is null))
          and (missing_to_sequence = #{missingTo} or (missing_to_sequence is null and #{missingTo} is null)) limit 1
    </select>
    <select id="selectOpenIntegrityIssues" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
        order by issue_id
    </select>
    <insert id="insertIntegrityIssue" parameterType="com.ruoyi.system.domain.TTestIntegrityIssue" useGeneratedKeys="true" keyProperty="issueId">
        insert into t_test_integrity_issue
        (test_id, interval_id, binding_id, source_type, source_id, source_channel_code, issue_type, issue_status,
         missing_from_sequence, missing_to_sequence, retry_count, detail, first_detected_at_utc, last_detected_at_utc)
        values (#{testId}, #{intervalId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{issueType}, 'OPEN',
                #{missingFromSequence}, #{missingToSequence}, #{retryCount}, #{detail}, #{firstDetectedAtUtc}, #{lastDetectedAtUtc})
    </insert>
    <update id="touchIntegrityIssue">
        update t_test_integrity_issue set retry_count = #{retryCount}, detail = #{detail}, last_detected_at_utc = #{detectedAtUtc}
        where issue_id = #{issueId} and issue_status = 'OPEN'
    </update>
    <update id="rewindBackfillCursor">
        update t_test_backfill_cursor set next_position = least(next_position, #{nextPosition}),
            source_status = 'PENDING', last_error = '区间序号缺口待重试', updated_at_utc = utc_timestamp(3)
        where test_id = #{testId} and binding_id = #{bindingId} and source_type = #{sourceType}
          and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
    </update>
    <update id="resolveIntegrityIssues">
        update t_test_integrity_issue set issue_status = 'RESOLVED', resolved_at_utc = #{resolvedAtUtc}
        where test_id = #{testId} and interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
    </update>
    <update id="resolveIntegrityIssue">
        update t_test_integrity_issue set issue_status = 'RESOLVED', resolved_at_utc = #{resolvedAtUtc}
        where issue_id = #{issueId} and issue_status = 'OPEN'
    </update>
    <select id="countOpenIntegrityIssues" resultType="int">
        select count(1) from t_test_integrity_issue where test_id = #{testId} and issue_status = 'OPEN'
    </select>
    <select id="countOpenExceptions" resultType="int">
        select count(1) from t_test_exception where test_id = #{testId} and exception_status = 'OPEN'
    </select>
    <select id="selectCollectorSampleId" resultType="long">
        select sample_id from t_test_collector_sample
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
@@ -641,7 +723,7 @@
    </update>
    <update id="updateReceptionStopped">
        update t_test
        set test_status = 'COMPLETED', reception_status = 'ENDED', ended_at_utc = #{endedAtUtc},
        set test_status = 'COMPLETION_PENDING', reception_status = 'ENDED', ended_at_utc = #{endedAtUtc},
            backfill_status = 'NOT_STARTED',
            update_by = #{updateBy}, update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED') and del_flag = 0