1fanwang commented on code in PR #66788:
URL: https://github.com/apache/airflow/pull/66788#discussion_r3454507309
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -257,6 +257,21 @@ def _update_dag_owner_links(dag_owner_links: dict[str,
str], dm: DagModel, *, se
)
+# Dialect-specific message prefixes for unique-constraint IntegrityErrors.
Kept in sync with
+# ``airflow.api_fastapi.common.exceptions._UniqueConstraintErrorHandler``.
+_UNIQUE_CONSTRAINT_ERROR_PREFIXES = (
+ "UNIQUE constraint failed", # SQLite
+ "Duplicate entry", # MySQL
+ "violates unique constraint", # Postgres
Review Comment:
Went back and raced two processors live on `main`: this arm never executes —
the `IntegrityError` fires at the outer `update_dag_parsing_results_in_db`
commit, not here, so there's no `exc.orig` to match or guard against.
`run_with_db_retries` already retries it into a no-op. Your write-site `ON
CONFLICT DO NOTHING` would be the right place if it were ever needed. Closing.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -295,6 +310,29 @@ def _serialize_dag_capturing_errors(
return []
except OperationalError:
raise
+ except IntegrityError as exc:
Review Comment:
Same finding from the live race — this line is never reached (the error
surfaces a frame up and the retry recovers), so there's nothing to downgrade to
a warning. Closing rather than re-tuning it.
##########
airflow-core/tests/unit/dag_processing/test_collection.py:
##########
@@ -651,6 +651,109 @@ def test_serialized_dag_errors_are_import_errors(
assert len(dag_import_error_listener.existing) == 0
assert dag_import_error_listener.new["abc.py"] ==
import_error.stacktrace
+ @pytest.mark.parametrize(
+ "orig_message",
+ [
+ pytest.param(
Review Comment:
You were right. I built exactly that — two sessions racing a brand-new dag
on real Postgres. It showed the mocked test was circular: the real
`IntegrityError` fires at the outer commit, so this arm never catches it, and
`run_with_db_retries` already recovers it. Raw output in the description.
That's what led to closing.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -295,6 +310,29 @@ def _serialize_dag_capturing_errors(
return []
except OperationalError:
raise
+ except IntegrityError as exc:
+ # Multiple Dag processors writing the same brand-new Dag can race on
the INSERT.
+ # The loser's transaction is already invalid, so we must roll the
session back to
+ # avoid PendingRollbackError on subsequent per-Dag work in this
parsing cycle.
+ # The winning peer already produced the correct row, so this is not an
import error
+ # and we don't retry. Non-unique IntegrityErrors (e.g. NOT-NULL
violations from a
+ # genuinely malformed Dag) fall through to the generic Exception arm.
Review Comment:
To answer directly: I reproduced it live on `main` (real Postgres, two
processors racing a brand-new dag). The duplicate-key `IntegrityError` is real,
but it surfaces at the outer `update_dag_parsing_results_in_db` commit — not
here in `_serialize_dag_capturing_errors`, which only `session.add`s and
returns `[]`. `run_with_db_retries` then retries it into a no-op. So this arm
never sees the real race. Full trace in the description.
--
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]
