This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 578c1a61830 Fix macOS `SIGSEGV` in triggerer and DAG processor via
`fork+exec` (#69008)
578c1a61830 is described below
commit 578c1a61830c11f39ad2601023d52dfe01614160
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Jul 1 16:47:18 2026 +0530
Fix macOS `SIGSEGV` in triggerer and DAG processor via `fork+exec` (#69008)
The triggerer and DAG processor fork worker children with a bare os.fork(),
which is not safe on macOS: the child inherits half-initialized Objective-C
runtime state and crashes with SIGSEGV/SIGABRT once it touches a system
framework (DNS resolution, secret backends, HTTP clients). #64874 fixed this
for task execution; both of these components hit the same crash because they
run user trigger and DAG-parsing code that makes network calls.
Generalize the fork+exec path so any importable entry point is rehydrated in
the exec'd child from its module:qualname, and inherit the structured-log
channel on a fixed fd so sync and async children are handled uniformly.
---
.../src/airflow/dag_processing/processor.py | 14 +++-
.../src/airflow/jobs/triggerer_job_runner.py | 13 +++-
.../tests/unit/dag_processing/test_processor.py | 46 +++++++++++-
airflow-core/tests/unit/jobs/test_triggerer_job.py | 28 +++++++
.../src/airflow/sdk/execution_time/supervisor.py | 87 ++++++++++++++--------
.../src/airflow/sdk/execution_time/task_runner.py | 7 --
.../task_sdk/execution_time/test_supervisor.py | 41 ++++++++--
7 files changed, 189 insertions(+), 47 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index 251b7b62974..88a486f99c6 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -40,6 +40,7 @@ from airflow.dag_processing.bundles.base import
BundleVersionLock
from airflow.dag_processing.dagbag import BundleDagBag, DagBag
from airflow.models.dag import DagModel
from airflow.sdk.exceptions import TaskNotFound
+from airflow.sdk.execution_time import supervisor
from airflow.sdk.execution_time.comms import (
ConnectionResult,
DeleteVariable,
@@ -586,13 +587,24 @@ class DagFileProcessorProcess(WatchedSubprocess):
) -> Self:
logger = kwargs["logger"]
- _pre_import_airflow_modules(os.fspath(path), logger)
+ # Parsing DAG files runs user code that can trigger macOS-unsafe ObjC
+ # initialization (secret backends, connection/variable lookups, HTTP
+ # clients). Fork+exec a clean interpreter there. Tests override
`target`
+ # with a stub to exercise the base infrastructure; keep bare fork for
those.
+ use_exec = target is _parse_file_entrypoint and
supervisor._should_use_exec()
+
+ # Pre-importing only helps the bare-fork child (it inherits the
imports via
+ # copy-on-write). An exec'd child re-imports from scratch, so skip it
there
+ # to avoid leaking user modules into the long-lived processor manager.
+ if not use_exec:
+ _pre_import_airflow_modules(os.fspath(path), logger)
proc: Self = super().start(
target=target,
client=client,
bundle_name=bundle_name,
dag_file_rel_path=dag_file_rel_path,
+ use_exec=use_exec,
**kwargs,
)
proc.had_callbacks = bool(callbacks) # Track if this process had
callbacks
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index eccd9d2fc04..355763aed5f 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -59,6 +59,7 @@ from airflow.models.trigger import Trigger
from airflow.observability.metrics import stats_utils
from airflow.sdk.api.datamodels._generated import HITLDetailResponse
from airflow.sdk.definitions.asset import Asset
+from airflow.sdk.execution_time import supervisor
from airflow.sdk.execution_time.comms import (
AssetStateStoreResult,
ClearAssetStateStoreByName,
@@ -539,7 +540,17 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
**kwargs,
):
proc_id = job.id if job is not None else uuid4()
- proc = super().start(id=proc_id, job=job, target=cls.run_in_process,
logger=logger, **kwargs)
+ # Triggers run user code that polls APIs / watches queues / hits HTTP
+ # endpoints -- almost always network calls that trigger macOS-unsafe
ObjC
+ # initialization. Fork+exec a clean interpreter for the runner child.
+ proc = super().start(
+ id=proc_id,
+ job=job,
+ target=cls.run_in_process,
+ logger=logger,
+ use_exec=supervisor._should_use_exec(),
+ **kwargs,
+ )
msg = messages.StartTriggerer()
proc.send_msg(msg, request_id=0)
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index 599c36d47f6..3f219c1a083 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -62,13 +62,14 @@ from airflow.dag_processing.processor import (
_execute_email_callbacks,
_execute_task_callbacks,
_parse_file,
+ _parse_file_entrypoint,
_pre_import_airflow_modules,
)
from airflow.models import DagRun
from airflow.sdk import DAG, BaseOperator
from airflow.sdk.api.client import Client
from airflow.sdk.api.datamodels._generated import ConnectionResponse,
DagRunState, VariableResponse
-from airflow.sdk.execution_time import comms
+from airflow.sdk.execution_time import comms, supervisor
from airflow.sdk.execution_time.comms import (
GetConnection,
GetTaskStates,
@@ -94,6 +95,19 @@ if TYPE_CHECKING:
pytestmark = pytest.mark.db_test
+
[email protected](autouse=True)
+def _force_bare_fork(monkeypatch):
+ """
+ Keep the parsing child on a bare ``os.fork`` on every platform.
+
+ On macOS the DAG processor forks+execs a clean interpreter for fork-safety
+ (see ``DagFileProcessorProcess.start``). Forcing bare fork keeps local
macOS
+ runs aligned with Linux/CI behavior.
+ """
+ monkeypatch.setattr(supervisor, "_should_use_exec", lambda: False)
+
+
DEFAULT_DATE = timezone.datetime(2016, 1, 1)
# Filename to be used for dags that are created in an ad-hoc manner and can be
removed/
@@ -468,6 +482,36 @@ class TestDagFileProcessor:
assert logger.warning.call_count == 1
[email protected](
+ ("platform_uses_exec", "target", "expected_use_exec"),
+ [
+ (True, _parse_file_entrypoint, True),
+ (False, _parse_file_entrypoint, False),
+ (True, lambda: None, False),
+ ],
+)
+def test_start_opts_into_fork_exec(monkeypatch, mocker, platform_uses_exec,
target, expected_use_exec):
+ """start() forks+execs only for the real parse entry point on fork-unsafe
platforms."""
+ monkeypatch.setattr(supervisor, "_should_use_exec", lambda:
platform_uses_exec)
+ base_start = mocker.patch(
+ "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start",
return_value=MagicMock()
+ )
+
mocker.patch("airflow.dag_processing.processor._pre_import_airflow_modules")
+
+ DagFileProcessorProcess.start(
+ path="some_dag.py",
+ bundle_path=pathlib.Path("/tmp/bundle"),
+ bundle_name="testing",
+ dag_file_rel_path="some_dag.py",
+ callbacks=[],
+ client=MagicMock(spec=Client),
+ target=target,
+ logger=MagicMock(),
+ )
+
+ assert base_start.call_args.kwargs["use_exec"] is expected_use_exec
+
+
def write_dag_in_a_fn_to_file(fn: Callable[[], None], folder: pathlib.Path) ->
pathlib.Path:
# Create the dag in a fn, and use inspect.getsource to write it to a file
so that
# a) the test dag is directly viewable here in the tests
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 9eced8f1f80..e8689510cf3 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -79,6 +79,7 @@ from airflow.sdk import DAG, Asset, BaseHook, BaseOperator
from airflow.sdk.api.client import Client
from airflow.sdk.api.datamodels._generated import AssetStateStoreResponse
from airflow.sdk.exceptions import ErrorType
+from airflow.sdk.execution_time import supervisor
from airflow.sdk.execution_time.comms import (
AssetStateStoreResult,
ClearAssetStateStoreByName,
@@ -123,6 +124,19 @@ if TYPE_CHECKING:
pytestmark = pytest.mark.db_test
[email protected](autouse=True)
+def _force_bare_fork(monkeypatch):
+ """
+ Keep the runner child on a bare ``os.fork`` on every platform.
+
+ On macOS the triggerer forks+execs a clean interpreter for fork-safety
+ (see ``TriggerRunnerSupervisor.start``). These tests fork the real runner
+ with trigger classes defined in this module, which a fresh exec'd
interpreter
+ cannot import. Forcing bare fork makes local macOS runs match Linux/CI.
+ """
+ monkeypatch.setattr(supervisor, "_should_use_exec", lambda: False)
+
+
@pytest.fixture(autouse=True)
def clean_database():
"""Fixture that cleans the database before and after every test."""
@@ -232,6 +246,20 @@ def test_triggerer_job_runner_stores_team_name(team_name):
assert runner.team_name == team_name
[email protected]("platform_uses_exec", [True, False])
+def test_start_opts_into_fork_exec(monkeypatch, mocker, platform_uses_exec):
+ """start() forks+execs the runner child on fork-unsafe platforms
(macOS)."""
+ monkeypatch.setattr(supervisor, "_should_use_exec", lambda:
platform_uses_exec)
+ base_start = mocker.patch(
+ "airflow.sdk.execution_time.supervisor.WatchedSubprocess.start",
return_value=MagicMock()
+ )
+
+ TriggerRunnerSupervisor.start(job=Job(id=999), capacity=10)
+
+ assert base_start.call_args.kwargs["use_exec"] is platform_uses_exec
+ assert base_start.call_args.kwargs["target"] ==
TriggerRunnerSupervisor.run_in_process
+
+
@pytest.fixture
def supervisor_builder(mocker, session):
def builder(job=None):
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index 76dfd2009ac..c533a9160f4 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -25,6 +25,7 @@ import functools
import io
import logging
import os
+import pkgutil
import selectors
import signal
import sys
@@ -498,42 +499,63 @@ runtime detects the corrupted state and crashes with
SIGABRT.
Calling ``os.execv`` immediately after ``os.fork`` replaces the child's address
space, giving it clean ObjC state. Before exec, the supervisor ``dup2``s the
-socketpairs onto FDs 0 (requests/stdin), 1 (stdout), 2 (stderr). The
duplicated
-FDs survive the upcoming exec because ``os.dup2(inheritable=True)`` (the
default)
-clears ``FD_CLOEXEC`` on the destination FDs. The log channel is obtained
after
-startup via the existing ``ResendLoggingFD`` mechanism.
+socketpairs onto fixed FDs the exec'd child reconstructs: 0 (requests/stdin),
+1 (stdout), 2 (stderr), 3 (structured logs). ``os.set_inheritable`` clears
+``FD_CLOEXEC`` on those FDs so they survive the upcoming exec.
-Currently only task execution opts in (via ``ActivitySubprocess.start``). DAG
-processor and triggerer can also hit this crash and will need the same
treatment
-as a follow-up (see https://github.com/apache/airflow/issues/65691).
+Task execution (``ActivitySubprocess``), the DAG processor
+(``DagFileProcessorProcess``) and the triggerer (``TriggerRunnerSupervisor``)
+all opt in -- each runs a different child entry point, named by
``module:qualname``
+in the ``_AIRFLOW_CHILD_TARGET`` env var and rehydrated by
:func:`_child_exec_main`.
See: https://github.com/python/cpython/issues/105912
https://github.com/apache/airflow/discussions/24463
+ https://github.com/apache/airflow/issues/65691
"""
+def _should_use_exec() -> bool:
+ """Whether forked children should ``exec`` a fresh interpreter on this
platform."""
+ return sys.platform in _FORK_EXEC_PLATFORMS
+
+
+def _resolve_child_target(dotted: str) -> Callable[[], None]:
+ """
+ Resolve a ``module:qualname`` string to the callable the exec'd child runs.
+
+ Empty input falls back to :func:`_subprocess_main` (task execution). The
+ qualname may name a nested attribute (e.g. ``ClassName.classmethod``) so
+ classmethod entry points like ``TriggerRunnerSupervisor.run_in_process``
+ rehydrate correctly across ``execv``.
+ """
+ if not dotted:
+ return _subprocess_main
+ return pkgutil.resolve_name(dotted)
+
+
def _child_exec_main():
"""
Entry point for the child process when using fork+exec (macOS).
- After exec, FDs 0/1/2 are already the requests/stdout/stderr sockets
- (dup2'd by the parent before exec). The log channel is NOT inherited;
- instead, the task runner requests it from the supervisor via the existing
- ``ResendLoggingFD`` mechanism after startup.
+ After exec, FDs 0/1/2/3 are the requests/stdout/stderr/log sockets the
parent
+ placed there via dup2. The target to run is named in
``_AIRFLOW_CHILD_TARGET``
+ (``module:qualname``); it is rehydrated and handed to :func:`_fork_main`,
which
+ sets up the structured log channel from FD 3 exactly as the bare-fork path
does.
"""
# FDs 0, 1, 2 were dup2'd onto the socketpairs before exec.
child_requests = socket(fileno=0)
child_stdout = socket(fileno=1)
child_stderr = socket(fileno=2)
+ target = _resolve_child_target(os.environ.pop("_AIRFLOW_CHILD_TARGET", ""))
+
# _fork_main always exits via os._exit(), so the socket objects above are
# never GC'd (which would close their underlying FDs). This is safe but
# depends on that invariant -- do not refactor _fork_main to return.
#
- # log_fd=0 tells _fork_main to skip structured log channel setup.
- # Signal to the task runner to request it via ResendLoggingFD after
startup.
- os.environ["_AIRFLOW_FORK_EXEC"] = "1"
- _fork_main(child_requests, child_stdout, child_stderr, 0, _subprocess_main)
+ # FD 3 is the inherited structured-log socket; _fork_main configures
logging
+ # over it just as it would with the FD inherited across a bare fork.
+ _fork_main(child_requests, child_stdout, child_stderr, 3, target)
class NeverRaised(Exception):
@@ -669,15 +691,14 @@ class WatchedSubprocess:
:param use_exec: If True, on platforms that need it (currently macOS),
immediately ``os.execv`` a fresh Python interpreter after
``os.fork``.
This avoids macOS fork-safety issues with Objective-C frameworks.
- Task execution opts in; DAG processor and triggerer do not.
-
- The exec'd child always runs ``_subprocess_main``, so ``use_exec=True``
- is only valid when ``target is _subprocess_main``.
+ ``target`` is rehydrated in the exec'd child from its
``module:qualname``,
+ so any importable entry point (task execution, DAG processor,
triggerer)
+ is supported.
"""
- if use_exec and target is not _subprocess_main:
- raise ValueError(
- f"use_exec=True is only supported with
target=_subprocess_main; got target={target!r}"
- )
+ if use_exec and "<" in getattr(target, "__qualname__", "<"):
+ # Closures/lambdas (``<locals>`` / ``<lambda>`` in the qualname)
and
+ # objects without a qualname can't be named for the exec'd child.
+ raise ValueError(f"use_exec=True requires a top-level importable
target, got {target!r}")
# Create socketpairs/"pipes" to connect to the stdin and out from the
subprocess
child_stdout, read_stdout = socketpair()
child_stderr, read_stderr = socketpair()
@@ -700,14 +721,22 @@ class WatchedSubprocess:
try:
if use_exec:
- # macOS: exec a fresh Python interpreter to replace the
- # inherited ObjC/CoreFoundation state that is not
fork-safe.
- # dup2 copies the socketpairs onto FDs 0/1/2; os.dup2
clears
- # FD_CLOEXEC on the destination FDs, so they survive exec.
- # The log channel is requested later via ResendLoggingFD.
+ # macOS: exec a fresh Python interpreter to drop the
inherited
+ # ObjC/CoreFoundation state that is not fork-safe.
Redirect the
+ # socketpairs onto the fixed FDs the exec'd child
reconstructs:
+ # 0 (requests/stdin), 1 (stdout), 2 (stderr), 3
(structured logs).
+ # The source fds are always >= 3 (0/1/2 stay open in every
launch
+ # path), so no dup2 clobbers a not-yet-placed source.
set_inheritable
+ # guarantees FD_CLOEXEC is clear on all four (dup2 leaves
it set when
+ # a source already equals its target), so they survive
execv. The
+ # entry point is passed to the child by name.
+ os.environ["_AIRFLOW_CHILD_TARGET"] =
f"{target.__module__}:{target.__qualname__}"
os.dup2(child_requests.fileno(), 0)
os.dup2(child_stdout.fileno(), 1)
os.dup2(child_stderr.fileno(), 2)
+ os.dup2(child_logs.fileno(), 3)
+ for fd in (0, 1, 2, 3):
+ os.set_inheritable(fd, True)
os.execv(
sys.executable,
[
@@ -1329,7 +1358,7 @@ class ActivitySubprocess(WatchedSubprocess):
# Opt in to fork+exec on platforms that need it (currently macOS).
# Tests override `target` with a local stub to exercise the base
# infrastructure; keep bare fork for those.
- use_exec = target is _subprocess_main and sys.platform in
_FORK_EXEC_PLATFORMS
+ use_exec = target is _subprocess_main and _should_use_exec()
proc: Self = super().start(
id=what.id, client=client, target=target, logger=logger,
use_exec=use_exec, **kwargs
)
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 ac8889daf02..ab79d4ba98e 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -2304,13 +2304,6 @@ def main():
log.info("::group::Pre Execute")
startup_details = get_startup_details()
- # On macOS fork+exec path, the structured log channel wasn't
- # inherited (exec replaces the address space). Request it from
- # the supervisor using the existing ResendLoggingFD mechanism.
- # Must happen after get_startup_details() so we don't read the
- # startup message as a ResendLoggingFD response.
- if os.environ.pop("_AIRFLOW_FORK_EXEC", None) == "1":
- reinit_supervisor_comms()
span_ctx_mgr = _make_task_span(msg=startup_details)
span = stack.enter_context(span_ctx_mgr)
ti, context, log = startup(msg=startup_details)
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 8dea3d0793f..54a426450b3 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -4057,15 +4057,38 @@ def
test_api_client_clears_dag_bag_override_when_dag_is_none():
in_process_api_server.cache_clear()
+class TestResolveChildTarget:
+ """Test rehydrating the exec'd child's entry point from
_AIRFLOW_CHILD_TARGET."""
+
+ def test_empty_defaults_to_subprocess_main(self):
+ assert supervisor._resolve_child_target("") is
supervisor._subprocess_main
+
+ def test_module_level_function(self):
+ resolved =
supervisor._resolve_child_target("airflow.sdk.execution_time.supervisor:_subprocess_main")
+ assert resolved is supervisor._subprocess_main
+
+ def test_classmethod_entry_point(self):
+ """A ``module:ClassName.method`` target rehydrates the bound
classmethod (e.g. the triggerer's)."""
+ resolved = supervisor._resolve_child_target(
+ "airflow.sdk.execution_time.supervisor:WatchedSubprocess.start"
+ )
+ assert resolved == supervisor.WatchedSubprocess.start
+ assert callable(resolved)
+
+ def test_start_rejects_non_importable_target_under_exec(self):
+ """A closure/lambda target can't be named for the exec'd child, so
start() fails fast."""
+ with pytest.raises(ValueError, match="top-level importable target"):
+ supervisor.WatchedSubprocess.start(target=lambda: None,
use_exec=True)
+
+
@pytest.mark.usefixtures("disable_capturing")
class TestChildExecMain:
"""Test the macOS fork+exec child entry point."""
- def test_uses_fds_012_and_requests_log_channel(self, monkeypatch):
- """_child_exec_main wraps FDs 0/1/2 as sockets, passes log_fd=0, sets
_AIRFLOW_FORK_EXEC."""
+ def test_uses_fds_0123_and_inherits_log_channel(self, monkeypatch):
+ """_child_exec_main wraps FDs 0/1/2 as sockets and passes log_fd=3
(inherited log channel)."""
# _child_exec_main expects FDs 0/1/2 to be sockets (dup2'd by the
- # parent before exec). It passes log_fd=0 to _fork_main (structured
- # logging is requested later via ResendLoggingFD).
+ # parent before exec) and the structured log channel inherited on FD 3.
req_a, req_b = socket.socketpair()
out_a, out_b = socket.socketpair()
err_a, err_b = socket.socketpair()
@@ -4075,6 +4098,9 @@ class TestChildExecMain:
saved_1 = os.dup(1)
saved_2 = os.dup(2)
+ # The parent names the entry point to run via this env var.
+ monkeypatch.setenv("_AIRFLOW_CHILD_TARGET",
"airflow.sdk.execution_time.supervisor:_subprocess_main")
+
try:
os.dup2(req_a.fileno(), 0)
os.dup2(out_a.fileno(), 1)
@@ -4101,11 +4127,10 @@ class TestChildExecMain:
assert captured["requests_fd"] == 0
assert captured["stdout_fd"] == 1
assert captured["stderr_fd"] == 2
- assert captured["log_fd"] == 0
+ assert captured["log_fd"] == 3
assert captured["target"] is supervisor._subprocess_main
- # _child_exec_main sets this so the task runner knows to request
- # the log channel via ResendLoggingFD.
- assert os.environ.pop("_AIRFLOW_FORK_EXEC") == "1"
+ # The env var is consumed (popped) so it does not leak to
grandchildren.
+ assert "_AIRFLOW_CHILD_TARGET" not in os.environ
finally:
# Restore original FDs.
os.dup2(saved_0, 0)
