From 283745a98da9f342f492e77e97bd864bd1af1be1 Mon Sep 17 00:00:00 2001 From: Ketan Umare Date: Wed, 5 Aug 2026 12:15:42 -0700 Subject: [PATCH] Add `flyte backfill ` for scheduled triggers Expands a cron or fixed-rate trigger's schedule across a window, shows exactly which runs it would create, and on approval launches a driver run that creates them from inside the cluster (so a long backfill does not depend on the client staying connected, and the work is observable and abortable like any run). De-duplication is the point of the naming. A backfilled slot is created the way the scheduler creates a real fire -- addressed by TriggerName, named by the deterministic slot hash, and marked schedule-sourced -- so a slot that already ran is recognised and skipped, and re-running the same backfill is a no-op. `--force` salts the name into a separate namespace to re-run those slots. The run-name hash mirrors the scheduler's runName() exactly: FNV-1 (not FNV-1a) 64-bit over org:project:domain:task:trigger plus unpadded wall-clock fields, rendered as unpadded hex. Expected values in tests/backfill/test_naming.py were generated by running the Go implementation; cron expansion is likewise pinned against robfig/cron v3, the parser the scheduler uses. Co-Authored-By: Claude Opus 5 --- src/flyte/backfill/__init__.py | 33 ++++ src/flyte/backfill/_driver.py | 152 ++++++++++++++++++ src/flyte/backfill/_execute.py | 170 +++++++++++++++++++++ src/flyte/backfill/_naming.py | 101 ++++++++++++ src/flyte/backfill/_plan.py | 180 ++++++++++++++++++++++ src/flyte/backfill/_schedule.py | 191 +++++++++++++++++++++++ src/flyte/cli/_backfill.py | 262 ++++++++++++++++++++++++++++++++ src/flyte/cli/main.py | 4 +- tests/backfill/test_naming.py | 108 +++++++++++++ tests/backfill/test_plan.py | 132 ++++++++++++++++ tests/backfill/test_schedule.py | 92 +++++++++++ tests/cli/test_backfill.py | 108 +++++++++++++ 12 files changed, 1532 insertions(+), 1 deletion(-) create mode 100644 src/flyte/backfill/__init__.py create mode 100644 src/flyte/backfill/_driver.py create mode 100644 src/flyte/backfill/_execute.py create mode 100644 src/flyte/backfill/_naming.py create mode 100644 src/flyte/backfill/_plan.py create mode 100644 src/flyte/backfill/_schedule.py create mode 100644 src/flyte/cli/_backfill.py create mode 100644 tests/backfill/test_naming.py create mode 100644 tests/backfill/test_plan.py create mode 100644 tests/backfill/test_schedule.py create mode 100644 tests/cli/test_backfill.py diff --git a/src/flyte/backfill/__init__.py b/src/flyte/backfill/__init__.py new file mode 100644 index 000000000..e671c6118 --- /dev/null +++ b/src/flyte/backfill/__init__.py @@ -0,0 +1,33 @@ +"""Backfill a scheduled trigger. + +Re-runs the slots a cron or fixed-rate trigger would have fired over a window, +naming each run exactly the way a real fire would so that slots which already ran +are recognised and skipped -- unless the backfill is forced, which salts the names +into a separate namespace and re-runs them. + +The runs are created by a driver run inside the cluster, not from the client. See +``flyte backfill --help``. +""" + +from ._driver import backfill_driver, launch_backfill +from ._execute import execute_plan, probe_existing +from ._naming import candidate_run_names, scheduled_run_name +from ._plan import DEFAULT_MAX_RUNS, BackfillPlan, BackfillSlot, build_plan, occurrences_for +from ._schedule import CronParseError, cron_occurrences, fixed_rate_occurrences + +__all__ = [ + "DEFAULT_MAX_RUNS", + "BackfillPlan", + "BackfillSlot", + "CronParseError", + "backfill_driver", + "build_plan", + "candidate_run_names", + "cron_occurrences", + "execute_plan", + "fixed_rate_occurrences", + "launch_backfill", + "occurrences_for", + "probe_existing", + "scheduled_run_name", +] diff --git a/src/flyte/backfill/_driver.py b/src/flyte/backfill/_driver.py new file mode 100644 index 000000000..95dc0748f --- /dev/null +++ b/src/flyte/backfill/_driver.py @@ -0,0 +1,152 @@ +"""The run that performs a backfill. + +``flyte backfill`` does not create the backfilled runs from your machine. It +launches one small driver run that creates them from inside the cluster, so a +long backfill does not depend on a laptop staying connected, and so the work is +itself observable, retryable, and abortable like any other run. + +Every run the driver creates is linked back to it as a spawned child, so the +backfill shows up as one tree rather than a scatter of unrelated runs. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from flyte.remote import Run + + from ._plan import BackfillPlan + +__all__ = ["backfill_driver", "encode_plan", "launch_backfill"] + + +@dataclass +class _EncodedPlan: + """The approved plan, in the form the driver receives it. + + The exact slot list is carried across rather than recomputed, so the driver + creates precisely what was shown and approved -- no re-expansion, no drift if + the trigger's schedule changes in between. + """ + + org: str + project: str + domain: str + task_name: str + trigger_name: str + force: bool + salt: str | None + slots: list[dict] + + +def encode_plan(plan: "BackfillPlan") -> str: + payload = _EncodedPlan( + org=plan.org, + project=plan.project, + domain=plan.domain, + task_name=plan.task_name, + trigger_name=plan.trigger_name, + force=plan.force, + salt=plan.salt, + slots=[ + { + "at": s.scheduled_at.isoformat(), + "name": s.run_name, + "already_ran": s.already_ran, + } + for s in plan.to_create + ], + ) + return json.dumps(payload.__dict__) + + +def _decode_plan(encoded: str) -> tuple["BackfillPlan", list]: + from ._plan import BackfillPlan, BackfillSlot + + raw = json.loads(encoded) + slots = [ + BackfillSlot( + scheduled_at=datetime.fromisoformat(s["at"]), + run_name=s["name"], + already_ran=s["already_ran"], + ) + for s in raw["slots"] + ] + plan = BackfillPlan( + trigger_name=raw["trigger_name"], + task_name=raw["task_name"], + project=raw["project"], + domain=raw["domain"], + org=raw["org"], + schedule="", + start=slots[0].scheduled_at if slots else datetime.now(timezone.utc), + end=slots[-1].scheduled_at if slots else datetime.now(timezone.utc), + force=raw["force"], + salt=raw["salt"], + queue=None, + max_runs=len(slots), + slots=slots, + ) + return plan, slots + + +async def backfill_driver(encoded_plan: str) -> str: + """Create every run in the encoded plan. Runs inside the cluster. + + Returns a short summary. Individual slot failures are reported rather than + raised, so one rejected slot does not abandon the rest of the backfill. + """ + from flyte.remote import Trigger + + from ._execute import execute_plan + + plan, _ = _decode_plan(encoded_plan) + details = await Trigger.get.aio(name=plan.trigger_name, task_name=plan.task_name) + + created = 0 + deduped = 0 + failed: list[str] = [] + + def _record(result) -> None: + nonlocal created, deduped + if result.error: + failed.append(f"{result.slot.scheduled_at.isoformat()}: {result.error}") + elif result.created: + created += 1 + else: + deduped += 1 + + await execute_plan(plan, details, on_slot=_record) + + summary = f"backfill {plan.trigger_name}: {created} created, {deduped} already existed, {len(failed)} failed" + if failed: + summary += "\n" + "\n".join(failed[:20]) + print(summary, flush=True) + return summary + + +def launch_backfill(plan: "BackfillPlan", name: str | None = None) -> "Run": + """Launch the driver run that performs ``plan``.""" + import flyte + + env = flyte.TaskEnvironment( + name="flyte-backfill", + resources=flyte.Resources(cpu=1, memory="500Mi"), + ) + driver = env.task(backfill_driver) + run = flyte.with_runcontext( + mode="remote", + name=name, + # The driver has no local source file of its own; ship it as a code bundle. + interactive_mode=True, + queue=plan.queue, + ).run(driver, encode_plan(plan)) + from typing import cast + + from flyte.remote import Run as _Run + + return cast(_Run, run) diff --git a/src/flyte/backfill/_execute.py b/src/flyte/backfill/_execute.py new file mode 100644 index 000000000..01fc35f05 --- /dev/null +++ b/src/flyte/backfill/_execute.py @@ -0,0 +1,170 @@ +"""Create the runs a backfill plan describes. + +A backfilled slot is created the same way the scheduler creates a real fire: the +run is addressed by ``TriggerName`` (so the control plane resolves the task, +inputs and run spec from the trigger itself), named by the deterministic slot +hash, and marked as schedule-sourced. + +Using the schedule-trigger source matters. The control plane rewrites the run +name's prefix for automation-sourced runs, and de-duplication happens against the +rewritten name. Creating these runs under any other source would store them under +a different name than a real fire, and the de-duplication that makes backfill safe +to re-run would silently stop working. +""" + +from __future__ import annotations + +from datetime import timezone +from typing import TYPE_CHECKING, Callable + +if TYPE_CHECKING: + from ._plan import BackfillPlan, BackfillSlot + +__all__ = ["SlotResult", "execute_plan"] + + +class SlotResult: + """Outcome of creating one slot's run.""" + + __slots__ = ("created", "error", "run_name", "slot") + + def __init__(self, slot: "BackfillSlot", run_name: str, created: bool, error: str | None = None): + self.slot = slot + self.run_name = run_name + #: False when the control plane returned a pre-existing run instead of creating one. + self.created = created + self.error = error + + +def _kickoff_arg_name(details) -> str: + """The input variable a trigger binds its scheduled time to, if any.""" + schedule = details.pb2.automation_spec.schedule + return schedule.kickoff_time_input_arg or "" + + +def _build_request( + plan: "BackfillPlan", + slot: "BackfillSlot", + kickoff_arg: str, +): + from flyteidl2.common import identifier_pb2 + from flyteidl2.core import literals_pb2 + from flyteidl2.task import common_pb2 as task_common_pb2 + from flyteidl2.workflow import run_definition_pb2, run_service_pb2 + from google.protobuf import timestamp_pb2 + + scheduled = slot.scheduled_at + if scheduled.tzinfo is None: + scheduled = scheduled.replace(tzinfo=timezone.utc) + ts = timestamp_pb2.Timestamp() + ts.FromDatetime(scheduled.astimezone(timezone.utc)) + + req = run_service_pb2.CreateRunRequest( + run_id=identifier_pb2.RunIdentifier( + org=plan.org, + project=plan.project, + domain=plan.domain, + name=slot.run_name, + ), + trigger_name=identifier_pb2.TriggerName( + org=plan.org, + project=plan.project, + domain=plan.domain, + task_name=plan.task_name, + name=plan.trigger_name, + ), + source=run_definition_pb2.RUN_SOURCE_SCHEDULE_TRIGGER, + run_start_time=ts, + ) + + # Only triggers that bind their scheduled time to an input need the literal; + # everything else reads the time from run_start_time. + if kickoff_arg: + req.inputs.CopyFrom( + task_common_pb2.Inputs( + literals=[ + task_common_pb2.NamedLiteral( + name=kickoff_arg, + value=literals_pb2.Literal( + scalar=literals_pb2.Scalar( + primitive=literals_pb2.Primitive(datetime=ts), + ) + ), + ) + ] + ) + ) + return req + + +async def execute_plan( + plan: "BackfillPlan", + details, + on_slot: Callable[[SlotResult], None] | None = None, +) -> list[SlotResult]: + """Create a run for every slot in ``plan.to_create``. + + Slots the plan marked as already run are skipped unless the plan is forced. + Creating a run whose name already exists is not an error -- the control plane + returns the existing run -- so a re-run of the same backfill is a no-op. + """ + from flyte._initialize import ensure_client, get_client + + ensure_client() + kickoff_arg = _kickoff_arg_name(details) + results: list[SlotResult] = [] + + for slot in plan.to_create: + req = _build_request(plan, slot, kickoff_arg) + try: + resp = await get_client().run_service.create_run(req) + returned = resp.run.action.id.run.name or slot.run_name + # The server hands back the pre-existing run when the name is taken, + # which is the de-duplication path rather than a failure. + created = returned == slot.run_name or not slot.already_ran + result = SlotResult(slot, returned, created) + except Exception as exc: # surfaced per slot; one bad slot must not sink the rest + result = SlotResult(slot, slot.run_name, False, error=str(exc)) + results.append(result) + if on_slot is not None: + on_slot(result) + return results + + +async def probe_existing( + plan_names: list[str], +) -> set[str]: + """Return which of ``plan_names`` already exist as runs. + + Used to mark slots as already run before anything is created, so the preview + and the confirmation prompt reflect what will actually happen. + """ + from flyteidl2.workflow import run_service_pb2 + + from flyte._initialize import ensure_client, get_client + + ensure_client() + from flyte._initialize import get_init_config + + cfg = get_init_config() + found: set[str] = set() + for name in plan_names: + try: + from flyteidl2.common import identifier_pb2 + + await get_client().run_service.get_run_details( + run_service_pb2.GetRunDetailsRequest( + run_id=identifier_pb2.RunIdentifier( + org=cfg.org, + project=cfg.project, + domain=cfg.domain, + name=name, + ) + ) + ) + found.add(name) + except Exception: + # Anything other than a hit is treated as "not there". A lookup failure + # only costs us a redundant create, which the server de-duplicates. + continue + return found diff --git a/src/flyte/backfill/_naming.py b/src/flyte/backfill/_naming.py new file mode 100644 index 000000000..c79e0e690 --- /dev/null +++ b/src/flyte/backfill/_naming.py @@ -0,0 +1,101 @@ +"""Deterministic run names for scheduled trigger fires. + +A scheduled fire's run name is a hash of the trigger's identity plus the exact +second it was scheduled for. That is what makes fires idempotent: re-firing the +same slot produces the same name, and the control plane returns the existing run +instead of creating a second one. + +Backfill reuses that scheme so a backfilled slot collides with the slot the +scheduler already ran -- which is exactly the behaviour we want by default, and +what ``force`` opts out of by salting the name. + +The construction here mirrors the scheduler's ``runName`` byte for byte. Any +divergence silently breaks de-duplication (the backfill would create a parallel +run rather than recognising the existing one), so ``tests/backfill/test_naming.py`` +pins the expected hashes. +""" + +from __future__ import annotations + +from datetime import datetime + +__all__ = ["candidate_run_names", "fnv1_64", "scheduled_run_name"] + +_FNV1_64_OFFSET = 0xCBF29CE484222325 +_FNV1_64_PRIME = 0x100000001B3 +_UINT64_MASK = 0xFFFFFFFFFFFFFFFF + +# The scheduler emits names prefixed "r". For automation-sourced runs routed to +# the actions engine the control plane swaps that prefix to "u", keeping the hash. +# A client cannot see the routing decision, so both spellings are candidates when +# checking whether a slot has already run. +_SCHEDULER_PREFIX = "r" +_ACTIONS_PREFIX = "u" + + +def fnv1_64(data: bytes) -> int: + """FNV-1 (not FNV-1a) 64-bit hash, matching Go's ``hash/fnv.New64()``. + + Note the operation order: FNV-1 multiplies then XORs; FNV-1a is the reverse. + Using the wrong variant produces plausible-looking but non-matching names. + """ + h = _FNV1_64_OFFSET + for byte in data: + h = (h * _FNV1_64_PRIME) & _UINT64_MASK + h ^= byte + return h + + +def _identity( + org: str, + project: str, + domain: str, + task_name: str, + trigger_name: str, + at: datetime, + salt: str | None = None, +) -> str: + """Build the string that gets hashed. + + Time components are the *wall clock* fields of ``at`` in whatever timezone it + carries -- the scheduler hashes the schedule's local time, not UTC -- and are + formatted as unpadded integers. + """ + base = ( + f"{org}:{project}:{domain}:{task_name}:{trigger_name}:" + f"{at.year}:{at.month}:{at.day}:{at.hour}:{at.minute}:{at.second}" + ) + # A salt is prepended rather than appended, following the artifact-trigger + # naming precedent, so salted names occupy a disjoint namespace. + return f"{salt}:{base}" if salt else base + + +def scheduled_run_name( + org: str, + project: str, + domain: str, + task_name: str, + trigger_name: str, + at: datetime, + salt: str | None = None, +) -> str: + """Return the run name a scheduled fire of ``trigger_name`` at ``at`` produces. + + With ``salt`` set the name lands in a separate namespace, so it will not + collide with -- and therefore will not be de-duplicated against -- the run the + scheduler created for the same slot. That is how ``--force`` re-runs a slot. + """ + digest = fnv1_64(_identity(org, project, domain, task_name, trigger_name, at, salt).encode()) + return f"{_SCHEDULER_PREFIX}{digest:x}" + + +def candidate_run_names(name: str) -> tuple[str, ...]: + """Both spellings a scheduled run may be stored under. + + The control plane rewrites the leading "r" to "u" for automation-sourced runs + routed to the actions engine. The routing decision is not visible from a + client, so an existence check has to consider both. + """ + if name.startswith(_SCHEDULER_PREFIX): + return (name, _ACTIONS_PREFIX + name[1:]) + return (name,) diff --git a/src/flyte/backfill/_plan.py b/src/flyte/backfill/_plan.py new file mode 100644 index 000000000..fbf57eb13 --- /dev/null +++ b/src/flyte/backfill/_plan.py @@ -0,0 +1,180 @@ +"""Work out which runs a backfill would create, before creating any of them. + +The plan is built once and then either printed (``--dry-run``), shown for +confirmation, or handed to the driver to execute. Keeping it a plain data +structure means the CLI preview and the driver agree on exactly what will happen. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Sequence + +from ._naming import candidate_run_names, scheduled_run_name +from ._schedule import cron_occurrences, fixed_rate_occurrences + +if TYPE_CHECKING: + from flyte.remote import TriggerDetails + +__all__ = ["DEFAULT_MAX_RUNS", "FORCE_SALT_PREFIX", "BackfillPlan", "BackfillSlot", "build_plan"] + +# The console caps a backfill at this many runs; the CLI matches it so both +# surfaces behave the same. Wider ranges are a deliberate, explicit choice. +DEFAULT_MAX_RUNS = 100 + +# Prepended to salted (forced) names so they occupy a namespace that can never +# collide with a real scheduled fire. +FORCE_SALT_PREFIX = "bf1" + + +@dataclass(frozen=True) +class BackfillSlot: + """One scheduled time and the run it maps to.""" + + scheduled_at: datetime + run_name: str + #: True when a run for this slot already exists, i.e. the slot already fired. + already_ran: bool = False + + @property + def candidates(self) -> tuple[str, ...]: + return candidate_run_names(self.run_name) + + +@dataclass +class BackfillPlan: + """Everything a backfill will do, decided up front.""" + + trigger_name: str + task_name: str + project: str + domain: str + org: str + schedule: str + start: datetime + end: datetime + force: bool + salt: str | None + queue: str | None + max_runs: int + slots: list[BackfillSlot] = field(default_factory=list) + #: Slots dropped because the plan hit ``max_runs``. + truncated: int = 0 + + @property + def to_create(self) -> list[BackfillSlot]: + return [s for s in self.slots if self.force or not s.already_ran] + + @property + def skipped(self) -> list[BackfillSlot]: + return [s for s in self.slots if not self.force and s.already_ran] + + @property + def overridden(self) -> list[BackfillSlot]: + """Slots that already ran and will be re-run because ``force`` is set.""" + return [s for s in self.slots if self.force and s.already_ran] + + +def _schedule_expression(details: "TriggerDetails") -> tuple[str, str | None, int | None, datetime | None]: + """Pull the schedule out of a trigger. + + Returns ``(human_expression, cron_expression, interval_minutes, rate_anchor)``. + Exactly one of ``cron_expression`` / ``interval_minutes`` is set. + """ + automation = details.pb2.automation_spec + schedule = automation.schedule + if schedule.HasField("cron"): + cron = schedule.cron + tz = cron.timezone or "UTC" + return (f"{cron.expression} ({tz})", cron.expression, None, None) + if schedule.cron_expression: # deprecated form, still accepted server-side + return (schedule.cron_expression, schedule.cron_expression, None, None) + if schedule.HasField("rate"): + rate = schedule.rate + # FixedRate.unit is an enum of MINUTE/HOUR/DAY; normalise to minutes. + unit_minutes = {0: 1, 1: 60, 2: 1440}.get(int(rate.unit), 1) + minutes = int(rate.value) * unit_minutes + anchor = rate.start_time.ToDatetime().replace(tzinfo=timezone.utc) if rate.HasField("start_time") else None + return (f"every {minutes}m", None, minutes, anchor) + raise ValueError(f"trigger {details.pb2.id.name!r} has no schedule -- only scheduled triggers can be backfilled") + + +def schedule_timezone(details: "TriggerDetails") -> str: + schedule = details.pb2.automation_spec.schedule + if schedule.HasField("cron") and schedule.cron.timezone: + return schedule.cron.timezone + return "UTC" + + +def occurrences_for( + details: "TriggerDetails", + start: datetime, + end: datetime, + limit: int | None = None, +) -> list[datetime]: + """Expand a trigger's schedule across ``[start, end]``.""" + _, cron_expr, interval, anchor = _schedule_expression(details) + if cron_expr is not None: + return cron_occurrences(cron_expr, start, end, limit) + assert interval is not None + return fixed_rate_occurrences(interval, start, end, anchor=anchor, limit=limit) + + +def build_plan( + *, + details: "TriggerDetails", + task_name: str, + org: str, + project: str, + domain: str, + start: datetime, + end: datetime, + force: bool = False, + suffix: str | None = None, + queue: str | None = None, + max_runs: int = DEFAULT_MAX_RUNS, + existing: Sequence[str] | None = None, +) -> BackfillPlan: + """Build the plan for backfilling ``details`` across ``[start, end]``. + + ``existing`` is the set of run names already known to exist; slots naming one + of those are marked as already run (and skipped unless ``force``). + """ + if start >= end: + raise ValueError("start must be before end") + + trigger_name = details.pb2.id.name.name or details.pb2.id.name + human_schedule, _, _, _ = _schedule_expression(details) + salt = f"{FORCE_SALT_PREFIX}:{suffix}" if force and suffix else (FORCE_SALT_PREFIX if force else None) + + # Ask for one more than the cap so we can report how much was left out. + raw = occurrences_for(details, start, end, limit=max_runs + 1 if max_runs else None) + truncated = max(0, len(raw) - max_runs) if max_runs else 0 + known = set(existing or ()) + + slots: list[BackfillSlot] = [] + for at in raw[:max_runs] if max_runs else raw: + name = scheduled_run_name(org, project, domain, task_name, trigger_name, at, salt=salt) + # Existence is always checked against the *unsalted* name -- that is the + # one a real scheduled fire would have produced. + unsalted = scheduled_run_name(org, project, domain, task_name, trigger_name, at) + already = any(c in known for c in candidate_run_names(unsalted)) + slots.append(BackfillSlot(scheduled_at=at, run_name=name, already_ran=already)) + + return BackfillPlan( + trigger_name=trigger_name, + task_name=task_name, + project=project, + domain=domain, + org=org, + schedule=human_schedule, + start=start, + end=end, + force=force, + salt=salt, + queue=queue, + max_runs=max_runs, + slots=slots, + truncated=truncated, + ) diff --git a/src/flyte/backfill/_schedule.py b/src/flyte/backfill/_schedule.py new file mode 100644 index 000000000..6d053cd8e --- /dev/null +++ b/src/flyte/backfill/_schedule.py @@ -0,0 +1,191 @@ +"""Expand a trigger's schedule into the individual times it fires. + +Backfill needs the same list of fire times the scheduler would have produced, so +it can name each slot the way a real fire would. Expansion happens in the +schedule's own timezone because the run name hashes local wall-clock fields. + +Only the five-field cron form is supported, which is what triggers accept: +``minute hour day-of-month month day-of-week``, each field being ``*``, a value, +a ``lo-hi`` range, a comma-separated list, or any of those with a ``/step``. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta +from typing import Iterator, Sequence + +__all__ = ["CronParseError", "cron_occurrences", "fixed_rate_occurrences", "parse_cron"] + +_CRON_TZ_PREFIX = re.compile(r"^\s*CRON_TZ=(?P\S+)\s+(?P.*)$") + +_FIELD_BOUNDS = ( + (0, 59), # minute + (0, 23), # hour + (1, 31), # day of month + (1, 12), # month + (0, 6), # day of week, Sunday = 0 +) + +_MONTH_ALIASES = { + m: i + 1 for i, m in enumerate(["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"]) +} +_DOW_ALIASES = {m: i for i, m in enumerate(["sun", "mon", "tue", "wed", "thu", "fri", "sat"])} + + +class CronParseError(ValueError): + """Raised when a cron expression cannot be understood.""" + + +def _alias(token: str, index: int) -> str: + lowered = token.lower() + if index == 3 and lowered in _MONTH_ALIASES: + return str(_MONTH_ALIASES[lowered]) + if index == 4 and lowered in _DOW_ALIASES: + return str(_DOW_ALIASES[lowered]) + return token + + +def _parse_field(raw: str, index: int) -> frozenset[int] | None: + """Parse one cron field. ``None`` means "unrestricted" (a bare ``*``).""" + lo_bound, hi_bound = _FIELD_BOUNDS[index] + if raw == "*": + return None + + values: set[int] = set() + for part in raw.split(","): + body, _, step_raw = part.partition("/") + try: + step = int(step_raw) if step_raw else 1 + except ValueError as exc: + raise CronParseError(f"invalid step in cron field {raw!r}") from exc + if step < 1: + raise CronParseError(f"step must be positive in cron field {raw!r}") + + if body == "*": + lo, hi = lo_bound, hi_bound + else: + bits = [_alias(b, index) for b in body.split("-")] + try: + lo = int(bits[0]) + hi = int(bits[1]) if len(bits) > 1 else (hi_bound if step_raw else lo) + except (ValueError, IndexError) as exc: + raise CronParseError(f"invalid range in cron field {raw!r}") from exc + + # Sunday is expressible as both 0 and 7; normalise 7 down. + if index == 4: + lo, hi = (0 if lo == 7 else lo), (0 if hi == 7 else hi) + if hi < lo: + lo, hi = hi, lo + if lo < lo_bound or hi > hi_bound or hi < lo: + raise CronParseError(f"cron field {raw!r} out of range {lo_bound}-{hi_bound}") + values.update(range(lo, hi + 1, step)) + + return frozenset(values) + + +def parse_cron(expression: str) -> tuple[tuple[frozenset[int] | None, ...], str | None]: + """Parse a cron expression into per-field value sets plus an optional timezone. + + Returns ``(fields, timezone)`` where ``fields`` is + ``(minute, hour, day_of_month, month, day_of_week)`` and each entry is either a + set of matching values or ``None`` for unrestricted. + """ + timezone_name = None + match = _CRON_TZ_PREFIX.match(expression) + if match: + timezone_name = match.group("tz") + expression = match.group("expr") + + parts = expression.split() + if len(parts) != 5: + raise CronParseError(f"expected a 5-field cron expression, got {len(parts)} field(s): {expression!r}") + return tuple(_parse_field(p, i) for i, p in enumerate(parts)), timezone_name + + +def _day_matches( + day: datetime, dom: frozenset[int] | None, month: frozenset[int] | None, dow: frozenset[int] | None +) -> bool: + if month is not None and day.month not in month: + return False + # Standard cron: when BOTH day-of-month and day-of-week are restricted, a day + # matches if EITHER does. Treating it as AND silently drops fire times. + py_dow = (day.weekday() + 1) % 7 # Python: Monday=0; cron: Sunday=0 + if dom is None and dow is None: + return True + if dom is not None and dow is not None: + return day.day in dom or py_dow in dow + if dom is not None: + return day.day in dom + return py_dow in dow + + +def cron_occurrences( + expression: str, + start: datetime, + end: datetime, + limit: int | None = None, +) -> list[datetime]: + """Every time ``expression`` fires in ``[start, end]``, inclusive. + + ``start`` and ``end`` are used as given -- pass times already localised to the + schedule's timezone, since the resulting datetimes are what get hashed into + run names. + """ + return list(_iter_cron(expression, start, end, limit)) + + +def _iter_cron(expression: str, start: datetime, end: datetime, limit: int | None) -> Iterator[datetime]: + fields, _ = parse_cron(expression) + minutes, hours, dom, month, dow = fields + minute_values: Sequence[int] = sorted(minutes) if minutes is not None else range(60) + hour_values: Sequence[int] = sorted(hours) if hours is not None else range(24) + + # Walk day by day and only expand hours/minutes on days that match, rather + # than testing every minute in the range. + day = start.replace(hour=0, minute=0, second=0, microsecond=0) + emitted = 0 + while day <= end: + if _day_matches(day, dom, month, dow): + for hour in hour_values: + for minute in minute_values: + moment = day.replace(hour=hour, minute=minute) + if moment < start: + continue + if moment > end: + return + yield moment + emitted += 1 + if limit is not None and emitted >= limit: + return + day += timedelta(days=1) + + +def fixed_rate_occurrences( + interval_minutes: int, + start: datetime, + end: datetime, + anchor: datetime | None = None, + limit: int | None = None, +) -> list[datetime]: + """Every fire time of a fixed-rate schedule in ``[start, end]``. + + ``anchor`` is the schedule's own start time, which sets the phase; without one + the window start is used. + """ + if interval_minutes < 1: + raise ValueError("interval_minutes must be at least 1") + step = timedelta(minutes=interval_minutes) + moment = anchor or start + if moment < start: + # Advance to the first tick at or after the window, without looping. + gap = (start - moment) // step + moment += step * gap + while moment < start: + moment += step + + out: list[datetime] = [] + while moment <= end and (limit is None or len(out) < limit): + out.append(moment) + moment += step + return out diff --git a/src/flyte/cli/_backfill.py b/src/flyte/cli/_backfill.py new file mode 100644 index 000000000..235a09a44 --- /dev/null +++ b/src/flyte/cli/_backfill.py @@ -0,0 +1,262 @@ +"""``flyte backfill `` -- re-run the slots a scheduled trigger missed. + +Expands the trigger's schedule across a window, shows exactly which runs it would +create, and on approval launches a driver run that creates them from inside the +cluster. + +Slots that already ran are skipped, because a backfilled run is named the same way +a real scheduled fire is. ``--force`` re-runs them under salted names instead. +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone +from typing import Optional + +import rich_click as click + +from . import _common as common + + +def _parse_time(value: str | None, *, what: str) -> datetime | None: + """Accept an ISO timestamp, a plain date, or a relative age like ``30d`` / ``12h``.""" + if not value: + return None + raw = value.strip() + if raw.lower() == "now": + return datetime.now(timezone.utc) + if len(raw) > 1 and raw[-1].lower() in "dhm" and raw[:-1].replace(".", "", 1).isdigit(): + amount = float(raw[:-1]) + unit = {"d": "days", "h": "hours", "m": "minutes"}[raw[-1].lower()] + return datetime.now(timezone.utc) - timedelta(**{unit: amount}) + try: + parsed = datetime.fromisoformat(raw) + except ValueError as exc: + raise click.BadParameter( + f"{what} must be an ISO timestamp (2026-05-01T02:00), a date (2026-05-01), " + f"or a relative age (30d, 12h). Got {value!r}." + ) from exc + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +@click.command("backfill", cls=click.RichCommand) +@click.argument("trigger_name", required=True) +@click.option("--task-name", default=None, help="Task the trigger belongs to. Looked up if omitted.") +@click.option("-p", "--project", default=None, help="Project (defaults to config).") +@click.option("-d", "--domain", default=None, help="Domain (defaults to config).") +@click.option( + "--from", + "start", + required=True, + help="Start of the window: ISO timestamp, a date, or a relative age such as 30d.", +) +@click.option("--to", "end", default=None, help="End of the window. Defaults to now.") +@click.option("--queue", default=None, help="Queue for the driver run. Backfilled runs use the trigger's own queue.") +@click.option( + "--force", + is_flag=True, + default=False, + help="Re-run slots that already ran. Names are salted so the new runs do not collide with the originals.", +) +@click.option("--suffix", default=None, help="With --force: extra salt distinguishing this backfill from earlier ones.") +@click.option( + "--max-runs", + default=None, + type=int, + help="Cap on runs created (default 100, matching the console).", +) +@click.option("--dry-run", is_flag=True, default=False, help="Show the plan and exit without creating anything.") +@click.option("-y", "--yes", is_flag=True, default=False, help="Skip the confirmation prompt.") +@click.option("--follow", "-f", is_flag=True, default=False, help="Stream the driver run's logs after launch.") +@click.pass_context +def backfill( + ctx: click.Context, + trigger_name: str, + task_name: Optional[str], + project: Optional[str], + domain: Optional[str], + start: str, + end: Optional[str], + queue: Optional[str], + force: bool, + suffix: Optional[str], + max_runs: Optional[int], + dry_run: bool, + yes: bool, + follow: bool, +) -> None: + """Backfill the scheduled trigger TRIGGER_NAME over a window of time. + + Every slot the schedule would have fired in the window becomes a run, named the + way a real fire is named. Slots that already ran are skipped, so running the + same backfill twice is a no-op. ``--force`` re-runs them under salted names. + + Examples: + + $ flyte backfill nightly_eval --from 30d + $ flyte backfill nightly_eval --from 2026-05-01 --to 2026-05-15 --dry-run + $ flyte backfill nightly_eval --from 7d --force --suffix rerun-2 + """ + if suffix and not force: + raise click.UsageError("--suffix requires --force") + started = _parse_time(start, what="--from") + ended = _parse_time(end, what="--to") or datetime.now(timezone.utc) + if started is None or started >= ended: + raise click.UsageError("--from must be earlier than --to") + + config = common.initialize_config(ctx, project=project, domain=domain) + asyncio.run( + _execute( + trigger_name=trigger_name, + task_name=task_name, + start=started, + end=ended, + queue=queue, + force=force, + suffix=suffix, + max_runs=max_runs, + dry_run=dry_run, + yes=yes, + follow=follow, + config=config, + ) + ) + + +async def _resolve_task_name(trigger_name: str, task_name: Optional[str]) -> str: + """Find which task a trigger belongs to, when the caller did not say.""" + if task_name: + return task_name + from flyte.remote import Trigger + + matches = [] + async for trig in await Trigger.listall.aio(): + if trig.name == trigger_name: + matches.append(trig) + if not matches: + raise click.ClickException( + f"No trigger named {trigger_name!r} in this project/domain. Pass --task-name to disambiguate." + ) + if len(matches) > 1: + tasks = ", ".join(sorted({m.task_name for m in matches})) + raise click.ClickException( + f"Several tasks have a trigger named {trigger_name!r} ({tasks}). Pass --task-name to choose one." + ) + return matches[0].task_name + + +def _render_plan(plan, console, output_format: str) -> None: + from rich.table import Table + + header = ( + f"Trigger {plan.trigger_name} → {plan.task_name}\n" + f"Schedule {plan.schedule}\n" + f"Window {plan.start.isoformat()} → {plan.end.isoformat()}\n" + f"Force {'on' + (f' (salt {plan.salt})' if plan.salt else '') if plan.force else 'off'}" + ) + console.print(common.get_panel("Backfill", header, output_format)) + + table = Table(show_header=True, header_style="bold") + table.add_column("Scheduled at") + table.add_column("Run name") + table.add_column("Action") + shown = plan.slots[:20] + for slot in shown: + if plan.force and slot.already_ran: + action = "[yellow]re-run (overrides existing)[/yellow]" + elif slot.already_ran: + action = "[dim]skip — already ran[/dim]" + else: + action = "[green]create[/green]" + table.add_row(slot.scheduled_at.isoformat(), slot.run_name, action) + console.print(table) + if len(plan.slots) > len(shown): + console.print(f"[dim]… and {len(plan.slots) - len(shown)} more[/dim]") + + console.print(f"\n[bold]{len(plan.to_create)}[/bold] run(s) to create, {len(plan.skipped)} skipped as already run.") + if plan.truncated: + console.print( + f"[yellow]{plan.truncated} further slot(s) fall outside the {plan.max_runs}-run cap " + f"and will not be created. Narrow the window or raise --max-runs.[/yellow]" + ) + + +async def _execute( + *, + trigger_name: str, + task_name: Optional[str], + start: datetime, + end: datetime, + queue: Optional[str], + force: bool, + suffix: Optional[str], + max_runs: Optional[int], + dry_run: bool, + yes: bool, + follow: bool, + config: common.CLIConfig, +) -> None: + from flyte._initialize import get_init_config + from flyte.backfill import DEFAULT_MAX_RUNS, build_plan + from flyte.backfill._driver import launch_backfill + from flyte.backfill._execute import probe_existing + from flyte.remote import Trigger + + console = common.get_console() + resolved_task = await _resolve_task_name(trigger_name, task_name) + details = await Trigger.get.aio(name=trigger_name, task_name=resolved_task) + cfg = get_init_config() + + # Build once without existence info to learn the candidate names, probe those, + # then rebuild so the displayed plan reflects what already ran. + def _build(existing): + return build_plan( + details=details, + task_name=resolved_task, + org=cfg.org or "", + project=cfg.project or "", + domain=cfg.domain or "", + start=start, + end=end, + force=force, + suffix=suffix, + queue=queue, + max_runs=max_runs or DEFAULT_MAX_RUNS, + existing=existing, + ) + + provisional = _build(None) + if not provisional.slots: + console.print("[yellow]No scheduled slots fall in that window — nothing to backfill.[/yellow]") + return + + with common.cli_status(config.output_format, "Checking which slots already ran..."): + candidates = [c for slot in provisional.slots for c in slot.candidates] + existing = await probe_existing(candidates) + plan = _build(existing) + + _render_plan(plan, console, config.output_format) + + if dry_run: + console.print("\n[dim]Dry run — nothing was created.[/dim]") + return + if not plan.to_create: + console.print("\n[dim]Every slot in this window has already run. Use --force to re-run them.[/dim]") + return + if not yes: + click.confirm(f"\nCreate {len(plan.to_create)} run(s)?", abort=True) + + run = launch_backfill(plan, name=None) + if config.output_format in ("json", "table-simple"): + info = f"Backfill run: {run.name}\nURL: {run.url}" + else: + info = ( + f"[green bold]Backfill run: {run.name}[/green bold]\n" + f"➡️ [blue bold][link={run.url}]{run.url}[/link][/blue bold]\n" + f"[dim]Creating {len(plan.to_create)} run(s) from inside the cluster.[/dim]" + ) + console.print(common.get_panel("Backfill", info, config.output_format)) + + if follow: + await run.show_logs.aio(max_lines=30, show_ts=True, raw=False) diff --git a/src/flyte/cli/main.py b/src/flyte/cli/main.py index 502d6d1b4..3df50648c 100644 --- a/src/flyte/cli/main.py +++ b/src/flyte/cli/main.py @@ -6,6 +6,7 @@ from . import _common as common from ._abort import abort +from ._backfill import backfill from ._build import build from ._common import CLIConfig from ._create import create @@ -32,7 +33,7 @@ "flyte": [ { "name": "Run and stop tasks", - "commands": ["run", "rerun", "abort", "signal"], + "commands": ["run", "rerun", "backfill", "abort", "signal"], }, { "name": "Serve Apps", @@ -285,6 +286,7 @@ def main( main.add_command(run) main.add_command(rerun) +main.add_command(backfill) main.add_command(deploy) main.add_command(get) # type: ignore main.add_command(create) # type: ignore diff --git a/tests/backfill/test_naming.py b/tests/backfill/test_naming.py new file mode 100644 index 000000000..809db7530 --- /dev/null +++ b/tests/backfill/test_naming.py @@ -0,0 +1,108 @@ +"""The run-name hash must match the scheduler's byte for byte. + +If it drifts, a backfilled slot no longer collides with the run the scheduler +already created, so de-duplication silently stops working and the backfill +quietly doubles up on work. The expected values below were produced by running +the scheduler's own Go implementation (hash/fnv New64 + the same format string), +so they pin the contract rather than the current Python behaviour. +""" + +from datetime import datetime, timezone + +import pytest + +from flyte.backfill._naming import candidate_run_names, fnv1_64, scheduled_run_name + +# (org, project, domain, task, trigger, (Y, M, D, h, m, s)) -> name from the Go scheduler +GO_REFERENCE = { + ( + "acme", + "ml-platform", + "production", + "evals.weekly.run", + "nightly_eval", + (2026, 5, 20, 2, 0, 0), + ): "r64640bf1ae650449", + ( + "acme", + "ml-platform", + "production", + "evals.weekly.run", + "nightly_eval", + (2026, 6, 1, 2, 0, 0), + ): "r5f16c7b4a799ddf7", + ("", "p", "d", "t", "n", (2026, 1, 1, 0, 0, 0)): "rf3c00da918b55b3", + ("o", "p", "d", "t", "n", (2026, 12, 31, 23, 59, 59)): "rae982f539aa14d00", + ( + "union", + "search", + "staging", + "search.rerank.rebuild", + "weekly_retraining", + (2026, 6, 1, 6, 0, 0), + ): "rd8c65d1b71ce83af", +} + + +@pytest.mark.parametrize(("key", "expected"), sorted(GO_REFERENCE.items())) +def test_matches_go_scheduler(key, expected): + org, project, domain, task, trigger, parts = key + at = datetime(*parts, tzinfo=timezone.utc) + assert scheduled_run_name(org, project, domain, task, trigger, at) == expected + + +def test_hash_is_fnv1_not_fnv1a(): + # FNV-1 and FNV-1a differ only in operation order; picking the wrong one still + # yields a plausible name, so assert against a known FNV-1 value. + assert fnv1_64(b"") == 0xCBF29CE484222325 + assert fnv1_64(b"a") == 0xAF63BD4C8601B7BE + # FNV-1a of "a" would be 0xAF63DC4C8601EC8C -- close enough to be worth pinning. + assert fnv1_64(b"a") != 0xAF63DC4C8601EC8C + + +def test_hex_is_unpadded(): + # Go's %x drops leading zeros, so names are not a fixed width. + name = scheduled_run_name("", "p", "d", "t", "n", datetime(2026, 1, 1, tzinfo=timezone.utc)) + assert name == "rf3c00da918b55b3" + assert len(name) == 16 # 'r' + 15 hex digits, i.e. one short of the usual 16 + + +def test_same_slot_is_stable(): + at = datetime(2026, 5, 20, 2, 0, tzinfo=timezone.utc) + first = scheduled_run_name("o", "p", "d", "t", "n", at) + second = scheduled_run_name("o", "p", "d", "t", "n", at) + assert first == second + + +def test_different_second_is_a_different_name(): + base = datetime(2026, 5, 20, 2, 0, 0, tzinfo=timezone.utc) + later = datetime(2026, 5, 20, 2, 0, 1, tzinfo=timezone.utc) + assert scheduled_run_name("o", "p", "d", "t", "n", base) != scheduled_run_name("o", "p", "d", "t", "n", later) + + +def test_salt_moves_the_name_into_a_separate_namespace(): + at = datetime(2026, 5, 20, 2, 0, tzinfo=timezone.utc) + plain = scheduled_run_name("o", "p", "d", "t", "n", at) + salted = scheduled_run_name("o", "p", "d", "t", "n", at, salt="bf1") + other = scheduled_run_name("o", "p", "d", "t", "n", at, salt="bf1:second") + assert len({plain, salted, other}) == 3 + + +def test_wall_clock_fields_are_used_verbatim(): + """The scheduler hashes local wall-clock fields, not an instant. + + Two datetimes describing the same instant in different offsets therefore hash + differently -- which is the scheduler's behaviour, not a bug here. + """ + from datetime import timedelta + + utc = datetime(2026, 5, 20, 2, 0, tzinfo=timezone.utc) + plus_two = datetime(2026, 5, 20, 4, 0, tzinfo=timezone(timedelta(hours=2))) + assert utc == plus_two # same instant + assert scheduled_run_name("o", "p", "d", "t", "n", utc) != scheduled_run_name("o", "p", "d", "t", "n", plus_two) + + +def test_candidate_names_cover_the_actions_prefix_rewrite(): + # Automation-sourced runs routed to the actions engine are stored under "u". + assert candidate_run_names("r64640bf1ae650449") == ("r64640bf1ae650449", "u64640bf1ae650449") + assert candidate_run_names("xdeadbeef") == ("xdeadbeef",) diff --git a/tests/backfill/test_plan.py b/tests/backfill/test_plan.py new file mode 100644 index 000000000..606f37c01 --- /dev/null +++ b/tests/backfill/test_plan.py @@ -0,0 +1,132 @@ +"""Planning decides what gets created, skipped, or re-run.""" + +from datetime import datetime, timezone + +import pytest + +from flyte.backfill._naming import scheduled_run_name +from flyte.backfill._plan import DEFAULT_MAX_RUNS, build_plan + +ORG, PROJECT, DOMAIN = "acme", "ml-platform", "production" +TASK, TRIGGER = "evals.weekly.run", "nightly_eval" +START = datetime(2026, 5, 1, 0, 0, tzinfo=timezone.utc) +END = datetime(2026, 5, 10, 23, 59, tzinfo=timezone.utc) + + +class _FakeCron: + def __init__(self, expression="0 2 * * *", timezone_name="UTC"): + self.expression = expression + self.timezone = timezone_name + + +class _FakeSchedule: + def __init__(self, expression="0 2 * * *", kickoff=""): + self._cron = _FakeCron(expression) + self.cron = self._cron + self.cron_expression = "" + self.kickoff_time_input_arg = kickoff + + def HasField(self, field): + return field == "cron" + + +class _FakeName: + def __init__(self, name): + self.name = name + + +class _FakeId: + def __init__(self, name): + self.name = _FakeName(name) + + +class _FakePb2: + def __init__(self, expression="0 2 * * *", kickoff=""): + self.id = _FakeId(TRIGGER) + self.automation_spec = type("A", (), {"schedule": _FakeSchedule(expression, kickoff)})() + + +class FakeTriggerDetails: + """Stands in for a remote TriggerDetails, exposing only what planning reads.""" + + def __init__(self, expression="0 2 * * *", kickoff=""): + self.pb2 = _FakePb2(expression, kickoff) + + +def _plan(**overrides): + kwargs = { + "details": FakeTriggerDetails(), + "task_name": TASK, + "org": ORG, + "project": PROJECT, + "domain": DOMAIN, + "start": START, + "end": END, + "max_runs": DEFAULT_MAX_RUNS, + } + kwargs.update(overrides) + return build_plan(**kwargs) + + +def test_one_slot_per_scheduled_fire(): + plan = _plan() + assert len(plan.slots) == 10 # daily at 02:00, May 1-10 + assert all(s.scheduled_at.hour == 2 for s in plan.slots) + + +def test_slot_names_match_a_real_scheduled_fire(): + plan = _plan() + first = plan.slots[0] + assert first.run_name == scheduled_run_name(ORG, PROJECT, DOMAIN, TASK, TRIGGER, first.scheduled_at) + + +def test_existing_runs_are_skipped(): + plan = _plan() + already = plan.slots[3].run_name + replanned = _plan(existing=[already]) + assert len(replanned.skipped) == 1 + assert len(replanned.to_create) == 9 + assert replanned.skipped[0].run_name == already + + +def test_existing_is_matched_against_the_rewritten_prefix_too(): + """A scheduled run routed to actions is stored under a 'u' prefix.""" + plan = _plan() + stored_as = "u" + plan.slots[0].run_name[1:] + replanned = _plan(existing=[stored_as]) + assert len(replanned.skipped) == 1 + + +def test_force_recreates_existing_slots_under_salted_names(): + plan = _plan() + already = plan.slots[0].run_name + forced = _plan(existing=[already], force=True) + assert len(forced.to_create) == 10 # nothing skipped + assert len(forced.overridden) == 1 + # The forced name must differ, or it would be de-duplicated against the original. + assert forced.slots[0].run_name != already + + +def test_suffix_distinguishes_repeated_forced_backfills(): + first = _plan(force=True, suffix="rerun-1") + second = _plan(force=True, suffix="rerun-2") + assert first.slots[0].run_name != second.slots[0].run_name + + +def test_plan_is_capped_and_reports_what_it_dropped(): + plan = _plan(end=datetime(2026, 8, 1, tzinfo=timezone.utc), max_runs=10) + assert len(plan.slots) == 10 + assert plan.truncated > 0 + + +def test_start_must_precede_end(): + with pytest.raises(ValueError): + _plan(start=END, end=START) + + +def test_trigger_without_a_schedule_is_rejected(): + details = FakeTriggerDetails() + details.pb2.automation_spec.schedule.cron_expression = "" + details.pb2.automation_spec.schedule.HasField = lambda field: False + with pytest.raises(ValueError, match="no schedule"): + _plan(details=details) diff --git a/tests/backfill/test_schedule.py b/tests/backfill/test_schedule.py new file mode 100644 index 000000000..4199a930b --- /dev/null +++ b/tests/backfill/test_schedule.py @@ -0,0 +1,92 @@ +"""Cron expansion has to agree with the scheduler's cron library. + +Expected counts and first fire times below come from running robfig/cron v3 -- +the parser the scheduler itself uses -- over May 2026. +""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from flyte.backfill._schedule import ( + CronParseError, + cron_occurrences, + fixed_rate_occurrences, + parse_cron, +) + +MAY_START = datetime(2026, 5, 1, 0, 0, 0, tzinfo=timezone.utc) +MAY_END = datetime(2026, 5, 31, 23, 59, 59, tzinfo=timezone.utc) + +# expression -> (count over May 2026, first three fire times as MM-DDTHH:MM) +ROBFIG_REFERENCE = { + "0 2 * * *": (31, ["05-01T02:00", "05-02T02:00", "05-03T02:00"]), + "0 6 * * 1": (4, ["05-04T06:00", "05-11T06:00", "05-18T06:00"]), + "*/15 * * * *": (2976, ["05-01T00:00", "05-01T00:15", "05-01T00:30"]), + "30 3 1,15 * *": (2, ["05-01T03:30", "05-15T03:30"]), + # Both day-of-month and day-of-week restricted: standard cron ORs them. + "0 0 1 * 1": (5, ["05-01T00:00", "05-04T00:00", "05-11T00:00"]), + "0 9-17/4 * * MON-FRI": (63, ["05-01T09:00", "05-01T13:00", "05-01T17:00"]), +} + + +@pytest.mark.parametrize(("expression", "expected"), sorted(ROBFIG_REFERENCE.items())) +def test_matches_robfig_cron(expression, expected): + count, first = expected + got = cron_occurrences(expression, MAY_START, MAY_END) + assert len(got) == count + assert [d.strftime("%m-%dT%H:%M") for d in got[: len(first)]] == first + + +def test_dom_and_dow_are_ored_not_anded(): + """A day matches if either field matches, when both are restricted.""" + got = cron_occurrences("0 0 1 * 1", MAY_START, MAY_END) + days = sorted({d.day for d in got}) + assert days == [1, 4, 11, 18, 25] # the 1st, plus every Monday + + +def test_window_bounds_are_inclusive(): + exact = datetime(2026, 5, 1, 2, 0, tzinfo=timezone.utc) + assert cron_occurrences("0 2 * * *", exact, exact) == [exact] + + +def test_limit_stops_expansion(): + got = cron_occurrences("*/15 * * * *", MAY_START, MAY_END, limit=10) + assert len(got) == 10 + + +def test_timezone_prefix_is_stripped(): + fields, tz = parse_cron("CRON_TZ=America/New_York 0 2 * * *") + assert tz == "America/New_York" + assert fields[1] == frozenset({2}) + + +def test_sunday_accepts_both_zero_and_seven(): + as_zero = cron_occurrences("0 0 * * 0", MAY_START, MAY_END) + as_seven = cron_occurrences("0 0 * * 7", MAY_START, MAY_END) + assert as_zero == as_seven + assert all(d.weekday() == 6 for d in as_zero) + + +@pytest.mark.parametrize("bad", ["0 2 * *", "", "0 2 * * * *", "99 2 * * *", "0 2 * * 1/0"]) +def test_invalid_expressions_raise(bad): + with pytest.raises(CronParseError): + cron_occurrences(bad, MAY_START, MAY_END) + + +def test_fixed_rate_respects_the_anchor_phase(): + anchor = datetime(2026, 5, 1, 0, 7, tzinfo=timezone.utc) + got = fixed_rate_occurrences(30, MAY_START, MAY_START + timedelta(hours=2), anchor=anchor) + assert [d.strftime("%H:%M") for d in got] == ["00:07", "00:37", "01:07", "01:37"] + + +def test_fixed_rate_skips_forward_to_the_window(): + anchor = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) + got = fixed_rate_occurrences(60, MAY_START, MAY_START + timedelta(hours=3), anchor=anchor) + assert got[0] == MAY_START + assert len(got) == 4 + + +def test_fixed_rate_rejects_a_nonpositive_interval(): + with pytest.raises(ValueError): + fixed_rate_occurrences(0, MAY_START, MAY_END) diff --git a/tests/cli/test_backfill.py b/tests/cli/test_backfill.py new file mode 100644 index 000000000..02a50d37b --- /dev/null +++ b/tests/cli/test_backfill.py @@ -0,0 +1,108 @@ +import re +from datetime import datetime, timedelta, timezone + +import mock +import pytest +from click.testing import CliRunner + +from flyte.cli._backfill import _parse_time, backfill +from flyte.cli.main import main + + +def _plain(output: str) -> str: + """rich-click styles its output; strip ANSI before matching.""" + return re.sub(r"\x1b\[[0-9;]*m", "", output) + + +def test_backfill_registered_on_main(): + assert "backfill" in main.commands + + +def test_backfill_takes_a_trigger_name_and_the_expected_options(): + opts = {o for p in backfill.params for o in p.opts} + assert {"--from", "--to", "--force", "--suffix", "--dry-run", "--max-runs", "--queue"} <= opts + assert any(p.name == "trigger_name" for p in backfill.params) + + +def test_from_is_required(): + result = CliRunner().invoke(backfill, ["nightly_eval"]) + assert result.exit_code != 0 + assert "--from" in _plain(result.output) + + +def test_suffix_requires_force(): + result = CliRunner().invoke(backfill, ["nightly_eval", "--from", "7d", "--suffix", "x"]) + assert result.exit_code != 0 + assert "--suffix requires --force" in _plain(result.output) + + +def test_window_must_be_ordered(): + result = CliRunner().invoke(backfill, ["nightly_eval", "--from", "2026-05-10", "--to", "2026-05-01"]) + assert result.exit_code != 0 + assert "earlier than" in _plain(result.output) + + +class TestParseTime: + def test_iso_timestamp(self): + assert _parse_time("2026-05-01T02:00", what="--from") == datetime(2026, 5, 1, 2, 0, tzinfo=timezone.utc) + + def test_plain_date(self): + assert _parse_time("2026-05-01", what="--from") == datetime(2026, 5, 1, tzinfo=timezone.utc) + + def test_explicit_offset_is_preserved(self): + parsed = _parse_time("2026-05-01T02:00+02:00", what="--from") + assert parsed.utcoffset() == timedelta(hours=2) + + @pytest.mark.parametrize( + ("value", "delta"), [("30d", timedelta(days=30)), ("12h", timedelta(hours=12)), ("45m", timedelta(minutes=45))] + ) + def test_relative_ages(self, value, delta): + before = datetime.now(timezone.utc) - delta + parsed = _parse_time(value, what="--from") + assert abs((parsed - before).total_seconds()) < 5 + + def test_now(self): + parsed = _parse_time("now", what="--to") + assert abs((parsed - datetime.now(timezone.utc)).total_seconds()) < 5 + + def test_garbage_is_rejected(self): + with pytest.raises(Exception, match="ISO timestamp"): + _parse_time("last tuesday", what="--from") + + def test_empty_is_none(self): + assert _parse_time(None, what="--to") is None + + +def test_dry_run_never_launches_anything(): + """--dry-run must print the plan and stop short of creating runs.""" + from tests.backfill.test_plan import FakeTriggerDetails + + trigger = mock.MagicMock() + trigger.get.aio = mock.AsyncMock(return_value=FakeTriggerDetails()) + + with ( + mock.patch("flyte.cli._common.initialize_config") as init_cfg, + mock.patch("flyte.remote.Trigger", trigger), + mock.patch("flyte.backfill._execute.probe_existing", mock.AsyncMock(return_value=set())), + mock.patch("flyte.backfill._driver.launch_backfill") as launch, + mock.patch("flyte._initialize.get_init_config") as init, + ): + init_cfg.return_value = mock.MagicMock(output_format="table") + init.return_value = mock.MagicMock(org="acme", project="ml-platform", domain="production") + result = CliRunner().invoke( + backfill, + [ + "nightly_eval", + "--task-name", + "evals.weekly.run", + "--from", + "2026-05-01", + "--to", + "2026-05-05", + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + launch.assert_not_called() + assert "Dry run" in _plain(result.output)