diff --git a/GitMirror/app.py b/GitMirror/app.py index bff1d298..29141a62 100644 --- a/GitMirror/app.py +++ b/GitMirror/app.py @@ -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: diff --git a/GitMirror/mirror.py b/GitMirror/mirror.py index 8f76d53b..0180d567 100644 --- a/GitMirror/mirror.py +++ b/GitMirror/mirror.py @@ -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)