Artifact [d00847794f]

Artifact d00847794fb7c3a45bd1c3c82562f993d733bc853e8cd603ff70195ca3fde995:

• File internal/xfer/xfer.go — part of check-in [2c157378a3] at 2026-08-18 01:24:41 on branch trunk — repo: add and remove repositories without restarting

  Adding a repository meant editing the server's command line and restarting, which took every other repository down with it. The registry now lives in OLTP where 0001 §8 always placed it, and repo.add / repo.remove (admin) register and tear down at runtime.

  registrar owns a repository's whole lifetime — index handle, pipeline, sync backend, poll goroutine. The startup loop could not: it relied on defer ix.Close() reaching serve's return and on a poll loop that never stops, and both leak the moment a repository is removed.

  Add persists before it serves, and rolls the record back if opening fails: a registration that serves without being recorded vanishes at the next restart with no trace of why. Remove stops serving and forgets, but leaves the store and index on disk — 'stop serving' and 'destroy' are different intentions and only one is reversible.

  --repo stays as bootstrap and override. (user: dmestas, size: 9213) [annotate] [blame] [check-ins using]

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

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

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