From de7ab45496ae9a6dc50ff03ac840c0ede5a40b5e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 20:54:45 -0400 Subject: [PATCH] fix(asap-tools): keep 10 s more engine history for late query sends The query engine's retention slack covered the client's query_time_offset plus one ingestion interval, with no room for the client sending late. When a 60s query group fires alongside a 1s group, the 1s queries go out ~2-3 s behind schedule, and a 1s-window instant query then found its window already evicted (7 of 300 evaluations in a classic sketch-bench smoke run). Add CLIENT_SEND_LAG_SLACK_MS (10 s) to get_retention_slack_ms. Co-Authored-By: Claude Opus 5.5 --- .../experiments/experiment_utils/config.py | 14 ++++++++++--- .../experiments/tests/test_retention_slack.py | 21 +++++++++++++++---- 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index 197232d0..3863f31c 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -558,11 +558,17 @@ def get_prometheus_data_ingestion_interval_ms(prometheus_config): raise ValueError(f"Invalid scrape interval string: {s}") +# The client can send a query seconds after its scheduled time when slow +# queries from other groups fire alongside it. +CLIENT_SEND_LAG_SLACK_MS = 10_000 + + def get_retention_slack_ms(query_groups, data_ingestion_interval_ms: int) -> int: """How far in the past the query client evaluates (the largest client_options.query_time_offset, in seconds), plus one ingestion interval - for windows still closing. The query engine keeps this much extra - CircularBuffer history so delayed queries find their windows.""" + for windows still closing, plus CLIENT_SEND_LAG_SLACK_MS for late sends. + The query engine keeps this much extra CircularBuffer history so delayed + queries find their windows.""" max_offset_s = max( ( (group.get("client_options") or {}).get("query_time_offset", 0) @@ -570,7 +576,9 @@ def get_retention_slack_ms(query_groups, data_ingestion_interval_ms: int) -> int ), default=0, ) - return int(max_offset_s * 1000) + data_ingestion_interval_ms + return ( + int(max_offset_s * 1000) + data_ingestion_interval_ms + CLIENT_SEND_LAG_SLACK_MS + ) PLANNER_OUTPUT_FILES = ("streaming_config.yaml", "inference_config.yaml") diff --git a/asap-tools/experiments/tests/test_retention_slack.py b/asap-tools/experiments/tests/test_retention_slack.py index caba93fb..5a2c9651 100644 --- a/asap-tools/experiments/tests/test_retention_slack.py +++ b/asap-tools/experiments/tests/test_retention_slack.py @@ -4,7 +4,7 @@ from omegaconf import OmegaConf -from experiment_utils.config import get_retention_slack_ms +from experiment_utils.config import CLIENT_SEND_LAG_SLACK_MS, get_retention_slack_ms class RetentionSlackTest(unittest.TestCase): @@ -17,14 +17,27 @@ def test_largest_query_time_offset_plus_one_ingestion_interval(self): {"client_options": {"query_time_offset": 4}}, ] ) - self.assertEqual(get_retention_slack_ms(groups, 1000), 11_000) + self.assertEqual( + get_retention_slack_ms(groups, 1000), 11_000 + CLIENT_SEND_LAG_SLACK_MS + ) + + def test_covers_a_send_lagging_behind_the_offset(self): + # A 1s-window instant query sent ~12.4s after its evaluation time + # (offset 10s, client ~2.4s behind while a 60s group fired) found its + # window already evicted with only offset + one interval of slack. + groups = OmegaConf.create([{"client_options": {"query_time_offset": 10}}]) + self.assertGreaterEqual(get_retention_slack_ms(groups, 1000), 12_400 + 1000) def test_missing_offset_counts_as_zero(self): groups = OmegaConf.create([{"client_options": {}}, {}]) - self.assertEqual(get_retention_slack_ms(groups, 15_000), 15_000) + self.assertEqual( + get_retention_slack_ms(groups, 15_000), 15_000 + CLIENT_SEND_LAG_SLACK_MS + ) def test_no_query_groups(self): - self.assertEqual(get_retention_slack_ms(None, 1000), 1000) + self.assertEqual( + get_retention_slack_ms(None, 1000), 1000 + CLIENT_SEND_LAG_SLACK_MS + ) if __name__ == "__main__":