jgoedeke opened a new issue, #74453: URL: https://github.com/apache/airflow/issues/74453
### Under which category would you file this issue? Airflow Core ### Apache Airflow version 3.3.2 ### What happened and how to reproduce it? In `SerializedDAG.partial_subset()` (https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/definitions/dag.py#L356-L385), `filter_task_group()` shallow-copies each task group and then deep-copies the attributes listed in `type(group).__slots__`: ```python copied = copy.copy(group) ... for attr in type(group).__slots__: value = getattr(group, attr) value = copy.deepcopy(value, memo) object.__setattr__(copied, attr, value) ``` On 3.3.2: ```text SerializedTaskGroup.__slots__ = ('_group_id', 'group_display_name', 'prefix_group_id', 'parent_group', 'dag', 'tooltip', 'doc_md', 'default_args', 'ui_color', 'ui_fgcolor', 'children', 'upstream_group_ids', 'downstream_group_ids', 'upstream_task_ids', 'downstream_task_ids') SerializedMappedTaskGroup.__slots__ = ('_expand_input',) ``` For a `SerializedMappedTaskGroup`, `children` is not in `type(group).__slots__`, so the copy keeps the original `children` dict (`copied.children is group.children` is `True` for the mapped group and `False` for the root group and the nested `SerializedTaskGroup`). `filter_task_group()` then assigns the subset's tasks and copied child groups into that dict, so after one call the original DAG's mapped group contains the subset's task objects and the subset's copy of the nested group. Behaviour after one `partial_subset()` call + `gc.collect()` on 3.3.2: | DAG shape | Original DAG modified | Original tasks with dead `task_group` weakref | |---|---|---| | mapped group containing a nested group | yes | tasks of the nested group | | mapped group, no nested group | yes | none | | unmapped group containing a nested group | no | none | Reproduction (no database needed): ```python import gc import tracemalloc from airflow.sdk import DAG, task, task_group from airflow.serialization.serialized_objects import DagSerialization @task def files(): return [1, 2] @task def prepare(x): return x @task def convert(x): return x @task_group def inner(x): a = convert.override(task_id='a')(x) b = convert.override(task_id='b')(a) return b @task def select(x): return x @task_group def outer(x): p = prepare(x) c = convert(p) i = inner(c) return select(i) with DAG('repro') as dag: outer.expand(x=files()) sdag = DagSerialization.from_dict(DagSerialization.to_dict(dag)) tracemalloc.start() gc.collect() base = tracemalloc.get_traced_memory()[0] for i in range(1, 201): sub = sdag.partial_subset( task_ids=['outer.prepare'], include_downstream=True, include_upstream=False ) del sub if i in (1, 2, 5, 50, 200): gc.collect() cur = tracemalloc.get_traced_memory()[0] inner_group = sdag.task_group.children['outer'].children.get('outer.inner') try: tg = sdag.task_dict['outer.inner.a'].task_group.group_id except ReferenceError as e: tg = f'ReferenceError: {e}' print( f'iter={i} mem_delta={(cur - base) / 1024:.0f}KiB ' f'inner_id={id(inner_group):x} task_group(a)={tg}' ) sdag.task_group.topological_sort() print('topological_sort ok') ``` Output on 3.3.2 (Python 3.11; ids and KiB values vary between runs): ```text iter=1 mem_delta=23KiB inner_id=70952194a410 task_group(a)=ReferenceError: weakly-referenced object no longer exists iter=2 mem_delta=61KiB inner_id=709521948890 task_group(a)=ReferenceError: weakly-referenced object no longer exists iter=5 mem_delta=132KiB inner_id=7095219a8ec0 task_group(a)=ReferenceError: weakly-referenced object no longer exists iter=50 mem_delta=937KiB inner_id=7095217bd650 task_group(a)=ReferenceError: weakly-referenced object no longer exists Traceback (most recent call last): File "/tmp/issue_exact.py", line 52, in <module> sub = sdag.partial_subset( File ".../airflow/serialization/definitions/dag.py", line 382, in partial_subset object.__setattr__(dag, "task_group", filter_task_group(self.task_group, None)) File ".../airflow/serialization/definitions/dag.py", line 374, in filter_task_group filtered_child = filter_task_group(child, proxy) File ".../airflow/serialization/definitions/dag.py", line 374, in filter_task_group filtered_child = filter_task_group(child, proxy) File ".../airflow/serialization/definitions/dag.py", line 363, in filter_task_group value = copy.deepcopy(value, memo) [... repeated copy.deepcopy frames ...] File ".../attr/_make.py", line 1034, in slots_getstate return {name: getattr(self, name) for name in state_attr_names} RecursionError: maximum recursion depth exceeded ``` The `outer.inner` group object in the original `sdag` is a different object at each sampled iteration (`inner_id`). The Grid code path produces the same `ReferenceError` after a single `partial_subset()` call (same `sdag` construction as above): ```python import gc from airflow.api_fastapi.core_api.services.ui.grid import _find_aggregates list(_find_aggregates(node=sdag.task_group, parent_node=None, ti_details={})) # ok (8 nodes) sdag.partial_subset(task_ids=['outer.prepare'], include_downstream=True, include_upstream=False) gc.collect() list(_find_aggregates(node=sdag.task_group, parent_node=None, ti_details={})) # ReferenceError ``` ```text File ".../airflow/api_fastapi/core_api/services/ui/grid.py", line 178, in _find_aggregates File ".../airflow/api_fastapi/core_api/services/ui/task_group.py", line 40, in <lambda> File ".../airflow/serialization/definitions/taskgroup.py", line 257, in topological_sort File ".../airflow/serialization/definitions/taskgroup.py", line 291, in _project_child_deps anc_idx = id_to_idx.get(tg.node_id) ReferenceError: weakly-referenced object no longer exists ``` In 3.3.2, `partial_subset()` is called by the API server in `POST .../clearTaskInstances` (`routes/public/task_instances.py`), `routes/ui/grid.py` (when `root` is given) and `routes/ui/structure.py`. In our deployment (DAG with a mapped task group containing a nested task group), the API server logged the following for the Grid TI summaries endpoint (`GET /ui/grid/ti_summaries/{dag_id}`): ```text 2026-10-08T06:24:14.205651359Z [2026-10-08T06:24:14.203182Z] {httptools_impl.py:427} ERROR - Exception in ASGI application 2026-10-08T06:24:14.205668653Z 2026-10-08T06:24:14.205672382Z Traceback (most recent call last): 2026-10-08T06:24:14.205675708Z File "/home/airflow/.local/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 422, in run_asgi 2026-10-08T06:24:14.205679291Z result = await app( # type: ignore[func-returns-value] 2026-10-08T06:24:14.205682579Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.205698870Z File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/applications.py", line 1159, in __call__ 2026-10-08T06:24:14.205703093Z await super().__call__(scope, receive, send) 2026-10-08T06:24:14.205706324Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/applications.py", line 96, in __call__ 2026-10-08T06:24:14.205709670Z await self.middleware_stack(scope, receive, send) 2026-10-08T06:24:14.205712846Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 186, in __call__ 2026-10-08T06:24:14.205716164Z raise exc 2026-10-08T06:24:14.205719346Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 164, in __call__ 2026-10-08T06:24:14.205722692Z await self.app(scope, receive, _send) 2026-10-08T06:24:14.205725973Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/common/http_access_log.py", line 109, in __call__ 2026-10-08T06:24:14.205729477Z await self.app(scope, receive, capture_send) 2026-10-08T06:24:14.205732674Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/gzip.py", line 78, in __call__ 2026-10-08T06:24:14.205736058Z await responder(scope, receive, send) 2026-10-08T06:24:14.205739348Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/gzip.py", line 103, in __call__ 2026-10-08T06:24:14.205742621Z await self.app(scope, receive, self.send_with_compression) 2026-10-08T06:24:14.205745830Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/base.py", line 198, in __call__ 2026-10-08T06:24:14.205749223Z raise app_exc 2026-10-08T06:24:14.205752294Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/base.py", line 144, in coro 2026-10-08T06:24:14.205755542Z await self.app(scope, receive_or_disconnect, send_no_error) 2026-10-08T06:24:14.205855421Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 63, in __call__ 2026-10-08T06:24:14.205859773Z await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) 2026-10-08T06:24:14.205863704Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app 2026-10-08T06:24:14.205867030Z raise exc 2026-10-08T06:24:14.205870052Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app 2026-10-08T06:24:14.205873264Z await app(scope, receive, sender) 2026-10-08T06:24:14.205876311Z File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__ 2026-10-08T06:24:14.205879575Z await self.app(scope, receive, send) 2026-10-08T06:24:14.205883039Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 670, in __call__ 2026-10-08T06:24:14.205886261Z await self.middleware_stack(scope, receive, send) 2026-10-08T06:24:14.205889376Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 690, in app 2026-10-08T06:24:14.205892552Z await route.handle(scope, receive, send) 2026-10-08T06:24:14.205895615Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 280, in handle 2026-10-08T06:24:14.205898803Z await self.app(scope, receive, send) 2026-10-08T06:24:14.205901916Z File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/routing.py", line 134, in app 2026-10-08T06:24:14.205905100Z await wrap_app_handling_exceptions(app, request)(scope, receive, send) 2026-10-08T06:24:14.205908277Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app 2026-10-08T06:24:14.205911864Z raise exc 2026-10-08T06:24:14.205914824Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app 2026-10-08T06:24:14.205918072Z await app(scope, receive, sender) 2026-10-08T06:24:14.206012036Z File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/routing.py", line 121, in app 2026-10-08T06:24:14.206016255Z await response(scope, receive, send) 2026-10-08T06:24:14.206019426Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 273, in __call__ 2026-10-08T06:24:14.206022767Z async with create_collapsing_task_group() as task_group: 2026-10-08T06:24:14.206025901Z File "/usr/python/lib/python3.11/contextlib.py", line 231, in __aexit__ 2026-10-08T06:24:14.206029113Z await self.gen.athrow(typ, value, traceback) 2026-10-08T06:24:14.206032247Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_utils.py", line 93, in create_collapsing_task_group 2026-10-08T06:24:14.206035548Z raise exc from exc.__cause__ or context 2026-10-08T06:24:14.206042854Z File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_core/_tasks.py", line 327, in _run_coro 2026-10-08T06:24:14.206046298Z retval = await self._coro 2026-10-08T06:24:14.206049457Z ^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206052970Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 276, in wrap 2026-10-08T06:24:14.206056357Z await func() 2026-10-08T06:24:14.206059422Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 250, in stream_response 2026-10-08T06:24:14.206062753Z async for chunk in self.body_iterator: 2026-10-08T06:24:14.206065871Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/concurrency.py", line 57, in iterate_in_threadpool 2026-10-08T06:24:14.206069210Z yield await anyio.to_thread.run_sync(_next, as_iterator) 2026-10-08T06:24:14.206072302Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206075401Z File "/home/airflow/.local/lib/python3.11/site-packages/anyio/to_thread.py", line 65, in run_sync 2026-10-08T06:24:14.206173524Z return await get_async_backend().run_sync_in_worker_thread( 2026-10-08T06:24:14.206177640Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206180835Z File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 2706, in run_sync_in_worker_thread 2026-10-08T06:24:14.206184296Z return await future 2026-10-08T06:24:14.206187506Z ^^^^^^^^^^^^ 2026-10-08T06:24:14.206190569Z File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 1100, in run 2026-10-08T06:24:14.206193887Z result = context.run(func, *args) 2026-10-08T06:24:14.206196966Z ^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206200114Z File "/home/airflow/.local/lib/python3.11/site-packages/starlette/concurrency.py", line 46, in _next 2026-10-08T06:24:14.206203466Z return next(iterator) 2026-10-08T06:24:14.206206560Z ^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206209779Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 516, in _generate 2026-10-08T06:24:14.206213208Z summary = _build_ti_summaries( 2026-10-08T06:24:14.206216620Z ^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206219666Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 430, in _build_ti_summaries 2026-10-08T06:24:14.206389195Z nodes = list(get_node_summaries()) 2026-10-08T06:24:14.206393373Z ^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206396537Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 407, in get_node_summaries 2026-10-08T06:24:14.206399875Z for node, _ in _find_aggregates( 2026-10-08T06:24:14.206406086Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/grid.py", line 179, in _find_aggregates 2026-10-08T06:24:14.206409664Z for child_node, child_summary in _find_aggregates( 2026-10-08T06:24:14.206413191Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/grid.py", line 178, in _find_aggregates 2026-10-08T06:24:14.206416626Z for child in get_task_group_children_getter()(node, group_dict): 2026-10-08T06:24:14.206419866Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206422946Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/task_group.py", line 40, in <lambda> 2026-10-08T06:24:14.206426589Z return lambda task_group, group_dict=None: task_group.topological_sort(group_dict=group_dict) 2026-10-08T06:24:14.206429793Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206433050Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/serialization/definitions/taskgroup.py", line 257, in topological_sort 2026-10-08T06:24:14.206436421Z deps = self._project_child_deps(i, child, id_to_idx, group_dict) 2026-10-08T06:24:14.206439499Z ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 2026-10-08T06:24:14.206442537Z File "/home/airflow/.local/lib/python3.11/site-packages/airflow/serialization/definitions/taskgroup.py", line 291, in _project_child_deps 2026-10-08T06:24:14.206445791Z anc_idx = id_to_idx.get(tg.node_id) 2026-10-08T06:24:14.206448813Z ^^^^^^^^^^ 2026-10-08T06:24:14.206451913Z ReferenceError: weakly-referenced object no longer exists ``` ### What you think should happen instead? `partial_subset()` should not modify the DAG it is called on. Replacing the loop with one that includes inherited attrs fields: ```python for attr in (field.name for field in attrs.fields(type(group))): ``` gives, for the same script on 3.3.2 (and the `_find_aggregates` walk still returns 8 nodes after 200 `partial_subset()` calls): ```text iter=1 mem_delta=4KiB inner_id=7841eb5c0260 task_group(a)=outer.inner iter=2 mem_delta=4KiB inner_id=7841eb5c0260 task_group(a)=outer.inner iter=5 mem_delta=5KiB inner_id=7841eb5c0260 task_group(a)=outer.inner iter=50 mem_delta=13KiB inner_id=7841eb5c0260 task_group(a)=outer.inner iter=200 mem_delta=28KiB inner_id=7841eb5c0260 task_group(a)=outer.inner topological_sort ok ``` After applying this change to our 3.3.2 deployment, the `ReferenceError` no longer appears in the API server logs. ### Operating System _No response_ ### Deployment Other Docker-based deployment ### Apache Airflow Provider(s) _No response_ ### Versions of Apache Airflow Providers _No response_ ### Official Helm Chart version Not Applicable ### Kubernetes Version _No response_ ### Helm Chart configuration _No response_ ### Docker Image customizations - Image based on `apache/airflow:slim-3.3.2-python3.11` - CeleryExecutor, Redis 7.2, PostgreSQL 13 ### Anything else? Possibly related: #58499 (`ReferenceError` in API Server during Dynamic Mapping of Nested TaskGroups), closed 2025-11-20. ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md) -- 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]
