diff options
| -rw-r--r-- | .dockerignore | 6 | ||||
| -rw-r--r-- | .gitignore | 3 | ||||
| -rw-r--r-- | Containerfile | 39 | ||||
| -rw-r--r-- | README | 322 | ||||
| -rw-r--r-- | fake-r2.go | 161 | ||||
| -rwxr-xr-x | fetch-forge | 28 | ||||
| -rw-r--r-- | forge.rev | 1 | ||||
| -rw-r--r-- | go.mod | 3 | ||||
| -rw-r--r-- | logs.go | 138 | ||||
| -rw-r--r-- | logs_test.go | 66 | ||||
| -rw-r--r-- | main.go | 160 | ||||
| -rw-r--r-- | main_test.go | 52 | ||||
| -rw-r--r-- | proxy.go | 153 | ||||
| -rw-r--r-- | proxy_test.go | 249 | ||||
| -rw-r--r-- | r2.go | 208 | ||||
| -rw-r--r-- | sync.go | 1043 | ||||
| -rw-r--r-- | sync_test.go | 1113 | ||||
| -rwxr-xr-x | test | 274 | ||||
| -rw-r--r-- | worker.js | 263 | ||||
| -rw-r--r-- | wrangler.jsonc | 35 |
20 files changed, 4317 insertions, 0 deletions
diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..f490a01 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +.wrangler +git-forge-cloudflare +*_test.go +README +fake-r2.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1e4c093 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +forge/ +git-forge-cloudflare +.wrangler/ diff --git a/Containerfile b/Containerfile new file mode 100644 index 0000000..73968f6 --- /dev/null +++ b/Containerfile @@ -0,0 +1,39 @@ +# Two binaries, one image, Alpine in both stages, matching git-forge's own +# Containerfile. +# +# forge/ is vendored by ./fetch-forge before a build and is not committed. + +FROM alpine:3.23 AS build +RUN apk add --no-cache go +WORKDIR /src + +COPY forge/ ./forge/ +COPY go.mod ./syncd/ +COPY *.go ./syncd/ + +# Static, so the runtime image needs no Go. buildvcs is off because neither tree +# carries a .git this build should read. +RUN cd forge && CGO_ENABLED=0 go build -trimpath -buildvcs=false -ldflags "-s -w" -o /git-forge . +RUN cd syncd && CGO_ENABLED=0 go build -trimpath -buildvcs=false -ldflags "-s -w" -o /syncd . + +FROM alpine:3.23 +# git-daemon is not the daemon here: in Alpine that subpackage is where +# git-http-backend lives, and without it every clone and push answers 500. +# tini reaps what git leaves: a detached gc is reparented to PID 1 and would +# otherwise stay a zombie. +RUN apk add --no-cache git git-daemon cgit highlight ca-certificates tini +COPY --from=build /git-forge /usr/local/bin/git-forge +COPY --from=build /syncd /usr/local/bin/syncd +# Which git-forge this is; syncd logs it at startup. +COPY --from=build /src/forge/.rev /etc/git-forge.rev + +# Non-root, and the datadir is credential material, so 0700 and one owner. +RUN adduser -D -H -u 1000 forge \ + && install -d -o forge -g forge -m 0700 /var/lib/git-forge +USER forge +ENV HOME=/tmp + +# No VOLUME: this disk is ephemeral by design and the durable copy is in the +# bucket. 8081 is syncd; git-forge listens on loopback only. +EXPOSE 8081 +ENTRYPOINT ["/sbin/tini", "--", "/usr/local/bin/syncd"] @@ -0,0 +1,322 @@ +git-forge-cloudflare +==================== +Runs git-forge on Cloudflare Containers with R2 for durable storage. A Worker +routes every request to a single container through one Durable Object. Inside +the container, syncd mirrors the data directory to an R2 bucket and restores +it at boot. + +git-forge is neither modified nor vendored here. The image builds it from its +own repository, at the commit named in `forge.rev`. + +Container disk is erased whenever the instance sleeps or its host is +recycled, and git-forge needs a real filesystem and one running process. +Cloudflare supplies the second: a Durable Object name resolves to one instance +globally. syncd supplies the first: the container keeps its filesystem on +ephemeral disk, R2 holds the durable copy, and git-forge is unaware of either. + + +Architecture +------------ + Worker -> Durable Object "forge" -> Container + :8080 git-forge, unmodified + :8081 syncd, in this repo + | + Outbound -> R2, logs, mail + + worker.js routes requests and answers the intercepted hosts + syncd mirrors and restores the datadir, makes writes durable + Containerfile git-forge, syncd, git, cgit, highlight, tini + wrangler.jsonc the bindings + +The Worker intercepts three hostnames, none of which resolve in DNS: + + r2.forge.internal objects, served from the R2 binding + logs.forge.internal stdout, logged where `wrangler tail` can see it + mail.forge.internal Mailgun's api shape, answered by whatever sends + +syncd carries no R2 credential: it sends plain HTTP to a hostname the Worker +intercepts on the same machine. Anything in the container that can make a +request has the whole bucket, cgit included; a binding has no narrower form. +Logs travel the same path because Cloudflare does not forward container +stdout to Workers observability. + + +Storage +------- + manifest.json which repositories exist. Authoritative + stale.json directories the manifest stopped listing, + and when + config/... the config directory, file by file + config/git/objects/... its audit log + config/REFS its refs + repos/{path}.git/objects/... immutable, uploaded once, never rewritten + repos/{path}.git/REFS `git show-ref` output, one PUT + +Git objects are content addressed and append only, so only refs change, and +syncd stores them as state rather than files: `gc --auto` moves loose refs +into `packed-refs`, and a restored loose ref would override it and resurrect +an old head. Objects go up first and refs last, the order git pushes in, so a +crash within a flush leaves the state from before the write. + +The manifest is what makes deletion work: restore creates only the +directories it names, so rewriting it deletes. The objects of a repository the +manifest no longer lists stay in the bucket for seven days, reported at every +boot and once a day while the container runs, and are removed by the first +check after that. The delay is the window to put back a manifest restored +from a backup or written short: a boot that believed one would otherwise +remove every repository it omits. + +A bucket holding objects under `repos/` and no manifest is fatal rather than +empty. A forge that never made a repository writes no manifest and starts +normally. + +Boot restores `config/` in full, then creates one empty directory per +repository in the manifest, which is all git-forge's startup scan looks at. +syncd hydrates a repository on its first request: `git init --bare`, copy the +objects, replay the refs. The issue and wiki siblings hydrate with it. A +repository that should have hydrated and did not fails loudly: git refuses an +empty directory, and git-forge does not treat that as an empty repository. + + +Durability +---------- +syncd puts a write in R2 before reporting success. It forwards the request to +git-forge and, for anything other than a known read, buffers the response, +flushes, and returns once the flush completes. A mutex serialises flushes and +each one rescans the whole datadir, so every request covers its own writes +whatever URL it had: + + held everything that is not a GET or HEAD, except git-upload-pack + streamed every GET and HEAD, except /confirm-email and /oidc/... + +A held response is capped at 8 MiB and answered 503 past it; the cap is for +cgit, which answers a snapshot to POST as well. + +One write is asynchronous: the date a push token was last used, which +git-forge treats as telemetry. A sweep every 15 seconds carries it up, and +backstops any misclassification above. + +tini is PID 1 and syncd its child, with git-forge as syncd's child, so the +container dies if syncd dies. On SIGTERM syncd signals git-forge, waits for +it to drain, flushes, and exits. + + +Limits +------ +Cloudflare caps a push at 100 MB of new objects at the edge, more on Business +and Enterprise, and nothing here can raise it. Clones are unbounded. git-forge +separately caps a wiki push at 16 MiB. Import an existing repository in +ranges, so that no single pack exceeds the cap: + + git push origin <sha>:refs/heads/master + +Local disk bounds the datadir. Instance type `basic` has 4 GB, and `git gc` +needs room to repack on top of it; `wrangler.jsonc` picks the type, and 20 GB +is the largest. + +Boot costs one round trip per object under `config/`. The sweep packs the +audit log at five hundred loose objects. A flush walks the whole datadir and +forks `git show-ref` per directory, so its cost, and every held write's +latency, grows with the number of repositories. + + +Test +---- + ./test + +Builds the image and exercises it end to end: create the first admin, create +a repository, push, destroy the container, then start a new one with only the +bucket to work from, and again after the audit log is packed. It needs docker, +go, git, curl and openssl on the host, and the git-forge checkout +`fetch-forge` reads. `fake-r2.go` stands in for the object handler in +`worker.js`, so no Cloudflare account is needed. + +`go test` covers syncd against a bucket: a flush racing a hydration, a gc, a +push, and a deletion, and the prune waiting out its grace. + +The rest of `worker.js` has no coverage: it runs only in a deployment. + + +Deploy +------ +Needs a Cloudflare account on Workers Paid with R2 enabled, wrangler, and +docker. + + ./fetch-forge + wrangler r2 bucket create git-forge + wrangler secret put GIT_FORGE_COOKIE_KEY at least 32 characters + wrangler deploy + +`fetch-forge` copies git-forge's source into `forge/` for the image build, +from `../git-forge` at the commit in `forge.rev`, or from `GIT_FORGE_SRC` and +`GIT_FORGE_REV`. `forge/` is derived and uncommitted. Upgrading means editing +`forge.rev`; the pin exists because this repository asserts which of +git-forge's routes write, which one removes a repository, and what a +repository holds besides objects and refs. syncd logs the revision at startup. + +`GIT_FORGE_COOKIE_KEY` must be a secret: the datadir is ephemeral, so a +generated key would sign everybody out on every restart. syncd refuses to +start without one. + +A custom domain is one route, with workers.dev switched off: + + "routes": [{ "pattern": "git.example.com", "custom_domain": true }], + "workers_dev": false, + +Switching workers.dev off matters: base_url is decided by the request that +starts the container, so two hostnames on one worker means whichever wakes it +wins and every form posted to the other returns 403 until it restarts. + +Leave `base_url` unset. worker.js supplies it from the origin of the request +that started the container, always as https. Set it in `wrangler.jsonc` only +when a custom domain fronts a hostname requests do not arrive on, and then +exactly. Put nothing else in `vars`: an environment value pins what the admin +page could otherwise change. worker.js sets `GIT_FORGE_CLIENT_IP_HEADER` to +`CF-Connecting-IP`, which the edge overwrites, so rate limits count real +addresses rather than syncd's. + +Against an empty bucket git-forge prints a one time setup token and serves +`/setup`. The container starts on the first request and prints the token +then, so start `wrangler tail` beforehand. A restart before setup issues a new +token; `wrangler deploy` forces a restart. + +The container stops after ten idle minutes and starts again on the next +request. A Durable Object alarm stops it with SIGTERM once nothing is in +flight; a request arriving mid-drain receives an error. Retry. + + +Mail +---- +git-forge's Mailgun driver issues one form-encoded POST, and `worker.js` +answers that shape, so any sender serves git-forge without changing it and +every key stays outside the container. The binding is declared and sends +nothing until git-forge's config points at the intercepted host; until then +git-forge's default driver drops each message with a log line naming the +recipient and subject. + +Edit the copy of `forge.json` in the bucket, since the web interface does not +expose mail settings, then redeploy: + + wrangler r2 object get git-forge/config/forge.json --remote --pipe >f + "$EDITOR" f + wrangler r2 object put git-forge/config/forge.json --remote --file=f + wrangler deploy + +`forge.json` reaches the bucket after the container has run once. git-forge +requires `domain` to be non-empty though nothing uses it: + + "mail": { + "driver": "mailgun", + "from": "git-forge <forge@example.com>", + "mailgun": { + "domain": "forge", + "base_url": "https://mail.forge.internal/v3" + } + } + +Without a sending domain onboarded to Email Service, a binding reaches only +addresses verified in the account. With no binding declared, `worker.js` logs +each message and reports success. The spool lives on container disk and is +not mirrored. + + +Sign in with a provider +----------------------- +git-forge implements generic OpenID Connect: discovery, authorization code +with PKCE, and a signed id_token carrying a verified address. Google works, +as does anything that publishes a discovery document; GitHub does not, since +its only OIDC issuer mints workload tokens. + +Declare the provider in the bucket's copy of `forge.json`, as with mail: + + "oidc": [{ + "key": "google", + "label": "Google", + "issuer": "https://accounts.google.com", + "client_id": "<id>.apps.googleusercontent.com", + "allow": ["someone@gmail.com", "@example.com"] + }] + +Supply the secret through the environment, never the file: + + wrangler secret put GIT_FORGE_OIDC_GOOGLE_CLIENT_SECRET + +git-forge derives the variable name from the key, upper-cased. Register the +redirect uri `{base_url}/oidc/{key}/callback` with the provider exactly. A +secret added while a container runs takes effect after a restart; both sides +report what they hold, without its value, in `wrangler tail`. + +An account minted this way is named `{key}:{name}` and holds no grants; the +admins receive mail when one first appears. Set `allow`, or every account the +provider vouches for can mint one. + +The container contacts the provider directly. This is the only feature that +depends on `enableInternet` in `worker.js` and on the ca bundle `main.go` +assembles; a forge with no provider can set it false. + + +Operations +---------- +R2 holds everything that has to survive. Two backups: + + every repository, cloned readable without any of this + the bucket, mirrored to another bucket exact, and restores as is + +Cloning omits the account table, so keep `users.json` beside it, and treat it +as credential material in full: + + wrangler r2 object get git-forge/config/users.json --remote --pipe + +R2 speaks the S3 api at `https://<account>.r2.cloudflarestorage.com`, so any +S3 tool mirrors the bucket. + +A repository deleted through the web is reported at every boot, and daily +while the container runs, for seven days: + + stale: notes.git, 412 objects no manifest lists, removed after 2026-09-18 + +To bring it back within that window, put `manifest.json` and `repos.json` +back as they were and redeploy; after it, a clone is the only recovery. + +Reverting any other mistake means editing the bucket and restarting, as with +mail. A restore reads the working tree, not the audit log, so putting the old +content back is the revert: + + wrangler r2 object get git-forge/config/repos.json --remote --pipe >f + "$EDITOR" f + wrangler r2 object put git-forge/config/repos.json --remote --file=f + wrangler deploy + +Do this while the forge is idle: syncd leaves a change made behind its back +alone, but a simultaneous change through the web wins. + +What the log means when something is wrong: + + restore: ... fatal, deliberately: an empty datadir + would be flushed over the real one + the change could not be made a flush failed, so the write is on a disk + durable that does not survive: a 503 on purpose. + The next flush carries it up + Not a git repository: hydration did not fire for a repository + /var/lib/git-forge/repos/... the matcher missed. The objects are still + in the bucket + config: audit log not the bucket is missing an object the log's + replayed refs name. This costs history and nothing + else + no sender bound mail is configured but the binding is not + GIT_FORGE_COOKIE_KEY must be the secret is unset + set + +Memory and disk are billed on the instance type while the container is +awake; a sleeping container costs nothing. + + +Debugging +--------- + wrangler tail logs, both processes + wrangler r2 object get git-forge/manifest.json --remote --pipe + wrangler r2 object get git-forge/repos/foo.git/REFS --remote --pipe + +`--remote` is mandatory: without it wrangler reads a local simulated bucket +and finds nothing. `REFS` is `git show-ref` output. syncd logs one line per +flush that uploaded anything and one per hydration. diff --git a/fake-r2.go b/fake-r2.go new file mode 100644 index 0000000..6bfdb3f --- /dev/null +++ b/fake-r2.go @@ -0,0 +1,161 @@ +//go:build ignore + +package main + +// A stand-in for the object handler in worker.js, backed by a directory, so +// the image can be driven end to end without a Cloudflare account. +// +// go run fake-r2.go -dir /tmp/bucket -listen :9000 + +import ( + "encoding/json" + "flag" + "io" + "io/fs" + "log" + "net/http" + "os" + "path/filepath" + "slices" + "strconv" + "strings" +) + +var dir string + +func main() { + listen := flag.String("listen", ":9000", "address to serve on") + flag.StringVar(&dir, "dir", "", "directory to keep objects in") + flag.Parse() + if dir == "" { + log.Fatal("-dir is required") + } + if err := os.MkdirAll(dir, 0700); err != nil { + log.Fatal(err) + } + log.Printf("fake r2 on %s backed by %s", *listen, dir) + log.Fatal(http.ListenAndServe(*listen, http.HandlerFunc(serve))) +} + +// path keeps a key inside the directory. +func path(key string) (string, bool) { + if !filepath.IsLocal(key) { + return "", false + } + return filepath.Join(dir, filepath.FromSlash(key)), true +} + +func serve(w http.ResponseWriter, r *http.Request) { + key := strings.TrimPrefix(r.URL.Path, "/") + // A post with no key is the log host; the lines are logged for the test to + // assert on. + if r.Method == http.MethodPost && key == "" { + body, _ := io.ReadAll(r.Body) + for _, line := range strings.Split(strings.TrimRight(string(body), "\n"), "\n") { + if line != "" { + log.Printf("container: %s", line) + } + } + w.WriteHeader(http.StatusNoContent) + return + } + // Logged, so the test can assert an idle sweep writes nothing. + log.Printf("%s %s", r.Method, key) + if r.Method == http.MethodGet && key == "" { + list(w, r) + return + } + file, ok := path(key) + if !ok { + http.Error(w, "bad key", http.StatusBadRequest) + return + } + + switch r.Method { + case http.MethodGet: + f, err := os.Open(file) + if err != nil { + http.Error(w, "not found", http.StatusNotFound) + return + } + defer f.Close() + info, err := f.Stat() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Length", strconv.FormatInt(info.Size(), 10)) + _, _ = io.Copy(w, f) + case http.MethodPut: + if r.ContentLength < 0 { + http.Error(w, "content-length required", http.StatusLengthRequired) + return + } + if err := os.MkdirAll(filepath.Dir(file), 0700); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + tmp, err := os.CreateTemp(filepath.Dir(file), ".put*") + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + defer os.Remove(tmp.Name()) + if _, err := io.Copy(tmp, r.Body); err != nil { + tmp.Close() + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + tmp.Close() + if err := os.Rename(tmp.Name(), file); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + case http.MethodDelete: + _ = os.Remove(file) + w.WriteHeader(http.StatusNoContent) + default: + http.Error(w, "bad method", http.StatusMethodNotAllowed) + } +} + +// Real r2 pages at a thousand; ten makes the container follow a cursor. +const page = 10 + +func list(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + var keys []string + err := filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() { + return err + } + rel, err := filepath.Rel(dir, p) + if err != nil { + return err + } + key := filepath.ToSlash(rel) + if strings.HasPrefix(key, prefix) && !strings.HasPrefix(filepath.Base(key), ".put") { + keys = append(keys, key) + } + return nil + }) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + slices.Sort(keys) + if keys == nil { + keys = []string{} // an empty listing is [], as the worker's map gives + } + from := 0 + if c := r.URL.Query().Get("cursor"); c != "" { + from, _ = strconv.Atoi(c) + } + to := min(from+page, len(keys)) + next := "" + if to < len(keys) { + next = strconv.Itoa(to) + } + _ = json.NewEncoder(w).Encode(map[string]any{"keys": keys[from:to], "cursor": next}) +} diff --git a/fetch-forge b/fetch-forge new file mode 100755 index 0000000..1a3aa99 --- /dev/null +++ b/fetch-forge @@ -0,0 +1,28 @@ +#!/bin/sh +# Vendor git-forge's source into forge/ for the image build. Derived state: +# gitignored, and running this again replaces it. +# +# GIT_FORGE_SRC where to take it from, default ../git-forge +# GIT_FORGE_REV which commit, default the one in forge.rev +set -eu + +src="${GIT_FORGE_SRC:-../git-forge}" +rev="${GIT_FORGE_REV:-$(cat forge.rev)}" + +# Resolved before anything is removed, so a revision that is not there fails +# with the tree that was vendored last still in place. --verify, because plain +# rev-parse echoes a forty character sha back without looking for it. +sha=$(git -C "$src" rev-parse --verify "$rev^{commit}") + +# Into a file rather than a pipe: a pipeline exits with the status of its last +# command, so a failed archive would be an empty forge/ and an exit of zero. +archive=$(mktemp) +trap 'rm -f "$archive"' EXIT +git -C "$src" archive -o "$archive" "$sha" + +rm -rf forge +mkdir forge +tar -xf "$archive" -C forge +echo "$sha" >forge/.rev + +echo "vendored git-forge $sha from $src" diff --git a/forge.rev b/forge.rev new file mode 100644 index 0000000..46f8697 --- /dev/null +++ b/forge.rev @@ -0,0 +1 @@ +bb6bc2387d58c737dbf92e73f8544bf39319dbd8 @@ -0,0 +1,3 @@ +module git-forge-cloudflare + +go 1.25.0 @@ -0,0 +1,138 @@ +package main + +// Cloudflare does not forward container stdout to observability, so syncd +// ships its own output and git-forge's to the worker. Drops rather than +// blocks, and never logs through log itself. + +import ( + "bytes" + "fmt" + "net/http" + "os" + "strings" + "sync/atomic" + "time" +) + +const ( + shipQueue = 256 + shipEvery = 500 * time.Millisecond + shipTimeout = 3 * time.Second + // The bound on getting run's attention; the flush itself is waited out, + // since those lines say how the container stopped. + shipWait = 5 * time.Second +) + +type shipper struct { + url string + lines chan string + flush chan chan struct{} + dropped atomic.Int64 // never queued, because the queue was full + unsent atomic.Int64 // queued, but the send failed +} + +// newShipper tees to stdout and to the worker, or to stdout alone with no +// endpoint. +func newShipper(base string) *shipper { + s := &shipper{ + url: base, + lines: make(chan string, shipQueue), + flush: make(chan chan struct{}), + } + if base != "" { + go s.run() + } + return s +} + +// close sends what is queued and waits: the last lines say how the container +// shut down. +func (s *shipper) close() { + if s.url == "" { + return + } + ack := make(chan struct{}) + select { + case s.flush <- ack: + <-ack + case <-time.After(shipWait): + } +} + +// Write always reports success: the caller is a logger. +func (s *shipper) Write(p []byte) (int, error) { + _, _ = os.Stdout.Write(p) + if s.url == "" { + return len(p), nil + } + for _, line := range strings.Split(string(bytes.TrimRight(p, "\n")), "\n") { + if line == "" { + continue + } + select { + case s.lines <- line: + default: + // Dropped rather than stalling a request; counted, so the gap is + // reported. + s.dropped.Add(1) + } + } + return len(p), nil +} + +func (s *shipper) run() { + client := &http.Client{Timeout: shipTimeout} + t := time.NewTicker(shipEvery) + defer t.Stop() + + var batch []string + send := func() { + lines := int64(len(batch)) + dropped, unsent := s.dropped.Swap(0), s.unsent.Swap(0) + if dropped > 0 { + batch = append(batch, fmt.Sprintf("syncd: dropped %d log lines, the queue was full", dropped)) + } + if unsent > 0 { + batch = append(batch, fmt.Sprintf("syncd: %d log lines did not reach the worker", unsent)) + } + if len(batch) == 0 { + return + } + body := strings.Join(batch, "\n") + batch = batch[:0] + // The status too: a 500 loses the batch as a refused connection does. + landed := false + if resp, err := client.Post(s.url, "text/plain", strings.NewReader(body)); err == nil { + landed = resp.StatusCode/100 == 2 + resp.Body.Close() + } + if !landed { + // Put back, so the next send that works reports one total. + s.dropped.Add(dropped) + s.unsent.Add(unsent + lines) + } + } + + for { + select { + case line := <-s.lines: + batch = append(batch, line) + if len(batch) >= shipQueue/2 { + send() + } + case <-t.C: + send() + case ack := <-s.flush: + for drained := false; !drained; { + select { + case line := <-s.lines: + batch = append(batch, line) + default: + drained = true + } + } + send() + close(ack) + } + } +} diff --git a/logs_test.go b/logs_test.go new file mode 100644 index 0000000..abef7eb --- /dev/null +++ b/logs_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// A gap in the log reads as nothing having happened, so a line that never +// reaches the worker has to be counted and reported by one that does. The two +// ways it fails are a send that errors and a send that is answered with an +// error, and only the first of them looks like a failure to the client. +func TestShipperReportsWhatDidNotLand(t *testing.T) { + var mu sync.Mutex + mode := "refuse" + var landed []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + switch mode { + case "refuse": + http.Error(w, "no", http.StatusInternalServerError) + case "drop": + // A send that errors rather than answering. + if conn, _, err := w.(http.Hijacker).Hijack(); err == nil { + conn.Close() + } + default: + body, _ := io.ReadAll(r.Body) + landed = append(landed, string(body)) + w.WriteHeader(http.StatusNoContent) + } + })) + defer srv.Close() + + s := newShipper(srv.URL) + // close sends what is queued and waits, so each of these is one send. + for _, step := range []struct{ mode, line string }{ + {"refuse", "answered with a 500"}, + {"drop", "the connection went away"}, + {"take", "this one lands"}, + } { + mu.Lock() + mode = step.mode + mu.Unlock() + if _, err := s.Write([]byte(step.line + "\n")); err != nil { + t.Fatal(err) + } + s.close() + } + + mu.Lock() + defer mu.Unlock() + if len(landed) != 1 { + t.Fatalf("the worker was handed %d bodies, want the one it accepted: %q", len(landed), landed) + } + if !strings.Contains(landed[0], "this one lands") { + t.Errorf("the line that should have landed is missing from %q", landed[0]) + } + if !strings.Contains(landed[0], "2 log lines did not reach the worker") { + t.Errorf("the two lines that did not land are not reported in %q", landed[0]) + } +} @@ -0,0 +1,160 @@ +package main + +// syncd fronts git-forge in the container. Container disk is wiped whenever +// the instance sleeps, so the datadir is restored before git-forge starts, +// mirrored while it runs, and flushed on the way out. + +import ( + "bytes" + "context" + "errors" + "log" + "net/http" + "os" + "os/exec" + "os/signal" + "syscall" + "time" +) + +const ( + datadir = "/var/lib/git-forge" + forgeAddr = "127.0.0.1:8080" + listen = ":8081" + + sweepInterval = 15 * time.Second + + // Cloudflare's interception ca is per instance and the container is not + // root, so SSL_CERT_FILE names a merged copy. + cfCert = "/etc/cloudflare/certs/cloudflare-containers-ca.crt" + sysCert = "/etc/ssl/certs/ca-certificates.crt" + bundle = "/tmp/ca-bundle.crt" + + revFile = "/etc/git-forge.rev" // written by the Containerfile from fetch-forge +) + +// Package scope because fatal flushes it before exiting. +var ship *shipper + +func main() { + // First, so a failure during startup is shipped too. + ship = newShipper(os.Getenv("SYNCD_LOGS")) + log.SetFlags(0) + log.SetPrefix("syncd: ") + log.SetOutput(ship) + + endpoint := os.Getenv("SYNCD_R2") + if endpoint == "" { + fatal("SYNCD_R2 is not set, so there is nowhere durable to write") + } + // Unset, git-forge mints one into a datadir that does not survive a + // restart. + if len(os.Getenv("GIT_FORGE_COOKIE_KEY")) < 32 { + fatal("GIT_FORGE_COOKIE_KEY must be set to at least 32 characters, " + + "because this datadir does not survive a restart: wrangler secret put GIT_FORGE_COOKIE_KEY") + } + + rev, err := os.ReadFile(revFile) + if err != nil { + rev = []byte("unknown") + } + log.Printf("git-forge %s", bytes.TrimSpace(rev)) + + s := newSync(newR2(endpoint), datadir) + + if err := caBundle(sysCert, cfCert, bundle); err != nil { + // Not fatal: only outbound mail needs the intercepted ca. + log.Printf("no ca bundle, outbound mail over tls will fail: %v", err) + } + + start := time.Now() + if err := s.restore(); err != nil { + // An empty datadir would be flushed over the real one. + fatal("restore: %v", err) + } + log.Printf("restored config and %d repository directories in %s", + len(s.manifest), time.Since(start).Round(time.Millisecond)) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + forge := exec.Command("git-forge") + // A later entry wins: git-forge has to listen where the proxy sends and + // write where syncd mirrors. + forge.Env = append(os.Environ(), + "SSL_CERT_FILE="+bundle, + "GIT_FORGE_LISTEN="+forgeAddr, + "GIT_FORGE_DATADIR="+datadir, + ) + forge.Stdout, forge.Stderr = ship, ship + if err := forge.Start(); err != nil { + fatal("git-forge: %v", err) + } + exited := make(chan error, 1) + go func() { exited <- forge.Wait() }() + + srv := &http.Server{ + Addr: listen, + Handler: s, + // No read or write deadline: a push sends a whole packfile in one + // request and a clone streams one back. + ReadHeaderTimeout: 30 * time.Second, + IdleTimeout: 2 * time.Minute, + } + go func() { + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + fatal("listen: %v", err) + } + }() + go s.sweep(ctx) + + code := 0 + select { + case err := <-exited: + // Exit nonzero and take the container down. + log.Printf("git-forge exited: %v", err) + code = 1 + case <-ctx.Done(): + log.Print("draining") + _ = forge.Process.Signal(syscall.SIGTERM) + <-exited + } + + // Only meaningful now that git-forge is stopped and cannot write again. + shut, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := srv.Shutdown(shut); err != nil { + log.Printf("shutdown: %v", err) + } + + if err := s.flush(); err != nil { + log.Printf("final flush: %v", err) + code = 1 + } else { + log.Print("flushed and stopping") + } + ship.close() + os.Exit(code) +} + +// Not log.Fatal, which exits with the line still queued in the shipper. +func fatal(format string, args ...any) { + log.Printf(format, args...) + ship.close() + os.Exit(1) +} + +// caBundle appends the interception ca to the system roots, so oidc still +// verifies against the real ones. +func caBundle(sys, cf, out string) error { + roots, err := os.ReadFile(sys) + if err != nil { + return err + } + intercept, err := os.ReadFile(cf) + if err != nil { + // No interception configured. + return os.WriteFile(out, roots, 0600) + } + return os.WriteFile(out, append(append(roots, '\n'), intercept...), 0600) +} diff --git a/main_test.go b/main_test.go new file mode 100644 index 0000000..b4b0af2 --- /dev/null +++ b/main_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// The interception ca is added to the system roots and never substituted for +// them: the one feature that reaches the internet directly is oidc, and it +// verifies a real provider against the roots the image ships. +func TestCABundle(t *testing.T) { + dir := t.TempDir() + sys := filepath.Join(dir, "ca-certificates.crt") + cf := filepath.Join(dir, "cloudflare.crt") + out := filepath.Join(dir, "ca-bundle.crt") + if err := os.WriteFile(sys, []byte("the system roots\n"), 0600); err != nil { + t.Fatal(err) + } + + // No interception certificate, which is what a bare docker run gets. + if err := caBundle(sys, cf, out); err != nil { + t.Fatalf("with no interception ca: %v", err) + } + if got, err := os.ReadFile(out); err != nil || string(got) != "the system roots\n" { + t.Errorf("the bundle is %q: %v, want the system roots alone", got, err) + } + + if err := os.WriteFile(cf, []byte("the interception ca\n"), 0600); err != nil { + t.Fatal(err) + } + if err := caBundle(sys, cf, out); err != nil { + t.Fatalf("with one: %v", err) + } + got, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(got), "the system roots") { + t.Errorf("the system roots were replaced rather than added to: %q", got) + } + if !strings.Contains(string(got), "the interception ca") { + t.Errorf("the interception ca is not in the bundle: %q", got) + } + + // A system store that is not there is a bundle that cannot be trusted, not + // an empty one. + if err := caBundle(filepath.Join(dir, "absent"), cf, out); err == nil { + t.Error("a missing system store was accepted") + } +} diff --git a/proxy.go b/proxy.go new file mode 100644 index 0000000..6ea4594 --- /dev/null +++ b/proxy.go @@ -0,0 +1,153 @@ +package main + +// syncd fronts git-forge: a repository is filled in before it is served and a +// write is durable before its response leaves. A flush walks the whole +// datadir, so a misclassified route costs delay, not durability. + +import ( + "bytes" + "fmt" + "log" + "net/http" + "net/http/httputil" + "strings" +) + +var proxy = &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.Out.URL.Scheme = "http" + pr.Out.URL.Host = forgeAddr + }, + // Unbuffered: a clone streams. + FlushInterval: -1, +} + +// Asked for by the idle alarm. Counted here: a fetch in the worker resolves at +// the headers, minutes before a clone's last byte. +const inflightPath = "/.syncd/inflight" + +// What a held response may weigh. A write's answer is kilobytes; the bound is +// for cgit, which answers a snapshot to POST as well. +const maxHeld = 8 << 20 + +func (s *Sync) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // RawPath is set only when the path arrived escaped, past the worker's + // check on the prefix: passed on to 404. + if r.URL.Path == inflightPath && r.URL.RawPath == "" { + fmt.Fprintln(w, s.inflight.Load()) + return + } + s.inflight.Add(1) + defer s.inflight.Add(-1) + + // Reads too. Deleting is the exception: git-forge removes the directories + // without reading them. + if base, ok := s.repoOf(r.URL.Path); ok && !strings.HasSuffix(r.URL.Path, "/settings/delete") { + if err := s.hydrate(base); err != nil { + log.Printf("%s: %v", r.URL.Path, err) + http.Error(w, "this repository could not be fetched from storage\n", http.StatusServiceUnavailable) + return + } + } + + if isRead(r) { + proxy.ServeHTTP(w, r) + return + } + + // A write: hold the response until it is in the bucket. + held := &buffered{header: http.Header{}} + proxy.ServeHTTP(held, r) + + if held.status/100 == 2 || held.status/100 == 3 { + if err := s.flush(); err != nil { + // A 503 for a write already on disk: the disk does not survive. A + // later flush carries it up. + log.Printf("flush for %s: %v", r.URL.Path, err) + http.Error(w, "the change could not be made durable\n", http.StatusServiceUnavailable) + return + } + } + // Flushed first, so the write is durable and only the body is lost; a + // truncated body would be worse than none. + if held.overflow { + log.Printf("%s %s: response past %d bytes, discarded rather than held", + r.Method, r.URL.Path, maxHeld) + http.Error(w, "this response is too large to hold\n", http.StatusServiceUnavailable) + return + } + held.writeTo(w) +} + +// isRead is a request that changes nothing that has to survive. +func isRead(r *http.Request) bool { + p := r.URL.Path + if r.Method != http.MethodGet && r.Method != http.MethodHead { + // A clone is a POST, and gigabytes. + return strings.HasSuffix(p, "/git-upload-pack") + } + // The two get routes that write: both edit users.json. + if p == "/confirm-email" || strings.HasPrefix(p, "/oidc/") { + return false + } + return true +} + +// repoOf resolves the repository a path names, two components before one as +// git-forge does. Stripping the suffixes is unambiguous because git-forge +// refuses names ending in them. +func (s *Sync) repoOf(p string) (string, bool) { + p = strings.TrimPrefix(p, "/") + parts := strings.SplitN(p, "/", 3) + for n := min(2, len(parts)); n >= 1; n-- { + base := strings.Join(parts[:n], "/") + base = strings.TrimSuffix(base, ".git") + base = strings.TrimSuffix(base, ".issues") + base = strings.TrimSuffix(base, ".wiki") + if s.known(base) { + return base, true + } + } + return "", false +} + +// buffered holds a response until it is durable. Past maxHeld it keeps nothing +// and says so. +type buffered struct { + header http.Header + status int + body bytes.Buffer + overflow bool +} + +func (b *buffered) Header() http.Header { return b.header } + +func (b *buffered) WriteHeader(status int) { + if b.status == 0 { + b.status = status + } +} + +// Write reports success even when it keeps nothing: a short write makes the +// proxy abort the connection before ServeHTTP can flush. +func (b *buffered) Write(p []byte) (int, error) { + if b.status == 0 { + b.status = http.StatusOK + } + if b.overflow || b.body.Len()+len(p) > maxHeld { + b.overflow = true + return len(p), nil + } + return b.body.Write(p) +} + +func (b *buffered) writeTo(w http.ResponseWriter) { + for k, v := range b.header { + w.Header()[k] = v + } + if b.status == 0 { + b.status = http.StatusOK + } + w.WriteHeader(b.status) + _, _ = w.Write(b.body.Bytes()) +} diff --git a/proxy_test.go b/proxy_test.go new file mode 100644 index 0000000..1e5a0e6 --- /dev/null +++ b/proxy_test.go @@ -0,0 +1,249 @@ +package main + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "strings" + "testing" + "time" +) + +// proxyTo stands a server running h where git-forge would be, for the length of +// the test, configured as proxy.go configures it so that a test drives the path +// a deployment does. +func proxyTo(t *testing.T, h http.HandlerFunc) { + t.Helper() + forge := httptest.NewServer(h) + t.Cleanup(forge.Close) + target, err := url.Parse(forge.URL) + if err != nil { + t.Fatal(err) + } + saved := proxy + proxy = &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { pr.SetURL(target) }, + FlushInterval: -1, + } + t.Cleanup(func() { proxy = saved }) +} + +// TestInflight pins what the idle alarm asks for: the count has to hold until +// the last byte of a response. +func TestInflight(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + proxyTo(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + close(started) + <-release + }) + + front := httptest.NewServer(newSync(nil, t.TempDir())) + defer front.Close() + + count := func() string { + resp, err := http.Get(front.URL + inflightPath) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + out, _ := io.ReadAll(resp.Body) + return strings.TrimSpace(string(out)) + } + if got := count(); got != "0" { + t.Errorf("with nothing happening the count is %q, want 0", got) + } + + done := make(chan struct{}) + go func() { + defer close(done) + resp, err := http.Get(front.URL + "/notes.git/git-upload-pack") + if err == nil { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + }() + + <-started + if got := count(); got != "1" { + t.Errorf("while a response is still streaming the count is %q, want 1", got) + } + + close(release) + <-done + // The decrement is deferred, so it lands just after the client is served. + for deadline := time.Now().Add(time.Second); ; { + if count() == "0" { + break + } + if time.Now().After(deadline) { + t.Errorf("the count never came back to 0") + break + } + time.Sleep(5 * time.Millisecond) + } +} + +// TestInflightPathIsExact pins that only the literal spelling is answered here: +// the worker refuses this prefix undecoded, so an escaped one has to go on. +func TestInflightPathIsExact(t *testing.T) { + proxyTo(t, func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no such page\n", http.StatusNotFound) + }) + + front := httptest.NewServer(newSync(nil, t.TempDir())) + defer front.Close() + + resp, err := http.Get(front.URL + "/%2Esyncd/inflight") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("an escaped %s was answered with %s %q, want it passed on", inflightPath, resp.Status, body) + } +} + +// TestHeldResponseIsBounded pins what a held response may weigh. cgit answers +// any method, so a snapshot asked for with POST is a read that arrives here as +// a write, and holding it would be the repository in memory. +func TestHeldResponseIsBounded(t *testing.T) { + body := bytes.Repeat([]byte("x"), 64<<10) + var weight int + proxyTo(t, func(w http.ResponseWriter, r *http.Request) { + for sent := 0; sent < weight; sent += len(body) { + _, _ = w.Write(body) + } + }) + + // A real bucket, so that the flush a write triggers succeeds and the only + // thing under test is the size of the answer. + _, bucketURL := newBucket(t) + s := newSync(newR2(bucketURL), t.TempDir()) + if err := s.restore(); err != nil { + t.Fatal(err) + } + front := httptest.NewServer(s) + defer front.Close() + + post := func() (int, int) { + resp, err := http.Post(front.URL+"/notes/snapshot/notes-master.tar.gz", "text/plain", nil) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + out, _ := io.ReadAll(resp.Body) + return resp.StatusCode, len(out) + } + + weight = maxHeld - len(body) + if status, n := post(); status != http.StatusOK || n != weight { + t.Errorf("a response under the bound came back %d with %d bytes, want 200 with %d", status, n, weight) + } + weight = maxHeld + len(body) + if status, _ := post(); status != http.StatusServiceUnavailable { + t.Errorf("a response over the bound came back %d, want it refused", status) + } +} + +// TestDeleteDoesNotHydrate pins the one exception to hydrating on any request +// that names a repository. git-forge removes the directories without reading +// them, so fetching first would be the whole repository onto disk on its way +// out. The exception costs nothing if it is lost, which is why it is pinned. +func TestDeleteDoesNotHydrate(t *testing.T) { + gitIdentity(t) + b, bucketURL := newBucket(t) + seed(t, b, bucketURL) + + proxyTo(t, func(w http.ResponseWriter, r *http.Request) {}) + + s := newSync(newR2(bucketURL), t.TempDir()) + if err := s.restore(); err != nil { + t.Fatal(err) + } + front := httptest.NewServer(s) + defer front.Close() + + post := func(path string) { + resp, err := http.Post(front.URL+path, "text/plain", nil) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + } + post("/x/settings/delete") + if s.hydrated("x.git") { + t.Error("the delete route fetched the repository it was about to remove") + } + post("/x/settings") + if !s.hydrated("x.git") { + t.Error("every other route still has to hydrate") + } +} + +// TestRepoOf pins the matcher that decides what a request hydrates. Matching +// something that is not a repository is wasted work; missing one serves an +// empty directory, which git reports as a fatal rather than an empty repository. +func TestRepoOf(t *testing.T) { + s := newSync(nil, t.TempDir()) + s.setManifest([]string{"notes.git", "notes.issues.git", "notes.wiki.git", "team/api.git"}) + + cases := []struct { + path string + want string + }{ + {"/notes.git/info/refs", "notes"}, + {"/notes.git/git-upload-pack", "notes"}, + {"/notes/tree/README", "notes"}, + {"/notes", "notes"}, + {"/notes.issues.git/info/refs", "notes"}, + {"/notes.wiki.git/info/refs", "notes"}, + {"/team/api.git/info/refs", "team/api"}, + {"/team/api/log", "team/api"}, + {"/team", ""}, + {"/settings", ""}, + {"/", ""}, + {"/nosuch.git/info/refs", ""}, + } + for _, c := range cases { + base, ok := s.repoOf(c.path) + if !ok { + base = "" + } + if base != c.want { + t.Errorf("repoOf(%q) = %q, want %q", c.path, base, c.want) + } + } +} + +// TestIsRead pins the rule that decides what is streamed and what is held. A +// mistake costs latency rather than correctness, except in the two directions +// that are not symmetric: a clone held is a gigabyte in memory, and the two get +// routes that edit users.json would be answered before they are durable. +func TestIsRead(t *testing.T) { + cases := []struct { + method, path string + want bool + }{ + {"GET", "/notes.git/info/refs", true}, + {"GET", "/notes/tree/README", true}, + {"HEAD", "/notes/tree/README", true}, + {"POST", "/notes.git/git-upload-pack", true}, + {"POST", "/notes.git/git-receive-pack", false}, + {"GET", "/confirm-email", false}, + {"GET", "/oidc/google", false}, + {"GET", "/oidc/google/callback", false}, + {"POST", "/settings/tokens", false}, + {"POST", "/admin/repo/create", false}, + } + for _, c := range cases { + if got := isRead(httptest.NewRequest(c.method, c.path, nil)); got != c.want { + t.Errorf("isRead(%s %s) = %v, want %v", c.method, c.path, got, c.want) + } + } +} @@ -0,0 +1,208 @@ +package main + +// The object store, over plain http to a host the worker intercepts on the +// same machine: no credential, no signing. + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "time" +) + +type R2 struct { + base string + client *http.Client +} + +func newR2(base string) *R2 { + return &R2{base: base, client: &http.Client{Timeout: 10 * time.Minute}} +} + +// Small: the peer is on this machine. Retrying a body is sound because putBody +// sets GetBody. +const tries = 3 + +func (r *R2) do(req *http.Request) (*http.Response, error) { + var err error + for attempt := range tries { + if attempt > 0 { + time.Sleep(time.Duration(attempt) * 200 * time.Millisecond) + if req.GetBody != nil { + if req.Body, err = req.GetBody(); err != nil { + return nil, err + } + } + } + var resp *http.Response + if resp, err = r.client.Do(req); err == nil { + if resp.StatusCode < 500 { + return resp, nil + } + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + resp.Body.Close() + err = fmt.Errorf("%s %s: %s: %s", req.Method, req.URL.Path, resp.Status, body) + } + } + return nil, err +} + +func (r *R2) url(key string) string { + return r.base + "/" + (&url.URL{Path: key}).EscapedPath() +} + +// get reads a whole object into memory; anything large goes through getFile. +func (r *R2) get(key string) ([]byte, bool, error) { + req, err := http.NewRequest(http.MethodGet, r.url(key), nil) + if err != nil { + return nil, false, err + } + resp, err := r.do(req) + if err != nil { + return nil, false, err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return nil, false, nil + } + if resp.StatusCode != http.StatusOK { + return nil, false, fmt.Errorf("get %s: %s", key, resp.Status) + } + data, err := io.ReadAll(resp.Body) + if err != nil { + return nil, false, err + } + return data, true, nil +} + +// getFile streams an object onto disk through a temporary name: a truncated +// object file would be believed. +func (r *R2) getFile(key, path string) error { + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return err + } + req, err := http.NewRequest(http.MethodGet, r.url(key), nil) + if err != nil { + return err + } + resp, err := r.do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("get %s: %s", key, resp.Status) + } + + tmp, err := os.CreateTemp(filepath.Dir(path), ".syncd*") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + if _, err := io.Copy(tmp, resp.Body); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmp.Name(), path) +} + +func (r *R2) put(key string, data []byte) error { + return r.putBody(key, int64(len(data)), func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(data)), nil + }) +} + +// putFile streams a file up: a packfile can be larger than the container's +// memory. +func (r *R2) putFile(key, path string) error { + info, err := os.Stat(path) + if err != nil { + return err + } + return r.putBody(key, info.Size(), func() (io.ReadCloser, error) { return os.Open(path) }) +} + +func (r *R2) putBody(key string, length int64, open func() (io.ReadCloser, error)) error { + if length == 0 { + // http.NoBody, or ContentLength 0 is sent as unknown and the handler + // rejects it. + open = func() (io.ReadCloser, error) { return http.NoBody, nil } + } + req, err := http.NewRequest(http.MethodPut, r.url(key), nil) + if err != nil { + return err + } + req.ContentLength = length + req.GetBody = open + if req.Body, err = req.GetBody(); err != nil { + return err + } + resp, err := r.do(req) + if err != nil { + return err + } + resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("put %s: %s", key, resp.Status) + } + return nil +} + +func (r *R2) delete(key string) error { + req, err := http.NewRequest(http.MethodDelete, r.url(key), nil) + if err != nil { + return err + } + resp, err := r.do(req) + if err != nil { + return err + } + resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("delete %s: %s", key, resp.Status) + } + return nil +} + +// list follows the cursor to the end: every key or an error, never a page. +func (r *R2) list(prefix string) ([]string, error) { + var keys []string + cursor := "" + for { + u := fmt.Sprintf("%s/?prefix=%s&cursor=%s", r.base, url.QueryEscape(prefix), url.QueryEscape(cursor)) + req, err := http.NewRequest(http.MethodGet, u, nil) + if err != nil { + return nil, err + } + resp, err := r.do(req) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, fmt.Errorf("list %s: %s", prefix, resp.Status) + } + var page struct { + Keys []string `json:"keys"` + Cursor string `json:"cursor"` + } + err = json.NewDecoder(resp.Body).Decode(&page) + resp.Body.Close() + if err != nil { + return nil, fmt.Errorf("list %s: %v", prefix, err) + } + keys = append(keys, page.Keys...) + if page.Cursor == "" { + return keys, nil + } + cursor = page.Cursor + } +} @@ -0,0 +1,1043 @@ +package main + +// The mirror. Objects are content addressed and append only; refs are the +// mutable part, stored as state rather than files, because a restored loose +// ref would win over packed-refs and resurrect an old head. + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io/fs" + "log" + "os" + "os/exec" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +const ( + manifestKey = "manifest.json" + // staleKey records the directories the manifest stopped listing, and when. + staleKey = "stale.json" + // pruneGrace is how long an unlisted repository stays in the bucket: the + // window to put back a manifest restored from a backup or written short. + pruneGrace = 7 * 24 * time.Hour + // pruneEvery is how often a running container looks again, so one that + // never idles still sees the grace out. + pruneEvery = 24 * time.Hour +) + +// SYNCD_LOOSE_LIMIT overrides it, for the test. +const defaultLooseLimit = 500 + +type Sync struct { + r2 *R2 + config string + repos string + staging string + + // One flush at a time, each rescanning, so a request covers its own + // writes. + flushMu sync.Mutex + + stateMu sync.Mutex + manifest map[string]bool // directory names under repos/, as "foo.git" + stored map[string]bool // object keys the bucket already holds + listed map[string]bool // prefixes read back from the bucket once + sent map[string][32]byte // digest per mutable key, written only when it moves + counts map[string]int // object files on disk per prefix, to notice a gc + gone map[string]map[string]bool // keys gone from disk at the last pass, per prefix + + looseLimit int + + hydrating sync.Map // repository path to *sync.Mutex + + // What the idle alarm asks for before it stops the container. + inflight atomic.Int64 +} + +func newSync(r2 *R2, dir string) *Sync { + limit := defaultLooseLimit + if n, err := strconv.Atoi(os.Getenv("SYNCD_LOOSE_LIMIT")); err == nil && n > 0 { + limit = n + } + return &Sync{ + r2: r2, + config: filepath.Join(dir, "config"), + repos: filepath.Join(dir, "repos"), + staging: filepath.Join(dir, ".hydrate"), + + manifest: map[string]bool{}, + stored: map[string]bool{}, + listed: map[string]bool{}, + sent: map[string][32]byte{}, + counts: map[string]int{}, + gone: map[string]map[string]bool{}, + + looseLimit: limit, + } +} + +// hydrated reports whether a directory has been filled in: hydrateOne moves +// objects in last. +func (s *Sync) hydrated(dir string) bool { + _, err := os.Stat(filepath.Join(s.repos, dir, "objects")) + return err == nil +} + +// restore brings back config/ in full and every repository as an empty +// directory, which is all git-forge's startup scan looks at. +func (s *Sync) restore() error { + // Debris from a hydration killed part way. + if err := os.RemoveAll(s.staging); err != nil { + return err + } + for _, d := range []string{s.config, s.repos, s.staging} { + if err := os.MkdirAll(d, 0700); err != nil { + return err + } + } + + data, found, err := s.r2.get(manifestKey) + if err != nil { + return err + } + if !found { + // A forge with no repositories never writes a manifest, and its config + // still has to come back. But an absent manifest also reads as every + // repository deleted, so one is believed only while the bucket holds + // no objects. + keys, err := s.r2.list("repos/") + if err != nil { + return err + } + // Objects, not keys: an empty REFS from a repository created in the + // window before its manifest is not worth refusing a boot over. + objects := 0 + for _, key := range keys { + if strings.Contains(key, "/objects/") { + objects++ + } + } + if objects > 0 { + return fmt.Errorf("no %s, but the bucket holds %d objects under repos/", manifestKey, objects) + } + log.Print("no manifest, so there are no repositories to restore") + return s.restoreConfig() + } + var dirs []string + if err := json.Unmarshal(data, &dirs); err != nil { + return fmt.Errorf("%s: %v", manifestKey, err) + } + for _, d := range dirs { + if !filepath.IsLocal(d) { + return fmt.Errorf("%s: %q is not a directory under repos/", manifestKey, d) + } + if err := os.MkdirAll(filepath.Join(s.repos, d), 0700); err != nil { + return err + } + s.manifest[d] = true + } + return s.restoreConfig() +} + +// restoreConfig is eager: git-forge reads users.json and repos.json before it +// serves anything. +func (s *Sync) restoreConfig() error { + keys, err := s.r2.list("config/") + if err != nil { + return err + } + if len(keys) == 0 { + return nil + } + gitdir := filepath.Join(s.config, ".git") + if err := gitInit(s.config, false); err != nil { + return err + } + want := make([]string, 0, len(keys)) + for _, key := range keys { + rest := strings.TrimPrefix(key, "config/") + // REFS is replayed below, once the objects it names are present. + if rest == "REFS" { + continue + } + // Skipped, not refused: a half written file is a key nothing reads. + if configTemp(rest) { + continue + } + if !configKey(rest) { + return fmt.Errorf("%s: refusing a key syncd did not write", key) + } + want = append(want, key) + } + err = s.fetchAll(want, func(key string) string { + rest := strings.TrimPrefix(key, "config/") + // git/ rather than .git/, so a listing is not full of dotted paths. + if after, ok := strings.CutPrefix(rest, "git/"); ok { + return filepath.Join(gitdir, after) + } + return filepath.Join(s.config, rest) + }) + if err != nil { + return err + } + // From the keys fetched, never the directory: a file only disk holds must + // not be remembered as stored. + for _, key := range want { + rest := strings.TrimPrefix(key, "config/") + if strings.HasPrefix(rest, "git/") { + continue // an object is immutable, so nothing compares it again + } + data, err := os.ReadFile(filepath.Join(s.config, rest)) + if err != nil { + return err + } + s.remember(key, data) + } + if err := s.fetchRefs(gitdir, "config/REFS"); err != nil { + // The log is only a log: one missing an object must not stop the + // forge. The next commit starts a fresh history. + log.Printf("config: audit log not replayed, its history restarts here: %v", err) + } + // Or every restored file looks untracked and the next change commits the + // whole tree. + if _, err := git(gitdir, s.config, nil, "reset", "--mixed", "--quiet"); err != nil { + log.Printf("config: index not reset, the next commit will be noisy: %v", err) + } + return nil +} + +// parallelDo runs fn over items and returns the first error. +func parallelDo(items []string, fn func(string) error) error { + const workers = 16 + if len(items) == 0 { + return nil + } + + jobs := make(chan string) + var wg sync.WaitGroup + var mu sync.Mutex + var failed error + + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + // Drained even after a failure, or the sender deadlocks. + for item := range jobs { + mu.Lock() + stop := failed != nil + mu.Unlock() + if stop { + continue + } + err := fn(item) + mu.Lock() + if err != nil && failed == nil { + failed = err + } + mu.Unlock() + } + }() + } + for _, item := range items { + jobs <- item + } + close(jobs) + wg.Wait() + return failed +} + +// checkKeys refuses a key whose path would leave the directory it is written +// to: only syncd writes this bucket. +func checkKeys(keys []string, prefix string) error { + for _, key := range keys { + if !filepath.IsLocal(strings.TrimPrefix(key, prefix)) { + return fmt.Errorf("%s: refusing a key that points outside %s", key, prefix) + } + } + return nil +} + +// configKey is a key syncd wrote: config/{file} or config/git/objects/{path}. +// Inside the directory is not enough: .git/config and .git/hooks are inside +// it, and git runs both. +func configKey(rest string) bool { + if after, ok := strings.CutPrefix(rest, "git/objects/"); ok { + return filepath.IsLocal(after) + } + return filepath.IsLocal(rest) && !strings.ContainsRune(rest, '/') && !strings.HasPrefix(rest, ".") +} + +// configTemp matches the {name}.tmp{n} git-forge writes before renaming over a +// config file. +func configTemp(name string) bool { + i := strings.LastIndex(name, ".tmp") + if i < 0 { + return false + } + _, err := strconv.Atoi(name[i+len(".tmp"):]) + return err == nil +} + +// fetchAll downloads objects and records that the bucket holds them. +func (s *Sync) fetchAll(keys []string, dest func(key string) string) error { + err := parallelDo(keys, func(key string) error { + return s.r2.getFile(key, dest(key)) + }) + if err != nil { + return err + } + s.stateMu.Lock() + for _, key := range keys { + s.stored[key] = true + } + s.stateMu.Unlock() + return nil +} + +// hydrate fills in one repository and its siblings. Safe for one already +// there, and concurrently. +func (s *Sync) hydrate(base string) error { + value, _ := s.hydrating.LoadOrStore(base, &sync.Mutex{}) + mu := value.(*sync.Mutex) + mu.Lock() + defer mu.Unlock() + + for _, suffix := range []string{".git", ".issues.git", ".wiki.git"} { + dir := base + suffix + s.stateMu.Lock() + want := s.manifest[dir] + s.stateMu.Unlock() + if !want || s.hydrated(dir) { + continue + } + start := time.Now() + n, err := s.hydrateOne(dir) + if err != nil { + return fmt.Errorf("hydrate %s: %v", dir, err) + } + log.Printf("hydrated %s, %d objects in %s", dir, n, time.Since(start).Round(time.Millisecond)) + } + return nil +} + +// hydrateOne builds a repository beside the placeholder and moves it in, +// objects last, so a flush never sees an objects directory with refs not yet +// replayed. Staged outside repos/, which scanRepos claims. +func (s *Sync) hydrateOne(dir string) (int, error) { + tmp, err := os.MkdirTemp(s.staging, "") + if err != nil { + return 0, err + } + // The empty shell after a rename, or the whole failed attempt. + defer os.RemoveAll(tmp) + + path := filepath.Join(tmp, "repo") + if err := gitInit(path, true); err != nil { + return 0, err + } + prefix := "repos/" + dir + "/objects/" + keys, err := s.r2.list(prefix) + if err != nil { + return 0, err + } + if err := checkKeys(keys, prefix); err != nil { + return 0, err + } + err = s.fetchAll(keys, func(key string) string { + return filepath.Join(path, "objects", strings.TrimPrefix(key, prefix)) + }) + if err != nil { + return 0, err + } + // Refs after the objects they name, the order git pushes in. + if err := s.fetchRefs(path, "repos/"+dir+"/REFS"); err != nil { + return 0, err + } + + // The placeholder itself is never moved: a flush that saw it gone would + // publish a deletion. + dest := filepath.Join(s.repos, dir) + entries, err := os.ReadDir(path) + if err != nil { + return 0, err + } + for _, e := range entries { + if e.Name() == "objects" { + continue + } + // Rename refuses an existing directory, so debris from a failed + // attempt would wedge every retry. + target := filepath.Join(dest, e.Name()) + if err := os.RemoveAll(target); err != nil { + return 0, err + } + if err := os.Rename(filepath.Join(path, e.Name()), target); err != nil { + return 0, err + } + } + return len(keys), os.Rename(filepath.Join(path, "objects"), filepath.Join(dest, "objects")) +} + +// fetchRefs replays stored ref state in one exec. +func (s *Sync) fetchRefs(gitdir, key string) error { + data, found, err := s.r2.get(key) + if err != nil || !found { + return err + } + s.remember(key, data) + var cmds bytes.Buffer + for line := range strings.Lines(string(data)) { + id, ref, ok := strings.Cut(strings.TrimSpace(line), " ") + if !ok { + continue + } + fmt.Fprintf(&cmds, "create %s %s\n", ref, id) + } + if cmds.Len() == 0 { + return nil + } + _, err = git(gitdir, "", cmds.Bytes(), "update-ref", "--stdin") + return err +} + +// flush makes everything on disk durable, walking the whole datadir whatever +// provoked it. +func (s *Sync) flush() error { + s.flushMu.Lock() + defer s.flushMu.Unlock() + + uploaded := 0 + n, err := s.flushConfig() + if err != nil { + return err + } + uploaded += n + + dirs, err := scanRepos(s.repos) + if err != nil { + return err + } + for _, dir := range dirs { + // A placeholder would publish empty refs over the real ones. + if !s.hydrated(dir) { + continue + } + n, err := s.flushRepo(dir) + if err != nil { + // A repository deleted under this pass is not a failure; the + // manifest below publishes it. + if !s.hydrated(dir) { + continue + } + return err + } + uploaded += n + } + + if s.manifestMoved(dirs) { + // Last, so a crash leaves a manifest naming only repositories whose + // objects are already up. + data, err := json.Marshal(dirs) + if err != nil { + return err + } + if err := s.r2.put(manifestKey, data); err != nil { + return err + } + // After the put, or one refused write would be permanent. + s.setManifest(dirs) + log.Printf("manifest: %d directories", len(dirs)) + } + if uploaded > 0 { + log.Printf("flushed %d objects", uploaded) + } + return nil +} + +// flushRepo reads the refs before walking the objects and writes them after: +// read afterwards, they could name objects of a push this pass never saw. +func (s *Sync) flushRepo(dir string) (int, error) { + path := filepath.Join(s.repos, dir) + refs, err := readRefs(path) + if err != nil { + return 0, err + } + n, err := s.flushObjects(path, "repos/"+dir+"/objects/") + if err != nil { + return n, err + } + // Empty refs are a repository with no branches, or one being deleted under + // the read; once its objects are gone too, nothing is published. + if len(refs) == 0 && !s.hydrated(dir) { + return n, nil + } + _, err = s.putIfMoved("repos/"+dir+"/REFS", refs) + return n, err +} + +// flushConfig mirrors the config directory, working tree first: a missing +// audit log must not take the account table with it. +func (s *Sync) flushConfig() (int, error) { + // Rewritten in place, so content decides, not presence. + n := 0 + entries, err := os.ReadDir(s.config) + if err != nil { + return n, err + } + for _, e := range entries { + if e.IsDir() || strings.HasPrefix(e.Name(), ".") || configTemp(e.Name()) { + continue + } + data, err := os.ReadFile(filepath.Join(s.config, e.Name())) + if err != nil { + return n, err + } + wrote, err := s.putIfMoved("config/"+e.Name(), data) + if err != nil { + return n, err + } + if wrote { + n++ + } + } + + gitdir := filepath.Join(s.config, ".git") + if _, err := os.Stat(gitdir); err != nil { + return n, nil // git would not initialise; git-forge logs that and runs on + } + // Refs read before the walk, as in flushRepo, and written after the tree: + // here the tree is what a restore reads. + refs, err := readRefs(gitdir) + if err != nil { + return n, err + } + objects, err := s.flushObjects(gitdir, "config/git/objects/") + n += objects + if err != nil { + return n, err + } + _, err = s.putIfMoved("config/REFS", refs) + return n, err +} + +// flushObjects uploads every object the bucket lacks. A detached gc can delete +// a loose object under the walk; git finishes the pack first, so walking again +// finds it. +func (s *Sync) flushObjects(gitdir, prefix string) (int, error) { + uploaded := 0 + for range 2 { + n, vanished, err := s.uploadObjects(gitdir, prefix) + uploaded += n + if err != nil { + return uploaded, err + } + if !vanished { + return uploaded, nil + } + } + return uploaded, fmt.Errorf("%s: objects kept vanishing under the walk", prefix) +} + +// uploadObjects sends what the bucket lacks and reports whether anything +// vanished under it. Objects are never rewritten, so presence is a complete +// answer. +func (s *Sync) uploadObjects(gitdir, prefix string) (int, bool, error) { + if err := s.learn(prefix); err != nil { + return 0, false, err + } + root := filepath.Join(gitdir, "objects") + local, vanished := 0, false + var pending []string + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + // gc removed a fanout directory it emptied; the caller walks again. + if path != root && os.IsNotExist(err) { + vanished = true + return fs.SkipDir + } + return err + } + if d.IsDir() { + return nil + } + rel, relErr := filepath.Rel(root, path) + if relErr != nil { + return relErr + } + if !mirrorable(rel, root) { + return nil + } + local++ + s.stateMu.Lock() + have := s.stored[prefix+filepath.ToSlash(rel)] + s.stateMu.Unlock() + if !have { + pending = append(pending, rel) + } + return nil + }) + // Only the root can be missing: a repository deleted under this pass. + if err != nil && !os.IsNotExist(err) { + return 0, false, err + } + + uploaded := 0 + err = parallelDo(pending, func(rel string) error { + key := prefix + filepath.ToSlash(rel) + if err := s.r2.putFile(key, filepath.Join(root, rel)); err != nil { + if !os.IsNotExist(err) { + return err + } + // gc took it; the caller walks again. + s.stateMu.Lock() + vanished = true + s.stateMu.Unlock() + return nil + } + s.stateMu.Lock() + s.stored[key] = true + uploaded++ + s.stateMu.Unlock() + return nil + }) + if err != nil { + return 0, false, err + } + + // A count below the last pass means a gc ran. Uploads are done, so the + // pack is up before anything is removed. + s.stateMu.Lock() + shrank := s.counts[prefix] > local + s.counts[prefix] = local + // What the last pass left waiting is this pass's to delete. + waiting := len(s.gone[prefix]) > 0 + s.stateMu.Unlock() + if shrank || waiting { + return uploaded, vanished, s.reconcile(gitdir, prefix) + } + return uploaded, vanished, nil +} + +// reconcile removes from the bucket what is no longer on disk, so only for a +// repository restored in full. A key goes on the second pass that finds it +// gone: the pack that replaced it reaches the bucket only after the walk that +// found it. +func (s *Sync) reconcile(gitdir, prefix string) error { + root := filepath.Join(gitdir, "objects") + keys, err := s.r2.list(prefix) + if err != nil { + return err + } + s.stateMu.Lock() + seen := s.gone[prefix] + s.stateMu.Unlock() + + waiting := map[string]bool{} + removed := 0 + for _, key := range keys { + rel := strings.TrimPrefix(key, prefix) + _, err := os.Stat(filepath.Join(root, filepath.FromSlash(rel))) + if err == nil { + continue + } + // This removes durable state, so a disk that cannot answer stops it. + if !os.IsNotExist(err) { + return err + } + if !seen[key] { + waiting[key] = true + continue + } + if err := s.r2.delete(key); err != nil { + return err + } + s.stateMu.Lock() + delete(s.stored, key) + s.stateMu.Unlock() + removed++ + } + // Recorded after the deletes, so a failed one is retried. + s.stateMu.Lock() + s.gone[prefix] = waiting + s.stateMu.Unlock() + if removed > 0 { + log.Printf("reconciled %s, removed %d objects gc had replaced", prefix, removed) + } + return nil +} + +// pruneDeleted removes what the bucket holds for a repository the manifest has +// not listed for pruneGrace. Deleting one removes its directories, so nothing +// walks its prefix again and reconcile never sees it. The delay is what makes +// a regressed manifest recoverable rather than a deletion at the next boot. +// +// The comparison is between the bucket and its own manifest, never a listing of +// the disk. Under flushMu because a new repository's objects go up before the +// manifest naming it. +func (s *Sync) pruneDeleted() error { + s.flushMu.Lock() + defer s.flushMu.Unlock() + + keys, err := s.r2.list("repos/") + if err != nil { + return err + } + stale := map[string][]string{} + for _, key := range keys { + dir, ok := repoDirOf(strings.TrimPrefix(key, "repos/")) + if !ok { + continue // not a key syncd wrote, so not one for this to decide + } + s.stateMu.Lock() + listed := s.manifest[dir] + s.stateMu.Unlock() + if !listed { + stale[dir] = append(stale[dir], key) + } + } + + since := map[string]time.Time{} + data, found, err := s.r2.get(staleKey) + if err != nil { + return err + } + if found { + if err := json.Unmarshal(data, &since); err != nil { + return fmt.Errorf("%s: %v", staleKey, err) + } + } + now := time.Now().UTC().Truncate(time.Second) + waiting := map[string]time.Time{} + for dir, keys := range stale { + first, ok := since[dir] + if !ok { + first = now + } + if now.Sub(first) < pruneGrace { + waiting[dir] = first + log.Printf("stale: %s, %d objects no manifest lists, removed after %s", + dir, len(keys), first.Add(pruneGrace).Format(time.DateOnly)) + continue + } + if err := parallelDo(keys, s.r2.delete); err != nil { + return err + } + s.stateMu.Lock() + for _, key := range keys { + delete(s.stored, key) + } + s.stateMu.Unlock() + log.Printf("pruned %s, %d objects the manifest stopped listing on %s", + dir, len(keys), first.Format(time.DateOnly)) + } + + if len(waiting) == 0 { + if found { + return s.r2.delete(staleKey) + } + return nil + } + data, err = json.Marshal(waiting) + if err != nil { + return err + } + return s.r2.put(staleKey, data) +} + +// repoDirOf names the directory a key under repos/ belongs to, by scanRepos's +// rule. +func repoDirOf(rest string) (string, bool) { + parts := strings.SplitN(rest, "/", 3) + if strings.HasSuffix(parts[0], ".git") { + return parts[0], true + } + if len(parts) == 3 && strings.HasSuffix(parts[1], ".git") { + return parts[0] + "/" + parts[1], true + } + return "", false +} + +// looseObjects counts what is in the two character fanout directories. +func looseObjects(gitdir string) int { + root := filepath.Join(gitdir, "objects") + fanout, err := os.ReadDir(root) + if err != nil { + return 0 + } + n := 0 + for _, d := range fanout { + if !d.IsDir() || len(d.Name()) != 2 { + continue + } + inner, err := os.ReadDir(filepath.Join(root, d.Name())) + if err != nil { + continue + } + n += len(inner) + } + return n +} + +// compactConfig packs the audit log: git's own auto gc waits for thousands of +// loose objects, which is a slow boot. Repositories are git-forge's to gc. +func (s *Sync) compactConfig() { + gitdir := filepath.Join(s.config, ".git") + n := looseObjects(gitdir) + if n < s.looseLimit { + return + } + if _, err := git(gitdir, s.config, nil, "gc", "--quiet"); err != nil { + log.Printf("config: not packed, the audit log will keep growing: %v", err) + return + } + log.Printf("config: packed %d loose objects", n) +} + +func mirrorable(rel, root string) bool { + // Derived and rewritten in place by gc: the one thing here not immutable. + if strings.HasPrefix(rel, "info/") { + return false + } + // Every component: receive-pack's quarantine directory holds ordinarily + // named files. + for _, part := range strings.Split(filepath.ToSlash(rel), "/") { + if strings.HasPrefix(part, "tmp_") || strings.HasPrefix(part, ".") { + return false + } + } + base := filepath.Base(rel) + // receive-pack's .keep is transient; mirrored, it would stop gc repacking + // that pack for ever. + if strings.HasSuffix(base, ".keep") { + return false + } + // A pack is complete only once its index is beside it. + if strings.HasSuffix(base, ".pack") { + if _, err := os.Stat(filepath.Join(root, strings.TrimSuffix(rel, ".pack")+".idx")); err != nil { + return false + } + } + return true +} + +// learn reads what the bucket holds under a prefix, once per boot: after that +// syncd is the only writer. +func (s *Sync) learn(prefix string) error { + s.stateMu.Lock() + known := s.listed[prefix] + s.stateMu.Unlock() + if known { + return nil + } + keys, err := s.r2.list(prefix) + if err != nil { + return err + } + s.stateMu.Lock() + for _, k := range keys { + s.stored[k] = true + } + s.listed[prefix] = true + s.stateMu.Unlock() + return nil +} + +// readRefs is show-ref's output. Exit 1 with nothing printed is no refs; +// everything else is an error, since this is written straight to the bucket. +func readRefs(gitdir string) ([]byte, error) { + out, err := git(gitdir, "", nil, "show-ref") + if err == nil { + return out, nil + } + var exit *exec.ExitError + if errors.As(err, &exit) && exit.ExitCode() == 1 && len(bytes.TrimSpace(out)) == 0 { + return nil, nil + } + return nil, fmt.Errorf("%s: %v", gitdir, err) +} + +// remember records what the bucket holds for a key, so the flush after a +// restore does not rewrite it. +func (s *Sync) remember(key string, data []byte) { + sum := sha256.Sum256(data) + s.stateMu.Lock() + s.sent[key] = sum + s.stateMu.Unlock() +} + +// putIfMoved writes a mutable key only when its content changed. +func (s *Sync) putIfMoved(key string, data []byte) (bool, error) { + sum := sha256.Sum256(data) + s.stateMu.Lock() + seen, ok := s.sent[key] + s.stateMu.Unlock() + if ok && seen == sum { + return false, nil + } + if err := s.r2.put(key, data); err != nil { + return false, err + } + s.stateMu.Lock() + s.sent[key] = sum + s.stateMu.Unlock() + return true, nil +} + +// manifestMoved reports whether the directory set differs from the bucket's. +func (s *Sync) manifestMoved(dirs []string) bool { + s.stateMu.Lock() + defer s.stateMu.Unlock() + if len(dirs) != len(s.manifest) { + return true + } + for _, d := range dirs { + if !s.manifest[d] { + return true + } + } + return false +} + +func (s *Sync) setManifest(dirs []string) { + next := make(map[string]bool, len(dirs)) + for _, d := range dirs { + next[d] = true + } + s.stateMu.Lock() + s.manifest = next + s.stateMu.Unlock() +} + +func (s *Sync) known(base string) bool { + s.stateMu.Lock() + defer s.stateMu.Unlock() + return s.manifest[base+".git"] +} + +// scanRepos lists every git directory under repos/ by git-forge's rule, +// siblings included: they have to be restored too. +func scanRepos(root string) ([]string, error) { + // Never nil: null in the manifest would read as no answer. + dirs := []string{} + top, err := os.ReadDir(root) + if err != nil { + return nil, err + } + for _, entry := range top { + if !entry.IsDir() { + continue + } + if strings.HasSuffix(entry.Name(), ".git") { + dirs = append(dirs, entry.Name()) + continue + } + inner, err := os.ReadDir(filepath.Join(root, entry.Name())) + if err != nil { + // A section can go between the two listings. + if os.IsNotExist(err) { + continue + } + return nil, err + } + for _, e := range inner { + if e.IsDir() && strings.HasSuffix(e.Name(), ".git") { + dirs = append(dirs, entry.Name()+"/"+e.Name()) + } + } + } + slices.Sort(dirs) + return dirs, nil +} + +// sweep backstops the writes no request holds for: token telemetry, and +// anything proxy.go's rule misses. +func (s *Sync) sweep(ctx context.Context) { + // Not in restore: a bucket this cannot list still serves. + if err := s.pruneDeleted(); err != nil { + log.Printf("prune: %v", err) + } + pruned := time.Now() + + t := time.NewTicker(sweepInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + // Packing first, so the flush uploads the pack before removing + // what it replaced. + s.compactConfig() + if err := s.flush(); err != nil { + log.Printf("sweep: %v", err) + } + if time.Since(pruned) >= pruneEvery { + if err := s.pruneDeleted(); err != nil { + log.Printf("prune: %v", err) + } + pruned = time.Now() + } + } + } +} + +// git returns stdout alone; stderr goes into the error, since readRefs stores +// stdout straight into the bucket. System and global configuration are +// neutralised. +func git(gitdir, worktree string, stdin []byte, args ...string) ([]byte, error) { + cmd := exec.Command("git", args...) + cmd.Env = append(os.Environ(), + "GIT_CONFIG_NOSYSTEM=1", + "GIT_CONFIG_GLOBAL=/dev/null", + "GIT_DIR="+gitdir, + ) + if worktree != "" { + cmd.Env = append(cmd.Env, "GIT_WORK_TREE="+worktree) + } + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + var out, fail bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &fail + if err := cmd.Run(); err != nil { + // %w, so readRefs can ask git's exit status. + if msg := bytes.TrimSpace(fail.Bytes()); len(msg) > 0 { + return out.Bytes(), fmt.Errorf("git %s: %w: %s", strings.Join(args, " "), err, msg) + } + return out.Bytes(), fmt.Errorf("git %s: %w", strings.Join(args, " "), err) + } + return out.Bytes(), nil +} + +// gitInit matches git-forge: an empty template and an explicit branch. +func gitInit(dir string, bare bool) error { + tmpl, err := os.MkdirTemp("", "syncd-template") + if err != nil { + return err + } + defer os.RemoveAll(tmpl) + + args := []string{"init", "--quiet", "--template=" + tmpl, "--initial-branch=master"} + if bare { + args = append(args, "--bare") + } + cmd := exec.Command("git", append(args, dir)...) + cmd.Env = append(os.Environ(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null") + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("git init %s: %v: %s", dir, err, out) + } + return nil +} diff --git a/sync_test.go b/sync_test.go new file mode 100644 index 0000000..2158626 --- /dev/null +++ b/sync_test.go @@ -0,0 +1,1113 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// TestMirrorable pins which files under objects/ belong in the bucket. The +// interesting cases are all state git is in the middle of writing. +func TestMirrorable(t *testing.T) { + root := t.TempDir() + + // mirrorable stats the index beside a pack, so these must be real files. + write := func(rel string) { + path := filepath.Join(root, filepath.FromSlash(rel)) + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("x"), 0600); err != nil { + t.Fatal(err) + } + } + write("ab/cdef0123456789") + write("pack/pack-done.pack") + write("pack/pack-done.idx") + write("pack/pack-half.pack") // no index yet + write("pack/pack-held.pack") + write("pack/pack-held.idx") + write("pack/pack-held.keep") // receive-pack still has it open + write("pack/tmp_pack_writing") + write("info/commit-graph") + // receive-pack's quarantine: a temporary directory holding files whose own + // names say nothing about being temporary. + write("tmp_objdir-incoming-HBbblc/pack/pack-incoming.idx") + write("tmp_objdir-incoming-HBbblc/pack/pack-incoming.pack") + write("tmp_objdir-incoming-HBbblc/pack/pack-incoming.keep") + + cases := []struct { + rel string + want bool + }{ + {"ab/cdef0123456789", true}, + {"pack/pack-done.pack", true}, + {"pack/pack-done.idx", true}, + {"pack/pack-half.pack", false}, + {"pack/pack-held.pack", true}, + {"pack/pack-held.keep", false}, + {"pack/tmp_pack_writing", false}, + {"info/commit-graph", false}, + {"tmp_objdir-incoming-HBbblc/pack/pack-incoming.idx", false}, + {"tmp_objdir-incoming-HBbblc/pack/pack-incoming.pack", false}, + {"tmp_objdir-incoming-HBbblc/pack/pack-incoming.keep", false}, + } + for _, c := range cases { + if got := mirrorable(filepath.FromSlash(c.rel), root); got != c.want { + t.Errorf("mirrorable(%q) = %v, want %v", c.rel, got, c.want) + } + } +} + +// TestScanRepos pins what the manifest holds, and so what a restore brings +// back: repositories at the root, one section deep, and the issue and wiki +// siblings. +func TestScanRepos(t *testing.T) { + root := t.TempDir() + for _, d := range []string{ + "notes.git", "notes.issues.git", "notes.wiki.git", + "team/api.git", "team/deeper/nested.git", "empty", + } { + if err := os.MkdirAll(filepath.Join(root, filepath.FromSlash(d)), 0700); err != nil { + t.Fatal(err) + } + } + // Named like a repository and not a directory, so not one. + if err := os.WriteFile(filepath.Join(root, "afile.git"), []byte("x"), 0600); err != nil { + t.Fatal(err) + } + + got, err := scanRepos(root) + if err != nil { + t.Fatal(err) + } + // team/deeper/nested.git is absent on purpose: a section is one level deep, + // which is the rule git-forge scans by. + want := "notes.git notes.issues.git notes.wiki.git team/api.git" + if strings.Join(got, " ") != want { + t.Errorf("scanRepos = %q, want %q", strings.Join(got, " "), want) + } +} + +// TestListFollowsTheCursor pins that a caller gets every key rather than the +// first page. A repository with more than a thousand loose objects is ordinary, +// and a listing that stops short is the one input that turns a hydration into a +// deletion: what it did not fetch reads afterwards as an object gc replaced. +func TestListFollowsTheCursor(t *testing.T) { + b, url := newBucket(t) + const prefix = "repos/x.git/objects/" + b.mu.Lock() + for i := range bucketPage*3 + 1 { + b.objs[fmt.Sprintf("%s%02d/object", prefix, i)] = []byte("x") + } + b.mu.Unlock() + + keys, err := newR2(url).list(prefix) + if err != nil { + t.Fatal(err) + } + if len(keys) != bucketPage*3+1 { + t.Errorf("a listing of %d keys came back with %d", bucketPage*3+1, len(keys)) + } +} + +// TestCheckKeys pins the one thing standing between a tampered bucket and a +// file written outside the datadir. +func TestCheckKeys(t *testing.T) { + const prefix = "repos/x.git/objects/" + cases := []struct { + key string + want bool + }{ + {prefix + "ab/cdef0123456789", true}, + {prefix + "pack/pack-1.pack", true}, + {prefix + "../../../../etc/passwd", false}, + {prefix + "/etc/passwd", false}, + {prefix, false}, + } + for _, c := range cases { + if got := checkKeys([]string{c.key}, prefix) == nil; got != c.want { + t.Errorf("checkKeys(%q) accepted = %v, want %v", c.key, got, c.want) + } + } +} + +// How many keys a listing answers at once. See the list case below. +const bucketPage = 25 + +// bucket is the object handler from worker.js, in memory. Object gets can be +// held open and any request made to fail, which turns races into tests. +type bucket struct { + mu sync.Mutex + objs map[string][]byte + failing func(method, key string) bool + gate chan struct{} // object gets wait on this while it is not nil + blocked chan struct{} // closed once one has + once sync.Once +} + +func newBucket(t *testing.T) (*bucket, string) { + t.Helper() + b := &bucket{objs: map[string][]byte{}, blocked: make(chan struct{})} + srv := httptest.NewServer(b) + t.Cleanup(srv.Close) + return b, srv.URL +} + +func (b *bucket) ServeHTTP(w http.ResponseWriter, r *http.Request) { + key := strings.TrimPrefix(r.URL.Path, "/") + + b.mu.Lock() + failing, gate := b.failing, b.gate + b.mu.Unlock() + if failing != nil && failing(r.Method, key) { + http.Error(w, "the bucket is having a moment", http.StatusInternalServerError) + return + } + if gate != nil && strings.Contains(key, "/objects/") { + b.once.Do(func() { close(b.blocked) }) + <-gate + } + + switch { + case r.Method == http.MethodGet && key == "": + prefix := r.URL.Query().Get("prefix") + keys := []string{} + b.mu.Lock() + for k := range b.objs { + if strings.HasPrefix(k, prefix) { + keys = append(keys, k) + } + } + b.mu.Unlock() + // Paged, because r2 pages at a thousand and a listing that stops short + // hydrates part of a repository, which the pass after it reads as + // objects gc replaced and deletes. Small, so every test that lists + // anything with a repository in it follows a cursor. + slices.Sort(keys) + from := 0 + if c := r.URL.Query().Get("cursor"); c != "" { + from, _ = strconv.Atoi(c) + } + to := min(from+bucketPage, len(keys)) + next := "" + if to < len(keys) { + next = strconv.Itoa(to) + } + _ = json.NewEncoder(w).Encode(map[string]any{"keys": keys[from:to], "cursor": next}) + case r.Method == http.MethodGet: + b.mu.Lock() + data, found := b.objs[key] + b.mu.Unlock() + if !found { + http.Error(w, "not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Length", strconv.Itoa(len(data))) + _, _ = w.Write(data) + case r.Method == http.MethodPut: + data, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + b.mu.Lock() + b.objs[key] = data + b.mu.Unlock() + w.WriteHeader(http.StatusNoContent) + case r.Method == http.MethodDelete: + b.mu.Lock() + delete(b.objs, key) + b.mu.Unlock() + w.WriteHeader(http.StatusNoContent) + default: + http.Error(w, "bad method", http.StatusMethodNotAllowed) + } +} + +// hold blocks every object get until release, and reports when the first one is +// waiting, which is a repository part way through being hydrated. +func (b *bucket) hold() <-chan struct{} { + b.mu.Lock() + defer b.mu.Unlock() + b.gate = make(chan struct{}) + return b.blocked +} + +func (b *bucket) release() { + b.mu.Lock() + gate := b.gate + b.gate = nil + b.mu.Unlock() + close(gate) +} + +func (b *bucket) fail(when func(method, key string) bool) { + b.mu.Lock() + b.failing = when + b.mu.Unlock() +} + +func (b *bucket) object(key string) string { + b.mu.Lock() + defer b.mu.Unlock() + return string(b.objs[key]) +} + +func (b *bucket) under(prefix string) []string { + b.mu.Lock() + defer b.mu.Unlock() + var keys []string + for k := range b.objs { + if strings.HasPrefix(k, prefix) { + keys = append(keys, k) + } + } + return keys +} + +// gitIdentity keeps these tests independent of whoever runs them: git refuses +// to write a commit without one. +func gitIdentity(t *testing.T) { + t.Helper() + for _, v := range []string{"GIT_AUTHOR_NAME", "GIT_COMMITTER_NAME"} { + t.Setenv(v, "syncd test") + } + for _, v := range []string{"GIT_AUTHOR_EMAIL", "GIT_COMMITTER_EMAIL"} { + t.Setenv(v, "test@example.com") + } +} + +// seed leaves the bucket holding what a forge that had taken one push would +// have, through the same flush, and returns the commit master points at. +func seed(t *testing.T, b *bucket, url string) string { + t.Helper() + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + repo := filepath.Join(dir, "repos", "x.git") + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + head := commitOne(t, repo) + if err := s.flush(); err != nil { + t.Fatal(err) + } + if refs := b.object("repos/x.git/REFS"); !strings.Contains(refs, head) { + t.Fatalf("the seed did not store the refs, the bucket holds %q", refs) + } + return head +} + +// commitOne puts one empty commit on master and returns it. +func commitOne(t *testing.T, repo string) string { + t.Helper() + tree, err := git(repo, "", []byte{}, "mktree") + if err != nil { + t.Fatalf("mktree: %v: %s", err, tree) + } + commit, err := git(repo, "", nil, "commit-tree", strings.TrimSpace(string(tree)), "-m", "first") + if err != nil { + t.Fatalf("commit-tree: %v: %s", err, commit) + } + head := strings.TrimSpace(string(commit)) + if out, err := git(repo, "", nil, "update-ref", "refs/heads/master", head); err != nil { + t.Fatalf("update-ref: %v: %s", err, out) + } + return head +} + +// commitBlobs puts one commit holding n blobs on master. Enough of them land in +// most of the fanout directories, which a gc racing a walk needs. +func commitBlobs(t *testing.T, repo string, n int) string { + t.Helper() + var paths, entries strings.Builder + dir := t.TempDir() + for i := range n { + name := filepath.Join(dir, fmt.Sprintf("f%d", i)) + if err := os.WriteFile(name, fmt.Appendf(nil, "content %d\n", i), 0600); err != nil { + t.Fatal(err) + } + paths.WriteString(name + "\n") + } + blobs, err := git(repo, "", []byte(paths.String()), "hash-object", "-w", "--stdin-paths") + if err != nil { + t.Fatalf("hash-object: %v: %s", err, blobs) + } + for i, sha := range strings.Fields(string(blobs)) { + fmt.Fprintf(&entries, "100644 blob %s\tf%d\n", sha, i) + } + tree, err := git(repo, "", []byte(entries.String()), "mktree") + if err != nil { + t.Fatalf("mktree: %v: %s", err, tree) + } + commit, err := git(repo, "", nil, "commit-tree", strings.TrimSpace(string(tree)), "-m", "many") + if err != nil { + t.Fatalf("commit-tree: %v: %s", err, commit) + } + head := strings.TrimSpace(string(commit)) + if out, err := git(repo, "", nil, "update-ref", "refs/heads/master", head); err != nil { + t.Fatalf("update-ref: %v: %s", err, out) + } + return head +} + +// Pruning waits a pass: deleting on the pass that first notices an object gone +// can take one whose replacement is still on disk alone. +func TestPruneWaitsForAPass(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + repo := filepath.Join(dir, "repos", "x.git") + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + head := commitOne(t, repo) + if err := s.flush(); err != nil { + t.Fatal(err) + } + out, err := git(repo, "", nil, "rev-parse", head+"^{tree}") + if err != nil { + t.Fatalf("rev-parse: %v: %s", err, out) + } + tree := strings.TrimSpace(string(out)) + key := "repos/x.git/objects/" + tree[:2] + "/" + tree[2:] + if b.object(key) == "" { + t.Fatalf("the flush did not store the tree at %s", key) + } + + // What gc does to a loose object the moment it has packed it. + if err := os.Remove(filepath.Join(repo, "objects", tree[:2], tree[2:])); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if b.object(key) == "" { + t.Fatal("the pass that first found the object gone deleted it") + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if b.object(key) != "" { + t.Error("the pass after that did not delete it") + } +} + +// A repository the manifest no longer lists leaves the bucket after the grace, +// not at the first boot that notices: a manifest restored from a backup, or +// written short, would otherwise take every repository it omits with it. +func TestPruneWaitsOutTheGrace(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + heads := map[string]string{} + for _, name := range []string{"keep.git", "drop.git"} { + repo := filepath.Join(dir, "repos", name) + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + heads[name] = commitOne(t, repo) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + + // What DeleteRepo leaves behind: the directory gone and the manifest + // rewritten without it, while everything it named is still in the bucket. + if err := os.RemoveAll(filepath.Join(dir, "repos", "drop.git")); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if n := len(b.under("repos/drop.git/")); n == 0 { + t.Fatal("the flush that rewrote the manifest already removed the objects") + } + + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if err := f.pruneDeleted(); err != nil { + t.Fatal(err) + } + if len(b.under("repos/drop.git/")) == 0 { + t.Fatal("the first boot to find it unlisted removed it") + } + if !strings.Contains(b.object(staleKey), "drop.git") { + t.Fatalf("the bucket does not record when it went unlisted: %q", b.object(staleKey)) + } + + // Listed again, by a manifest put back: the record goes and nothing is + // removed. + f.setManifest([]string{"keep.git", "drop.git"}) + if err := f.pruneDeleted(); err != nil { + t.Fatal(err) + } + if got := b.object(staleKey); got != "" { + t.Errorf("a repository listed again is still recorded as stale: %q", got) + } + if len(b.under("repos/drop.git/")) == 0 { + t.Fatal("a repository listed again was removed") + } + + // Unlisted for longer than the grace. + f.setManifest([]string{"keep.git"}) + b.mu.Lock() + b.objs[staleKey] = []byte(`{"drop.git":"2020-01-01T00:00:00Z"}`) + b.mu.Unlock() + if err := f.pruneDeleted(); err != nil { + t.Fatal(err) + } + if keys := b.under("repos/drop.git/"); len(keys) > 0 { + t.Errorf("the bucket still holds %d keys for a repository unlisted past the grace: %v", len(keys), keys) + } + if got := b.object(staleKey); got != "" { + t.Errorf("the record outlived the deletion: %q", got) + } + + // The question worth asking of anything that deletes: whether it took + // something still reachable. + if keys := b.under("repos/keep.git/"); len(keys) == 0 { + t.Fatal("a repository the manifest lists was pruned too") + } + if err := f.hydrate("keep"); err != nil { + t.Fatalf("hydrating what the prune left: %v", err) + } + out, _ := git(filepath.Join(fresh, "repos", "keep.git"), "", nil, "show-ref") + if !strings.Contains(string(out), heads["keep.git"]) { + t.Errorf("the repository that survived is missing its refs, show-ref says %q", out) + } +} + +// gc packs loose objects and deletes them, detached, while the flush a push +// triggered is still uploading. Skipping what it took would write refs naming +// objects the bucket never received, so the flush has to walk again. +func TestFlushWhileGitRepacks(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + repo := filepath.Join(dir, "repos", "x.git") + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + + // Enough reachable objects that the uploads outlast the walk: sixteen are in + // flight at a time, so the rest are still queued when repack runs. + commitBlobs(t, repo, 60) + + blocked := b.hold() + done := make(chan error, 1) + go func() { done <- s.flush() }() + <-blocked + if out, err := git(repo, "", nil, "repack", "-a", "-d", "-q"); err != nil { + t.Fatalf("repack: %v: %s", err, out) + } + b.release() + if err := <-done; err != nil { + t.Fatalf("flush: %v", err) + } + + // Not what the bucket holds but whether it is a repository, which only a + // fresh container answers. + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if err := f.hydrate("x"); err != nil { + t.Fatalf("hydrating what the flush published: %v", err) + } + out, err := git(filepath.Join(fresh, "repos", "x.git"), "", nil, "rev-list", "--objects", "--all") + if err != nil { + t.Fatalf("the published repository is missing objects its refs name: %v: %s", err, out) + } + if n := len(strings.Split(strings.TrimSpace(string(out)), "\n")); n < 61 { + t.Errorf("only %d objects came back, want the commit, its tree and sixty blobs", n) + } +} + +// flushRepo reads the ref state before it walks the objects. A push that lands +// during the upload is left to the next flush rather than published by this +// one: refs read afterwards would name a commit the walk never saw. +func TestFlushWhileAPushLands(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + repo := filepath.Join(dir, "repos", "x.git") + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + // Enough that the uploads are still running when the commit below lands. + head := commitBlobs(t, repo, 60) + + blocked := b.hold() + done := make(chan error, 1) + go func() { done <- s.flush() }() + <-blocked + + // The walk is over by the time the first upload blocks, so this commit is + // exactly what a push arriving mid-flush leaves behind. + late := commitOne(t, repo) + b.release() + if err := <-done; err != nil { + t.Fatalf("flush: %v", err) + } + + refs := b.object("repos/x.git/REFS") + if strings.Contains(refs, late) { + t.Fatalf("the flush published a commit it did not walk, refs hold %q", refs) + } + if !strings.Contains(refs, head) { + t.Fatalf("the flush published no refs at all, the bucket holds %q", refs) + } + + // Whether the published state is servable, which only a fresh container + // answers. + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if err := f.hydrate("x"); err != nil { + t.Fatalf("hydrating what the flush published: %v", err) + } + out, err := git(filepath.Join(fresh, "repos", "x.git"), "", nil, "rev-list", "--objects", head) + if err != nil { + t.Fatalf("the published repository is missing objects its refs name: %v: %s", err, out) + } +} + +// The other half of a detached gc: it removes the fanout directories it +// emptied, and a walk inside one when it goes stops there. Treating that as a +// finished walk publishes refs naming objects the bucket never received. +// +// The window is small, so this takes several rounds. A flush that reports the +// walk was short is a correct answer; success over a bucket that cannot serve +// what it published is not. +func TestFlushWhileGitPrunes(t *testing.T) { + gitIdentity(t) + + published := 0 + for round := range 8 { + _, url := newBucket(t) + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + repo := filepath.Join(dir, "repos", "x.git") + if err := gitInit(repo, true); err != nil { + t.Fatal(err) + } + head := commitBlobs(t, repo, 600) + + // Where gc is the moment before it prunes: everything reachable is in a + // pack, and the loose copies it replaced are still on disk. + if out, err := git(repo, "", nil, "repack", "-a", "-q"); err != nil { + t.Fatalf("repack: %v: %s", err, out) + } + + // prune-packed alongside the flush: the listing the flush does first is + // long enough for this to be well under way. + done := make(chan struct{}) + go func() { + defer close(done) + root := filepath.Join(repo, "objects") + fanout, err := os.ReadDir(root) + if err != nil { + return + } + for _, d := range fanout { + if d.IsDir() && len(d.Name()) == 2 { + _ = os.RemoveAll(filepath.Join(root, d.Name())) + time.Sleep(20 * time.Microsecond) + } + } + }() + err := s.flush() + <-done + if err != nil { + continue + } + published++ + + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if err := f.hydrate("x"); err != nil { + t.Fatalf("round %d: the flush published refs the bucket cannot serve: %v", round, err) + } + out, err := git(filepath.Join(fresh, "repos", "x.git"), "", nil, "rev-list", "--objects", head) + if err != nil { + t.Fatalf("round %d: the published repository is missing objects its refs name: %v: %s", round, err, out) + } + } + // Every round refusing is a pass that checked nothing, and a flush that + // always refuses is its own bug. + if published == 0 { + t.Error("no round got as far as publishing, so nothing above was checked") + } +} + +// Deleting a repository must not fail the flush of everything else. DeleteRepo +// removes the directory, and the section with it when it held the last one, so +// a pass can list one and find it gone by the time it reads it. +// +// The directories are renamed away rather than removed, so each stops existing +// at an instant: that is the moment the pass has to survive. A removal caught +// half done leaves something git will not read, which flush still fails on. +// +// The deletions start when the bucket sees the first upload, so they land while +// the pass is inside the repositories. +func TestFlushWhileRepositoriesAreDeleted(t *testing.T) { + gitIdentity(t) + + for round := range 20 { + b := &bucket{objs: map[string][]byte{}, blocked: make(chan struct{})} + dir := t.TempDir() + away := t.TempDir() + + var doomed []string + var once sync.Once + upload, deleted := make(chan struct{}), make(chan struct{}) + release := func() { once.Do(func() { close(upload) }) } + go func() { + defer close(deleted) + <-upload + for i, p := range doomed { + _ = os.Rename(p, filepath.Join(away, strconv.Itoa(i))) + // The section goes with the last repository in it. + _ = os.Remove(filepath.Dir(p)) + time.Sleep(time.Millisecond) + } + }() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut && strings.Contains(r.URL.Path, "/repos/") { + release() + } + b.ServeHTTP(w, r) + })) + + s := newSync(newR2(srv.URL), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + keep := filepath.Join(dir, "repos", "keep.git") + if err := gitInit(keep, true); err != nil { + t.Fatal(err) + } + head := commitOne(t, keep) + // Sorted after keep.git, so the pass is still working when they go. + for i := range 12 { + p := filepath.Join(dir, "repos", fmt.Sprintf("team%d", i), "api.git") + if err := gitInit(p, true); err != nil { + t.Fatal(err) + } + commitOne(t, p) + doomed = append(doomed, p) + } + + err := s.flush() + // The deleter has to be done with the round's directories before + // t.TempDir takes them away underneath it. + srv.Close() + release() + <-deleted + if err != nil { + t.Fatalf("round %d: a deletion elsewhere failed the flush: %v", round, err) + } + if refs := b.object("repos/keep.git/REFS"); !strings.Contains(refs, head) { + t.Fatalf("round %d: the repository that stayed is missing its refs, the bucket holds %q", round, refs) + } + } +} + +// A repository being hydrated must be invisible to a flush: an objects +// directory that git init has just made holds no refs yet, and a flush that saw +// one would write that emptiness over the real ref state. +func TestFlushDuringHydration(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + head := seed(t, b, url) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + + blocked := b.hold() + done := make(chan error, 1) + go func() { done <- s.hydrate("x") }() + <-blocked + + if err := s.flush(); err != nil { + t.Fatal(err) + } + b.release() + if err := <-done; err != nil { + t.Fatal(err) + } + + if refs := b.object("repos/x.git/REFS"); !strings.Contains(refs, head) { + t.Errorf("the flush overwrote the stored refs, the bucket holds %q", refs) + } + out, _ := git(filepath.Join(dir, "repos", "x.git"), "", nil, "show-ref") + if !strings.Contains(string(out), head) { + t.Errorf("the hydrated repository has no refs, show-ref says %q", out) + } +} + +// A move into the placeholder that failed part way must not wedge every attempt +// after it: rename refuses an existing directory, so the debris of one failure +// would otherwise be permanent for the life of the container. +func TestHydrationAfterPartialMove(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + head := seed(t, b, url) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + // What a rename that died in the middle of the loop leaves: part of a + // repository, no objects directory, so still a placeholder. + for _, d := range []string{"refs/heads", "info"} { + if err := os.MkdirAll(filepath.Join(dir, "repos", "x.git", d), 0700); err != nil { + t.Fatal(err) + } + } + headfile := filepath.Join(dir, "repos", "x.git", "HEAD") + if err := os.WriteFile(headfile, []byte("ref: refs/heads/master\n"), 0600); err != nil { + t.Fatal(err) + } + + if err := s.hydrate("x"); err != nil { + t.Fatalf("hydrating over the debris: %v", err) + } + out, _ := git(filepath.Join(dir, "repos", "x.git"), "", nil, "show-ref") + if !strings.Contains(string(out), head) { + t.Errorf("the repository was not hydrated, show-ref says %q", out) + } +} + +// A manifest write that failed must be carried up by the next flush: recording +// it before the put would make one refused write permanent. +func TestManifestRetriedAfterFailure(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + if err := gitInit(filepath.Join(dir, "repos", "x.git"), true); err != nil { + t.Fatal(err) + } + + b.fail(func(method, key string) bool { + return method == http.MethodPut && key == manifestKey + }) + if err := s.flush(); err == nil { + t.Fatal("a flush whose manifest the bucket refused reported success") + } + b.fail(nil) + + if err := s.flush(); err != nil { + t.Fatal(err) + } + if got := b.object(manifestKey); !strings.Contains(got, "x.git") { + t.Errorf("the repository is missing from the manifest, which holds %q", got) + } +} + +// A forge with no repositories writes no manifest, because the directory set +// never moved. A restore has to tell that apart from an empty bucket, or the +// next container comes up with no accounts and flushes that over the real ones. +func TestConfigRestoredWithoutManifest(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + if err := gitInit(filepath.Join(dir, "config"), false); err != nil { + t.Fatal(err) + } + users := `{"alice":{}}` + if err := os.WriteFile(filepath.Join(dir, "config", "users.json"), []byte(users), 0600); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if got := b.object(manifestKey); got != "" { + t.Fatalf("a forge with no repositories wrote a manifest holding %q", got) + } + + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(filepath.Join(fresh, "config", "users.json")) + if err != nil { + t.Fatalf("the account table did not come back: %v", err) + } + if string(got) != users { + t.Fatalf("the account table came back as %q", got) + } + if err := f.flush(); err != nil { + t.Fatal(err) + } + if got := b.object("config/users.json"); got != users { + t.Fatalf("the flush after the restore left %q in the bucket", got) + } +} + +// An absent manifest means no repositories, which is true of a forge that has +// none and a lie about a bucket that has lost one. Believing the lie hands the +// sweep every repository in the bucket as one nothing lists. +func TestRestoreRefusesAMissingManifest(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + seed(t, b, url) + + b.mu.Lock() + delete(b.objs, manifestKey) + b.mu.Unlock() + + f := newSync(newR2(url), t.TempDir()) + if err := f.restore(); err == nil { + t.Fatal("a restore believed a bucket that holds repositories and no manifest") + } + if keys := b.under("repos/x.git/"); len(keys) == 0 { + t.Error("the repository was pruned anyway") + } + + // And the case it has to stay true for: a forge that never had one. + empty, emptyURL := newBucket(t) + empty.mu.Lock() + empty.objs["config/users.json"] = []byte(`{"alice":{}}`) + empty.mu.Unlock() + if err := newSync(newR2(emptyURL), t.TempDir()).restore(); err != nil { + t.Errorf("a forge with no repositories was refused: %v", err) + } +} + +// What a restore knows the bucket holds has to come from the keys it fetched. +// Taking it from the directory instead would mean a file the bucket does not +// hold is remembered as stored, and the flush after it leaves it on a disk that +// does not survive. +func TestConfigOnDiskOnlyIsStillMirrored(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + + seedDir := t.TempDir() + s := newSync(newR2(url), seedDir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + if err := gitInit(filepath.Join(seedDir, "config"), false); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(seedDir, "config", "users.json"), []byte(`{"alice":{}}`), 0600); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + + // A datadir that outlived the syncd that wrote it, holding a file whose + // flush never succeeded. + dirty := t.TempDir() + if err := os.MkdirAll(filepath.Join(dirty, "config"), 0700); err != nil { + t.Fatal(err) + } + repos := `{"notes":{"visibility":"public"}}` + if err := os.WriteFile(filepath.Join(dirty, "config", "repos.json"), []byte(repos), 0600); err != nil { + t.Fatal(err) + } + + f := newSync(newR2(url), dirty) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if err := f.flush(); err != nil { + t.Fatal(err) + } + if got := b.object("config/repos.json"); got != repos { + t.Errorf("the bucket holds %q for a file only disk had", got) + } +} + +// git-forge writes a config file by creating {name}.tmp{n} beside it and +// renaming over the top, so a pass can catch one between the two. Mirroring it +// stores a half written account table under a key nothing removes. +func TestConfigTempFilesAreNotMirrored(t *testing.T) { + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + users := `{"alice":{}}` + if err := os.WriteFile(filepath.Join(dir, "config", "users.json"), []byte(users), 0600); err != nil { + t.Fatal(err) + } + // What git-forge's writeFile leaves beside it for the length of an fsync. + temp := "users.json.tmp3877658021" + if err := os.WriteFile(filepath.Join(dir, "config", temp), []byte(`{"alice":{"par`), 0600); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if got := b.object("config/" + temp); got != "" { + t.Errorf("the bucket holds a file git-forge was still writing: %q", got) + } + if got := b.object("config/users.json"); got != users { + t.Errorf("the account table is not in the bucket, which holds %q", got) + } + + // And one already there is left where it is rather than restored, because + // putting it back on disk is what makes it permanent. + b.mu.Lock() + b.objs["config/"+temp] = []byte(`{"alice":{"par`) + b.mu.Unlock() + + fresh := t.TempDir() + f := newSync(newR2(url), fresh) + if err := f.restore(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(fresh, "config", temp)); err == nil { + t.Error("a restore wrote a half written config file back to disk") + } + if got, err := os.ReadFile(filepath.Join(fresh, "config", "users.json")); err != nil || string(got) != users { + t.Errorf("the account table did not come back: %q: %v", got, err) + } +} + +// Only syncd writes this bucket, so a key of a shape it never writes means the +// bucket has been tampered with. +func TestRestoreRefusesAKeyItDidNotWrite(t *testing.T) { + for _, key := range []string{ + "config/.git/config", + "config/.git/hooks/pre-commit", + "config/git/hooks/pre-commit", + "config/sub/dir/users.json", + } { + b, url := newBucket(t) + b.mu.Lock() + b.objs["config/users.json"] = []byte(`{}`) + b.objs[key] = []byte("#!/bin/sh\nid\n") + b.mu.Unlock() + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err == nil { + t.Errorf("a restore believed %s", key) + } + } +} + +// git-forge serves without an audit log when git will not initialise. The +// working tree still has to be mirrored: it is what a restore reads. +func TestConfigFlushedWithoutAuditLog(t *testing.T) { + b, url := newBucket(t) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + users := `{"alice":{}}` + if err := os.WriteFile(filepath.Join(dir, "config", "users.json"), []byte(users), 0600); err != nil { + t.Fatal(err) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if got := b.object("config/users.json"); got != users { + t.Errorf("the account table is not in the bucket, which holds %q", got) + } +} + +// A hydration that fails part way must leave the placeholder alone, so the next +// request starts over. An objects directory left behind counts as hydrated. +func TestHydrationRetriedAfterFailure(t *testing.T) { + gitIdentity(t) + b, url := newBucket(t) + head := seed(t, b, url) + + dir := t.TempDir() + s := newSync(newR2(url), dir) + if err := s.restore(); err != nil { + t.Fatal(err) + } + + b.fail(func(method, key string) bool { + return method == http.MethodGet && strings.Contains(key, "/objects/") + }) + if err := s.hydrate("x"); err == nil { + t.Fatal("a hydration that could not read the objects reported success") + } + b.fail(nil) + + if err := s.hydrate("x"); err != nil { + t.Fatalf("second attempt: %v", err) + } + out, _ := git(filepath.Join(dir, "repos", "x.git"), "", nil, "show-ref") + if !strings.Contains(string(out), head) { + t.Errorf("the second request did not hydrate it, show-ref says %q", out) + } + if err := s.flush(); err != nil { + t.Fatal(err) + } + if refs := b.object("repos/x.git/REFS"); !strings.Contains(refs, head) { + t.Errorf("the flush that followed overwrote the stored refs with %q", refs) + } +} @@ -0,0 +1,274 @@ +#!/bin/sh +# Drives the real image end to end against a local object store: the container +# is destroyed and the forge comes back from the bucket alone. +# +# Needs docker, go, git, curl and openssl here, nothing extra inside the image. +# fake-r2.go stands in for the object handler in worker.js, so no Cloudflare +# account is needed. +set -eu + +IMAGE=${IMAGE:-git-forge-cf:smoke} +PORT=${PORT:-18081} +R2PORT=${R2PORT:-19000} +BASE=http://127.0.0.1:$PORT +NAME=git-forge-cf-smoke + +WORK=$(mktemp -d) +BUCKET=$WORK/bucket +R2PID="" +cleanup() { + docker rm -f "$NAME" >/dev/null 2>&1 || true + [ -n "$R2PID" ] && kill "$R2PID" 2>/dev/null || true + rm -rf "$WORK" +} +trap cleanup EXIT + +ok() { printf 'ok %s\n' "$1"; } +no() { + printf 'FAIL %s\n' "$1" >&2 + docker logs "$NAME" 2>&1 | tail -25 >&2 || true + exit 1 +} +want() { grep -q -- "$2" "$3" || no "$1"; ok "$1"; } +code() { + description=$1 expected=$2 + shift 2 + got=$(curl -s -o "$WORK/body" -w '%{http_code}' "$@") + [ "$got" = "$expected" ] || no "$description: got $got, wanted $expected" + ok "$description" +} +csrf() { grep -o 'name="csrf" value="[^"]*"' "$1" | head -1 | cut -d'"' -f4; } + +# Same trick git-forge's own test uses: the secret is the base32 of the ascii +# "12345678901234567890", so a code needs openssl and no base32 decoder. +SECRET=GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ +KEYHEX=3132333435363738393031323334353637383930 +totp() { + step=$(( $(date -u +%s) / 30 )) + # The counter as eight raw bytes. Octal escapes rather than \x, which is not + # POSIX and which dash does not interpret. + msg="" + for byte in $(printf '%016x' "$step" | sed 's/../& /g'); do + msg="$msg$(printf '\\%03o' "$((0x$byte))")" + done + digest=$(printf "$msg" | openssl dgst -sha1 -mac HMAC -macopt hexkey:$KEYHEX | sed 's/.*= *//') + offset=$(( 0x$(printf %s "$digest" | cut -c40) * 2 + 1 )) + chunk=$(printf %s "$digest" | cut -c${offset}-$((offset + 7))) + printf '%06d' $(( (0x$chunk & 0x7fffffff) % 1000000 )) +} + +export GIT_TERMINAL_PROMPT=0 GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null +export GIT_AUTHOR_NAME=Smoke GIT_AUTHOR_EMAIL=smoke@example.com +export GIT_COMMITTER_NAME=Smoke GIT_COMMITTER_EMAIL=smoke@example.com +# alice's cookie jar, as two curl flags. Left unquoted at every use on purpose. +alice="-c $WORK/alice -b $WORK/alice" + +# start <description> brings up a container against the same bucket. No volume, +# so the datadir it starts from is whatever it can fetch. +# +# GIT_FORGE_DATADIR is deliberately wrong: syncd mirrors /var/lib/git-forge and +# is what tells git-forge where to write, so a git-forge that took this value +# would serve normally while every flush went nowhere. +start() { + docker run -d --name "$NAME" \ + -p "$PORT:8081" \ + --add-host=host.docker.internal:host-gateway \ + -e SYNCD_R2="http://host.docker.internal:$R2PORT" \ + -e GIT_FORGE_DATADIR=/tmp/not-mirrored \ + -e GIT_FORGE_BASE_URL="$BASE" \ + -e GIT_FORGE_COOKIE_KEY=0123456789abcdef0123456789abcdef \ + -e SYNCD_LOOSE_LIMIT=1 \ + -e SYNCD_LOGS="http://host.docker.internal:$R2PORT/" \ + "$IMAGE" >/dev/null + i=0 + while [ $i -lt 100 ]; do + if curl -fsS "$BASE/healthz" >/dev/null 2>&1; then + ok "$1" + return + fi + i=$((i + 1)) + sleep 0.2 + done + docker logs "$NAME" >&2 || true + no "$1" +} + +echo "== build" +./fetch-forge >/dev/null +docker build -q -f Containerfile -t "$IMAGE" . >/dev/null +ok "the image builds" + +# Built rather than `go run`, so that R2PID is the server and not a wrapper +# whose child would outlive the kill and hold the port. +go build -o "$WORK/fake-r2" fake-r2.go +"$WORK/fake-r2" -dir "$BUCKET" -listen ":$R2PORT" >"$WORK/r2.log" 2>&1 & +R2PID=$! +sleep 1 +[ -d "$BUCKET" ] || no "the object store did not start" +ok "the object store is up" + +echo "== first start, on an empty bucket" +start "the container starts with nothing to restore" + +docker logs "$NAME" 2>&1 | grep -o 'token: .*' | cut -d' ' -f2 >"$WORK/tok" +TOKEN=$(cat "$WORK/tok") +[ -n "$TOKEN" ] || no "no setup token in the log" +curl -s $alice "$BASE/setup" >"$WORK/p" +code "the setup form makes the first admin" 303 $alice -o /dev/null \ + -d "csrf=$(csrf "$WORK/p")&token=$TOKEN&name=alice&password=hunter2hunter2&again=hunter2hunter2" \ + "$BASE/setup" + +curl -s $alice "$BASE/settings/totp" >"$WORK/p" +curl -s $alice -o /dev/null \ + -d "csrf=$(csrf "$WORK/p")&secret=$SECRET&code=$(totp)¤t=hunter2hunter2" \ + "$BASE/settings/totp" +code "the admin holds a second factor" 200 $alice "$BASE/admin" + +curl -s $alice "$BASE/admin" >"$WORK/p" +code "a repository is created" 303 $alice -o /dev/null \ + -d "csrf=$(csrf "$WORK/p")&name=notes&desc=Notes&visibility=public&issues=members&wiki=members" \ + "$BASE/admin/repo/create" + +curl -s $alice "$BASE/settings" >"$WORK/p" +curl -s $alice -d "csrf=$(csrf "$WORK/p")&repo=notes" "$BASE/settings/tokens" >"$WORK/minted" +PUSH=$(sed -n 's/.*<pre id="token">\(.*\)<\/pre>.*/\1/p' "$WORK/minted") +[ -n "$PUSH" ] || no "no push token was minted" +ok "a push token is minted" + +echo "== push" +git init -q "$WORK/src" +echo "first" >"$WORK/src/a.txt" +git -C "$WORK/src" add a.txt +git -C "$WORK/src" -c commit.gpgsign=false commit -qm "first" +git -C "$WORK/src" push -q "http://alice:$PUSH@127.0.0.1:$PORT/notes.git" master:master +ok "push over http succeeds" +HEAD=$(git -C "$WORK/src" rev-parse HEAD) + +# The push returned, so it is already in the bucket. Nothing is waited for: a +# sleep would hide exactly the bug this checks for. +[ -f "$BUCKET/repos/notes.git/REFS" ] || no "the refs are not in the bucket" +grep -q "$HEAD" "$BUCKET/repos/notes.git/REFS" || no "the bucket has the wrong head" +ok "the push was durable before the response returned" +find "$BUCKET/repos/notes.git/objects" -type f >"$WORK/objs" 2>/dev/null || true +[ -s "$WORK/objs" ] || no "no objects in the bucket" +ok "objects are in the bucket" +[ -f "$BUCKET/config/users.json" ] || no "the account table is not in the bucket" +ok "the config is in the bucket" + +# git sends anything past http.postBuffer, a megabyte by default, with chunked +# transfer encoding, and that body crosses syncd's proxy. Every other push here +# is a few hundred bytes. Random, so the pack is still over the buffer after +# compression. +head -c 3000000 /dev/urandom >"$WORK/src/big.bin" +git -C "$WORK/src" add big.bin +git -C "$WORK/src" -c commit.gpgsign=false commit -qm "big" +git -C "$WORK/src" push -q "http://alice:$PUSH@127.0.0.1:$PORT/notes.git" master:master +ok "a push past git's post buffer succeeds" +# Reassigned, so everything after the restart is checked against the large push +# rather than the small one. +HEAD=$(git -C "$WORK/src" rev-parse HEAD) +grep -q "$HEAD" "$BUCKET/repos/notes.git/REFS" || no "the chunked push is not in the bucket" +ok "and it was durable before the response returned too" + +echo "== destroy the container, keep only the bucket" +docker rm -f "$NAME" >/dev/null +ok "the container and its disk are gone" + +start "a fresh container starts from the bucket alone" + +# Lazy restore: the directory has to exist so git-forge's startup scan finds it, +# but nothing has asked for the repository yet, so it must still be empty. +docker exec "$NAME" ls /var/lib/git-forge/repos/notes.git >"$WORK/before" 2>&1 || true +[ -s "$WORK/before" ] && no "the repository was fetched before anything asked for it" +ok "the repository is a placeholder until it is asked for" + +curl -s "$BASE/" >"$WORK/home" +want "the index lists it anyway" "notes" "$WORK/home" + +echo "== hydrate on first touch" +git clone -q "http://127.0.0.1:$PORT/notes.git" "$WORK/clone" 2>"$WORK/cloneerr" \ + || { cat "$WORK/cloneerr" >&2; no "the clone failed"; } +[ "$(git -C "$WORK/clone" rev-parse HEAD)" = "$HEAD" ] || no "the clone is at the wrong commit" +ok "the clone comes back at the same commit" +want "and the content survived" "first" "$WORK/clone/a.txt" + +docker exec "$NAME" ls /var/lib/git-forge/repos/notes.git >"$WORK/after" 2>&1 || true +grep -q objects "$WORK/after" || no "the repository was not filled in" +ok "the repository is filled in only after it was asked for" + +echo "== the accounts survived too" +# The push token is a digest in users.json, so a push that authenticates proves +# the account table came back, and proves a restored repository takes writes. +echo "second" >"$WORK/clone/b.txt" +git -C "$WORK/clone" add b.txt +git -C "$WORK/clone" -c commit.gpgsign=false commit -qm "second" +git -C "$WORK/clone" push -q "http://alice:$PUSH@127.0.0.1:$PORT/notes.git" master:master +ok "the push token still authenticates after the restore" +SECOND=$(git -C "$WORK/clone" rev-parse HEAD) +grep -q "$SECOND" "$BUCKET/repos/notes.git/REFS" || no "the second push is not in the bucket" +ok "and the second push is durable too" + +# alice enrolled a second factor before the restart, so a password alone must +# now be answered with a demand for a code rather than a session. +curl -s -o "$WORK/login" -w '%{http_code}' \ + -d "name=alice&password=hunter2hunter2" "$BASE/login" >"$WORK/lc" +grep -qx 200 "$WORK/lc" || no "signing in did not reach the second factor" +want "the second factor survived the restore" 'name="code"' "$WORK/login" + +echo "== compaction packs the audit log and prunes what it replaced" +# SYNCD_LOOSE_LIMIT is 1 here, so a sweep packs rather than waiting for the five +# hundred a deployment would. Long enough for two of them, because pruning waits +# for the pass after the one that finds an object gone. +loose() { find "$BUCKET/config/git/objects" -type f -not -path '*/pack/*' 2>/dev/null | wc -l; } +before=$(loose) +[ "$before" -gt 0 ] || no "there were no loose objects to pack" +sleep 35 +ls "$BUCKET"/config/git/objects/pack/*.pack >/dev/null 2>&1 || no "the audit log was not packed" +ok "the audit log is packed" +after=$(loose) +[ "$after" -lt "$before" ] || no "nothing was pruned: $before loose objects before, $after after" +ok "what packing replaced was pruned from the bucket ($before down to $after)" + +# Pruning deletes from object storage, so the question is whether it deleted +# something still reachable. +docker rm -f "$NAME" >/dev/null +start "a third container starts from the compacted bucket" +curl -s "$BASE/" >"$WORK/home3" +want "the index still lists the repository" "notes" "$WORK/home3" +git clone -q "http://127.0.0.1:$PORT/notes.git" "$WORK/clone3" 2>"$WORK/err3" \ + || { cat "$WORK/err3" >&2; no "the clone failed after compaction"; } +[ "$(git -C "$WORK/clone3" rev-parse HEAD)" = "$SECOND" ] || no "the clone is at the wrong commit after compaction" +ok "and a clone after compaction is still at the right commit" + +echo "== an idle forge writes nothing" +# The sweep runs every fifteen seconds and must find nothing to do. +before=$(grep -c 'PUT ' "$WORK/r2.log" || true) +sleep 20 +after=$(grep -c 'PUT ' "$WORK/r2.log" || true) +[ "$before" = "$after" ] || no "an idle sweep wrote $((after - before)) objects, it should write none" +ok "a sweep over unchanged state writes nothing" + +echo "== a graceful stop is durable and says so" +# SIGTERM, not a kill: syncd drains git-forge, flushes once more, and ships +# those last lines rather than losing them with the process. +grep -q 'container: syncd:' "$WORK/r2.log" || no "no container logs were shipped at all" +ok "container logs are shipped" +docker stop -t 30 "$NAME" >/dev/null +grep -q 'container: syncd: draining' "$WORK/r2.log" || no "the drain was never reported" +ok "SIGTERM drains git-forge" +grep -q 'container: syncd: flushed and stopping' "$WORK/r2.log" || no "the final flush was never reported" +ok "and the final flush is reported before the process exits" + +echo "== a container that refuses to start says why" +# Container stdout is forwarded nowhere, so a fatal must ship its line before it +# goes. No cookie key is the cheapest way to make it refuse. +docker run --rm --name "$NAME-fatal" \ + --add-host=host.docker.internal:host-gateway \ + -e SYNCD_R2="http://host.docker.internal:$R2PORT" \ + -e SYNCD_LOGS="http://host.docker.internal:$R2PORT/" \ + "$IMAGE" >/dev/null 2>&1 || true +sleep 1 +want "the reason it refused to start reaches the log" \ + 'container: syncd: GIT_FORGE_COOKIE_KEY must be set' "$WORK/r2.log" + +printf '\nall good\n' diff --git a/worker.js b/worker.js new file mode 100644 index 0000000..ae7cc0d --- /dev/null +++ b/worker.js @@ -0,0 +1,263 @@ +// The worker routes every request to the one container and answers the +// container's outbound calls from bindings. Everything about git is syncd's. +import { DurableObject, WorkerEntrypoint } from "cloudflare:workers"; + +// syncd fronts git-forge, so this is the only port the worker talks to. +const PORT = 8081; + +// syncd answers this prefix itself; refused below, so only the alarm asks. +const SYNCD = "/.syncd/"; +const INFLIGHT = SYNCD + "inflight"; + +// None of these resolve: all three are intercepted and served from bindings. +const R2_HOST = "r2.forge.internal"; +const MAIL_HOST = "mail.forge.internal"; +// Cloudflare does not forward container stdout to observability, so syncd sends +// its own and git-forge's here. +const LOG_HOST = "logs.forge.internal"; + +const BOOT_TRIES = 40; +const BOOT_INTERVAL_MS = 250; + +// An alarm is the idle timer. +const SLEEP_AFTER_MS = 10 * 60 * 1000; + +// oidcEnv picks out the client secrets by prefix: the provider keys are in +// forge.json, not here. +function oidcEnv(env) { + const out = {}; + for (const [name, value] of Object.entries(env)) { + if (name.startsWith("GIT_FORGE_OIDC") && typeof value === "string") { + out[name] = value; + } + } + // Names, never values. + console.log("container env: oidc variables forwarded:", Object.keys(out).join(", ") || "none"); + return out; +} + +export default { + fetch(request, env) { + // One name, so one container globally. + return env.FORGE.get(env.FORGE.idFromName("forge")).fetch(request); + }, +}; + +export class Forge extends DurableObject { + async fetch(request) { + const url = new URL(request.url); + // Only the alarm may ask, and it does not come through here. + if (url.pathname.startsWith(SYNCD)) { + return new Response("not found\n", { status: 404 }); + } + // base_url from the request, as https: git-forge refuses a form post + // whose Origin disagrees with it, and the scheme puts Secure on every + // cookie for the life of the container. + await this.boot(url.origin.replace(/^http:/, "https:")); + await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS); + // The container listens on http; the rest passes through untouched. + return await this.ctx.container + .getTcpPort(PORT) + .fetch(request.url.replace(/^https:/, "http:"), request); + } + + // SIGTERM rather than destroy: syncd drains git-forge and flushes once more + // on the way out. + async alarm() { + if (!this.ctx.container.running) { + console.log("alarm: already stopped"); + return; + } + const busy = await this.inflight(); + if (busy > 0) { + console.log(`alarm: ${busy} in flight, staying up`); + await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS); + return; + } + console.log("alarm: idle, stopping the container"); + this.ctx.container.signal(15); + // Armed again, so a container that did not stop is signalled once more. + await this.ctx.storage.setAlarm(Date.now() + SLEEP_AFTER_MS); + } + + // Asked of syncd: a fetch here resolves at the headers, minutes before a + // clone's last byte. + async inflight() { + try { + const r = await this.ctx.container.getTcpPort(PORT).fetch(`http://forge${INFLIGHT}`); + return parseInt(await r.text(), 10) || 0; + } catch (e) { + // Unreachable counts as idle, so a wedged container is stopped. + console.log("alarm: asked what was in flight and got:", e); + return 0; + } + } + + // blockConcurrencyWhile makes this once for two requests arriving together. + async boot(origin) { + if (this.ctx.container.running) return; + await this.ctx.blockConcurrencyWhile(async () => { + if (this.ctx.container.running) return; + + // Wired before start, so no early outbound call escapes to the + // internet. Absent on an old runtime. + if (this.ctx.exports?.Outbound === undefined) { + throw new Error( + "ctx.exports.Outbound is undefined, so outbound interception cannot be wired: " + + "raise compatibility_date in wrangler.jsonc", + ); + } + const proxy = this.ctx.exports.Outbound({ props: {} }); + await this.ctx.container.interceptOutboundHttp(R2_HOST, proxy); + await this.ctx.container.interceptOutboundHttp(LOG_HOST, proxy); + // https because git-forge validates mail.mailgun.base_url as https. + await this.ctx.container.interceptOutboundHttps(MAIL_HOST, proxy); + + // No datadir: syncd tells git-forge where it is. + this.ctx.container.start({ + enableInternet: true, // OIDC, and mail if it goes out directly + env: { + GIT_FORGE_BASE_URL: this.env.GIT_FORGE_BASE_URL || origin, + // Every request arrives from syncd otherwise; the edge + // overwrites this header. + GIT_FORGE_CLIENT_IP_HEADER: "CF-Connecting-IP", + // A secret, not a file in a datadir that does not survive a + // restart. + GIT_FORGE_COOKIE_KEY: this.env.GIT_FORGE_COOKIE_KEY ?? "", + SYNCD_R2: `http://${R2_HOST}`, + SYNCD_LOGS: `http://${LOG_HOST}/`, + // git-forge requires a key for the mailgun driver; the + // handler below ignores it. + GIT_FORGE_MAIL_MAILGUN_API_KEY: "intercepted", + ...oidcEnv(this.env), + }, + }); + // Not awaited: it exists so a container that dies is logged. + this.ctx.container.monitor().catch((e) => console.log("container exited:", e)); + + await this.waitReady(); + }); + } + + // waitReady polls git-forge's health endpoint through syncd, so a pass + // proves both are up. + async waitReady() { + const port = this.ctx.container.getTcpPort(PORT); + for (let i = 0; i < BOOT_TRIES; i++) { + try { + const r = await port.fetch("http://forge/healthz"); + if (r.ok) return; + } catch { + // Not listening yet. + } + await new Promise((r) => setTimeout(r, BOOT_INTERVAL_MS)); + } + throw new Error("container did not become ready"); + } +} + +// Outbound answers the intercepted hosts from outside the container, where the +// bindings live. +export class Outbound extends WorkerEntrypoint { + fetch(request) { + const url = new URL(request.url); + if (url.hostname === LOG_HOST) return this.log(request); + if (url.hostname === MAIL_HOST) return this.mail(request); + if (url.hostname === R2_HOST) return this.object(request, url); + // Named above, not defaulted: a new interception has to say what + // serves it. + return new Response("no binding serves this host\n", { status: 404 }); + } + + // One console.log per line, as written. + async log(request) { + const body = await request.text(); + for (const line of body.split("\n")) { + if (line !== "") console.log(line); + } + return new Response(null, { status: 204 }); + } + + // Four verbs, all syncd needs. Keys are the path; a get on / with a prefix + // is a list. + async object(request, url) { + const key = decodeURIComponent(url.pathname.slice(1)); + const bucket = this.env.BUCKET; + + if (request.method === "GET" && key === "") { + const out = await bucket.list({ + prefix: url.searchParams.get("prefix") ?? undefined, + cursor: url.searchParams.get("cursor") || undefined, + limit: 1000, + }); + return Response.json({ + keys: out.objects.map((o) => o.key), + cursor: out.truncated ? out.cursor : "", + }); + } + if (key === "") return new Response("no key", { status: 400 }); + + switch (request.method) { + case "GET": { + const obj = await bucket.get(key); + if (obj === null) return new Response("not found", { status: 404 }); + return new Response(obj.body, { + headers: { "content-length": String(obj.size) }, + }); + } + case "PUT": { + // Streamed: a packfile can be larger than a worker's memory. + const length = request.headers.get("content-length"); + if (length === null) return new Response("content-length required", { status: 411 }); + await bucket.put(key, request.body, { + httpMetadata: { contentType: "application/octet-stream" }, + }); + return new Response(null, { status: 204 }); + } + case "DELETE": + await bucket.delete(key); + return new Response(null, { status: 204 }); + } + return new Response("bad method", { status: 405 }); + } + + // git-forge's mailgun driver is one form-encoded post, so answering that + // shape gives it any sender. + async mail(request) { + const form = await request.formData(); + const to = form.get("to"); + const subject = form.get("subject") ?? ""; + const text = form.get("text") ?? ""; + const from = form.get("from") ?? ""; + + if (this.env.SEND_EMAIL === undefined) { + // Success, or git-forge's spool retries six times. + console.log(`mail: no sender bound, dropping message to ${to}: ${subject}`); + return Response.json({ id: "dropped", message: "no sender bound" }); + } + + // The binding wants a bare address; git-forge sends "Name <addr>". + const addr = (s) => { + const m = /<([^>]+)>/.exec(s); + return m ? m[1] : s.trim(); + }; + + const message = { + from: addr(from), + to: addr(to), + subject, + text, + }; + const unsub = form.get("h:List-Unsubscribe"); + if (unsub) message.headers = { "List-Unsubscribe": unsub }; + + try { + await this.env.SEND_EMAIL.send(message); + } catch (e) { + // A 5xx is what makes git-forge's spool retry. + console.log("mail: send failed:", e); + return new Response(String(e), { status: 502 }); + } + return Response.json({ id: "sent", message: "Queued. Thank you." }); + } +} diff --git a/wrangler.jsonc b/wrangler.jsonc new file mode 100644 index 0000000..7144cdc --- /dev/null +++ b/wrangler.jsonc @@ -0,0 +1,35 @@ +{ + // One worker, one durable object, one container. The name is fixed at + // "forge" in worker.js, so exactly one container serves the bucket. + "name": "git-forge", + "main": "worker.js", + "compatibility_date": "2026-08-01", + "observability": { "enabled": true }, + + "containers": [ + { + "class_name": "Forge", + "image": "./Containerfile", + // 1/4 vCPU, 1 GiB, 4 GB disk. Disk is the ceiling on the datadir, + // and git gc needs room to repack on top of it. + "instance_type": "basic", + "max_instances": 1 + } + ], + + "durable_objects": { + "bindings": [{ "name": "FORGE", "class_name": "Forge" }] + }, + "migrations": [{ "tag": "v1", "new_sqlite_classes": ["Forge"] }], + + "r2_buckets": [{ "binding": "BUCKET", "bucket_name": "git-forge" }], + + // No vars: an environment value pins what the admin page could otherwise + // change. GIT_FORGE_BASE_URL is the exception, for a custom domain fronting + // a hostname requests do not arrive on. + + // Outbound mail: worker.js answers Mailgun's api shape and decides what + // sends. Nothing is sent until forge.json names the intercepted host; + // delete this line to rule mail out. + "send_email": [{ "name": "SEND_EMAIL" }] +} |