feat(worker): git 锁从 flock 升级 DB 锁表(pipeline_git_locks)，跨主机多 worker 生效，TTL 180s 崩溃自释放

pipeline_service/agent_loop.py

@@ -11,7 +11,6 @@ v3.4.0 新增：
 """
 
 import asyncio
-import fcntl
 import hashlib
 import json
 import os
@@ -263,37 +262,46 @@ async def _write_code_file(filepath, content):
         return False, str(e)
 
 
-_GIT_LOCK_DIR = os.environ.get('PIPELINE_GIT_LOCK_DIR', '/tmp/pipeline_git_locks')
+_GIT_LOCK_TTL = 180  # 秒：覆盖 git 单次操作最坏时长(clone 120s)；过期可被原子接管
 
 
-def _git_lock_file(repo_dir):
+def _git_lock_key(repo_dir):
-    """同 repo 的 git 操作共用一把锁文件。
+    """同 repo 的 git 操作共用一把锁。lock_key = sha256(repo_abs_path)。"""
-
+    return hashlib.sha256(os.path.abspath(repo_dir).encode('utf-8')).hexdigest()
-    本机多 worker 用 flock 串行（锁文件在本地 /tmp，不落 NFS，flock 可靠）；
-    跨主机分布时需换 DB/Redis 锁，锁文件位置可用 PIPELINE_GIT_LOCK_DIR 覆盖。
-    """
-    try:
-        os.makedirs(_GIT_LOCK_DIR, exist_ok=True)
-    except OSError:
-        pass
-    key = hashlib.sha256(os.path.abspath(repo_dir).encode('utf-8')).hexdigest()[:32]
-    return os.path.join(_GIT_LOCK_DIR, key + '.lock')
 
 
 @asynccontextmanager
 async def _git_lock(repo_dir, timeout=90):
-    """git 操作串行锁：只锁 git 那几秒，任务其它部分(LLM/写文件)完全并行。"""
+    """git 操作串行锁（DB 版，跨主机多 worker 生效）：只锁 git 那几秒，任务其它部分完全并行。
-    path = _git_lock_file(repo_dir)
+
-    fd = open(path, 'w')
+    用 pipeline_git_locks 表 + ON DUPLICATE KEY UPDATE + expires_at TTL：
+    - 原子抢占：INSERT ... ON DUPLICATE KEY UPDATE，仅当 expires_at 过期才接管他人锁
+    - 崩溃自动释放：expires_at 过期后下一个申请者原子接管
+    - 释放：DELETE ... WHERE token 匹配（避免误删他人锁）
+    """
+    from appPublic.uniqueID import getID
+
+    key = _git_lock_key(repo_dir)
+    token = getID()
     deadline = time.time() + timeout
     acquired = False
+    db = _get_db()
     try:
-        while True:
+        while not acquired:
-            try:
+            async with db.sqlorContext("pipeline") as sor:
-                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                await sor.sqlExe(
-                acquired = True
+                    "INSERT INTO pipeline_git_locks (lock_key, token, expires_at) "
-                break
+                    "VALUES (${k}$, ${t}$, DATE_ADD(NOW(), INTERVAL " + str(_GIT_LOCK_TTL) + " SECOND)) "
-            except BlockingIOError:
+                    "ON DUPLICATE KEY UPDATE "
+                    "token = IF(expires_at < NOW(), VALUES(token), token), "
+                    "expires_at = IF(expires_at < NOW(), DATE_ADD(NOW(), INTERVAL " + str(_GIT_LOCK_TTL) + " SECOND), expires_at)",
+                    {"k": key, "t": token})
+                r = await sor.sqlExe(
+                    "SELECT token FROM pipeline_git_locks WHERE lock_key=${k}$ AND token=${t}$",
+                    {"k": key, "t": token})
+                if r:
+                    acquired = True
+            if not acquired:
                 if time.time() > deadline:
                     raise TimeoutError(f'git lock timeout after {timeout}s: {repo_dir}')
                 await asyncio.sleep(0.3)
@@ -301,10 +309,12 @@ async def _git_lock(repo_dir, timeout=90):
     finally:
         if acquired:
             try:
-                fcntl.flock(fd, fcntl.LOCK_UN)
+                async with db.sqlorContext("pipeline") as sor:
-            except OSError:
+                    await sor.sqlExe(
+                        "DELETE FROM pipeline_git_locks WHERE lock_key=${k}$ AND token=${t}$",
+                        {"k": key, "t": token})
+            except Exception:
                 pass
-        fd.close()
 
 
 async def _git_setup(workdir):