From 0a3fc7b96a70e99a780dba93fd972d0940963c7d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:49:16 +0800 Subject: [PATCH 1/2] refactor: Share digest schedule semantics across clouds Co-Authored-By: Codex --- python/scripts/azure_timer_digest.py | 65 +++++++++++++++++++++ python/scripts/cloud_run_daily_digest.py | 21 ++++++- python/tests/test_azure_timer_digest.py | 72 ++++++++++++++++++++++++ 3 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 python/scripts/azure_timer_digest.py create mode 100644 python/tests/test_azure_timer_digest.py diff --git a/python/scripts/azure_timer_digest.py b/python/scripts/azure_timer_digest.py new file mode 100644 index 00000000..98c59f1a --- /dev/null +++ b/python/scripts/azure_timer_digest.py @@ -0,0 +1,65 @@ +"""Azure timer adapter for the existing digest and GCS delivery ledger.""" +from datetime import datetime, time, timedelta + +import cloud_run_daily_digest as core + + +class ScheduleUnavailable(Exception): + """The timer did not supply a valid original slot.""" + + +def validate_slot(scheduled, now): + try: + if scheduled.tzinfo is None or scheduled > now: + raise ValueError() + local = scheduled.astimezone(core.SGT) + if local.time() not in (time(7), time(7, 30)): + raise ValueError() + except (AttributeError, TypeError, ValueError): + raise ScheduleUnavailable('invalid_original_slot') from None + report_date, _, check = core.business_slot(scheduled) + return report_date, check, scheduled + + +def resolve_slot(schedule_status, now): + try: + # Next is the persisted original slot, including a past-due startup. + scheduled = datetime.fromisoformat(schedule_status['Next']) + except (KeyError, TypeError, ValueError): + raise ScheduleUnavailable('invalid_original_slot') from None + return validate_slot(scheduled, now) + + +def window_slot(begin, end, seconds, now): + try: + first, last = datetime.fromisoformat(begin), datetime.fromisoformat(end) + if (first.tzinfo is None or last.tzinfo is None or first.second or first.microsecond + or last.second or last.microsecond or last > now + or last - first != timedelta(seconds=seconds)): + raise ValueError() + except (ValueError, TypeError): + raise ValueError('unverified_sliding_window') from None + return last + + +def execute_timer(schedule_status, configuration, store, send, *, now=core.utc_now, platform_config=None, + provenance='azure_functions_timer'): + _, _, scheduled = resolve_slot(schedule_status, now()) + return _execute_validated(scheduled, configuration, store, send, now=now, + platform_config=platform_config, provenance=provenance) + + +def execute_scheduled(scheduled, configuration, store, send, *, now=core.utc_now, platform_config=None, + provenance='azure_functions_timer'): + validate_slot(scheduled, now()) + return _execute_validated(scheduled, configuration, store, send, now=now, + platform_config=platform_config, provenance=provenance) + + +def _execute_validated(scheduled, configuration, store, send, *, now, platform_config, provenance): + report_date, day, check = core.business_slot(scheduled) + return core.execute_slot( + report_date, day, check, + {'scheduled_at': scheduled.isoformat(), 'trigger_provenance': provenance}, + configuration, store, send, now=now, platform_config=platform_config, + ) diff --git a/python/scripts/cloud_run_daily_digest.py b/python/scripts/cloud_run_daily_digest.py index 07585287..4f9faf65 100644 --- a/python/scripts/cloud_run_daily_digest.py +++ b/python/scripts/cloud_run_daily_digest.py @@ -121,6 +121,13 @@ def utc_now(): return datetime.now(UTC) +def business_slot(stamp): + """Derive the digest operation from an already validated trigger time.""" + local = stamp.astimezone(SGT) + report_date = local.date() + return report_date, (report_date - timedelta(days=1)).isoformat(), local.minute == 30 + + def report_slot(headers, configuration, now): values = {key.lower(): value for key, value in headers.items()} job = values.get('x-cloudscheduler-jobname') @@ -136,8 +143,8 @@ def report_slot(headers, configuration, now): local = stamp.astimezone(SGT) if (local.hour, local.minute) not in {(7, 0), (7, 30)} or stamp > now: raise ValueError('digest_schedule_slot_invalid') - report_date = local.date() - return report_date, (report_date - timedelta(days=1)).isoformat(), stamp, local.minute == 30 + report_date, day, check = business_slot(stamp) + return report_date, day, stamp, check def _sources(configuration, store, day, end, *, now, monotonic): @@ -211,6 +218,14 @@ def execute(headers, configuration, store, send, *, preview=False, now=utc_now, reason = str(error) if isinstance(error, ValueError) and str(error) in known else ( 'digest_schedule_header_missing' if isinstance(error, KeyError) else 'digest_schedule_header_invalid') return 400, {'delivery': 'invalid_slot', 'reason_code': reason} + return execute_slot(report_date, day, check, {'scheduled_at': scheduled_at.isoformat()}, + configuration, store, send, preview=preview, now=now, + monotonic=monotonic, platform_config=platform_config) + + +def execute_slot(report_date, day, check, schedule_record, configuration, store, send, *, + preview=False, now=utc_now, monotonic=clock.monotonic, platform_config=None): + """Run the unchanged business logic after the calling platform validates its trigger.""" safe = {'report_date': report_date.isoformat(), 'business_day': day, 'delivery': 'not_attempted'} if not preview and now().astimezone(SGT).date() != report_date: return 200, {**safe, 'delivery': 'expired_slot'} @@ -250,7 +265,7 @@ def execute(headers, configuration, store, send, *, preview=False, now=utc_now, key = f'{configuration.recipient_scope}/{report_date.isoformat()}.json' record = {**safe, 'schema_version': 'qsl.daily_digest_delivery.v1', 'delivery': 'sending', 'recipient_scope': configuration.recipient_scope, 'attempt_id': str(uuid4()), - 'scheduled_at': scheduled_at.isoformat(), 'message_sha256': hashlib.sha256(text.encode()).hexdigest()} + **schedule_record, 'message_sha256': hashlib.sha256(text.encode()).hexdigest()} try: created, generation = store.reserve(key, record) except StoreUnavailable: diff --git a/python/tests/test_azure_timer_digest.py b/python/tests/test_azure_timer_digest.py new file mode 100644 index 00000000..1ce4f575 --- /dev/null +++ b/python/tests/test_azure_timer_digest.py @@ -0,0 +1,72 @@ +"""Boundary regressions for the Azure timer adapter.""" +from datetime import datetime +import unittest +from unittest.mock import Mock, patch + +import cloud_run_daily_digest as core +from azure_timer_digest import ScheduleUnavailable, execute_scheduled, execute_timer, resolve_slot, window_slot + +NOW = datetime.fromisoformat('2026-11-10T07:31:00+08:00') + + +class ScheduleTests(unittest.TestCase): + def test_next_is_original_slot_with_singapore_date(self): + schedule = {'Last': '2026-11-08T23:30:00Z', 'Next': '2026-11-09T23:00:00Z'} + send = resolve_slot(schedule, NOW) + schedule['Next'] = '2026-11-09T23:30:00Z' + check = resolve_slot(schedule, NOW) + self.assertEqual(send[0].isoformat(), '2026-11-10') + self.assertEqual(check[0], send[0]) + self.assertIs(send[1], False) + self.assertIs(check[1], True) + + def test_no_schedule_never_touches_ledger_or_sender(self): + store, send = Mock(), Mock() + variants = [{}, None, *({'Next': stamp} for stamp in ( + None, '2026-11-09T23:00:12Z', '2026-11-09T23:00:00', '2026-11-10T23:00:00Z', + ))] + for schedule in variants: + with self.subTest(schedule=schedule), self.assertRaises(ScheduleUnavailable): + execute_timer(schedule, Mock(), store, send, now=lambda: NOW) + self.assertEqual(store.mock_calls, []) + send.assert_not_called() + + def test_past_due_previous_day_expires_without_touching_ledger(self): + store, send = Mock(), Mock() + code, result = execute_timer( + {'Next': '2026-11-08T23:00:00Z'}, Mock(), store, send, now=lambda: NOW, + ) + self.assertEqual((code, result['delivery']), (200, 'expired_slot')) + self.assertEqual(store.mock_calls, []) + send.assert_not_called() + + def test_direct_window_timestamp_preserves_slot_and_provenance(self): + stamp = datetime.fromisoformat('2026-11-09T23:30:00+00:00') + with patch.object(core, 'execute_slot', return_value=(200, {})) as execute: + execute_scheduled(stamp, Mock(), Mock(), Mock(), now=lambda: NOW, + provenance='azure_logic_sliding_window') + args = execute.call_args.args + self.assertEqual((args[0].isoformat(), args[1], args[2]), ('2026-11-10', '2026-11-09', True)) + self.assertEqual(args[3], {'scheduled_at': stamp.isoformat(), + 'trigger_provenance': 'azure_logic_sliding_window'}) + + def test_google_retains_subminute_tolerance_while_azure_rejects_it(self): + stamp = '2026-11-09T23:00:12.123456+00:00' + config = Mock(scheduler_job='projects/p/locations/l/jobs/j') + result = core.report_slot({'X-CloudScheduler-JobName': 'j', + 'X-CloudScheduler-ScheduleTime': stamp}, config, NOW) + self.assertEqual((result[0].isoformat(), result[1], result[3]), + ('2026-11-10', '2026-11-09', False)) + with self.assertRaises(ScheduleUnavailable): + resolve_slot({'Next': stamp}, NOW) + + def test_container_window_supplies_original_time_at_the_boundary(self): + start, end = '2026-11-08T23:00:00Z', '2026-11-09T23:00:00Z' + stamp = window_slot(start, end, 86400, NOW) + self.assertEqual(stamp.isoformat(), '2026-11-09T23:00:00+00:00') + for bad_start, bad_end in ((None, end), (start, '2026-11-09T23:00:01Z'), + (start, '2026-11-10T23:00:00Z')): + with self.subTest(start=bad_start, end=bad_end), self.assertRaisesRegex( + ValueError, 'unverified_sliding_window' + ): + window_slot(bad_start, bad_end, 86400, NOW) From afbbe3f55c5299c50561b774a1a75225a84e1e22 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:11:47 +0800 Subject: [PATCH 2/2] test: Fix digest adapter test discovery Co-Authored-By: Codex --- python/tests/test_azure_timer_digest.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/python/tests/test_azure_timer_digest.py b/python/tests/test_azure_timer_digest.py index 1ce4f575..4e7b1ee5 100644 --- a/python/tests/test_azure_timer_digest.py +++ b/python/tests/test_azure_timer_digest.py @@ -1,8 +1,11 @@ """Boundary regressions for the Azure timer adapter.""" -from datetime import datetime +import sys import unittest +from datetime import datetime +from pathlib import Path from unittest.mock import Mock, patch +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import cloud_run_daily_digest as core from azure_timer_digest import ScheduleUnavailable, execute_scheduled, execute_timer, resolve_slot, window_slot