diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index 17cbc22..bbd9c65 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -115,6 +115,12 @@ or access Secret Manager. Archive reads use the Storage SDK and the job's attached identity; the image does not need gcloud. Existing candidate writes, merge semantics and identity checks remain authoritative. +The heartbeat candidate publisher uses the same bounded SDK archive reader. +A successfully listed empty month is empty input, including when a completed +window spans two UTC months. Permission, timeout and JSON failures still abort +publication. An empty window produces idle evidence only after the existing +schedule check confirms that no run was due. + Use one job per existing central target, with its exact protected runtime JSON, archive prefix, opaque UID, target ID, destination and slot. No identifiers can be inferred from a display label. Supply the existing notification schedule so diff --git a/notifications/digest_snapshot/digest_archive.py b/notifications/digest_snapshot/digest_archive.py index 5202da9..2665e01 100644 --- a/notifications/digest_snapshot/digest_archive.py +++ b/notifications/digest_snapshot/digest_archive.py @@ -8,7 +8,6 @@ import json import os import re -import subprocess from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo @@ -17,11 +16,49 @@ _SAFE = re.compile(r"[a-z0-9][a-z0-9_-]{0,40}\Z") +class ArchiveReader: + """Native Storage reads with the job identity; no gcloud or secret access.""" + + def __init__(self, client): + self.client = client + + def read(self, args): + uri = args[-1] + if not uri.startswith("gs://"): + raise ValueError("candidate_archive_uri_invalid") + bucket, separator, key = uri[5:].partition("/") + if not bucket or not separator or not key: + raise ValueError("candidate_archive_uri_invalid") + if args[0] == "ls": + blobs = self.client.list_blobs( + bucket, prefix=key.split("*", 1)[0], match_glob=key, + max_results=10001, retry=None, timeout=30, + ) + rows = [] + for blob in blobs: + rows.append({"url": f"gs://{bucket}/{blob.name}", + "metadata": {"updated": blob.updated.isoformat()}}) + if len(rows) > 10000: + raise ValueError("candidate_archive_budget_exceeded") + return rows + if args[0] != "cat": + raise ValueError("candidate_archive_operation_invalid") + blob = self.client.bucket(bucket).blob(key) + blob.reload(retry=None, timeout=30) + if blob.size is None or blob.size > 2 * 1024**2: + raise ValueError("candidate_archive_object_too_large") + return json.loads(blob.download_as_bytes( + if_generation_match=int(blob.generation), retry=None, timeout=30, + )) + + def _read(args): - result = subprocess.run(["gcloud", "storage", *args], capture_output=True, text=True, check=False) - if result.returncode: - raise ValueError("candidate_archive_read_failed") - return json.loads(result.stdout) + from google.cloud import storage + try: + return ArchiveReader(storage.Client()).read(args) + except Exception as error: + # Empty SDK listings return []; all failed reads remain failures. + raise ValueError("candidate_archive_read_failed") from error def emit(env, *, now=None, read=_read, upload=None): diff --git a/scripts/publish_digest_before_read.py b/scripts/publish_digest_before_read.py index 90f1881..941743c 100644 --- a/scripts/publish_digest_before_read.py +++ b/scripts/publish_digest_before_read.py @@ -8,6 +8,8 @@ from datetime import datetime, time, timedelta, timezone from zoneinfo import ZoneInfo +from notifications.digest_snapshot.digest_archive import ArchiveReader + SHANGHAI = ZoneInfo("Asia/Shanghai") PLATFORM = "firstrade" @@ -25,42 +27,6 @@ def publication_day(now): return (local.date() - timedelta(days=1)).isoformat() -class ArchiveReader: - """Native Storage reads with the job identity; no gcloud or secret access.""" - - def __init__(self, client): - self.client = client - - def read(self, args): - uri = args[-1] - if not uri.startswith("gs://"): - raise ValueError("candidate_archive_uri_invalid") - bucket, separator, key = uri[5:].partition("/") - if not bucket or not separator or not key: - raise ValueError("candidate_archive_uri_invalid") - if args[0] == "ls": - blobs = self.client.list_blobs( - bucket, prefix=key.split("*", 1)[0], match_glob=key, - max_results=10001, retry=None, timeout=30, - ) - rows = [] - for blob in blobs: - rows.append({"url": f"gs://{bucket}/{blob.name}", - "metadata": {"updated": blob.updated.isoformat()}}) - if len(rows) > 10000: - raise ValueError("candidate_archive_budget_exceeded") - return rows - if args[0] != "cat": - raise ValueError("candidate_archive_operation_invalid") - blob = self.client.bucket(bucket).blob(key) - blob.reload(retry=None, timeout=30) - if blob.size is None or blob.size > 2 * 1024**2: - raise ValueError("candidate_archive_object_too_large") - return json.loads(blob.download_as_bytes( - if_generation_match=int(blob.generation), retry=None, timeout=30, - )) - - def _emit(env, *, now, read): from notifications.unified import emit_candidates return emit_candidates(env, now=now, read=read) diff --git a/tests/test_heartbeat_archive_reader.py b/tests/test_heartbeat_archive_reader.py new file mode 100644 index 0000000..b294b38 --- /dev/null +++ b/tests/test_heartbeat_archive_reader.py @@ -0,0 +1,103 @@ +"""Heartbeat uses bounded SDK archive reads, including empty month partitions.""" +from datetime import datetime, timezone +import json +import subprocess + +import pytest +from google.api_core.exceptions import Forbidden + +from notifications import unified +from test_publish_digest_before_read import ENV, PLATFORM +import test_publish_digest_before_read as prepub + +client = prepub.client + +NOW = datetime(2026, 11, 2, 1, 35, tzinfo=timezone.utc) +SOURCE_PLATFORM = "charles_schwab" if PLATFORM == "schwab" else PLATFORM + + +def emit(): + return unified.emit_candidates(ENV, now=NOW) + + +@pytest.fixture(autouse=True) +def no_cli(monkeypatch): + def fail(*args, **kwargs): + pytest.fail("heartbeat must use the Storage SDK") + monkeypatch.setattr(subprocess, "run", fail) + + +def candidate(client): + return client.blobs[f"gs://synthetic-candidates/digest/{PLATFORM}/2026-11-01/synthetic-slot.json"] + + +@pytest.mark.parametrize("other_month", [None, "2026-10", "2026-11"]) +def test_cross_month_empty_partition_reaches_not_due_check(client, other_month): + if other_month: + target = {**json.loads(ENV["RUNTIME_TARGET_JSON"]), "service_name": "synthetic-other"} + report = {"runtime_target": target, "started_at": "2026-11-01T20:00:00Z"} + client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/{other_month}/other.json").upload_from_string( + json.dumps(report), retry=None) + result = emit() + assert result["uploaded"] is True + rows = json.loads(candidate(client).body)["runs"] + assert len(rows) == 1 and rows[0]["actually_ran"] is False + assert rows[0]["scheduled_idle"] is True + assert rows[0]["business_day"] == "2026-11-01" + assert rows[0].get("order_count") is None and rows[0].get("fill_count") is None + assert [kwargs["match_glob"] for _, kwargs in client.lists] == [ + f"reports/{SOURCE_PLATFORM}/**/2026-10/*.json", + f"reports/{SOURCE_PLATFORM}/**/2026-11/*.json", + ] + assert all(kwargs["retry"] is None and kwargs["timeout"] == 30 for _, kwargs in client.lists) + + +def test_cross_month_empty_due_window_still_fails_without_writing(client, monkeypatch): + monkeypatch.setitem(ENV, "DIGEST_NOTIFICATION_SCHEDULE_JSON", json.dumps({"scheduler": { + "main_time": "45 15 * * *", "timezone": "America/New_York"}})) + with pytest.raises(ValueError, match="candidate_due_source_absent"): + emit() + assert not any(blob.writes for blob in client.blobs.values()) + + +@pytest.mark.parametrize("error", [Forbidden("synthetic-private"), TimeoutError("synthetic-private")]) +@pytest.mark.parametrize("failed_month", ["2026-10", "2026-11"]) +def test_listing_failure_is_never_empty_even_after_an_empty_month(client, monkeypatch, error, failed_month): + real_list = client.list_blobs + def listing(*args, **kwargs): + if failed_month in kwargs["match_glob"]: + def failed_page(): + raise error + yield + return failed_page() + return real_list(*args, **kwargs) + monkeypatch.setattr(client, "list_blobs", listing) + with pytest.raises(ValueError, match="candidate_archive_read_failed"): + emit() + assert not any(blob.writes for blob in client.blobs.values()) + + +def test_invalid_archive_json_remains_a_read_failure(client): + blob = client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/2026-10/run.json") + blob.upload_from_string('{"synthetic-private":', retry=None) + with pytest.raises(ValueError, match="candidate_archive_read_failed"): + emit() + assert all("/digest/" not in uri for uri in client.blobs) + + +def test_cross_month_report_in_october_is_preserved_when_november_is_empty(client): + from notifications.digest_snapshot.digest_candidates import cycle_evidence + target = json.loads(ENV["RUNTIME_TARGET_JSON"]) + report = {"schema_version": "runtime_report.v1", "platform": SOURCE_PLATFORM, + "strategy_profile": target["strategy_profile"], "service_name": target["service_name"], + "runtime_target": target, "run_id": "synthetic-run", "dry_run": False, + "status": "ok", "started_at": "2026-10-31T23:00:00+00:00", + "summary": {"notification_digest": cycle_evidence( + {"execution_status": "no_action", "broker_submission_done": False}, [])}} + client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/2026-10/run.json").upload_from_string( + json.dumps(report), retry=None) + assert emit()["uploaded"] is True + rows = json.loads(candidate(client).body)["runs"] + assert len(rows) == 1 and rows[0]["actually_ran"] is True + assert rows[0]["source_run_id"] == "synthetic-run" + assert rows[0]["business_day"] == "2026-11-01"