commit a006483bbc26b9dd210fa8e5c8c244e79fe12aea Author: iceBear67 Date: Fri Aug 14 07:13:22 2026 +0000 init diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..981917c --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +.git +.github +.gitignore +*.md +!README.md +config.toml +data/ +docker-compose.yml +Dockerfile +.dockerignore diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6608a11 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,46 @@ +name: ci + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version: "1.24" + cache: true + + - name: Check formatting + run: | + unformatted=$(gofmt -l .) + if [ -n "$unformatted" ]; then + echo "::error::gofmt needed for: $unformatted" + gofmt -d . + exit 1 + fi + + - name: Vet + run: go vet ./... + + # The syncer tests drive a real git binary against temporary repositories. + - name: Test + run: go test -race -count=1 ./... + + - name: Build + run: go build ./... + + - name: Validate the example config + env: + SRC_TOKEN: dummy-token-for-validation + run: go run . -config config.example.toml -check diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml new file mode 100644 index 0000000..38e59d3 --- /dev/null +++ b/.github/workflows/docker.yml @@ -0,0 +1,77 @@ +name: docker + +on: + push: + branches: [main] + tags: ["v*"] + pull_request: + workflow_dispatch: + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +concurrency: + group: docker-${{ github.ref }} + cancel-in-progress: true + +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - uses: actions/checkout@v4 + + # QEMU lets one runner produce the arm64 image alongside amd64. + - uses: docker/setup-qemu-action@v3 + - uses: docker/setup-buildx-action@v3 + + # Pull requests build but never publish: forks must not be able to push. + - name: Log in to ${{ env.REGISTRY }} + if: github.event_name != 'pull_request' + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Derive tags and labels + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=semver,pattern={{version}} + type=semver,pattern={{major}}.{{minor}} + type=semver,pattern={{major}} + type=ref,event=branch + type=ref,event=pr + type=sha,format=short + type=raw,value=latest,enable={{is_default_branch}} + + - name: Build and push + uses: docker/build-push-action@v6 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: ${{ github.event_name != 'pull_request' }} + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + build-args: | + VERSION=${{ steps.meta.outputs.version }} + cache-from: type=gha + cache-to: type=gha,mode=max + + - name: Summary + if: github.event_name != 'pull_request' + run: | + { + echo "### Published" + echo + echo '```' + echo "${{ steps.meta.outputs.tags }}" + echo '```' + } >> "$GITHUB_STEP_SUMMARY" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..463e4d1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +# Local runtime state and secrets — never commit these. +/config.toml +/data/ +/keys/ +/syncbot +*.pem +id_ed25519* +id_rsa* diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..e21a2a4 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,146 @@ +# AGENTS.md + +Notes for anyone (human or agent) changing this repository. +User-facing docs live in [README.md](README.md); this file is about the code. + +## What this is + +A daemon that mirrors git repositories from `src` to `dst` on a timer, driven by a +hot-reloadable TOML file. Go 1.24, one third-party dependency +(`github.com/BurntSushi/toml`), ~1500 lines of non-test code. + +Design goals, in priority order: **do not corrupt or delete anything on dst**, +stay small in memory, stay simple. When a change trades any of these for +convenience, it is probably the wrong change. + +## Commands + +```bash +go build ./... +go test ./... # includes end-to-end tests that shell out to real git +go test -race -count=1 ./... # what CI runs +go vet ./... +gofmt -l . # must print nothing; CI fails otherwise +go run . -config config.example.toml -check # needs SRC_TOKEN set to anything +``` + +The tests need a `git` binary on `PATH`. They create real repositories under +`t.TempDir()` and never touch the network. + +## Layout + +``` +main.go flags, logger, signal handling, config file watcher +signal_unix.go/_other.go build-tagged SIGHUP wiring (no-op on non-unix) +internal/config/ TOML parsing, validation, defaulting → []Job +internal/gitx/ hermetic wrapper around the git CLI + proc_unix.go/_other.go build-tagged process-group creation and kill +internal/syncer/ one sync cycle for one repository +internal/manager/ per-repo goroutines, reload diffing, backoff + http.go /healthz /readyz /status /metrics +``` + +Dependency direction is strictly `main → manager → syncer → gitx → config`. +Nothing under `internal/` imports `manager`. + +## Invariants + +These are load-bearing. Breaking one is a bug even if the tests still pass. + +1. **`config.Job` is fully resolved.** All defaulting, env expansion and + validation happen in `config.resolve`. Downstream code never asks "was this + field set?" — it reads the value. This is also what makes reload diffing work: + `Apply` decides whether a repo changed with `reflect.DeepEqual` on `Job`, so a + `Job` must contain no pointers, maps, funcs, or timestamps, and must be + deterministic for identical input. + +2. **Repository URLs never reach disk.** They are passed as command-line + arguments to `git fetch` / `git push` / `git ls-remote`, never written into + `.git/config` via `git remote add`. URLs may embed tokens; mirror directories + may outlive the process. + +3. **Secrets are scrubbed from every string that escapes.** `gitx.Run` runs both + its log lines and git's output through `scrub` using `Options.Secrets`, and + URLs through `RedactURL`. Any new code path that logs or wraps git output must + do the same. There is a test asserting a token never appears in an error. + +4. **Every git invocation goes through `gitx.Run`.** It supplies the isolated + `HOME`, disables system/global gitconfig, drops inherited `GIT_SSH_COMMAND` + and friends, sets `GIT_TERMINAL_PROMPT=0`, and puts the child in its own + process group so a timeout kills `ssh` too. Calling `exec.Command("git", ...)` + directly anywhere else reintroduces all of those problems. + +5. **The syncer holds no state between cycles.** Every cycle re-derives truth + from `ls-remote` on both ends. Do not add a state file, a "last synced SHA" + cache, or an in-memory `map[repo]refs` shortcut — the self-healing behaviour + (recovering from external force-pushes and partial pushes) comes entirely from + not trusting anything remembered. + +6. **An empty source never empties a destination** unless `allow_empty = true`. + The guard is in `syncer.Sync`; keep it before the push, not inside it. + +7. **One goroutine per mirror directory, ever.** `Apply` cancels the goroutine + being replaced and passes its `done` channel to the successor as `waitFor`, so + the successor blocks until the predecessor has exited. Two goroutines running + `git` in the same bare repo will corrupt it. + +8. **`Apply` must not block.** It is called from the config watcher; a repo may + be 25 minutes into a 30-minute sync. Hence the `waitFor` handoff above rather + than waiting inline. + +## Gotchas discovered the hard way + +- **`limitedWriter.Write` must return `len(p)`**, not the number of bytes it + chose to keep. `os/exec` treats a short write as an error and would abort an + otherwise healthy git run once output exceeded the cap. + +- **`signal.Notify` with an empty slice subscribes to every signal.** The + `reloadSignals()` call site is guarded with a length check for the non-unix + build where the slice is empty. + +- **Git refuses to delete the branch that HEAD points at** (`receive.denyDeleteCurrent`). + This shows up when pruning a renamed default branch; GitHub behaves the same + way. It is server-side behaviour, not something to work around in the client. + The syncer test fixture sets `receive.denyDeleteCurrent ignore` on its bare dst + for exactly this reason. Documented in the README's troubleshooting section. + +- **`url.UserPassword` percent-encodes.** `RedactURL` uses the literal string + `redacted` as the placeholder; `***` came back as `%2A%2A%2A`. + +- **Deploy keys mounted read-only are usually 0644** and ssh rejects them. + `gitx.usableKey` stages a 0600 copy in the per-sync temp dir. Do not "fix" this + by chmod'ing the original — the operator often cannot make it writable. + +- **`IdentitiesOnly=yes` is required.** Without it ssh offers every key the agent + knows about and GitHub closes the connection with "too many authentication + failures" before reaching the right one. + +- **`expandEnv` deliberately only understands `${VAR}`.** Bare `$VAR` is left + alone because `$` is common in passwords, and an undefined variable is a hard + error rather than an empty expansion, which would otherwise produce a URL that + fails in a confusing way much later. + +## Testing conventions + +`internal/syncer/syncer_test.go` has a harness that builds real source and bare +destination repositories and drives `Sync` against them. New sync behaviour +belongs there rather than in a mock — the interesting bugs in this program are +all in how git actually behaves. Existing cases cover first sync, no-op cycles, +new commits, prune, force-push after a rewrite, repairing drift on dst, the +empty-source refusal, ref filtering, unreachable sources, and timeouts. + +Manager tests assert reload semantics: unchanged repos keep running, changed +repos restart, removed repos stop. They run under `-race`; the `wg` in `Manager` +exists because a goroutine replaced by a reload could otherwise outlive `Stop` +and race `t.TempDir()` cleanup. + +## When adding a config field + +1. Add it to `Repo` (and `Global` if it should be inheritable) in `config.go`. +2. Resolve it in `resolveJob` — with an explicit default, never a zero value that + downstream code has to interpret. +3. Add it to `Job`, keeping the type comparable (see invariant 1). +4. Thread it through `syncer.Sync` / `gitx`. +5. Document it in the README table and, if it is interesting, `config.example.toml`. +6. Add a case to `config_test.go`. `unknownKeys` rejects unrecognised keys, so a + field that is parsed but not registered will make previously valid configs fail. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..47dc3e3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..422cf36 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,36 @@ +FROM golang:1.24-alpine AS build + +WORKDIR /src + +# Dependencies first: this layer is reused whenever only source files change. +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +ARG VERSION=dev +RUN CGO_ENABLED=0 go build \ + -trimpath \ + -ldflags="-s -w -X main.version=${VERSION}" \ + -o /out/syncbot . + + +FROM alpine:3.22 + +LABEL org.opencontainers.image.title="syncbot" \ + org.opencontainers.image.description="Mirrors git repositories from a source to a destination on a timer." \ + org.opencontainers.image.licenses="MIT" + +# git and ssh do the actual transfer work; syncbot only orchestrates them. +RUN apk add --no-cache git openssh-client ca-certificates tzdata \ + && adduser -D -H -u 10001 syncbot \ + && mkdir -p /var/lib/syncbot /etc/syncbot \ + && chown -R syncbot:syncbot /var/lib/syncbot + +COPY --from=build /out/syncbot /usr/local/bin/syncbot + +USER syncbot + +# Mount the config read-only at this path, or override it with -config. +ENTRYPOINT ["/usr/local/bin/syncbot"] +CMD ["-config", "/etc/syncbot/config.toml"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..b8f8aa7 --- /dev/null +++ b/README.md @@ -0,0 +1,280 @@ +# syncbot + +定时把一个 git 仓库（src）镜像推送到另一个 git 仓库（dst）的小机器人。 +典型用途：把内网 GitLab / Gitea 上的仓库持续同步到 GitHub，dst 侧用一把 deploy key 授权。 + +- **单文件、低占用** — 静态二进制 6.4 MB，空转常驻内存约 11 MB，同步时的重活交给短命的 `git` 子进程 +- **配置热更新** — 改完 `config.toml` 直接保存，几秒内生效；只有被改动的仓库会重启，其它仓库的计时器和进行中的同步不受影响 +- **无状态、自愈** — 不存任何同步进度。每轮都直接问 src 和 dst 各自有哪些引用，只做必要的操作。有人手贱 force push 了 dst、上一次推送推到一半失败、容器被重建 —— 下一轮自动收敛 +- **省流量** — 每轮先用 `ls-remote` 探一下，src 没动就不 fetch，dst 已经一致就不 push +- **不会误删** — src 突然变空（URL 写错、token 过期长这样）时拒绝把 dst 清空 +- **凭据不落盘** — 带 token 的 URL 只在命令行里传，不写进 `.git/config`；日志和报错里的密码一律脱敏 + +--- + +## 快速开始 + +### Docker Compose（推荐） + +```bash +mkdir -p syncbot/keys && cd syncbot +curl -O https://raw.githubusercontent.com/OWNER/syncbot/main/docker-compose.yml +curl -o config.toml https://raw.githubusercontent.com/OWNER/syncbot/main/config.example.toml +# 编辑 docker-compose.yml 里的 image 和 config.toml 里的仓库地址 +docker compose up -d +docker compose logs -f +``` + +### 直接跑二进制 + +```bash +go build -o syncbot . +./syncbot -config ./config.toml -check # 先校验配置 +./syncbot -config ./config.toml +``` + +### 最小配置 + +```toml +[global] +work_dir = "/var/lib/syncbot" + +[[repo]] +name = "my-project" +src = "https://gitlab.internal/team/my-project.git" +dst = "git@github.com:me/my-project.git" +interval = "5m" +ssh_key = "/etc/syncbot/keys/my-project" +``` + +完整示例见 [`config.example.toml`](config.example.toml)。 + +--- + +## 给 bot 配 GitHub deploy key + +```bash +# 1. 生成一把不带密码的 key（bot 无人值守，不能有 passphrase） +ssh-keygen -t ed25519 -N "" -C "syncbot" -f keys/my-project + +# 2. 把公钥加到 GitHub：仓库 → Settings → Deploy keys → Add deploy key +cat keys/my-project.pub +# ⚠️ 必须勾选 "Allow write access"，否则只能读不能推 + +# 3. 配置里指向私钥 +# ssh_key = "/etc/syncbot/keys/my-project" +``` + +几个坑： + +- **一把 deploy key 只能绑一个仓库。** GitHub 不允许同一把公钥重复添加到多个仓库（会报 key is already in use）。要镜像多个仓库就每个仓库生成一把，或者改用 machine user / GitHub App token 走 HTTPS。 +- **私钥权限不用操心。** 从 Kubernetes Secret 或 `:ro` 挂载进来的 key 常常是 0644，`ssh` 会直接拒绝。syncbot 会自动在私有临时目录做一份 0600 的副本再用，任务结束即删。 +- **主机密钥校验。** 默认 `accept-new`（首次连接自动信任并记到 `work_dir/home/known_hosts`）。想更严格就先固定下来： + + ```bash + ssh-keyscan github.com > keys/known_hosts + ``` + + ```toml + known_hosts = "/etc/syncbot/keys/known_hosts" + strict_host_key = "yes" + ``` + +--- + +## 配置 + +### `[global]` + +| 字段 | 默认值 | 说明 | +| --- | --- | --- | +| `work_dir` | `/var/lib/syncbot` | 本地镜像仓库和 ssh 状态的存放目录，需要可写 | +| `interval` | `5m` | 默认轮询间隔 | +| `timeout` | `30m` | 单次同步的超时 | +| `concurrency` | `min(4, CPU 核数)` | 同时运行的 git 进程数上限，**内存主要靠它控制** | +| `max_backoff` | `1h` | 连续失败后的退避上限 | +| `reload_interval` | `5s` | 多久检查一次配置文件有没有变 | +| `listen` | 空（不监听） | HTTP 端点地址，如 `:8080` | +| `log_level` | `info` | `debug` / `info` / `warn` / `error` | +| `log_format` | `text` | `text` / `json` | + +`[global]` 里还可以写下面所有 `[[repo]]` 字段作为默认值。 + +### `[[repo]]` + +| 字段 | 默认值 | 说明 | +| --- | --- | --- | +| `name` | 必填 | 唯一标识，同时用作目录名和监控标签，限 `[A-Za-z0-9._-]` | +| `src` | 必填 | 源仓库，字符串或表（见下） | +| `dst` | 必填 | 目标仓库 | +| `enabled` | `true` | 设为 `false` 可临时停掉而不用删配置 | +| `interval` / `timeout` / `max_backoff` | 继承 global | | +| `refs` | `["refs/heads/*", "refs/tags/*"]` | 镜像哪些引用，每条最多一个 `*` | +| `prune` | `true` | src 上删掉的分支/标签，也在 dst 上删掉 | +| `force` | `true` | 允许非快进推送。上游 rebase 过就必须开 | +| `atomic` | `false` | `true` 表示所有引用要么全成功要么全不动 | +| `allow_empty` | `false` | 允许「空的 src 清空 dst」，见下方保护机制 | +| `ssh_key` | 无 | 私钥绝对路径，同时作用于 src 和 dst | +| `known_hosts` | 无 | 固定的 known_hosts 文件 | +| `strict_host_key` | 有 `known_hosts` 时 `yes`，否则 `accept-new` | `yes` / `no` / `accept-new` | +| `git_config` | 无 | 原样传给 git 的 `-c key=value`，用于压内存等 | + +### 端点写法 + +src / dst 可以直接写成字符串： + +```toml +src = "https://gitlab.internal/team/repo.git" +dst = "git@github.com:me/repo.git" +``` + +两边需要各自的凭据时写成表： + +```toml +[repo.src] +url = "git@gitlab.internal:team/repo.git" +ssh_key = "/etc/syncbot/keys/gitlab" + +[repo.dst] +url = "git@github.com:me/repo.git" +ssh_key = "/etc/syncbot/keys/github" +known_hosts = "/etc/syncbot/keys/known_hosts" +strict_host_key = "yes" +``` + +### 环境变量 + +URL 和路径里的 `${VAR}` 会用环境变量展开，适合放 token： + +```toml +src = "https://x-access-token:${SRC_TOKEN}@github.com/upstream/repo.git" +``` + +- 变量没定义会**直接启动失败并指出名字**，不会静默展开成空串给你一个坏掉的 URL +- 只认 `${VAR}`，裸写的 `$VAR` 原样保留（密码里 `$` 很常见） +- 环境变量在进程启动时读取，改环境变量需要重启 + +--- + +## 热更新 + +syncbot 每 `reload_interval` 比对一次配置文件内容的哈希，变了就重新加载；也可以 `kill -HUP` 立即触发。 + +重新加载时会把新配置和正在跑的任务做 diff： + +- 新增的仓库 → 启动 +- 删掉或 `enabled = false` 的仓库 → 停掉，其正在进行的 git 操作会被取消 +- 配置有改动的仓库 → 重启（新任务会等旧任务完全退出，避免两个进程同时动同一个镜像目录） +- **没改动的仓库 → 完全不动**，计时器和进行中的同步都不受影响 + +几点说明： + +- **配置写错了不会挂。** 解析失败会记一条 ERROR，然后继续用上一份能跑的配置。修好保存即可恢复 +- `log_level` 改了立即生效 +- `listen` 改了需要重启进程，日志里会 WARN 提示 +- 累计的 `syncs` / `pushes` 计数在仓库重启后保留，只有仓库被删掉才清零 + +--- + +## 运维 + +### 命令行 + +``` +syncbot -config PATH 配置文件路径（默认 /etc/syncbot/config.toml） + -check 校验配置并打印解析结果后退出 + -once 所有仓库同步一次就退出（适合 cron / CI） + -version 打印版本 +``` + +### 信号 + +| 信号 | 行为 | +| --- | --- | +| `SIGHUP` | 立即重新加载配置 | +| `SIGTERM` / `SIGINT` | 优雅退出，最多等 20 秒让进行中的同步收尾 | + +### HTTP 端点（需要设置 `listen`） + +| 路径 | 说明 | +| --- | --- | +| `/healthz` | 进程活着就返回 200，适合容器 healthcheck | +| `/readyz` | 所有仓库都至少成功同步过一次才返回 200，否则 503 | +| `/status` | JSON，每个仓库的最后运行时间、失败次数、引用数、最后一次错误 | +| `/metrics` | Prometheus 文本格式 | + +指标：`syncbot_build_info`、`syncbot_uptime_seconds`、`syncbot_sync_total`、 +`syncbot_push_total`、`syncbot_consecutive_failures`、`syncbot_refs`、 +`syncbot_last_success_timestamp_seconds`、`syncbot_last_duration_seconds`。 + +告警建议盯 `syncbot_consecutive_failures > 0`，或者 +`time() - syncbot_last_success_timestamp_seconds` 超过 interval 的若干倍。 + +--- + +## 工作原理 + +每个仓库一个 goroutine、一个独立计时器，所以某个仓库卡住不会拖累别的。 +一个全局信号量限制同时运行的 git 进程数（`concurrency`）。 + +单轮同步： + +1. `git ls-remote` 问 src 现在有哪些引用 +2. 和本地镜像（`work_dir/mirrors/.git`）比对 → 一样就**跳过 fetch** +3. 不一样才 `git fetch --prune` +4. `git ls-remote` 问 dst 现在有哪些引用 +5. 和本地镜像比对 → 一样就**跳过 push** +6. 不一样才 `git push --prune --force` + +因为每轮都重新问过两边，所以不需要任何本地状态文件，删掉 `work_dir` 也只是让下次重新克隆一遍而已。 + +失败时按 `interval → 2×→ 4× …` 退避，上限 `max_backoff`；成功后立刻恢复正常节奏。 + +所有 git 调用都在独立进程组里，超时或退出时整组一起收掉，不会漏下 `ssh` 之类的子进程。 +每次调用都用隔离的 `HOME` 并禁用系统/全局 gitconfig，行为不受宿主机环境影响。 + +--- + +## 常见问题 + +**推送被拒：`refusing to delete the current branch`** +dst 的默认分支（HEAD 指向的那个）不允许被删除，GitHub 也一样。 +上游把 `master` 改名成 `main` 之后会撞上这个：先去 GitHub 仓库设置里把默认分支改成新名字，再让 syncbot 同步。 + +**日志出现 `refusing to mirror an empty source`** +src 一个引用都没返回，但 dst 有内容。绝大多数情况是 URL 写错或者 token 过期， +所以默认拒绝推送以免把 dst 清空。确实想清空的话给那个仓库加 `allow_empty = true`。 + +**`Permission denied (publickey)`** +依次检查：deploy key 是否勾了 *Allow write access*；配置里 `ssh_key` 是否指向**私钥**（不是 `.pub`）； +key 是否没有 passphrase；同一把 key 是否被重复用在了多个 GitHub 仓库上。 +`log_level = "debug"` 可以看到每条 git 命令。 + +**只想同步分支，不要标签 / PR 引用** +`refs = ["refs/heads/*"]`。默认就不会碰 `refs/pull/*` 这类引用。 + +**大仓库把内存吃满了** +吃内存的是 `git`，不是 syncbot 自己。调小 `concurrency`，并给该仓库加： + +```toml +git_config = ["pack.threads=1", "pack.windowMemory=64m", "core.bigFileThreshold=8m"] +``` + +**想立刻同步一次，不等下个周期** +`docker compose restart syncbot`，或者用 `-once` 单独跑一次。 + +--- + +## 开发 + +```bash +go test ./... # 单元测试 + 跑真实 git 的端到端测试 +go test -race ./... +go vet ./... +go build . +``` + +代码结构和设计约定见 [AGENTS.md](AGENTS.md)。 + +镜像由 GitHub Actions 构建并推送到 `ghcr.io/OWNER/syncbot`， +`main` 分支推 `latest`，打 `v*` 标签推对应语义化版本，支持 amd64 / arm64。 diff --git a/config.example.toml b/config.example.toml new file mode 100644 index 0000000..9819d98 --- /dev/null +++ b/config.example.toml @@ -0,0 +1,106 @@ +# syncbot 配置示例 +# +# 所有字段都有默认值，最小可用配置只需要一个 [[repo]] 块和 src/dst 两个地址。 +# 修改本文件后无需重启：syncbot 会在几秒内自动应用（详见 README 的「热更新」）。 + +[global] +# 本地镜像仓库和 SSH 状态的存放位置，需要可写并建议持久化。 +work_dir = "/var/lib/syncbot" + +# 所有仓库的默认轮询间隔，可被单个 [[repo]] 覆盖。 +interval = "5m" + +# 单次同步（fetch + push）的超时时间。大仓库首次克隆可能较久。 +timeout = "30m" + +# 同时进行的 git 进程数上限。这是内存占用的主要控制项： +# 每个并发任务约等于一个 git 进程的峰值内存。 +concurrency = 2 + +# 连续失败后的退避上限，避免上游长时间挂掉时刷屏。 +max_backoff = "1h" + +# 日志：level = debug|info|warn|error，format = text|json +log_level = "info" +log_format = "text" + +# 可选的 HTTP 端点：/healthz /readyz /status /metrics +# 留空（或删掉本行）则完全不监听端口。 +listen = ":8080" + +# 多久检查一次本文件是否变化。 +reload_interval = "5s" + +# 下面这些是所有仓库的默认值，同样可以在 [[repo]] 里逐个覆盖。 +# refs = ["refs/heads/*", "refs/tags/*"] # 镜像哪些引用 +# prune = true # src 上删掉的分支/标签，也在 dst 上删掉 +# force = true # 允许非快进推送（上游 rebase 后必须） +# atomic = false # true 表示所有引用要么全部更新、要么全不更新 +# allow_empty = false # 允许「空的 src 清空 dst」，默认拒绝，见 README +# ssh_key = "/etc/syncbot/keys/id_ed25519" +# known_hosts = "/etc/syncbot/known_hosts" + + +# --- 典型场景：内网 GitLab -> GitHub，用 deploy key 推送 ------------------- +[[repo]] +name = "my-project" +src = "https://gitlab.internal/team/my-project.git" +dst = "git@github.com:me/my-project.git" + +interval = "1m" + +# GitHub 分配给这个 bot 的 deploy key（需要勾选 "Allow write access"）。 +# 只读挂载、权限是 0644 也没关系，syncbot 会自动做一份 0600 的私有副本。 +ssh_key = "/etc/syncbot/keys/my-project" + + +# --- src 需要 token 的场景 ------------------------------------------------- +# URL 里的 ${VAR} 会用环境变量展开；变量未定义时启动直接报错，不会静默变成空串。 +# 注意只支持 ${VAR} 这种写法，裸写的 $VAR 会原样保留（密码里常有 $）。 +[[repo]] +name = "private-upstream" +src = "https://x-access-token:${SRC_TOKEN}@github.com/upstream/repo.git" +dst = "git@github.com:me/repo-mirror.git" + +# 只镜像分支，不同步标签。 +refs = ["refs/heads/*"] + + +# --- src 和 dst 各自使用不同凭据 ------------------------------------------- +[[repo]] +name = "cross-host" +interval = "10m" + +[repo.src] +url = "git@gitlab.internal:team/repo.git" +ssh_key = "/etc/syncbot/keys/gitlab" +known_hosts = "/etc/syncbot/known_hosts" +strict_host_key = "yes" # yes | no | accept-new + +[repo.dst] +url = "git@github.com:me/repo.git" +ssh_key = "/etc/syncbot/keys/github" + + +# --- 暂时停掉某个仓库，不用删配置 ------------------------------------------ +[[repo]] +name = "paused" +enabled = false +src = "https://example.com/paused.git" +dst = "git@github.com:me/paused.git" + + +# --- 超大仓库：限制 git 自身的内存占用 -------------------------------------- +[[repo]] +name = "huge-monorepo" +src = "https://git.internal/huge.git" +dst = "git@github.com:me/huge.git" +interval = "30m" +timeout = "2h" + +# 原样传给 git 的 -c 参数，用于压住打包时的内存峰值。 +git_config = [ + "pack.threads=1", + "pack.windowMemory=64m", + "core.bigFileThreshold=8m", +] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..ca979c2 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,35 @@ +services: + syncbot: + image: ghcr.io/OWNER/syncbot:latest + container_name: syncbot + restart: unless-stopped + + volumes: + # The config is re-read while running — edit it in place, no restart. + - ./config.toml:/etc/syncbot/config.toml:ro + # Deploy keys. Read-only 0644 mounts are fine; syncbot stages a 0600 copy. + - ./keys:/etc/syncbot/keys:ro + # Local mirrors. Persisting these avoids re-cloning after every restart. + - syncbot-data:/var/lib/syncbot + + # Only needed if you want to reach /healthz, /status or /metrics. + ports: + - "127.0.0.1:8080:8080" + + # Requires `listen` to be set in config.toml. + healthcheck: + test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/healthz"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 10s + + # git is the only thing that needs real memory here; cap it to taste and + # keep `concurrency` in config.toml in line with whatever you allow. + mem_limit: 512m + + security_opt: + - no-new-privileges:true + +volumes: + syncbot-data: diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..ff6f8d2 --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module syncbot + +go 1.24 + +require github.com/BurntSushi/toml v1.4.0 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..8bc10f6 --- /dev/null +++ b/go.sum @@ -0,0 +1,2 @@ +github.com/BurntSushi/toml v1.4.0 h1:kuoIxZQy2WRRk1pttg9asf+WVv6tWQuBNVmK8+nqPr0= +github.com/BurntSushi/toml v1.4.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..7da02a7 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,478 @@ +// Package config loads, validates and resolves syncbot's TOML configuration. +// +// The on-disk document is deliberately forgiving: every knob has a default, +// [global] supplies fallbacks for all repositories, and each [[repo]] overrides +// only what it needs. Load flattens all of that into a []Job in which every +// value is already resolved, so the rest of the daemon never has to reason +// about defaults or inheritance — and so the hot-reload logic can decide +// whether a job changed with a plain reflect.DeepEqual. +package config + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "runtime" + "sort" + "strings" + "time" + + "github.com/BurntSushi/toml" +) + +// Duration is a time.Duration that decodes from a TOML string such as "5m". +type Duration time.Duration + +// UnmarshalText implements encoding.TextUnmarshaler. +func (d *Duration) UnmarshalText(b []byte) error { + v, err := time.ParseDuration(string(b)) + if err != nil { + return fmt.Errorf("invalid duration %q (want e.g. \"30s\", \"5m\", \"2h\")", b) + } + *d = Duration(v) + return nil +} + +// D returns the underlying time.Duration. +func (d Duration) D() time.Duration { return time.Duration(d) } + +// Endpoint is one side of a mirror. In TOML it may be written either as a bare +// URL string or as a table when it needs its own credentials: +// +// dst = "git@github.com:me/repo.git" +// +// [repo.dst] +// url = "git@github.com:me/repo.git" +// ssh_key = "/etc/syncbot/keys/repo" +type Endpoint struct { + URL string `json:"url"` + SSHKey string `json:"-"` + KnownHosts string `json:"-"` + StrictHostKey string `json:"-"` +} + +// UnmarshalTOML accepts both the string and the table spelling of an endpoint. +func (e *Endpoint) UnmarshalTOML(v any) error { + switch t := v.(type) { + case string: + e.URL = t + return nil + case map[string]any: + for _, k := range sortedKeys(t) { + s, ok := t[k].(string) + if !ok { + return fmt.Errorf("key %q must be a string", k) + } + switch k { + case "url": + e.URL = s + case "ssh_key": + e.SSHKey = s + case "known_hosts": + e.KnownHosts = s + case "strict_host_key": + e.StrictHostKey = s + default: + return fmt.Errorf("unknown key %q (want url, ssh_key, known_hosts or strict_host_key)", k) + } + } + if e.URL == "" { + return fmt.Errorf("missing required key \"url\"") + } + return nil + default: + return fmt.Errorf("must be a URL string or a table, got %T", v) + } +} + +// Global holds process-wide settings plus the defaults inherited by every repo. +type Global struct { + WorkDir string `toml:"work_dir"` + Listen string `toml:"listen"` + LogLevel string `toml:"log_level"` + LogFormat string `toml:"log_format"` + Concurrency int `toml:"concurrency"` + ReloadInterval Duration `toml:"reload_interval"` + + // Inherited by every [[repo]] unless overridden there. + Interval Duration `toml:"interval"` + Timeout Duration `toml:"timeout"` + MaxBackoff Duration `toml:"max_backoff"` + Refs []string `toml:"refs"` + Prune *bool `toml:"prune"` + Force *bool `toml:"force"` + Atomic *bool `toml:"atomic"` + AllowEmpty *bool `toml:"allow_empty"` + SSHKey string `toml:"ssh_key"` + KnownHosts string `toml:"known_hosts"` + StrictHostKey string `toml:"strict_host_key"` + GitConfig []string `toml:"git_config"` +} + +// Repo is one [[repo]] block as written by the user. +type Repo struct { + Name string `toml:"name"` + Src Endpoint `toml:"src"` + Dst Endpoint `toml:"dst"` + Enabled *bool `toml:"enabled"` + + Interval Duration `toml:"interval"` + Timeout Duration `toml:"timeout"` + MaxBackoff Duration `toml:"max_backoff"` + Refs []string `toml:"refs"` + Prune *bool `toml:"prune"` + Force *bool `toml:"force"` + Atomic *bool `toml:"atomic"` + AllowEmpty *bool `toml:"allow_empty"` + SSHKey string `toml:"ssh_key"` + KnownHosts string `toml:"known_hosts"` + StrictHostKey string `toml:"strict_host_key"` + GitConfig []string `toml:"git_config"` +} + +// file mirrors the TOML document itself. +type file struct { + Global Global `toml:"global"` + Repos []Repo `toml:"repo"` +} + +// Job is a fully resolved sync unit: all defaults folded in, ready to run. +// Every field is comparable with reflect.DeepEqual, which is how the manager +// detects that a reloaded config actually changed something for this repo. +type Job struct { + Name string + Src Endpoint + Dst Endpoint + Dir string // local bare mirror + + Interval time.Duration + Timeout time.Duration + MaxBackoff time.Duration + + Refs []string + Prune bool + Force bool + Atomic bool + AllowEmpty bool + GitConfig []string +} + +// Config is the resolved configuration the daemon runs on. +type Config struct { + WorkDir string + Listen string + LogLevel string + LogFormat string + Concurrency int + ReloadInterval time.Duration + Jobs []Job +} + +// Defaults applied when the document leaves a value out. +const ( + DefaultWorkDir = "/var/lib/syncbot" + DefaultInterval = 5 * time.Minute + DefaultTimeout = 30 * time.Minute + DefaultMaxBackoff = time.Hour + DefaultReloadInterval = 5 * time.Second + DefaultLogLevel = "info" + DefaultLogFormat = "text" +) + +// DefaultRefs mirrors branches and tags, which is what almost everyone wants. +var DefaultRefs = []string{"refs/heads/*", "refs/tags/*"} + +// nameRe keeps repo names usable as directory names and Prometheus labels. +var nameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`) + +// Load reads, validates and resolves the config file at path. +func Load(path string) (*Config, error) { + b, err := os.ReadFile(path) + if err != nil { + return nil, err + } + return Parse(b) +} + +// Parse resolves an in-memory TOML document. Load is the usual entry point; +// Parse exists so tests (and -check) can work without touching disk. +func Parse(b []byte) (*Config, error) { + var f file + md, err := toml.Decode(string(b), &f) + if err != nil { + return nil, err + } + if undec := unknownKeys(md); len(undec) > 0 { + return nil, fmt.Errorf("unknown config key(s): %s", strings.Join(undec, ", ")) + } + return resolve(&f) +} + +// unknownKeys reports keys the decoder did not recognise, so a typo like +// "intervall" fails loudly at startup instead of silently doing nothing. +func unknownKeys(md toml.MetaData) []string { + var out []string + for _, k := range md.Undecoded() { + s := k.String() + // Endpoint consumes its own subtree via UnmarshalTOML and validates the + // keys itself; the decoder cannot see into it, so ignore those paths. + if strings.Contains(s, ".src.") || strings.Contains(s, ".dst.") { + continue + } + out = append(out, s) + } + sort.Strings(out) + return out +} + +func resolve(f *file) (*Config, error) { + g := f.Global + + workDir, err := expandEnv(orString(g.WorkDir, DefaultWorkDir)) + if err != nil { + return nil, fmt.Errorf("global.work_dir: %w", err) + } + if !filepath.IsAbs(workDir) { + if workDir, err = filepath.Abs(workDir); err != nil { + return nil, fmt.Errorf("global.work_dir: %w", err) + } + } + + cfg := &Config{ + WorkDir: workDir, + Listen: g.Listen, + LogLevel: strings.ToLower(orString(g.LogLevel, DefaultLogLevel)), + LogFormat: strings.ToLower(orString(g.LogFormat, DefaultLogFormat)), + Concurrency: g.Concurrency, + ReloadInterval: orDuration(g.ReloadInterval, DefaultReloadInterval), + } + if cfg.Concurrency <= 0 { + cfg.Concurrency = min(4, runtime.NumCPU()) + } + switch cfg.LogLevel { + case "debug", "info", "warn", "error": + default: + return nil, fmt.Errorf("global.log_level: %q is not one of debug, info, warn, error", cfg.LogLevel) + } + switch cfg.LogFormat { + case "text", "json": + default: + return nil, fmt.Errorf("global.log_format: %q is not one of text, json", cfg.LogFormat) + } + if cfg.ReloadInterval < time.Second { + return nil, fmt.Errorf("global.reload_interval: must be at least 1s") + } + if len(f.Repos) == 0 { + return nil, fmt.Errorf("no [[repo]] blocks defined: nothing to sync") + } + + seen := make(map[string]bool, len(f.Repos)) + for i := range f.Repos { + r := &f.Repos[i] + where := fmt.Sprintf("repo[%d]", i) + if r.Name != "" { + where = fmt.Sprintf("repo %q", r.Name) + } + if !nameRe.MatchString(r.Name) { + return nil, fmt.Errorf("%s: name must match %s", where, nameRe) + } + if seen[r.Name] { + return nil, fmt.Errorf("%s: duplicate name", where) + } + seen[r.Name] = true + + if enabled := r.Enabled; enabled != nil && !*enabled { + continue + } + + job, err := resolveJob(r, &g, workDir) + if err != nil { + return nil, fmt.Errorf("%s: %w", where, err) + } + cfg.Jobs = append(cfg.Jobs, *job) + } + if len(cfg.Jobs) == 0 { + return nil, fmt.Errorf("every [[repo]] is disabled: nothing to sync") + } + sort.Slice(cfg.Jobs, func(i, j int) bool { return cfg.Jobs[i].Name < cfg.Jobs[j].Name }) + return cfg, nil +} + +func resolveJob(r *Repo, g *Global, workDir string) (*Job, error) { + j := &Job{ + Name: r.Name, + Dir: filepath.Join(workDir, "mirrors", r.Name+".git"), + Interval: pick(r.Interval, g.Interval, DefaultInterval), + Timeout: pick(r.Timeout, g.Timeout, DefaultTimeout), + MaxBackoff: pick(r.MaxBackoff, g.MaxBackoff, DefaultMaxBackoff), + Refs: orSlice(r.Refs, g.Refs, DefaultRefs), + Prune: orBool(true, r.Prune, g.Prune), + Force: orBool(true, r.Force, g.Force), + Atomic: orBool(false, r.Atomic, g.Atomic), + AllowEmpty: orBool(false, r.AllowEmpty, g.AllowEmpty), + GitConfig: orSlice(r.GitConfig, g.GitConfig, nil), + } + + var err error + // SSH settings cascade: endpoint table -> [[repo]] -> [global]. + if j.Src, err = resolveEndpoint(r.Src, r, g); err != nil { + return nil, fmt.Errorf("src: %w", err) + } + if j.Dst, err = resolveEndpoint(r.Dst, r, g); err != nil { + return nil, fmt.Errorf("dst: %w", err) + } + if j.Src.URL == "" { + return nil, fmt.Errorf("src is required") + } + if j.Dst.URL == "" { + return nil, fmt.Errorf("dst is required") + } + if j.Src.URL == j.Dst.URL { + return nil, fmt.Errorf("src and dst are the same repository") + } + if j.Interval <= 0 { + return nil, fmt.Errorf("interval must be positive") + } + if j.Timeout <= 0 { + return nil, fmt.Errorf("timeout must be positive") + } + if j.MaxBackoff < j.Interval { + j.MaxBackoff = j.Interval + } + if len(j.Refs) == 0 { + return nil, fmt.Errorf("refs must not be empty") + } + for _, p := range j.Refs { + if !strings.HasPrefix(p, "refs/") { + return nil, fmt.Errorf("refs: %q must start with \"refs/\"", p) + } + if strings.Count(p, "*") > 1 { + return nil, fmt.Errorf("refs: %q may contain at most one \"*\"", p) + } + } + for _, kv := range j.GitConfig { + if !strings.Contains(kv, "=") { + return nil, fmt.Errorf("git_config: %q must be in key=value form", kv) + } + } + return j, nil +} + +func resolveEndpoint(e Endpoint, r *Repo, g *Global) (Endpoint, error) { + out := Endpoint{ + URL: e.URL, + SSHKey: orString(e.SSHKey, r.SSHKey, g.SSHKey), + KnownHosts: orString(e.KnownHosts, r.KnownHosts, g.KnownHosts), + StrictHostKey: orString(e.StrictHostKey, r.StrictHostKey, g.StrictHostKey), + } + + var err error + for _, p := range []*string{&out.URL, &out.SSHKey, &out.KnownHosts} { + if *p, err = expandEnv(*p); err != nil { + return out, err + } + } + + if out.StrictHostKey == "" { + // With a pinned known_hosts file we can afford to be strict; without + // one, trust-on-first-use is the only thing that can work unattended. + if out.KnownHosts != "" { + out.StrictHostKey = "yes" + } else { + out.StrictHostKey = "accept-new" + } + } + switch out.StrictHostKey { + case "yes", "no", "accept-new": + default: + return out, fmt.Errorf("strict_host_key: %q is not one of yes, no, accept-new", out.StrictHostKey) + } + if out.SSHKey != "" && !filepath.IsAbs(out.SSHKey) { + return out, fmt.Errorf("ssh_key: %q must be an absolute path", out.SSHKey) + } + return out, nil +} + +// envRe matches ${VAR}. Bare $VAR is deliberately not expanded so that secrets +// containing a literal '$' survive unharmed. +var envRe = regexp.MustCompile(`\$\{([A-Za-z_][A-Za-z0-9_]*)\}`) + +// expandEnv substitutes ${VAR} references, failing loudly on undefined names — +// silently expanding to "" would produce a subtly broken URL instead. +func expandEnv(s string) (string, error) { + if !strings.Contains(s, "${") { + return s, nil + } + var missing []string + out := envRe.ReplaceAllStringFunc(s, func(m string) string { + name := m[2 : len(m)-1] + v, ok := os.LookupEnv(name) + if !ok { + missing = append(missing, name) + return "" + } + return v + }) + if len(missing) > 0 { + return "", fmt.Errorf("undefined environment variable(s): %s", strings.Join(missing, ", ")) + } + return out, nil +} + +func orString(vs ...string) string { + for _, v := range vs { + if v != "" { + return v + } + } + return "" +} + +func orBool(def bool, vs ...*bool) bool { + for _, v := range vs { + if v != nil { + return *v + } + } + return def +} + +func orSlice(vs ...[]string) []string { + for _, v := range vs { + if len(v) > 0 { + return append([]string(nil), v...) + } + } + return nil +} + +func orDuration(v Duration, def time.Duration) time.Duration { + if v != 0 { + return v.D() + } + return def +} + +func pick(repo, global Duration, def time.Duration) time.Duration { + if repo != 0 { + return repo.D() + } + if global != 0 { + return global.D() + } + return def +} + +func sortedKeys(m map[string]any) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// orBool with a literal default needs a *bool; these make the call sites read +// naturally without sprinkling helper variables around. +func boolPtr(b bool) *bool { return &b } diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..f691b99 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,314 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +const minimal = ` +[[repo]] +name = "demo" +src = "https://git.example.com/demo.git" +dst = "git@github.com:me/demo.git" +` + +func mustParse(t *testing.T, doc string) *Config { + t.Helper() + cfg, err := Parse([]byte(doc)) + if err != nil { + t.Fatalf("parse: %v", err) + } + return cfg +} + +func TestDefaultsAreApplied(t *testing.T) { + cfg := mustParse(t, minimal) + + if cfg.WorkDir != DefaultWorkDir { + t.Errorf("work_dir = %q, want %q", cfg.WorkDir, DefaultWorkDir) + } + if cfg.LogLevel != "info" || cfg.LogFormat != "text" { + t.Errorf("log defaults = %q/%q", cfg.LogLevel, cfg.LogFormat) + } + if cfg.Concurrency < 1 { + t.Errorf("concurrency = %d, want >= 1", cfg.Concurrency) + } + + j := cfg.Jobs[0] + if j.Interval != DefaultInterval || j.Timeout != DefaultTimeout { + t.Errorf("interval/timeout = %s/%s", j.Interval, j.Timeout) + } + if !j.Prune || !j.Force { + t.Error("prune and force should default to true for a mirror") + } + if j.Atomic || j.AllowEmpty { + t.Error("atomic and allow_empty should default to false") + } + if strings.Join(j.Refs, ",") != strings.Join(DefaultRefs, ",") { + t.Errorf("refs = %v, want %v", j.Refs, DefaultRefs) + } + if !strings.HasSuffix(j.Dir, "mirrors/demo.git") { + t.Errorf("mirror dir = %q", j.Dir) + } +} + +func TestGlobalDefaultsCascadeAndRepoOverrides(t *testing.T) { + cfg := mustParse(t, ` +[global] +work_dir = "/data" +interval = "10m" +prune = false +ssh_key = "/keys/shared" + +[[repo]] +name = "inherits" +src = "https://example.com/a.git" +dst = "git@github.com:me/a.git" + +[[repo]] +name = "overrides" +src = "https://example.com/b.git" +dst = "git@github.com:me/b.git" +interval = "30s" +prune = true +ssh_key = "/keys/b" +`) + + byName := map[string]Job{} + for _, j := range cfg.Jobs { + byName[j.Name] = j + } + + a := byName["inherits"] + if a.Interval != 10*time.Minute { + t.Errorf("inherited interval = %s, want 10m", a.Interval) + } + if a.Prune { + t.Error("inherited prune should be false") + } + if a.Dst.SSHKey != "/keys/shared" { + t.Errorf("inherited ssh_key = %q", a.Dst.SSHKey) + } + + b := byName["overrides"] + if b.Interval != 30*time.Second { + t.Errorf("overridden interval = %s, want 30s", b.Interval) + } + if !b.Prune { + t.Error("overridden prune should be true") + } + if b.Dst.SSHKey != "/keys/b" { + t.Errorf("overridden ssh_key = %q", b.Dst.SSHKey) + } +} + +func TestEndpointAcceptsStringOrTable(t *testing.T) { + cfg := mustParse(t, ` +[[repo]] +name = "demo" +src = "https://git.example.com/demo.git" + +[repo.dst] +url = "git@github.com:me/demo.git" +ssh_key = "/keys/demo" +known_hosts = "/etc/syncbot/known_hosts" +strict_host_key = "yes" +`) + + j := cfg.Jobs[0] + if j.Src.URL != "https://git.example.com/demo.git" { + t.Errorf("src url = %q", j.Src.URL) + } + if j.Dst.URL != "git@github.com:me/demo.git" { + t.Errorf("dst url = %q", j.Dst.URL) + } + if j.Dst.SSHKey != "/keys/demo" || j.Dst.KnownHosts != "/etc/syncbot/known_hosts" { + t.Errorf("dst ssh settings = %+v", j.Dst) + } + if j.Dst.StrictHostKey != "yes" { + t.Errorf("strict_host_key = %q", j.Dst.StrictHostKey) + } + // src has no key of its own and none was inherited. + if j.Src.SSHKey != "" { + t.Errorf("src ssh_key = %q, want empty", j.Src.SSHKey) + } +} + +func TestStrictHostKeyDefaultFollowsKnownHosts(t *testing.T) { + cfg := mustParse(t, minimal) + if got := cfg.Jobs[0].Dst.StrictHostKey; got != "accept-new" { + t.Errorf("without known_hosts: %q, want accept-new", got) + } + + cfg = mustParse(t, minimal+` +[global] +known_hosts = "/etc/syncbot/known_hosts" +`) + if got := cfg.Jobs[0].Dst.StrictHostKey; got != "yes" { + t.Errorf("with known_hosts: %q, want yes", got) + } +} + +func TestEnvExpansion(t *testing.T) { + t.Setenv("SYNCBOT_TEST_TOKEN", "s3cr#t$") + + cfg := mustParse(t, ` +[[repo]] +name = "demo" +src = "https://x-access-token:${SYNCBOT_TEST_TOKEN}@github.com/me/demo.git" +dst = "git@github.com:me/mirror.git" +`) + want := "https://x-access-token:s3cr#t$@github.com/me/demo.git" + if got := cfg.Jobs[0].Src.URL; got != want { + t.Errorf("expanded src = %q, want %q", got, want) + } +} + +func TestUndefinedEnvVarIsAnError(t *testing.T) { + _, err := Parse([]byte(` +[[repo]] +name = "demo" +src = "https://${SYNCBOT_DEFINITELY_UNSET}@example.com/a.git" +dst = "git@github.com:me/a.git" +`)) + if err == nil || !strings.Contains(err.Error(), "SYNCBOT_DEFINITELY_UNSET") { + t.Fatalf("want an error naming the missing variable, got %v", err) + } +} + +func TestDisabledRepoIsSkipped(t *testing.T) { + cfg := mustParse(t, ` +[[repo]] +name = "on" +src = "https://example.com/a.git" +dst = "git@github.com:me/a.git" + +[[repo]] +name = "off" +enabled = false +src = "https://example.com/b.git" +dst = "git@github.com:me/b.git" +`) + if len(cfg.Jobs) != 1 || cfg.Jobs[0].Name != "on" { + t.Fatalf("want only the enabled repo, got %d: %+v", len(cfg.Jobs), cfg.Jobs) + } +} + +func TestValidationErrors(t *testing.T) { + cases := []struct{ name, doc, want string }{ + {"no repos", `[global] +work_dir = "/data"`, "nothing to sync"}, + {"missing name", `[[repo]] +src = "a" +dst = "b"`, "name must match"}, + {"bad name", `[[repo]] +name = "../escape" +src = "a" +dst = "b"`, "name must match"}, + {"duplicate name", `[[repo]] +name = "x" +src = "a" +dst = "b" +[[repo]] +name = "x" +src = "c" +dst = "d"`, "duplicate name"}, + {"missing dst", `[[repo]] +name = "x" +src = "a"`, "dst is required"}, + {"same src and dst", `[[repo]] +name = "x" +src = "a" +dst = "a"`, "same repository"}, + {"unknown key", `[[repo]] +name = "x" +src = "a" +dst = "b" +intervall = "5m"`, "unknown config key"}, + {"bad duration", `[[repo]] +name = "x" +src = "a" +dst = "b" +interval = "5 minutes"`, "invalid duration"}, + {"relative ssh key", `[[repo]] +name = "x" +src = "a" +dst = "b" +ssh_key = "keys/x"`, "absolute path"}, + {"ref without prefix", `[[repo]] +name = "x" +src = "a" +dst = "b" +refs = ["heads/*"]`, `must start with "refs/"`}, + {"two globs", `[[repo]] +name = "x" +src = "a" +dst = "b" +refs = ["refs/*/*"]`, `at most one`}, + {"bad log level", `[global] +log_level = "verbose" +[[repo]] +name = "x" +src = "a" +dst = "b"`, "log_level"}, + {"bad endpoint key", `[[repo]] +name = "x" +src = "a" +[repo.dst] +url = "b" +sshkey = "/k"`, "unknown key"}, + {"endpoint table without url", `[[repo]] +name = "x" +src = "a" +[repo.dst] +ssh_key = "/k"`, "url"}, + {"bad git_config", `[[repo]] +name = "x" +src = "a" +dst = "b" +git_config = ["pack.threads"]`, "key=value"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := Parse([]byte(tc.doc)) + if err == nil { + t.Fatalf("want an error mentioning %q", tc.want) + } + if !strings.Contains(err.Error(), tc.want) { + t.Errorf("error = %q, want it to mention %q", err, tc.want) + } + }) + } +} + +func TestMaxBackoffNeverBelowInterval(t *testing.T) { + cfg := mustParse(t, ` +[[repo]] +name = "x" +src = "a" +dst = "b" +interval = "10m" +max_backoff = "1m" +`) + if got := cfg.Jobs[0].MaxBackoff; got != 10*time.Minute { + t.Errorf("max_backoff = %s, want it raised to the interval (10m)", got) + } +} + +func TestJobsAreSortedForStableDiffs(t *testing.T) { + cfg := mustParse(t, ` +[[repo]] +name = "zulu" +src = "a" +dst = "b" +[[repo]] +name = "alpha" +src = "c" +dst = "d" +`) + if cfg.Jobs[0].Name != "alpha" || cfg.Jobs[1].Name != "zulu" { + t.Errorf("jobs not sorted: %s, %s", cfg.Jobs[0].Name, cfg.Jobs[1].Name) + } +} diff --git a/internal/gitx/gitx.go b/internal/gitx/gitx.go new file mode 100644 index 0000000..0bfbaa4 --- /dev/null +++ b/internal/gitx/gitx.go @@ -0,0 +1,444 @@ +// Package gitx is a thin, hermetic wrapper around the git command line. +// +// Shelling out to git — rather than linking a pure-Go implementation — keeps +// the binary small and the memory profile flat: the heavy lifting happens in a +// short-lived child process that the kernel reclaims when it exits. +// +// Every invocation runs with an isolated HOME and with the system/global git +// config disabled, so syncbot behaves identically no matter whose account or +// container it runs in. +package gitx + +import ( + "bytes" + "context" + "fmt" + "io" + "log/slog" + "net/url" + "os" + "os/exec" + "path/filepath" + "strings" + "time" +) + +// SSH describes how to authenticate to one endpoint over SSH. +type SSH struct { + KeyPath string // deploy key; empty means "use the agent / default keys" + KnownHosts string // pinned host key file; empty means use Home/known_hosts + StrictHostKey string // yes | no | accept-new +} + +// Options carry everything a git invocation needs beyond its arguments. +type Options struct { + Log *slog.Logger + Home string // isolated HOME for git and ssh; must exist + GitConfig []string // extra "key=value" settings passed as -c + SSHCmd string // pre-built GIT_SSH_COMMAND, see PrepareSSH + Secrets []string // substrings scrubbed from logs and error messages +} + +// Refs maps a full ref name to the object it points at. +type Refs map[string]string + +// Equal reports whether two ref sets are identical. +func (r Refs) Equal(other Refs) bool { + if len(r) != len(other) { + return false + } + for k, v := range r { + if other[k] != v { + return false + } + } + return true +} + +// termGrace is how long a git child gets to exit after SIGTERM before we +// escalate to SIGKILL on its whole process group. +const termGrace = 3 * time.Second + +// Run executes git with the given arguments and returns its stdout. +// +// The child is put in its own process group so that a timeout or a shutdown +// takes down the helpers git spawns (ssh, git-remote-https) instead of leaking +// them. +func Run(ctx context.Context, o Options, dir string, args ...string) (string, error) { + full := make([]string, 0, len(args)+2*len(o.GitConfig)) + for _, kv := range o.GitConfig { + full = append(full, "-c", kv) + } + full = append(full, args...) + + cmd := exec.Command("git", full...) + cmd.Dir = dir + cmd.Env = environ(o) + cmd.SysProcAttr = sysProcAttr() + + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &limitedWriter{W: &stderr, N: 64 << 10} + + started := time.Now() + if err := cmd.Start(); err != nil { + return "", fmt.Errorf("git %s: %w", args[0], err) + } + + // Watchdog: translate context cancellation into signals for the group. + watchdogDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + terminate(cmd.Process.Pid) + case <-watchdogDone: + } + }() + + err := cmd.Wait() + close(watchdogDone) + + if o.Log != nil && o.Log.Enabled(ctx, slog.LevelDebug) { + o.Log.Debug("git", "args", scrub(strings.Join(args, " "), o.Secrets), + "dur", time.Since(started).Round(time.Millisecond), "err", err) + } + + if err != nil { + msg := scrub(strings.TrimSpace(stderr.String()), o.Secrets) + if ctx.Err() != nil { + return "", fmt.Errorf("git %s: %w (%s)", args[0], ctx.Err(), firstLines(msg, 3)) + } + if msg == "" { + msg = err.Error() + } + return "", fmt.Errorf("git %s: %s", args[0], firstLines(msg, 8)) + } + return stdout.String(), nil +} + +// environ builds a deterministic environment: the parent's, minus anything that +// could redirect git's authentication or config, plus our own settings. +func environ(o Options) []string { + drop := map[string]bool{ + "HOME": true, "XDG_CONFIG_HOME": true, + "GIT_SSH": true, "GIT_SSH_COMMAND": true, "GIT_ASKPASS": true, "SSH_ASKPASS": true, + "GIT_CONFIG": true, "GIT_CONFIG_GLOBAL": true, "GIT_CONFIG_SYSTEM": true, + "GIT_DIR": true, "GIT_WORK_TREE": true, "GIT_TERMINAL_PROMPT": true, + } + out := make([]string, 0, 16) + for _, kv := range os.Environ() { + if k, _, ok := strings.Cut(kv, "="); ok && !drop[k] { + out = append(out, kv) + } + } + out = append(out, + "HOME="+o.Home, + "GIT_CONFIG_GLOBAL="+os.DevNull, + "GIT_CONFIG_SYSTEM="+os.DevNull, + "GIT_TERMINAL_PROMPT=0", // never block waiting for a password + "SSH_ASKPASS_REQUIRE=never", + "LC_ALL=C", + ) + if o.SSHCmd != "" { + out = append(out, "GIT_SSH_COMMAND="+o.SSHCmd) + } + return out +} + +// PrepareSSH builds a GIT_SSH_COMMAND for the endpoint. +// +// Deploy keys are usually mounted read-only from a secret store, which often +// means mode 0644 — and ssh flatly refuses group- or world-readable keys. When +// that happens we copy the key into tmpDir at 0600 rather than asking the +// operator to fix permissions they may not control. +func PrepareSSH(s SSH, home, tmpDir string) (string, error) { + knownHosts := s.KnownHosts + if knownHosts == "" { + knownHosts = filepath.Join(home, "known_hosts") + if _, err := os.Stat(knownHosts); os.IsNotExist(err) { + if err := os.WriteFile(knownHosts, nil, 0o600); err != nil { + return "", fmt.Errorf("create known_hosts: %w", err) + } + } + } + + strict := s.StrictHostKey + if strict == "" { + strict = "accept-new" + } + + args := []string{"ssh", + "-o", "BatchMode=yes", + "-o", "StrictHostKeyChecking=" + strict, + "-o", "UserKnownHostsFile=" + knownHosts, + "-o", "ConnectTimeout=30", + } + + if s.KeyPath != "" { + key, err := usableKey(s.KeyPath, tmpDir) + if err != nil { + return "", err + } + // IdentitiesOnly stops ssh from offering an agent's keys first and + // tripping GitHub's "too many authentication failures". + args = append(args, "-i", key, "-o", "IdentitiesOnly=yes") + } + + quoted := make([]string, len(args)) + for i, a := range args { + quoted[i] = shellQuote(a) + } + return strings.Join(quoted, " "), nil +} + +func usableKey(path, tmpDir string) (string, error) { + fi, err := os.Stat(path) + if err != nil { + return "", fmt.Errorf("ssh_key: %w", err) + } + if fi.IsDir() { + return "", fmt.Errorf("ssh_key: %s is a directory", path) + } + if fi.Mode().Perm()&0o077 == 0 { + return path, nil + } + // Too permissive for ssh: stage a private copy. + b, err := os.ReadFile(path) + if err != nil { + return "", fmt.Errorf("ssh_key: %w", err) + } + dst := filepath.Join(tmpDir, "id_"+filepath.Base(path)) + if err := os.WriteFile(dst, b, 0o600); err != nil { + return "", fmt.Errorf("ssh_key: stage private copy: %w", err) + } + return dst, nil +} + +// EnsureMirror makes sure dir holds a usable bare repository, creating it on +// first run. A directory that exists but is not a bare repo is reported rather +// than deleted — that is almost always a misconfigured mount, and silently +// wiping it would be the wrong kind of helpful. +func EnsureMirror(ctx context.Context, o Options, dir string) error { + if _, err := os.Stat(filepath.Join(dir, "HEAD")); err == nil { + out, err := Run(ctx, o, dir, "rev-parse", "--is-bare-repository") + if err != nil { + return fmt.Errorf("%s exists but is not a git repository: %w", dir, err) + } + if strings.TrimSpace(out) != "true" { + return fmt.Errorf("%s is not a bare repository", dir) + } + return nil + } else if !os.IsNotExist(err) { + return err + } + + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + _, err := Run(ctx, o, o.Home, "init", "--bare", "--quiet", "--initial-branch=main", dir) + return err +} + +// LsRemote asks a remote which refs it currently has, without transferring any +// objects. This is the "check for updates" probe: cheap enough to run on a +// short interval even against large repositories. +func LsRemote(ctx context.Context, o Options, dir, repoURL string, patterns []string) (Refs, error) { + args := append([]string{"ls-remote", "--refs", "--", repoURL}, patterns...) + out, err := Run(ctx, o, dir, args...) + if err != nil { + return nil, err + } + return parseRefs(out, "\t", true, patterns), nil +} + +// LocalRefs reads the mirror's own refs, filtered to the managed patterns. +func LocalRefs(ctx context.Context, o Options, dir string, patterns []string) (Refs, error) { + out, err := Run(ctx, o, dir, "for-each-ref", "--format=%(objectname)\t%(refname)") + if err != nil { + return nil, err + } + return parseRefs(out, "\t", true, patterns), nil +} + +// Fetch updates the mirror from src. Refs that vanished upstream are pruned so +// the mirror is an exact copy of the managed namespace, not an accumulation. +func Fetch(ctx context.Context, o Options, dir, src string, patterns []string) error { + args := []string{"fetch", "--force", "--no-tags", "--no-write-fetch-head", "--prune", "--quiet", "--", src} + for _, p := range patterns { + args = append(args, "+"+p+":"+p) + } + _, err := Run(ctx, o, dir, args...) + return err +} + +// PushOptions controls how the mirror is written to the destination. +type PushOptions struct { + Prune bool // delete destination refs that no longer exist upstream + Force bool // allow non-fast-forward updates (a mirror must) + Atomic bool // all refs update, or none do +} + +// Push writes the mirror's managed refs to dst and returns the porcelain lines +// describing what actually changed. +func Push(ctx context.Context, o Options, dir, dst string, patterns []string, po PushOptions) ([]string, error) { + args := []string{"push", "--porcelain"} + if po.Prune { + args = append(args, "--prune") + } + if po.Force { + args = append(args, "--force") + } + if po.Atomic { + args = append(args, "--atomic") + } + args = append(args, "--", dst) + for _, p := range patterns { + args = append(args, p+":"+p) + } + + out, err := Run(ctx, o, dir, args...) + if err != nil { + return nil, err + } + + var changed []string + for _, line := range strings.Split(out, "\n") { + line = strings.TrimRight(line, "\r") + // Porcelain format: "\t:\t". '=' means the ref + // was already up to date, which is the boring majority. + if line == "" || strings.HasPrefix(line, "To ") || line == "Done" || strings.HasPrefix(line, "=\t") { + continue + } + changed = append(changed, scrub(line, o.Secrets)) + } + return changed, nil +} + +// MatchRef implements git's refspec globbing: at most one "*", which matches +// any run of characters including "/". +func MatchRef(pattern, ref string) bool { + i := strings.IndexByte(pattern, '*') + if i < 0 { + return pattern == ref + } + prefix, suffix := pattern[:i], pattern[i+1:] + return len(ref) >= len(prefix)+len(suffix) && + strings.HasPrefix(ref, prefix) && + strings.HasSuffix(ref, suffix) +} + +// MatchAny reports whether ref matches any of the patterns. +func MatchAny(patterns []string, ref string) bool { + for _, p := range patterns { + if MatchRef(p, ref) { + return true + } + } + return false +} + +// parseRefs reads "" lines, optionally filtering to patterns. +func parseRefs(out, sep string, filter bool, patterns []string) Refs { + refs := make(Refs) + for _, line := range strings.Split(out, "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + obj, ref, ok := strings.Cut(line, sep) + if !ok { + continue + } + // Peeled entries ("refs/tags/v1^{}") describe the commit behind an + // annotated tag; the tag object itself is what we mirror. + if strings.HasSuffix(ref, "^{}") { + continue + } + if filter && !MatchAny(patterns, ref) { + continue + } + refs[ref] = obj + } + return refs +} + +// RedactURL strips the password from a URL so it can be logged. +func RedactURL(raw string) string { + if !strings.Contains(raw, "://") { + return raw // scp-style (git@host:path) carries no inline secret + } + u, err := url.Parse(raw) + if err != nil || u.User == nil { + return raw + } + if _, hasPassword := u.User.Password(); hasPassword { + // Plain letters: anything punctuation-ish would come back + // percent-encoded from URL.String() and read as noise in a log line. + u.User = url.UserPassword(u.User.Username(), "redacted") + } else { + u.User = url.User(u.User.Username()) + } + return u.String() +} + +// URLSecret returns the credential embedded in a URL, if any, so callers can +// register it with Options.Secrets and keep it out of logs. +func URLSecret(raw string) string { + if !strings.Contains(raw, "://") { + return "" + } + u, err := url.Parse(raw) + if err != nil || u.User == nil { + return "" + } + if pw, ok := u.User.Password(); ok && pw != "" { + return pw + } + return "" +} + +func scrub(s string, secrets []string) string { + for _, sec := range secrets { + if sec != "" { + s = strings.ReplaceAll(s, sec, "***") + } + } + return s +} + +func firstLines(s string, n int) string { + lines := strings.Split(s, "\n") + if len(lines) > n { + lines = append(lines[:n], "...") + } + return strings.Join(lines, "; ") +} + +// shellQuote makes a token safe for GIT_SSH_COMMAND, which git hands to a shell. +func shellQuote(s string) string { + if s != "" && !strings.ContainsAny(s, " \t\n\"'\\$`&;|<>()*?[]{}#~!") { + return s + } + return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" +} + +// limitedWriter keeps a runaway stderr from growing without bound. +type limitedWriter struct { + W io.Writer + N int +} + +// Write always reports the full length: a short write would be treated as an +// error by os/exec and would abort an otherwise healthy git invocation. +func (l *limitedWriter) Write(p []byte) (int, error) { + total := len(p) + if l.N <= 0 { + return total, nil + } + if len(p) > l.N { + p = p[:l.N] + } + n, err := l.W.Write(p) + l.N -= n + return total, err +} diff --git a/internal/gitx/gitx_test.go b/internal/gitx/gitx_test.go new file mode 100644 index 0000000..f4e7eaa --- /dev/null +++ b/internal/gitx/gitx_test.go @@ -0,0 +1,221 @@ +package gitx + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestMatchRef(t *testing.T) { + cases := []struct { + pattern, ref string + want bool + }{ + {"refs/heads/*", "refs/heads/main", true}, + {"refs/heads/*", "refs/heads/feature/nested/deep", true}, // '*' spans '/' + {"refs/heads/*", "refs/tags/v1", false}, + {"refs/heads/*", "refs/heads/", true}, + {"refs/heads/main", "refs/heads/main", true}, + {"refs/heads/main", "refs/heads/maint", false}, + {"refs/tags/v*", "refs/tags/v1.2.3", true}, + {"refs/tags/v*", "refs/tags/rc1", false}, + {"refs/heads/*-stable", "refs/heads/2.0-stable", true}, + {"refs/heads/*-stable", "refs/heads/2.0-beta", false}, + } + for _, tc := range cases { + if got := MatchRef(tc.pattern, tc.ref); got != tc.want { + t.Errorf("MatchRef(%q, %q) = %v, want %v", tc.pattern, tc.ref, got, tc.want) + } + } +} + +func TestParseRefsSkipsPeeledAndFilters(t *testing.T) { + out := strings.Join([]string{ + "aaa\trefs/heads/main", + "bbb\trefs/tags/v1", + "ccc\trefs/tags/v1^{}", // peeled annotated tag + "ddd\trefs/pull/7/head", + "", + }, "\n") + + refs := parseRefs(out, "\t", true, []string{"refs/heads/*", "refs/tags/*"}) + + if len(refs) != 2 { + t.Fatalf("got %d refs, want 2: %v", len(refs), refs) + } + if refs["refs/heads/main"] != "aaa" || refs["refs/tags/v1"] != "bbb" { + t.Errorf("unexpected refs: %v", refs) + } + if _, ok := refs["refs/tags/v1^{}"]; ok { + t.Error("peeled tag entry should be dropped") + } + if _, ok := refs["refs/pull/7/head"]; ok { + t.Error("unmanaged ref should be filtered out") + } +} + +func TestRefsEqual(t *testing.T) { + a := Refs{"refs/heads/main": "1", "refs/tags/v1": "2"} + if !a.Equal(Refs{"refs/tags/v1": "2", "refs/heads/main": "1"}) { + t.Error("same contents should compare equal regardless of order") + } + if a.Equal(Refs{"refs/heads/main": "1"}) { + t.Error("different sizes should not compare equal") + } + if a.Equal(Refs{"refs/heads/main": "1", "refs/tags/v1": "9"}) { + t.Error("different objects should not compare equal") + } +} + +func TestRedactURL(t *testing.T) { + cases := []struct{ in, want string }{ + {"https://user:token@github.com/me/x.git", "https://user:redacted@github.com/me/x.git"}, + {"https://token@github.com/me/x.git", "https://token@github.com/me/x.git"}, + {"https://github.com/me/x.git", "https://github.com/me/x.git"}, + {"git@github.com:me/x.git", "git@github.com:me/x.git"}, + {"/srv/git/local.git", "/srv/git/local.git"}, + } + for _, tc := range cases { + if got := RedactURL(tc.in); got != tc.want { + t.Errorf("RedactURL(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +func TestURLSecret(t *testing.T) { + if got := URLSecret("https://x-access-token:ghp_abc@github.com/me/x.git"); got != "ghp_abc" { + t.Errorf("URLSecret = %q, want ghp_abc", got) + } + if got := URLSecret("git@github.com:me/x.git"); got != "" { + t.Errorf("URLSecret = %q, want empty", got) + } +} + +func TestScrubRemovesSecrets(t *testing.T) { + got := scrub("fatal: auth failed for ghp_abc123", []string{"ghp_abc123"}) + if strings.Contains(got, "ghp_abc123") { + t.Errorf("secret leaked: %q", got) + } +} + +func TestShellQuote(t *testing.T) { + cases := []struct{ in, want string }{ + {"ssh", "ssh"}, + {"/etc/keys/id_ed25519", "/etc/keys/id_ed25519"}, + {"/keys/my key", `'/keys/my key'`}, + {"it's", `'it'\''s'`}, + {"StrictHostKeyChecking=accept-new", "StrictHostKeyChecking=accept-new"}, + } + for _, tc := range cases { + if got := shellQuote(tc.in); got != tc.want { + t.Errorf("shellQuote(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +// A deploy key mounted from a secret store is often world-readable, which ssh +// rejects outright. PrepareSSH must stage a 0600 copy instead of failing. +func TestPrepareSSHStagesPermissiveKey(t *testing.T) { + home, tmp := t.TempDir(), t.TempDir() + key := filepath.Join(t.TempDir(), "deploy_key") + if err := os.WriteFile(key, []byte("PRIVATE KEY"), 0o644); err != nil { + t.Fatal(err) + } + + cmd, err := PrepareSSH(SSH{KeyPath: key, StrictHostKey: "accept-new"}, home, tmp) + if err != nil { + t.Fatal(err) + } + if strings.Contains(cmd, key) { + t.Errorf("should use a staged copy, not the 0644 original: %s", cmd) + } + + staged := filepath.Join(tmp, "id_deploy_key") + fi, err := os.Stat(staged) + if err != nil { + t.Fatalf("staged copy missing: %v", err) + } + if perm := fi.Mode().Perm(); perm != 0o600 { + t.Errorf("staged key mode = %o, want 600", perm) + } + for _, want := range []string{"IdentitiesOnly=yes", "BatchMode=yes", "StrictHostKeyChecking=accept-new"} { + if !strings.Contains(cmd, want) { + t.Errorf("ssh command missing %q: %s", want, cmd) + } + } +} + +func TestPrepareSSHKeepsPrivateKeyPath(t *testing.T) { + home, tmp := t.TempDir(), t.TempDir() + key := filepath.Join(t.TempDir(), "deploy_key") + if err := os.WriteFile(key, []byte("PRIVATE KEY"), 0o600); err != nil { + t.Fatal(err) + } + + cmd, err := PrepareSSH(SSH{KeyPath: key}, home, tmp) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(cmd, key) { + t.Errorf("an already-private key should be used in place: %s", cmd) + } +} + +func TestPrepareSSHCreatesKnownHosts(t *testing.T) { + home, tmp := t.TempDir(), t.TempDir() + + if _, err := PrepareSSH(SSH{}, home, tmp); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(home, "known_hosts")); err != nil { + t.Errorf("known_hosts not created in HOME: %v", err) + } +} + +func TestEnvironIsHermetic(t *testing.T) { + t.Setenv("GIT_SSH_COMMAND", "ssh -i /attacker/key") + t.Setenv("GIT_CONFIG_GLOBAL", "/attacker/gitconfig") + t.Setenv("HTTPS_PROXY", "http://proxy.internal:3128") + + env := environ(Options{Home: "/var/lib/syncbot/home", SSHCmd: "ssh -o BatchMode=yes"}) + + got := map[string]string{} + for _, kv := range env { + if k, v, ok := strings.Cut(kv, "="); ok { + got[k] = v // later entries win, matching exec's behaviour + } + } + + if got["GIT_SSH_COMMAND"] != "ssh -o BatchMode=yes" { + t.Errorf("inherited GIT_SSH_COMMAND not overridden: %q", got["GIT_SSH_COMMAND"]) + } + if got["GIT_CONFIG_GLOBAL"] != os.DevNull { + t.Errorf("global git config not disabled: %q", got["GIT_CONFIG_GLOBAL"]) + } + if got["HOME"] != "/var/lib/syncbot/home" { + t.Errorf("HOME = %q", got["HOME"]) + } + if got["GIT_TERMINAL_PROMPT"] != "0" { + t.Error("git must never prompt for credentials") + } + if got["HTTPS_PROXY"] != "http://proxy.internal:3128" { + t.Error("proxy settings should be inherited") + } +} + +func TestLimitedWriterCaps(t *testing.T) { + var sb strings.Builder + w := &limitedWriter{W: &sb, N: 10} + + n, err := w.Write([]byte(strings.Repeat("x", 100))) + if err != nil { + t.Fatal(err) + } + if n != 100 { + t.Errorf("Write reported %d, want the full 100 so callers do not see a short write", n) + } + if sb.Len() != 10 { + t.Errorf("captured %d bytes, want the 10-byte cap", sb.Len()) + } +} diff --git a/internal/gitx/proc_other.go b/internal/gitx/proc_other.go new file mode 100644 index 0000000..d2f466f --- /dev/null +++ b/internal/gitx/proc_other.go @@ -0,0 +1,19 @@ +//go:build !unix + +package gitx + +import ( + "os" + "syscall" +) + +// sysProcAttr has no portable equivalent outside unix; the default is fine. +func sysProcAttr() *syscall.SysProcAttr { return nil } + +// terminate kills just the child. Helper processes it spawned may outlive it, +// but syncbot is deployed on Linux, where proc_unix.go handles this properly. +func terminate(pid int) { + if p, err := os.FindProcess(pid); err == nil { + _ = p.Kill() + } +} diff --git a/internal/gitx/proc_unix.go b/internal/gitx/proc_unix.go new file mode 100644 index 0000000..a88c016 --- /dev/null +++ b/internal/gitx/proc_unix.go @@ -0,0 +1,24 @@ +//go:build unix + +package gitx + +import ( + "syscall" + "time" +) + +// sysProcAttr puts git in its own process group so we can signal the whole +// tree — git itself plus the ssh or git-remote-https helper it spawned. +func sysProcAttr() *syscall.SysProcAttr { + return &syscall.SysProcAttr{Setpgid: true} +} + +// terminate asks the process group to exit, then insists. +func terminate(pid int) { + if pid <= 0 { + return + } + _ = syscall.Kill(-pid, syscall.SIGTERM) + time.Sleep(termGrace) + _ = syscall.Kill(-pid, syscall.SIGKILL) +} diff --git a/internal/manager/http.go b/internal/manager/http.go new file mode 100644 index 0000000..c551f39 --- /dev/null +++ b/internal/manager/http.go @@ -0,0 +1,152 @@ +package manager + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "strconv" + "strings" + "time" +) + +// Handler exposes the manager over HTTP: +// +// /healthz the process is alive (use this for a container health check) +// /readyz every repo has synced successfully at least once +// /status JSON snapshot of every repo +// /metrics Prometheus text format +func (m *Manager) Handler(version string) http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { + writeText(w, http.StatusOK, "ok\n") + }) + + mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) { + var pending []string + for _, s := range m.Statuses() { + if s.LastSuccess.IsZero() { + pending = append(pending, s.Name) + } + } + if len(pending) > 0 { + writeText(w, http.StatusServiceUnavailable, + "awaiting first successful sync: "+strings.Join(pending, ", ")+"\n") + return + } + writeText(w, http.StatusOK, "ready\n") + }) + + mux.HandleFunc("GET /status", func(w http.ResponseWriter, r *http.Request) { + statuses := m.Statuses() + body := struct { + Version string `json:"version"` + Uptime string `json:"uptime"` + Repos []Status `json:"repos"` + Healthy bool `json:"healthy"` + Failing int `json:"failing"` + Reported string `json:"reported_at"` + }{ + Version: version, + Uptime: m.Uptime().Round(time.Second).String(), + Repos: statuses, + Healthy: true, + Reported: time.Now().UTC().Format(time.RFC3339), + } + for _, s := range statuses { + if s.Failures > 0 { + body.Failing++ + body.Healthy = false + } + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(body) + }) + + mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + var b strings.Builder + fmt.Fprintf(&b, "# HELP syncbot_build_info Version of the running binary.\n") + fmt.Fprintf(&b, "# TYPE syncbot_build_info gauge\n") + fmt.Fprintf(&b, "syncbot_build_info{version=%q} 1\n", version) + fmt.Fprintf(&b, "# HELP syncbot_uptime_seconds Time since start.\n") + fmt.Fprintf(&b, "# TYPE syncbot_uptime_seconds gauge\n") + fmt.Fprintf(&b, "syncbot_uptime_seconds %s\n", seconds(m.Uptime())) + + metric(&b, "syncbot_sync_total", "counter", "Sync cycles started.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_sync_total{repo=%q} %d\n", s.Name, s.Syncs) + } + metric(&b, "syncbot_push_total", "counter", "Cycles that pushed to dst.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_push_total{repo=%q} %d\n", s.Name, s.Pushes) + } + metric(&b, "syncbot_consecutive_failures", "gauge", "Failed cycles since the last success.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_consecutive_failures{repo=%q} %d\n", s.Name, s.Failures) + } + metric(&b, "syncbot_refs", "gauge", "Mirrored refs.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_refs{repo=%q} %d\n", s.Name, s.Refs) + } + metric(&b, "syncbot_last_success_timestamp_seconds", "gauge", "Unix time of the last successful sync.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_last_success_timestamp_seconds{repo=%q} %d\n", s.Name, unix(s.LastSuccess)) + } + metric(&b, "syncbot_last_duration_seconds", "gauge", "Duration of the last sync cycle.") + for _, s := range m.Statuses() { + fmt.Fprintf(&b, "syncbot_last_duration_seconds{repo=%q} %s\n", s.Name, + strconv.FormatFloat(float64(s.LastDurMS)/1000, 'f', 3, 64)) + } + _, _ = w.Write([]byte(b.String())) + }) + + return mux +} + +// Serve runs the HTTP endpoint until ctx is cancelled. +func (m *Manager) Serve(ctx context.Context, addr, version string, log *slog.Logger) error { + srv := &http.Server{ + Addr: addr, + Handler: m.Handler(version), + ReadHeaderTimeout: 10 * time.Second, + } + go func() { + <-ctx.Done() + shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(shutdown) + }() + + log.Info("http listening", "addr", addr, + "endpoints", "/healthz /readyz /status /metrics") + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + return err + } + return nil +} + +func metric(b *strings.Builder, name, kind, help string) { + fmt.Fprintf(b, "# HELP %s %s\n# TYPE %s %s\n", name, help, name, kind) +} + +func seconds(d time.Duration) string { + return strconv.FormatFloat(d.Seconds(), 'f', 3, 64) +} + +func unix(t time.Time) int64 { + if t.IsZero() { + return 0 + } + return t.Unix() +} + +func writeText(w http.ResponseWriter, code int, body string) { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(code) + _, _ = w.Write([]byte(body)) +} diff --git a/internal/manager/manager.go b/internal/manager/manager.go new file mode 100644 index 0000000..8fbbd69 --- /dev/null +++ b/internal/manager/manager.go @@ -0,0 +1,412 @@ +// Package manager owns the running set of sync jobs. +// +// Each repository gets its own goroutine with its own timer, so a slow or +// broken repo never delays the others. A shared semaphore caps how many git +// processes run at once, which is what actually bounds memory use. +// +// Reloading is incremental: Apply diffs the new configuration against the +// running jobs and touches only what changed. Repos whose settings are +// unchanged keep their timer and their in-flight work. +package manager + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "reflect" + "sort" + "sync" + "time" + + "syncbot/internal/config" + "syncbot/internal/gitx" + "syncbot/internal/syncer" +) + +// Status is the observable state of one repository. +type Status struct { + Name string `json:"name"` + Src string `json:"src"` + Dst string `json:"dst"` + Interval string `json:"interval"` + Syncing bool `json:"syncing"` + Refs int `json:"refs"` + Syncs int64 `json:"syncs"` + Pushes int64 `json:"pushes"` + Failures int `json:"consecutive_failures"` + LastRun time.Time `json:"last_run,omitzero"` + LastSuccess time.Time `json:"last_success,omitzero"` + NextRun time.Time `json:"next_run,omitzero"` + LastError string `json:"last_error,omitempty"` + LastDurMS int64 `json:"last_duration_ms"` +} + +// Manager supervises one goroutine per repository. +type Manager struct { + ctx context.Context + log *slog.Logger + + mu sync.Mutex + jobs map[string]*job + stats map[string]*Status + sem chan struct{} + conc int + workDir string + sync *syncer.Syncer + started time.Time + + // wg covers every goroutine ever started, including ones already replaced + // by a reload, so shutdown does not leave a git process behind. + wg sync.WaitGroup +} + +type job struct { + spec config.Job + cancel context.CancelFunc + done chan struct{} +} + +// New returns a Manager whose jobs all derive from ctx. +func New(ctx context.Context, log *slog.Logger) *Manager { + return &Manager{ + ctx: ctx, + log: log, + jobs: make(map[string]*job), + stats: make(map[string]*Status), + sem: make(chan struct{}, 1), + conc: 1, + started: time.Now(), + } +} + +// Apply reconciles the running jobs with cfg. It never blocks on in-flight +// syncs: a job being replaced is cancelled, and its successor waits for the +// handover before touching the same mirror directory. +func (m *Manager) Apply(cfg *config.Config) error { + home, err := prepareDirs(cfg.WorkDir) + if err != nil { + return err + } + + if cfg.Concurrency <= 0 { + cfg.Concurrency = 1 + } + + m.mu.Lock() + + if m.conc != cfg.Concurrency { + // Callers release into the channel they acquired from, so swapping the + // semaphore is safe even while syncs are in flight. + m.log.Info("concurrency changed", "from", m.conc, "to", cfg.Concurrency) + m.sem = make(chan struct{}, cfg.Concurrency) + m.conc = cfg.Concurrency + } + + workDirChanged := m.workDir != cfg.WorkDir + if workDirChanged { + m.workDir = cfg.WorkDir + m.sync = &syncer.Syncer{Log: m.log, Home: home} + } + + want := make(map[string]config.Job, len(cfg.Jobs)) + for _, j := range cfg.Jobs { + want[j.Name] = j + } + + // Stop jobs that disappeared or whose resolved spec changed. + var stopped []*job + for name, j := range m.jobs { + w, keep := want[name] + if keep && !workDirChanged && reflect.DeepEqual(w, j.spec) { + continue + } + delete(m.jobs, name) + stopped = append(stopped, j) + if !keep { + delete(m.stats, name) + m.log.Info("repo removed", "repo", name) + } + } + handover := make(map[string]<-chan struct{}, len(stopped)) + for _, j := range stopped { + j.cancel() + handover[j.spec.Name] = j.done + } + + // Start whatever is now missing. + var added, restarted int + for i, spec := range cfg.Jobs { + if _, running := m.jobs[spec.Name]; running { + continue + } + prev := handover[spec.Name] + if prev != nil { + restarted++ + } else { + added++ + } + m.startLocked(spec, prev, stagger(i)) + } + m.mu.Unlock() + + if added+restarted+len(stopped) > 0 { + m.log.Info("configuration applied", + "repos", len(cfg.Jobs), "started", added, "restarted", restarted, + "stopped", len(stopped)-restarted, "concurrency", cfg.Concurrency) + } + return nil +} + +// startLocked launches a job goroutine. m.mu must be held. +func (m *Manager) startLocked(spec config.Job, waitFor <-chan struct{}, delay time.Duration) { + ctx, cancel := context.WithCancel(m.ctx) + j := &job{spec: spec, cancel: cancel, done: make(chan struct{})} + m.jobs[spec.Name] = j + + st, ok := m.stats[spec.Name] + if !ok { + st = &Status{Name: spec.Name} + m.stats[spec.Name] = st + } + // Counters survive a restart; the descriptive fields follow the new spec. + st.Src = gitx.RedactURL(spec.Src.URL) + st.Dst = gitx.RedactURL(spec.Dst.URL) + st.Interval = spec.Interval.String() + + m.wg.Add(1) + go m.loop(ctx, j, waitFor, delay) +} + +func (m *Manager) loop(ctx context.Context, j *job, waitFor <-chan struct{}, delay time.Duration) { + defer m.wg.Done() + defer close(j.done) + + // When replacing a job, let the previous goroutine finish unwinding before + // operating on its mirror directory. + if waitFor != nil { + select { + case <-waitFor: + case <-ctx.Done(): + return + } + } + + log := m.log.With("repo", j.spec.Name) + log.Info("watching", "src", gitx.RedactURL(j.spec.Src.URL), + "dst", gitx.RedactURL(j.spec.Dst.URL), "interval", j.spec.Interval) + + timer := time.NewTimer(delay) + defer timer.Stop() + + for { + m.update(j.spec.Name, func(s *Status) { s.NextRun = time.Now().Add(delay) }) + + select { + case <-ctx.Done(): + return + case <-timer.C: + } + + failures := m.runOnce(ctx, j.spec, log) + if ctx.Err() != nil { + return + } + + delay = j.spec.Interval + if failures > 0 { + delay = backoff(j.spec.Interval, j.spec.MaxBackoff, failures) + if delay > j.spec.Interval { + log.Warn("backing off", "failures", failures, "retry_in", delay) + } + } + timer.Reset(delay) + } +} + +// runOnce performs a single sync and returns the consecutive failure count. +func (m *Manager) runOnce(ctx context.Context, spec config.Job, log *slog.Logger) int { + release, err := m.acquire(ctx) + if err != nil { + return 0 // shutting down + } + defer release() + + m.update(spec.Name, func(s *Status) { + s.Syncing = true + s.LastRun = time.Now() + s.Syncs++ + }) + + res, err := m.syncer().Sync(ctx, spec) + + failures := 0 + m.update(spec.Name, func(s *Status) { + s.Syncing = false + s.LastDurMS = res.Duration.Milliseconds() + if err != nil { + s.Failures++ + s.LastError = err.Error() + } else { + s.Failures = 0 + s.LastError = "" + s.LastSuccess = time.Now() + s.Refs = res.Refs + if res.Pushed { + s.Pushes++ + } + } + failures = s.Failures + }) + + switch { + case err != nil && ctx.Err() != nil: + log.Info("sync cancelled") + case err != nil: + log.Error("sync failed", "err", err, "consecutive_failures", failures) + case res.Pushed: + log.Info("pushed", "refs", res.Refs, "changes", len(res.Changes), + "dur", res.Duration.Round(time.Millisecond)) + for _, c := range res.Changes { + log.Debug("ref updated", "change", c) + } + } + return failures +} + +// RunOnce syncs every job exactly once and reports whether all succeeded. It +// is the engine behind the -once flag, for cron-style or one-shot deployments. +func (m *Manager) RunOnce(cfg *config.Config) error { + if err := m.Apply(&config.Config{ // reuse Apply's dir/semaphore setup, no jobs + WorkDir: cfg.WorkDir, Concurrency: cfg.Concurrency, + ReloadInterval: cfg.ReloadInterval, LogLevel: cfg.LogLevel, LogFormat: cfg.LogFormat, + }); err != nil { + return err + } + + var wg sync.WaitGroup + errs := make([]error, len(cfg.Jobs)) + for i, spec := range cfg.Jobs { + wg.Add(1) + go func() { + defer wg.Done() + release, err := m.acquire(m.ctx) + if err != nil { + errs[i] = err + return + } + defer release() + + log := m.log.With("repo", spec.Name) + res, err := m.syncer().Sync(m.ctx, spec) + if err != nil { + errs[i] = fmt.Errorf("%s: %w", spec.Name, err) + log.Error("sync failed", "err", err) + return + } + log.Info("synced", "refs", res.Refs, "pushed", res.Pushed, + "dur", res.Duration.Round(time.Millisecond)) + }() + } + wg.Wait() + return errors.Join(errs...) +} + +// Stop cancels every job and waits up to grace for them to unwind. No further +// Apply may be called afterwards. +func (m *Manager) Stop(grace time.Duration) { + m.mu.Lock() + for _, j := range m.jobs { + j.cancel() + } + m.jobs = make(map[string]*job) + m.mu.Unlock() + + // Waiting on the group rather than on the current jobs also covers + // goroutines a recent reload replaced but that have not finished unwinding. + done := make(chan struct{}) + go func() { + m.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(grace): + m.log.Warn("shutdown grace period expired; abandoning in-flight syncs", "grace", grace) + } +} + +// Statuses returns a snapshot of every known repository, sorted by name. +func (m *Manager) Statuses() []Status { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]Status, 0, len(m.stats)) + for _, s := range m.stats { + out = append(out, *s) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} + +// Uptime reports how long the manager has been running. +func (m *Manager) Uptime() time.Duration { return time.Since(m.started) } + +// acquire takes a slot from the concurrency semaphore. The release function +// returns the slot to the same channel it came from, which keeps the count +// correct even if Apply swapped the semaphore in the meantime. +func (m *Manager) acquire(ctx context.Context) (func(), error) { + m.mu.Lock() + sem := m.sem + m.mu.Unlock() + + select { + case sem <- struct{}{}: + return func() { <-sem }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func (m *Manager) syncer() *syncer.Syncer { + m.mu.Lock() + defer m.mu.Unlock() + return m.sync +} + +func (m *Manager) update(name string, fn func(*Status)) { + m.mu.Lock() + defer m.mu.Unlock() + if s, ok := m.stats[name]; ok { + fn(s) + } +} + +// prepareDirs creates the mirror store and the isolated HOME git/ssh will use. +func prepareDirs(workDir string) (home string, err error) { + home = filepath.Join(workDir, "home") + for _, d := range []string{filepath.Join(workDir, "mirrors"), home} { + if err := os.MkdirAll(d, 0o700); err != nil { + return "", fmt.Errorf("prepare %s: %w", d, err) + } + } + return home, nil +} + +// backoff grows the retry delay geometrically, capped at max, so a repo that +// is down for hours stops hammering it (and our logs) every interval. +func backoff(interval, max time.Duration, failures int) time.Duration { + d := interval + for i := 1; i < failures && d < max; i++ { + d *= 2 + } + return min(d, max) +} + +// stagger spreads the first run of each repo so a restart with many repos does +// not fire every git process in the same instant. +func stagger(index int) time.Duration { + return min(time.Duration(index)*250*time.Millisecond, 15*time.Second) +} diff --git a/internal/manager/manager_test.go b/internal/manager/manager_test.go new file mode 100644 index 0000000..3875d8c --- /dev/null +++ b/internal/manager/manager_test.go @@ -0,0 +1,243 @@ +package manager + +import ( + "context" + "io" + "log/slog" + "os" + "path/filepath" + "testing" + "time" + + "syncbot/internal/config" +) + +func testConfig(workDir string, jobs ...config.Job) *config.Config { + return &config.Config{ + WorkDir: workDir, + Concurrency: 2, + ReloadInterval: time.Second, + Jobs: jobs, + } +} + +// newJob builds a spec pointing at paths that do not exist. The job goroutine +// will fail its sync quickly and harmlessly, which is all these tests need — +// they are about supervision, not about git. +func newJob(name, workDir string, interval time.Duration) config.Job { + return config.Job{ + Name: name, + Src: config.Endpoint{URL: filepath.Join(workDir, name+"-src.git")}, + Dst: config.Endpoint{URL: filepath.Join(workDir, name+"-dst.git")}, + Dir: filepath.Join(workDir, "mirrors", name+".git"), + Interval: interval, + Timeout: 5 * time.Second, + MaxBackoff: time.Hour, + Refs: config.DefaultRefs, + Prune: true, + Force: true, + } +} + +func newTestManager(t *testing.T) (*Manager, string) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + workDir := t.TempDir() + m := New(ctx, slog.New(slog.NewTextHandler(io.Discard, nil))) + t.Cleanup(func() { m.Stop(5 * time.Second) }) + return m, workDir +} + +func (m *Manager) jobPointers() map[string]*job { + m.mu.Lock() + defer m.mu.Unlock() + out := make(map[string]*job, len(m.jobs)) + for k, v := range m.jobs { + out[k] = v + } + return out +} + +func TestApplyStartsJobs(t *testing.T) { + m, dir := newTestManager(t) + + if err := m.Apply(testConfig(dir, newJob("a", dir, time.Hour), newJob("b", dir, time.Hour))); err != nil { + t.Fatal(err) + } + + if got := len(m.jobPointers()); got != 2 { + t.Fatalf("running jobs = %d, want 2", got) + } + if got := len(m.Statuses()); got != 2 { + t.Errorf("statuses = %d, want 2", got) + } +} + +// The whole point of an incremental reload: editing one repo must not disturb +// the others' timers or in-flight work. +func TestApplyLeavesUnchangedJobsRunning(t *testing.T) { + m, dir := newTestManager(t) + cfg := testConfig(dir, newJob("a", dir, time.Hour), newJob("b", dir, time.Hour)) + if err := m.Apply(cfg); err != nil { + t.Fatal(err) + } + before := m.jobPointers() + + // Identical configuration: nothing should be touched. + if err := m.Apply(testConfig(dir, newJob("a", dir, time.Hour), newJob("b", dir, time.Hour))); err != nil { + t.Fatal(err) + } + after := m.jobPointers() + for name, j := range before { + if after[name] != j { + t.Errorf("job %q was restarted despite an identical config", name) + } + } + + // Change only "a": "b" must survive untouched. + if err := m.Apply(testConfig(dir, newJob("a", dir, 30*time.Minute), newJob("b", dir, time.Hour))); err != nil { + t.Fatal(err) + } + changed := m.jobPointers() + if changed["a"] == before["a"] { + t.Error("job \"a\" should have been restarted after its interval changed") + } + if changed["b"] != before["b"] { + t.Error("job \"b\" should not have been restarted") + } + if got := changed["a"].spec.Interval; got != 30*time.Minute { + t.Errorf("restarted job carries interval %s, want 30m", got) + } +} + +func TestApplyStopsRemovedJobs(t *testing.T) { + m, dir := newTestManager(t) + if err := m.Apply(testConfig(dir, newJob("a", dir, time.Hour), newJob("b", dir, time.Hour))); err != nil { + t.Fatal(err) + } + removed := m.jobPointers()["b"] + + if err := m.Apply(testConfig(dir, newJob("a", dir, time.Hour))); err != nil { + t.Fatal(err) + } + + if _, still := m.jobPointers()["b"]; still { + t.Fatal("job \"b\" is still registered after being removed from the config") + } + select { + case <-removed.done: + case <-time.After(5 * time.Second): + t.Fatal("goroutine for the removed job did not exit") + } + for _, s := range m.Statuses() { + if s.Name == "b" { + t.Error("status for the removed job should be dropped") + } + } +} + +func TestCountersSurviveJobRestart(t *testing.T) { + m, dir := newTestManager(t) + if err := m.Apply(testConfig(dir, newJob("a", dir, time.Hour))); err != nil { + t.Fatal(err) + } + + m.update("a", func(s *Status) { s.Syncs, s.Pushes = 7, 3 }) + + if err := m.Apply(testConfig(dir, newJob("a", dir, 5*time.Minute))); err != nil { + t.Fatal(err) + } + + got := m.Statuses()[0] + if got.Syncs != 7 || got.Pushes != 3 { + t.Errorf("counters reset on restart: syncs=%d pushes=%d, want 7/3", got.Syncs, got.Pushes) + } + if got.Interval != "5m0s" { + t.Errorf("status interval = %q, want it to follow the new spec", got.Interval) + } +} + +func TestConcurrencyChangeKeepsSemaphoreConsistent(t *testing.T) { + m, dir := newTestManager(t) + cfg := testConfig(dir, newJob("a", dir, time.Hour)) + cfg.Concurrency = 1 + if err := m.Apply(cfg); err != nil { + t.Fatal(err) + } + + // Hold a slot from the old semaphore, then resize. + release, err := m.acquire(context.Background()) + if err != nil { + t.Fatal(err) + } + + cfg2 := testConfig(dir, newJob("a", dir, time.Hour)) + cfg2.Concurrency = 4 + if err := m.Apply(cfg2); err != nil { + t.Fatal(err) + } + release() // must return the slot to the channel it came from, not the new one + + // The new semaphore should have its full capacity available. + var releases []func() + for i := 0; i < 4; i++ { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + r, err := m.acquire(ctx) + cancel() + if err != nil { + t.Fatalf("acquire %d/4 blocked after resize: %v", i+1, err) + } + releases = append(releases, r) + } + for _, r := range releases { + r() + } +} + +func TestApplyRejectsUnusableWorkDir(t *testing.T) { + m, dir := newTestManager(t) + // A regular file cannot host the mirror directories. + blocked := filepath.Join(dir, "not-a-dir") + if err := os.WriteFile(blocked, []byte("x"), 0o600); err != nil { + t.Fatal(err) + } + + if err := m.Apply(testConfig(blocked, newJob("a", dir, time.Hour))); err == nil { + t.Fatal("want an error for a work_dir that cannot be created") + } +} + +func TestBackoff(t *testing.T) { + const interval, max = time.Minute, 30 * time.Minute + cases := []struct { + failures int + want time.Duration + }{ + {0, time.Minute}, + {1, time.Minute}, + {2, 2 * time.Minute}, + {3, 4 * time.Minute}, + {4, 8 * time.Minute}, + {6, 30 * time.Minute}, // capped + {100, 30 * time.Minute}, + } + for _, tc := range cases { + if got := backoff(interval, max, tc.failures); got != tc.want { + t.Errorf("backoff(failures=%d) = %s, want %s", tc.failures, got, tc.want) + } + } +} + +func TestStaggerIsBounded(t *testing.T) { + if got := stagger(0); got != 0 { + t.Errorf("the first repo should start immediately, got %s", got) + } + if got := stagger(3); got != 750*time.Millisecond { + t.Errorf("stagger(3) = %s, want 750ms", got) + } + if got := stagger(1000); got != 15*time.Second { + t.Errorf("stagger should be capped at 15s, got %s", got) + } +} diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go new file mode 100644 index 0000000..c3ba4b1 --- /dev/null +++ b/internal/syncer/syncer.go @@ -0,0 +1,163 @@ +// Package syncer performs a single src -> dst mirror cycle. +// +// The design is deliberately stateless: nothing is remembered between runs. +// Each cycle asks both remotes what they currently hold and does only the work +// needed to make them agree. That makes the bot self-healing — if someone +// force-pushes the destination, or a push half-fails, or the container is +// rebuilt from scratch, the next cycle simply notices and converges. +package syncer + +import ( + "context" + "fmt" + "log/slog" + "os" + "time" + + "syncbot/internal/config" + "syncbot/internal/gitx" +) + +// Result summarises what a cycle did, for logging and metrics. +type Result struct { + Fetched bool // objects were pulled from src + Pushed bool // refs were written to dst + Refs int // number of managed refs after the cycle + Changes []string // porcelain lines for the refs that moved + Duration time.Duration +} + +// Syncer runs mirror cycles. It holds no per-repo state and is safe for +// concurrent use. +type Syncer struct { + Log *slog.Logger + Home string // isolated HOME for git/ssh; must exist and be writable +} + +// Sync brings job.Dst in line with job.Src. +func (s *Syncer) Sync(ctx context.Context, job config.Job) (Result, error) { + start := time.Now() + var res Result + + ctx, cancel := context.WithTimeout(ctx, job.Timeout) + defer cancel() + + // Staged deploy-key copies live here and are shredded when we are done. + tmp, err := os.MkdirTemp(s.Home, "ssh-") + if err != nil { + return res, fmt.Errorf("create temp dir: %w", err) + } + defer os.RemoveAll(tmp) + + log := s.Log.With("repo", job.Name) + secrets := []string{gitx.URLSecret(job.Src.URL), gitx.URLSecret(job.Dst.URL)} + + srcOpts, err := s.options(job.Src, job.GitConfig, tmp, secrets, log) + if err != nil { + return res, fmt.Errorf("src: %w", err) + } + dstOpts, err := s.options(job.Dst, job.GitConfig, tmp, secrets, log) + if err != nil { + return res, fmt.Errorf("dst: %w", err) + } + + if err := gitx.EnsureMirror(ctx, srcOpts, job.Dir); err != nil { + return res, err + } + + // 1. Ask src what it has. This is the cheap poll that runs every interval. + srcRefs, err := gitx.LsRemote(ctx, srcOpts, job.Dir, job.Src.URL, job.Refs) + if err != nil { + return res, fmt.Errorf("read src %s: %w", gitx.RedactURL(job.Src.URL), err) + } + + localRefs, err := gitx.LocalRefs(ctx, srcOpts, job.Dir, job.Refs) + if err != nil { + return res, fmt.Errorf("read mirror: %w", err) + } + + // 2. Only transfer objects when the mirror is actually behind. + if !srcRefs.Equal(localRefs) { + log.Info("fetching", "src", gitx.RedactURL(job.Src.URL), + "local_refs", len(localRefs), "src_refs", len(srcRefs)) + if err := gitx.Fetch(ctx, srcOpts, job.Dir, job.Src.URL, job.Refs); err != nil { + return res, fmt.Errorf("fetch from %s: %w", gitx.RedactURL(job.Src.URL), err) + } + res.Fetched = true + if localRefs, err = gitx.LocalRefs(ctx, srcOpts, job.Dir, job.Refs); err != nil { + return res, fmt.Errorf("read mirror after fetch: %w", err) + } + } + res.Refs = len(localRefs) + + // 3. Ask dst what it has, so external drift is detected too. + dstRefs, err := gitx.LsRemote(ctx, dstOpts, job.Dir, job.Dst.URL, job.Refs) + if err != nil { + return res, fmt.Errorf("read dst %s: %w", gitx.RedactURL(job.Dst.URL), err) + } + + if !needsPush(localRefs, dstRefs, job.Prune) { + res.Duration = time.Since(start) + log.Debug("already in sync", "refs", res.Refs, "dur", res.Duration.Round(time.Millisecond)) + return res, nil + } + + // Safety net: an upstream that suddenly reports zero refs is far more + // likely to be a broken URL or a revoked token than a genuine wipe, and + // pushing that through with --prune would delete the destination. + if len(localRefs) == 0 && len(dstRefs) > 0 && !job.AllowEmpty { + return res, fmt.Errorf("refusing to mirror an empty source over %d ref(s) on dst; "+ + "set allow_empty = true if this is intended", len(dstRefs)) + } + + log.Info("pushing", "dst", gitx.RedactURL(job.Dst.URL), "refs", res.Refs) + changes, err := gitx.Push(ctx, dstOpts, job.Dir, job.Dst.URL, job.Refs, gitx.PushOptions{ + Prune: job.Prune, + Force: job.Force, + Atomic: job.Atomic, + }) + if err != nil { + return res, fmt.Errorf("push to %s: %w", gitx.RedactURL(job.Dst.URL), err) + } + res.Pushed = true + res.Changes = changes + res.Duration = time.Since(start) + return res, nil +} + +// options builds the git invocation environment for one endpoint. +func (s *Syncer) options(e config.Endpoint, gitConfig []string, tmp string, secrets []string, log *slog.Logger) (gitx.Options, error) { + o := gitx.Options{ + Log: log, + Home: s.Home, + GitConfig: gitConfig, + Secrets: secrets, + } + sshCmd, err := gitx.PrepareSSH(gitx.SSH{ + KeyPath: e.SSHKey, + KnownHosts: e.KnownHosts, + StrictHostKey: e.StrictHostKey, + }, s.Home, tmp) + if err != nil { + return o, err + } + o.SSHCmd = sshCmd + return o, nil +} + +// needsPush reports whether dst differs from the mirror in any way we manage. +func needsPush(local, dst gitx.Refs, prune bool) bool { + for ref, obj := range local { + if dst[ref] != obj { + return true + } + } + if prune { + for ref := range dst { + if _, ok := local[ref]; !ok { + return true + } + } + } + return false +} diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go new file mode 100644 index 0000000..3cbb7b6 --- /dev/null +++ b/internal/syncer/syncer_test.go @@ -0,0 +1,322 @@ +package syncer + +import ( + "context" + "io" + "log/slog" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "syncbot/internal/config" + "syncbot/internal/gitx" +) + +// harness wires up a real src repo, a real bare dst repo and a Syncer, so the +// tests exercise the actual git plumbing rather than a mock of it. +type harness struct { + t *testing.T + src string + dst string + job config.Job + sync *Syncer +} + +func newHarness(t *testing.T) *harness { + t.Helper() + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + + root := t.TempDir() + src := filepath.Join(root, "src") + dst := filepath.Join(root, "dst.git") + home := filepath.Join(root, "home") + for _, d := range []string{src, dst, home} { + if err := os.MkdirAll(d, 0o700); err != nil { + t.Fatal(err) + } + } + + git(t, src, "init", "--quiet", "-b", "main") + git(t, dst, "init", "--bare", "--quiet", "-b", "main") + // A bare repo refuses to have the branch its HEAD points at deleted, which + // would otherwise make the allow_empty case untestable. GitHub behaves the + // same way for its default branch; see the README's troubleshooting notes. + git(t, dst, "config", "receive.denyDeleteCurrent", "ignore") + + h := &harness{ + t: t, + src: src, + dst: dst, + job: config.Job{ + Name: "test", + Src: config.Endpoint{URL: src}, + Dst: config.Endpoint{URL: dst}, + Dir: filepath.Join(root, "mirror.git"), + Interval: time.Minute, + Timeout: 2 * time.Minute, + MaxBackoff: time.Minute, + Refs: config.DefaultRefs, + Prune: true, + Force: true, + }, + sync: &Syncer{Log: slog.New(slog.NewTextHandler(io.Discard, nil)), Home: home}, + } + h.commit("first") + return h +} + +func (h *harness) commit(msg string) { + h.t.Helper() + path := filepath.Join(h.src, "file.txt") + if err := os.WriteFile(path, []byte(msg+"\n"), 0o600); err != nil { + h.t.Fatal(err) + } + git(h.t, h.src, "add", "-A") + git(h.t, h.src, "commit", "--quiet", "-m", msg) +} + +func (h *harness) run() Result { + h.t.Helper() + res, err := h.sync.Sync(context.Background(), h.job) + if err != nil { + h.t.Fatalf("sync: %v", err) + } + return res +} + +func (h *harness) refs(dir string) gitx.Refs { + h.t.Helper() + out := git(h.t, dir, "for-each-ref", "--format=%(objectname)\t%(refname)") + refs := gitx.Refs{} + for _, line := range strings.Split(out, "\n") { + if obj, ref, ok := strings.Cut(strings.TrimSpace(line), "\t"); ok { + refs[ref] = obj + } + } + return refs +} + +// assertMirrored checks that dst holds exactly what src holds. +func (h *harness) assertMirrored() { + h.t.Helper() + src, dst := h.refs(h.src), h.refs(h.dst) + if !src.Equal(dst) { + h.t.Fatalf("dst does not mirror src\n src: %v\n dst: %v", src, dst) + } +} + +func TestFirstSyncCopiesEverything(t *testing.T) { + h := newHarness(t) + git(t, h.src, "tag", "-a", "v1.0.0", "-m", "release") + git(t, h.src, "branch", "feature/x") + + res := h.run() + if !res.Fetched || !res.Pushed { + t.Fatalf("want fetch and push on first sync, got %+v", res) + } + h.assertMirrored() + + dst := h.refs(h.dst) + for _, want := range []string{"refs/heads/main", "refs/heads/feature/x", "refs/tags/v1.0.0"} { + if _, ok := dst[want]; !ok { + t.Errorf("dst missing %s (has %v)", want, dst) + } + } +} + +func TestNoChangesDoesNothing(t *testing.T) { + h := newHarness(t) + h.run() + + res := h.run() + if res.Fetched { + t.Error("fetched despite src being unchanged") + } + if res.Pushed { + t.Error("pushed despite dst already being in sync") + } +} + +func TestNewCommitPropagates(t *testing.T) { + h := newHarness(t) + h.run() + + h.commit("second") + res := h.run() + if !res.Fetched || !res.Pushed { + t.Fatalf("want fetch and push after a new commit, got %+v", res) + } + h.assertMirrored() +} + +func TestDeletedBranchIsPruned(t *testing.T) { + h := newHarness(t) + git(t, h.src, "branch", "temp") + h.run() + if _, ok := h.refs(h.dst)["refs/heads/temp"]; !ok { + t.Fatal("setup: dst should have refs/heads/temp") + } + + git(t, h.src, "branch", "-D", "temp") + h.run() + + if _, ok := h.refs(h.dst)["refs/heads/temp"]; ok { + t.Error("refs/heads/temp still on dst after being deleted upstream") + } + h.assertMirrored() +} + +func TestForcePushAfterRewrite(t *testing.T) { + h := newHarness(t) + h.commit("second") + h.run() + + // Rewrite history the way a rebase or an amended commit would. + git(t, h.src, "reset", "--hard", "--quiet", "HEAD~1") + h.commit("rewritten") + res := h.run() + + if !res.Pushed { + t.Fatal("want a push after history was rewritten") + } + h.assertMirrored() +} + +// The bot keeps no state between runs, so damage done directly to dst must heal +// on the next cycle even though src has not moved. +func TestDestinationDriftIsRepaired(t *testing.T) { + h := newHarness(t) + git(t, h.src, "branch", "keep") + h.run() + + git(t, h.dst, "update-ref", "-d", "refs/heads/keep") + if _, ok := h.refs(h.dst)["refs/heads/keep"]; ok { + t.Fatal("setup: refs/heads/keep should be gone from dst") + } + + res := h.run() + if res.Fetched { + t.Error("fetched even though src had not changed") + } + if !res.Pushed { + t.Fatal("want a push to repair dst") + } + h.assertMirrored() +} + +func TestEmptySourceIsRefused(t *testing.T) { + h := newHarness(t) + h.run() + + // Simulate a source that answers but has nothing to offer — a revoked + // token or a wrong URL looks exactly like this. + empty := filepath.Join(t.TempDir(), "empty.git") + git(t, t.TempDir(), "init", "--bare", "--quiet", empty) + h.job.Src = config.Endpoint{URL: empty} + + _, err := h.sync.Sync(context.Background(), h.job) + if err == nil { + t.Fatal("want an error when an empty source would wipe dst") + } + if !strings.Contains(err.Error(), "allow_empty") { + t.Errorf("error should point at allow_empty, got: %v", err) + } + if len(h.refs(h.dst)) == 0 { + t.Error("dst was wiped despite the guard") + } + + // With the guard lifted the wipe goes through, as documented. + h.job.AllowEmpty = true + if _, err := h.sync.Sync(context.Background(), h.job); err != nil { + t.Fatalf("sync with allow_empty: %v", err) + } + if n := len(h.refs(h.dst)); n != 0 { + t.Errorf("dst should be empty with allow_empty = true, has %d refs", n) + } +} + +func TestRefsFilterLimitsWhatIsMirrored(t *testing.T) { + h := newHarness(t) + git(t, h.src, "tag", "v1") + h.job.Refs = []string{"refs/heads/*"} + + h.run() + + dst := h.refs(h.dst) + if _, ok := dst["refs/heads/main"]; !ok { + t.Error("branches should be mirrored") + } + if _, ok := dst["refs/tags/v1"]; ok { + t.Error("tags should not be mirrored when refs excludes them") + } +} + +func TestUnreachableSourceReportsError(t *testing.T) { + h := newHarness(t) + h.job.Src = config.Endpoint{URL: filepath.Join(t.TempDir(), "does-not-exist.git")} + h.job.Timeout = 30 * time.Second + + if _, err := h.sync.Sync(context.Background(), h.job); err == nil { + t.Fatal("want an error for an unreachable source") + } +} + +func TestTimeoutIsEnforced(t *testing.T) { + h := newHarness(t) + h.job.Timeout = time.Nanosecond + + _, err := h.sync.Sync(context.Background(), h.job) + if err == nil { + t.Fatal("want an error when the timeout expires") + } +} + +func TestNeedsPush(t *testing.T) { + local := gitx.Refs{"refs/heads/main": "aaa"} + + cases := []struct { + name string + local gitx.Refs + dst gitx.Refs + prune bool + want bool + }{ + {"identical", local, gitx.Refs{"refs/heads/main": "aaa"}, true, false}, + {"moved", local, gitx.Refs{"refs/heads/main": "bbb"}, true, true}, + {"missing on dst", local, gitx.Refs{}, true, true}, + {"extra on dst, pruning", local, gitx.Refs{"refs/heads/main": "aaa", "refs/heads/x": "c"}, true, true}, + {"extra on dst, not pruning", local, gitx.Refs{"refs/heads/main": "aaa", "refs/heads/x": "c"}, false, false}, + {"both empty", gitx.Refs{}, gitx.Refs{}, true, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := needsPush(tc.local, tc.dst, tc.prune); got != tc.want { + t.Errorf("needsPush = %v, want %v", got, tc.want) + } + }) + } +} + +func git(t *testing.T, dir string, args ...string) string { + t.Helper() + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), + "GIT_CONFIG_GLOBAL="+os.DevNull, + "GIT_CONFIG_SYSTEM="+os.DevNull, + "GIT_AUTHOR_NAME=syncbot test", + "GIT_AUTHOR_EMAIL=test@example.invalid", + "GIT_COMMITTER_NAME=syncbot test", + "GIT_COMMITTER_EMAIL=test@example.invalid", + ) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %s: %v\n%s", strings.Join(args, " "), err, out) + } + return string(out) +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..fd4a59f --- /dev/null +++ b/main.go @@ -0,0 +1,262 @@ +// Command syncbot mirrors git repositories from a source to a destination on a +// timer. See README.md for the configuration format. +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "flag" + "fmt" + "io" + "log/slog" + "os" + "os/signal" + "runtime/debug" + "syscall" + "time" + + "syncbot/internal/config" + "syncbot/internal/gitx" + "syncbot/internal/manager" +) + +// version is stamped at build time with -ldflags "-X main.version=...". +var version = "dev" + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "syncbot: "+err.Error()) + os.Exit(1) + } +} + +func run() error { + var ( + cfgPath = flag.String("config", "/etc/syncbot/config.toml", "path to the TOML config file") + check = flag.Bool("check", false, "validate the config file and exit") + once = flag.Bool("once", false, "sync every repository once, then exit") + showVer = flag.Bool("version", false, "print the version and exit") + ) + flag.Parse() + + if *showVer { + fmt.Println("syncbot", buildVersion()) + return nil + } + + cfg, err := config.Load(*cfgPath) + if err != nil { + return err + } + if *check { + printSummary(cfg, *cfgPath) + return nil + } + + level := new(slog.LevelVar) + log := newLogger(cfg, level, os.Stderr) + log.Info("starting", "version", buildVersion(), "config", *cfgPath, + "work_dir", cfg.WorkDir, "repos", len(cfg.Jobs)) + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + mgr := manager.New(ctx, log) + + if *once { + return mgr.RunOnce(cfg) + } + if err := mgr.Apply(cfg); err != nil { + return err + } + + if cfg.Listen != "" { + go func() { + if err := mgr.Serve(ctx, cfg.Listen, buildVersion(), log); err != nil { + log.Error("http server stopped", "err", err) + } + }() + } + + go watchConfig(ctx, *cfgPath, cfg, level, log, mgr) + + <-ctx.Done() + log.Info("shutting down") + mgr.Stop(20 * time.Second) + log.Info("stopped") + return nil +} + +// watchConfig re-reads the config file whenever its contents change, and on +// SIGHUP. A file that fails to parse is reported and ignored — the daemon keeps +// running on the last configuration that worked, which is what you want when a +// typo lands in production at 3am. +func watchConfig(ctx context.Context, path string, initial *config.Config, level *slog.LevelVar, log *slog.Logger, mgr *manager.Manager) { + digest, _ := fileDigest(path) + current := initial + + // Note: signal.Notify with no signals subscribes to *every* signal, so only + // register when the platform actually has a reload signal to offer. + hup := make(chan os.Signal, 1) + if sigs := reloadSignals(); len(sigs) > 0 { + signal.Notify(hup, sigs...) + defer signal.Stop(hup) + } + + ticker := time.NewTicker(current.ReloadInterval) + defer ticker.Stop() + + for { + forced := false + select { + case <-ctx.Done(): + return + case <-ticker.C: + case <-hup: + forced = true + log.Info("reload requested by signal") + } + + d, err := fileDigest(path) + if err != nil { + log.Error("cannot read config", "path", path, "err", err) + continue + } + if d == digest && !forced { + continue + } + // Record the digest even on failure so a broken edit is reported once + // rather than on every tick. + digest = d + + cfg, err := config.Load(path) + if err != nil { + log.Error("config reload failed, keeping the previous configuration", "err", err) + continue + } + + if cfg.LogLevel != current.LogLevel { + level.Set(parseLevel(cfg.LogLevel)) + log.Info("log level changed", "level", cfg.LogLevel) + } + if cfg.Listen != current.Listen { + log.Warn("listen address changed; this takes effect after a restart", + "current", current.Listen, "configured", cfg.Listen) + } + if cfg.ReloadInterval != current.ReloadInterval { + ticker.Reset(cfg.ReloadInterval) + } + + if err := mgr.Apply(cfg); err != nil { + log.Error("cannot apply new configuration, keeping the previous one", "err", err) + continue + } + current = cfg + } +} + +func fileDigest(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func newLogger(cfg *config.Config, level *slog.LevelVar, w io.Writer) *slog.Logger { + level.Set(parseLevel(cfg.LogLevel)) + opts := &slog.HandlerOptions{Level: level} + + var h slog.Handler + if cfg.LogFormat == "json" { + h = slog.NewJSONHandler(w, opts) + } else { + h = slog.NewTextHandler(w, opts) + } + return slog.New(h) +} + +func parseLevel(s string) slog.Level { + switch s { + case "debug": + return slog.LevelDebug + case "warn": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} + +func printSummary(cfg *config.Config, path string) { + fmt.Printf("%s: OK\n\n", path) + fmt.Printf("work_dir %s\n", cfg.WorkDir) + fmt.Printf("concurrency %d\n", cfg.Concurrency) + fmt.Printf("reload every %s\n", cfg.ReloadInterval) + if cfg.Listen != "" { + fmt.Printf("listen %s\n", cfg.Listen) + } + fmt.Printf("\n%d repositor%s:\n", len(cfg.Jobs), plural(len(cfg.Jobs))) + for _, j := range cfg.Jobs { + fmt.Printf("\n %s\n", j.Name) + fmt.Printf(" src %s\n", gitx.RedactURL(j.Src.URL)) + fmt.Printf(" dst %s\n", gitx.RedactURL(j.Dst.URL)) + fmt.Printf(" every %s (timeout %s)\n", j.Interval, j.Timeout) + fmt.Printf(" refs %v\n", j.Refs) + fmt.Printf(" flags prune=%t force=%t atomic=%t allow_empty=%t\n", + j.Prune, j.Force, j.Atomic, j.AllowEmpty) + if j.Dst.SSHKey != "" { + fmt.Printf(" dst key %s (strict_host_key=%s)\n", j.Dst.SSHKey, j.Dst.StrictHostKey) + } + if j.Src.SSHKey != "" { + fmt.Printf(" src key %s (strict_host_key=%s)\n", j.Src.SSHKey, j.Src.StrictHostKey) + } + fmt.Printf(" mirror %s\n", j.Dir) + } +} + +func plural(n int) string { + if n == 1 { + return "y" + } + return "ies" +} + +// buildVersion prefers the ldflags value and falls back to VCS data stamped in +// by the Go toolchain, so a `go build` without flags still says something. +func buildVersion() string { + if version != "dev" { + return version + } + info, ok := debug.ReadBuildInfo() + if !ok { + return version + } + var rev, dirty string + for _, s := range info.Settings { + switch s.Key { + case "vcs.revision": + if len(s.Value) > 12 { + rev = s.Value[:12] + } else { + rev = s.Value + } + case "vcs.modified": + if s.Value == "true" { + dirty = "-dirty" + } + } + } + if rev == "" { + return version + } + return version + "+" + rev + dirty +} diff --git a/signal_other.go b/signal_other.go new file mode 100644 index 0000000..9d43d7b --- /dev/null +++ b/signal_other.go @@ -0,0 +1,9 @@ +//go:build !unix + +package main + +import "os" + +// reloadSignals is empty off unix: there is no SIGHUP, so reloads happen only +// through the file watcher. +func reloadSignals() []os.Signal { return nil } diff --git a/signal_unix.go b/signal_unix.go new file mode 100644 index 0000000..19b7749 --- /dev/null +++ b/signal_unix.go @@ -0,0 +1,11 @@ +//go:build unix + +package main + +import ( + "os" + "syscall" +) + +// reloadSignals are the signals that force an immediate config re-read. +func reloadSignals() []os.Signal { return []os.Signal{syscall.SIGHUP} }