Don't include future samples in resampling windows

Currently only too-old samples are being filtered out from the relevant
samples by the resampler, but we should also filter out future samples,
as it might happen that the timer was delayed and was triggered a while
after a resampling period has passed, so we could have received samples
in the future for the current window's perspective.

This commit also filters these future samples and only consider relevant
samples that were emitted during the current resampling period.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/timeseries/_resampling.py

@@ -661,14 +661,15 @@ def resample(self, timestamp: datetime) -> Sample:
         # We need to pass a dummy Sample to bisect because it only support
         # specifying a key extraction function in Python 3.10, so we need to
         # compare samples at the moment.
-        cut_index = bisect(self._buffer, Sample(minimum_relevant_timestamp, None))
+        min_index = bisect(self._buffer, Sample(minimum_relevant_timestamp, None))
+        max_index = bisect(self._buffer, Sample(timestamp, None))
         # Using itertools for slicing doesn't look very efficient, but
         # experiements with a custom (ring) buffer that can slice showed that
         # it is not that bad. See:
         # https://github.com/frequenz-floss/frequenz-sdk-python/pull/130
         # So if we need more performance beyond this point, we probably need to
         # resort to some C (or similar) implementation.
-        relevant_samples = list(itertools.islice(self._buffer, cut_index, None))
+        relevant_samples = list(itertools.islice(self._buffer, min_index, max_index))
         value = (
             conf.resampling_function(relevant_samples, conf, props)
             if relevant_samples

tests/timeseries/test_resampling.py

@@ -349,6 +349,91 @@ async def test_timer_errors_are_logged(
     resampling_fun_mock.reset_mock()
 
 
+async def test_future_samples_not_included(
+    fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
+) -> None:
+    """Test resampling window size is consistent."""
+    timestamp = datetime.now(timezone.utc)
+
+    resampling_period_s = 2
+    expected_resampled_value = 42.0
+
+    resampling_fun_mock = MagicMock(
+        spec=ResamplingFunction, return_value=expected_resampled_value
+    )
+    config = ResamplerConfig(
+        resampling_period_s=resampling_period_s,
+        max_data_age_in_periods=2.0,
+        resampling_function=resampling_fun_mock,
+        initial_buffer_len=4,
+    )
+    resampler = Resampler(config)
+
+    source_receiver = source_chan.new_receiver()
+    source_sender = source_chan.new_sender()
+
+    sink_mock = AsyncMock(spec=Sink, return_value=True)
+
+    resampler.add_timeseries("test", source_receiver, sink_mock)
+    source_props = resampler.get_source_properties(source_receiver)
+
+    # Test timeline
+    #
+    # t(s)   0          1      1.9  2          3          4  4.1 4.2
+    #        |----------|--------|--R----------|----------R--|---|------------>
+    # value  5.0                7.0           4.0            3.0 timer fires
+    #                       (with ts=2.1)
+    #
+    # R = resampling is done
+
+    # Send a few samples and run a resample tick, advancing the fake time by one period
+    sample0s = Sample(timestamp, value=5.0)
+    sample1s = Sample(timestamp + timedelta(seconds=1), value=12.0)
+    sample2_1s = Sample(timestamp + timedelta(seconds=2.1), value=7.0)
+    await source_sender.send(sample0s)
+    await source_sender.send(sample1s)
+    await source_sender.send(sample2_1s)
+    fake_time.shift(resampling_period_s)
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == 2
+    sink_mock.assert_called_once_with(
+        Sample(
+            timestamp + timedelta(seconds=resampling_period_s), expected_resampled_value
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample0s, sample1s), config, source_props  # sample2_1s is not here
+    )
+    assert source_props == SourceProperties(
+        sampling_start=timestamp, received_samples=3, sampling_period_s=None
+    )
+    assert _get_buffer_len(resampler, source_receiver) == config.initial_buffer_len
+    sink_mock.reset_mock()
+    resampling_fun_mock.reset_mock()
+
+    # Second resampling run
+    sample3s = Sample(timestamp + timedelta(seconds=3), value=4.0)
+    sample4_1s = Sample(timestamp + timedelta(seconds=4.1), value=3.0)
+    await source_sender.send(sample3s)
+    await source_sender.send(sample4_1s)
+    fake_time.shift(resampling_period_s + 0.2)
+    await resampler.resample(one_shot=True)
+
+    assert datetime.now(timezone.utc).timestamp() == 4.2
+    sink_mock.assert_called_once_with(
+        Sample(
+            timestamp + timedelta(seconds=resampling_period_s * 2),
+            expected_resampled_value,
+        )
+    )
+    resampling_fun_mock.assert_called_once_with(
+        a_sequence(sample1s, sample2_1s, sample3s),
+        config,
+        source_props,  # sample4_1s is not here
+    )
+
+
 async def test_resampling_with_one_window(
     fake_time: time_machine.Coordinates, source_chan: Broadcast[Sample]
 ) -> None: