fix(telemetry-report): stream remote rows as JSON-lines to avoid droplet OOM

fetch_rows_remote loaded every row in the window into a list and
json.dumps'd the whole payload on the remote side; on the 1GB
pulse-license droplet a 30-day window over a 104k-row DB got the
process OOM-killed. The remote helper now prints a db_stats header
line and then one JSON line per row while iterating the cursor, so
it never holds the full result set. Local parsing updated to match
and covered by unit tests; fetch_rows_local is unchanged.

Verified end to end against the droplet with --since-days 30.

Contract-Neutral: fetch_rows_remote OOM fix: stream JSON-lines instead of one giant payload; no contract delta, report output unchanged
This commit is contained in:
rcourtman 2026-07-17 23:17:14 +01:00
parent 9e88c0dfcb
commit 1e267566af
2 changed files with 34 additions and 6 deletions

View file

@ -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]]:

View file

@ -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",