fix(GitMirror): 同步失败时删除本地镜像并排到队尾,大仓强制 HTTP/1.1
This commit is contained in:
1 parent
c81637dc71
commit
3ba3740624
2 files changed
+62
-22
No files matched your search
+43
-22
@@ -59,7 +59,14 @@ Manager = MirrorManager(Cfg, Gitea, Github, Store)
|
||||
|
||||
|
||||
class SyncWorker:
|
||||
"""去重的同步队列 + 工作线程(每仓串行,失败退避重试)"""
|
||||
"""去重的同步队列 + 工作线程
|
||||
|
||||
失败处理:丢弃本地镜像并把该仓库**追加到队尾**(不阻塞其它仓库)。
|
||||
每个仓库在单轮回填中最多重排 MAX_REQUEUE 次,成功即清零;
|
||||
每 15 分钟的定时回填会重置预算,继续兜底。
|
||||
"""
|
||||
|
||||
MAX_REQUEUE = 10
|
||||
|
||||
def __init__(self, Workers: int):
|
||||
self.Queue: queue.Queue[tuple[str, dict | None]] = queue.Queue()
|
||||
@@ -67,6 +74,7 @@ class SyncWorker:
|
||||
self._Guard = threading.Lock()
|
||||
self._Workers = Workers
|
||||
self._Threads: list[threading.Thread] = []
|
||||
self._Requeues: dict[str, int] = {}
|
||||
|
||||
def Start(self) -> None:
|
||||
for Index in range(self._Workers):
|
||||
@@ -87,38 +95,50 @@ class SyncWorker:
|
||||
with self._Guard:
|
||||
return len(self._Pending)
|
||||
|
||||
def ResetRequeueBudget(self) -> None:
|
||||
"""新一轮回填:清空重排计数"""
|
||||
with self._Guard:
|
||||
self._Requeues.clear()
|
||||
|
||||
def _Run(self) -> None:
|
||||
while True:
|
||||
RepoName, Repo = self.Queue.get()
|
||||
Requeue = False
|
||||
try:
|
||||
self._SyncWithRetry(RepoName, Repo)
|
||||
Requeue = self._SyncOnce(RepoName, Repo)
|
||||
except Exception: # 兜底,避免工作线程退出
|
||||
logger.exception('同步任务异常:%s', RepoName)
|
||||
finally:
|
||||
with self._Guard:
|
||||
self._Pending.discard(RepoName)
|
||||
self.Queue.task_done()
|
||||
if Requeue:
|
||||
self.Put(RepoName, Repo)
|
||||
|
||||
def _SyncWithRetry(self, RepoName: str, Repo: dict | None) -> None:
|
||||
Delays = [10, 30, 60, 120]
|
||||
LastError: Exception | None = None
|
||||
for Attempt, Delay in enumerate([0] + Delays):
|
||||
if Delay:
|
||||
time.sleep(Delay)
|
||||
try:
|
||||
Info = Repo or Gitea.GetRepo(RepoName)
|
||||
if not Info:
|
||||
raise MirrorError(f'Gitea 中不存在仓库 {RepoName}')
|
||||
Manager.Sync(Info)
|
||||
return
|
||||
except MirrorError as Error:
|
||||
LastError = Error
|
||||
logger.warning('同步失败(第 %s 次)%s:%s', Attempt + 1, RepoName, Error)
|
||||
except Exception as Error: # 网络异常等
|
||||
LastError = Error
|
||||
logger.warning('同步异常(第 %s 次)%s:%s', Attempt + 1, RepoName, Error)
|
||||
Store.SetStatus(RepoName, Status='failed', LastError=str(LastError)[:1000])
|
||||
logger.error('同步最终失败 %s:%s', RepoName, LastError)
|
||||
def _SyncOnce(self, RepoName: str, Repo: dict | None) -> bool:
|
||||
"""执行一次同步;返回 True 表示需要重新排到队尾"""
|
||||
try:
|
||||
Info = Repo or Gitea.GetRepo(RepoName)
|
||||
if not Info:
|
||||
raise MirrorError(f'Gitea 中不存在仓库 {RepoName}')
|
||||
Manager.Sync(Info)
|
||||
with self._Guard:
|
||||
self._Requeues.pop(RepoName, None)
|
||||
return False
|
||||
except Exception as Error:
|
||||
logger.warning('同步失败 %s:%s', RepoName, Error)
|
||||
# 丢弃可能损坏的本地镜像(大仓 HTTP/2 断流后无法续传)
|
||||
Manager.DropLocal(RepoName)
|
||||
with self._Guard:
|
||||
Count = self._Requeues.get(RepoName, 0) + 1
|
||||
self._Requeues[RepoName] = Count
|
||||
if Count <= self.MAX_REQUEUE:
|
||||
Store.SetStatus(RepoName, Status='pending', LastError=str(Error)[:1000])
|
||||
logger.info('已排到队尾(第 %s 次重排):%s', Count, RepoName)
|
||||
return True
|
||||
Store.SetStatus(RepoName, Status='failed', LastError=str(Error)[:1000])
|
||||
logger.error('超过重排上限,标记失败 %s:%s', RepoName, Error)
|
||||
return False
|
||||
|
||||
|
||||
Worker = SyncWorker(Cfg.Concurrency)
|
||||
@@ -130,6 +150,7 @@ Worker = SyncWorker(Cfg.Concurrency)
|
||||
|
||||
def Backfill() -> dict:
|
||||
"""枚举组织全部仓库并入队(幂等)"""
|
||||
Worker.ResetRequeueBudget()
|
||||
Repos = Gitea.ListRepos()
|
||||
Queued = 0
|
||||
for Repo in Repos:
|
||||
|
||||
@@ -21,6 +21,7 @@ import json
|
||||
import logging
|
||||
import os
|
||||
import secrets as py_secrets
|
||||
import shutil
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import threading
|
||||
@@ -372,6 +373,9 @@ class MirrorManager:
|
||||
Cmd = [
|
||||
'git',
|
||||
'-c', 'core.longpaths=true',
|
||||
# 大仓库(数百 MB)走 HTTP/2 常被服务端 reset(curl 92),强制 HTTP/1.1
|
||||
'-c', 'http.version=HTTP/1.1',
|
||||
'-c', 'http.postBuffer=524288000',
|
||||
# 清空 credential helper,杜绝全局 GCM/OAuth 凭据抢先命中
|
||||
'-c', 'credential.helper=',
|
||||
] + Args
|
||||
@@ -438,6 +442,21 @@ class MirrorManager:
|
||||
logger.info('镜像完成 %s', RepoName)
|
||||
return {'RepoName': RepoName, 'GithubName': GithubName, 'Status': 'ok'}
|
||||
|
||||
def DropLocal(self, RepoName: str) -> None:
|
||||
"""删除本地 bare 镜像
|
||||
|
||||
大仓库 fetch 中途断流(curl 92 / early EOF)后本地 pack 可能残缺,
|
||||
继续在同一目录重试容易重复失败;直接丢弃,下次从头干净拉起。
|
||||
"""
|
||||
Local = os.path.join(self.Cfg.ReposDir, f'{RepoName}.git')
|
||||
if not os.path.isdir(Local):
|
||||
return
|
||||
try:
|
||||
shutil.rmtree(Local)
|
||||
logger.info('已删除本地镜像:%s', RepoName)
|
||||
except OSError as Error:
|
||||
logger.warning('删除本地镜像失败 %s:%s', RepoName, Error)
|
||||
|
||||
def HandleRepoDeleted(self, RepoName: str) -> None:
|
||||
"""源仓库被删除:按配置归档(默认)/ 真删 / 忽略"""
|
||||
GithubName = self._GithubName(RepoName)
|
||||
|
||||
Reference in new issue
Block a user