fix(telemetry): restore worker agent delivery and truthful system status

deploy/telemetry-plane/normalizer/normalizer.py

@@ -31,7 +31,8 @@ SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")
 MAX_PAYLOAD_BYTES: Final = 1024 * 1024
 MAX_TAGS: Final = 256
 MAX_SERIES_KEY_BYTES: Final = 4096
-RETENTION_INTERVAL_SECONDS: Final = 86_400
+RETENTION_INTERVAL_SECONDS: Final = 60
+RETENTION_BATCH_ROWS: Final = 1000
 ALLOWED_TELEMETRY_TAGS: Final = frozenset(
     {
         "agent_id",
@@ -434,6 +435,39 @@ def _start_query_server(
     return server
 
 
+def _retention_batch(connection: Any) -> int:
+    """Preserve the existing 30-day policy without an unbounded startup DELETE."""
+    with connection.transaction(), connection.cursor() as cursor:
+        cursor.execute("SET LOCAL statement_timeout = '2s'")
+        cursor.execute("SET LOCAL lock_timeout = '250ms'")
+        cursor.execute(
+            """
+            WITH expired AS (
+                SELECT tableoid, ctid FROM contour_telemetry_samples
+                WHERE observed_at < CURRENT_TIMESTAMP - INTERVAL '30 days'
+                ORDER BY observed_at LIMIT %s
+            )
+            DELETE FROM contour_telemetry_samples AS samples USING expired
+            WHERE samples.tableoid = expired.tableoid AND samples.ctid = expired.ctid
+            """,
+            (RETENTION_BATCH_ROWS,),
+        )
+        return cursor.rowcount
+
+
+def _retention_loop(dsn: str) -> None:
+    import psycopg  # type: ignore[import-not-found]
+
+    while True:
+        # Neither a retention timeout nor database recovery blocks MQTT I/O.
+        time.sleep(RETENTION_INTERVAL_SECONDS)
+        try:
+            with psycopg.connect(dsn, connect_timeout=3) as connection:
+                _retention_batch(connection)
+        except psycopg.Error:
+            print("telemetry retention deferred", flush=True)
+
+
 def main() -> None:
     import paho.mqtt.client as mqtt
     import psycopg  # type: ignore[import-not-found]
@@ -447,7 +481,9 @@ def main() -> None:
     password = _required("MISSIONCORE_MQTT_PASSWORD")
     runtime_health = TelemetryRuntimeHealth()
     connection = psycopg.connect(dsn, autocommit=True)
-    last_retention_monotonic = 0.0
+    threading.Thread(
+        target=_retention_loop, args=(dsn,), name="telemetry-retention", daemon=True
+    ).start()
     runtime_health.set_database(True)
     _start_query_server(dsn, runtime_health)
     client = mqtt.Client(
@@ -480,7 +516,7 @@ def main() -> None:
         runtime_health.set_mqtt(False, f"MQTT disconnected: {reason_code}")
 
     def on_message(_client: Any, _userdata: object, message: Any) -> None:
-        nonlocal connection, last_retention_monotonic
+        nonlocal connection
         try:
             row = _normalize(message.topic, message.payload)
         except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
@@ -489,19 +525,6 @@ def main() -> None:
         for attempt in range(2):
             try:
                 with connection.cursor() as cursor:
-                    now_monotonic = time.monotonic()
-                    if (
-                        now_monotonic - last_retention_monotonic
-                        >= RETENTION_INTERVAL_SECONDS
-                    ):
-                        cursor.execute(
-                            """
-                            DELETE FROM contour_telemetry_samples
-                            WHERE observed_at
-                                < CURRENT_TIMESTAMP - INTERVAL '30 days'
-                            """
-                        )
-                        last_retention_monotonic = now_monotonic
                     cursor.execute(
                     """
                     INSERT INTO contour_telemetry_samples (