优化采集器长时历史的有界读取

yeg

2026-08-29 8761644af158c5e7f05e999e9834b6496ce414e7

优化采集器长时历史的有界读取

历史下载按目标区间分页读取，故障视图和重启恢复按最多一千条流式处理。补充六万条历史、Redis 精确范围及跨页损坏状态测试，避免单次检测历史整体进入 JVM。

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

@@ -25,7 +25,6 @@
    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:";
    private static final long RECENT_REQUESTS_TTL_HOURS = 1;
    private static final int HISTORY_READ_BATCH_SIZE = 1000;

    private final RedisTemplate<Object, Object> redisTemplate;
    private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of();
@@ -99,25 +98,56 @@
    }

    @Override
    public List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    public CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit,
            boolean reverse)
    {
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE);
        String key = historyKey(collectorCode);
        Long storedSize = redisTemplate.opsForList().size(key);
        List<CollectorMeasurementRecord> records = new ArrayList<>();
        if (storedSize != null)
        long redisCount = storedSize == null ? 0 : storedSize;
        List<CollectorMeasurementRecord> embedded = pendingEmbeddedHistory.stream()
                .filter(append -> append.collectorCode().equals(collectorCode))
                .findFirst().map(CollectorHistoryAppend::records).orElse(List.of());
        long total = redisCount + embedded.size();
        long from = Math.min(offset, total);
        int count = (int) Math.min(limit, total - from);
        List<CollectorMeasurementRecord> records = new ArrayList<>(count);
        if (reverse)
        {
            for (long start = 0; start < storedSize; start += HISTORY_READ_BATCH_SIZE)
            long lower = total - from - count;
            long upper = total - from;
            long embeddedLower = Math.max(lower, redisCount);
            for (long index = upper - 1; index >= embeddedLower; index--)
                records.add(embedded.get((int) (index - redisCount)));
            long redisUpper = Math.min(upper, redisCount);
            if (lower < redisUpper)
            {
                long end = Math.min(start + HISTORY_READ_BATCH_SIZE, storedSize) - 1;
                List<Object> stored = redisTemplate.opsForList().range(key, start, end);
                List<Object> stored = redisTemplate.opsForList().range(key, lower, redisUpper - 1);
                if (stored != null)
                    stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
                    for (int index = stored.size() - 1; index >= 0; index--)
                        records.add(collectorMeasurementRecord(stored.get(index)));
            }
        }
        pendingEmbeddedHistory.stream()
                .filter(append -> append.collectorCode().equals(collectorCode))
                .forEach(append -> records.addAll(append.records()));
        return List.copyOf(records);
        else
        {
            long upper = from + count;
            long redisUpper = Math.min(upper, redisCount);
            if (from < redisUpper)
            {
                List<Object> stored = redisTemplate.opsForList().range(key, from, redisUpper - 1);
                if (stored != null) stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
            }
            long embeddedFrom = Math.max(from, redisCount);
            if (embeddedFrom < upper)
                records.addAll(embedded.subList((int) (embeddedFrom - redisCount),
                        (int) (upper - redisCount)));
        }
        if (records.size() != count)
            throw new IllegalStateException("Collector history page changed while reading Redis");
        long nextOffset = from + records.size();
        return new CollectorHistoryBatch(offset, nextOffset, nextOffset < total, total, records);
    }

    @Override

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

@@ -5,7 +5,9 @@
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -115,7 +117,8 @@
        SimulatorState restored = store.loadOrInitialize(state(1, "127.0.0.2"));

        assertEquals(1L, restored.collectors().get(0).historyRecordCount());
        assertEquals(1, store.loadCollectorHistory("dev-sim-collector-1").size());
        assertEquals(1, store.loadCollectorHistoryPage("dev-sim-collector-1", 0, 1000, false)
                .records().size());
        assertEquals(true, stored.get().toString().contains("\"history\":[{"));
        store.completeValidatedRestore(restored);
        assertEquals(false, stored.get().toString().contains("\"history\":[{"));
@@ -182,7 +185,36 @@
        store.saveAndDeleteCollectorHistory(next, "dev-sim-collector-1");

        assertEquals(next, store.loadOrInitialize(current));
        assertEquals(List.of(), store.loadCollectorHistory("dev-sim-collector-1"));
        assertEquals(List.of(), store.loadCollectorHistoryPage("dev-sim-collector-1", 0, 1000, false)
                .records());
    }

    @Test
    void readsOnlyRequestedCollectorHistoryRange()
    {
        AtomicReference<Object> stored = new AtomicReference<>();
        RedisTemplate<Object, Object> redis = redis(stored);
        RedisBtsSimulatorStateStore store = new RedisBtsSimulatorStateStore(redis);
        SimulatorState current = state(1, "127.0.0.2");
        store.loadOrInitialize(current);
        var record = new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord(
                "dev-sim", "dev-sim-collector-1", "S-01", 1L,
                java.time.Instant.parse("2026-08-29T05:00:00Z"), "2026-08-29 13:00:00.000",
                java.time.Instant.parse("2026-08-29T05:00:00Z"), "temperature",
                new java.math.BigDecimal("25.1"), "℃");
        store.save(current, List.of(
                new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend(
                        "dev-sim-collector-1", java.util.Collections.nCopies(1010, record))));
        ListOperations<Object, Object> lists = redis.opsForList();
        clearInvocations(lists);

        var page = store.loadCollectorHistoryPage("dev-sim-collector-1", 1000, 10, false);

        assertEquals(1010, page.total());
        assertEquals(10, page.records().size());
        assertEquals(false, page.hasMore());
        verify(lists).range(RedisBtsSimulatorStateStore.HISTORY_KEY_PREFIX + "dev-sim-collector-1",
                1000L, 1009L);
    }

    @Test
@@ -290,8 +322,14 @@
                    listValues.put(key, records);
                    return (long) records.size();
                });
        when(lists.range(any(), anyLong(), anyLong())).thenAnswer(invocation ->
                List.copyOf(listValues.getOrDefault(invocation.getArgument(0), List.of())));
        when(lists.range(any(), anyLong(), anyLong())).thenAnswer(invocation -> {
            List<Object> records = listValues.getOrDefault(invocation.getArgument(0), List.of());
            long requestedStart = invocation.getArgument(1);
            long requestedEnd = invocation.getArgument(2);
            int start = (int) Math.min(requestedStart, records.size());
            int end = (int) Math.min(requestedEnd + 1, records.size());
            return start >= end ? List.of() : List.copyOf(records.subList(start, end));
        });
        when(lists.size(any())).thenAnswer(invocation ->
                (long) listValues.getOrDefault(invocation.getArgument(0), List.of()).size());
        when(redis.delete(org.mockito.ArgumentMatchers.<Object>any()))

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

@@ -210,8 +210,7 @@
    public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
    {
        CollectorState collector = requireCollector(collectorId);
        return DevelopmentCollectorSimulation.history(collector,
                stateStore.loadCollectorHistory(collector.deviceCode()), offset, limit);
        return DevelopmentCollectorSimulation.history(collector, stateStore, offset, limit);
    }

    /** 返回控制页需要的运行状态快照。 */
@@ -518,8 +517,7 @@
                    || !isIpv4(collector.ip()) || !ips.add(collector.ip())
                    || collector.channelCount() < 1 || collector.channelCount() > MAX_CHANNEL_COUNT)
                throw new IllegalStateException("Invalid BTS simulator collector state in Redis");
            DevelopmentCollectorSimulation.validate(collector,
                    stateStore.loadCollectorHistory(collector.deviceCode()));
            DevelopmentCollectorSimulation.validate(collector, stateStore);
            maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
        }
        java.util.Set<String> tombstoneIds = new java.util.HashSet<>();

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

@@ -9,6 +9,7 @@
public interface BtsSimulatorStateStore
{
    int CURRENT_VERSION = 1;
    int MAX_HISTORY_BATCH_SIZE = 1000;

    /** 读取已有状态；仅在状态不存在时原子写入并返回默认状态。 */
    SimulatorState loadOrInitialize(SimulatorState defaultState);
@@ -27,10 +28,14 @@
        save(state);
    }

    /** 读取指定采集器的只追加原始历史。 */
    default List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    /** 按原始写入顺序或反向顺序读取有界历史批次。 */
    default CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit,
            boolean reverse)
    {
        return List.of();
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE);
        return new CollectorHistoryBatch(offset, 0, false, 0, List.of());
    }

    /** 原子保存删除采集器后的主状态，并删除该采集器的独立历史。 */
@@ -110,6 +115,12 @@
        public CollectorHistoryAppend { records = List.copyOf(records); }
    }

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

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

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

@@ -8,7 +8,6 @@
import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -26,7 +25,7 @@
    public static final int DEFAULT_SAMPLE_INTERVAL_MS = 1000;
    public static final int MIN_SAMPLE_INTERVAL_MS = 100;
    public static final int MAX_SAMPLE_INTERVAL_MS = 60000;
    public static final int MAX_HISTORY_PAGE_SIZE = 1000;
    public static final int MAX_HISTORY_PAGE_SIZE = BtsSimulatorStateStore.MAX_HISTORY_BATCH_SIZE;
    private static final int MAX_SAMPLES_PER_ADVANCE = 1000;
    private static final ZoneId DEVICE_ZONE = ZoneId.of("Asia/Hong_Kong");
    private static final DateTimeFormatter DEVICE_TIME_FORMATTER =
@@ -145,22 +144,26 @@
    }

    /** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
    public static CollectorHistoryPage history(CollectorState current, List<CollectorMeasurementRecord> rawHistory,
    public static CollectorHistoryPage history(CollectorState current, BtsSimulatorStateStore stateStore,
            long offset, int limit)
    {
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_PAGE_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_PAGE_SIZE);

        List<CollectorMeasurementRecord> download = faultedHistory(current, rawHistory);
        int from = (int) Math.min(offset, download.size());
        int to = Math.min(from + limit, download.size());
        return new CollectorHistoryPage(current.deviceCode(), offset, to, to < download.size(),
                download.size(), download.subList(from, to));
        CollectorHistoryFaults faults = current.historyFaults();
        if (faults.missingSequences().isEmpty() && faults.duplicateSequence() == null
                && faults.conflictingSequence() == null && !faults.outOfOrder())
        {
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    current.deviceCode(), offset, limit, false);
            return new CollectorHistoryPage(current.deviceCode(), offset, batch.nextOffset(), batch.hasMore(),
                    batch.total(), batch.records());
        }
        return faultedHistoryPage(current, stateStore, offset, limit);
    }

    /** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
    public static void validate(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
    public static void validate(CollectorState state, BtsSimulatorStateStore stateStore)
    {
        validateSampleInterval(state.sampleIntervalMs());
        if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
@@ -173,7 +176,7 @@
            validateMetrics(channel.metrics());
        }
        validateLatestMeasurements(state);
        validateHistory(state, rawHistory);
        validateHistory(state, stateStore);
        validateFaults(state.historyFaults());
    }

@@ -193,25 +196,49 @@
        }
    }

    private static void validateHistory(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
    private static void validateHistory(CollectorState state, BtsSimulatorStateStore stateStore)
    {
        long maximumSequence = 0;
        long previousSequence = 0;
        long recordCount = 0;
        long offset = 0;
        Set<String> identities = new java.util.HashSet<>();
        for (CollectorMeasurementRecord record : rawHistory)
        while (true)
        {
            String identity = record.sequenceId() + "\u0000" + record.channelCode() + "\u0000" + record.metricCode();
            if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
                    record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
                    record.metricCode(), record.value(), record.unit()) || !identities.add(identity)
                    || record.sequenceId() < previousSequence)
                throw new IllegalStateException("Invalid collector history state in Redis");
            maximumSequence = Math.max(maximumSequence, record.sequenceId());
            previousSequence = record.sequenceId();
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    state.deviceCode(), offset, MAX_HISTORY_PAGE_SIZE, false);
            if (batch.offset() != offset || batch.records().size() > MAX_HISTORY_PAGE_SIZE)
                throw new IllegalStateException("Invalid collector history page in Redis");
            for (CollectorMeasurementRecord record : batch.records())
            {
                if (record.sequenceId() < previousSequence)
                    throw new IllegalStateException("Invalid collector history state in Redis");
                if (record.sequenceId() != previousSequence)
                {
                    identities.clear();
                    previousSequence = record.sequenceId();
                }
                String identity = record.channelCode() + "\u0000" + record.metricCode();
                if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
                        record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
                        record.metricCode(), record.value(), record.unit()) || !identities.add(identity))
                    throw new IllegalStateException("Invalid collector history state in Redis");
                maximumSequence = Math.max(maximumSequence, record.sequenceId());
                recordCount++;
            }
            if (!batch.hasMore())
            {
                if (batch.total() != recordCount)
                    throw new IllegalStateException("Invalid collector history count in Redis");
                break;
            }
            if (batch.nextOffset() <= offset)
                throw new IllegalStateException("Invalid collector history page in Redis");
            offset = batch.nextOffset();
        }
        if (state.nextSequence() <= maximumSequence)
            throw new IllegalStateException("Invalid collector history sequence in Redis");
        if (state.historyRecordCount() != rawHistory.size())
        if (state.historyRecordCount() != recordCount)
            throw new IllegalStateException("Invalid collector history count in Redis");
    }

@@ -272,28 +299,50 @@
        return channelCode + "\u0000" + metricCode;
    }

    private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current,
            List<CollectorMeasurementRecord> rawHistory)
    private static CollectorHistoryPage faultedHistoryPage(CollectorState current,
            BtsSimulatorStateStore stateStore, long offset, int limit)
    {
        // 故障视图需返回精确 total，因此逐批扫描原始历史；正常下载会直接读取目标页。
        CollectorHistoryFaults faults = current.historyFaults();
        Set<Long> missing = Set.copyOf(faults.missingSequences());
        List<CollectorMeasurementRecord> records = new ArrayList<>();
        for (CollectorMeasurementRecord record : rawHistory)
        List<CollectorMeasurementRecord> page = new ArrayList<>(limit);
        long rawOffset = 0;
        long total = 0;
        while (true)
        {
            if (missing.contains(record.sequenceId())) continue;
            records.add(record);
            if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence())) records.add(record);
            if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
                records.add(new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(),
                        record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(),
                        record.serverReceivedAt(), record.metricCode(),
                        record.value().add(new BigDecimal("0.001")), record.unit()));
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    current.deviceCode(), rawOffset, MAX_HISTORY_PAGE_SIZE, faults.outOfOrder());
            for (CollectorMeasurementRecord record : batch.records())
            {
                if (missing.contains(record.sequenceId())) continue;
                total = appendFaultedRecord(page, record, total, offset, limit);
                if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence()))
                    total = appendFaultedRecord(page, record, total, offset, limit);
                if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
                    total = appendFaultedRecord(page, conflictingRecord(record), total, offset, limit);
            }
            if (!batch.hasMore()) break;
            if (batch.nextOffset() <= rawOffset)
                throw new IllegalStateException("Invalid collector history page in Redis");
            rawOffset = batch.nextOffset();
        }
        if (faults.outOfOrder())
            records.sort(Comparator.comparingLong(CollectorMeasurementRecord::sequenceId).reversed()
                    .thenComparing(CollectorMeasurementRecord::channelCode)
                    .thenComparing(CollectorMeasurementRecord::metricCode));
        return records;
        long from = Math.min(offset, total);
        long nextOffset = from + page.size();
        return new CollectorHistoryPage(current.deviceCode(), offset, nextOffset, nextOffset < total, total, page);
    }

    private static long appendFaultedRecord(List<CollectorMeasurementRecord> page,
            CollectorMeasurementRecord record, long index, long offset, int limit)
    {
        if (index >= offset && page.size() < limit) page.add(record);
        return index + 1;
    }

    private static CollectorMeasurementRecord conflictingRecord(CollectorMeasurementRecord record)
    {
        return new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(), record.channelCode(),
                record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
                record.metricCode(), record.value().add(new BigDecimal("0.001")), record.unit());
    }

    private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,

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

@@ -5,6 +5,7 @@
import java.util.List;
import java.util.Map;
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.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
@@ -14,7 +15,7 @@
{
    private SimulatorState state;
    private List<RecentRequest> recentRequests = List.of();
    private final Map<String, List<CollectorMeasurementRecord>> collectorHistory = new LinkedHashMap<>();
    private final Map<String, ArrayList<CollectorMeasurementRecord>> collectorHistory = new LinkedHashMap<>();

    public InMemoryBtsSimulatorStateStore() { }

@@ -40,19 +41,35 @@
    public synchronized void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
    {
        for (CollectorHistoryAppend append : historyAppends)
        {
            List<CollectorMeasurementRecord> records = new ArrayList<>(
                    collectorHistory.getOrDefault(append.collectorCode(), List.of()));
            records.addAll(append.records());
            collectorHistory.put(append.collectorCode(), List.copyOf(records));
        }
            collectorHistory.computeIfAbsent(append.collectorCode(), key -> new ArrayList<>())
                    .addAll(append.records());
        this.state = state;
    }

    @Override
    public synchronized List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    public synchronized CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset,
            int limit, boolean reverse)
    {
        return collectorHistory.getOrDefault(collectorCode, List.of());
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE);
        List<CollectorMeasurementRecord> history = collectorHistory.getOrDefault(collectorCode,
                new ArrayList<>());
        long total = history.size();
        int from = (int) Math.min(offset, total);
        int count = Math.min(limit, history.size() - from);
        List<CollectorMeasurementRecord> records = new ArrayList<>(count);
        if (reverse)
        {
            for (int index = history.size() - 1 - from; index >= history.size() - from - count; index--)
                records.add(history.get(index));
        }
        else
        {
            records.addAll(history.subList(from, from + count));
        }
        long nextOffset = from + records.size();
        return new CollectorHistoryBatch(offset, nextOffset, nextOffset < total, total, records);
    }

    @Override

 ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java

@@ -24,12 +24,14 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryBatch;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;

class SimulatorPublicContractTest
@@ -121,15 +123,28 @@
    void collectorKeepsAppendOnlyHistoryBeyondPreviousDisplayLimit() throws Exception
    {
        Instant startedAt = Instant.parse("2026-08-29T03:00:00Z");
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
        TrackingHistoryStore store = new TrackingHistoryStore();
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, store))
        {
            simulator.configureCollectorSampling(1, 100, CollectorHistoryFaults.none());
            simulator.advanceCollectors(startedAt);
            for (int batch = 1; batch <= 5; batch++)
                simulator.advanceCollectors(startedAt.plusMillis(batch * 100_000L));

            assertEquals(60_012, simulator.collectorHistory(1, 0, 1).total());
            assertEquals(1, simulator.collectorHistory(1, 0, 1).records().get(0).sequenceId());
            store.resetReadStats();
            var page = simulator.collectorHistory(1, 0, 1);
            assertEquals(60_012, page.total());
            assertEquals(1, page.records().get(0).sequenceId());
            assertEquals(1, store.historyReads);
            assertEquals(1, store.recordsRead);
        }

        store.resetReadStats();
        try (BtsDevelopmentSimulator ignored = new BtsDevelopmentSimulator(0, store))
        {
            assertTrue(store.historyReads > 1);
            assertEquals(1000, store.maximumRequestedLimit);
            assertEquals(60_012, store.recordsRead);
        }
    }

@@ -480,4 +495,98 @@
        assertThrows(IllegalStateException.class,
                () -> new BtsDevelopmentSimulator(0, store));
    }

    @Test
    void developmentSimulatorRejectsHistorySequenceRegressionAcrossPages()
    {
        Instant observedAt = Instant.parse("2026-08-29T06:30:00Z");
        List<CollectorChannelState> channels = List.of(new CollectorChannelState("S-01", List.of(
                new CollectorMetricState("temperature", new BigDecimal("25.0"), "℃"))));
        List<CollectorMeasurementRecord> records = new java.util.ArrayList<>();
        for (long sequence = 1; sequence <= 1000; sequence++)
            records.add(new CollectorMeasurementRecord("dev-sim", "dev-sim-collector-1", "S-01", sequence,
                    observedAt, "2026-08-29 14:30:00.000", observedAt, "temperature",
                    new BigDecimal("25.0"), "℃"));
        records.add(new CollectorMeasurementRecord("dev-sim", "dev-sim-collector-1", "S-01", 999,
                observedAt, "2026-08-29 14:30:00.000", observedAt, "temperature",
                new BigDecimal("25.0"), "℃"));
        CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
                "02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
                1000, 1001L, observedAt, channels, List.of(), records.size(), CollectorHistoryFaults.none());
        SimulatorState damaged = new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
                List.of(), List.of(collector), 1, 2, List.of());
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore(damaged);
        store.save(damaged, List.of(new CollectorHistoryAppend(collector.deviceCode(), records)));

        assertThrows(IllegalStateException.class,
                () -> new BtsDevelopmentSimulator(0, store));
    }

    private static final class TrackingHistoryStore implements BtsSimulatorStateStore
    {
        private final InMemoryBtsSimulatorStateStore delegate = new InMemoryBtsSimulatorStateStore();
        private int historyReads;
        private int maximumRequestedLimit;
        private long recordsRead;

        @Override
        public SimulatorState loadOrInitialize(SimulatorState defaultState)
        {
            return delegate.loadOrInitialize(defaultState);
        }

        @Override
        public void save(SimulatorState state)
        {
            delegate.save(state);
        }

        @Override
        public void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
        {
            delegate.save(state, historyAppends);
        }

        @Override
        public CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit,
                boolean reverse)
        {
            CollectorHistoryBatch batch = delegate.loadCollectorHistoryPage(collectorCode, offset, limit, reverse);
            historyReads++;
            maximumRequestedLimit = Math.max(maximumRequestedLimit, limit);
            recordsRead += batch.records().size();
            return batch;
        }

        @Override
        public void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
        {
            delegate.saveAndDeleteCollectorHistory(state, collectorCode);
        }

        @Override
        public List<RecentRequest> loadRecentRequests()
        {
            return delegate.loadRecentRequests();
        }

        @Override
        public void saveRecentRequests(List<RecentRequest> requests)
        {
            delegate.saveRecentRequests(requests);
        }

        @Override
        public void reset(SimulatorState defaultState)
        {
            delegate.reset(defaultState);
        }

        private void resetReadStats()
        {
            historyReads = 0;
            maximumRequestedLimit = 0;
            recordsRead = 0;
        }
    }
}