From 4e390b7dade663e56bf4c7e4bd38c35954e9a85f Mon Sep 17 00:00:00 2001
From: Maksim Totmin <maksim@totmin.ru>
Date: Wed, 8 Apr 2026 18:57:51 +0700
Subject: [PATCH] =?UTF-8?q?fix:=20=D0=B7=D0=B0=D1=89=D0=B8=D1=82=D0=B0=20?=
 =?UTF-8?q?=D0=BE=D1=82=20=D0=BF=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5=D0=BB?=
 =?UTF-8?q?=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=B2=D1=8B=D0=BF=D0=BE=D0=BB?=
 =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D0=B8=D1=8F=20DAG=20=D0=BF=D1=80=D0=B8=20ret?=
 =?UTF-8?q?ry?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- двойная проверка: в actions (по списку issues) и в client (API запрос)
- если DAG уже running — retry отклоняется, сразу alert
- предотвращает дублирование выполнения отчётов
---
 airflow_monitor/actions.py | 22 +++++++++++++++++++++-
 airflow_monitor/client.py  | 29 +++++++++++++++++++++++++++++
 airflow_monitor/monitor.py |  2 +-
 3 files changed, 51 insertions(+), 2 deletions(-)

diff --git a/airflow_monitor/actions.py b/airflow_monitor/actions.py
index 1a9647d..9b6bd24 100644
--- a/airflow_monitor/actions.py
+++ b/airflow_monitor/actions.py
@@ -131,13 +131,18 @@ class ActionHandler:
             config.max_retries, config.retry_wait,
         )
 
-    def handle_issue(self, issue: DagIssue) -> str:
+    def handle_issue(self, issue: DagIssue, all_issues: list[DagIssue]) -> str:
         """Process a single issue.
 
+        Args:
+            issue: The issue to handle.
+            all_issues: All issues in this cycle (to check for running runs of same DAG).
+
         Returns:
             "already_handled" - previously alerted, skip
             "retried" - retry initiated, needs re-check
             "needs_alert" - retry exhausted, alert needed
+            "skip_running" - DAG already has a running instance, unsafe to retry
         """
         dag_id = issue.dag_id
         run_id = issue.dag_run_id
@@ -161,6 +166,21 @@ class ActionHandler:
             logger.debug("  → already_handled: alert was sent previously")
             return "already_handled"
 
+        # SAFETY: Do not retry if this DAG already has a running instance.
+        # Prevents parallel execution of the same DAG (critical for reports).
+        if issue.issue_type == "failed":
+            has_running = any(
+                i.dag_id == dag_id and i.state == "running"
+                for i in all_issues
+            )
+            if has_running:
+                logger.warning(
+                    "  → skip_running: DAG %s has a running instance, "
+                    "will NOT retry failed run %s to avoid parallel execution",
+                    dag_id, run_id,
+                )
+                return "needs_alert"
+
         # Can we retry?
         if retry_count < self._config.max_retries:
             logger.debug(
diff --git a/airflow_monitor/client.py b/airflow_monitor/client.py
index 5ffd20d..5f217b4 100644
--- a/airflow_monitor/client.py
+++ b/airflow_monitor/client.py
@@ -259,11 +259,40 @@ class AirflowClient:
             )
             return []
 
+    def has_running_run(self, dag_id: str) -> bool:
+        """Check if DAG has any currently running runs.
+
+        Used as a safety check before retrying a failed run
+        to prevent parallel execution.
+        """
+        try:
+            runs = self.get_dag_runs(dag_id, states=["running"])
+            if runs:
+                logger.warning(
+                    "DAG %s has %d running run(s), unsafe to retry",
+                    dag_id, len(runs),
+                )
+                return True
+            return False
+        except AirflowAPIError:
+            # If we can't check, assume it's running (safe side)
+            logger.warning("Cannot check running state for %s, assuming running", dag_id)
+            return True
+
     def clear_dag_run(self, dag_id: str, dag_run_id: str) -> bool:
         """Clear (retry) a DAG run by resetting failed task instances.
 
+        Safety: checks for running instances before clearing.
         Returns True on success, False on failure.
         """
+        # Double-check: do not retry if DAG already has a running instance
+        if self.has_running_run(dag_id):
+            logger.warning(
+                "SAFETY: Refusing to clear %s/%s — DAG has running instance",
+                dag_id, dag_run_id,
+            )
+            return False
+
         encoded_dag_id = quote(dag_id, safe="")
         encoded_run_id = quote(dag_run_id, safe="")
 
diff --git a/airflow_monitor/monitor.py b/airflow_monitor/monitor.py
index 53dc759..820aa6c 100644
--- a/airflow_monitor/monitor.py
+++ b/airflow_monitor/monitor.py
@@ -140,7 +140,7 @@ class Monitor:
         phase_start = time.monotonic()
         needs_recheck = []
         for issue in all_issues:
-            action = self._actions.handle_issue(issue)
+            action = self._actions.handle_issue(issue, all_issues)
             logger.info(
                 "Phase 2/5 [Handle]: DAG %s run %s → %s (type=%s, duration=%.0fs)",
                 issue.dag_id, issue.dag_run_id, action,