diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index ae036a7774..319b6c0115 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -112,6 +112,8 @@ Configuration for the `sqlmesh run` command. Please note that this is only appli | `environment_check_interval` | The number of seconds to wait between attempts to check the target environment for readiness (Default: 30 seconds) | int | N | | `environment_check_max_wait` | The maximum number of seconds to wait for the target environment to be ready (Default: 6 hours) | int | N | +`sqlmesh run` checks the target environment's persisted finalization state. This check does not establish whether a plan process is active. A failed promotion can leave the environment unfinalized, so `run` continues to wait. If the plan failed, inspect its logs, resolve the errors, and successfully reapply it with `sqlmesh plan ` before retrying the run. + ## Format Formatting settings for the `sqlmesh format` command and UI. diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index 39590a0fa3..3bf9249005 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -920,13 +920,18 @@ def _block_until_finalized() -> str: if environment_state.finalized_ts: return environment_state.plan_id self.console.log_warning( - f"Environment '{environment}' is being updated by plan '{environment_state.plan_id}'. " - f"Retrying in {self.config.run.environment_check_interval} seconds..." + f"Environment '{environment}' is not finalized after plan '{environment_state.plan_id}'. " + "The plan may still be running or may have failed. " + "Waiting for environment finalization, " + f"retrying in {self.config.run.environment_check_interval} seconds " + f"(configured maximum wait: {self.config.run.environment_check_max_wait} seconds)." ) time.sleep(self.config.run.environment_check_interval) raise SQLMeshError( - f"Exceeded the maximum wait time for environment '{environment}' to be ready. " - "This means that the environment either failed to update or the update is taking longer than expected. " + f"Exceeded the maximum wait time for environment '{environment}' to finalize " + f"(configured maximum wait: {self.config.run.environment_check_max_wait} seconds). " + "The plan may still be running or may have failed. Inspect the plan logs and, if it failed, " + f"resolve its errors and reapply it with `sqlmesh plan {environment}` to finalize the environment. " "See https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#run to adjust the timeout settings." ) diff --git a/tests/core/integration/test_run.py b/tests/core/integration/test_run.py index c3e6626ad0..dbb4ff6e08 100644 --- a/tests/core/integration/test_run.py +++ b/tests/core/integration/test_run.py @@ -1,6 +1,7 @@ from __future__ import annotations import typing as t +from pathlib import Path import pytest import time_machine from pytest_mock.plugin import MockerFixture @@ -8,12 +9,22 @@ from sqlmesh.core import constants as c from sqlmesh.core import dialect as d from sqlmesh.core.config.categorizer import CategorizerConfig +from sqlmesh.core.config import ( + Config, + DuckDBConnectionConfig, + GatewayConfig, + ModelDefaultsConfig, + RunConfig, +) +from sqlmesh.core.context import Context from sqlmesh.core.model import ( SqlModel, PythonModel, load_sql_based_model, ) from sqlmesh.utils.date import to_timestamp +from sqlmesh.utils.concurrency import NodeExecutionFailedError +from sqlmesh.utils.errors import SQLMeshError if t.TYPE_CHECKING: pass @@ -21,6 +32,77 @@ pytestmark = pytest.mark.slow +def test_run_waits_for_finalization_after_failed_plan(tmp_path: Path, mocker: MockerFixture): + model_path = tmp_path / "models" / "example.sql" + model_path.parent.mkdir() + original_model = "MODEL (name repro.example, kind FULL); SELECT 1::INT AS id;" + updated_model = original_model.replace("SELECT 1", "SELECT 2") + model_path.write_text(original_model) + context = Context( + paths=tmp_path, + config=Config( + gateways={ + "local": GatewayConfig( + connection=DuckDBConnectionConfig(database=str(tmp_path / "repro.duckdb")) + ) + }, + default_gateway="local", + model_defaults=ModelDefaultsConfig(dialect="duckdb"), + disable_anonymized_analytics=True, + run=RunConfig(environment_check_interval=1, environment_check_max_wait=2), + ), + ) + try: + context.plan("prod", auto_apply=True, no_prompts=True) + initial_environment = context.state_sync.get_environment("prod") + assert initial_environment is not None and initial_environment.finalized_ts + + model_path.write_text( + updated_model + + "\nON_VIRTUAL_UPDATE_BEGIN;" + + "\nSELECT CAST('intentional_promotion_failure' AS INTEGER);" + + "\nON_VIRTUAL_UPDATE_END;" + ) + context.load() + with pytest.raises(NodeExecutionFailedError): + context.plan("prod", auto_apply=True, no_prompts=True) + + failed_environment = context.state_sync.get_environment("prod") + assert failed_environment is not None + assert failed_environment.finalized_ts is None + run = mocker.spy(context, "_run") + warnings = mocker.spy(context.console, "log_warning") + with pytest.raises(SQLMeshError, match="Exceeded the maximum wait time") as exc_info: + context.run("prod", skip_janitor=True) + + run.assert_not_called() + assert context.state_sync.get_environment("prod") == failed_environment + assert warnings.call_count == 2 + + model_path.write_text(updated_model) + context.load() + context.plan("prod", auto_apply=True, no_prompts=True) + recovered_environment = context.state_sync.get_environment("prod") + assert recovered_environment is not None and recovered_environment.finalized_ts + assert not context.run("prod", skip_janitor=True).is_failure + run.assert_called_once() + + warning = warnings.call_args_list[0].args[0] + assert failed_environment.plan_id in warning + assert "finaliz" in warning + assert "1 seconds" in warning + assert "2 seconds" in warning + assert "failed" in warning + assert "is being updated" not in warning + assert "finaliz" in str(exc_info.value) + assert "sqlmesh plan prod" in str(exc_info.value) + assert "https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#run" in str( + exc_info.value + ) + finally: + context.close() + + @time_machine.travel("2023-01-08 15:00:00 UTC") def test_run_with_select_models( init_and_plan_context: t.Callable,