fix(asap-tools): keep 10 s more engine history for late query sends

asap-tools/experiments/experiment_utils/config.py

@@ -558,19 +558,27 @@ 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)
             for group in query_groups or []
         ),
         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")

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__":