try overload fix 2, backend upsert and test script fix

backend/src/config.py

@@ -43,6 +43,7 @@ class BackendSettings:
     packet_tracker_retention_seconds: float
     packet_tracker_min_flush_interval_seconds: float
     packet_tracker_persist_timeout_seconds: float
+    packet_tracker_batch_persist_timeout_seconds: float
     packet_tracker_persist_retry_backoff_seconds: float
     packet_tracker_persist_retry_backoff_max_seconds: float
     packet_tracker_error_log_interval_seconds: float
@@ -89,6 +90,10 @@ def load_settings() -> BackendSettings:
         packet_tracker_retention_seconds=_env_float("BACKEND_PACKET_TRACKER_RETENTION_SECONDS", 10.0),
         packet_tracker_min_flush_interval_seconds=_env_float("BACKEND_PACKET_TRACKER_MIN_FLUSH_INTERVAL_SECONDS", 0.05),
         packet_tracker_persist_timeout_seconds=_env_float("BACKEND_PACKET_TRACKER_PERSIST_TIMEOUT_SECONDS", 2.0),
+        packet_tracker_batch_persist_timeout_seconds=_env_float(
+            "BACKEND_PACKET_TRACKER_BATCH_PERSIST_TIMEOUT_SECONDS",
+            10.0,
+        ),
         packet_tracker_persist_retry_backoff_seconds=_env_float(
             "BACKEND_PACKET_TRACKER_PERSIST_RETRY_BACKOFF_SECONDS",
             0.25,

backend/src/utilities/database.py

@@ -263,6 +263,42 @@ class DatabasePool:
         if self._pool is None:
             await self.init_pool()
 
+        if self._pool is None:
+            return
+
+        try:
+            async with self._pool.acquire() as conn:
+                row = await self._upsert_packet_with_conn(conn, pkt_info)
+        except Exception:
+            logger.exception("DB upsert failed")
+            return
+
+        self._publish_packet_row(pkt_info, row)
+
+    async def upsert_packets(self, packets: List[Dict[str, Any]]) -> None:
+        """Insert or update packet records using one database connection for the batch."""
+        if not packets:
+            return
+        if self._pool is None:
+            await self.init_pool()
+        if self._pool is None:
+            return
+
+        published_rows: List[tuple[Dict[str, Any], Optional[Dict[str, Any]]]] = []
+        try:
+            async with self._pool.acquire() as conn:
+                for pkt_info in packets:
+                    row = await self._upsert_packet_with_conn(conn, pkt_info)
+                    published_rows.append((pkt_info, row))
+        except Exception:
+            logger.exception("DB packet batch upsert failed")
+            raise
+
+        for pkt_info, row in published_rows:
+            self._publish_packet_row(pkt_info, row)
+
+    async def _upsert_packet_with_conn(self, conn: asyncpg.Connection, pkt_info: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """Insert or update one packet record using an already-acquired connection."""
         _normalize_json_fields(pkt_info)
         _attach_derived_fields(pkt_info)
 
@@ -270,8 +306,6 @@ class DatabasePool:
         telemetry_metadata = pkt_info.get("telemetry_metadata")
         capture_observations = pkt_info.get("capture_observations")
 
-        try:
-            async with self._pool.acquire() as conn:
         row = await conn.fetchrow(
             """
             INSERT INTO packets (
@@ -412,12 +446,11 @@ class DatabasePool:
             json.dumps(capture_observations) if capture_observations is not None else None,
             pkt_info.get("raw"),
         )
-        except Exception:
-            logger.exception("DB upsert failed")
-            return
+        return dict(row) if row else None
 
+    def _publish_packet_row(self, pkt_info: Dict[str, Any], row: Optional[Dict[str, Any]]) -> None:
         if row:
-            persisted = _serialize_row_for_broadcast(dict(row))
+            persisted = _serialize_row_for_broadcast(row)
             pkt_info.update(persisted)
 
         if self.broadcaster:

backend/src/utilities/packet_tracker.py

@@ -178,6 +178,8 @@ class PacketTracker:
             "persist_failed_total": 0,
             "persist_timeout_total": 0,
             "persist_failure_log_suppressed": 0,
+            "persist_batch_total": 0,
+            "persist_batch_failed_total": 0,
             "evicted_persisted_total": 0,
             "evicted_unpersisted_total": 0,
         }
@@ -689,8 +691,7 @@ class PacketTracker:
                 for correlation_key in expired_keys:
                     self._entries.pop(correlation_key, None)
 
-            for entry in due_entries:
-                self._persist(entry)
+            self._persist_batch(due_entries)
 
     def _persist_backoff_elapsed(self, entry: Dict[str, Any], now_ts: float) -> bool:
         failures = int(entry.get("persist_failures") or 0)
@@ -707,23 +708,36 @@ class PacketTracker:
         )
         return now_ts - float(entry.get("last_persist_attempt_at") or 0.0) >= backoff
 
-    def _persist(self, entry: Dict[str, Any]) -> None:
-        payload = dict(entry["payload"])
+    def _persist_batch(self, entries: List[Dict[str, Any]]) -> None:
+        if not entries:
+            return
 
+        payloads = [dict(entry["payload"]) for entry in entries]
         web_loop = getattr(shared_objects, "web_loop", None)
         web_db = getattr(shared_objects, "db", None)
         if web_loop is None or web_db is None:
             return
 
+        fut: concurrent.futures.Future[Any]
         try:
-            fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packet(payload), web_loop)
-            fut.result(timeout=settings.packet_tracker_persist_timeout_seconds)
+            if hasattr(web_db, "upsert_packets"):
+                fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packets(payloads), web_loop)
+            else:
+                fut = asyncio.run_coroutine_threadsafe(self._persist_payloads_one_by_one(web_db, payloads), web_loop)
+            timeout = max(
+                settings.packet_tracker_persist_timeout_seconds,
+                settings.packet_tracker_batch_persist_timeout_seconds,
+            )
+            fut.result(timeout=timeout)
             with self._lock:
+                self._stats["persist_batch_total"] += 1
+                persisted_at = time.time()
+                for entry in entries:
                     current = self._entries.get(entry["correlation_key"])
                     if current is not None:
                         current["persisted"] = True
                         current["dirty"] = False
-                    current["last_persisted_at"] = time.time()
+                        current["last_persisted_at"] = persisted_at
                         current["persist_failures"] = 0
         except Exception as exc:
             try:
@@ -733,9 +747,11 @@ class PacketTracker:
 
             is_timeout = isinstance(exc, (TimeoutError, concurrent.futures.TimeoutError, asyncio.TimeoutError))
             with self._lock:
-                self._stats["persist_failed_total"] += 1
+                self._stats["persist_batch_failed_total"] += 1
+                self._stats["persist_failed_total"] += len(entries)
                 if is_timeout:
-                    self._stats["persist_timeout_total"] += 1
+                    self._stats["persist_timeout_total"] += len(entries)
+                for entry in entries:
                     current = self._entries.get(entry["correlation_key"])
                     if current is not None:
                         current["dirty"] = True
@@ -745,14 +761,18 @@ class PacketTracker:
             if now_ts - self._last_persist_error_log_at >= settings.packet_tracker_error_log_interval_seconds:
                 self._last_persist_error_log_at = now_ts
                 logger.warning(
-                    "Packet persistence is overloaded; failed to persist %s (%s). Further errors are rate-limited.",
-                    entry["correlation_key"],
+                    "Packet persistence is overloaded; failed to persist batch of %s packets (%s). Further errors are rate-limited.",
+                    len(entries),
                     type(exc).__name__,
                 )
             else:
                 with self._lock:
                     self._stats["persist_failure_log_suppressed"] += 1
 
+    async def _persist_payloads_one_by_one(self, web_db: Any, payloads: List[Dict[str, Any]]) -> None:
+        for payload in payloads:
+            await web_db.upsert_packet(payload)
+
     def _record_stats(self, payload: Dict[str, Any]) -> None:
         capture_sources = set(payload.get("capture_sources") or [])
         self._stats["persisted_total"] += 1

tools/network_timing_benchmark.py

@@ -24,6 +24,7 @@ import os
 import platform
 import re
 import shutil
+import signal
 import socket
 import statistics
 import subprocess
@@ -264,33 +265,42 @@ def run_command(
     start_monotonic = time.perf_counter()
     start_wall = now_iso()
     sampler.start()
+    process: subprocess.Popen[str] | None = None
     try:
-        completed = subprocess.run(
+        process = subprocess.Popen(
             command,
             cwd=str(cwd),
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
-            timeout=timeout,
-            check=False,
+            start_new_session=True,
         )
+        stdout, stderr = process.communicate(timeout=timeout)
         result = CommandRun(
             args=command,
-            returncode=completed.returncode,
-            stdout=completed.stdout or "",
-            stderr=completed.stderr or "",
+            returncode=process.returncode,
+            stdout=stdout or "",
+            stderr=stderr or "",
         )
     except subprocess.TimeoutExpired as exc:
+        if process is not None:
+            terminate_process_group(process)
+            stdout, stderr = process.communicate()
+        else:
             stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
             stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
         result = CommandRun(
             args=command,
-            returncode=None,
+            returncode=process.returncode if process is not None else None,
             stdout=stdout,
             stderr=stderr,
             timed_out=True,
             error=f"command timed out after {timeout} seconds",
         )
+    except KeyboardInterrupt:
+        if process is not None:
+            terminate_process_group(process)
+        raise
     except OSError as exc:
         result = CommandRun(
             args=command,
@@ -337,6 +347,22 @@ def run_command(
     return result, meta
 
 
+def terminate_process_group(process: subprocess.Popen[str], grace_seconds: float = 2.0) -> None:
+    """Terminate a command and any children it spawned."""
+    if process.poll() is not None:
+        return
+    try:
+        os.killpg(process.pid, signal.SIGTERM)
+        process.wait(timeout=grace_seconds)
+    except subprocess.TimeoutExpired:
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+    except ProcessLookupError:
+        pass
+
+
 def parse_ping_output(text: str, sent_fallback: int) -> dict[str, Any]:
     rtts: list[float] = []
     for line in text.splitlines():
@@ -1031,6 +1057,12 @@ def parse_csv_strings(value: str) -> list[str]:
     return items
 
 
+def cooldown(seconds: float) -> None:
+    if seconds <= 0:
+        return
+    time.sleep(seconds)
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         description="Measure latency, jitter, throughput, loss, CPU, and interface counters.",
@@ -1066,6 +1098,12 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--mtr-count", type=int, default=100)
     parser.add_argument("--no-archive", action="store_true")
     parser.add_argument("--command-timeout-extra", type=float, default=15.0)
+    parser.add_argument(
+        "--cooldown-seconds",
+        type=float,
+        default=0.0,
+        help="Sleep after each test so local queues, CPU, and the backend can drain before the next test.",
+    )
     return parser
 
 
@@ -1083,6 +1121,8 @@ def main() -> int:
         parser.error("--flent-duration must be >= 1")
     if args.sockperf_duration < 1:
         parser.error("--sockperf-duration must be >= 1")
+    if args.cooldown_seconds < 0:
+        parser.error("--cooldown-seconds must be >= 0")
     for protocol in args.sockperf_protocols:
         if protocol not in {"tcp", "udp"}:
             parser.error("--sockperf-protocols may only contain tcp and/or udp")
@@ -1095,14 +1135,23 @@ def main() -> int:
 
     tests: list[dict[str, Any]] = []
     tests.extend(run_ping_tests(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_arping(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_iperf_tcp(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_iperf_tcp_reverse(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_iperf_udp(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_iperf_udp_reverse(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.extend(run_sockperf(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.extend(run_flent(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_traceroute(args, out_dir))
+    cooldown(args.cooldown_seconds)
     tests.append(run_mtr(args, out_dir))
 
     summary = {