feat: allow bounded trailing gaps during production promotion

docs/integrations/github.md

@@ -300,11 +300,56 @@ Below is an example of how to define the default config for the bot in either YA
 | `skip_pr_backfill`                    | Indicates if the bot should skip backfilling models in the PR environment. Default: `True`                                                                                                                                                                                                                                                                                                                |  bool  |    N     |
 | `pr_include_unmodified`               | Indicates whether to include unmodified models in the PR environment. Default to the project's config value (which defaults to `False`)                                                                                                                                                                                                                                                                   |  bool  |    N     |
 | `run_on_deploy_to_prod`               | Indicates whether to run latest intervals when deploying to prod. If set to false, the deployment will backfill only the changed models up to the existing latest interval in production, ignoring any missing intervals beyond this point. Default: `False`                                                                                                                                              |  bool  |    N     |
+| `max_trailing_gap_intervals_on_deploy_to_prod` | Maximum number of scheduler boundaries of newly observed trailing gaps that a production deployment may accept. `0` keeps the strict no-gap behavior (default); a positive value opts in to bounded trailing-gap promotion.                                                                                                                                                         |  int   |    N     |
 | `pr_environment_name`                 | The name of the PR environment to create for which a PR number will be appended to. Defaults to the repo name if not provided. Note: The name will be normalized to alphanumeric + underscore and lowercase.                                                                                                                                                                                              |  str   |    N     |
 | `prod_branch_name`                    | The name of the git branch associated with production. Ex: `prod`. Default: `main` or `master` is considered prod                                                                                                                                                                                                                                                                                         |  str   |    N     |
 | `forward_only_branch_suffix`          | If the git branch has this suffix, trigger a [forward-only](../concepts/plans.md#forward-only-plans) plan instead of a normal plan. Default: `-forward-only`                                                                                                                                                                                                                                              |  str   |    N     |
 | `check_if_blocked_on_deploy_to_prod`  | The bot normally checks if a PR is blocked from merging before deploying to production. Setting this to `False` will skip that check. Default: `True`                                                                                                                                                                                                                                                     |  bool  |    N     |
 
+### Bounded trailing gaps during production promotion
+
+`max_trailing_gap_intervals_on_deploy_to_prod` handles a scheduler race while a
+production plan is being built. It defaults to `0`, which preserves strict
+no-gap promotion. Set it to a finite,
+nonnegative integer to allow a production plan with `no_gaps=True` to accept a
+bounded, contiguous trailing suffix that is observed after the plan's immutable
+plan-time frontier. The candidate must still contain complete coverage through
+that frontier. Historical, internal, ambiguous, and over-bound gaps continue to
+block promotion.
+
+As a conservative first-release guard, promotion also rejects an otherwise
+eligible suffix when any transitive downstream model already has recorded
+coverage overlapping that suffix. A normal scheduler repair can fill the
+upstream gap, but it does not automatically invalidate downstream intervals
+that are already recorded, so the deployment fails closed in this case.
+
+The bound counts actual scheduler boundaries in stored interval coverage. It
+does not count lookback-expanded work, so one accepted missing interval can
+cause Cron to plan more intervals when it repairs a model with lookback. After
+promotion, the normal Cron scheduler discovers the missing suffix and repairs
+it using the usual dependency ordering and lookback behavior. Any subsequent
+scheduler selection must include the affected models' upstream dependencies (or
+use the scheduler's normal automatic upstream selection). A downstream-only
+selection can record incomplete data before its upstream models are repaired;
+this option does not automatically invalidate that work.
+
+The option is passed only to the production `no_gaps` plan. PR, development,
+test, and production plans built with `no_gaps=False` retain their existing
+behavior. The promotion warning identifies the environment, plan, snapshots,
+frontiers, and accepted range so the temporary freshness debt is observable.
+The warning reports the observed debt at classification time. SQLMesh does not
+fence a scheduler that is already in flight, so additional intervals may be
+recorded on the incumbent after that check; the next Cron run must reconcile
+any remaining coverage on the promoted candidate.
+
+For example:
+
+```yaml
+cicd_bot:
+  type: github
+  max_trailing_gap_intervals_on_deploy_to_prod: 1
+```
+
 Example with all properties defined:
 
 === "YAML"

sqlmesh/core/context.py

@@ -91,6 +91,7 @@
 )
 from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals, PlanExplainer
 from sqlmesh.core.plan.definition import UserProvidedFlags
+from sqlmesh.core.promotion import PromotionFrontier
 from sqlmesh.core.reference import ReferenceGraph
 from sqlmesh.core.scheduler import Scheduler, CompletionStatus
 from sqlmesh.core.schema_loader import create_external_models_file
@@ -1461,6 +1462,7 @@ def plan(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
+        max_trailing_gap_intervals: t.Optional[int] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1511,6 +1513,8 @@ def plan(
             explain: Whether to explain the plan instead of applying it.
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
+            max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that production
+                promotion may accept as debt when no_gaps is enabled.
 
         Returns:
             The populated Plan object.
@@ -1543,6 +1547,7 @@ def plan(
             explain=explain,
             ignore_cron=ignore_cron,
             min_intervals=min_intervals,
+            max_trailing_gap_intervals=max_trailing_gap_intervals,
         )
 
         plan = plan_builder.build()
@@ -1601,6 +1606,7 @@ def plan_builder(
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
         always_include_local_changes: t.Optional[bool] = None,
+        max_trailing_gap_intervals: t.Optional[int] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1644,6 +1650,8 @@ def plan_builder(
                 on every model when checking for missing intervals
             always_include_local_changes: Usually when restatements are present, local changes in the filesystem are ignored.
                 However, it can be desirable to deploy changes + restatements in the same plan, so this flag overrides the default behaviour.
+            max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that production
+                promotion may accept as debt when no_gaps is enabled.
 
         Returns:
             The plan builder.
@@ -1696,6 +1704,9 @@ def plan_builder(
         diff_rendered = diff_rendered or False
         skip_linter = skip_linter or False
         min_intervals = min_intervals or 0
+        max_trailing_gap_intervals = (
+            0 if max_trailing_gap_intervals is None else max_trailing_gap_intervals
+        )
 
         environment = environment or self.config.default_target_environment
         environment = Environment.sanitize_name(environment)
@@ -1821,12 +1832,29 @@ def plan_builder(
             backfill_models = modified_model_names or None
 
         max_interval_end_per_model = None
+        planned_frontiers: t.Dict[str, PromotionFrontier] = {}
         default_start, default_end = None, None
-        if not run:
-            ignore_cron = False
-            max_interval_end_per_model = self._get_max_interval_end_per_model(
+        should_capture_frontiers = not is_dev and no_gaps and max_trailing_gap_intervals > 0
+        if not run or should_capture_frontiers:
+            captured_max_interval_end_per_model = self._get_max_interval_end_per_model(
                 snapshots, backfill_models
             )
+            if should_capture_frontiers:
+                planned_frontiers = self._get_planned_frontiers(
+                    context_diff, captured_max_interval_end_per_model
+                )
+
+            if run:
+                # `--run` deliberately does not cap backfills at the current production
+                # frontier. Keep the captured values separate so recording promotion
+                # frontiers does not change that behavior.
+                max_interval_end_per_model = None
+            else:
+                max_interval_end_per_model = captured_max_interval_end_per_model
+
+        if not run:
+            ignore_cron = False
+            assert max_interval_end_per_model is not None
             # If no end date is specified, use the max interval end from prod
             # to prevent unintended evaluation of the entire DAG.
             default_start, default_end = self._get_plan_default_start_end(
@@ -1925,6 +1953,8 @@ def plan_builder(
             },
             explain=explain or False,
             ignore_cron=ignore_cron or False,
+            max_trailing_gap_intervals=max_trailing_gap_intervals,
+            planned_frontiers=planned_frontiers,
         )
 
     def apply(
@@ -3588,6 +3618,38 @@ def _get_max_interval_end_per_model(
             ).items()
         }
 
+    @staticmethod
+    def _get_planned_frontiers(
+        context_diff: ContextDiff,
+        max_interval_end_per_model: t.Mapping[str, datetime],
+    ) -> t.Dict[str, PromotionFrontier]:
+        """Capture production frontiers for changed models at plan-build time.
+
+        The interval-end query is intentionally performed before any execution-time or
+        explicit-end adjustment. The incumbent identity comes from the context diff's
+        modified snapshot tuple and is carried with the frontier so promotion can verify
+        it against the incumbent observed under its environment-row lock. A missing
+        previous plan means there is no incumbent identity to carry, so no frontier is
+        recorded and promotion remains strict.
+        """
+        if not context_diff.previous_plan_id:
+            return {}
+
+        frontiers: t.Dict[str, PromotionFrontier] = {}
+
+        for name, (current_snapshot, previous_snapshot) in context_diff.modified_snapshots.items():
+            if not current_snapshot.is_incremental or not previous_snapshot.is_incremental:
+                continue
+            if name not in max_interval_end_per_model:
+                continue
+
+            frontiers[name] = PromotionFrontier(
+                snapshot_id=previous_snapshot.snapshot_id,
+                end=to_timestamp(max_interval_end_per_model[name]),
+            )
+
+        return frontiers
+
     @staticmethod
     def _get_models_for_interval_end(
         snapshots: t.Dict[str, Snapshot], backfill_models: t.Set[str]

sqlmesh/core/plan/builder.py

@@ -23,6 +23,7 @@
     UserProvidedFlags,
     earliest_interval_start,
 )
+from sqlmesh.core.promotion import PromotionFrontier
 from sqlmesh.core.schema_diff import (
     get_schema_differ,
     has_drop_alteration,
@@ -72,6 +73,9 @@ class PlanBuilder:
         no_gaps:  Whether to ensure that new snapshots for nodes that are already a
             part of the target environment have no data gaps when compared against previous
             snapshots for same nodes.
+        max_trailing_gap_intervals: The maximum number of trailing scheduler intervals that
+            production promotion may accept as debt when no_gaps is enabled.
+        planned_frontiers: The production snapshot frontiers captured when the plan was built.
         skip_backfill: Whether to skip the backfill step.
         empty_backfill: Like skip_backfill, but also records processed intervals.
         is_dev: Whether this plan is for development purposes.
@@ -138,6 +142,8 @@ def __init__(
         console: t.Optional[PlanBuilderConsole] = None,
         user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None,
         selected_models: t.Optional[t.Set[str]] = None,
+        max_trailing_gap_intervals: int = 0,
+        planned_frontiers: t.Optional[t.Dict[str, PromotionFrontier]] = None,
     ):
         self._context_diff = context_diff
         self._no_gaps = no_gaps
@@ -184,6 +190,8 @@ def __init__(
         self._user_provided_flags = user_provided_flags
         self._selected_models = selected_models
         self._explain = explain
+        self._max_trailing_gap_intervals = max_trailing_gap_intervals
+        self._planned_frontiers = dict(planned_frontiers or {})
 
         self._start = start
         if not self._start and self._forward_only_preview_needed:
@@ -375,6 +383,8 @@ def build(self) -> Plan:
             ignore_cron=self._ignore_cron,
             user_provided_flags=self._user_provided_flags,
             selected_models=self._selected_models,
+            max_trailing_gap_intervals=self._max_trailing_gap_intervals,
+            planned_frontiers=self._planned_frontiers,
         )
         self._latest_plan = plan
         return plan

sqlmesh/core/plan/definition.py

@@ -9,6 +9,7 @@
 
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements
+from sqlmesh.core.promotion import PromotionFrontier
 from sqlmesh.utils.metaprogramming import Executable  # noqa
 from sqlmesh.core.node import IntervalUnit
 from sqlmesh.core.snapshot import (
@@ -83,6 +84,10 @@ class Plan(PydanticModel, frozen=True):
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
     selected_models: t.Optional[t.Set[str]] = None
     """Models that have been selected for this plan (used for dbt selected_resources)"""
+    max_trailing_gap_intervals: int = Field(default=0, ge=0, strict=True)
+    """Maximum number of trailing scheduler intervals that promotion may accept as debt."""
+    planned_frontiers: t.Dict[str, PromotionFrontier] = Field(default_factory=dict)
+    """The production snapshot frontier captured when the plan was built, keyed by model name."""
 
     @cached_property
     def start(self) -> TimeLike:
@@ -299,6 +304,8 @@ def to_evaluatable(self) -> EvaluatablePlan:
             environment_statements=self.context_diff.environment_statements,
             user_provided_flags=self.user_provided_flags,
             selected_models=self.selected_models,
+            max_trailing_gap_intervals=self.max_trailing_gap_intervals,
+            planned_frontiers=dict(self.planned_frontiers),
         )
 
     @cached_property
@@ -338,6 +345,8 @@ class EvaluatablePlan(PydanticModel):
     environment_statements: t.Optional[t.List[EnvironmentStatements]] = None
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
     selected_models: t.Optional[t.Set[str]] = None
+    max_trailing_gap_intervals: int = Field(default=0, ge=0, strict=True)
+    planned_frontiers: t.Dict[str, PromotionFrontier] = Field(default_factory=dict)
 
     def is_selected_for_backfill(self, model_fqn: str) -> bool:
         return self.models_to_backfill is None or model_fqn in self.models_to_backfill

sqlmesh/core/plan/evaluator.py

@@ -363,11 +363,16 @@ def visit_restatement_stage(
     def visit_environment_record_update_stage(
         self, stage: stages.EnvironmentRecordUpdateStage, plan: EvaluatablePlan
     ) -> None:
-        self.state_sync.promote(
-            plan.environment,
-            no_gaps_snapshot_names=stage.no_gaps_snapshot_names if plan.no_gaps else set(),
-            environment_statements=plan.environment_statements,
-        )
+        promote_kwargs: t.Dict[str, t.Any] = {
+            "no_gaps_snapshot_names": (stage.no_gaps_snapshot_names if plan.no_gaps else set()),
+            "environment_statements": plan.environment_statements,
+        }
+        if stage.max_trailing_gap_intervals:
+            promote_kwargs.update(
+                max_trailing_gap_intervals=stage.max_trailing_gap_intervals,
+                planned_frontiers=stage.planned_frontiers,
+            )
+        self.state_sync.promote(plan.environment, **promote_kwargs)
 
     def visit_migrate_schemas_stage(
         self, stage: stages.MigrateSchemasStage, plan: EvaluatablePlan

sqlmesh/core/plan/stages.py

@@ -1,10 +1,11 @@
 import typing as t
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from sqlmesh.core import constants as c
 from sqlmesh.core.environment import EnvironmentStatements, EnvironmentNamingInfo, Environment
 from sqlmesh.core.plan.common import should_force_rebuild
 from sqlmesh.core.plan.definition import EvaluatablePlan
+from sqlmesh.core.promotion import PromotionFrontier
 from sqlmesh.core.state_sync import StateReader
 from sqlmesh.core.scheduler import merged_missing_intervals, SnapshotToIntervals
 from sqlmesh.core.snapshot.definition import (
@@ -141,9 +142,14 @@ class EnvironmentRecordUpdateStage:
 
     Args:
         no_gaps_snapshot_names: Names of snapshots for which there should be no interval gaps.
+        max_trailing_gap_intervals: Maximum trailing scheduler intervals that may be accepted
+            during promotion.
+        planned_frontiers: Production snapshot frontiers captured when the plan was built.
     """
 
     no_gaps_snapshot_names: t.Set[str]
+    max_trailing_gap_intervals: int = 0
+    planned_frontiers: t.Dict[str, PromotionFrontier] = field(default_factory=dict)
 
 
 @dataclass
@@ -370,7 +376,11 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
 
         stages.append(
             EnvironmentRecordUpdateStage(
-                no_gaps_snapshot_names={s.name for s in before_promote_snapshots}
+                no_gaps_snapshot_names={s.name for s in before_promote_snapshots},
+                max_trailing_gap_intervals=(
+                    plan.max_trailing_gap_intervals if plan.is_prod and plan.no_gaps else 0
+                ),
+                planned_frontiers=(plan.planned_frontiers if plan.is_prod and plan.no_gaps else {}),
             )
         )