Add tick_delay to mitigate boundary races in cascaded resampling (#1394)

## Problem

In cascaded timer-based resamplers (for example `1s -> 10s`), a race can
happen at window boundaries:
the downstream resampler can tick and compute its window before the
upstream resampler has emitted the last sample for the previous window.

Example with `max_data_age_in_periods=1`:
- The upstream resampler computes `[9,10)` and emits its result (label
`9`) slightly **after** `t=10`.
- The downstream resampler computes `[0,10)` at `t=10` and does not see
that label-`9` sample yet.
- At the next tick (`t=20`), that delayed sample is too old and is no
longer considered.

Result: the affected downstream window stays permanently incomplete.

## Change

- Added `tick_delay` to `ResamplerConfig` and `ResamplerConfig2`.
- `tick_delay` is keyword-only to avoid positional-argument API
breakage.
- The resampler can now optionally wait `tick_delay` after a tick, while
still computing with the original window boundary.
- This delays processing without shifting the actual window definitions.
- Added validation: `tick_delay` must be `>= 0` and `<
resampling_period`.
- Added documentation for the cascaded-resampling use case.
- Marked `tick_delay` as experimental (it may be changed or deprecated
in the future).

## Tests

- Added validation tests for invalid `tick_delay` values (`< 0`, `>=
resampling_period`).
- Added fixture-based timing tests that separate sample arrival time
from timestamp-based window filtering.
- Added a with/without-delay late-arrival comparison (same timestamps
and arrival timing, only `tick_delay` differs) to make the delay effect
explicit.

Author: llucax

RELEASE_NOTES.md

@@ -10,7 +10,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* A new `tick_delay` option was added to `ResamplerConfig` and `ResamplerConfig2` to delay resampling execution after each timer tick. The delay was designed to postpone processing while keeping window boundaries aligned to the original tick times, which can be used for cascaded resampling pipelines. This option is experimental and may be changed or deprecated in a future release.
 
 ## Bug Fixes
 

src/frequenz/sdk/timeseries/_resampling/_config.py

@@ -201,6 +201,25 @@ class ResamplerConfig:
     time the resampler is created.
     """
 
+    tick_delay: timedelta = field(default=timedelta(0), kw_only=True)
+    """Delay before processing each resampling tick.
+
+    This delays when resampling computation happens, while keeping the
+    resampling windows aligned to the original timer tick boundaries. This
+    delay is only a time-based buffer, not a strict synchronization mechanism.
+
+    Warning:
+        This is an experimental option and may be changed or deprecated in
+        the future.
+
+    Example:
+        This can be used in cascaded resampling setups to reduce timing races
+        where downstream windows are processed before upstream resampled values
+        are emitted.
+
+    It must be non-negative and smaller than `resampling_period`.
+    """
+
     def __post_init__(self) -> None:
         """Check that config values are valid.
 
@@ -245,6 +264,13 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f"align_to ({self.align_to}) should be a timezone aware datetime"
             )
+        if self.tick_delay < timedelta(0):
+            raise ValueError(f"tick_delay ({self.tick_delay}) should be non-negative")
+        if self.tick_delay >= self.resampling_period:
+            raise ValueError(
+                f"tick_delay ({self.tick_delay}) should be smaller than "
+                f"resampling_period ({self.resampling_period})"
+            )
 
 
 class ResamplingFunction2(Protocol):
@@ -415,3 +441,10 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f"align_to ({self.align_to}) must be specified via timer_config"
             )
+        if self.tick_delay < timedelta(0):
+            raise ValueError(f"tick_delay ({self.tick_delay}) should be non-negative")
+        if self.tick_delay >= self.resampling_period:
+            raise ValueError(
+                f"tick_delay ({self.tick_delay}) should be smaller than "
+                f"resampling_period ({self.resampling_period})"
+            )

src/frequenz/sdk/timeseries/_resampling/_resampler.py

@@ -200,6 +200,11 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 case unexpected:
                     assert_never(unexpected)
 
+            # Delay processing to let upstream cascaded resamplers emit their
+            # boundary samples first; window boundaries still use next_tick_time.
+            if self._config.tick_delay:
+                await asyncio.sleep(self._config.tick_delay.total_seconds())
+
             # We need to make a copy here because we need to match the results to the
             # current resamplers, and since we await here, new resamplers could be added
             # or removed from the dict while we awaiting the resampling, which would