Skip to content
Draft
65 changes: 63 additions & 2 deletions functions-python/tasks_executor/src/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,18 @@
notifications_dispatch_monitor_handler,
)
from tasks.changelog.backfill_changelog import backfill_changelog_handler
from tasks.seal_of_reliability.backfill.backfill_seal_of_reliability import (
backfill_seal_of_reliability_handler,
)

from tasks.seal_of_reliability.backfill.seal_backfill_orchestrator import (
seal_backfill_orchestrator_handler,
)

from tasks.seal_of_reliability.backfill.seal_backfill_worker import (
seal_backfill_worker_handler,
)

from tasks.seal_of_reliability.update_seal_of_reliability import (
update_seal_of_reliability_handler,
)
Expand Down Expand Up @@ -302,6 +314,53 @@
),
"handler": update_seal_of_reliability_handler,
},
"backfill_seal_of_reliability": {
"description": (
"Establishes a starting Seal of Reliability state for feeds that have none "
"(issue #1763), by cold-starting each feed 12 months back and replaying the "
"nightly evaluation forward one day at a time to end_date, writing only the "
"final day. The intermediate days are held in memory and discarded unless "
"snapshot_mode says otherwise. "
"Parameters: stable_feed_ids (required, non-empty), start_date (ISO date, "
"default end_date minus days_back; clamped up to each feed's created_at), "
"end_date (ISO date, default yesterday UTC), days_back (default 365), "
"dry_run (default true), limit (default null), criteria (default null "
"meaning every implemented criterion), batch_size (default 200), "
"only_missing (default true; skips feeds that already have seal state), "
"snapshot_mode (final|all|none, default final), resume_from_snapshot "
"(default false; the #1803 hook), max_reported_feeds (default 50)."
),
"handler": backfill_seal_of_reliability_handler,
},
"seal_backfill_orchestrator": {
"description": (
"Cloud Tasks producer for the Seal of Reliability backfill across the whole "
"catalog (issue #1763). Resolves every seal-eligible GTFS feed that has no "
"seal state yet, chunks it, registers a run in TaskExecutionTracker (feeds "
"DB), and enqueues one 'seal_backfill_worker' task per batch plus a single "
"'seal_orchestrator_monitor' barrier task. The window is resolved here once "
"and passed to every worker, so all batches of a run end on the same day. "
"Parameters: dry_run (default true), batch_size (default 100), start_date "
"(ISO date, default end_date minus days_back), end_date (ISO date, default "
"yesterday UTC), days_back (default 365), criteria (default null), limit "
"(default null), stable_feed_ids (restrict eligibility to these ids, default "
"null), only_missing (default true), snapshot_mode (final|all|none, default "
"final), resume_from_snapshot (default false), deadline_seconds (default "
"7200), monitor_delay_seconds (default 300)."
),
"handler": seal_backfill_orchestrator_handler,
},
"seal_backfill_worker": {
"description": (
"Cloud Tasks worker: march one batch's worth of feeds for the Seal of "
"Reliability backfill and report completion/failure to TaskExecutionTracker. "
"Parameters: run_id (required), batch_id (required), stable_feed_ids "
"(required, non-empty), start_date (required, ISO date), end_date (required, "
"ISO date), criteria (default null), only_missing (default true), "
"snapshot_mode (default final), resume_from_snapshot (default false)."
),
"handler": seal_backfill_worker_handler,
},
"seal_orchestrator": {
"description": (
"Cloud Tasks producer for the nightly Seal of Reliability run across the "
Expand Down Expand Up @@ -333,8 +392,10 @@
"batch of a seal orchestrator run has reported, or the run's "
"deadline_seconds passes, then aggregates each batch's report and marks "
"the run completed (every batch succeeded) or failed (any batch failed, "
"or the deadline was reached with batches still unaccounted for). "
"Parameters: run_id (required)."
"or the deadline was reached with batches still unaccounted for). Settles "
"both the nightly fan-out and the backfill fan-out; task_name selects which. "
"Parameters: run_id (required), task_name (default 'seal_orchestrator_run'; "
"pass 'seal_backfill_run' for a backfill run)."
),
"handler": seal_orchestrator_monitor_handler,
},
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
#
# MobilityData 2026
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Task entry point for the Seal of Reliability backfill (issue #1763)."""

from datetime import date, datetime
from typing import Optional

from tasks.seal_of_reliability.backfill.seal_backfill import (
DEFAULT_DAYS_BACK,
DEFAULT_SNAPSHOT_MODE,
backfill_seals,
)
from tasks.seal_of_reliability.seal_updater import (
DEFAULT_BATCH_SIZE,
DEFAULT_MAX_REPORTED_FEEDS,
)


def _parse_day(value: Optional[str], field: str) -> Optional[date]:
"""Parse a payload date string to a `date`, or None when absent.

Accepts a full timestamp too — an operator pasting the nightly task's `now` should not
hit a parse error. The march is day-granular, so the time is dropped either way.
"""
if value is None:
return None
try:
return date.fromisoformat(value)
except ValueError:
pass
try:
return datetime.fromisoformat(value).date()
except ValueError:
raise ValueError(
f"{field} must be an ISO date such as 2026-01-31, got {value!r}"
)


def get_parameters(payload: dict):
"""Read the task parameters from the payload, applying defaults."""
payload = payload or {}
return (
payload.get("stable_feed_ids"),
_parse_day(payload.get("start_date"), "start_date"),
_parse_day(payload.get("end_date"), "end_date"),
payload.get("days_back", DEFAULT_DAYS_BACK),
payload.get("dry_run", True),
payload.get("limit", None),
payload.get("criteria", None),
payload.get("batch_size", DEFAULT_BATCH_SIZE),
payload.get("only_missing", True),
payload.get("snapshot_mode", DEFAULT_SNAPSHOT_MODE),
payload.get("resume_from_snapshot", False),
payload.get("max_reported_feeds", DEFAULT_MAX_REPORTED_FEEDS),
payload.get("simulate", None),
payload.get("trace", False),
)


def backfill_seal_of_reliability_handler(payload: dict) -> dict:
"""Handler for the Seal of Reliability backfill. A dry run returns the plan only.

Payload, all optional but `stable_feed_ids`:
stable_feed_ids required, non-empty; there is no run-the-whole-catalogue mode
start_date ISO date, clamped up to each feed's created_at. Default: end_date
minus days_back
end_date ISO date, last day simulated. Default: yesterday UTC
days_back window length when start_date is absent. Default: 365
dry_run Default: True
limit cap the number of feeds. Default: no limit
criteria restrict to these criteria. Default: every implemented one
batch_size Default: 200
only_missing skip feeds that already have seal state. Default: True
snapshot_mode final | all | none. Default: final
resume_from_snapshot seed from the snapshot before march_start (#1803).
Default: False
max_reported_feeds cap on the `feeds` list in the response. Default: 50
simulate force statuses per criterion, on days counted from each feed's
march start: {"fresh_coverage": {"default": "pass", "fail": [3]}}.
Requires dry_run; see `parse_simulation` for the full shape
trace return the march day by day: every seal_criterion field, plus where
it came from. Marches without writing when dry_run. Consecutive days
in which nothing changed are always collapsed into one entry — its
first day, its last, and the count between
"""
(
stable_feed_ids,
start_date,
end_date,
days_back,
dry_run,
limit,
criteria,
batch_size,
only_missing,
snapshot_mode,
resume_from_snapshot,
max_reported_feeds,
simulate,
trace,
) = get_parameters(payload)
return backfill_seals(
stable_feed_ids=stable_feed_ids,
start_date=start_date,
end_date=end_date,
days_back=days_back,
dry_run=dry_run,
limit=limit,
criteria=criteria,
batch_size=batch_size,
only_missing=only_missing,
snapshot_mode=snapshot_mode,
resume_from_snapshot=resume_from_snapshot,
max_reported_feeds=max_reported_feeds,
simulate=simulate,
trace=trace,
)
Loading
Loading