// Package xfer fronts the zit sync protocol (0001 §7): quarry
// terminates HTTP for POST /r/{repo} — the URL zit clients use verbatim
// — and proxies each request to a per-repo `zit serve` child. TLS,
// keep-alive, and N-repos-on-one-port live here; zit's deliberately
// minimal one-shot server stays untouched. Children are started lazily
// and restarted if they die; a proxied push lands artifacts in the
// store, where the ingest watcher picks them up on its next poll.
package xfer
import (
"bytes"
"fmt"
"io"
"log"
"net"
"net/http"
"os"
"os/exec"
"strconv"
"strings"
"sync"
"time"
)
// maxRequest caps a buffered sync request. Clients batch rounds well
// below this; the cap only bounds the gate's memory.
const maxRequest = 256 << 20
// Auth is the push-gating credential source (0001 §7: auth lives at
// quarry's edge). Gated reports whether a repo requires authenticated
// pushes; Secret returns a user's stored shared secret (RFC §6.3).
type Auth interface {
Gated(repo string) bool
Secret(repo, user string) (string, bool)
}
// ReadGate answers whether this sync request may read the repository at
// all. nil leaves every repository readable, which is v1's behaviour.
//
// It exists because the web gate is decoration without it: a private
// repository whose pages refuse anonymous readers, while `zit clone`
// against the same name hands over every artifact, is not private.
type ReadGate func(req *http.Request, repo string) bool
// WriteGuard runs a store-mutating proxy under the store's single-writer
// lease (0001 §3); it returns an error when this node may not write.
type WriteGuard func(repo string, proxy func() error) error
// Protect answers, for one authenticated pusher, which protected-branch
// globs they do NOT own (RFC 0007 §Enforcement). The gate turns an
// identity into a rule; the child turns the rule into a decision,
// because only the store can say which branch a pushed check-in
// resolves to. nil: nothing is protected.
type Protect func(repo, user string) ([]string, error)
// ProtectHeader carries the per-request rule to the child. Honoured by
// zit serve only when started with --fronted, which every child here
// is; the value is a comma-separated glob list, the shape zit's
// ignore-glob lists already use. Never on the fossil wire: loopback,
// between quarry and its own child.
const ProtectHeader = "X-Zit-Protect"
// Fronter proxies sync requests to per-repo zit serve children.
type Fronter struct {
mu sync.Mutex
repos map[string]*backend
auth Auth // nil: every push is anonymous
guard WriteGuard // nil: pushes are not lease-gated (single node)
readGate ReadGate // nil: every repository is readable
writeGate WriteGate // nil: every repository accepts pushes
counted Counted // nil: nothing is tallied
identify Identify // nil: readers are never named
protect Protect // nil: no branch is protected
known Known // nil: a missing backend is always "no such repository"
}
// Reader is what a served read said about who was asking. Every field
// is best-effort and most of them are empty most of the time: a public
// repository is mostly cloned by strangers, and that is not a defect in
// the capture.
//
// The three identity states are kept APART on purpose and must never be
// merged by a caller. Account is proven — a credential this forge
// checked. Claimed is a name the client asserted and nothing verified;
// nothing signs a pull, so a claim is free to forge. Neither set means
// nobody said who they were. Folding a claim in with an account would
// make a forged name indistinguishable from a real one at exactly the
// moment somebody reads the total.
type Reader struct {
// Clone distinguishes a new clone from a later fetch round.
Clone bool
// Client is the software, from the User-Agent: "zit", "fossil", or
// "other". Self-reported, so it answers "what is still on the old
// path" and never "who is this".
Client string
// Version is the client-version pragma's sequence. Also
// self-reported, and empty when the client sent no pragma.
Version string
// Account is a VERIFIED identity: a Basic credential this forge
// authenticated, or a login card whose signature checked out
// against the shared secret. Empty when nobody proved anything.
Account string
// Claimed is a login-card name whose signature did NOT check out,
// or could not be checked. Never treat it as an identity.
Claimed string
}
// Anonymous reports a read that named nobody at all.
func (r Reader) Anonymous() bool { return r.Account == "" && r.Claimed == "" }
// Counted is told that a repository was cloned or fetched, and what the
// request said about the reader. Stated in primitives rather than the
// events package's types for the same reason every other seam here is:
// the fronter knows what the protocol said, and the composition root
// owns what that is worth recording.
//
// It is called only for requests that were actually SERVED — after the
// visibility gate, after auth, and after the backend answered. A refused
// pull is not a fetch, and a repository that answered 404 to a stranger
// must not tell that stranger's arrival to a counter either.
type Counted func(repo string, r Reader)
// SetCounted installs the clone/fetch tally.
func (f *Fronter) SetCounted(c Counted) { f.counted = c }
// Known reports whether the forge serves a repository by this name at
// all, independently of whether its backend has been registered yet.
//
// The two answers differ for tens of seconds after a restart: stores
// are registered one at a time in name order, and a large forge takes a
// while to walk the list. A sync arriving in that window found no
// backend and was told the repository does not exist -- so a clone or
// push during a deploy reported a missing repository rather than a
// forge that was not ready yet, and the client had no reason to retry.
type Known func(name string) bool
// SetKnown installs the registry lookup that tells "not yet" from "no".
func (f *Fronter) SetKnown(k Known) { f.known = k }
// Identify resolves the VERIFIED account behind a request's Basic
// credential, or "" for none. It is the same answer the visibility gate
// already computes for the same request — deliberately, so that "who
// may read this" and "who is recorded as having read it" can never
// drift apart into two different notions of identity.
//
// It must never challenge and never refuse: it observes, it does not
// gate. A wrong password here costs a name off a statistic, never a
// clone.
type Identify func(req *http.Request) string
// SetIdentify installs the reader-identity resolver for tallies.
func (f *Fronter) SetIdentify(i Identify) { f.identify = i }
// WriteGate answers whether the repository accepts pushes at all: a
// mirror that tracks an upstream normally does not, whoever is asking.
// The string names the upstream for the refusal. nil: every store is
// writable.
type WriteGate func(repo string) (ok bool, upstream string)
// SetWriteGate installs the posture gate for pushes.
func (f *Fronter) SetWriteGate(g WriteGate) { f.writeGate = g }
// SetAuth installs the push gate.
func (f *Fronter) SetAuth(a Auth) { f.auth = a }
// SetProtect installs the branch-protection rule source.
func (f *Fronter) SetProtect(p Protect) { f.protect = p }
// SetReadGate installs the visibility gate for pulls.
func (f *Fronter) SetReadGate(g ReadGate) { f.readGate = g }
// SetGuard installs the single-writer lease gate for pushes.
func (f *Fronter) SetGuard(g WriteGuard) { f.guard = g }
type backend struct {
// name is the repository as the forge calls it, carried so the
// child's log lines can say which store they came from.
name string
dir string
port int
cmd *exec.Cmd
// errR is our end of the child's stderr pipe, owned here rather
// than by os/exec. See startChild for why that distinction is the
// difference between a forge that stops and one that wedges.
errR *os.File
}
// stop kills the child and releases its stderr reader.
//
// The order matters. Wait first: with an *os.File on Stderr it reaps
// the process and returns, because os/exec has no copier to await.
// Then close our read end, which unblocks the logging goroutine even
// if some grandchild still holds the write end open forever.
func (b *backend) stop() {
if b.cmd != nil && b.cmd.Process != nil {
b.cmd.Process.Kill()
b.cmd.Wait()
}
if b.errR != nil {
b.errR.Close()
b.errR = nil
}
b.cmd = nil
}
func New() *Fronter {
return &Fronter{repos: map[string]*backend{}}
}
// Register makes one repository frontable. dir is the repository
// directory (the one holding .zit), same as the ingest side uses.
func (f *Fronter) Register(name, dir string) {
f.mu.Lock()
defer f.mu.Unlock()
f.repos[name] = &backend{name: name, dir: dir}
}
// Unregister removes a repository and stops its child. Removing the
// entry without killing the child would leave a `zit serve` process
// holding a store nothing routes to any more — invisible, and still
// holding file handles.
func (f *Fronter) Unregister(name string) {
f.mu.Lock()
defer f.mu.Unlock()
b, ok := f.repos[name]
if !ok {
return
}
b.stop()
delete(f.repos, name)
}
// Close stops every child.
func (f *Fronter) Close() {
f.mu.Lock()
defer f.mu.Unlock()
for _, b := range f.repos {
b.stop()
}
}
// ServeHTTP handles POST /r/{repo} (and its /xfer alias): ensure the
// repo's child is alive, then relay the request body and response
// verbatim — the payload is the sync protocol's business, not ours.
func (f *Fronter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
name := req.PathValue("repo")
// Visibility first, and answering NOT FOUND rather than forbidden:
// a sync client told 403 has learned the repository exists, which is
// the one thing a private repository must not tell it.
if f.readGate != nil && !f.readGate(req, name) {
http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
return
}
b, err := f.ensure(name)
if err != nil {
status := http.StatusBadGateway
// Logged AFTER the status is decided, so the line says what the
// client was actually told. It used to read "unknown repository"
// even while answering "starting", which is the misleading
// signal this whole change exists to remove -- and it cost real
// time reading it during a deploy.
reason := err.Error()
if err == errUnknownRepo {
status = http.StatusNotFound
// A repository the forge serves but has not registered yet
// is starting, not missing. Say so with a retry hint: a
// clone or push arriving during a deploy should wait rather
// than conclude the repository is gone.
//
// This cannot leak the existence of a private repository:
// the visibility gate above already answered 404 for anyone
// who may not see it, so only a reader who is allowed to
// know the name reaches here at all.
if f.known != nil && f.known(name) {
w.Header().Set("Retry-After", "5")
status = http.StatusServiceUnavailable
reason = "not registered yet; the forge is still starting"
}
}
log.Printf("xfer: %s: %s (%d)", name, reason, status)
http.Error(w, http.StatusText(status), status)
return
}
// The gate buffers the request to inspect its cards; the child gets
// the original bytes untouched. Pulls stay anonymous (v1).
raw, err := io.ReadAll(io.LimitReader(req.Body, maxRequest+1))
if err != nil || len(raw) > maxRequest {
http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge)
return
}
// Inspect cards once when either gate is active: auth verifies the
// login on pushes, the lease routes pushes to the write-holder.
var pr pushRequest
if f.auth != nil || f.guard != nil || f.writeGate != nil || f.counted != nil {
pr = scanRequest(decodeContainer(raw))
}
// Posture before credentials: a mirror refuses the push whoever
// signed it, and the message says where the push belongs.
if pr.hasPush && f.writeGate != nil {
if ok, upstream := f.writeGate(name); !ok {
log.Printf("xfer: %s: push refused, store mirrors %s", name, upstream)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("not accepted here: this store mirrors " + upstream + " and does not accept local writes"))
return
}
}
// pusher is the AUTHENTICATED identity — the login card verified
// against the shared secret — and never the U card, which is a
// string any clone can write. Empty means anonymous, and anonymous
// owns nothing.
pusher := ""
if pr.hasPush && f.auth != nil && f.auth.Gated(name) {
// Body login when present; else the x-f-l-c cookie form
// (RFC §6.5), whose nonce covers the whole body.
login := pr
if login.user == "" {
login, _ = cookieLogin(req, decodeContainer(raw))
}
secret, known := f.auth.Secret(name, login.user)
if !known || !verifyLogin(login, secret) {
log.Printf("xfer: %s: push rejected (user %q)", name, login.user)
w.Header().Set("Content-Type", "application/x-fossil")
// Name what is actually missing. "not authorized to write"
// sent two people straight to the capability letters, which
// are only half of the answer: a push needs the letters AND
// a credential, and the credential is the half with no page
// pointing at it. A refusal that does not say which half is
// wrong costs an afternoon.
w.Write(errorReply(refusalFor(login.user)))
return
}
pusher = login.user
}
// The per-request protection rule, computed for EVERY request and
// sent even when empty. zit#151 treats an absent header as no rule
// — right for a bare server, but behind a front it would let a
// forgotten code path silently disable protection. So "empty
// because computed" is the only shape that exists here; "absent
// because forgotten" cannot happen, and the harness proves it.
//
// A rule source that cannot answer fails the push closed: the one
// thing worse than refusing an owner is admitting a stranger
// because a database was busy.
rule := ""
if f.protect != nil {
refused, err := f.protect(name, pusher)
if err != nil {
if pr.hasPush {
log.Printf("xfer: %s: push refused, protection unavailable: %v", name, err)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("branch protection is unavailable; push refused"))
return
}
} else {
rule = strings.Join(refused, ",")
}
}
proxy := func() error {
target := "http://127.0.0.1:" + strconv.Itoa(b.port) + "/xfer"
prox, err := http.NewRequestWithContext(req.Context(), "POST", target, bytes.NewReader(raw))
if err != nil {
return err
}
// A fresh request, not a copy: the client's headers never reach
// the child. That is what makes the protect header the gate's
// alone — a forwarded client header with that name would be a
// bypass on loopback behind --fronted.
prox.Header.Set("Content-Type", req.Header.Get("Content-Type"))
prox.Header.Set(ProtectHeader, rule)
resp, err := backendClient.Do(prox)
if err != nil {
// The child may have died mid-flight; drop it so the next
// request restarts it.
f.drop(name)
return err
}
defer resp.Body.Close()
if ct := resp.Header.Get("Content-Type"); ct != "" {
w.Header().Set("Content-Type", ct)
}
w.WriteHeader(resp.StatusCode)
_, err = io.Copy(w, resp.Body)
return err
}
// A push mutates the store, so it routes through the write lease; a
// node that does not hold it answers with a read-only error card
// rather than writing behind the holder's back.
if pr.hasPush && f.guard != nil {
if err := f.guard(name, proxy); err != nil {
log.Printf("xfer: %s: push not served here: %v", name, err)
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(errorReply("store is read-only on this node"))
}
return
}
if err := proxy(); err != nil {
log.Printf("xfer: %s: proxy: %v", name, err)
http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
return
}
f.tally(name, pr, req, raw)
}
// refusalFor says which half of a push's authorization is missing, in
// the one sentence a fossil error card gets.
func refusalFor(user string) string {
if user == "" {
return "anonymous push is not accepted here: push with --user and a push credential"
}
return "not authorized to write as " + user +
": a push needs check-in capability AND a push credential, and one of them is missing" +
" (mint a credential with POST /api/v1/me/sync-token)"
}
// tally records a served read. Pushes are not counted here: they are
// already the timeline's business, where a clone leaves no other trace
// at all — which is the whole reason to count these two.
func (f *Fronter) tally(name string, pr pushRequest, req *http.Request, raw []byte) {
switch {
case f.counted == nil:
case pr.newClone:
f.counted(name, f.reader(name, pr, req, raw, true))
case pr.hasPull:
f.counted(name, f.reader(name, pr, req, raw, false))
}
}
// reader assembles what this request said about who was asking.
//
// Nothing here can refuse anything: it runs AFTER the response has been
// written. That ordering is the safety property — a bug in identity
// resolution cannot turn into a refused clone, because by the time this
// runs the clone has already been served.
func (f *Fronter) reader(name string, pr pushRequest, req *http.Request, raw []byte, clone bool) Reader {
r := Reader{Clone: clone, Client: clientKind(req.UserAgent()), Version: pr.version}
// Basic first: it is the form zit uses, it is checked against the
// account database, and it arrives on the clone round itself.
if f.identify != nil {
r.Account = f.identify(req)
}
if r.Account != "" {
return r
}
// Else a login card, which is fossil's form. It arrives on later
// rounds rather than on the clone, so this mostly names fetches.
login := pr
if login.user == "" {
login, _ = cookieLogin(req, decodeContainer(raw))
}
if login.user == "" {
return r
}
// The signature check is the whole difference between an account
// and a claim. It is the same verification pushes already run; the
// only thing new is running it on a request shape that skipped it.
if f.auth != nil {
if secret, known := f.auth.Secret(name, login.user); known && verifyLogin(login, secret) {
r.Account = login.user
return r
}
}
r.Claimed = login.user
return r
}
// clientKind names the software from its User-Agent. zit presents a
// fossil-shaped client-version pragma for compatibility, so the pragma
// card cannot tell the two apart and the User-Agent is the only signal
// that does.
//
// Self-reported, and trivially spoofable. It is a population statistic,
// never evidence.
func clientKind(ua string) string {
switch {
case strings.HasPrefix(ua, "zig/"), strings.Contains(ua, "zit/"):
return "zit"
case strings.HasPrefix(ua, "Fossil/"), strings.HasPrefix(ua, "fossil/"):
return "fossil"
default:
return "other"
}
}
// backendClient has no overall timeout: sync rounds on big stores are
// legitimately long. Cancellation rides the inbound request context.
var backendClient = &http.Client{}
var errUnknownRepo = fmt.Errorf("unknown repository")
func (f *Fronter) drop(name string) {
f.mu.Lock()
defer f.mu.Unlock()
if b, ok := f.repos[name]; ok && b.cmd != nil {
if b.cmd.Process != nil {
b.cmd.Process.Kill()
b.cmd.Wait()
}
b.cmd = nil
}
}
// ensure returns a live backend for name, starting its child if needed.
func (f *Fronter) ensure(name string) (*backend, error) {
f.mu.Lock()
defer f.mu.Unlock()
b, ok := f.repos[name]
if !ok {
return nil, errUnknownRepo
}
if b.cmd != nil {
// A quick dial beats trusting ProcessState: the reaper may not
// have run yet for a just-dead child, and a hung child that no
// longer accepts deserves a restart too.
if b.cmd.ProcessState == nil {
if c, err := net.DialTimeout("tcp4", "127.0.0.1:"+strconv.Itoa(b.port), 300*time.Millisecond); err == nil {
c.Close()
return b, nil
}
b.cmd.Process.Kill()
}
b.stop()
}
port, err := freePort()
if err != nil {
return nil, err
}
// --fronted: this child's only client is the gate on loopback, and
// it may honour the gate's per-request protect header. A zit serve
// reachable directly is never started this way.
cmd := exec.Command("zit", "serve", "--port", strconv.Itoa(port), "--fronted")
cmd.Dir = b.dir
// The child's diagnostics reach the forge log (childlog.go) — but
// through a pipe THIS code owns, never by handing os/exec a plain
// io.Writer.
//
// os/exec connects an *os.File to the child directly and copies
// anything else on a goroutine that Wait() blocks on until every
// write end of the pipe is closed. Killing `zit serve` does not
// close a write end some grandchild inherited, so Wait() never
// returned: Close() and Unregister() hold f.mu while they wait, and
// ensure() and ServeHTTP take that same mutex, so one removed
// repository wedged sync for every repository on the forge. It hung
// two CI runs at the ten-minute timeout before it was believed.
//
// With an *os.File there is no copier to await, the reading is ours
// to end, and stop() ends it by closing errR.
pr, pw, err := os.Pipe()
if err != nil {
return nil, fmt.Errorf("stderr pipe for zit serve: %w", err)
}
cmd.Stderr = pw
if err := cmd.Start(); err != nil {
pr.Close()
pw.Close()
return nil, fmt.Errorf("start zit serve: %w", err)
}
// The parent's write end goes now: only the child (and whatever it
// spawns) should hold one, or the reader would never see EOF even
// after everybody exits.
pw.Close()
b.errR = pr
go func(log io.Writer, r *os.File) {
io.Copy(log, r)
r.Close()
}(newChildLog(b.name), pr)
// Reap in the background so ProcessState is set when it exits.
go cmd.Wait()
if err := waitReady(port, 5*time.Second); err != nil {
cmd.Process.Kill()
return nil, err
}
b.cmd, b.port = cmd, port
return b, nil
}
// ProjectCode reads a store's identity through `zit stat -R` (zit#85).
// It works on a BARE store — no checkout, no server, no port — which is
// what lets quarry serve a repository that is nothing but a .zit file.
//
// This replaced a clone handshake against a transient `zit serve` child:
// find a free port, spawn, poll for readiness, POST a container, scrape
// the push card. That cost a subprocess and a listening socket per
// repository at startup, and because serve is checkout-only it was the
// single thing keeping quarry from bare-store serving.
func ProjectCode(dir string) (string, error) {
cmd := exec.Command("zit", "stat", "-R", dir)
var stderr bytes.Buffer
cmd.Stderr = &stderr
out, err := cmd.Output()
if err != nil {
return "", fmt.Errorf("zit stat -R %s: %w: %s", dir, err, strings.TrimSpace(stderr.String()))
}
for _, line := range strings.Split(string(out), "\n") {
if rest, ok := strings.CutPrefix(line, "project-code "); ok {
if code := strings.TrimSpace(rest); code != "" {
return code, nil
}
}
}
return "", ErrNoProjectCode
}
// freePort asks the kernel for an unused port. The listen-then-close
// window is racy in principle; waitReady failing surfaces a collision
// and the next request retries with a fresh port.
func freePort() (int, error) {
l, err := net.Listen("tcp4", "127.0.0.1:0")
if err != nil {
return 0, err
}
port := l.Addr().(*net.TCPAddr).Port
l.Close()
return port, nil
}
func waitReady(port int, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
c, err := net.DialTimeout("tcp4", "127.0.0.1:"+strconv.Itoa(port), 200*time.Millisecond)
if err == nil {
c.Close()
return nil
}
time.Sleep(50 * time.Millisecond)
}
return fmt.Errorf("zit serve on :%d not ready within %s", port, timeout)
}
// ReapOrphans kills stray `zit serve --fronted` children left by a
// previous quarry.
//
// The fronter starts one per repository on demand and kills it only
// when that repository's backend is restarted. Nothing kills them when
// QUARRY exits — and the deploy path uses SIGKILL, so no graceful
// shutdown runs — leaving one orphan per synced repository per restart,
// reparented to init and running forever. Twenty deploys in a day left
// twenty-four of them, the oldest eight hours old, each holding a port
// and its own memory.
//
// A freshly started quarry has no backends by definition, so any child
// carrying --fronted is a leftover. The flag is the marker: it is set
// only here, for children whose sole client is the loopback gate.
func ReapOrphans() int {
out, err := exec.Command("pgrep", "-f", "zit serve --port .* --fronted").Output()
if err != nil {
return 0 // no pgrep, or nothing matched: nothing to do either way
}
killed := 0
for _, line := range strings.Fields(string(out)) {
pid, perr := strconv.Atoi(line)
if perr != nil || pid <= 1 {
continue
}
p, ferr := os.FindProcess(pid)
if ferr != nil {
continue
}
if p.Kill() == nil {
killed++
}
}
return killed
}