457 lines
17 KiB
Python
457 lines
17 KiB
Python
"""Mirror lifecycle management."""
|
||
|
||
import fcntl
|
||
import json
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
from contextlib import contextmanager
|
||
from pathlib import Path
|
||
from typing import Any, Dict, Generator, List, Optional, Tuple
|
||
|
||
from .config import MirrorConfig
|
||
from .db import MappingDB
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MirrorError(Exception):
|
||
pass
|
||
|
||
|
||
class Mirror:
|
||
"""Manages a single SVN↔Git mirror's lifecycle on disk."""
|
||
|
||
def __init__(self, config: MirrorConfig, data_dir: Path):
|
||
self.config = config
|
||
self.base_dir = data_dir / "mirrors" / config.id
|
||
self.canonical_dir = self.base_dir / "canonical-repo.git"
|
||
self.svn_wc_dir = self.base_dir / "svn-wc"
|
||
self.mapping_db_path = self.base_dir / "mapping.db"
|
||
self.authors_path = self.base_dir / "authors.txt"
|
||
self.state_path = self.base_dir / "state.json"
|
||
self.lock_path = self.base_dir / "sync.lock"
|
||
self._db: Optional[MappingDB] = None
|
||
|
||
@contextmanager
|
||
def sync_lock(self) -> Generator[None, None, None]:
|
||
"""Acquire an exclusive flock on the mirror's lock file.
|
||
|
||
Guarantees that only one process/thread syncs this mirror at a
|
||
time. Releases automatically when the context exits.
|
||
"""
|
||
self.lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||
fd = os.open(str(self.lock_path), os.O_CREAT | os.O_RDWR, 0o644)
|
||
try:
|
||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||
yield
|
||
finally:
|
||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||
os.close(fd)
|
||
|
||
@property
|
||
def db(self) -> MappingDB:
|
||
if self._db is None:
|
||
self._db = MappingDB(self.mapping_db_path)
|
||
return self._db
|
||
|
||
@property
|
||
def exists(self) -> bool:
|
||
return self.base_dir.exists()
|
||
|
||
# ─── Lifecycle ────────────────────────────────────────────
|
||
|
||
def create(self) -> None:
|
||
"""Create directory structure, init bare Git repo and mapping DB."""
|
||
if self.exists:
|
||
raise MirrorError(
|
||
f"Mirror {self.config.id} already exists at {self.base_dir}"
|
||
)
|
||
|
||
logger.info("Creating mirror %s at %s", self.config.id, self.base_dir)
|
||
self.base_dir.mkdir(parents=True, exist_ok=True)
|
||
self.svn_wc_dir.mkdir(exist_ok=True)
|
||
|
||
# Init bare Git repo (our canonical copy)
|
||
self._git("init", "--bare", str(self.canonical_dir))
|
||
|
||
# Init mapping DB
|
||
_ = self.db
|
||
|
||
# Write authors file for git svn
|
||
self._write_authors_file()
|
||
|
||
# Write initial state
|
||
self._write_state({
|
||
"initial_import_done": False,
|
||
"last_svn_revision": None,
|
||
"last_git_commit": None,
|
||
"created_at": _now_iso(),
|
||
"last_sync_at": None,
|
||
})
|
||
|
||
logger.info("Mirror %s created successfully", self.config.id)
|
||
|
||
def destroy(self) -> None:
|
||
"""Remove all mirror data from disk."""
|
||
if not self.exists:
|
||
raise MirrorError(f"Mirror {self.config.id} does not exist")
|
||
|
||
logger.warning("Destroying mirror %s at %s", self.config.id, self.base_dir)
|
||
shutil.rmtree(self.base_dir)
|
||
logger.info("Mirror %s destroyed", self.config.id)
|
||
|
||
def init_import(self) -> None:
|
||
"""Run initial SVN import via git svn clone --no-metadata."""
|
||
if not self.exists:
|
||
self.create()
|
||
|
||
state = self._read_state()
|
||
if state.get("initial_import_done"):
|
||
logger.info("Mirror %s already imported (skipping)", self.config.id)
|
||
return
|
||
|
||
self._check_git_svn_available()
|
||
|
||
import_dir = self.base_dir / ".import-tmp"
|
||
if import_dir.exists():
|
||
shutil.rmtree(import_dir)
|
||
|
||
try:
|
||
self._cache_svn_credentials()
|
||
self._run_git_svn_clone(import_dir)
|
||
self._push_import_to_bare(import_dir)
|
||
mappings = self._read_rev_maps(import_dir)
|
||
self._write_mappings_to_db(mappings)
|
||
self._finalize_import_state()
|
||
except BaseException:
|
||
logger.exception("Initial import failed for %s", self.config.id)
|
||
if import_dir.exists():
|
||
shutil.rmtree(import_dir)
|
||
raise
|
||
else:
|
||
shutil.rmtree(import_dir, ignore_errors=True)
|
||
|
||
# ─── Status ───────────────────────────────────────────────
|
||
|
||
def status(self) -> Dict[str, Any]:
|
||
"""Return a snapshot of mirror state for display."""
|
||
if not self.base_dir.exists():
|
||
return {
|
||
"id": self.config.id,
|
||
"status": "not_created",
|
||
"svn_url": self.config.svn.url,
|
||
"enabled": self.config.enabled,
|
||
}
|
||
|
||
state = self._read_state()
|
||
rev = self.db.get_max_revision_across_all_branches() if self.mapping_db_path.exists() else None
|
||
count = self.db.mapping_count() if self.mapping_db_path.exists() else 0
|
||
|
||
return {
|
||
"id": self.config.id,
|
||
"status": "created",
|
||
"imported": state.get("initial_import_done", False),
|
||
"svn_url": self.config.svn.url,
|
||
"enabled": self.config.enabled,
|
||
"last_svn_revision": rev,
|
||
"mapping_count": count,
|
||
"base_dir": str(self.base_dir),
|
||
"created_at": state.get("created_at"),
|
||
"last_sync_at": state.get("last_sync_at"),
|
||
}
|
||
|
||
# ─── Private helpers ──────────────────────────────────────
|
||
|
||
def _check_git_svn_available(self):
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "svn", "--version"],
|
||
capture_output=True, text=True, timeout=30,
|
||
)
|
||
if r.returncode != 0:
|
||
raise MirrorError(
|
||
"git svn not available (exit code %d): %s" % (r.returncode, r.stderr)
|
||
)
|
||
except FileNotFoundError:
|
||
raise MirrorError("git svn not found – is git-svn installed?")
|
||
except subprocess.TimeoutExpired:
|
||
raise MirrorError("git svn --version timed out")
|
||
|
||
def _cache_svn_credentials(self):
|
||
"""Pre-populate SVN auth cache so git svn doesn't prompt."""
|
||
svn_cfg = self.config.svn
|
||
if not svn_cfg.username:
|
||
return
|
||
import subprocess
|
||
try:
|
||
r = subprocess.run(
|
||
["svn", "info", "--non-interactive",
|
||
svn_cfg.url.rstrip("/") + "/" + svn_cfg.trunk]
|
||
+ svn_cfg.auth_args(),
|
||
capture_output=True, timeout=30,
|
||
)
|
||
if r.returncode != 0:
|
||
err = (r.stderr or b"").decode("utf-8", errors="replace").strip()
|
||
logger.warning("SVN credential cache failed (non-fatal): %s", err)
|
||
else:
|
||
logger.debug("SVN credentials cached")
|
||
except Exception as e:
|
||
logger.warning("SVN credential cache skipped: %s", e)
|
||
|
||
def _run_git_svn_clone(self, target: Path):
|
||
"""Execute git svn clone --no-metadata for the initial import."""
|
||
svn_cfg = self.config.svn
|
||
args = ["git", "svn", "clone", "--no-metadata"]
|
||
|
||
if svn_cfg.layout == "std":
|
||
args.append("--stdlayout")
|
||
else:
|
||
args.extend(["-T", svn_cfg.trunk])
|
||
args.extend(["-b", svn_cfg.branches])
|
||
args.extend(["-t", svn_cfg.tags])
|
||
|
||
if svn_cfg.username:
|
||
args.extend(["--username", svn_cfg.username])
|
||
|
||
if self.config.authors:
|
||
args.extend(["--authors-file", str(self.authors_path)])
|
||
|
||
args.extend([svn_cfg.url, str(target)])
|
||
|
||
logger.info("Starting git svn clone (this may take a while)…")
|
||
logger.debug("Running: %s", " ".join(str(a) for a in args))
|
||
|
||
password = svn_cfg.password or ""
|
||
process = subprocess.Popen(
|
||
args,
|
||
stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
)
|
||
|
||
for line in process.stdout or []:
|
||
line = line.rstrip()
|
||
if line:
|
||
logger.info("[git svn] %s", line)
|
||
if "Password for" in line and password:
|
||
process.stdin.write(password + "\n")
|
||
process.stdin.flush()
|
||
|
||
ret = process.wait()
|
||
if ret != 0:
|
||
raise MirrorError(f"git svn clone failed with exit code {ret}")
|
||
|
||
logger.info("git svn clone completed")
|
||
|
||
def _push_import_to_bare(self, import_dir: Path):
|
||
"""Push all refs from the temporary clone into our canonical bare repo.
|
||
|
||
git svn stores branches under refs/remotes/origin/*, NOT refs/heads/*.
|
||
We need to map:
|
||
refs/remotes/origin/trunk → refs/heads/master
|
||
refs/remotes/origin/branches/* → refs/heads/*
|
||
refs/remotes/origin/tags/* → refs/tags/*
|
||
"""
|
||
logger.info("Pushing imported refs to canonical bare repo…")
|
||
|
||
# -- setup remote ---------------------------------------------------
|
||
self._git("-C", str(import_dir), "remote", "remove", "origin",
|
||
check=False)
|
||
self._git("-C", str(import_dir), "remote", "add", "bare",
|
||
str(self.canonical_dir))
|
||
|
||
# -- push trunk → master --------------------------------------------
|
||
self._git("-C", str(import_dir), "push", "bare",
|
||
"refs/remotes/origin/trunk:refs/heads/master",
|
||
"--force")
|
||
|
||
# -- push other branches --------------------------------------------
|
||
# List remote refs and push each as a local branch
|
||
r = self._git("-C", str(import_dir), "for-each-ref",
|
||
"--format=%(refname)", "refs/remotes/origin/",
|
||
check=True)
|
||
for ref in r.stdout.strip().splitlines():
|
||
ref = ref.strip()
|
||
if not ref:
|
||
continue
|
||
# Determine target ref
|
||
if ref == "refs/remotes/origin/trunk":
|
||
continue # already pushed above
|
||
if ref.startswith("refs/remotes/origin/tags/"):
|
||
# git svn uses refs/remotes/origin/tags/* – convert to real tags
|
||
tag_name = ref.removeprefix("refs/remotes/origin/tags/")
|
||
target = f"refs/tags/{tag_name}"
|
||
else:
|
||
branch_name = ref.removeprefix("refs/remotes/origin/")
|
||
target = f"refs/heads/{branch_name}"
|
||
self._git("-C", str(import_dir), "push", "bare",
|
||
f"{ref}:{target}", "--force")
|
||
|
||
# -- push any real tags git svn might have created ------------------
|
||
self._git("-C", str(import_dir), "push", "bare", "--tags", "--force",
|
||
check=False)
|
||
|
||
def _read_rev_maps(self, import_dir: Path) -> List[Tuple[int, str, str, str]]:
|
||
"""
|
||
Parse .rev_map.* files from the git svn temp clone.
|
||
|
||
Returns list of (svn_revision, git_hash, svn_branch, git_ref).
|
||
"""
|
||
svn_dir = import_dir / ".git" / "svn"
|
||
if not svn_dir.exists():
|
||
logger.warning("No .git/svn directory found – cannot build mapping")
|
||
return []
|
||
|
||
mappings: List[Tuple[int, str, str, str]] = []
|
||
for rev_map in sorted(svn_dir.rglob(".rev_map.*")):
|
||
if not rev_map.is_file():
|
||
continue
|
||
|
||
# Relative path from .git/svn/ gives us the ref, e.g.:
|
||
# .git/svn/refs/remotes/origin/trunk/.rev_map.<UUID>
|
||
# → ref = refs/remotes/origin/trunk
|
||
ref_path = rev_map.parent.relative_to(svn_dir)
|
||
git_ref = str(ref_path).replace("\\", "/")
|
||
svn_branch = self._ref_to_svn_branch(git_ref)
|
||
|
||
data = rev_map.read_bytes()
|
||
size = 24 # 4 bytes revision + 20 bytes SHA-1
|
||
zero_hash = "0000000000000000000000000000000000000000"
|
||
entries = 0
|
||
for i in range(0, len(data) - size + 1, size):
|
||
rev_raw = data[i:i+4]
|
||
sha_raw = data[i+4:i+24]
|
||
if len(rev_raw) < 4 or len(sha_raw) < 20:
|
||
break
|
||
rev = int.from_bytes(rev_raw, "big")
|
||
sha = sha_raw.hex()
|
||
# git svn sometimes pads the file with a zero-hash record;
|
||
# skip those.
|
||
if sha == zero_hash:
|
||
logger.debug(
|
||
"Skipping zero-hash rev_map entry at offset %d", i
|
||
)
|
||
continue
|
||
mappings.append((rev, sha, svn_branch, git_ref))
|
||
entries += 1
|
||
|
||
logger.debug(
|
||
"Read %d entries from %s (branch: %s)",
|
||
entries, rev_map.name, svn_branch,
|
||
)
|
||
|
||
return mappings
|
||
|
||
def _ref_to_svn_branch(self, git_ref: str) -> str:
|
||
"""Convert a git svn remote ref path to an SVN branch name.
|
||
|
||
Input examples:
|
||
refs/remotes/origin/trunk → trunk
|
||
refs/remotes/origin/branches/feat → branches/feat
|
||
refs/remotes/origin/tags/v1.0 → tags/v1.0
|
||
"""
|
||
prefix = "refs/remotes/origin/"
|
||
if git_ref.startswith(prefix):
|
||
return git_ref[len(prefix):]
|
||
return git_ref
|
||
|
||
def _write_mappings_to_db(self, mappings: List[Tuple[int, str, str, str]]):
|
||
"""Bulk-insert parsed mappings into SQLite."""
|
||
logger.info("Writing %d mapping entries to DB…", len(mappings))
|
||
for rev, sha, branch, ref in mappings:
|
||
self.db.record_mapping(rev, sha, branch, ref, source="svn")
|
||
logger.info("Mapping DB now has %d entries", self.db.mapping_count())
|
||
|
||
def _finalize_import_state(self):
|
||
"""Update state.json + DB after successful import."""
|
||
last_rev = self.db.get_max_revision_across_all_branches()
|
||
|
||
self._write_state({
|
||
"initial_import_done": True,
|
||
"last_svn_revision": last_rev,
|
||
"last_git_commit": None,
|
||
"created_at": self._read_state().get("created_at", _now_iso()),
|
||
"last_sync_at": _now_iso(),
|
||
})
|
||
|
||
self.db.set_state("last_svn_revision", str(last_rev))
|
||
self.db.set_state("last_git_commit", "")
|
||
logger.info(
|
||
"Initial import complete for %s (last SVN rev: %s)",
|
||
self.config.id, last_rev,
|
||
)
|
||
|
||
# ─── File I/O ─────────────────────────────────────────────
|
||
|
||
def _write_authors_file(self):
|
||
"""Write authors.txt for git svn including ALL SVN authors.
|
||
|
||
Scans the SVN log for all unique authors, then merges with
|
||
configured mappings. Any author without a configured mapping
|
||
defaults to ``user <user@svn>`` so that git svn does not abort.
|
||
"""
|
||
all_authors = dict(self.config.authors)
|
||
|
||
# Discover all SVN authors from the remote repo
|
||
try:
|
||
r = subprocess.run(
|
||
["svn", "log", "-q", "--xml", self.config.svn.url],
|
||
capture_output=True, text=True, check=True, timeout=300,
|
||
)
|
||
import xml.etree.ElementTree as ET
|
||
root = ET.fromstring(r.stdout)
|
||
for entry in root.findall(".//logentry"):
|
||
el = entry.find("author")
|
||
if el is not None and el.text:
|
||
svn_user = el.text.strip()
|
||
if svn_user and svn_user not in all_authors:
|
||
all_authors[svn_user] = f"{svn_user} <{svn_user}@svn>"
|
||
except Exception:
|
||
logger.warning(
|
||
"Could not discover SVN authors (svn log failed); "
|
||
"using configured authors only"
|
||
)
|
||
|
||
with open(self.authors_path, "w") as f:
|
||
for svn_user in sorted(all_authors):
|
||
f.write(f"{svn_user} = {all_authors[svn_user]}\n")
|
||
|
||
def _read_state(self) -> Dict[str, Any]:
|
||
if not self.state_path.exists():
|
||
return {}
|
||
with open(self.state_path) as f:
|
||
return json.load(f)
|
||
|
||
def _write_state(self, state: Dict[str, Any]):
|
||
with open(self.state_path, "w") as f:
|
||
json.dump(state, f, indent=2, default=str)
|
||
|
||
# ─── Git subprocess helper ────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _git(*args: str, check: bool = True) -> subprocess.CompletedProcess:
|
||
try:
|
||
r = subprocess.run(
|
||
["git", *args],
|
||
capture_output=True, text=True, timeout=600,
|
||
)
|
||
except FileNotFoundError:
|
||
raise MirrorError("git not found on PATH")
|
||
except subprocess.TimeoutExpired:
|
||
raise MirrorError("git command timed out: git " + " ".join(args))
|
||
|
||
if check and r.returncode != 0:
|
||
raise MirrorError(
|
||
f"git command failed (exit {r.returncode}): "
|
||
f"git {' '.join(args)}\n{r.stderr.strip()}"
|
||
)
|
||
return r
|
||
|
||
|
||
def _now_iso() -> str:
|
||
from datetime import datetime, timezone
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|