kaxil commented on code in PR #65993:
URL: https://github.com/apache/airflow/pull/65993#discussion_r3160934712
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -19,6 +19,7 @@
import json
import time
+from concurrent.futures import TimeoutError
Review Comment:
This shadows the builtin `TimeoutError`. In Python 3.11+ they're the same
class, but Airflow still supports 3.10 where `concurrent.futures.TimeoutError`
and the builtin are distinct. Anywhere later in this module that does `except
TimeoutError` will now silently mean `concurrent.futures.TimeoutError`.
Prefer `from concurrent.futures import TimeoutError as FutureTimeoutError`,
or move the import inside `transport()` since that's the only use site.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -358,7 +359,12 @@ async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
self._cm = AsyncExitStack()
- asyncio.run_coroutine_threadsafe(start_lifespan(self._cm, self.app),
middleware.loop)
+ try:
+ asyncio.run_coroutine_threadsafe(start_lifespan(self._cm,
self.app), middleware.loop).result(
+ timeout=30
+ )
+ except TimeoutError as err:
+ raise RuntimeError("Timed out while starting the in-process
execution API lifespan") from err
Review Comment:
This fix improves observability (loud failure instead of silent hang) but
does not actually resolve #65945. The reporter's stack trace shows their custom
`SlurmSharedFileTaskHandler.format` crashing inside `svcs.register_factory`'s
`log.debug()` call -- the lifespan startup raises because of a buggy user log
handler. After this PR merges they'll see a clear `RuntimeError` propagated
from `transport`, but the triggerer will still fail to start; the user-side
handler bug is the actual cause.
Two follow-ups worth considering: (a) call this out explicitly in the PR
description so the reporter understands what behavior to expect post-merge; (b)
think about whether the `log.debug` call inside the svcs/lifespan setup should
be made more defensive (or routed through a logger that bypasses the user task
handler) so a single broken user handler doesn't take down triggerer startup.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -358,7 +359,12 @@ async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
self._cm = AsyncExitStack()
Review Comment:
Pre-existing, but visible here: `self._cm` is set up to register the
lifespan's exit handlers, yet I don't see anywhere in `InProcessExecutionAPI`
that calls `self._cm.aclose()`. That means the lifespan's shutdown code (svcs
registry teardown, anything yielded back from registered factories) never runs.
Since you're already touching this block, worth confirming whether the
missing teardown is intentional (process exits anyway) or a separate latent bug
worth filing.
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -60,6 +66,38 @@ def test_ti_self_routes_have_task_instance_id_param(client):
)
+class TestInProcessExecutionAPI:
+ def test_transport_waits_for_lifespan_startup(self):
+ entered_lifespan = threading.Event()
+
+ @asynccontextmanager
+ async def lifespan(app):
+ await asyncio.sleep(0.05)
+ app.state.lifespan_called = True
+ entered_lifespan.set()
+ yield
+
+ api = InProcessExecutionAPI()
+ api._app = FastAPI(lifespan=lifespan)
+
+ api.transport
+
Review Comment:
Stylistic nit: accessing a property purely for its side effect reads like a
typo to a future reader. Prefer `_ = api.transport` or `assert api.transport is
not None` to make the intent explicit.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -358,7 +359,12 @@ async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
self._cm = AsyncExitStack()
- asyncio.run_coroutine_threadsafe(start_lifespan(self._cm, self.app),
middleware.loop)
+ try:
+ asyncio.run_coroutine_threadsafe(start_lifespan(self._cm,
self.app), middleware.loop).result(
+ timeout=30
Review Comment:
Hardcoded magic number. Lift to a module-level constant (e.g.
`_LIFESPAN_STARTUP_TIMEOUT_SECONDS = 30`) so it's discoverable and tunable, and
include the value in the `RuntimeError` message on line 367 so operators have a
hint when a slow real-world startup gets clipped (`f"Timed out after
{_LIFESPAN_STARTUP_TIMEOUT_SECONDS}s ..."`).
Has 30s been sanity-checked against the slowest realistic deployment (cold
filesystems, large dag_bag warmup, Postgres connection setup on a sleepy
network)? If not, it might be worth bumping it or making it configurable.
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -60,6 +66,38 @@ def test_ti_self_routes_have_task_instance_id_param(client):
)
+class TestInProcessExecutionAPI:
+ def test_transport_waits_for_lifespan_startup(self):
+ entered_lifespan = threading.Event()
+
+ @asynccontextmanager
+ async def lifespan(app):
+ await asyncio.sleep(0.05)
+ app.state.lifespan_called = True
+ entered_lifespan.set()
+ yield
+
+ api = InProcessExecutionAPI()
+ api._app = FastAPI(lifespan=lifespan)
+
+ api.transport
+
+ assert entered_lifespan.is_set()
+ assert api.app.state.lifespan_called
+
+ def test_transport_surfaces_lifespan_startup_errors(self):
+ @asynccontextmanager
+ async def lifespan(app):
+ raise RuntimeError("lifespan failed")
+ yield
+
+ api = InProcessExecutionAPI()
+ api._app = FastAPI(lifespan=lifespan)
+
+ with pytest.raises(RuntimeError, match="lifespan failed"):
+ api.transport
+
Review Comment:
Both tests bypass `api.app` (the real `cached_property` that wires up
`dependency_overrides`, `dag_bag`, and the actual Execution API routes at
`app.py:325-342`) by assigning `api._app = FastAPI(lifespan=lifespan)`
directly. That isolates the timeout/exception logic cleanly, which is fine for
a unit test, but it doesn't exercise the *real* Execution API lifespan -- which
is the thing that actually broke in #65945.
Consider also adding an integration-style test that uses the real `api.app`
and asserts startup completes, plus one that injects a slow lifespan to drive
the 30s timeout path end-to-end.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
