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

yeg

2026-08-29 8761644af158c5e7f05e999e9834b6496ce414e7

 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;
        }
    }
}