refactor: Share digest schedule semantics across clouds

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,
+    )

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:

python/tests/test_azure_timer_digest.py

@@ -0,0 +1,75 @@
+"""Boundary regressions for the Azure timer adapter."""
+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
+
+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)