fix self heal restart

iaai_scraper/worker/self_heal.py

@@ -12,6 +12,8 @@ logger = logging.getLogger("iaai_scraper.worker.self_heal")
 
 GLOBAL_PROGRESS_TS_KEY = "iaai:state:last_progress_ts"
 SELF_HEAL_RESTART_LOCK_KEY = "iaai:state:self_heal_restart_in_progress"
+SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -76,6 +78,22 @@ def _read_last_progress_ts(redis_client: Redis) -> int | None:
 	return max_ts or None
 
 
+def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict[str, int]]:
+	"""Есть ли признаки активной/зависшей работы, даже если очередь пуста."""
+	queue_len = _safe_int(redis_client.llen(queue_name), 0)
+	has_lock = 1 if redis_client.get(SYNC_LISTING_LOCK_KEY) else 0
+	has_task_progress = 0
+	for _ in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+		has_task_progress = 1
+		break
+	flags = {
+		"queue_len": queue_len,
+		"has_lock": has_lock,
+		"has_task_progress": has_task_progress,
+	}
+	return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
+
+
 def _kill_worker_process() -> None:
 	pid_file = "/tmp/celery-worker.pid"
 	pid: int | None = None
@@ -101,7 +119,7 @@ def _kill_worker_process() -> None:
 		# Если процесс ещё жив — принудительно убиваем.
 		os.kill(pid, 0)
 		logger.error("Self-heal: worker pid=%s did not stop after SIGTERM; sending SIGKILL", pid)
-		os.kill(pid, signal.SIGKILL)
+		os.kill(pid, SIGKILL_FALLBACK)
 	except ProcessLookupError:
 		pass
 	except Exception:
@@ -137,8 +155,8 @@ def main() -> None:
 				redis_client = _get_redis()
 				redis_client.ping()
 
-			queue_len = _safe_int(redis_client.llen(queue_name), 0)
+			has_inflight, inflight = _has_inflight_work(redis_client, queue_name)
-			if queue_len <= 0:
+			if not has_inflight:
 				time.sleep(check_interval)
 				continue
 
@@ -151,8 +169,8 @@ def main() -> None:
 					time.sleep(check_interval)
 					continue
 				logger.warning(
-					"Self-heal: queue=%d but no progress timestamp found after startup grace",
+					"Self-heal: inflight=%s but no progress timestamp found after startup grace",
-					queue_len,
+					inflight,
 				)
 				age = stall_seconds + 1
 
@@ -174,8 +192,8 @@ def main() -> None:
 				continue
 
 			logger.error(
-				"Self-heal: detected global stall (queue=%d, progress_age=%ss > %ss). Restarting worker process...",
+				"Self-heal: detected global stall (inflight=%s, progress_age=%ss > %ss). Restarting worker process...",
-				queue_len,
+				inflight,
 				age,
 				stall_seconds,
 			)