Skip to content
quarry Artifact [a4375ba898]
Login

Artifact [a4375ba898]

Check-ins Using Download Hex

Artifact a4375ba89896b91c46324bb9c3241b981b820207c0d7a9ae9d14feff1ada0ac8:

// 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/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

// 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
}

// Counted is told that a repository was cloned or fetched. 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, clone bool)

// SetCounted installs the clone/fetch tally.
func (f *Fronter) SetCounted(c Counted) { f.counted = c }

// 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 }

// 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 {
	dir  string
	port int
	cmd  *exec.Cmd
}

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{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
	}
	if b.cmd != nil && b.cmd.Process != nil {
		b.cmd.Process.Kill()
		b.cmd.Wait()
	}
	delete(f.repos, name)
}

// Close stops every child.
func (f *Fronter) Close() {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, b := range f.repos {
		if b.cmd != nil && b.cmd.Process != nil {
			b.cmd.Process.Kill()
			b.cmd.Wait()
			b.cmd = nil
		}
	}
}

// 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 {
		log.Printf("xfer: %s: %v", name, err)
		status := http.StatusBadGateway
		if err == errUnknownRepo {
			status = http.StatusNotFound
		}
		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
		}
	}
	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")
			w.Write(errorReply("not authorized to write"))
			return
		}
	}

	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
		}
		prox.Header.Set("Content-Type", req.Header.Get("Content-Type"))
		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)
}

// 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) {
	switch {
	case f.counted == nil:
	case pr.newClone:
		f.counted(name, true)
	case pr.hasPull:
		f.counted(name, false)
	}
}

// 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.cmd = nil
	}
	port, err := freePort()
	if err != nil {
		return nil, err
	}
	cmd := exec.Command("zit", "serve", "--port", strconv.Itoa(port))
	cmd.Dir = b.dir
	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("start zit serve: %w", err)
	}
	// 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)
}