Artifact [c8f16f0ceb]

Artifact c8f16f0ceb80653cd5ca0e5d99ab2661131bdb9bc87790a6d1df5738a3c7395f:

• File internal/xfer/xfer.go — part of check-in [31b695c5cb] at 2026-08-15 21:30:33 on branch trunk — xfer: sync fronting (0001 §7 v1 slice)

  POST /r/{repo} (+/xfer alias) proxies to lazily-started per-repo zit serve children; dead children restart on the next request. E2E: clone through quarry, push back, watcher folds the push. Push auth waits for the forge-OLTP slice; leases stay v-next. (user: dmestas, size: 5001) [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 (
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os/exec"
	"strconv"
	"sync"
	"time"
)

// Fronter proxies sync requests to per-repo zit serve children.
type Fronter struct {
	mu    sync.Mutex
	repos map[string]*backend
}

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
	}
	target := "http://127.0.0.1:" + strconv.Itoa(b.port) + "/xfer"
	prox, err := http.NewRequestWithContext(req.Context(), "POST", target, req.Body)
	if err != nil {
		http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
		return
	}
	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.
		log.Printf("xfer: %s: proxy: %v", name, err)
		f.drop(name)
		http.Error(w, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
		return
	}
	defer resp.Body.Close()
	if ct := resp.Header.Get("Content-Type"); ct != "" {
		w.Header().Set("Content-Type", ct)
	}
	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)
}

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

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