From bca6a687c025d8c42ce3b43f4c25d3c4a2813b76 Mon Sep 17 00:00:00 2001 From: khup Date: Fri, 28 Aug 2026 14:20:12 -0400 Subject: [PATCH 1/4] Keep Dag bundle generations available while in use Concurrent users must share one stable tracking inode so cleanup cannot delete code that a task or parser still needs. Lease acquisition therefore fails closed, while release remains best effort to avoid hiding the protected operation's outcome. --- .../airflow/dag_processing/bundles/base.py | 82 +++++++++---------- .../unit/dag_processing/bundles/test_base.py | 56 +++++++++++++ 2 files changed, 96 insertions(+), 42 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py b/airflow-core/src/airflow/dag_processing/bundles/base.py index 344a3349fecab..383da1d3feeae 100644 --- a/airflow-core/src/airflow/dag_processing/bundles/base.py +++ b/airflow-core/src/airflow/dag_processing/bundles/base.py @@ -24,7 +24,7 @@ import tempfile import warnings from abc import ABC, abstractmethod -from contextlib import contextmanager +from contextlib import contextmanager, suppress from dataclasses import dataclass, field from datetime import timedelta from fcntl import LOCK_SH, LOCK_UN, flock @@ -33,8 +33,8 @@ from typing import TYPE_CHECKING, Any import pendulum -from pendulum.parsing import ParserError +from airflow._shared.timezones import timezone from airflow.configuration import conf if TYPE_CHECKING: @@ -96,13 +96,6 @@ class BundleUsageTrackingManager: :meta private: """ - def _parse_dt(self, val) -> DateTime | None: - try: - dt = pendulum.parse(val) - return dt if isinstance(dt, pendulum.DateTime) else None - except ParserError: - return None - @staticmethod def _filter_for_min_versions(val: list[TrackedBundleVersionInfo]) -> list[TrackedBundleVersionInfo]: min_versions_to_keep = conf.getint( @@ -134,15 +127,10 @@ def _find_all_tracking_files(self, bundle_name) -> list[TrackedBundleVersionInfo for file in tracking_dir.iterdir(): log.debug("found bundle tracking file, path=%s", file) version = file.name - dt_str = file.read_text() - dt = self._parse_dt(val=dt_str) - if not dt: - log.error( - "could not parse val as datetime bundle_name=%s val=%s version=%s", - bundle_name, - dt_str, - version, - ) + try: + dt = timezone.from_timestamp(file.stat().st_mtime) + except FileNotFoundError: + # A concurrent cleanup may remove a tracking file after iterdir() observed it. continue found.append(TrackedBundleVersionInfo(lock_file_path=file, version=version, dt=dt)) return found @@ -169,9 +157,10 @@ def log_info(msg): with open(info.lock_file_path, "a") as f: flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) # exclusive lock, do not wait # remove the actual bundle copy - shutil.rmtree(bundle_version_path) + with suppress(FileNotFoundError): + shutil.rmtree(bundle_version_path) # remove the lock file - os.remove(info.lock_file_path) + info.lock_file_path.unlink(missing_ok=True) except BlockingIOError: log_info("could not obtain lock. stale bundle will not be removed.") return @@ -285,8 +274,9 @@ class BaseDagBundle(ABC): that bundle version. This also means, that on a single worker, it's possible that multiple versions of the same bundle are used at the same time. - In contrast, the DAG processor uses a bundle to keep the DAGs from that bundle up to date. There will not be - multiple versions of the same bundle in use at the same time. The DAG processor will always use the latest version. + In contrast, the DAG processor uses a bundle to keep the DAGs from that bundle up to date. It discovers the + latest version while allowing in-flight and version-pinned callback work to finish against its original + generation. :param name: String identifier for the DAG bundle :param refresh_interval: How often the bundle should be refreshed from the source in seconds @@ -297,6 +287,8 @@ class BaseDagBundle(ABC): """ supports_versioning: bool = False + refreshes_to_versioned_paths: bool = False + """Whether refreshing publishes a new immutable path while older published paths remain usable.""" _locked: bool = False @@ -452,6 +444,9 @@ class BundleVersionLock: """ Lock version of bundle when in use to prevent deletion. + Acquisition failures propagate because running without the lease could allow cleanup to remove code in use. + Release failures are logged so they do not mask the outcome of the protected operation. + :meta private: """ @@ -476,28 +471,35 @@ def _log_exc(self, msg): self.lock_file_path, ) - def _update_version_file(self): - """Create a version file containing last-used timestamp.""" - if TYPE_CHECKING: - assert self.lock_file_path - self.lock_file_path.parent.mkdir(parents=True, exist_ok=True) - - with tempfile.TemporaryDirectory() as td: - temp_file = Path(td, self.lock_file_path) - now = pendulum.now(tz=pendulum.UTC) - temp_file.write_text(now.isoformat()) - os.replace(temp_file, self.lock_file_path) - def acquire(self): if not self.version: return if self.lock_file: return - self._update_version_file() if TYPE_CHECKING: assert self.lock_file_path - self.lock_file = open(self.lock_file_path) - flock(self.lock_file, LOCK_SH) + self.lock_file_path.parent.mkdir(parents=True, exist_ok=True) + while True: + lock_file = open(self.lock_file_path, "a+") + flock(lock_file, LOCK_SH) + try: + path_stat = self.lock_file_path.stat() + except FileNotFoundError: + flock(lock_file, LOCK_UN) + lock_file.close() + continue + descriptor_stat = os.fstat(lock_file.fileno()) + if (descriptor_stat.st_dev, descriptor_stat.st_ino) != ( + path_stat.st_dev, + path_stat.st_ino, + ): + flock(lock_file, LOCK_UN) + lock_file.close() + continue + now = pendulum.now(tz=pendulum.UTC).timestamp() + os.utime(self.lock_file_path, (now, now)) + self.lock_file = lock_file + return def release(self): if self.lock_file: @@ -506,11 +508,7 @@ def release(self): self.lock_file = None def __enter__(self) -> Self: - # wrapping in try except here is just extra cautious since this is in task execution path - try: - self.acquire() - except Exception: - self._log_exc("error when attempting to acquire lock") + self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): diff --git a/airflow-core/tests/unit/dag_processing/bundles/test_base.py b/airflow-core/tests/unit/dag_processing/bundles/test_base.py index f092f3e00e770..713d868d99872 100644 --- a/airflow-core/tests/unit/dag_processing/bundles/test_base.py +++ b/airflow-core/tests/unit/dag_processing/bundles/test_base.py @@ -26,6 +26,7 @@ from pathlib import Path from unittest.mock import call, patch +import pendulum import pytest import time_machine @@ -35,7 +36,9 @@ BundleUsageTrackingManager, BundleVersion, BundleVersionLock, + TrackedBundleVersionInfo, get_bundle_storage_root_path, + get_bundle_version_path, ) from tests_common.test_utils.config import conf_vars @@ -74,6 +77,10 @@ def path(self): pass +def test_bundle_refresh_does_not_publish_versioned_paths_by_default(): + assert BasicBundle(name="basic").refreshes_to_versioned_paths is False + + def test_dag_bundle_root_storage_path(): with conf_vars({("dag_processor", "dag_bundle_storage_path"): None}): assert get_bundle_storage_root_path() == Path(tempfile.gettempdir(), "airflow", "dag_bundles") @@ -194,6 +201,31 @@ def test_that_shared_lock_blocks_ex_lock(self): lth1.stop = True t1.join() + def test_first_shared_holder_remains_protected_after_second_releases(self): + bundle_name = "abc" + version = "shared-version" + bundle_path = get_bundle_version_path(bundle_name=bundle_name, version=version) + bundle_path.mkdir(parents=True) + first = BundleVersionLock(bundle_name=bundle_name, bundle_version=version) + second = BundleVersionLock(bundle_name=bundle_name, bundle_version=version) + first.acquire() + second.acquire() + second.release() + info = TrackedBundleVersionInfo( + lock_file_path=first.lock_file_path, + version=version, + dt=pendulum.now(tz=pendulum.UTC), + ) + + BundleUsageTrackingManager._remove_stale_bundle(bundle_name, info) + + assert bundle_path.exists() + first.release() + + BundleUsageTrackingManager._remove_stale_bundle(bundle_name, info) + + assert not bundle_path.exists() + def test_that_no_version_is_noop(self): with BundleVersionLock( bundle_name="Yer face", @@ -203,6 +235,16 @@ def test_that_no_version_is_noop(self): assert b.lock_file_path is None assert b.lock_file is None + def test_context_manager_propagates_acquire_failure(self): + lock = BundleVersionLock(bundle_name="abc", bundle_version="v1") + + with ( + patch.object(lock, "acquire", side_effect=OSError("tracking unavailable")), + pytest.raises(OSError, match="tracking unavailable"), + ): + with lock: + pass + def test_log_exc_formats_message_correctly(self): """Test that _log_exc correctly formats the log message with all parameters.""" from airflow.dag_processing.bundles.base import log as bundle_log @@ -241,6 +283,20 @@ def refresh(self) -> None: ... class TestBundleUsageTrackingManager: + def test_cleanup_removes_tracking_file_when_generation_is_missing(self): + lock = BundleVersionLock(bundle_name="abc", bundle_version="missing") + lock.acquire() + lock.release() + info = TrackedBundleVersionInfo( + lock_file_path=lock.lock_file_path, + version="missing", + dt=pendulum.now(tz=pendulum.UTC), + ) + + BundleUsageTrackingManager._remove_stale_bundle("abc", info) + + assert not lock.lock_file_path.exists() + @pytest.mark.parametrize( ("threshold_hours", "min_versions", "when_hours", "expected_remaining"), [ From ef59035b32ff2b8b1a73d420280d1374cb216b15 Mon Sep 17 00:00:00 2001 From: khup Date: Fri, 28 Aug 2026 14:20:43 -0400 Subject: [PATCH 2/4] Preserve immutable Dag bundle generations during processing Refreshing into immutable paths is only safe when queued, in-flight, callback, and task work retain the exact generation they reference. Carrying that identity across the supervisor protocol also prevents stale parse results from replacing newer serialized state. --- .../src/airflow/dag_processing/manager.py | 235 +++++++++- .../src/airflow/dag_processing/processor.py | 11 +- .../tests/unit/dag_processing/test_manager.py | 406 ++++++++++++++++-- .../unit/dag_processing/test_processor.py | 43 ++ .../sdk/execution_time/schema/schema.json | 12 + .../schema/versions/__init__.py | 7 +- .../schema/versions/v2026_10_30.py | 11 + .../airflow/sdk/execution_time/task_runner.py | 43 +- .../execution_time/test_task_runner.py | 14 +- 9 files changed, 703 insertions(+), 79 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index d892f5d7f238c..7f0e6d1d78a33 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -51,6 +51,7 @@ from airflow.configuration import conf from airflow.dag_processing.bundles.base import ( BundleUsageTrackingManager, + BundleVersionLock, unpack_bundle_version, ) from airflow.dag_processing.bundles.manager import DagBundlesManager @@ -140,6 +141,7 @@ class DagFileInfo: bundle_name: str bundle_path: Path | None = field(compare=False, default=None) bundle_version: str | None = None + bundle_version_data: dict[str, Any] | None = field(compare=False, default=None) @property def absolute_path(self) -> Path: @@ -265,6 +267,8 @@ class DagFileProcessorManager(LoggingMixin): _dag_bundles: list[BaseDagBundle] = attrs.field(factory=list, init=False) _bundle_versions: dict[str, str | None] = attrs.field(factory=dict, init=False) _bundle_version_data: dict[str, dict | None] = attrs.field(factory=dict, init=False) + _published_bundle_versions: dict[str, str | None] = attrs.field(factory=dict, init=False) + _bundle_version_locks: dict[tuple[str, str], BundleVersionLock] = attrs.field(factory=dict, init=False) _multi_team: bool = attrs.field(factory=lambda: conf.getboolean("core", "multi_team"), init=False) _bundle_name_to_team_name: dict[str, str | None] = attrs.field(factory=dict, init=False) @@ -378,7 +382,18 @@ def before_run(self) -> None: gc.freeze() def after_run(self) -> None: - """Tear down state after the parsing loop exits. Default no-op; override to customize.""" + """Release bundle generation leases after the parsing loop exits.""" + locks = list(self._bundle_version_locks.items()) + self._bundle_version_locks.clear() + for (bundle_name, bundle_version), lock in locks: + try: + lock.release() + except Exception: + self.log.exception( + "Failed to release bundle generation lease for %s at version %s", + bundle_name, + bundle_version, + ) def prepare_server_process_context(self) -> None: """ @@ -572,9 +587,10 @@ def _run_parsing_loop(self): self._kill_timed_out_processors() - self._queue_requested_files_for_parsing() - + priority_files = self._claim_requested_files_for_parsing() self._refresh_dag_bundles(known_files=known_files) + self._queue_requested_files_for_parsing(priority_files, known_files=known_files) + self._reconcile_bundle_version_locks(known_files) if not self._file_queue: # Generate more file paths to process if we processed all the files already. Note for this to @@ -591,6 +607,7 @@ def _run_parsing_loop(self): for callback in self.fetch_callbacks(): self._add_callback_to_queue(callback) self._scan_stale_dags() + self._reconcile_bundle_version_locks(known_files) self._cleanup_stale_bundle_versions() self.purge_inactive_dag_warnings() @@ -635,13 +652,66 @@ def _service_processor_sockets(self, timeout: float | None = 1.0): on_close(sock) sock.close() - def _queue_requested_files_for_parsing(self) -> None: - """Queue any files requested for parsing as requested by users via UI/API.""" + def _claim_requested_files_for_parsing(self) -> list[DagFileInfo]: + """Claim priority files and request their bundles be refreshed before queueing them.""" files = self.claim_priority_files() - self._add_files_to_queue(files, mode="frontprio") self.request_bundle_refresh(file.bundle_name for file in files) if self._force_refresh_bundles: self.log.info("Bundles being force refreshed: %s", ", ".join(self._force_refresh_bundles)) + return files + + def _queue_requested_files_for_parsing( + self, + files: list[DagFileInfo] | None = None, + *, + known_files: dict[str, set[DagFileInfo]] | None = None, + ) -> None: + """Queue priority files against their bundle's successfully refreshed generation.""" + if files is None: + files = self._claim_requested_files_for_parsing() + bundles = {bundle.name: bundle for bundle in self._dag_bundles} + resolved_files = [] + for file in files: + bundle = bundles.get(file.bundle_name) + if bundle is None: + resolved_files.append(file) + continue + if bundle.refreshes_to_versioned_paths is True and known_files is not None: + published_file = next( + ( + known_file + for known_file in known_files.get(bundle.name, ()) + if known_file.presence_key == file.presence_key + ), + None, + ) + if published_file is None: + self.log.info( + "Skipping priority parse for %s in bundle %s because it is not in a published generation", + file.rel_path, + bundle.name, + ) + continue + resolved_files.append(published_file) + continue + + if bundle.refreshes_to_versioned_paths is True: + bundle_version, bundle_version_data = unpack_bundle_version( + bundle.get_current_version(), bundle + ) + else: + bundle_version = None + bundle_version_data = None + resolved_files.append( + DagFileInfo( + rel_path=file.rel_path, + bundle_name=file.bundle_name, + bundle_path=bundle.path, + bundle_version=bundle_version, + bundle_version_data=bundle_version_data, + ) + ) + self._add_files_to_queue(resolved_files, mode="frontprio") def claim_priority_files(self) -> list[DagFileInfo]: """ @@ -692,7 +762,19 @@ def _claim_priority_files(self, *, session: Session = NEW_SESSION) -> list[DagFi bundle = bundles[request.bundle_name] files.append( DagFileInfo( - rel_path=Path(request.relative_fileloc), bundle_name=bundle.name, bundle_path=bundle.path + rel_path=Path(request.relative_fileloc), + bundle_name=bundle.name, + bundle_path=bundle.path, + bundle_version=( + self._bundle_versions.get(bundle.name) + if bundle.refreshes_to_versioned_paths is True + else None + ), + bundle_version_data=( + self._bundle_version_data.get(bundle.name) + if bundle.refreshes_to_versioned_paths is True + else None + ), ) ) session.delete(request) @@ -759,9 +841,20 @@ def prepare_callback_bundle(self, request: CallbackRequest) -> BaseDagBundle | N self.log.error("Bundle %s no longer configured, skipping callback", request.bundle_name) return None if bundle.supports_versioning and request.bundle_version: + key = (request.bundle_name, request.bundle_version) + already_leased = key in self._bundle_version_locks try: + self._acquire_bundle_version_lock(*key) bundle.initialize() except Exception: + if not already_leased and (lock := self._bundle_version_locks.pop(key, None)): + try: + lock.release() + except Exception: + self.log.exception( + "Failed to release callback bundle lease for %s at version %s", + *key, + ) self.log.exception( "Error initializing bundle %s version %s for callback, skipping", request.bundle_name, @@ -781,6 +874,7 @@ def _add_callback_to_queue(self, request: CallbackRequest) -> None: bundle_path=bundle.path, bundle_name=request.bundle_name, bundle_version=request.bundle_version, + bundle_version_data=request.version_data, ) self._callback_to_execute[file_info].append(request) self._add_files_to_queue([file_info], mode="front") @@ -909,6 +1003,26 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]): version_after_refresh, version_data_after_refresh = unpack_bundle_version( bundle.get_current_version(), bundle ) + if bundle.refreshes_to_versioned_paths is True: + try: + self._acquire_bundle_version_lock(bundle.name, version_after_refresh) + except Exception: + self.log.exception( + "Could not lease published generation %s for bundle %s", + version_after_refresh, + bundle.name, + ) + self._force_refresh_bundles.add(bundle.name) + continue + if not bundle.path.is_dir(): + self.log.error( + "Published generation %s for bundle %s disappeared before it could be leased", + version_after_refresh, + bundle.name, + ) + self._force_refresh_bundles.add(bundle.name) + continue + self._published_bundle_versions[bundle.name] = version_after_refresh if previously_seen and pre_refresh_version == version_after_refresh: self.log.debug( "Bundle %s version not changed after refresh: %s", @@ -937,7 +1051,17 @@ def _refresh_dag_bundles(self, known_files: dict[str, set[DagFileInfo]]): self._bundle_version_data[bundle.name] = version_data_after_refresh found_files = { - DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path) + DagFileInfo( + rel_path=p, + bundle_name=bundle.name, + bundle_path=bundle.path, + bundle_version=( + version_after_refresh if bundle.refreshes_to_versioned_paths is True else None + ), + bundle_version_data=( + version_data_after_refresh if bundle.refreshes_to_versioned_paths is True else None + ), + ) for p in self._find_files_in_bundle(bundle) } @@ -1204,13 +1328,20 @@ def handle_removed_files(self, known_files: dict[str, set[DagFileInfo]]): def purge_removed_files_from_queue(self, present: set[DagFileInfo]): """Remove from queue any files no longer observed locally.""" present_keys = {file.presence_key for file in present} - self._file_queue = OrderedDict((x, None) for x in self._file_queue if x.presence_key in present_keys) + self._file_queue = OrderedDict( + (file, None) + for file in self._file_queue + if file in self._callback_to_execute + or (file.bundle_version is not None and file in present) + or (file.bundle_version is None and file.presence_key in present_keys) + ) stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue)) def remove_orphaned_file_stats(self, present: set[DagFileInfo]): """Remove the stats for any dag files that don't exist anymore.""" - present_keys = {file.presence_key for file in present} - stats_to_remove = {file for file in self._file_stats if file.presence_key not in present_keys} + stats_to_remove = { + file for file in self._file_stats if file not in present and file not in self._processors + } for file in stats_to_remove: del self._file_stats[file] @@ -1221,7 +1352,9 @@ def terminate_orphan_processes(self, present: set[DagFileInfo]): bundle_to_team = self._get_team_names({file.bundle_name for file in self._processors}) for file in list(self._processors.keys()): - if file.presence_key not in present_keys: + # An immutable generation can disappear from the latest file listing while a historical callback + # is still running against it. Its generation lease keeps the files alive; let the processor finish. + if file.bundle_version is None and file.presence_key not in present_keys: processor = self._processors.pop(file, None) if not processor: continue @@ -1285,12 +1418,32 @@ def handle_parsing_result( team_name=team_name, ) - if proc.parsing_result is not None: + is_stale_generation = ( + file.bundle_version is not None + and self._published_bundle_versions.get(file.bundle_name, file.bundle_version) + != file.bundle_version + ) + if proc.parsing_result is not None and is_stale_generation: + self.log.info( + "Discarding parse result for %s in superseded bundle %s generation %s", + file.rel_path, + file.bundle_name, + file.bundle_version, + ) + elif proc.parsing_result is not None: try: self.persist_parsing_result( bundle_name=file.bundle_name, - bundle_version=self._bundle_versions[file.bundle_name], - version_data=self._bundle_version_data.get(file.bundle_name), + bundle_version=( + file.bundle_version + if file.bundle_version is not None + else self._bundle_versions[file.bundle_name] + ), + version_data=( + file.bundle_version_data + if file.bundle_version is not None + else self._bundle_version_data.get(file.bundle_name) + ), parsing_result=proc.parsing_result, run_duration=run_duration, relative_fileloc=str(file.rel_path), @@ -1439,6 +1592,7 @@ def _create_process(self, dag_file: DagFileInfo) -> DagFileProcessorProcess: path=dag_file.absolute_path, bundle_path=cast("Path", dag_file.bundle_path), bundle_name=dag_file.bundle_name, + bundle_version=dag_file.bundle_version, dag_file_rel_path=str(dag_file.rel_path), callbacks=callback_to_execute_for_file, selector=self.selector, @@ -1448,6 +1602,42 @@ def _create_process(self, dag_file: DagFileInfo) -> DagFileProcessorProcess: client=self.client, ) + def _reconcile_bundle_version_locks(self, known_files: dict[str, set[DagFileInfo]]) -> None: + """Keep generation leases for every current, queued, or in-flight parse.""" + referenced_files = set(self._file_queue) + referenced_files.update(self._processors) + for files in known_files.values(): + referenced_files.update(files) + referenced_versions = { + (file.bundle_name, file.bundle_version) + for file in referenced_files + if file.bundle_version is not None + } + + for key in referenced_versions - self._bundle_version_locks.keys(): + self._acquire_bundle_version_lock(*key) + + for key in self._bundle_version_locks.keys() - referenced_versions: + lock = self._bundle_version_locks.pop(key) + try: + lock.release() + except Exception: + self.log.exception( + "Failed to release bundle generation lease for %s at version %s", + *key, + ) + + present = set().union(*known_files.values()) if known_files else set() + self.remove_orphaned_file_stats(present) + + def _acquire_bundle_version_lock(self, bundle_name: str, bundle_version: str | None) -> None: + """Acquire one manager-owned lease, deduplicated by bundle generation.""" + if bundle_version is None or (bundle_name, bundle_version) in self._bundle_version_locks: + return + lock = BundleVersionLock(bundle_name=bundle_name, bundle_version=bundle_version) + lock.acquire() + self._bundle_version_locks[(bundle_name, bundle_version)] = lock + def _start_new_processes(self): """Start more processors if we have enough slots and files to process.""" bundle_to_team = self._get_team_names({file.bundle_name for file in self._file_queue}) @@ -1480,14 +1670,19 @@ def _add_new_files_to_queue(self, known_files: dict[str, set[DagFileInfo]]): A "new" file is a file that has not been processed yet and is not currently being processed. """ new_files = [] - tracked_presence_keys = {file.presence_key for file in self._file_queue} - tracked_presence_keys.update(file.presence_key for file in self._file_stats) - tracked_presence_keys.update(file.presence_key for file in self._processors) + tracked_files = set(self._file_queue) | self._file_stats.keys() | self._processors.keys() + tracked_presence_keys = {file.presence_key for file in tracked_files if file.bundle_version is None} for files in known_files.values(): for file in files: - if file.presence_key not in tracked_presence_keys: - new_files.append(file) + if file.bundle_version is not None: + if file in tracked_files: + continue + tracked_files.add(file) + else: + if file.presence_key in tracked_presence_keys: + continue tracked_presence_keys.add(file.presence_key) + new_files.append(file) if new_files: self.log.info("Adding %d new files to the front of the queue", len(new_files)) diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 3805de9fdf55b..f72724a92348e 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -126,6 +126,9 @@ class DagFileParseRequest(BaseModel): bundle_name: str """Bundle name for team-specific executor validation.""" + bundle_version: str | None = None + """Exact bundle generation used for this parse.""" + callback_requests: list[CallbackRequest] = Field(default_factory=list) type: Literal["DagFileParseRequest"] = "DagFileParseRequest" @@ -226,7 +229,8 @@ def _parse_file_entrypoint(): task_runner.SUPERVISOR_COMMS = comms_decoder log = structlog.get_logger(logger_name="task") - result = _parse_file(msg, log) + with BundleVersionLock(bundle_name=msg.bundle_name, bundle_version=msg.bundle_version): + result = _parse_file(msg, log) if result is not None: comms_decoder.send(result) @@ -604,6 +608,7 @@ def start( # type: ignore[override] path: str | os.PathLike[str], bundle_path: Path, bundle_name: str, + bundle_version: str | None = None, dag_file_rel_path: str, callbacks: list[CallbackRequest], target: Callable[[], None] = _parse_file_entrypoint, @@ -633,7 +638,7 @@ def start( # type: ignore[override] **kwargs, ) proc.had_callbacks = bool(callbacks) # Track if this process had callbacks - proc._on_child_started(callbacks, path, bundle_path, bundle_name) + proc._on_child_started(callbacks, path, bundle_path, bundle_name, bundle_version) return proc def _on_child_started( @@ -642,11 +647,13 @@ def _on_child_started( path: str | os.PathLike[str], bundle_path: Path, bundle_name: str, + bundle_version: str | None, ) -> None: msg = DagFileParseRequest( file=os.fspath(path), bundle_path=bundle_path, bundle_name=bundle_name, + bundle_version=bundle_version, callback_requests=callbacks, ) self.send_msg(msg, request_id=0) diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 0c6fdbd1deb31..5df677513f5a9 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -45,7 +45,7 @@ from airflow._shared.timezones import timezone from airflow.callbacks.callback_requests import DagCallbackRequest -from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.dag_processing.bundles.base import BaseDagBundle, BundleVersion from airflow.dag_processing.bundles.manager import DagBundlesManager from airflow.dag_processing.collection import update_dag_parsing_results_in_db from airflow.dag_processing.dagbag import DagBag @@ -95,12 +95,17 @@ def _get_file_infos(files: list[str | Path]) -> list[DagFileInfo]: return [DagFileInfo(bundle_name="testing", bundle_path=TEST_DAGS_FOLDER, rel_path=Path(f)) for f in files] -def _get_versioned_file_info(file: str | Path, bundle_version: str = "v1") -> DagFileInfo: +def _get_versioned_file_info( + file: str | Path, + bundle_version: str = "v1", + bundle_version_data: dict | None = None, +) -> DagFileInfo: return DagFileInfo( bundle_name="testing", bundle_path=TEST_DAGS_FOLDER, rel_path=Path(file), bundle_version=bundle_version, + bundle_version_data=bundle_version_data, ) @@ -633,6 +638,7 @@ def test_purge_removed_files_keeps_versioned_callback_file_when_unversioned_file present_file = _get_file_infos(["callbacks.py"])[0] manager._file_queue = OrderedDict.fromkeys([versioned_file]) + manager._callback_to_execute[versioned_file] = [MagicMock()] manager.purge_removed_files_from_queue(present={present_file}) @@ -648,27 +654,24 @@ def test_purge_removed_files_drops_versioned_callback_file_when_truly_absent(sel assert manager._file_queue == OrderedDict() - def test_terminate_orphan_processes_keeps_versioned_callback_processor_when_unversioned_file_is_present( - self, - ): + def test_terminate_orphan_processes_keeps_versioned_callback_processor_when_file_is_absent(self): manager = DagFileProcessorManager(max_runs=1) versioned_file = _get_versioned_file_info("callbacks.py") - present_file = _get_file_infos(["callbacks.py"])[0] processor = MagicMock() manager._processors[versioned_file] = processor - manager.terminate_orphan_processes(present={present_file}) + manager.terminate_orphan_processes(present=set()) assert manager._processors == {versioned_file: processor} processor.kill.assert_not_called() def test_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(self): manager = DagFileProcessorManager(max_runs=1) - versioned_file = _get_versioned_file_info("callbacks with spaces.py") + unversioned_file = _get_file_infos(["callbacks with spaces.py"])[0] processor = MagicMock() - manager._processors[versioned_file] = processor + manager._processors[unversioned_file] = processor with mock.patch("airflow.dag_processing.manager.stats.decr") as stats_decr_mock: manager.terminate_orphan_processes(present=set()) @@ -683,11 +686,11 @@ def test_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(se def test_terminate_orphan_processes_tolerates_stale_file_handle_on_close(self): """A stale NFS file handle on close (e.g. OpenShift) must not crash the manager.""" manager = DagFileProcessorManager(max_runs=1) - versioned_file = _get_versioned_file_info("callbacks.py") + unversioned_file = _get_file_infos(["callbacks.py"])[0] processor, _ = self.mock_processor() processor.logger_filehandle.close.side_effect = OSError(116, "Stale file handle") - manager._processors[versioned_file] = processor + manager._processors[unversioned_file] = processor with ( mock.patch.object(type(processor), "kill"), @@ -697,7 +700,7 @@ def test_terminate_orphan_processes_tolerates_stale_file_handle_on_close(self): assert manager._processors == {} - def test_remove_orphaned_file_stats_keeps_versioned_callback_stats_when_unversioned_file_is_present(self): + def test_remove_orphaned_file_stats_drops_old_generation_when_current_file_is_present(self): manager = DagFileProcessorManager(max_runs=1) versioned_file = _get_versioned_file_info("callbacks.py") present_file = _get_file_infos(["callbacks.py"])[0] @@ -706,7 +709,19 @@ def test_remove_orphaned_file_stats_keeps_versioned_callback_stats_when_unversio manager.remove_orphaned_file_stats(present={present_file}) - assert manager._file_stats == {versioned_file: DagFileStat()} + assert manager._file_stats == {} + + def test_remove_orphaned_file_stats_keeps_generation_with_active_processor(self): + manager = DagFileProcessorManager(max_runs=1) + old_file = _get_versioned_file_info("callbacks.py", bundle_version="v1") + current_file = _get_versioned_file_info("callbacks.py", bundle_version="v2") + + manager._file_stats[old_file] = DagFileStat() + manager._processors[old_file] = MagicMock() + + manager.remove_orphaned_file_stats(present={current_file}) + + assert manager._file_stats == {old_file: DagFileStat()} def test_remove_orphaned_file_stats_drops_versioned_callback_stats_when_truly_absent(self): manager = DagFileProcessorManager(max_runs=1) @@ -849,29 +864,26 @@ def test_add_new_files_to_queue_behavior(self): # file_1 should remain (already in queue) assert list(manager._file_queue) == [file_2, file_1] - def test_add_new_files_to_queue_skips_versioned_files_already_represented(self): + def test_add_new_files_to_queue_adds_new_immutable_generation(self): manager = DagFileProcessorManager(max_runs=1) queued_versioned_file = _get_versioned_file_info("file_1.py") processed_versioned_file = _get_versioned_file_info("file_3.py") parsed_versioned_file = _get_versioned_file_info("file_4.py") - new_file = _get_file_infos(["file_2.py"])[0] + current_files = { + _get_versioned_file_info(name, bundle_version="v2") + for name in ("file_1.py", "file_2.py", "file_3.py", "file_4.py") + } manager._file_queue = OrderedDict.fromkeys([queued_versioned_file]) manager._processors[processed_versioned_file] = MagicMock() manager._file_stats[parsed_versioned_file] = DagFileStat(num_dags=1) - known_files = { - "testing": { - _get_file_infos(["file_1.py"])[0], - new_file, - _get_file_infos(["file_3.py"])[0], - _get_file_infos(["file_4.py"])[0], - } - } + known_files = {"testing": current_files} manager._add_new_files_to_queue(known_files) - assert list(manager._file_queue) == [new_file, queued_versioned_file] + assert set(manager._file_queue) == current_files | {queued_versioned_file} + assert list(manager._file_queue)[-1] == queued_versioned_file @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"}) @mock.patch("airflow.utils.file.os.path.getmtime", new=mock_get_mtime) @@ -1129,6 +1141,76 @@ def claim_priority_files(self) -> list[DagFileInfo]: assert manager._file_queue == OrderedDict.fromkeys([file1, file2]) assert manager._force_refresh_bundles == {"dags-folder"} + def test_priority_file_is_rebound_to_refreshed_immutable_generation(self, tmp_path): + old_file = DagFileInfo( + bundle_name="versioned", + rel_path=Path("dag.py"), + bundle_path=tmp_path / "before-refresh", + bundle_version="v1", + bundle_version_data={"manifest": "v1"}, + ) + bundle = MagicMock( + name="versioned", + path=tmp_path / "versions" / "v2", + refreshes_to_versioned_paths=True, + ) + bundle.name = "versioned" + bundle.get_current_version.return_value = BundleVersion(version="v2", data={"manifest": "v2"}) + manager = DagFileProcessorManager(max_runs=1) + manager._dag_bundles = [bundle] + manager._bundle_versions["versioned"] = "v2" + manager._bundle_version_data["versioned"] = {"manifest": "v2"} + + manager._queue_requested_files_for_parsing([old_file]) + + [queued_file] = manager._file_queue + assert queued_file.bundle_path == tmp_path / "versions" / "v2" + assert queued_file.bundle_version == "v2" + assert queued_file.bundle_version_data == {"manifest": "v2"} + + def test_priority_file_uses_published_snapshot_when_state_persistence_failed(self, tmp_path): + requested_file = DagFileInfo( + bundle_name="versioned", + rel_path=Path("dag.py"), + bundle_path=tmp_path / "versions" / "v1", + bundle_version="v1", + ) + published_file = DagFileInfo( + bundle_name="versioned", + rel_path=Path("dag.py"), + bundle_path=tmp_path / "versions" / "v2", + bundle_version="v2", + bundle_version_data={"manifest": "v2"}, + ) + bundle = MagicMock(refreshes_to_versioned_paths=True) + bundle.name = "versioned" + manager = DagFileProcessorManager(max_runs=1) + manager._dag_bundles = [bundle] + manager._bundle_versions["versioned"] = "v1" + manager._bundle_version_data["versioned"] = {"manifest": "v1"} + + manager._queue_requested_files_for_parsing( + [requested_file], known_files={"versioned": {published_file}} + ) + + assert list(manager._file_queue) == [published_file] + + def test_priority_file_is_not_queued_without_a_published_generation(self, tmp_path): + requested_file = DagFileInfo( + bundle_name="versioned", + rel_path=Path("dag.py"), + bundle_path=tmp_path / "versions" / "v1", + bundle_version="v1", + ) + bundle = MagicMock(refreshes_to_versioned_paths=True) + bundle.name = "versioned" + manager = DagFileProcessorManager(max_runs=1) + manager._dag_bundles = [bundle] + + manager._queue_requested_files_for_parsing([requested_file], known_files={}) + + assert manager._file_queue == OrderedDict() + def test_request_bundle_refresh_marks_bundles_for_refresh(self): """`request_bundle_refresh` adds the bundles to the force-refresh set.""" manager = DagFileProcessorManager(max_runs=1) @@ -1497,6 +1579,66 @@ def test_cleanup_stale_bundle_versions(self, mock_bundle_manager): manager.cleanup_stale_bundle_versions() mock_bundle_manager.return_value.remove_stale_bundle_versions.assert_called_once_with() + def test_reconcile_bundle_version_locks_tracks_current_queued_and_inflight_versions(self): + manager = DagFileProcessorManager(max_runs=1) + current = _get_versioned_file_info("current.py", bundle_version="v3") + queued = _get_versioned_file_info("queued.py", bundle_version="v2") + inflight = _get_versioned_file_info("inflight.py", bundle_version="v1") + manager._file_queue[queued] = None + manager._processors[inflight] = MagicMock() + + created_locks = {} + + def create_lock(*, bundle_name, bundle_version): + lock = MagicMock() + created_locks[(bundle_name, bundle_version)] = lock + return lock + + with mock.patch( + "airflow.dag_processing.manager.BundleVersionLock", side_effect=create_lock + ) as lock_class: + manager._reconcile_bundle_version_locks({"testing": {current}}) + + assert set(manager._bundle_version_locks) == { + ("testing", "v1"), + ("testing", "v2"), + ("testing", "v3"), + } + assert lock_class.call_count == 3 + for lock in created_locks.values(): + lock.acquire.assert_called_once_with() + + manager._file_queue.clear() + manager._processors.clear() + manager._reconcile_bundle_version_locks({"testing": {current}}) + + assert set(manager._bundle_version_locks) == {("testing", "v3")} + created_locks[("testing", "v1")].release.assert_called_once_with() + created_locks[("testing", "v2")].release.assert_called_once_with() + created_locks[("testing", "v3")].release.assert_not_called() + + def test_after_run_releases_bundle_version_locks(self): + manager = DagFileProcessorManager(max_runs=1) + locks = {("testing", "v1"): MagicMock(), ("testing", "v2"): MagicMock()} + manager._bundle_version_locks.update(locks) + + manager.after_run() + + assert manager._bundle_version_locks == {} + for lock in locks.values(): + lock.release.assert_called_once_with() + + def test_reconcile_bundle_version_lock_acquisition_failure_is_fail_closed(self): + manager = DagFileProcessorManager(max_runs=1) + current = _get_versioned_file_info("current.py") + + with mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class: + lock_class.return_value.acquire.side_effect = OSError("read-only tracking directory") + with pytest.raises(OSError, match="read-only tracking directory"): + manager._reconcile_bundle_version_locks({"testing": {current}}) + + assert manager._bundle_version_locks == {} + @pytest.mark.parametrize( ("log_target", "expected_subprocess_logs_to_stdout"), [ @@ -1608,7 +1750,7 @@ def test_terminate_normalizes_file_path_stats_tag(self): def test_handle_parsing_result_provides_its_own_session_when_caller_omits(self): """``handle_parsing_result`` is wrapped in ``@provide_session`` so subclasses overriding it can run without a caller-supplied session.""" manager = DagFileProcessorManager(max_runs=1) - file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER) + file = _get_versioned_file_info("abc.txt") manager._file_stats[file] = DagFileStat() manager._bundle_versions["testing"] = "v1" @@ -1625,7 +1767,7 @@ def test_handle_parsing_result_provides_its_own_session_when_caller_omits(self): def test_handle_parsing_result_throttles_retry_when_first_persist_fails(self, session): """Persist errors should throttle retries without claiming persistence succeeded.""" manager = DagFileProcessorManager(max_runs=1) - file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER) + file = _get_versioned_file_info("abc.txt") original_stat = DagFileStat() manager._file_stats[file] = original_stat manager._bundle_versions["testing"] = "v1" @@ -1648,7 +1790,7 @@ def test_handle_parsing_result_throttles_retry_when_first_persist_fails(self, se def test_handle_parsing_result_updates_stats_after_successful_persist(self, session): manager = DagFileProcessorManager(max_runs=1) - file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER) + file = _get_versioned_file_info("abc.txt") original_stat = DagFileStat( num_dags=1, import_errors=0, @@ -1682,6 +1824,42 @@ def test_handle_parsing_result_updates_stats_after_successful_persist(self, sess assert manager._file_stats[file].last_finish_time > original_stat.last_finish_time assert manager._file_stats[file].num_dags == 0 + def test_handle_parsing_result_persists_file_generation_snapshot(self, session): + manager = DagFileProcessorManager(max_runs=1) + old_data = {"manifest_version_id": "manifest-v1"} + file = _get_versioned_file_info( + "abc.txt", + bundle_version="v1", + bundle_version_data=old_data, + ) + manager._file_stats[file] = DagFileStat() + manager._bundle_versions["testing"] = "v2" + manager._bundle_version_data["testing"] = {"manifest_version_id": "manifest-v2"} + processor, _ = self.mock_processor(start_time=time.monotonic() - 1) + processor.had_callbacks = False + processor.parsing_result = DagFileParsingResult(fileloc="abc.txt", serialized_dags=[]) + + with mock.patch.object(manager, "persist_parsing_result") as mock_persist: + manager.handle_parsing_result(file, processor, session=session) + + assert mock_persist.call_args.kwargs["bundle_version"] == "v1" + assert mock_persist.call_args.kwargs["version_data"] == old_data + + def test_handle_parsing_result_discards_superseded_immutable_generation(self, session): + manager = DagFileProcessorManager(max_runs=1) + old_file = _get_versioned_file_info("abc.txt", bundle_version="v1") + manager._file_stats[old_file] = DagFileStat() + manager._published_bundle_versions["testing"] = "v2" + processor, _ = self.mock_processor(start_time=time.monotonic() - 1) + processor.had_callbacks = False + processor.parsing_result = DagFileParsingResult(fileloc="abc.txt", serialized_dags=[]) + + with mock.patch.object(manager, "persist_parsing_result") as mock_persist: + manager.handle_parsing_result(old_file, processor, session=session) + + mock_persist.assert_not_called() + assert manager._file_stats[old_file].run_count == 1 + def test_collect_results_processes_remaining_files_when_one_persist_fails(self, session): manager = DagFileProcessorManager(max_runs=1) file_a = DagFileInfo(bundle_name="testing", rel_path=Path("a.py"), bundle_path=TEST_DAGS_FOLDER) @@ -1754,6 +1932,7 @@ def test_collect_results_tolerates_stale_file_handle_on_close(self): "file": "/opt/airflow/dags/test_dag.py", "bundle_path": "/opt/airflow/dags", "bundle_name": "testing", + "bundle_version": None, "callback_requests": [], "type": "DagFileParseRequest", }, @@ -1775,6 +1954,7 @@ def test_collect_results_tolerates_stale_file_handle_on_close(self): "file": "/opt/airflow/dags/dag_callback_dag.py", "bundle_path": "/opt/airflow/dags", "bundle_name": "testing", + "bundle_version": None, "callback_requests": [ { "filepath": "dag_callback_dag.py", @@ -1799,7 +1979,11 @@ def test_serialize_callback_requests(self, callbacks, path, expected_body): processor, read_socket = self.mock_processor() processor._on_child_started( - callbacks, path, bundle_path=Path("/opt/airflow/dags"), bundle_name="testing" + callbacks, + path, + bundle_path=Path("/opt/airflow/dags"), + bundle_name="testing", + bundle_version=None, ) read_socket.settimeout(0.1) @@ -2349,6 +2533,7 @@ def test_callback_queue(self, mock_get_logger, configure_testing_dag_bundle): path=Path(dag2_path.bundle_path, dag2_path.rel_path), bundle_path=dag2_path.bundle_path, bundle_name="testing", + bundle_version=dag2_path.bundle_version, dag_file_rel_path=str(dag2_path.rel_path), callbacks=[dag2_req1], selector=mock.ANY, @@ -2362,6 +2547,7 @@ def test_callback_queue(self, mock_get_logger, configure_testing_dag_bundle): path=Path(dag1_path.bundle_path, dag1_path.rel_path), bundle_path=dag1_path.bundle_path, bundle_name="testing", + bundle_version=dag1_path.bundle_version, dag_file_rel_path=str(dag1_path.rel_path), callbacks=[dag1_req1, dag1_req2], selector=mock.ANY, @@ -2381,6 +2567,7 @@ def test_prepare_callback_bundle_initializes_versioned_bundle(self, mock_bundle_ bundle = MagicMock(spec=BaseDagBundle) bundle.supports_versioning = True mock_bundle_manager.return_value.get_bundle.return_value = bundle + events = [] request = DagCallbackRequest( filepath="file1.py", @@ -2392,7 +2579,14 @@ def test_prepare_callback_bundle_initializes_versioned_bundle(self, mock_bundle_ msg=None, ) - assert manager.prepare_callback_bundle(request) is bundle + with mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class: + lock_class.return_value.acquire.side_effect = lambda: events.append("lease") + bundle.initialize.side_effect = lambda: events.append("initialize") + + assert manager.prepare_callback_bundle(request) is bundle + + assert events == ["lease", "initialize"] + assert manager._bundle_version_locks == {("testing", "some_commit_hash"): lock_class.return_value} bundle.initialize.assert_called_once() @mock.patch("airflow.dag_processing.manager.DagBundlesManager") @@ -2414,7 +2608,8 @@ def test_prepare_callback_bundle_forwards_version_data(self, mock_bundle_manager msg=None, ) - manager.prepare_callback_bundle(request) + with mock.patch("airflow.dag_processing.manager.BundleVersionLock"): + manager.prepare_callback_bundle(request) mock_bundle_manager.return_value.get_bundle.assert_called_once_with( name="testing", version="some_commit_hash", version_data=version_data ) @@ -2494,8 +2689,62 @@ def test_prepare_callback_bundle_returns_none_when_initialize_fails(self, mock_b msg=None, ) - assert manager.prepare_callback_bundle(request) is None + with mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class: + assert manager.prepare_callback_bundle(request) is None + bundle.initialize.assert_called_once() + lock_class.return_value.acquire.assert_called_once_with() + lock_class.return_value.release.assert_called_once_with() + assert manager._bundle_version_locks == {} + + @mock.patch("airflow.dag_processing.manager.DagBundlesManager") + def test_prepare_callback_bundle_release_failure_does_not_escape(self, mock_bundle_manager): + manager = DagFileProcessorManager(max_runs=1) + bundle = MagicMock(spec=BaseDagBundle) + bundle.supports_versioning = True + bundle.initialize.side_effect = RuntimeError("clone failed") + mock_bundle_manager.return_value.get_bundle.return_value = bundle + request = DagCallbackRequest( + filepath="file1.py", + dag_id="dag1", + run_id="run1", + is_failure_callback=False, + bundle_name="testing", + bundle_version="some_commit_hash", + msg=None, + ) + + with mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class: + lock_class.return_value.release.side_effect = OSError("tracking write failed") + + assert manager.prepare_callback_bundle(request) is None + + assert manager._bundle_version_locks == {} + + @mock.patch("airflow.dag_processing.manager.DagBundlesManager") + def test_prepare_callback_bundle_acquire_failure_skips_initialize(self, mock_bundle_manager): + manager = DagFileProcessorManager(max_runs=1) + bundle = MagicMock(spec=BaseDagBundle) + bundle.supports_versioning = True + mock_bundle_manager.return_value.get_bundle.return_value = bundle + request = DagCallbackRequest( + filepath="file1.py", + dag_id="dag1", + run_id="run1", + is_failure_callback=False, + bundle_name="testing", + bundle_version="some_commit_hash", + msg=None, + ) + + with mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class: + lock_class.return_value.acquire.side_effect = OSError("tracking directory is read-only") + + assert manager.prepare_callback_bundle(request) is None + + bundle.initialize.assert_not_called() + lock_class.return_value.release.assert_not_called() + assert manager._bundle_version_locks == {} @mock.patch("airflow.dag_processing.manager.DagBundlesManager") def test_add_callback_queues_file_info_on_success(self, mock_bundle_manager): @@ -2515,7 +2764,8 @@ def test_add_callback_queues_file_info_on_success(self, mock_bundle_manager): msg=None, ) - manager._add_callback_to_queue(request) + with mock.patch("airflow.dag_processing.manager.BundleVersionLock"): + manager._add_callback_to_queue(request) bundle.initialize.assert_called_once() assert manager._callback_to_execute @@ -3409,6 +3659,98 @@ def test_refresh_dag_bundles_versioned_first_seen_skips_short_circuit(self): mock_update.assert_called_once_with("mock_bundle", last_refreshed=mock.ANY, version="v1") assert manager._bundle_versions["mock_bundle"] == "v1" + def test_refresh_immutable_bundle_acquires_generation_lease_before_file_scan(self, tmp_path): + manager = DagFileProcessorManager(max_runs=1) + bundle = self._make_refresh_bundle( + supports_versioning=True, + current_version=BundleVersion(version="v2"), + ) + bundle.refreshes_to_versioned_paths = True + bundle.path = tmp_path / "v2" + bundle.path.mkdir() + manager._dag_bundles = [bundle] + events = [] + + def find_files(_bundle): + events.append("scan") + assert ("mock_bundle", "v2") in manager._bundle_version_locks + return [] + + with ( + mock.patch.object( + manager, "get_bundle_state", return_value=BundleState(last_refreshed=None, version="v1") + ), + mock.patch.object(manager, "update_bundle_state"), + mock.patch.object(manager, "_find_files_in_bundle", side_effect=find_files), + mock.patch.object(manager, "deactivate_deleted_dags"), + mock.patch.object(manager, "clear_orphaned_import_errors"), + mock.patch.object(manager, "handle_removed_files"), + mock.patch.object(manager, "_resort_file_queue"), + mock.patch.object(manager, "_add_new_files_to_queue"), + mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class, + ): + lock_class.return_value.acquire.side_effect = lambda: events.append("lease") + manager._refresh_dag_bundles({}) + + assert events == ["lease", "scan"] + assert manager._published_bundle_versions == {"mock_bundle": "v2"} + + def test_refresh_immutable_bundle_lease_failure_skips_file_scan(self, tmp_path): + manager = DagFileProcessorManager(max_runs=1) + bundle = self._make_refresh_bundle( + supports_versioning=True, + current_version=BundleVersion(version="v2"), + ) + bundle.refreshes_to_versioned_paths = True + bundle.path = tmp_path / "v2" + bundle.path.mkdir() + manager._dag_bundles = [bundle] + + with ( + mock.patch.object( + manager, "get_bundle_state", return_value=BundleState(last_refreshed=None, version="v1") + ), + mock.patch.object(manager, "update_bundle_state"), + mock.patch.object(manager, "_find_files_in_bundle") as find_files, + mock.patch("airflow.dag_processing.manager.BundleVersionLock") as lock_class, + ): + lock_class.return_value.acquire.side_effect = OSError("tracking directory is read-only") + manager._refresh_dag_bundles({}) + + find_files.assert_not_called() + assert manager._bundle_version_locks == {} + assert manager._published_bundle_versions == {} + assert manager._force_refresh_bundles == {"mock_bundle"} + + def test_refresh_immutable_bundle_retries_if_generation_disappears_before_lease(self, tmp_path): + manager = DagFileProcessorManager(max_runs=1) + bundle = self._make_refresh_bundle( + supports_versioning=True, + current_version=BundleVersion(version="v2"), + ) + bundle.refreshes_to_versioned_paths = True + bundle.path = tmp_path / "v2" + bundle.path.mkdir() + manager._dag_bundles = [bundle] + + def remove_generation(): + shutil.rmtree(bundle.path) + + with ( + mock.patch.object( + manager, "get_bundle_state", return_value=BundleState(last_refreshed=None, version="v1") + ), + mock.patch.object(manager, "update_bundle_state") as update_state, + mock.patch.object(manager, "_find_files_in_bundle") as find_files, + mock.patch.object(manager, "_acquire_bundle_version_lock", side_effect=remove_generation), + ): + manager._refresh_dag_bundles({}) + + update_state.assert_not_called() + find_files.assert_not_called() + assert manager._published_bundle_versions == {} + assert manager._force_refresh_bundles == {"mock_bundle"} + def test_refresh_dag_bundles_get_bundle_state_failure_skips_bundle(self): """A failure in get_bundle_state() logs and skips the bundle without aborting the loop.""" manager = DagFileProcessorManager(max_runs=1) diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py index f54b82fef8291..3d818ff588751 100644 --- a/airflow-core/tests/unit/dag_processing/test_processor.py +++ b/airflow-core/tests/unit/dag_processing/test_processor.py @@ -585,6 +585,49 @@ def test_parse_file_entrypoint_parses_dag_callbacks(mocker): ] +def test_parse_file_entrypoint_holds_bundle_generation_lease(mocker): + request = DagFileParseRequest( + file="/files/dags/dag.py", + bundle_path="/files/dags", + bundle_name="testing", + bundle_version="v1", + ) + decoder = MagicMock() + decoder._get_response.return_value = request + decoder_class = mocker.patch("airflow.sdk.execution_time.comms.CommsDecoder") + decoder_class.__getitem__.return_value.return_value = decoder + lock = mocker.patch("airflow.dag_processing.processor.BundleVersionLock") + parse_file = mocker.patch( + "airflow.dag_processing.processor._parse_file", + return_value=DagFileParsingResult(fileloc=request.file, serialized_dags=[]), + ) + + _parse_file_entrypoint() + + lock.assert_called_once_with(bundle_name="testing", bundle_version="v1") + lock.return_value.__enter__.assert_called_once_with() + lock.return_value.__exit__.assert_called_once_with(None, None, None) + parse_file.assert_called_once_with(request, mocker.ANY) + + +def test_parse_process_forwards_bundle_version_to_child(): + process = MagicMock(spec=DagFileProcessorProcess) + + DagFileProcessorProcess._on_child_started( + process, + callbacks=[], + path="/files/dags/dag.py", + bundle_path=pathlib.Path("/files/dags"), + bundle_name="testing", + bundle_version="v1", + ) + + request = process.send_msg.call_args.args[0] + assert request.bundle_version == "v1" + assert request.bundle_name == "testing" + assert request.bundle_path == pathlib.Path("/files/dags") + + def test_parse_file_with_dag_callbacks(spy_agency): from airflow import DAG diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json index 85f8b7cfa4e2b..e9c84e4f4871d 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json +++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json @@ -933,6 +933,18 @@ "title": "Bundle Name", "type": "string" }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Bundle Version" + }, "callback_requests": { "items": { "discriminator": { diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py index 7e5ce93f86bdc..9e566e2a6059e 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py @@ -39,11 +39,16 @@ def get_bundle() -> VersionBundle: from airflow.sdk.execution_time.schema.versions.v2026_10_30 import ( AddArgBindingsToSupervisorTIRunContext, + AddBundleVersionToDagFileParseRequest, ) return VersionBundle( HeadVersion(), - Version("2026-10-30", AddArgBindingsToSupervisorTIRunContext), + Version( + "2026-10-30", + AddArgBindingsToSupervisorTIRunContext, + AddBundleVersionToDagFileParseRequest, + ), Version("2026-06-16"), ) diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py index e6b93f5dea805..84420ea737dd5 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py @@ -19,6 +19,7 @@ from cadwyn import VersionChange, schema +from airflow.dag_processing.processor import DagFileParseRequest # noqa: SDK002 from airflow.sdk.api.datamodels._generated import TIRunContext @@ -34,3 +35,13 @@ class AddArgBindingsToSupervisorTIRunContext(VersionChange): description = __doc__ instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("arg_bindings").didnt_exist,) + + +class AddBundleVersionToDagFileParseRequest(VersionChange): + """Add the immutable bundle version to Dag file parse requests.""" + + description = __doc__ + + instructions_to_migrate_to_previous_version = ( + schema(DagFileParseRequest).field("bundle_version").didnt_exist, + ) diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py b/task-sdk/src/airflow/sdk/execution_time/task_runner.py index eeb2788062248..d7c99438a836e 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -2420,30 +2420,29 @@ def main(): span = INVALID_SPAN with stack: try: - try: - log.info("::group::Pre Execute") - startup_details = get_startup_details() - - span_ctx_mgr = _make_task_span(msg=startup_details) - span = stack.enter_context(span_ctx_mgr) - ti, context, log = startup(msg=startup_details) - except AirflowRescheduleException as reschedule: - log.warning("Rescheduling task during startup, marking task as UP_FOR_RESCHEDULE") - SUPERVISOR_COMMS.send( - msg=RescheduleTask( - reschedule_date=reschedule.reschedule_date, - end_date=datetime.now(tz=timezone.utc), - ) - ) - span.record_exception(reschedule) - span.set_status( - Status(StatusCode.ERROR, description=f"Exception: {type(reschedule).__name__}") - ) - sys.exit(0) + log.info("::group::Pre Execute") + startup_details = get_startup_details() with BundleVersionLock( - bundle_name=ti.bundle_instance.name, - bundle_version=ti.bundle_instance.version, + bundle_name=startup_details.bundle_info.name, + bundle_version=startup_details.bundle_info.version, ): + try: + span_ctx_mgr = _make_task_span(msg=startup_details) + span = stack.enter_context(span_ctx_mgr) + ti, context, log = startup(msg=startup_details) + except AirflowRescheduleException as reschedule: + log.warning("Rescheduling task during startup, marking task as UP_FOR_RESCHEDULE") + SUPERVISOR_COMMS.send( + msg=RescheduleTask( + reschedule_date=reschedule.reschedule_date, + end_date=datetime.now(tz=timezone.utc), + ) + ) + span.record_exception(reschedule) + span.set_status( + Status(StatusCode.ERROR, description=f"Exception: {type(reschedule).__name__}") + ) + sys.exit(0) state, _, error = run(ti, context, log) context["exception"] = error # run() funnels every failure path into `error` rather than diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py index cc9fb77e08921..14caa4e9561b4 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py +++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py @@ -458,6 +458,7 @@ def test_main_sends_reschedule_task_when_startup_reschedules( sentry_integration="", ) mock_get_startup_details.return_value = what + mock_startup.side_effect = AirflowRescheduleException(reschedule_date=reschedule_date) # Move time @@ -526,12 +527,17 @@ def test_main_marks_worker_span_error_on_failure( ) mock_get_startup_details.return_value = what + events = [] + mock_bundle_lock.return_value.__enter__.side_effect = lambda: events.append("lease-entered") + mock_bundle_lock.return_value.__exit__.side_effect = lambda *args: events.append("lease-exited") + ti = mock.Mock() ti.bundle_instance.name = "my-bundle" ti.bundle_instance.version = None ti._terminal_state_send_failed = False - mock_startup.return_value = (ti, {}, mock.Mock()) - mock_run.return_value = (state, mock.Mock(), error) + mock_startup.side_effect = lambda **kwargs: events.append("startup") or (ti, {}, mock.Mock()) + mock_run.side_effect = lambda *args, **kwargs: events.append("run") or (state, mock.Mock(), error) + mock_finalize.side_effect = lambda *args: events.append("finalize") exporter = InMemorySpanExporter() provider = TracerProvider() @@ -541,6 +547,10 @@ def test_main_marks_worker_span_error_on_failure( with mock.patch("airflow.sdk.execution_time.task_runner.tracer", t): task_runner.main() + mock_bundle_lock.assert_called_once_with(bundle_name="my-bundle", bundle_version=None) + mock_bundle_lock.return_value.__exit__.assert_called_once() + assert events == ["lease-entered", "startup", "run", "finalize", "lease-exited"] + worker = {s.name: s for s in exporter.get_finished_spans()}["worker.my_task"] if expect_error_status: assert worker.status.status_code == trace.StatusCode.ERROR From ad5d8670c5f8a5b6165ce4de0e2877ef05f3467d Mon Sep 17 00:00:00 2001 From: khup Date: Fri, 28 Aug 2026 14:21:10 -0400 Subject: [PATCH 3/4] Keep callbacks pinned to their Dag run bundle version A later Dag parse can advance the current DagVersion while an existing run must remain reproducible. Resolving callback metadata against the run's copied version prevents internally inconsistent version and version-data pairs. --- .../src/airflow/jobs/scheduler_job_runner.py | 49 ++++++------------- .../src/airflow/models/dag_version.py | 6 ++- .../src/airflow/models/serialized_dag.py | 5 +- airflow-core/src/airflow/models/trigger.py | 14 ++---- .../tests/unit/executors/test_workloads.py | 2 + .../tests/unit/jobs/test_scheduler_job.py | 37 +++++++++----- .../tests/unit/models/test_dag_version.py | 10 +++- .../tests/unit/models/test_trigger.py | 23 +++++++++ 8 files changed, 82 insertions(+), 64 deletions(-) diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index f8b2df4aac2ca..9f9aa27792fac 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -209,16 +209,11 @@ def _resolve_ti_callback_bundle_info(ti: TaskInstance) -> tuple[str, str | None, Used by the heartbeat-timeout purge path. Encapsulates the bundle-pinning semantics: fall back to ``dag_model`` for legacy tasks with no ``dag_version`` (pre-AIP-66 migrations), and leave the bundle version unpinned when the dag run itself wasn't pinned (``disable_bundle_versioning``), - so the callback runs against the same code as the task did. ``process_executor_events`` inlines - the same resolution for its externally-killed-task path. + so the callback runs against the same code as the task did. """ bundle_name = ti.dag_version.bundle_name if ti.dag_version else ti.dag_model.bundle_name - bundle_version = ( - ti.dag_version.bundle_version - if ti.dag_version and ti.dag_run.bundle_version is not None - else ti.dag_run.bundle_version - ) - version_data = _resolve_version_data(ti.dag_version, ti.dag_run.bundle_version) + bundle_version = ti.dag_run.bundle_version + version_data = _resolve_version_data(ti.dag_version, bundle_version) return bundle_name, bundle_version, version_data @@ -1626,15 +1621,7 @@ def process_executor_events( # Safely extract bundle info: prefer dag_version when available, # fall back to dag_model/dag_run for legacy tasks migrated from # Airflow 2 where dag_version may be None (AIP-66). - _bundle_name = ti.dag_version.bundle_name if ti.dag_version else ti.dag_model.bundle_name - # Mirror dag_run pinning: if the run wasn't pinned (e.g. dag.disable_bundle_versioning=True), - # leave the callback unpinned so it runs against the same code as the task. - _bundle_version = ( - ti.dag_version.bundle_version - if ti.dag_version and ti.dag_run.bundle_version is not None - else ti.dag_run.bundle_version - ) - _version_data = _resolve_version_data(ti.dag_version, ti.dag_run.bundle_version) + _bundle_name, _bundle_version, _version_data = _resolve_ti_callback_bundle_info(ti) # Backfill dag_version_id for legacy tasks (Pydantic requires uuid.UUID). if not _ensure_ti_has_dag_version_id(ti, session, cls.logger()): continue @@ -1679,13 +1666,11 @@ def process_executor_events( ) # Safely extract bundle info with fallback for legacy tasks # (dag_version may be None after Airflow 2 → 3 migration). - _email_bundle_name = ( - ti.dag_version.bundle_name if ti.dag_version else ti.dag_model.bundle_name - ) - _email_bundle_version = ( - ti.dag_version.bundle_version if ti.dag_version else ti.dag_run.bundle_version - ) - _email_version_data = _resolve_version_data(ti.dag_version, ti.dag_run.bundle_version) + ( + _email_bundle_name, + _email_bundle_version, + _email_version_data, + ) = _resolve_ti_callback_bundle_info(ti) # Backfill dag_version_id for legacy tasks (Pydantic requires uuid.UUID). if not _ensure_ti_has_dag_version_id(ti, session, cls.logger()): continue @@ -3200,17 +3185,11 @@ def _maybe_requeue_stuck_ti(self, *, ti, session, executor): ti = session.merge(ti) # Safely extract bundle info with fallback for legacy tasks # (dag_version may be None after Airflow 2 → 3 migration). - _stuck_bundle_name = ( - ti.dag_version.bundle_name if ti.dag_version else ti.dag_model.bundle_name - ) - # Mirror dag_run pinning: if the run wasn't pinned (e.g. dag.disable_bundle_versioning=True), - # leave the callback unpinned so it runs against the same code as the task. - _stuck_bundle_version = ( - ti.dag_version.bundle_version - if ti.dag_version and ti.dag_run.bundle_version is not None - else ti.dag_run.bundle_version - ) - _stuck_version_data = _resolve_version_data(ti.dag_version, ti.dag_run.bundle_version) + ( + _stuck_bundle_name, + _stuck_bundle_version, + _stuck_version_data, + ) = _resolve_ti_callback_bundle_info(ti) # Backfill dag_version_id for legacy tasks (Pydantic requires uuid.UUID). # Note: we cannot use `continue` here because this method is not # inside a loop. If backfilling fails we simply skip the callback. diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py index 602364e71b658..7c30952503616 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -253,6 +253,10 @@ def _resolve_version_data( # Expose version_data only when the run is pinned (bundle_version set) and a DagVersion is # present, so the bundle initializes against the exact version the run used. Unpinned runs # follow the latest bundle state, and legacy rows have no DagVersion. - if dag_version is not None and bundle_version is not None: + if ( + dag_version is not None + and bundle_version is not None + and dag_version.bundle_version == bundle_version + ): return dag_version.version_data return None diff --git a/airflow-core/src/airflow/models/serialized_dag.py b/airflow-core/src/airflow/models/serialized_dag.py index d49c168013f7c..53686e526eca3 100644 --- a/airflow-core/src/airflow/models/serialized_dag.py +++ b/airflow-core/src/airflow/models/serialized_dag.py @@ -703,9 +703,8 @@ def write_dag( if serialized_dag_hash == new_dag_hash and dag_version and dag_version.bundle_name == bundle_name: # Serialized content is unchanged, so we don't create a new DagVersion. - # But if the bundle advanced, refresh the latest version's pointer in place — tasks resolve - # their code from ``ti.dag_version.bundle_version`` at run time, so a stale - # pointer makes runs execute an outdated commit. + # If the bundle advanced, refresh the latest DagVersion in place for future runs. Existing + # DagRuns keep their copied ``bundle_version`` and continue to resolve the source they started with. bundle_metadata_changed = ( dag_version.bundle_version != bundle_version or dag_version.version_data != version_data ) diff --git a/airflow-core/src/airflow/models/trigger.py b/airflow-core/src/airflow/models/trigger.py index 75578974db826..79f261fde2caf 100644 --- a/airflow-core/src/airflow/models/trigger.py +++ b/airflow-core/src/airflow/models/trigger.py @@ -676,22 +676,14 @@ def _submit_callback_if_necessary() -> None: raise RuntimeError("relative_fileloc should not be None for a finished task") from airflow.models.dag_version import _resolve_version_data - # Derive bundle identity from the TI's dag_version (falling back to dag_run/dag_model - # for legacy/unpinned runs), mirroring the other callback sites so bundle_version and - # version_data always describe the same version. + # The DagRun's bundle version is the immutable source identifier used by task execution. bundle_name = ( task_instance.dag_version.bundle_name if task_instance.dag_version else task_instance.dag_model.bundle_name ) - bundle_version = ( - task_instance.dag_version.bundle_version - if task_instance.dag_version and task_instance.dag_run.bundle_version is not None - else task_instance.dag_run.bundle_version - ) - version_data = _resolve_version_data( - task_instance.dag_version, task_instance.dag_run.bundle_version - ) + bundle_version = task_instance.dag_run.bundle_version + version_data = _resolve_version_data(task_instance.dag_version, bundle_version) request = TaskCallbackRequest( filepath=task_instance.dag_model.relative_fileloc, ti=task_instance, diff --git a/airflow-core/tests/unit/executors/test_workloads.py b/airflow-core/tests/unit/executors/test_workloads.py index 37fbcd96ce950..f221b75b89685 100644 --- a/airflow-core/tests/unit/executors/test_workloads.py +++ b/airflow-core/tests/unit/executors/test_workloads.py @@ -228,6 +228,7 @@ def _make_mock_ti( ti.dag_version.version_data = ti_dag_version_data if has_created_dag_version: + ti.dag_run.created_dag_version.bundle_version = bundle_version ti.dag_run.created_dag_version.version_data = version_data else: ti.dag_run.created_dag_version = None @@ -303,6 +304,7 @@ def _make_mocks(bundle_version, version_data, *, has_created_dag_version=True): dag_run.dag_model.bundle_name = "test-bundle" dag_run.dag_model.relative_fileloc = "dags/test_dag.py" if has_created_dag_version: + dag_run.created_dag_version.bundle_version = bundle_version dag_run.created_dag_version.version_data = version_data else: dag_run.created_dag_version = None diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index a7b332e763f1c..41414d33cf50a 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -64,7 +64,11 @@ from airflow.executors.executor_utils import ExecutorName from airflow.executors.local_executor import LocalExecutor from airflow.jobs.job import Job, run_job -from airflow.jobs.scheduler_job_runner import SCHEDULER_DAG_CACHE_SIZE, SchedulerJobRunner +from airflow.jobs.scheduler_job_runner import ( + SCHEDULER_DAG_CACHE_SIZE, + SchedulerJobRunner, + _resolve_ti_callback_bundle_info, +) from airflow.models.asset import ( AssetActive, AssetAliasModel, @@ -13097,11 +13101,7 @@ def _extract_bundle_name(ti): def _extract_bundle_version(ti): """Mirror the inline fallback logic from scheduler_job_runner.py.""" - return ( - ti.dag_version.bundle_version - if ti.dag_version and ti.dag_run.bundle_version is not None - else ti.dag_run.bundle_version - ) + return ti.dag_run.bundle_version class TestSchedulerCallbackBundleInfoDagVersionNullable: @@ -13109,7 +13109,7 @@ class TestSchedulerCallbackBundleInfoDagVersionNullable: Verify the bundle_name / bundle_version extraction logic used at all five TaskCallbackRequest / EmailRequest creation sites in scheduler_job_runner.py. - When dag_version is present -> use dag_version.bundle_name / bundle_version. + When dag_version is present -> use its bundle_name and the DagRun's stable bundle_version. When dag_version is None -> fall back to dag_model.bundle_name / dag_run.bundle_version. """ @@ -13129,7 +13129,7 @@ def test_bundle_info_from_dag_version_when_present(self, dv_bundle_name, dv_bund ti = _make_ti_with_dag_version(dag_version=dv, dag_model_bundle_name="SHOULD-NOT-USE") assert _extract_bundle_name(ti) == dv_bundle_name - assert _extract_bundle_version(ti) == dv_bundle_version + assert _extract_bundle_version(ti) == "v1.0-fallback" # ── With dag_version None (legacy Airflow 2 task) ───────────────────── @@ -13174,10 +13174,9 @@ def test_no_attribute_error_regardless_of_dag_version(self, dag_version_present) assert isinstance(name, str) assert version is None or isinstance(version, str) - # ── Precedence: dag_version wins over fallback ───────────────────────── + # ── Precedence: DagVersion owns the name; DagRun owns the version ────── - def test_dag_version_takes_precedence_over_fallback_values(self): - """When dag_version is set, dag_model/dag_run fallbacks must NOT be used.""" + def test_dag_version_name_and_dag_run_version_take_precedence(self): dv = _make_dag_version(bundle_name="preferred-bundle", bundle_version="preferred-v1") ti = _make_ti_with_dag_version( dag_version=dv, @@ -13186,7 +13185,7 @@ def test_dag_version_takes_precedence_over_fallback_values(self): ) assert _extract_bundle_name(ti) == "preferred-bundle" - assert _extract_bundle_version(ti) == "preferred-v1" + assert _extract_bundle_version(ti) == "fallback-v1" def test_fallback_values_used_only_when_dag_version_is_none(self): """When dag_version is None, fallback values must be used.""" @@ -13214,6 +13213,20 @@ def test_unpinned_dag_run_overrides_dag_version_bundle_version(self): # but bundle_version follows the dag_run's unpinned state assert _extract_bundle_version(ti) is None + def test_callback_uses_run_version_when_dag_version_advances(self): + current_dag_version = _make_dag_version(bundle_name="my-bundle", bundle_version="new-version") + current_dag_version.version_data = {"manifest": "new"} + ti = _make_ti_with_dag_version( + dag_version=current_dag_version, + dag_run_bundle_version="old-version", + ) + + bundle_name, bundle_version, version_data = _resolve_ti_callback_bundle_info(ti) + + assert bundle_name == "my-bundle" + assert bundle_version == "old-version" + assert version_data is None + def _make_scheduler_runner_for_connection_tests( executors: list[BaseExecutor], diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py index 670a82b7a5207..e902fe2844cdd 100644 --- a/airflow-core/tests/unit/models/test_dag_version.py +++ b/airflow-core/tests/unit/models/test_dag_version.py @@ -230,13 +230,19 @@ class TestResolveVersionData: ("dag_version", "bundle_version", "expected"), [ pytest.param( - mock.Mock(version_data={"schema_version": 1}), + mock.Mock(bundle_version="abc123", version_data={"schema_version": 1}), "abc123", {"schema_version": 1}, id="pinned-with-data", ), pytest.param( - mock.Mock(version_data={"schema_version": 1}), + mock.Mock(bundle_version="abc123", version_data={"schema_version": 1}), + "different-version", + None, + id="mismatched-version-suppresses-data", + ), + pytest.param( + mock.Mock(bundle_version="abc123", version_data={"schema_version": 1}), None, None, id="unpinned-suppresses-present-data", diff --git a/airflow-core/tests/unit/models/test_trigger.py b/airflow-core/tests/unit/models/test_trigger.py index 0ab5a89e7ca03..b64032c567c79 100644 --- a/airflow-core/tests/unit/models/test_trigger.py +++ b/airflow-core/tests/unit/models/test_trigger.py @@ -434,6 +434,29 @@ def test_submit_event_task_end_callback_includes_version_data(mock_send, session assert request.version_data == version_data +@patch("airflow.callbacks.database_callback_sink.DatabaseCallbackSink.send") +def test_submit_event_callback_uses_run_version_after_dag_version_advances( + mock_send, session, create_task_instance +): + trigger = Trigger(classpath="does.not.matter", kwargs={}) + session.add(trigger) + task_instance = create_task_instance( + session=session, logical_date=timezone.utcnow(), state=State.DEFERRED + ) + task_instance.trigger_id = trigger.id + task_instance.dag_run.bundle_version = "old-version" + task_instance.dag_version.bundle_version = "new-version" + task_instance.dag_version.version_data = {"manifest": "new"} + session.commit() + + Trigger.submit_event(trigger.id, TaskSuccessEvent(), session=session) + session.flush() + + request = mock_send.call_args.kwargs["callback"] + assert request.bundle_version == "old-version" + assert request.version_data is None + + @pytest.mark.parametrize( ("retries", "expected_state", "expected_callback_type", "expect_history_row"), [ From ffaf426e17f2cb5ab3fce8dac80935fa2675f4ff Mon Sep 17 00:00:00 2001 From: khup Date: Fri, 28 Aug 2026 14:21:35 -0400 Subject: [PATCH 4/4] Add atomic versioned deployments to S3 Dag bundles Mutable prefix synchronization cannot reproduce historical Dag runs or make multi-object deployments atomic. A content-addressed release protocol lets publishers expose one complete generation at a time while retaining an independently verifiable last-known-good local copy. --- .../dag-bundles.rst | 8 +- generated/known_airflow_exceptions.txt | 1 - providers/amazon/docs/bundles/index.rst | 109 +++ .../providers/amazon/aws/bundles/s3.py | 537 ++++++++++- .../providers/amazon/version_compat.py | 2 + .../tests/unit/amazon/aws/bundles/test_s3.py | 911 +++++++++++++++++- 6 files changed, 1518 insertions(+), 50 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/dag-bundles.rst b/airflow-core/docs/administration-and-deployment/dag-bundles.rst index ef65b5eaa7135..1d261847a69a0 100644 --- a/airflow-core/docs/administration-and-deployment/dag-bundles.rst +++ b/airflow-core/docs/administration-and-deployment/dag-bundles.rst @@ -51,7 +51,9 @@ Airflow supports multiple types of Dag Bundles, each catering to specific use ca These bundles integrate with Git repositories, allowing Airflow to fetch Dags directly from a repository. The `GitDagBundle` does support versioning. **airflow.providers.amazon.aws.bundles.s3.S3DagBundle** - These bundles reference an S3 bucket containing Dag files. They do not support versioning of the bundle, meaning tasks always run using the latest code. + These bundles reference an S3 bucket containing Dag files. They use the latest code by default. On Airflow 3.4 + and later, an optional publisher-managed, content-addressed manifest protocol supports versioned Dag runs and + atomic deployments. **airflow.providers.google.cloud.bundles.gcs.GCSDagBundle** These bundles reference a GCS bucket containing Dag files. They do not support versioning of the bundle, meaning tasks always run using the latest code. @@ -140,7 +142,9 @@ For an S3 Dag bundle, the required kwarg is ``bucket_name``. You can optionally .. note:: - ``S3DagBundle`` does not support versioning. Tasks always run against the latest code in the bucket. + ``S3DagBundle`` uses the latest code unless ``manifest_key`` is configured. Manifest mode requires Airflow 3.4 + or later and an S3 Versioning-enabled bucket. Each Dag run then records a content-addressed release and can + retrieve the exact object versions used when the run was created. See :doc:`apache-airflow-providers-amazon:bundles/index` for the full list of kwargs and more examples. diff --git a/generated/known_airflow_exceptions.txt b/generated/known_airflow_exceptions.txt index acfd9ae3eb825..5bfa36dac4ed1 100644 --- a/generated/known_airflow_exceptions.txt +++ b/generated/known_airflow_exceptions.txt @@ -39,7 +39,6 @@ providers/alibaba/src/airflow/providers/alibaba/cloud/hooks/oss.py::11 providers/alibaba/src/airflow/providers/alibaba/cloud/operators/analyticdb_spark.py::1 providers/alibaba/src/airflow/providers/alibaba/cloud/sensors/oss_key.py::2 providers/amazon/src/airflow/providers/amazon/aws/auth_manager/avp/facade.py::3 -providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py::5 providers/amazon/src/airflow/providers/amazon/aws/executors/aws_lambda/lambda_executor.py::2 providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py::1 providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py::1 diff --git a/providers/amazon/docs/bundles/index.rst b/providers/amazon/docs/bundles/index.rst index 0c1d3ea63020a..b2ce9cb8f39bd 100644 --- a/providers/amazon/docs/bundles/index.rst +++ b/providers/amazon/docs/bundles/index.rst @@ -45,3 +45,112 @@ Example of using the S3DagBundle: } } ]' + +Without ``manifest_key``, the bundle keeps its original mutable behavior and synchronizes the latest objects under +``prefix``. Airflow does not pin Dag runs in this mode. + +Versioned, atomic deployments +----------------------------- + +Airflow 3.4 and later can pin an S3 Dag bundle to an immutable deployment. Enable this mode by configuring a +publisher-managed current pointer key: + +.. code-block:: json + + { + "name": "my-versioned-s3-dags", + "classpath": "airflow.providers.amazon.aws.bundles.s3.S3DagBundle", + "kwargs": { + "aws_conn_id": "aws_default", + "bucket_name": "my-airflow-bucket", + "prefix": "dags/", + "manifest_key": "airflow-bundles/current.json", + "refresh_interval": 60 + } + } + +The bucket must have S3 Versioning enabled. A release manifest names every bundle object by its exact S3 +``VersionId`` and records its size and SHA-256 digest: + +.. code-block:: json + + { + "schema_version": 1, + "bucket_name": "my-airflow-bucket", + "prefix": "dags", + "objects": [ + { + "key": "dags/example.py", + "version_id": "3Lg...", + "size": 418, + "sha256": "d7a8fbb307d7809469ca9abcb0082e4f8d5651e46d3cdb762d02d0bf37c9e592" + } + ] + } + +Every ``version_id`` must be a nonempty, non-``null`` S3 version. The current pointer and the entire +``.releases/`` metadata namespace are reserved and must not appear in ``objects``. +Schema version 1 is strict: a release has exactly the four fields shown above, each object has exactly its four +shown fields, and a pointer has exactly ``schema_version`` and ``bundle_version``. Additional fields require a new +schema version and are rejected by this reader. + +The bundle version is the lowercase SHA-256 digest of the canonical release manifest. To produce the canonical +bytes, normalize ``prefix`` by removing its optional trailing slash, sort ``objects`` by ``key``, then encode the +whole release object as UTF-8 JSON with keys sorted, no insignificant whitespace, and non-ASCII characters left +unescaped. For example: + +.. code-block:: python + + import hashlib + import json + + release["prefix"] = release["prefix"].rstrip("/") + release["objects"] = sorted(release["objects"], key=lambda item: item["key"]) + canonical = json.dumps( + release, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + bundle_version = hashlib.sha256(canonical).hexdigest() + +Upload deployment artifacts in this order: + +1. Upload every Dag and support file, recording the returned S3 ``VersionId``, byte size, and SHA-256 digest. +2. Upload the release manifest to + ``.releases/.json``. Create this content-addressed key only if absent (for + example with ``If-None-Match: *``), or verify that an existing object has identical canonical content. +3. Last, atomically replace ``manifest_key`` with the current pointer: + + .. code-block:: json + + {"schema_version":1,"bundle_version":"<64-character lowercase SHA-256>"} + +Publishing the pointer last is the deployment transaction boundary. Uploading objects or a release manifest alone +does not make them visible to Airflow. A rollback only requires publishing a pointer to an earlier retained release. +Airflow reads the pointer once per refresh. When the trusted local generation is absent, it validates the +content-addressed release, downloads each exact object version into a staging directory, verifies size and SHA-256, +and publishes the complete local generation with one rename. Cached pinned runs skip both pointer and release reads. +A missing, malformed, or incomplete release leaves the previous generation active. + +Airflow verifies source integrity when it first publishes a generation. Later task startups use the atomic +completion marker in Airflow's private ``dag_bundle_storage_path`` instead of rereading the entire bundle. Protect +that local directory as trusted Airflow state. + +Pinned Dag runs resolve their release directly from the recorded bundle version and do not depend on mutable +``version_data`` in the metadata database. The configured ``bucket_name``, ``prefix``, and ``manifest_key`` must +therefore remain stable for a bundle name until every run, retry, and callback that could need an old release is +past the retention horizon. Relocate only after that horizon, then use a new bundle name. Keeping old and new +bundle configurations active at the same time is only safe when they cannot expose duplicate Dag IDs. + +Retention and permissions +------------------------- + +Retain every release manifest and every referenced noncurrent object version for at least the maximum Dag run, +retry, clearing, and backfill horizon. An S3 lifecycle policy that deletes either artifact prevents recovering the +corresponding historical generation on a cache miss or a new worker. Protect the ``.releases/`` namespace from +overwrite where possible; Airflow still recomputes its content hash and rejects mismatches. + +The AWS connection needs permission to check the bucket and read the current pointer, release manifests, and exact +object versions. This normally includes ``s3:ListBucket``, ``s3:GetObject``, and ``s3:GetObjectVersion`` on the +configured bucket and keys. Set ``requester_pays`` to ``true`` when the bucket uses Requester Pays. diff --git a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py index 65bacb4b38811..511d8e800fb38 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py @@ -16,27 +16,114 @@ # under the License. from __future__ import annotations +import hashlib +import json import os -from pathlib import Path +import re +import shutil +import tempfile +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any import structlog +from boto3.s3.transfer import S3Transfer from airflow.dag_processing.bundles.base import BaseDagBundle from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook from airflow.providers.amazon.aws.hooks.s3 import S3Hook +from airflow.providers.amazon.version_compat import AIRFLOW_V_3_3_PLUS, AIRFLOW_V_3_4_PLUS from airflow.providers.common.compat.sdk import AirflowException +if AIRFLOW_V_3_3_PLUS: + from airflow.dag_processing.bundles.base import BundleVersion + + +_MANIFEST_SCHEMA_VERSION = 1 +_POINTER_SCHEMA_VERSION = 1 +_STAGING_DIR_PREFIX = ".s3-staging-" +_COMPLETION_MARKER = ".airflow-s3-generation.json" +_RELEASE_MANIFESTS_SUFFIX = ".releases" + + +class S3DagBundleConfigError(AirflowException): + """Raised when an S3 Dag bundle manifest is configured incorrectly.""" + + +class S3DagBundleManifestError(AirflowException): + """Raised when an S3 Dag bundle manifest cannot be loaded or validated.""" + + +class S3DagBundleIntegrityError(AirflowException): + """Raised when a local S3 Dag bundle generation fails integrity validation.""" + + +@dataclass(frozen=True) +class _ManifestObject: + key: str + relative_path: PurePosixPath + version_id: str + size: int + sha256: str + + def as_dict(self) -> dict[str, str | int]: + return { + "key": self.key, + "sha256": self.sha256, + "size": self.size, + "version_id": self.version_id, + } + + +@dataclass(frozen=True) +class _Manifest: + bucket_name: str + prefix: str + objects: tuple[_ManifestObject, ...] + + @property + def canonical_data(self) -> dict[str, Any]: + return { + "bucket_name": self.bucket_name, + "objects": [obj.as_dict() for obj in self.objects], + "prefix": self.prefix, + "schema_version": _MANIFEST_SCHEMA_VERSION, + } + + @property + def version(self) -> str: + payload = json.dumps( + self.canonical_data, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode() + return hashlib.sha256(payload).hexdigest() + + +@dataclass(frozen=True) +class _PublishedGeneration: + path: Path + bundle_version: BundleVersion + class S3DagBundle(BaseDagBundle): """ S3 Dag bundle - exposes a directory in S3 as a Dag bundle. - This allows Airflow to load Dags directly from an S3 bucket. + This allows Airflow to load Dags directly from an S3 bucket. By default, the bundle synchronizes the + latest objects under ``prefix`` into one local directory. Supplying ``manifest_key`` enables versioning + and atomic, last-known-good publication. In that mode, the publisher uploads versioned objects, an + immutable content-addressed release manifest, and finally a small current-version pointer. - :param aws_conn_id: Airflow connection ID for AWS. Defaults to AwsBaseHook.default_conn_name. + :param aws_conn_id: Airflow connection ID for AWS. Defaults to AwsBaseHook.default_conn_name. :param bucket_name: The name of the S3 bucket containing the Dag files. - :param prefix: Optional subdirectory within the S3 bucket where the Dags are stored. - If None, Dags are assumed to be at the root of the bucket (Optional). + :param prefix: Optional prefix within the S3 bucket where the Dags are stored. + :param manifest_key: Optional S3 key for a publisher-managed deployment manifest. This requires Airflow + 3.4 or later and S3 bucket versioning. The configured object is the current-version pointer; + immutable release manifests live below ``.releases/`` and contain the exact S3 + VersionId, size, and SHA-256 digest for every object in the bundle. + :param requester_pays: Whether requests to the S3 bucket should include ``RequestPayer="requester"``. """ supports_versioning = False @@ -47,14 +134,59 @@ def __init__( aws_conn_id: str = AwsBaseHook.default_conn_name, bucket_name: str, prefix: str = "", + manifest_key: str | None = None, + requester_pays: bool = False, **kwargs, ) -> None: super().__init__(**kwargs) + if manifest_key is not None and not AIRFLOW_V_3_4_PLUS: + raise S3DagBundleConfigError("S3 Dag bundle manifests require Airflow 3.4 or later") + if manifest_key is not None and ( + not manifest_key + or manifest_key.startswith("/") + or manifest_key.lower().startswith("s3://") + or manifest_key.endswith("/") + or "\\" in manifest_key + or "\0" in manifest_key + or any(part in {"", ".", ".."} for part in manifest_key.split("/")) + ): + raise S3DagBundleConfigError("manifest_key must identify an S3 object") + if manifest_key is not None: + longest_release_key = f"{manifest_key}{_RELEASE_MANIFESTS_SUFFIX}/{'0' * 64}.json" + if len(longest_release_key.encode()) > 1024: + raise S3DagBundleConfigError("manifest_key is too long for derived S3 release keys") + if self.version is not None and ( + not isinstance(self.version, str) or re.fullmatch(r"[0-9a-f]{64}", self.version) is None + ): + raise S3DagBundleConfigError("S3 Dag bundle version must be a lowercase SHA-256 digest") + normalized_prefix = prefix.rstrip("/") + if ( + prefix.startswith("/") + or prefix.endswith("//") + or "\\" in prefix + or "\0" in prefix + or any(part in {"", ".", ".."} for part in normalized_prefix.split("/")) + ) and prefix != "": + raise S3DagBundleConfigError("prefix must be a safe S3 key prefix in manifest mode") + self.aws_conn_id = aws_conn_id self.bucket_name = bucket_name self.prefix = prefix - # Local path where S3 Dags are downloaded - self.s3_dags_dir: Path = self.base_dir + self.manifest_key = manifest_key + self.requester_pays = requester_pays + # This is intentionally an instance attribute. Manifest mode is opt-in and must not change legacy + # S3DagBundle instances in the same process. + self.supports_versioning = manifest_key is not None + self.refreshes_to_versioned_paths = self.supports_versioning + self._s3_hook: S3Hook | None = None + self._published_generation: _PublishedGeneration | None = None + + if self.version and self.supports_versioning: + self.s3_dags_dir = self.versions_dir / self.version + else: + # Keep mutable legacy synchronization outside ``versions`` so file discovery and stale deletion + # can never traverse immutable generations. + self.s3_dags_dir = self.base_dir / "tracking" log = structlog.get_logger(__name__) self._log = log.bind( @@ -62,28 +194,30 @@ def __init__( version=self.version, bucket_name=self.bucket_name, prefix=self.prefix, + manifest_key=self.manifest_key, aws_conn_id=self.aws_conn_id, + requester_pays=self.requester_pays, ) - self._s3_hook: S3Hook | None = None - def _initialize(self): + def _initialize(self) -> None: with self.lock(): - if not self.s3_dags_dir.exists(): - self._log.info("Creating local Dags directory: %s", self.s3_dags_dir) - os.makedirs(self.s3_dags_dir) - - if not self.s3_dags_dir.is_dir(): - raise AirflowException(f"Local Dags path: {self.s3_dags_dir} is not a directory.") + if self.supports_versioning: + self.versions_dir.mkdir(parents=True, exist_ok=True) + if self.version and self._publish_existing_generation(self.version): + return + else: + self.s3_dags_dir.mkdir(parents=True, exist_ok=True) + if not self.s3_dags_dir.is_dir(): + raise S3DagBundleConfigError(f"Local Dags path: {self.s3_dags_dir} is not a directory.") if not self.s3_hook.check_for_bucket(bucket_name=self.bucket_name): - raise AirflowException(f"S3 bucket '{self.bucket_name}' does not exist.") + raise S3DagBundleConfigError(f"S3 bucket '{self.bucket_name}' does not exist.") - if self.prefix: - # don't check when prefix is "" - if not self.s3_hook.check_for_prefix( + if not self.supports_versioning: + if self.prefix and not self.s3_hook.check_for_prefix( bucket_name=self.bucket_name, prefix=self.prefix, delimiter="/" ): - raise AirflowException( + raise S3DagBundleConfigError( f"S3 prefix 's3://{self.bucket_name}/{self.prefix}' does not exist." ) self.refresh() @@ -93,41 +227,358 @@ def initialize(self) -> None: super().initialize() @property - def s3_hook(self): + def s3_hook(self) -> S3Hook: if self._s3_hook is None: - try: - self._s3_hook: S3Hook = S3Hook(aws_conn_id=self.aws_conn_id) # Initialize S3 hook. - except AirflowException as e: - self._log.warning("Could not create S3Hook for connection %s: %s", self.aws_conn_id, e) + self._s3_hook = S3Hook( + aws_conn_id=self.aws_conn_id, + requester_pays=self.requester_pays, + ) return self._s3_hook - def __repr__(self): + def __repr__(self) -> str: return ( f"" ) - def get_current_version(self) -> str | None: - """Return the current version of the Dag bundle. Currently not supported.""" - return None + def _requester_pays_args(self) -> dict[str, str]: + if self.requester_pays: + return {"RequestPayer": "requester"} + return {} + + def _download_extra_args(self) -> dict[str, Any]: + return { + name: value + for name, value in self.s3_hook.extra_args.items() + if name in S3Transfer.ALLOWED_DOWNLOAD_ARGS and name not in {"RequestPayer", "VersionId"} + } + + def _read_json_object(self, key: str) -> Any: + if self.manifest_key is None: + raise S3DagBundleConfigError("S3 Dag bundle manifest mode is not enabled") + + request: dict[str, Any] = { + "Bucket": self.bucket_name, + "Key": key, + **self._download_extra_args(), + **self._requester_pays_args(), + } + + try: + response = self.s3_hook.get_conn().get_object(**request) + body = response["Body"] + try: + payload = body.read() + finally: + body.close() + manifest_data = json.loads(payload) + except Exception as e: + raise S3DagBundleManifestError(f"Could not read S3 Dag bundle metadata {key!r}") from e + + return manifest_data + + def _release_manifest_key(self, version: str) -> str: + if self.manifest_key is None: + raise S3DagBundleConfigError("S3 Dag bundle manifest mode is not enabled") + return f"{self.manifest_key}{_RELEASE_MANIFESTS_SUFFIX}/{version}.json" + + def _read_current_pointer(self) -> str: + if self.manifest_key is None: + raise S3DagBundleConfigError("S3 Dag bundle manifest mode is not enabled") + data = self._read_json_object(self.manifest_key) + if ( + not isinstance(data, dict) + or set(data) != {"schema_version", "bundle_version"} + or type(data.get("schema_version")) is not int + or data.get("schema_version") != _POINTER_SCHEMA_VERSION + ): + raise S3DagBundleManifestError( + f"S3 Dag bundle pointer must use schema_version {_POINTER_SCHEMA_VERSION}" + ) + version = data.get("bundle_version") + if not isinstance(version, str) or re.fullmatch(r"[0-9a-f]{64}", version) is None: + raise S3DagBundleManifestError("S3 Dag bundle pointer has an invalid version") + return version + + def _read_release_manifest(self, version: str) -> _Manifest: + manifest = self._parse_manifest(self._read_json_object(self._release_manifest_key(version))) + if manifest.version != version: + raise S3DagBundleManifestError( + f"S3 Dag bundle release manifest content hash {manifest.version!r} " + f"does not match requested version {version!r}" + ) + return manifest + + def _parse_manifest(self, data: Any) -> _Manifest: + if ( + not isinstance(data, dict) + or set(data) != {"schema_version", "bucket_name", "prefix", "objects"} + or type(data.get("schema_version")) is not int + or data.get("schema_version") != _MANIFEST_SCHEMA_VERSION + ): + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest must use schema_version {_MANIFEST_SCHEMA_VERSION}" + ) + prefix = self.prefix.rstrip("/") + if data.get("bucket_name") != self.bucket_name or data.get("prefix") != prefix: + raise S3DagBundleManifestError( + "S3 Dag bundle release manifest bucket_name and prefix must match bundle configuration" + ) + raw_objects = data.get("objects") + if not isinstance(raw_objects, list): + raise S3DagBundleManifestError("S3 Dag bundle manifest objects must be a list") + + expected_key_prefix = f"{prefix}/" if prefix else "" + parsed_objects: list[_ManifestObject] = [] + relative_paths: set[str] = set() + + for index, raw_object in enumerate(raw_objects): + if not isinstance(raw_object, dict): + raise S3DagBundleManifestError(f"S3 Dag bundle manifest object {index} must be an object") + if set(raw_object) != {"key", "version_id", "size", "sha256"}: + raise S3DagBundleManifestError(f"S3 Dag bundle manifest object {index} has invalid fields") + key = raw_object.get("key") + version_id = raw_object.get("version_id") + size = raw_object.get("size") + sha256 = raw_object.get("sha256") + if not isinstance(key, str) or not key or key.endswith("/"): + raise S3DagBundleManifestError(f"S3 Dag bundle manifest object {index} has an invalid key") + if self.manifest_key is not None and ( + key == self.manifest_key or key.startswith(f"{self.manifest_key}{_RELEASE_MANIFESTS_SUFFIX}/") + ): + raise S3DagBundleManifestError("S3 Dag bundle manifest must not include bundle metadata") + if expected_key_prefix and not key.startswith(expected_key_prefix): + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest key {key!r} is outside configured prefix {self.prefix!r}" + ) + + relative_key = key[len(expected_key_prefix) :] if expected_key_prefix else key + path_parts = relative_key.split("/") + if ( + any(part in {"", ".", ".."} for part in path_parts) + or "\\" in relative_key + or "\0" in relative_key + ): + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest key {key!r} is not a safe relative path" + ) + relative_path = PurePosixPath(*path_parts) + normalized_relative_path = relative_path.as_posix() + if normalized_relative_path == _COMPLETION_MARKER or normalized_relative_path.startswith( + f"{_COMPLETION_MARKER}/" + ): + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest path {_COMPLETION_MARKER!r} and its subtree are reserved" + ) + if normalized_relative_path in relative_paths: + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest contains duplicate path {normalized_relative_path!r}" + ) + if not isinstance(version_id, str) or not version_id or version_id == "null": + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest object {key!r} has an invalid version_id" + ) + if not isinstance(size, int) or isinstance(size, bool) or size < 0: + raise S3DagBundleManifestError(f"S3 Dag bundle manifest object {key!r} has an invalid size") + if not isinstance(sha256, str) or re.fullmatch(r"[0-9a-f]{64}", sha256) is None: + raise S3DagBundleManifestError(f"S3 Dag bundle manifest object {key!r} has an invalid sha256") + + relative_paths.add(normalized_relative_path) + parsed_objects.append( + _ManifestObject( + key=key, + relative_path=relative_path, + version_id=version_id, + size=size, + sha256=sha256, + ) + ) + + for relative_path in relative_paths: + parts = relative_path.split("/") + for index in range(1, len(parts)): + parent = "/".join(parts[:index]) + if parent in relative_paths: + raise S3DagBundleManifestError( + f"S3 Dag bundle manifest path {relative_path!r} conflicts with file {parent!r}" + ) + + return _Manifest( + bucket_name=self.bucket_name, + prefix=prefix, + objects=tuple(sorted(parsed_objects, key=lambda obj: obj.key)), + ) + + @staticmethod + def _bundle_version(manifest: _Manifest) -> BundleVersion: + return BundleVersion(version=manifest.version) + + def get_current_version(self) -> str | BundleVersion | None: + """Return the locally published manifest version, or ``None`` for legacy mode.""" + if not self.supports_versioning: + return None + if self._published_generation is not None: + return self._published_generation.bundle_version + + version = self.version or self._read_current_pointer() + return BundleVersion(version=version) @property def path(self) -> Path: """Return the local path to the Dag files.""" - return self.s3_dags_dir # Path where Dags are downloaded. + if self._published_generation is not None: + return self._published_generation.path + return self.s3_dags_dir - def refresh(self) -> None: - """Refresh the Dag bundle by re-downloading the Dags from S3.""" - if self.version: - raise AirflowException("Refreshing a specific version is not supported") + def _remove_orphaned_staging_dirs(self) -> None: + for stage_path in self.versions_dir.glob(f"{_STAGING_DIR_PREFIX}*"): + if stage_path.is_dir() and not stage_path.is_symlink(): + self._log.warning("Removing incomplete S3 Dag bundle staging directory", path=stage_path) + shutil.rmtree(stage_path) + @staticmethod + def _generation_is_published(path: Path, version: str) -> bool: + # Reuse is intentionally O(1): first publication verifies every object before an atomic rename. + # Airflow treats its private bundle storage as trusted, as Git bundle worktrees do; synchronous + # rehashing here would read the whole bundle during every task and callback startup. + if path.is_symlink(): + return False + marker_path = path / _COMPLETION_MARKER + if marker_path.is_symlink(): + return False + try: + marker = json.loads(marker_path.read_text()) + except (OSError, ValueError, TypeError): + return False + return ( + isinstance(marker, dict) + and set(marker) == {"schema_version", "bundle_version"} + and type(marker.get("schema_version")) is int + and marker["schema_version"] == 1 + and marker["bundle_version"] == version + ) + + def _get_published_generation_path(self, version: str) -> Path | None: + path = self.versions_dir / version + if not path.exists() and not path.is_symlink(): + return None + if not path.is_dir() or not self._generation_is_published(path, version): + raise S3DagBundleIntegrityError( + f"Existing S3 Dag bundle generation {path} is incomplete or corrupted" + ) + return path + + def _publish_existing_generation(self, version: str) -> bool: + if not (published_path := self._get_published_generation_path(version)): + return False + self._published_generation = _PublishedGeneration( + path=published_path, + bundle_version=BundleVersion(version=version), + ) + return True + + @staticmethod + def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as file: + for chunk in iter(lambda: file.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def _download_generation(self, manifest: _Manifest, target: Path) -> None: + client = self.s3_hook.get_conn() + requester_pays_args = self._requester_pays_args() + hook_extra_args = self._download_extra_args() + for obj in manifest.objects: + local_path = target.joinpath(*obj.relative_path.parts) + local_path.parent.mkdir(parents=True, exist_ok=True) + extra_args = {**hook_extra_args, "VersionId": obj.version_id, **requester_pays_args} + try: + client.download_file( + Bucket=self.bucket_name, + Key=obj.key, + Filename=os.fspath(local_path), + ExtraArgs=extra_args, + Config=self.s3_hook.transfer_config, + ) + except Exception as e: + raise S3DagBundleIntegrityError( + f"Could not download S3 object {obj.key!r} at version {obj.version_id!r}" + ) from e + actual_size = local_path.stat().st_size + if actual_size != obj.size: + raise S3DagBundleIntegrityError( + f"Downloaded S3 object {obj.key!r} at version {obj.version_id!r} has size " + f"{actual_size}, expected {obj.size}" + ) + actual_sha256 = self._file_sha256(local_path) + if actual_sha256 != obj.sha256: + raise S3DagBundleIntegrityError( + f"Downloaded S3 object {obj.key!r} at version {obj.version_id!r} has SHA-256 " + f"{actual_sha256}, expected {obj.sha256}" + ) + + def _materialize_generation(self, manifest: _Manifest, version: str) -> Path: + if published_path := self._get_published_generation_path(version): + return published_path + final_path = self.versions_dir / version + + staging_path = Path(tempfile.mkdtemp(prefix=_STAGING_DIR_PREFIX, dir=self.versions_dir)) + try: + self._download_generation(manifest, staging_path) + marker_path = staging_path / _COMPLETION_MARKER + marker_path.write_text(json.dumps({"schema_version": 1, "bundle_version": version})) + for obj in manifest.objects: + staging_path.joinpath(*obj.relative_path.parts).chmod(0o444) + marker_path.chmod(0o444) + try: + staging_path.rename(final_path) + except FileExistsError: + if not final_path.is_dir() or not self._generation_is_published(final_path, version): + raise S3DagBundleIntegrityError( + f"Concurrent S3 Dag bundle generation {final_path} is incomplete or corrupted" + ) from None + return final_path + finally: + if staging_path.exists(): + shutil.rmtree(staging_path) + + def _refresh_versioned(self) -> None: + self._remove_orphaned_staging_dirs() + version = self.version or self._read_current_pointer() + if self._publish_existing_generation(version): + return + manifest = self._read_release_manifest(version) + bundle_version = self._bundle_version(manifest) + + generation_path = self._materialize_generation(manifest, bundle_version.version) + # Publish instance state last. A failed download or validation therefore leaves both the visible path + # and get_current_version() pinned to the previous complete generation. + self._published_generation = _PublishedGeneration( + path=generation_path, + bundle_version=bundle_version, + ) + + def refresh(self) -> None: + """Refresh the Dag bundle from S3.""" with self.lock(): + if self.supports_versioning: + self._refresh_versioned() + return + if self.version: + raise S3DagBundleConfigError("Refreshing a specific version is not supported") + self._log.debug( - "Downloading Dags from s3://%s/%s to %s", self.bucket_name, self.prefix, self.s3_dags_dir + "Downloading Dags from s3://%s/%s to %s", + self.bucket_name, + self.prefix, + self.s3_dags_dir, ) self.s3_hook.sync_to_local_dir( bucket_name=self.bucket_name, @@ -138,23 +589,21 @@ def refresh(self) -> None: def view_url(self, version: str | None = None) -> str | None: """ - Return a URL for viewing the Dags in S3. Currently, versioning is not supported. + Return a URL for viewing the Dags in S3. This method is deprecated and will be removed when the minimum supported Airflow version is 3.1. - Use `view_url_template` instead. + Use ``view_url_template`` instead. """ return self.view_url_template() def view_url_template(self) -> str | None: - """Return a URL for viewing the Dags in S3. Currently, versioning is not supported.""" - if self.version: - raise AirflowException("S3 url with version is not supported") + """Return a URL for viewing the Dags in S3.""" + if self.version and not self.supports_versioning: + raise S3DagBundleConfigError("S3 url with version is not supported") if hasattr(self, "_view_url_template") and self._view_url_template: - # Because we use this method in the view_url method, we need to handle - # backward compatibility for Airflow versions that doesn't have the - # _view_url_template attribute. Should be removed when we drop support for Airflow 3.0 + # Because we use this method in the view_url method, we need to handle backward compatibility for + # Airflow versions that don't have the _view_url_template attribute. Remove with Airflow 3.0 support. return self._view_url_template - # https://.s3..amazonaws.com/ url = f"https://{self.bucket_name}.s3" if self.s3_hook.region_name: url += f".{self.s3_hook.region_name}" diff --git a/providers/amazon/src/airflow/providers/amazon/version_compat.py b/providers/amazon/src/airflow/providers/amazon/version_compat.py index 4cb19eb3642e2..fa6faf2ba419b 100644 --- a/providers/amazon/src/airflow/providers/amazon/version_compat.py +++ b/providers/amazon/src/airflow/providers/amazon/version_compat.py @@ -41,6 +41,7 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]: AIRFLOW_V_3_1_1_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 1) AIRFLOW_V_3_1_8_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 8) AIRFLOW_V_3_3_PLUS: bool = get_base_airflow_version_tuple() >= (3, 3, 0) +AIRFLOW_V_3_4_PLUS: bool = get_base_airflow_version_tuple() >= (3, 4, 0) try: from airflow.sdk.definitions._internal.types import NOTSET, ArgNotSet @@ -60,6 +61,7 @@ def is_arg_set(value): # type: ignore[misc,no-redef] "AIRFLOW_V_3_1_1_PLUS", "AIRFLOW_V_3_1_8_PLUS", "AIRFLOW_V_3_3_PLUS", + "AIRFLOW_V_3_4_PLUS", "NOTSET", "ArgNotSet", "is_arg_set", diff --git a/providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py b/providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py index f886ebe436718..5997f3eed6883 100644 --- a/providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py +++ b/providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py @@ -16,8 +16,11 @@ # under the License. from __future__ import annotations +import hashlib +import json import os -from unittest.mock import MagicMock, call +from concurrent.futures import ThreadPoolExecutor +from unittest.mock import MagicMock, call, patch import boto3 import pytest @@ -29,15 +32,26 @@ from airflow.providers.common.compat.sdk import AirflowException from tests_common.test_utils.config import conf_vars +from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS, AIRFLOW_V_3_4_PLUS AWS_CONN_ID_WITH_REGION = "s3_dags_connection" AWS_CONN_ID_REGION = "eu-central-1" AWS_CONN_ID_DEFAULT = "aws_default" S3_BUCKET_NAME = "my-airflow-dags-bucket" S3_BUCKET_PREFIX = "project1/dags" +S3_MANIFEST_KEY = "deployments/current.json" +TEST_SHA256 = "0" * 64 if airflow.version.version.strip().startswith("3"): - from airflow.providers.amazon.aws.bundles.s3 import S3DagBundle + from airflow.providers.amazon.aws.bundles.s3 import ( + S3DagBundle, + S3DagBundleConfigError, + S3DagBundleIntegrityError, + S3DagBundleManifestError, + ) + +if AIRFLOW_V_3_3_PLUS: + from airflow.dag_processing.bundles.base import BundleVersion @pytest.fixture @@ -68,6 +82,72 @@ def s3_bucket(mocked_s3_resource, s3_client): return bucket +@pytest.fixture +def versioned_s3_bucket(s3_client): + s3_client.create_bucket(Bucket=S3_BUCKET_NAME) + s3_client.put_bucket_versioning( + Bucket=S3_BUCKET_NAME, + VersioningConfiguration={"Status": "Enabled"}, + ) + return S3_BUCKET_NAME + + +def manifest_version(manifest: dict) -> str: + canonical_manifest = { + "bucket_name": manifest["bucket_name"], + "objects": sorted(manifest["objects"], key=lambda obj: obj["key"]), + "prefix": manifest["prefix"], + "schema_version": 1, + } + return hashlib.sha256( + json.dumps( + canonical_manifest, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode() + ).hexdigest() + + +def publish_release(s3_client, manifest: dict, *, update_pointer: bool = True) -> str: + version = manifest_version(manifest) + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=f"{S3_MANIFEST_KEY}.releases/{version}.json", + Body=json.dumps(manifest).encode(), + ) + if update_pointer: + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=S3_MANIFEST_KEY, + Body=json.dumps({"schema_version": 1, "bundle_version": version}).encode(), + ) + return version + + +def publish_manifest(s3_client, files: dict[str, bytes]) -> tuple[dict, str]: + """Upload objects, their immutable release manifest, and the current pointer in that order.""" + objects = [] + for key, body in files.items(): + response = s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=key, Body=body) + objects.append( + { + "key": key, + "version_id": response["VersionId"], + "size": len(body), + "sha256": hashlib.sha256(body).hexdigest(), + } + ) + manifest = { + "schema_version": 1, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "objects": objects, + } + version = publish_release(s3_client, manifest) + return manifest, version + + @pytest.fixture(autouse=True) def bundle_temp_dir(tmp_path): with conf_vars({("dag_processor", "dag_bundle_storage_path"): str(tmp_path)}): @@ -131,7 +211,7 @@ def test_correct_bundle_path_used(self): bundle = S3DagBundle( name="test", aws_conn_id=AWS_CONN_ID_DEFAULT, prefix="project1_dags", bucket_name="airflow_dags" ) - assert str(bundle.base_dir) == str(bundle.s3_dags_dir) + assert bundle.s3_dags_dir == bundle.base_dir / "tracking" def test_s3_bucket_and_prefix_validated(self, s3_bucket): hook = S3Hook(aws_conn_id=AWS_CONN_ID_DEFAULT) @@ -221,3 +301,828 @@ def test_refresh_without_prefix(self, s3_bucket, s3_client): bundle.refresh() assert bundle._log.debug.call_count == 2 assert bundle._log.debug.call_args_list == [download_log_call, download_log_call] + + +@pytest.mark.skipif(not AIRFLOW_V_3_4_PLUS, reason="S3 manifest versioning requires Airflow >=3.4") +class TestS3DagBundleManifest: + @pytest.fixture(autouse=True) + def setup_connections(self, create_connection_without_db): + create_connection_without_db( + Connection( + conn_id=AWS_CONN_ID_DEFAULT, + conn_type="aws", + extra={"config_kwargs": {"s3": {"bucket_name": S3_BUCKET_NAME}}}, + ) + ) + + @staticmethod + def _bundle(name="manifest-test", **kwargs): + return S3DagBundle( + name=name, + aws_conn_id=AWS_CONN_ID_DEFAULT, + bucket_name=S3_BUCKET_NAME, + prefix=S3_BUCKET_PREFIX, + manifest_key=S3_MANIFEST_KEY, + **kwargs, + ) + + def test_manifest_mode_is_opt_in_per_instance(self): + manifest_bundle = self._bundle() + legacy_bundle = S3DagBundle( + name="legacy", + aws_conn_id=AWS_CONN_ID_DEFAULT, + bucket_name=S3_BUCKET_NAME, + prefix=S3_BUCKET_PREFIX, + ) + + assert manifest_bundle.supports_versioning is True + assert legacy_bundle.supports_versioning is False + assert S3DagBundle.supports_versioning is False + + def test_manifest_mode_requires_airflow_3_4(self): + with ( + patch("airflow.providers.amazon.aws.bundles.s3.AIRFLOW_V_3_4_PLUS", False), + pytest.raises(S3DagBundleConfigError, match="Airflow 3.4 or later"), + ): + self._bundle() + + @pytest.mark.parametrize( + "kwargs", + [ + {"manifest_key": ""}, + {"manifest_key": "../current.json"}, + {"manifest_key": "nested//current.json"}, + {"manifest_key": "x" * 950}, + {"prefix": "dags//"}, + {"prefix": "../dags"}, + ], + ) + def test_manifest_configuration_rejects_ambiguous_keys(self, kwargs): + defaults = { + "name": "invalid-config", + "aws_conn_id": AWS_CONN_ID_DEFAULT, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "manifest_key": S3_MANIFEST_KEY, + } + + with pytest.raises(S3DagBundleConfigError): + S3DagBundle(**{**defaults, **kwargs}) + + def test_manifest_publishes_complete_generation(self, versioned_s3_bucket): + publish_manifest( + boto3.client("s3"), + { + f"{S3_BUCKET_PREFIX}/dag.py": b"from helpers import VALUE\n", + f"{S3_BUCKET_PREFIX}/helpers.py": b"VALUE = 'v1'\n", + }, + ) + bundle = self._bundle() + + bundle.initialize() + current = bundle.get_current_version() + + assert isinstance(current, BundleVersion) + assert bundle.path == bundle.versions_dir / current.version + assert (bundle.path / "dag.py").read_bytes() == b"from helpers import VALUE\n" + assert (bundle.path / "helpers.py").read_bytes() == b"VALUE = 'v1'\n" + assert current.data is None + + def test_objects_are_not_published_until_manifest_changes(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + version_one = bundle.get_current_version() + path_one = bundle.path + + response = s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=f"{S3_BUCKET_PREFIX}/dag.py", + Body=b"VERSION = 2\n", + ) + manifest = { + "schema_version": 1, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": response["VersionId"], + "size": len(b"VERSION = 2\n"), + "sha256": hashlib.sha256(b"VERSION = 2\n").hexdigest(), + } + ], + } + version_two = publish_release(s3_client, manifest, update_pointer=False) + bundle.refresh() + + assert bundle.path == path_one + assert bundle.get_current_version() == version_one + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=S3_MANIFEST_KEY, + Body=json.dumps({"schema_version": 1, "bundle_version": version_two}).encode(), + ) + bundle.refresh() + + assert bundle.path != path_one + assert bundle.get_current_version() != version_one + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 2\n" + + def test_pinned_bundle_downloads_historical_manifest_and_objects(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + current_bundle = self._bundle(name="current") + current_bundle.initialize() + version_one = current_bundle.get_current_version() + + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 2\n"}) + pinned_bundle = S3DagBundle( + name="pinned", + aws_conn_id=AWS_CONN_ID_DEFAULT, + bucket_name=S3_BUCKET_NAME, + prefix=S3_BUCKET_PREFIX, + manifest_key=S3_MANIFEST_KEY, + version=version_one.version, + version_data=None, + ) + pinned_bundle.initialize() + + assert pinned_bundle.manifest_key == S3_MANIFEST_KEY + assert pinned_bundle.get_current_version() == version_one + assert (pinned_bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + def test_get_current_version_describes_published_path(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + published_version = bundle.get_current_version() + + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 2\n"}) + + assert bundle.get_current_version() == published_version + assert bundle.path == bundle.versions_dir / published_version.version + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + def test_partial_download_keeps_last_good_generation(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest( + s3_client, + { + f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n", + f"{S3_BUCKET_PREFIX}/helper.py": b"VALUE = 1\n", + }, + ) + bundle = self._bundle() + bundle.initialize() + previous_path = bundle.path + previous_version = bundle.get_current_version() + publish_manifest( + s3_client, + { + f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 2\n", + f"{S3_BUCKET_PREFIX}/helper.py": b"VALUE = 2\n", + }, + ) + failed_generation_path = bundle.versions_dir / bundle._read_current_pointer() + + client = bundle.s3_hook.get_conn() + original_download = client.download_file + download_count = 0 + + def fail_second_download(**kwargs): + nonlocal download_count + download_count += 1 + if download_count == 2: + raise OSError("injected download failure") + return original_download(**kwargs) + + client.download_file = fail_second_download + with pytest.raises(S3DagBundleIntegrityError, match="Could not download S3 object"): + bundle.refresh() + + assert bundle.path == previous_path + assert bundle.get_current_version() == previous_version + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + assert (bundle.path / "helper.py").read_bytes() == b"VALUE = 1\n" + assert not list(bundle.versions_dir.glob(".s3-staging-*")) + assert not failed_generation_path.exists() + + def test_missing_object_version_keeps_last_good_generation(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + previous_path = bundle.path + previous_version = bundle.get_current_version() + + manifest, _ = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 2\n"}) + obj = manifest["objects"][0] + s3_client.delete_object( + Bucket=S3_BUCKET_NAME, + Key=obj["key"], + VersionId=obj["version_id"], + ) + + with pytest.raises(S3DagBundleIntegrityError, match="Could not download S3 object"): + bundle.refresh() + + assert bundle.path == previous_path + assert bundle.get_current_version() == previous_version + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + def test_pointer_to_missing_release_keeps_last_good_generation(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + previous_path = bundle.path + previous_version = bundle.get_current_version() + missing_version = "f" * 64 + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=S3_MANIFEST_KEY, + Body=json.dumps({"schema_version": 1, "bundle_version": missing_version}).encode(), + ) + + with pytest.raises(S3DagBundleManifestError, match="Could not read.*release"): + bundle.refresh() + + assert bundle.path == previous_path + assert bundle.get_current_version() == previous_version + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + assert not (bundle.versions_dir / missing_version).exists() + assert not list(bundle.versions_dir.glob(".s3-staging-*")) + + def test_pinned_manifest_hash_mismatch_fails_before_download(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + manifest, _ = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + current_bundle = self._bundle(name="current") + current_bundle.initialize() + pinned_bundle = self._bundle( + name="pinned", + version="0" * 64, + version_data=None, + ) + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=f"{S3_MANIFEST_KEY}.releases/{'0' * 64}.json", + Body=json.dumps(manifest).encode(), + ) + client = pinned_bundle.s3_hook.get_conn() + client.download_file = MagicMock() + + with pytest.raises(S3DagBundleManifestError, match="does not match requested version"): + pinned_bundle.initialize() + + client.download_file.assert_not_called() + assert not pinned_bundle.path.exists() + + def test_existing_generation_is_reused_without_download(self, versioned_s3_bucket): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + first_bundle = self._bundle() + first_bundle.initialize() + current = first_bundle.get_current_version() + pinned_bundle = self._bundle(version=current.version, version_data=None) + pinned_bundle.s3_hook.check_for_bucket = MagicMock() + client = pinned_bundle.s3_hook.get_conn() + client.get_object = MagicMock() + client.download_file = MagicMock() + + pinned_bundle.initialize() + + pinned_bundle.s3_hook.check_for_bucket.assert_not_called() + client.get_object.assert_not_called() + client.download_file.assert_not_called() + assert pinned_bundle.path == first_bundle.path + + def test_current_instance_reuses_generation_after_reading_only_pointer(self, versioned_s3_bucket, mocker): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + first_bundle = self._bundle() + first_bundle.initialize() + second_bundle = self._bundle() + client = second_bundle.s3_hook.get_conn() + get_object = mocker.spy(client, "get_object") + download_file = mocker.spy(client, "download_file") + + second_bundle.initialize() + + assert [call.kwargs["Key"] for call in get_object.call_args_list] == [S3_MANIFEST_KEY] + download_file.assert_not_called() + assert second_bundle.path == first_bundle.path + + def test_pinned_get_current_version_does_not_read_release(self): + pinned_bundle = self._bundle(version=TEST_SHA256, version_data=None) + pinned_bundle.s3_hook.get_conn().get_object = MagicMock() + + assert pinned_bundle.get_current_version() == BundleVersion(version=TEST_SHA256) + pinned_bundle.s3_hook.get_conn().get_object.assert_not_called() + + def test_concurrent_initializers_publish_one_complete_generation(self, versioned_s3_bucket): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundles = [self._bundle(), self._bundle()] + + with ThreadPoolExecutor(max_workers=2) as executor: + list(executor.map(lambda bundle: bundle.initialize(), bundles)) + + assert bundles[0].path == bundles[1].path + assert (bundles[0].path / "dag.py").read_bytes() == b"VERSION = 1\n" + assert not list(bundles[0].versions_dir.glob(".s3-staging-*")) + + def test_pinned_bundle_skips_pointer_and_downloads_exact_object_versions( + self, versioned_s3_bucket, mocker + ): + s3_client = boto3.client("s3") + manifest, version = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=f"{S3_BUCKET_PREFIX}/dag.py", + Body=b"VERSION = 2\n", + ) + bundle = self._bundle(name="exact-version", version=version, version_data=None) + client = bundle.s3_hook.get_conn() + get_object = mocker.spy(client, "get_object") + download_file = mocker.spy(client, "download_file") + + bundle.initialize() + + requested_keys = [item.kwargs["Key"] for item in get_object.call_args_list] + assert requested_keys[0] == f"{S3_MANIFEST_KEY}.releases/{version}.json" + assert S3_MANIFEST_KEY not in requested_keys + download_file.assert_called_once() + assert ( + download_file.call_args.kwargs["ExtraArgs"]["VersionId"] == manifest["objects"][0]["version_id"] + ) + assert (bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + def test_manifest_mode_enforces_requester_pays_for_metadata_and_object_requests( + self, versioned_s3_bucket, mocker + ): + s3_client = boto3.client("s3") + manifest, version = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle(requester_pays=True) + # Connection-level extras must not weaken the manifest's exact-version or requester-pays contract. + bundle.s3_hook._extra_args = {"RequestPayer": "bucket-owner", "VersionId": "wrong-version"} + client = bundle.s3_hook.get_conn() + get_object = mocker.spy(client, "get_object") + download_file = mocker.spy(client, "download_file") + + bundle.initialize() + + metadata_keys = {S3_MANIFEST_KEY, f"{S3_MANIFEST_KEY}.releases/{version}.json"} + metadata_calls = [call for call in get_object.call_args_list if call.kwargs["Key"] in metadata_keys] + assert {call.kwargs["Key"] for call in metadata_calls} == metadata_keys + assert all(call.kwargs["RequestPayer"] == "requester" for call in metadata_calls) + assert all("VersionId" not in call.kwargs for call in metadata_calls) + assert download_file.call_args.kwargs["ExtraArgs"] == { + "VersionId": manifest["objects"][0]["version_id"], + "RequestPayer": "requester", + } + + def test_orphaned_stage_is_removed(self, versioned_s3_bucket): + publish_manifest(boto3.client("s3"), {}) + bundle = self._bundle() + orphan = bundle.versions_dir / ".s3-staging-abandoned" + orphan.mkdir(parents=True) + (orphan / "partial.py").write_text("partial") + + bundle.initialize() + + assert not orphan.exists() + assert bundle.path.is_dir() + assert {path.name for path in bundle.path.iterdir()} == {".airflow-s3-generation.json"} + + def test_semantic_manifest_hash_is_order_independent(self): + bundle = self._bundle() + objects = [ + { + "key": f"{S3_BUCKET_PREFIX}/b.py", + "version_id": "b-version", + "size": 2, + "sha256": TEST_SHA256, + }, + { + "key": f"{S3_BUCKET_PREFIX}/a.py", + "version_id": "a-version", + "size": 1, + "sha256": TEST_SHA256, + }, + ] + + first = bundle._parse_manifest( + { + "schema_version": 1, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "objects": objects, + } + ) + second = bundle._parse_manifest( + { + "objects": list(reversed(objects)), + "prefix": S3_BUCKET_PREFIX, + "bucket_name": S3_BUCKET_NAME, + "schema_version": 1, + } + ) + + assert first.version == second.version + + def test_semantic_manifest_hash_matches_canonical_unicode_vector(self): + manifest = { + "schema_version": 1, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/café.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + } + + expected_version = "42a81dc4cd69060b6eebcfb5a71af45c4a6528114b7ddb5c8e8120024bbb66d0" + assert manifest_version(manifest) == expected_version + assert self._bundle()._parse_manifest(manifest).version == expected_version + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("key", f"{S3_BUCKET_PREFIX}/changed.py"), + ("version_id", "changed-version"), + ("size", 2), + ("sha256", "1" * 64), + ], + ) + def test_each_object_field_changes_semantic_version(self, field, value): + bundle = self._bundle() + original = { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + changed = {**original, field: value} + release = { + "schema_version": 1, + "bucket_name": S3_BUCKET_NAME, + "prefix": S3_BUCKET_PREFIX, + "objects": [original], + } + changed_release = {**release, "objects": [changed]} + + assert bundle._parse_manifest(release).version != bundle._parse_manifest(changed_release).version + + @pytest.mark.parametrize( + ("manifest", "error"), + [ + ({"schema_version": 2, "objects": []}, "schema_version"), + ({"schema_version": True, "objects": []}, "schema_version"), + ({"schema_version": 1, "objects": {}}, "objects must be a list"), + ( + {"schema_version": 1, "objects": [], "unexpected": True}, + "schema_version", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + "unexpected": True, + } + ], + }, + "invalid fields", + ), + ({"schema_version": 1, "objects": ["not-an-object"]}, "must be an object"), + ({"schema_version": 1, "bucket_name": "different", "objects": []}, "must match"), + ({"schema_version": 1, "prefix": "different", "objects": []}, "must match"), + ( + { + "schema_version": 1, + "objects": [ + { + "key": S3_MANIFEST_KEY, + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "must not include bundle metadata", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_MANIFEST_KEY}.releases/{TEST_SHA256}.json", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "must not include bundle metadata", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/duplicate.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + }, + { + "key": f"{S3_BUCKET_PREFIX}/duplicate.py", + "version_id": "v2", + "size": 1, + "sha256": TEST_SHA256, + }, + ], + }, + "duplicate path", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": "outside/dag.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "outside configured prefix", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/../escape.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "safe relative path", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/a", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + }, + { + "key": f"{S3_BUCKET_PREFIX}/a/b.py", + "version_id": "v2", + "size": 1, + "sha256": TEST_SHA256, + }, + ], + }, + "conflicts with file", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "null", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "invalid version_id", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": 123, + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "invalid key", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "v1", + "size": True, + "sha256": TEST_SHA256, + } + ], + }, + "invalid size", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "v1", + "size": -1, + "sha256": TEST_SHA256, + } + ], + }, + "invalid size", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/dag.py", + "version_id": "v1", + "size": 1, + "sha256": "not-a-sha256", + } + ], + }, + "invalid sha256", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/.airflow-s3-generation.json", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "reserved", + ), + ( + { + "schema_version": 1, + "objects": [ + { + "key": f"{S3_BUCKET_PREFIX}/.airflow-s3-generation.json/child.py", + "version_id": "v1", + "size": 1, + "sha256": TEST_SHA256, + } + ], + }, + "reserved", + ), + ], + ) + def test_invalid_manifest_is_rejected(self, manifest, error): + manifest.setdefault("bucket_name", S3_BUCKET_NAME) + manifest.setdefault("prefix", S3_BUCKET_PREFIX) + with pytest.raises(S3DagBundleManifestError, match=error): + self._bundle()._parse_manifest(manifest) + + def test_pointer_to_missing_release_is_rejected(self, s3_client): + s3_client.create_bucket(Bucket=S3_BUCKET_NAME) + s3_client.put_object( + Bucket=S3_BUCKET_NAME, + Key=S3_MANIFEST_KEY, + Body=json.dumps({"schema_version": 1, "bundle_version": TEST_SHA256}).encode(), + ) + + with pytest.raises(S3DagBundleManifestError, match="Could not read.*release"): + self._bundle().initialize() + + def test_corrupted_completion_marker_is_not_replaced(self, versioned_s3_bucket): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + current = bundle.get_current_version() + marker = bundle.path / ".airflow-s3-generation.json" + marker.chmod(0o644) + marker.write_text("corrupt") + pinned_bundle = self._bundle(version=current.version, version_data=None) + + with pytest.raises(S3DagBundleIntegrityError, match="incomplete or corrupted"): + pinned_bundle.initialize() + + assert marker.read_text() == "corrupt" + + def test_boolean_completion_marker_schema_is_rejected(self, versioned_s3_bucket): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle() + bundle.initialize() + current = bundle.get_current_version() + marker = bundle.path / ".airflow-s3-generation.json" + marker.chmod(0o644) + marker.write_text(json.dumps({"schema_version": True, "bundle_version": current.version})) + + with pytest.raises(S3DagBundleIntegrityError, match="incomplete or corrupted"): + self._bundle(version=current.version, version_data=None).initialize() + + def test_downloaded_object_checksum_must_match_manifest(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + manifest, _ = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + manifest["objects"][0]["sha256"] = TEST_SHA256 + publish_release(s3_client, manifest) + bundle = self._bundle() + + with pytest.raises(S3DagBundleIntegrityError, match="SHA-256"): + bundle.initialize() + + assert not list(bundle.versions_dir.glob(".s3-staging-*")) + + def test_downloaded_object_size_must_match_manifest(self, versioned_s3_bucket): + s3_client = boto3.client("s3") + manifest, _ = publish_manifest(s3_client, {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + manifest["objects"][0]["size"] += 1 + version = publish_release(s3_client, manifest) + bundle = self._bundle() + + with pytest.raises(S3DagBundleIntegrityError, match="has size"): + bundle.initialize() + + assert not (bundle.versions_dir / version).exists() + assert not list(bundle.versions_dir.glob(".s3-staging-*")) + + @pytest.mark.parametrize("version_data", [None, {"arbitrary": "value"}, {"manifest_version_id": "old"}]) + def test_pinned_run_does_not_depend_on_version_data(self, versioned_s3_bucket, version_data): + publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + current_bundle = self._bundle(name="current") + current_bundle.initialize() + current = current_bundle.get_current_version() + + pinned_bundle = S3DagBundle( + name="pinned", + aws_conn_id=AWS_CONN_ID_DEFAULT, + bucket_name=S3_BUCKET_NAME, + prefix=S3_BUCKET_PREFIX, + manifest_key=S3_MANIFEST_KEY, + version=current.version, + version_data=version_data, + ) + pinned_bundle.initialize() + + assert pinned_bundle.bucket_name == S3_BUCKET_NAME + assert pinned_bundle.prefix == S3_BUCKET_PREFIX + assert pinned_bundle.manifest_key == S3_MANIFEST_KEY + assert (pinned_bundle.path / "dag.py").read_bytes() == b"VERSION = 1\n" + + @pytest.mark.parametrize("version", ["../escape", "abc", "A" * 64, 123]) + def test_pinned_version_must_be_lowercase_sha256(self, version): + with pytest.raises(S3DagBundleConfigError, match="lowercase SHA-256"): + self._bundle(version=version) + + @pytest.mark.parametrize( + "body", + [ + b"not-json", + json.dumps({"schema_version": 2, "bundle_version": TEST_SHA256}).encode(), + json.dumps({"schema_version": True, "bundle_version": TEST_SHA256}).encode(), + json.dumps({"schema_version": 1, "bundle_version": "short"}).encode(), + json.dumps({"schema_version": 1, "bundle_version": TEST_SHA256, "unexpected": True}).encode(), + ], + ) + def test_invalid_current_pointer_is_rejected(self, versioned_s3_bucket, body): + boto3.client("s3").put_object(Bucket=S3_BUCKET_NAME, Key=S3_MANIFEST_KEY, Body=body) + + with pytest.raises(S3DagBundleManifestError): + self._bundle().initialize() + + def test_release_root_symlink_is_rejected(self, versioned_s3_bucket, tmp_path): + _, version = publish_manifest(boto3.client("s3"), {f"{S3_BUCKET_PREFIX}/dag.py": b"VERSION = 1\n"}) + bundle = self._bundle(version=version) + bundle.versions_dir.mkdir(parents=True) + external = tmp_path / "external-generation" + external.mkdir() + (bundle.versions_dir / version).symlink_to(external, target_is_directory=True) + + with pytest.raises(S3DagBundleIntegrityError, match="incomplete or corrupted"): + bundle.initialize()