Skip to content
quarry Artifact [ea77fc73af]
Login

Artifact [ea77fc73af]

Check-ins Using Download Hex

Artifact ea77fc73af70878d70a96b8b5608d27a2dc7022f153ecd81838b1924970567db:

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

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

// SetAuth installs the push gate.
func (f *Fronter) SetAuth(a Auth) { f.auth = a }

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

// 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")
	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 {
		pr = scanRequest(decodeContainer(raw))
	}
	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)
	}
}

// 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 learns a store's project code the boundary-clean way: a
// clone handshake against a transient zit serve child. The reply to a
// `clone 3 1` request carries a push card naming servercode and
// projectcode (sync RFC §3.2) — no .zit bytes are read.
func ProjectCode(dir string) (string, error) {
	port, err := freePort()
	if err != nil {
		return "", err
	}
	cmd := exec.Command("zit", "serve", "--port", strconv.Itoa(port))
	cmd.Dir = dir
	if err := cmd.Start(); err != nil {
		return "", fmt.Errorf("start zit serve: %w", err)
	}
	defer func() {
		cmd.Process.Kill()
		cmd.Wait()
	}()
	if err := waitReady(port, 5*time.Second); err != nil {
		return "", err
	}
	body := encodeContainer([]byte("clone 3 1\n"))
	resp, err := http.Post("http://127.0.0.1:"+strconv.Itoa(port)+"/xfer",
		"application/x-fossil", bytes.NewReader(body))
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	raw, err := io.ReadAll(io.LimitReader(resp.Body, maxRequest))
	if err != nil {
		return "", err
	}
	for _, line := range bytes.Split(decodeContainer(raw), []byte("\n")) {
		if bytes.HasPrefix(line, []byte("push ")) {
			f := bytes.Fields(line)
			if len(f) == 3 {
				return string(f[2]), 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)
}