diff --git a/scripts/telemetry_adoption_report.py b/scripts/telemetry_adoption_report.py index 1f7fb21b7e..322f2c8091 100644 --- a/scripts/telemetry_adoption_report.py +++ b/scripts/telemetry_adoption_report.py @@ -1138,6 +1138,8 @@ def fetch_rows_local(db_path: str, since_days: int) -> dict[str, Any]: def fetch_rows_remote(ssh_host: str, db_path: str, since_days: int) -> dict[str, Any]: + # Streams JSON-lines (db_stats header, then one row per line) so the remote + # process never holds the full result set — the license droplet has 1GB RAM. remote_script = """ import json import sqlite3 @@ -1161,11 +1163,9 @@ rows_sql = ( ) try: db_stats = dict(conn.execute(db_stats_sql).fetchone()) - rows = [ - dict(row) - for row in conn.execute(rows_sql, (since_days,)).fetchall() - ] - print(json.dumps({"db_stats": db_stats, "rows": rows})) + print(json.dumps({"db_stats": db_stats})) + for row in conn.execute(rows_sql, (since_days,)): + print(json.dumps(dict(row))) finally: conn.close() """ @@ -1176,7 +1176,12 @@ finally: capture_output=True, check=True, ) - return json.loads(result.stdout) + lines = (line for line in result.stdout.splitlines() if line.strip()) + try: + header = json.loads(next(lines)) + except StopIteration: + raise RuntimeError(f"empty response from remote telemetry fetch on {ssh_host}") from None + return {"db_stats": header["db_stats"], "rows": [json.loads(line) for line in lines]} def counter_entries(counter: Counter[str], key_name: str) -> list[dict[str, Any]]: diff --git a/scripts/tests/test_telemetry_adoption_report.py b/scripts/tests/test_telemetry_adoption_report.py index 05f0837bcd..8dba193028 100644 --- a/scripts/tests/test_telemetry_adoption_report.py +++ b/scripts/tests/test_telemetry_adoption_report.py @@ -5,8 +5,11 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path +import json +import subprocess import sys import unittest +from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parents[1] @@ -26,6 +29,26 @@ class TelemetryAdoptionReportTest(unittest.TestCase): "6.0.0-rc.1+git.45.gabcdef", ) + def test_fetch_rows_remote_parses_json_lines_stream(self) -> None: + db_stats = {"latest_ping": "2026-07-17 00:00:00", "total_rows": 2, "total_distinct_installs": 2} + rows = [ + {"install_id": "a", "received_at": "2026-07-17 00:00:00"}, + {"install_id": "b", "received_at": "2026-07-16 00:00:00"}, + ] + stdout = "\n".join([json.dumps({"db_stats": db_stats}), *(json.dumps(row) for row in rows), ""]) + completed = subprocess.CompletedProcess(args=[], returncode=0, stdout=stdout, stderr="") + with mock.patch.object(report.subprocess, "run", return_value=completed) as run_mock: + result = report.fetch_rows_remote("pulse-license", "/opt/licenses.sqlite", 30) + self.assertEqual(result, {"db_stats": db_stats, "rows": rows}) + remote_script = run_mock.call_args.kwargs["input"] + self.assertNotIn("fetchall", remote_script) + + def test_fetch_rows_remote_rejects_empty_response(self) -> None: + completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="\n", stderr="") + with mock.patch.object(report.subprocess, "run", return_value=completed): + with self.assertRaisesRegex(RuntimeError, "empty response"): + report.fetch_rows_remote("pulse-license", "/opt/licenses.sqlite", 30) + def test_classify_reported_version_requires_real_published_tag(self) -> None: identity = report.classify_reported_version( "v6.0.0-rc.2",