host
42b295fc71
tmpfiles 行 d /run/chv/syncd 0755 chv chv 没用：tmpfiles-setup 跑在 cloud-init 建 chv 用户之前，属主解析失败整行被跳过， 客内 /run/chv/syncd 根本不存在（2026-11-07 真机实测：镜像 里有这行、目录没有，syncd 反复退出、restart counter 57）。 改走单元 RuntimeDirectory=chv-syncd（服务启动时由 systemd 按服务用户建目录，不依赖启动时序）；状态写失败改为记日志 不杀进程；daemon 与驱动的 syncd 路径同步改到 /run/chv-syncd/。
287 lines
10 KiB
Python
287 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""chv-syncd：客机运行时 sync 常驻（阶段 2 三分之一）。
|
||
|
||
职责：常驻循环调用 /usr/local/bin/chv-sync 做 Gitea 双向同步，
|
||
状态落盘 /run/chv/sync-status.json，socket /run/chv/syncd.sock
|
||
接受 ping/status/sync-now。
|
||
|
||
同步规则一条不改（与 chv-sync.sh 同款，复用它不重写 git 逻辑）：
|
||
- Gitea https://gitea.tose.top 是权威远端；启动 clone 选定仓
|
||
最新 HEAD（仍由 cloud-init runcmd 的 chv-sync boot 做；
|
||
syncd 只在仓已 clone 的前提下循环，仓不存在记 needs-boot
|
||
不擅自 clone——boot 须 root 写 credential，syncd 以 chv
|
||
身份跑，无权也不该碰）。
|
||
- 每轮先 pull --ff-only 再 push；pull 分叉即停住、跳过 push，
|
||
状态记 error、绝不 --force、不自动 rebase、不覆盖任何一方。
|
||
- 工作树脏时 pull 拒绝，同样记 error 等下一轮，不丢改动。
|
||
- push 只推已提交的提交；未提交改动不参与同步。
|
||
- token 只走 /home/chv/.git-credentials（0600），syncd
|
||
自身不读 guest-env、不打印、不记入状态文件与日志。
|
||
|
||
以 chv 身份运行。默认每 300 秒一轮（CHV_SYNC_INTERVAL，
|
||
下限 60 秒）；sync-now 立即触发一轮（轮内重复触发跳过）。
|
||
"""
|
||
import json
|
||
import os
|
||
import signal
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
# systemd RuntimeDirectory（见 build.sh 的 syncd 单元）：服务启动
|
||
# 时由 systemd 按服务用户（chv）建 /run/chv-syncd，启动时序
|
||
# 不依赖 cloud-init 先建好 chv 用户。不要放回 /run/chv/ 下——
|
||
# tmpfiles-setup 跑在 cloud-init 建 chv 用户之前，带 chv 属主
|
||
# 的 tmpfiles 行会被跳过（2026-11-07 真机实测：目录根本没建，
|
||
# syncd 反复退出、长期 activating）。
|
||
SYNC_SOCK = "/run/chv-syncd/syncd.sock"
|
||
STATUS_PATH = "/run/chv-syncd/sync-status.json"
|
||
WORKSPACE = "/home/chv/workspace"
|
||
SYNC_BIN = "/usr/local/bin/chv-sync"
|
||
DEFAULT_INTERVAL = 300.0
|
||
MIN_INTERVAL = 60.0
|
||
MAX_REQUEST_BYTES = 65536
|
||
|
||
|
||
def _atomic_write(path, data: bytes):
|
||
tmp = "%s.tmp.%d" % (path, os.getpid())
|
||
with open(tmp, "wb") as fh:
|
||
fh.write(data)
|
||
fh.flush()
|
||
os.fsync(fh.fileno())
|
||
os.rename(tmp, path)
|
||
|
||
|
||
def _read_json(path):
|
||
try:
|
||
with open(path) as fh:
|
||
return json.load(fh)
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
def resolve_repo(workspace=WORKSPACE):
|
||
"""WORKSPACE 下唯一的 git 仓库；多个/零个返回 (None, reason)。"""
|
||
try:
|
||
names = sorted(os.listdir(workspace))
|
||
except OSError as exc:
|
||
return None, "workspace unreadable: %s" % exc
|
||
found = [os.path.join(workspace, name) for name in names
|
||
if os.path.isdir(os.path.join(workspace, name, ".git"))]
|
||
if len(found) == 1:
|
||
return found[0], None
|
||
if not found:
|
||
return None, "needs-boot: no cloned repo in workspace"
|
||
return None, "ambiguous: %d repos, explicit repo required" % len(found)
|
||
|
||
|
||
def _run_sync(action, repo):
|
||
"""跑 chv-sync pull/push；返回 (ok, tail)。tail 只取末行
|
||
（git 报错不含 token：remote URL 是干净 https，credential
|
||
走 store 文件），截断 500 字符。"""
|
||
try:
|
||
proc = subprocess.run(
|
||
[SYNC_BIN, action, repo], capture_output=True,
|
||
text=True, errors="replace", timeout=300)
|
||
except OSError as exc:
|
||
return False, "sync binary failed: %s" % exc
|
||
except subprocess.TimeoutExpired:
|
||
return False, "sync %s timeout after 300s" % action
|
||
tail = ((proc.stdout or "") + (proc.stderr or "")).strip()
|
||
tail = tail.splitlines()[-1] if tail else ""
|
||
return proc.returncode == 0, tail[:500]
|
||
|
||
|
||
def _git_head(repo):
|
||
try:
|
||
proc = subprocess.run(
|
||
["git", "-C", repo, "rev-parse", "HEAD"],
|
||
capture_output=True, text=True, errors="replace",
|
||
timeout=30)
|
||
except (OSError, subprocess.TimeoutExpired):
|
||
return None
|
||
if proc.returncode != 0:
|
||
return None
|
||
return proc.stdout.strip()
|
||
|
||
|
||
class Syncd:
|
||
def __init__(self, interval=DEFAULT_INTERVAL,
|
||
status_path=STATUS_PATH, workspace=WORKSPACE):
|
||
self.interval = max(interval, MIN_INTERVAL)
|
||
self.status_path = status_path
|
||
self.workspace = workspace
|
||
self._stop = threading.Event()
|
||
self._cycle_lock = threading.Lock()
|
||
self._cycle_running = False
|
||
self._status = _read_json(status_path) or {
|
||
"service": "chv-syncd", "last": "never",
|
||
"at": None, "head": None, "error": None,
|
||
"consecutive_errors": 0, "cycles": 0}
|
||
|
||
def _write_status(self):
|
||
try:
|
||
_atomic_write(self.status_path,
|
||
(json.dumps(self._status) + "\n").encode())
|
||
except OSError as exc:
|
||
# 状态文件是可观测性，不是同步本身：写失败记一
|
||
# 行日志、不杀进程（2026-11-07 真机教训：写失败
|
||
# 抛到主线程致 syncd 反复退出、长期 activating）。
|
||
print("chv-syncd: status write failed: %s" % exc,
|
||
file=sys.stderr, flush=True)
|
||
|
||
def run_cycle(self):
|
||
"""一轮同步：先 pull 再 push；任一步失败即停住记 error。"""
|
||
if not self._cycle_lock.acquire(blocking=False):
|
||
return {"ok": False, "error": "cycle already running"}
|
||
try:
|
||
self._cycle_running = True
|
||
return self._cycle()
|
||
finally:
|
||
self._cycle_running = False
|
||
self._cycle_lock.release()
|
||
|
||
def _cycle(self):
|
||
repo, reason = resolve_repo(self.workspace)
|
||
now = time.time()
|
||
self._status["cycles"] = self._status.get("cycles", 0) + 1
|
||
if repo is None:
|
||
self._status.update(
|
||
{"last": "error", "at": now, "error": reason,
|
||
"consecutive_errors":
|
||
self._status.get("consecutive_errors", 0) + 1})
|
||
self._write_status()
|
||
return {"ok": False, "error": reason}
|
||
head_before = _git_head(repo)
|
||
ok, tail = _run_sync("pull", repo)
|
||
if not ok:
|
||
self._status.update(
|
||
{"last": "error", "at": now, "repo": repo,
|
||
"head": head_before,
|
||
"error": "pull failed: %s" % tail,
|
||
"consecutive_errors":
|
||
self._status.get("consecutive_errors", 0) + 1})
|
||
self._write_status()
|
||
return {"ok": False, "error": "pull failed: %s" % tail}
|
||
ok, tail = _run_sync("push", repo)
|
||
head_after = _git_head(repo)
|
||
if not ok:
|
||
self._status.update(
|
||
{"last": "error", "at": now, "repo": repo,
|
||
"head": head_after,
|
||
"error": "push failed: %s" % tail,
|
||
"consecutive_errors":
|
||
self._status.get("consecutive_errors", 0) + 1})
|
||
self._write_status()
|
||
return {"ok": False, "error": "push failed: %s" % tail}
|
||
self._status.update(
|
||
{"last": "ok", "at": now, "repo": repo,
|
||
"head": head_after, "head_before": head_before,
|
||
"error": None, "consecutive_errors": 0})
|
||
self._write_status()
|
||
return {"ok": True, "repo": repo, "head": head_after}
|
||
|
||
def handle(self, req):
|
||
if not isinstance(req, dict):
|
||
return {"ok": False, "error": "request must be an object"}
|
||
op = req.get("op")
|
||
if op == "ping":
|
||
return {"ok": True, "service": "chv-syncd",
|
||
"pid": os.getpid()}
|
||
if op == "status":
|
||
return {"ok": True, "status": self._status}
|
||
if op == "sync-now":
|
||
return self.run_cycle()
|
||
return {"ok": False, "error": "unknown op: %s" % (op,)}
|
||
|
||
def _serve(self):
|
||
if os.path.exists(SYNC_SOCK):
|
||
os.remove(SYNC_SOCK)
|
||
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||
srv.bind(SYNC_SOCK)
|
||
os.chmod(SYNC_SOCK, 0o600)
|
||
srv.listen(8)
|
||
srv.settimeout(0.5)
|
||
try:
|
||
while not self._stop.is_set():
|
||
try:
|
||
conn, _ = srv.accept()
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
break
|
||
with conn:
|
||
try:
|
||
conn.settimeout(10.0)
|
||
buf = b""
|
||
while len(buf) < MAX_REQUEST_BYTES:
|
||
chunk = conn.recv(4096)
|
||
if not chunk:
|
||
break
|
||
buf += chunk
|
||
if buf.endswith(b"\n"):
|
||
break
|
||
req = json.loads(buf.decode())
|
||
resp = self.handle(req)
|
||
except (OSError, ValueError) as exc:
|
||
resp = {"ok": False, "error": str(exc)}
|
||
try:
|
||
conn.sendall((json.dumps(resp) + "\n").encode())
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
try:
|
||
srv.close()
|
||
except OSError:
|
||
pass
|
||
try:
|
||
os.remove(SYNC_SOCK)
|
||
except OSError:
|
||
pass
|
||
|
||
def run(self):
|
||
thread = threading.Thread(target=self._serve, daemon=True)
|
||
thread.start()
|
||
print("chv-syncd pid=%d interval=%.0fs status=%s"
|
||
% (os.getpid(), self.interval, self.status_path),
|
||
flush=True)
|
||
try:
|
||
while not self._stop.is_set():
|
||
self.run_cycle()
|
||
self._stop.wait(self.interval)
|
||
finally:
|
||
self._stop.set()
|
||
thread.join(timeout=5)
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
|
||
|
||
def main(argv=None):
|
||
argv = list(sys.argv[1:] if argv is None else argv)
|
||
interval = float(os.environ.get("CHV_SYNC_INTERVAL")
|
||
or DEFAULT_INTERVAL)
|
||
i = 0
|
||
while i < len(argv):
|
||
if argv[i] == "--interval" and i + 1 < len(argv):
|
||
interval = float(argv[i + 1])
|
||
i += 2
|
||
else:
|
||
print("用法：chv-syncd [--interval 秒]", file=sys.stderr)
|
||
return 2
|
||
syncd = Syncd(interval=interval)
|
||
|
||
def _on_signal(signum, _frame):
|
||
syncd.stop()
|
||
|
||
signal.signal(signal.SIGTERM, _on_signal)
|
||
signal.signal(signal.SIGINT, _on_signal)
|
||
syncd.run()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|