实现设备自然结束与区间补采恢复

yeg

2026-08-30 9c0fd80309b277420c492180ae9f98f779f0f5c8

实现设备自然结束与区间补采恢复

增加自然结束协调、来源区间补采游标和检测柜采集器历史持久化，补齐暂停边界与重启恢复测试。

 .gitignore

@@ -14,5 +14,6 @@
!/ruoyi-admin/src/main/resources/db/migration/V8__complete_collector_page_contract.sql
!/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
/.mvn-review-repository/
/ruoyi-admin/logs/

 ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java

@@ -15,6 +15,8 @@
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsDeviceProfile;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryBatch;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
import com.ruoyi.device.model.DeviceWorkStatus;

/** 将开发模拟器完整状态写入 Redis，主状态不设置普通 TTL。 */
@@ -25,6 +27,8 @@
    static final String RECENT_REQUESTS_KEY = "dev:bts-simulator:recent-requests:v1";
    static final String HISTORY_KEYS_KEY = "dev:bts-simulator:collector-history-keys:v1";
    static final String HISTORY_KEY_PREFIX = "dev:bts-simulator:collector-history:v1:";
    static final String BTS_HISTORY_KEYS_KEY = "dev:bts-simulator:bts-history-keys:v1";
    static final String BTS_HISTORY_KEY_PREFIX = "dev:bts-simulator:bts-history:v1:";
    private static final long RECENT_REQUESTS_TTL_HOURS = 1;

    private final RedisTemplate<Object, Object> redisTemplate;
@@ -50,24 +54,26 @@
    @Override
    public SimulatorState resetHistoryForDemoStartup(SimulatorState state)
    {
        if (pendingEmbeddedHistory.isEmpty()) return state;
        java.util.Map<String, Integer> embeddedCounts = new java.util.HashMap<>();
        pendingEmbeddedHistory.forEach(append -> embeddedCounts.put(append.collectorCode(), append.records().size()));
        List<CollectorState> collectors = state.collectors().stream().map(collector -> {
            int embeddedCount = embeddedCounts.getOrDefault(collector.deviceCode(), 0);
            if (embeddedCount == 0) return collector;
            long remaining = Math.max(0L, collector.historyRecordCount() - embeddedCount);
            return new CollectorState(collector.collectorId(), collector.deviceCode(), collector.ip(), collector.mac(),
                    collector.name(), collector.manufacturer(), collector.model(), collector.channelCount(),
                    collector.online(), collector.sampleIntervalMs(), collector.nextSequence(), collector.lastSampledAt(),
                    collector.channels(), collector.latestMeasurements().stream()
                            .filter(item -> collector.deviceCode().equals(item.collectorCode()))
                            .toList(), remaining, collector.historyFaults());
        }).toList();
        // 不删除独立历史 Redis key；这里只阻止旧内嵌数据在本次启动时迁移回来。
        pendingEmbeddedHistory = List.of();
        Set<Object> historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY);
        if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys);
        redisTemplate.delete(HISTORY_KEYS_KEY);

        List<com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState> collectors = state.collectors()
                .stream()
                .map(collector -> new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState(
                        collector.collectorId(), collector.deviceCode(), collector.ip(), collector.mac(),
                        collector.name(), collector.manufacturer(), collector.model(), collector.channelCount(),
                        collector.online(), collector.sampleIntervalMs(), 1L,
                        null, collector.channels(), List.of(), 0L,
                        collector.historyFaults()))
                .toList();
        SimulatorState reset = new SimulatorState(state.version(), state.devices(), collectors,
                state.nextDeviceId(), state.nextCollectorId(), state.tombstones());
        redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(reset));
        return reset;
        SimulatorState cleaned = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
                state.nextCollectorId(), state.tombstones());
        save(cleaned);
        return cleaned;
    }

    private SimulatorState restoreState(Object stored)
@@ -176,6 +182,34 @@
    }

    @Override
    public BtsHistoryBatch loadBtsHistoryPage(String historyKey, long offset, int limit)
    {
        if (offset < 0 || limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
            throw new IllegalArgumentException("invalid BTS history page");
        String key = BTS_HISTORY_KEY_PREFIX + historyKey;
        Long size = redisTemplate.opsForList().size(key);
        long total = size == null ? 0 : size;
        long from = Math.min(offset, total);
        int count = (int) Math.min(limit, total - from);
        List<Object> values = count == 0 ? List.of() : redisTemplate.opsForList().range(key, from, from + count - 1);
        List<BtsHistoryRecord> records = values == null ? List.of() : values.stream()
                .map(item -> JSON.parseObject(item.toString(), BtsHistoryRecord.class)).toList();
        long next = from + records.size();
        return new BtsHistoryBatch(offset, next, next < total, total, records);
    }

    @Override
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void appendBtsHistory(String historyKey, BtsHistoryRecord record)
    {
        String key = BTS_HISTORY_KEY_PREFIX + historyKey;
        RedisOperations<Object, Object> operations = redisTemplate;
        operations.opsForList().rightPush(key, JSON.toJSONString(record));
        operations.opsForList().trim(key, -MAX_HISTORY_RECORDS, -1);
        operations.opsForSet().add(BTS_HISTORY_KEYS_KEY, key);
    }

    @Override
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
    {
@@ -218,6 +252,9 @@
        Set<Object> historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY);
        if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys);
        redisTemplate.delete(HISTORY_KEYS_KEY);
        Set<Object> btsHistoryKeys = redisTemplate.opsForSet().members(BTS_HISTORY_KEYS_KEY);
        if (btsHistoryKeys != null && !btsHistoryKeys.isEmpty()) redisTemplate.delete(btsHistoryKeys);
        redisTemplate.delete(BTS_HISTORY_KEYS_KEY);
        redisTemplate.delete(RECENT_REQUESTS_KEY);
        save(defaultState);
    }

 ruoyi-admin/src/main/java/com/ruoyi/config/TestBackfillScheduler.java

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

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.ruoyi.system.service.ITestBackfillService;

/** 以较低优先级执行历史补采；实时设备观测由既有调度器优先完成。 */
@Component
public class TestBackfillScheduler
{
    private final ITestBackfillService backfill;

    public TestBackfillScheduler(ITestBackfillService backfill)
    {
        this.backfill = backfill;
    }

    @Scheduled(fixedDelay = 3000)
    public void reconcileAndBackfill()
    {
        backfill.reconcileNaturalEnd();
        backfill.resumePendingBackfill();
    }
}

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

New file
@@ -0,0 +1,58 @@
-- Ticket13：自然结束与区间补采的来源游标、采集器原始历史和恢复元数据。
ALTER TABLE t_test
    ADD COLUMN backfill_status VARCHAR(32) NOT NULL DEFAULT 'NOT_STARTED'
        COMMENT '区间补采状态：NOT_STARTED、RUNNING、WAITING、FAILED' AFTER runtime_attach,
    ADD COLUMN backfill_started_at_utc DATETIME(3) NULL AFTER backfill_status,
    ADD COLUMN backfill_completed_at_utc DATETIME(3) NULL AFTER backfill_started_at_utc,
    ADD KEY idx_test_backfill_status (reception_status, test_status, backfill_status);

CREATE TABLE t_test_backfill_cursor (
    cursor_id BIGINT NOT NULL AUTO_INCREMENT,
    test_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,
    protocol_test_id VARCHAR(64) NULL,
    next_position BIGINT NOT NULL DEFAULT 1,
    source_status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
    protocol_metadata JSON NULL,
    last_error VARCHAR(1000) NULL,
    updated_at_utc DATETIME(3) NOT NULL,
    PRIMARY KEY (cursor_id),
    UNIQUE KEY uk_test_backfill_source (test_id, binding_id, source_type, source_id, source_channel_code),
    KEY idx_test_backfill_work (source_status, updated_at_utc),
    CONSTRAINT fk_backfill_cursor_test FOREIGN KEY (test_id) REFERENCES t_test (test_id),
    CONSTRAINT fk_backfill_cursor_binding FOREIGN KEY (binding_id) REFERENCES t_test_channel_binding (binding_id),
    CONSTRAINT chk_backfill_cursor_source CHECK (source_type IN ('CABINET_CHANNEL', 'COLLECTOR_CHANNEL')),
    CONSTRAINT chk_backfill_cursor_status CHECK (source_status IN ('PENDING', 'RUNNING', 'COMPLETED', 'FAILED')),
    CONSTRAINT chk_backfill_cursor_position CHECK (next_position > 0)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='测试来源区间补采游标';

CREATE TABLE t_test_collector_sample (
    sample_id BIGINT NOT NULL AUTO_INCREMENT,
    test_id BIGINT NOT NULL,
    binding_id BIGINT NOT NULL,
    collector_id BIGINT NOT NULL,
    collector_channel_code VARCHAR(64) NOT NULL,
    sequence_id BIGINT NOT NULL,
    device_time_utc DATETIME(3) NOT NULL,
    device_time_raw VARCHAR(64) NOT NULL,
    server_received_at_utc DATETIME(3) NOT NULL,
    metric_code VARCHAR(191) NOT NULL,
    metric_value DECIMAL(30,12) NULL,
    unit VARCHAR(32) NOT NULL,
    content_sha256 BINARY(32) NOT NULL,
    PRIMARY KEY (sample_id),
    UNIQUE KEY uk_collector_history_sample (collector_id, collector_channel_code, sequence_id, metric_code),
    KEY idx_collector_sample_test_time (test_id, binding_id, device_time_utc),
    CONSTRAINT fk_collector_sample_test FOREIGN KEY (test_id) REFERENCES t_test (test_id),
    CONSTRAINT fk_collector_sample_binding FOREIGN KEY (binding_id) REFERENCES t_test_channel_binding (binding_id),
    CONSTRAINT fk_collector_sample_collector FOREIGN KEY (collector_id) REFERENCES t_collector (collector_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='采集器原始历史测量';

ALTER TABLE t_test_sample
    ADD COLUMN reception_interval_id BIGINT NULL AFTER binding_id,
    ADD KEY idx_sample_interval (reception_interval_id),
    ADD CONSTRAINT fk_sample_reception_interval FOREIGN KEY (reception_interval_id)
        REFERENCES t_test_reception_interval (interval_id);

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

New file
@@ -0,0 +1,28 @@
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;

class Ticket13MigrationContractTest
{
    @Test
    void addsRecoverableSourceCursorsAndCollectorHistoryWithoutChangingEarlierTables()
    {
        String sql;
        try (InputStream stream = getClass().getClassLoader().getResourceAsStream(
                "db/migration/V21__natural_end_interval_backfill.sql"))
        {
            if (stream == null) throw new AssertionError("缺少 Ticket13 迁移");
            sql = new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (Exception exception) { throw new AssertionError(exception); }
        assertTrue(sql.contains("t_test_backfill_cursor"));
        assertTrue(sql.contains("t_test_collector_sample"));
        assertTrue(sql.contains("protocol_metadata JSON"));
        assertTrue(sql.contains("reception_interval_id"));
        assertFalse(sql.matches("(?s).*\\bDROP\\s+(TABLE|COLUMN)\\b.*"));
    }
}

 ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java

@@ -4,6 +4,8 @@
import java.io.StringReader;
import java.math.BigDecimal;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import javax.xml.stream.XMLInputFactory;
@@ -22,6 +24,7 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsDeviceProfile;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
@@ -50,6 +53,9 @@
    private SimulatorState state;
    private boolean running = true;
    private long observationSequence;
    private final java.util.Map<String, java.util.List<SimBtsHistory>> btsHistory = new java.util.LinkedHashMap<>();
    private static final DateTimeFormatter BTS_HISTORY_TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd HH.mm.ss")
            .withZone(ZoneId.of("Asia/Hong_Kong"));

    /** 启动仅监听 IPv4 本机回环地址的开发模拟器；传入 0 时由系统分配测试端口。 */
    public BtsDevelopmentSimulator(int port) throws IOException
@@ -62,7 +68,9 @@
    {
        this.stateStore = stateStore;
        state = stateStore.loadOrInitialize(defaultState());
        // 仅清理 Ticket13 之前嵌在主状态中的旧演示历史；独立历史由存储实现保留并继续补采。
        state = stateStore.resetHistoryForDemoStartup(state);
        restoreBtsHistory();
        validateRestoredState();
        stateStore.completeValidatedRestore(state);
        simulator = ScriptedBtsTcpSimulator.responding(port, this::respond);
@@ -341,6 +349,8 @@
        {
            case "getdevinfo" -> deviceInfoResponse();
            case "inquire" -> observationResponse(request);
            case "inquiredf" -> historyAvailabilityResponse(request);
            case "download" -> historyResponse(request);
            default -> "<bts version=\"1.0\"><cmd>error</cmd></bts>";
        };
    }
@@ -445,6 +455,16 @@
                    profile.voltageMaxV().doubleValue());
            double capacity = working ? Math.max(0d, tick * 0.001d * current) : 0d;
            double energy = capacity * voltage;
            if (working)
            {
                String key = historyKey(profile.devtype(), observed.deviceId(), profile.subDeviceId(), channel.channelId());
                java.util.List<SimBtsHistory> history = btsHistory.computeIfAbsent(key, ignored -> new ArrayList<>());
                history.add(new SimBtsHistory(history.size() + 1L, Instant.now(), current, voltage, capacity, energy));
                SimBtsHistory latest = history.get(history.size() - 1);
                stateStore.appendBtsHistory(key, new BtsHistoryRecord(latest.sequence(), latest.at(), latest.current(),
                        latest.voltage(), latest.capacity(), latest.energy()));
                if (history.size() > BtsSimulatorStateStore.MAX_HISTORY_RECORDS) history.remove(0);
            }
            response.append("<inquire dev=\"").append(profile.devtype()).append('-').append(observed.deviceId())
                    .append('-').append(profile.subDeviceId()).append('-').append(channel.channelId())
                    .append("-0\" cycle_id=\"1\" step_id=\"1\" step_type=\"cc\" workstatus=\"")
@@ -458,6 +478,88 @@
        return response.append("</list></bts>").toString();
    }

    private synchronized String historyAvailabilityResponse(String request)
    {
        java.util.regex.Matcher matcher = java.util.regex.Pattern.compile("<chl\\s+([^>]*)/?>").matcher(request);
        StringBuilder response = new StringBuilder("<bts version=\"1.0\"><cmd>inquiredf_resp</cmd><list>");
        while (matcher.find())
        {
            String attrs = matcher.group(1);
            int devtype = Integer.parseInt(attribute(attrs, "devtype"));
            int devid = Integer.parseInt(attribute(attrs, "devid"));
            int subdevid = Integer.parseInt(attribute(attrs, "subdevid"));
            int chlid = Integer.parseInt(attribute(attrs, "chlid"));
            int count = btsHistory.getOrDefault(historyKey(devtype, devid, subdevid, chlid), List.of()).size();
            response.append("<chl devtype=\"").append(devtype).append("\" devid=\"").append(devid)
                    .append("\" subdevid=\"").append(subdevid).append("\" chlid=\"").append(chlid)
                    .append("\" testid=\"SIM-").append(devid).append('-').append(chlid).append("\" count=\"")
                    .append(count).append("\">true</chl>");
        }
        return response.append("</list></bts>").toString();
    }

    private synchronized String historyResponse(String request)
    {
        int devtype = Integer.parseInt(attribute(request, "devtype"));
        int devid = Integer.parseInt(attribute(request, "devid"));
        int subdevid = Integer.parseInt(attribute(request, "subdevid"));
        int chlid = Integer.parseInt(attribute(request, "chlid"));
        String testId = attribute(request, "testid");
        long start = Long.parseLong(attribute(request, "startpos"));
        int count = Integer.parseInt(attribute(request, "count"));
        List<SimBtsHistory> history = btsHistory.getOrDefault(historyKey(devtype, devid, subdevid, chlid), List.of());
        int from = (int) Math.min(Math.max(0L, start - 1L), history.size());
        int to = Math.min(history.size(), from + Math.max(1, count));
        StringBuilder response = new StringBuilder("<bts version=\"1.0\"><cmd>download_resp</cmd><download devtype=\"")
                .append(devtype).append("\" devid=\"").append(devid).append("\" subdevid=\"").append(subdevid)
                .append("\" chlid=\"").append(chlid).append("\" testid=\"").append(testId)
                .append("\"><list>");
        for (SimBtsHistory item : history.subList(from, to))
            response.append("<data seqid=\"").append(item.sequence()).append("\" stepid=\"1\" cycleid=\"1\" steptype=\"cc\" atime=\"")
                    .append(BTS_HISTORY_TIME.format(item.at())).append("\" testtime=\"1000\" volt=\"")
                    .append(decimal(item.voltage())).append("\" curr=\"").append(decimal(item.current()))
                    .append("\" cap=\"").append(decimal(item.capacity())).append("\" eng=\"").append(decimal(item.energy())).append("\"/>");
        return response.append("</list></download></bts>").toString();
    }

    private String attribute(String xml, String name)
    {
        java.util.regex.Matcher matcher = java.util.regex.Pattern.compile(name + "=\\\"([^\\\"]+)\\\"").matcher(xml);
        if (!matcher.find()) throw new IllegalArgumentException("Missing simulator attribute: " + name);
        return matcher.group(1);
    }

    private String historyKey(int devtype, int devid, int subdevid, int chlid)
    {
        return devtype + ":" + devid + ":" + subdevid + ":" + chlid;
    }

    private void restoreBtsHistory()
    {
        for (DeviceState device : state.devices())
        {
            BtsDeviceProfile profile = profileFor(device);
            for (ChannelState channel : device.channels())
            {
                String key = historyKey(profile.devtype(), device.deviceId(), profile.subDeviceId(), channel.channelId());
                ArrayList<SimBtsHistory> restored = new ArrayList<>();
                long offset = 0;
                while (true)
                {
                    var page = stateStore.loadBtsHistoryPage(key, offset, BtsSimulatorStateStore.MAX_HISTORY_BATCH_SIZE);
                    restored.addAll(page.records().stream().map(item -> new SimBtsHistory(item.sequenceId(),
                            item.deviceTime(), item.currentA(), item.voltageV(), item.capacityAh(), item.energyWh())).toList());
                    if (!page.hasMore()) break;
                    if (page.nextOffset() <= offset) throw new IllegalStateException("Invalid BTS history page");
                    offset = page.nextOffset();
                }
                if (!restored.isEmpty()) btsHistory.put(key, restored);
            }
        }
    }

    private record SimBtsHistory(long sequence, Instant at, double current, double voltage, double capacity, double energy) { }

    private java.util.Set<String> requestedChannels(String request)
    {
        java.util.Set<String> channels = new java.util.HashSet<>();

 ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java

@@ -43,6 +43,15 @@
        return new CollectorHistoryBatch(offset, 0, false, 0, List.of());
    }

    /** 读取检测柜模拟历史；正式适配器不依赖该演示扩展。 */
    default BtsHistoryBatch loadBtsHistoryPage(String historyKey, long offset, int limit)
    {
        return new BtsHistoryBatch(offset, 0, false, 0, List.of());
    }

    /** 追加检测柜模拟历史，供重启续采合同使用。 */
    default void appendBtsHistory(String historyKey, BtsHistoryRecord record) { }

    /** 原子保存删除采集器后的主状态，并删除该采集器的独立历史。 */
    default void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
    {
@@ -137,6 +146,15 @@
        public CollectorHistoryBatch { records = List.copyOf(records); }
    }

    record BtsHistoryRecord(long sequenceId, Instant deviceTime, double currentA, double voltageV,
            double capacityAh, double energyWh) { }

    record BtsHistoryBatch(long offset, long nextOffset, boolean hasMore, long total,
            List<BtsHistoryRecord> records)
    {
        public BtsHistoryBatch { records = List.copyOf(records); }
    }

    record CollectorHistoryFaults(List<Long> missingSequences, Long duplicateSequence,
            Long conflictingSequence, boolean outOfOrder)
    {

 ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java

@@ -7,6 +7,8 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryBatch;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryBatch;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;

@@ -16,6 +18,7 @@
    private SimulatorState state;
    private List<RecentRequest> recentRequests = List.of();
    private final Map<String, ArrayList<CollectorMeasurementRecord>> collectorHistory = new LinkedHashMap<>();
    private final Map<String, ArrayList<BtsHistoryRecord>> btsHistory = new LinkedHashMap<>();

    public InMemoryBtsSimulatorStateStore() { }

@@ -85,6 +88,25 @@
    }

    @Override
    public synchronized BtsHistoryBatch loadBtsHistoryPage(String historyKey, long offset, int limit)
    {
        List<BtsHistoryRecord> history = btsHistory.getOrDefault(historyKey, new ArrayList<>());
        int from = (int) Math.min(offset, history.size());
        int count = Math.min(limit, history.size() - from);
        List<BtsHistoryRecord> records = new ArrayList<>(history.subList(from, from + count));
        long next = from + records.size();
        return new BtsHistoryBatch(offset, next, next < history.size(), history.size(), records);
    }

    @Override
    public synchronized void appendBtsHistory(String historyKey, BtsHistoryRecord record)
    {
        ArrayList<BtsHistoryRecord> history = btsHistory.computeIfAbsent(historyKey, key -> new ArrayList<>());
        history.add(record);
        if (history.size() > MAX_HISTORY_RECORDS) history.remove(0);
    }

    @Override
    public synchronized List<RecentRequest> loadRecentRequests()
    {
        return recentRequests;
@@ -102,5 +124,6 @@
        state = defaultState;
        recentRequests = List.of();
        collectorHistory.clear();
        btsHistory.clear();
    }
}

 ruoyi-device/src/test/java/com/ruoyi/device/bts/BtsDevelopmentSimulatorHistoryContractTest.java

New file
@@ -0,0 +1,58 @@
package com.ruoyi.device.bts;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import java.time.Duration;
import java.time.ZoneId;
import org.junit.jupiter.api.Test;
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;

/** Ticket13：开发检测柜历史合同与实时快照保持两条独立路径。 */
class BtsDevelopmentSimulatorHistoryContractTest
{
    @Test
    void workingObservationProducesPagedHistoryWithoutControlCommand() throws Exception
    {
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, new InMemoryBtsSimulatorStateStore()))
        {
            simulator.updateChannel(1, 1, true, DeviceWorkStatus.WORKING);
            try (BtsDeviceAdapter adapter = new BtsDeviceAdapter("dev-sim",
                    new BtsTcpTransport("127.0.0.1", simulator.port(), Duration.ofSeconds(2), Duration.ofSeconds(2)),
                    new BtsXmlCodec("dev-sim", ZoneId.of("Asia/Hong_Kong"))))
            {
                var channel = adapter.discoverChannels().stream().filter(item -> item.identity().chlid() == 1).findFirst().orElseThrow();
                adapter.inquire(java.util.Set.of(channel.identity()));
                var availability = adapter.inquireHistory(java.util.Set.of(channel.identity())).get(0);
                assertFalse(availability.btsTestId().isBlank());
                var page = adapter.downloadHistory(channel.identity(), availability.btsTestId(), 1, 1);
                assertEquals(1, page.samples().size());
                assertEquals(1, page.samples().get(0).sequenceId());
            }
        }
    }

    @Test
    void historySurvivesSimulatorRestartForResume() throws Exception
    {
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, store))
        {
            simulator.updateChannel(1, 1, true, DeviceWorkStatus.WORKING);
            try (BtsDeviceAdapter adapter = new BtsDeviceAdapter("dev-sim",
                    new BtsTcpTransport("127.0.0.1", simulator.port(), Duration.ofSeconds(2), Duration.ofSeconds(2)),
                    new BtsXmlCodec("dev-sim", ZoneId.of("Asia/Hong_Kong"))))
            {
                adapter.inquire(java.util.Set.of(adapter.discoverChannels().get(0).identity()));
            }
        }
        try (BtsDevelopmentSimulator restarted = new BtsDevelopmentSimulator(0, store);
                BtsDeviceAdapter adapter = new BtsDeviceAdapter("dev-sim",
                        new BtsTcpTransport("127.0.0.1", restarted.port(), Duration.ofSeconds(2), Duration.ofSeconds(2)),
                        new BtsXmlCodec("dev-sim", ZoneId.of("Asia/Hong_Kong"))))
        {
            var channel = adapter.discoverChannels().get(0);
            assertEquals(1, adapter.inquireHistory(java.util.Set.of(channel.identity())).get(0).availableRecordCount());
        }
    }
}

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

@@ -39,6 +39,9 @@
    private Date endedAtUtc;
    /** 是否从已运行设备的后续观测边界接入。 */
    private Boolean runtimeAttach;
    private String backfillStatus;
    private Date backfillStartedAtUtc;
    private Date backfillCompletedAtUtc;
    private String delFlag;
    private String batteryNo;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
@@ -90,6 +93,12 @@
    public void setEndedAtUtc(Date value) { endedAtUtc = value; }
    public Boolean getRuntimeAttach() { return runtimeAttach; }
    public void setRuntimeAttach(Boolean value) { runtimeAttach = value; }
    public String getBackfillStatus() { return backfillStatus; }
    public void setBackfillStatus(String value) { backfillStatus = value; }
    public Date getBackfillStartedAtUtc() { return backfillStartedAtUtc; }
    public void setBackfillStartedAtUtc(Date value) { backfillStartedAtUtc = value; }
    public Date getBackfillCompletedAtUtc() { return backfillCompletedAtUtc; }
    public void setBackfillCompletedAtUtc(Date value) { backfillCompletedAtUtc = value; }
    public String getDelFlag() { return delFlag; }
    public void setDelFlag(String value) { delFlag = value; }
    public String getBatteryNo() { return batteryNo; }

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

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

import java.util.Date;

/** 测试来源历史补采的可恢复游标。 */
public class TTestBackfillCursor
{
    private Long cursorId;
    private Long testId;
    private Long bindingId;
    private String sourceType;
    private Long sourceId;
    private String sourceChannelCode;
    private String protocolTestId;
    private Long nextPosition;
    private String sourceStatus;
    private String protocolMetadata;
    private String lastError;
    private Date updatedAtUtc;

    public Long getCursorId() { return cursorId; }
    public void setCursorId(Long value) { cursorId = value; }
    public Long getTestId() { return testId; }
    public void setTestId(Long value) { testId = 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 getProtocolTestId() { return protocolTestId; }
    public void setProtocolTestId(String value) { protocolTestId = value; }
    public Long getNextPosition() { return nextPosition; }
    public void setNextPosition(Long value) { nextPosition = value; }
    public String getSourceStatus() { return sourceStatus; }
    public void setSourceStatus(String value) { sourceStatus = value; }
    public String getProtocolMetadata() { return protocolMetadata; }
    public void setProtocolMetadata(String value) { protocolMetadata = value; }
    public String getLastError() { return lastError; }
    public void setLastError(String value) { lastError = value; }
    public Date getUpdatedAtUtc() { return updatedAtUtc; }
    public void setUpdatedAtUtc(Date value) { updatedAtUtc = value; }
}

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

@@ -14,6 +14,7 @@
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.TTestBackfillCursor;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.TestBatteryOptionVo;
import com.ruoyi.system.domain.vo.TestChannelOptionVo;
@@ -59,6 +60,7 @@
    int countRegisteredChannel(Long channelId);
    int countCollectorEnabled(Long collectorId);
    Integer selectCollectorChannelCount(Long collectorId);
    String selectCollectorCode(Long collectorId);
    List<TestUserOptionVo> selectEnabledUserOptions();
    List<TestTypeOptionVo> selectEnabledTestTypeOptions();
    List<TestBatteryOptionVo> selectEnabledBatteryOptions();
@@ -111,4 +113,48 @@
    int closeCollectorExceptionInterval(@Param("intervalId") Long intervalId,
            @Param("recoveredAtUtc") Date recoveredAtUtc, @Param("missingToUtc") Date missingToUtc);
    List<TTestCollectorExceptionInterval> selectCollectorExceptionIntervals(Long testId);

    /** 查询需要自然结束或继续补采的测试。 */
    List<TTest> selectTestsForNaturalEnd();
    List<TTest> selectTestsForBackfill();
    /** 按稳定来源身份锁定补采游标，避免应用重启重复推进。 */
    TTestBackfillCursor selectBackfillCursorForUpdate(@Param("testId") Long testId,
            @Param("bindingId") Long bindingId, @Param("sourceType") String sourceType,
            @Param("sourceId") Long sourceId, @Param("sourceChannelCode") String sourceChannelCode);
    int insertBackfillCursor(TTestBackfillCursor cursor);
    int updateBackfillCursor(TTestBackfillCursor cursor);
    int updateTestBackfillStarted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    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 updateTestNaturallyEnded(@Param("testId") Long testId, @Param("endedAtUtc") Date endedAtUtc);
    String selectSampleContentHash(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
    int insertBtsSample(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("channelId") Long channelId, @Param("intervalId") Long intervalId,
            @Param("btsTestId") String btsTestId, @Param("sequenceId") Long sequenceId,
            @Param("deviceTimeUtc") Date deviceTimeUtc, @Param("deviceTimeRaw") String deviceTimeRaw,
            @Param("receivedAtUtc") Date receivedAtUtc, @Param("cycleNo") Integer cycleNo,
            @Param("stepNo") Integer stepNo, @Param("stepType") String stepType,
            @Param("testTimeMs") Long testTimeMs, @Param("voltageV") java.math.BigDecimal voltageV,
            @Param("currentA") java.math.BigDecimal currentA, @Param("capacityAh") java.math.BigDecimal capacityAh,
            @Param("energyWh") java.math.BigDecimal energyWh, @Param("contentSha256") byte[] contentSha256);
    int insertBtsMeasurement(@Param("sampleId") Long sampleId, @Param("metricCode") String metricCode,
            @Param("metricValue") java.math.BigDecimal metricValue, @Param("unit") String unit,
            @Param("sourceType") String sourceType);
    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("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("sequenceId") Long sequenceId, @Param("deviceTimeUtc") Date deviceTimeUtc,
            @Param("deviceTimeRaw") String deviceTimeRaw, @Param("receivedAtUtc") Date receivedAtUtc,
            @Param("metricCode") String metricCode, @Param("metricValue") java.math.BigDecimal metricValue,
            @Param("unit") String unit, @Param("contentSha256") byte[] contentSha256);
    Long selectCollectorSampleId(@Param("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("sequenceId") Long sequenceId, @Param("metricCode") String metricCode);
    String selectCollectorSampleContentHash(@Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode, @Param("sequenceId") Long sequenceId,
            @Param("metricCode") String metricCode);
    int insertBackfillConflict(@Param("testId") Long testId, @Param("channelId") Long channelId,
            @Param("detail") String detail);
}

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

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

/** 设备自然结束后的区间历史补采与恢复协调接口。 */
public interface ITestBackfillService
{
    /** 对接收中的测试检查检测柜自然结束并关闭其接收边界。 */
    void reconcileNaturalEnd();

    /** 从持久化游标继续所有处于完成确认中的测试补采。 */
    void resumePendingBackfill();

    /** 立即结束指定测试的接收并进入完成确认中，供自然结束协调器使用。 */
    void endNaturally(Long testId, String reason);
}

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

@@ -236,6 +236,8 @@
        Date actionAt = new Date();
        if (testMapper.updateReceptionPaused(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        // 暂停边界不属于本次测试的确认接收区间，继续时由新的区间重新开始计时。
        testMapper.closeOpenReceptionIntervals(testId, actionAt, "RECEPTION_PAUSED");
        testMapper.insertTestAudit(testId, "RECEPTION_PAUSED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "polling", "STOPPED")));
        return 1;
@@ -254,6 +256,15 @@
        Date actionAt = new Date();
        if (testMapper.updateReceptionResumed(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        for (TTestBattery battery : current.getBatteries())
        {
            TTestChannelBinding binding = battery.getBinding();
            testMapper.insertReceptionInterval(testId, binding.getBindingId(), "CABINET_CHANNEL",
                    binding.getChannelId(), null, actionAt, "RECEPTION_RESUMED");
            for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
                testMapper.insertReceptionInterval(testId, binding.getBindingId(), "COLLECTOR_CHANNEL",
                        battery.getCollectorId(), mapping.getChannelCode(), actionAt, "RECEPTION_RESUMED");
        }
        testMapper.insertTestAudit(testId, "RECEPTION_RESUMED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "polling", "RESUMED")));
        return 1;

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

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

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Instant;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.Objects;
import com.alibaba.fastjson2.JSON;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.transaction.annotation.Transactional;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.device.api.DeviceAdapter;
import com.ruoyi.device.model.ChannelIdentity;
import com.ruoyi.device.model.HistoryAvailability;
import com.ruoyi.device.model.HistoryPage;
import com.ruoyi.device.model.HistorySample;
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBackfillCursor;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestChannelBinding;
import com.ruoyi.system.domain.TTestCollectorChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestBackfillService;
import com.ruoyi.system.service.ITTestService;

/** 以测试动作边界筛选来源历史，并以数据库游标保证补采可恢复。 */
@Service
public class TestBackfillServiceImpl implements ITestBackfillService
{
    private static final int PAGE_SIZE = 1000;
    private final TTestMapper mapper;
    private final DeviceAdapterRegistry registry;
    private final BtsSimulatorStateStore collectorHistoryStore;
    private final ITTestService tests;
    private final TransactionTemplate transactionTemplate;

    public TestBackfillServiceImpl(TTestMapper mapper, DeviceAdapterRegistry registry,
            BtsSimulatorStateStore collectorHistoryStore, ITTestService tests,
            PlatformTransactionManager transactionManager)
    {
        this.mapper = mapper;
        this.registry = registry;
        this.collectorHistoryStore = collectorHistoryStore;
        this.tests = tests;
        this.transactionTemplate = new TransactionTemplate(transactionManager);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void reconcileNaturalEnd()
    {
        List<TTest> summaries = mapper.selectTestsForNaturalEnd();
        if (summaries == null) return;
        for (TTest summary : summaries)
        {
            TTest test = tests.selectTestById(summary.getTestId());
            if (test == null || test.getBatteries() == null || test.getBatteries().isEmpty()) continue;
            boolean allFinished = true;
            for (TTestBattery battery : test.getBatteries())
            {
                TTestChannelBinding binding = battery.getBinding();
                if (binding == null) { allFinished = false; break; }
                var channel = mapper.selectReceptionChannelForUpdate(binding.getChannelId());
                if (channel == null || !"FINISHED".equalsIgnoreCase(channel.getWorkStatus()))
                { allFinished = false; break; }
            }
            if (allFinished) endNaturally(test.getTestId(), "NATURAL_FINISH");
        }
    }

    @Override
    public void resumePendingBackfill()
    {
        List<TTest> pending = mapper.selectTestsForBackfill();
        if (pending == null) return;
        for (TTest test : pending)
        {
            try { transactionTemplate.executeWithoutResult(status -> backfill(test)); }
            catch (RuntimeException exception)
            {
                mapper.updateTestBackfillWaiting(test.getTestId(), new Date(), exception.getMessage());
            }
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void endNaturally(Long testId, String reason)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        Date endedAt = new Date();
        if (mapper.updateTestNaturallyEnded(testId, endedAt) != 1) return;
        mapper.closeOpenReceptionIntervals(testId, endedAt, reason == null ? "NATURAL_FINISH" : reason);
        mapper.deleteChannelLeasesByTestId(testId);
        mapper.deleteCollectorChannelLeasesByTestId(testId);
        mapper.insertTestAudit(testId, "RECEPTION_NATURALLY_ENDED", null, "system", null,
                JSON.toJSONString(java.util.Map.of("testStatus", "COMPLETION_PENDING",
                        "receptionStatus", "ENDED", "reason", reason == null ? "NATURAL_FINISH" : reason)));
    }

    @Transactional(rollbackFor = Exception.class)
    protected void backfill(TTest test)
    {
        if (mapper.updateTestBackfillStarted(test.getTestId(), new Date()) == 0) return;
        // 来源按检测柜、采集器分别提交；任何一页失败都不推进该来源游标。
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            TTestChannelBinding binding = battery.getBinding();
            if (binding != null) backfillCabinet(test, battery, binding);
            for (TTestCollectorChannel mapping : battery.getCollectorChannels() == null
                    ? List.<TTestCollectorChannel>of() : battery.getCollectorChannels())
                backfillCollector(test, battery, mapping);
        }
        mapper.updateTestBackfillCompleted(test.getTestId(), new Date());
    }

    private void backfillCabinet(TTest test, TTestBattery battery, TTestChannelBinding binding)
    {
        String adapterCode = binding.getAdapterCode();
        if (adapterCode == null || adapterCode.isBlank()) throw new ServiceException("检测柜适配器身份缺失");
        ChannelIdentity identity = new ChannelIdentity(adapterCode, binding.getDevtype(), binding.getDevid(),
                binding.getSubdevid(), binding.getChlid());
        List<TTestReceptionInterval> intervals = intervals(test.getTestId(), binding.getBindingId(), "CABINET_CHANNEL",
                binding.getChannelId(), null);
        if (intervals.isEmpty()) return;
        HistoryAvailability availability = registry.execute(adapterCode,
                adapter -> adapter.inquireHistory(java.util.Set.of(identity)).stream().findFirst()
                        .orElseThrow(() -> new ServiceException("检测柜未返回历史清单")));
        TTestBackfillCursor cursor = cursor(test, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                availability.btsTestId());
        cursor.setProtocolMetadata(JSON.toJSONString(java.util.Map.of("availableRecordCount",
                availability.availableRecordCount(), "uploadComplete", availability.uploadComplete(),
                "adapterCode", adapterCode)));
        cursor.setUpdatedAtUtc(new Date());
        mapper.updateBackfillCursor(cursor);
        long position = Math.max(1L, cursor.getNextPosition() == null ? 1L : cursor.getNextPosition());
        while (true)
        {
            final long requestPosition = position;
            HistoryPage page = registry.execute(adapterCode,
                    adapter -> adapter.downloadHistory(identity, availability.btsTestId(), requestPosition, PAGE_SIZE));
            for (HistorySample sample : page.samples())
            {
                if (!within(sample.deviceTime(), intervals)) continue;
                saveBtsSample(test, binding, intervals, availability.btsTestId(), sample);
            }
            long next = page.nextPosition() > position ? page.nextPosition() : position + page.samples().size();
            cursor.setNextPosition(next);
            cursor.setSourceStatus(page.samples().isEmpty() && page.nextPosition() <= position ? "COMPLETED" : "RUNNING");
            cursor.setUpdatedAtUtc(new Date());
            mapper.updateBackfillCursor(cursor);
            if (page.samples().isEmpty() || page.nextPosition() <= position) break;
            position = next;
        }
        cursor.setSourceStatus("COMPLETED");
        cursor.setUpdatedAtUtc(new Date());
        mapper.updateBackfillCursor(cursor);
    }

    private void saveBtsSample(TTest test, TTestChannelBinding binding, List<TTestReceptionInterval> intervals,
            String btsTestId, HistorySample sample)
    {
        byte[] hash = hash(sample.sequenceId() + "|" + sample.stepNo() + "|" + sample.cycleNo() + "|"
                + sample.stepType() + "|" + sample.deviceTimeRaw() + "|" + sample.voltageV() + "|"
                + sample.currentA() + "|" + sample.capacityAh() + "|" + sample.energyWh());
        String old = mapper.selectSampleContentHash(binding.getChannelId(), btsTestId, sample.sequenceId());
        if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
        {
            mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
                    "BTS testid=" + btsTestId + ", seqid=" + sample.sequenceId());
            return;
        }
        mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(),
                intervalId(intervals, sample.deviceTime()), btsTestId, sample.sequenceId(), Date.from(sample.deviceTime()),
                sample.deviceTimeRaw(), new Date(), sample.cycleNo(), sample.stepNo(), sample.stepType(), sample.testTimeMs(),
                sample.voltageV(), sample.currentA(), sample.capacityAh(), sample.energyWh(), hash);
        Long sampleId = mapper.selectBtsSampleId(binding.getChannelId(), btsTestId, sample.sequenceId());
        if (sampleId != null)
            sample.measurements().forEach((code, measurement) -> mapper.insertBtsMeasurement(sampleId, code,
                    measurement.value(), measurement.unit() == null ? "" : measurement.unit(), measurement.sourceType()));
    }

    private void backfillCollector(TTest test, TTestBattery battery, TTestCollectorChannel mapping)
    {
        if (battery.getCollectorId() == null) return;
        if (battery.getBinding() == null) throw new ServiceException("采集器绑定缺少检测柜通道");
        String collectorCode = mapper.selectCollectorCode(battery.getCollectorId());
        if (collectorCode == null || collectorCode.isBlank()) throw new ServiceException("采集器来源不存在");
        List<TTestReceptionInterval> intervals = intervals(test.getTestId(), battery.getBinding().getBindingId(),
                "COLLECTOR_CHANNEL", battery.getCollectorId(), mapping.getChannelCode());
        if (intervals.isEmpty()) return;
        TTestBackfillCursor cursor = cursor(test, battery.getBinding().getBindingId(), "COLLECTOR_CHANNEL",
                battery.getCollectorId(), mapping.getChannelCode(), collectorCode);
        long offset = Math.max(0L, (cursor.getNextPosition() == null ? 1L : cursor.getNextPosition()) - 1L);
        while (true)
        {
            var page = collectorHistoryStore.loadCollectorHistoryPage(collectorCode, offset, PAGE_SIZE, false);
            for (CollectorMeasurementRecord record : page.records())
            {
                if (!mapping.getMeasurementCode().equalsIgnoreCase(record.metricCode()) || !within(record.deviceTime(), intervals)) continue;
                saveCollectorSample(test, battery, mapping, record);
            }
            offset = page.nextOffset();
            cursor.setNextPosition(offset + 1L);
            cursor.setSourceStatus(page.hasMore() ? "RUNNING" : "COMPLETED");
            cursor.setUpdatedAtUtc(new Date());
            mapper.updateBackfillCursor(cursor);
            if (!page.hasMore()) break;
        }
    }

    private void saveCollectorSample(TTest test, TTestBattery battery, TTestCollectorChannel mapping,
            CollectorMeasurementRecord record)
    {
        byte[] hash = hash(record.sequenceId() + "|" + record.deviceTimeRaw() + "|" + record.metricCode() + "|"
                + record.value() + "|" + record.unit());
        String old = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), record.channelCode(),
                record.sequenceId(), record.metricCode());
        if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
        {
            mapper.insertBackfillConflict(test.getTestId(), null,
                    "采集器=" + record.collectorCode() + ", seqid=" + record.sequenceId());
            return;
        }
        mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(), battery.getCollectorId(),
                record.channelCode(), record.sequenceId(), Date.from(record.deviceTime()), record.deviceTimeRaw(),
                Date.from(record.serverReceivedAt()), record.metricCode(), record.value(), record.unit(), hash);
    }

    private TTestBackfillCursor cursor(TTest test, Long bindingId, String sourceType, Long sourceId,
            String channelCode, String protocolTestId)
    {
        TTestBackfillCursor value = mapper.selectBackfillCursorForUpdate(test.getTestId(), bindingId, sourceType, sourceId, channelCode);
        if (value != null)
        {
            if (value.getProtocolTestId() != null && !Objects.equals(value.getProtocolTestId(), protocolTestId))
                throw new ServiceException("历史协议测试标识发生变化");
            return value;
        }
        value = new TTestBackfillCursor();
        value.setTestId(test.getTestId()); value.setBindingId(bindingId); value.setSourceType(sourceType);
        value.setSourceId(sourceId); value.setSourceChannelCode(channelCode); value.setProtocolTestId(protocolTestId);
        value.setNextPosition(1L); value.setSourceStatus("PENDING"); value.setProtocolMetadata("{}");
        value.setUpdatedAtUtc(new Date());
        mapper.insertBackfillCursor(value);
        return value;
    }

    private List<TTestReceptionInterval> intervals(Long testId, Long bindingId, String sourceType, Long sourceId,
            String channelCode)
    {
        return mapper.selectTestReceptionIntervals(testId).stream()
                .filter(item -> Objects.equals(item.getBindingId(), bindingId) && Objects.equals(item.getSourceType(), sourceType)
                        && Objects.equals(item.getSourceId(), sourceId)
                        && Objects.equals(item.getSourceChannelCode(), channelCode)).toList();
    }

    private boolean within(Instant time, List<TTestReceptionInterval> intervals)
    {
        return time != null && intervals.stream().anyMatch(item -> !time.isBefore(item.getStartBoundaryAtUtc().toInstant())
                && (item.getEndBoundaryAtUtc() == null || time.isBefore(item.getEndBoundaryAtUtc().toInstant())));
    }

    private Long intervalId(List<TTestReceptionInterval> intervals, Instant time)
    {
        return intervals.stream().filter(item -> !time.isBefore(item.getStartBoundaryAtUtc().toInstant())
                && (item.getEndBoundaryAtUtc() == null || time.isBefore(item.getEndBoundaryAtUtc().toInstant())))
                .map(TTestReceptionInterval::getIntervalId).findFirst().orElse(null);
    }

    private byte[] hash(String value)
    {
        try { return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8)); }
        catch (Exception exception) { throw new IllegalStateException("SHA-256不可用", exception); }
    }
}

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

@@ -19,6 +19,9 @@
        <result property="startedAtUtc" column="started_at_utc" />
        <result property="endedAtUtc" column="ended_at_utc" />
        <result property="runtimeAttach" column="runtime_attach" />
        <result property="backfillStatus" column="backfill_status" />
        <result property="backfillStartedAtUtc" column="backfill_started_at_utc" />
        <result property="backfillCompletedAtUtc" column="backfill_completed_at_utc" />
        <result property="delFlag" column="del_flag" />
        <result property="createBy" column="create_by" />
        <result property="createTime" column="create_time" />
@@ -153,7 +156,8 @@
               t.started_at_utc, t.ended_at_utc, t.del_flag,
               t.create_by, t.create_time, t.update_by, t.update_time, t.remark,
               (select count(1) from t_test_battery b where b.test_id = t.test_id and b.del_flag = 0) battery_count,
               t.runtime_attach runtime_attach
               t.runtime_attach runtime_attach, t.backfill_status, t.backfill_started_at_utc,
               t.backfill_completed_at_utc
        from t_test t
        left join sys_user creator on creator.user_id = t.creator_id and creator.del_flag = '0'
        left join sys_user principal on principal.user_id = t.principal_id and principal.del_flag = '0'
@@ -371,6 +375,9 @@
    <select id="selectCollectorChannelCount" resultType="int">
        select channel_count from t_collector where collector_id = #{collectorId} and status = '0' and del_flag = 0
    </select>
    <select id="selectCollectorCode" resultType="string">
        select collector_code from t_collector where collector_id = #{collectorId} and status = '0' and del_flag = 0
    </select>

    <select id="selectEnabledUserOptions" resultType="com.ruoyi.system.domain.vo.TestUserOptionVo">
        select user_id userId, user_name userName, nick_name nickName
@@ -430,6 +437,119 @@
          and cabinet.registration_status = 'REGISTERED'
        for update
    </select>

    <select id="selectTestsForNaturalEnd" resultMap="TTestResultMap">
        <include refid="selectTestBase" />
        where t.del_flag = 0 and t.reception_status in ('RECEIVING', 'PAUSED')
        order by t.test_id
    </select>
    <select id="selectTestsForBackfill" resultMap="TTestResultMap">
        <include refid="selectTestBase" />
        where t.del_flag = 0 and t.reception_status = 'ENDED'
          and t.test_status in ('COMPLETION_PENDING', 'COMPLETED')
          and coalesce(t.backfill_status, 'NOT_STARTED') &lt;&gt; 'COMPLETED'
        order by t.ended_at_utc, t.test_id
    </select>
    <resultMap id="TTestBackfillCursorResultMap" type="com.ruoyi.system.domain.TTestBackfillCursor">
        <id property="cursorId" column="cursor_id" />
        <result property="testId" column="test_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="protocolTestId" column="protocol_test_id" />
        <result property="nextPosition" column="next_position" />
        <result property="sourceStatus" column="source_status" />
        <result property="protocolMetadata" column="protocol_metadata" />
        <result property="lastError" column="last_error" />
        <result property="updatedAtUtc" column="updated_at_utc" />
    </resultMap>
    <select id="selectBackfillCursorForUpdate" resultMap="TTestBackfillCursorResultMap">
        select * from t_test_backfill_cursor
        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))
        for update
    </select>
    <insert id="insertBackfillCursor" parameterType="com.ruoyi.system.domain.TTestBackfillCursor" useGeneratedKeys="true" keyProperty="cursorId">
        insert into t_test_backfill_cursor
        (test_id, binding_id, source_type, source_id, source_channel_code, protocol_test_id,
         next_position, source_status, protocol_metadata, last_error, updated_at_utc)
        values (#{testId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{protocolTestId},
                #{nextPosition}, #{sourceStatus}, cast(#{protocolMetadata} as json), #{lastError}, #{updatedAtUtc})
    </insert>
    <update id="updateBackfillCursor" parameterType="com.ruoyi.system.domain.TTestBackfillCursor">
        update t_test_backfill_cursor set protocol_test_id = #{protocolTestId}, next_position = #{nextPosition},
            source_status = #{sourceStatus}, protocol_metadata = cast(#{protocolMetadata} as json),
            last_error = #{lastError}, updated_at_utc = #{updatedAtUtc}
        where cursor_id = #{cursorId}
    </update>
    <update id="updateTestBackfillStarted">
        update t_test set test_status = 'COMPLETION_PENDING', backfill_status = 'RUNNING',
            backfill_started_at_utc = coalesce(backfill_started_at_utc, #{updatedAtUtc}),
            update_time = #{updatedAtUtc} where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestBackfillWaiting">
        update t_test set backfill_status = 'WAITING', update_time = #{updatedAtUtc}, remark = #{error}
        where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestBackfillCompleted">
        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="updateTestNaturallyEnded">
        update t_test set test_status = 'COMPLETION_PENDING', reception_status = 'ENDED',
            ended_at_utc = #{endedAtUtc}, backfill_status = 'NOT_STARTED', update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED')
    </update>
    <select id="selectSampleContentHash" resultType="string">
        select hex(content_sha256) from t_test_sample
        where channel_id = #{channelId} and bts_test_id = #{btsTestId} and seq_id = #{sequenceId}
    </select>
    <insert id="insertBtsSample">
        insert ignore into t_test_sample
        (test_id, binding_id, channel_id, reception_interval_id, bts_test_id, seq_id, device_time_utc,
         device_time_raw, received_at_utc, cycle_no, step_no, step_type, test_time_ms, voltage_v,
         current_a, capacity_ah, energy_wh, content_sha256)
        values (#{testId}, #{bindingId}, #{channelId}, #{intervalId}, #{btsTestId}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{cycleNo}, #{stepNo}, #{stepType}, #{testTimeMs}, #{voltageV},
                #{currentA}, #{capacityAh}, #{energyWh}, #{contentSha256})
    </insert>
    <insert id="insertBtsMeasurement">
        insert ignore into t_test_sample_measurement
        (sample_id, metric_code, metric_value, unit, source_type)
        values (#{sampleId}, #{metricCode}, #{metricValue}, #{unit}, #{sourceType})
    </insert>
    <select id="selectBtsSampleId" resultType="long">
        select sample_id from t_test_sample
        where channel_id = #{channelId} and bts_test_id = #{btsTestId} and seq_id = #{sequenceId}
    </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,
         device_time_raw, server_received_at_utc, metric_code, metric_value, unit, content_sha256)
        values (#{testId}, #{bindingId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{metricCode}, #{metricValue}, #{unit}, #{contentSha256})
    </insert>
    <select id="selectCollectorSampleId" resultType="long">
        select sample_id from t_test_collector_sample
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and sequence_id = #{sequenceId} and metric_code = #{metricCode}
    </select>
    <select id="selectCollectorSampleContentHash" resultType="string">
        select hex(content_sha256) from t_test_collector_sample
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and sequence_id = #{sequenceId} and metric_code = #{metricCode}
    </select>
    <insert id="insertBackfillConflict">
        insert into t_test_exception
        (test_id, channel_id, exception_type, exception_status, severity,
         first_occurred_at_utc, last_occurred_at_utc, occurrence_count, detail)
        values (#{testId}, #{channelId}, 'HISTORY_CONTENT_CONFLICT', 'OPEN', 'WARNING', utc_timestamp(3),
                utc_timestamp(3), 1, #{detail})
        on duplicate key update last_occurred_at_utc = utc_timestamp(3), occurrence_count = occurrence_count + 1,
            detail = values(detail)
    </insert>
    <select id="selectChannelLeaseForUpdate" parameterType="Long" resultMap="TTestBindingResultMap">
        select binding.binding_id, binding.test_id, binding.battery_relation_id, binding.channel_id,
               binding.claim_status
@@ -522,6 +642,7 @@
    <update id="updateReceptionStopped">
        update t_test
        set test_status = 'COMPLETED', 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
    </update>