package api
import (
"errors"
"net"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"time"
"unicode/utf8"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/uv"
"craftdesign.group/quarry/internal/xfer"
)
// RFC 0020 rung 6 against a stock `fossil server`: a mirror imports the
// upstream's unversioned files into its bucket after each pull, as a
// fossil client of the upstream.
// oltpMirrors is the composition root's mirror adapter, over a real
// OLTP database, so the round is tested against the columns it writes.
type oltpMirrors struct{ db *oltp.DB }
func (m oltpMirrors) Mirrors() ([]Mirror, error) {
rows, err := m.db.Mirrors()
out := make([]Mirror, 0, len(rows))
for _, r := range rows {
out = append(out, toMirror(r))
}
return out, err
}
func (m oltpMirrors) MirrorFor(repo string) (Mirror, bool, error) {
r, ok, err := m.db.MirrorFor(repo)
return toMirror(r), ok, err
}
func (m oltpMirrors) SetMirror(repo, url string, every int) error {
return m.db.SetMirror(repo, url, time.Duration(every)*time.Second)
}
func (m oltpMirrors) DetachMirror(repo string) error { return m.db.DetachMirror(repo) }
func (m oltpMirrors) SetMirrorWrites(repo string, accept bool) error {
return m.db.SetMirrorWrites(repo, accept)
}
func (m oltpMirrors) RecordMirrorRun(repo string, took time.Duration, changed bool, err error) (int, error) {
return m.db.RecordMirrorRun(repo, time.Now(), took, changed, err)
}
func (m oltpMirrors) MirrorPulls(repo string, n int) ([]MirrorPull, error) {
rows, err := m.db.MirrorPulls(repo, n)
var out []MirrorPull
for _, r := range rows {
out = append(out, MirrorPull{At: r.At, Ms: r.Ms, Changed: r.Changed, Err: r.Err, UVRan: r.UVRan, UVChanged: r.UVChanged, UVErr: r.UVErr})
}
return out, err
}
func (m oltpMirrors) MirrorDiverged(repo, checkin string, at time.Time) error {
return m.db.MirrorDiverged(repo, checkin, at)
}
func (m oltpMirrors) SetMirrorUV(repo string, on bool) error { return m.db.SetMirrorUV(repo, on) }
func (m oltpMirrors) SetMirrorUVCredential(repo, user, pw string) error {
return m.db.SetMirrorUVCredential(repo, user, pw)
}
func (m oltpMirrors) MirrorUVCredential(repo string) (string, string, bool, error) {
return m.db.MirrorUVCredential(repo)
}
func (m oltpMirrors) RecordMirrorUV(repo, pullAt string, changed int, msg string) error {
return m.db.RecordMirrorUV(repo, pullAt, time.Now(), changed, msg)
}
func toMirror(r oltp.Mirror) Mirror {
return Mirror{Repo: r.Repo, URL: r.URL, Every: int(r.Every.Seconds()), Enabled: r.Enabled,
LastAt: r.LastAt, LastOK: r.LastOK, LastErr: r.LastErr, Failures: r.Failures,
AcceptWrites: r.AcceptWrites, DetachedFrom: r.DetachedFrom,
UV: r.UV, UVUser: r.UVUser, UVCredential: r.UVCredential, UVAt: r.UVAt, UVChanged: r.UVChanged, UVErr: r.UVErr}
}
type uvMirrorRig struct {
t *testing.T
dir string
home string
up string // the upstream's repository file
url string // the fossil server's address
store string // the mirror's zit store
db *oltp.DB
objects *uvBucket
buckets *memBuckets
bucket string
r *ops.Registry
d Deps
// pullAs makes the DAG pull sign in as this upstream login, as a
// store that remembers a login pulls; the round's identity is its own.
pullAs, pullPW string
private bool
}
// newUVMirrorRig serves a fresh fossil repository with `fossil server`,
// clones it with zit as the mirror "test", and binds a bucket with
// limit bytes per file (none when limit is 0). The upstream has a login
// "puller" holding only o.
func newUVMirrorRig(t *testing.T, limit int64) *uvMirrorRig {
t.Helper()
for _, bin := range []string{"zit", "fossil"} {
if _, err := exec.LookPath(bin); err != nil {
if os.Getenv("CI") == "true" {
t.Fatalf("FAIL: %s not on PATH under CI: the mirror import test needs both", bin)
}
t.Skip(bin + " not on PATH: the mirror import test needs both")
}
}
g := &uvMirrorRig{t: t, dir: t.TempDir(), home: t.TempDir()}
g.up = filepath.Join(g.dir, "up.fossil")
g.fossil("init", g.up)
// Not fossil's own "reader": check_login refuses the logins fossil
// init makes (anonymous, nobody, developer, reader) whatever their
// password.
g.fossil("user", "new", "puller", "", "pullerpw", "-R", g.up)
g.fossil("user", "capabilities", "puller", "o", "-R", g.up)
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
port := strconv.Itoa(l.Addr().(*net.TCPAddr).Port)
l.Close()
srv := exec.Command("fossil", "server", g.up, "--port", port, "--localhost")
srv.Env = g.env()
if err := srv.Start(); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { srv.Process.Kill(); srv.Wait() })
g.url = "http://127.0.0.1:" + port
for i := 0; ; i++ {
resp, err := http.Get(g.url)
if err == nil {
resp.Body.Close()
break
}
if i > 100 {
t.Fatalf("FAIL: fossil server did not answer: %v", err)
}
time.Sleep(50 * time.Millisecond)
}
g.store = filepath.Join(g.dir, "store")
g.run(g.dir, "zit", "clone", g.url, g.store)
code, err := xfer.ProjectCode(g.store)
if err != nil {
t.Fatal(err)
}
g.db, err = oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { g.db.Close() })
srvS3, objects := newUVBucket(t)
g.objects, g.buckets, g.bucket = objects, newMemBuckets(), srvS3.URL
if limit > 0 {
g.buckets.rows["test"] = oltp.BucketBinding{Endpoint: srvS3.URL, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK", LimitBytes: limit}
}
mirrors := oltpMirrors{g.db}
if err := g.db.SetMirror("test", g.url, time.Hour); err != nil {
t.Fatal(err)
}
repo := &Repo{Mirror: func(u string) (bool, error) {
args := []string{"pull", u}
if g.pullAs != "" {
args = append(args, "--user", g.pullAs)
}
cmd := exec.Command("zit", args...)
cmd.Dir, cmd.Env = g.store, append(os.Environ(), "ZIT_PASSWORD="+g.pullPW)
out, err := cmd.CombinedOutput()
if err != nil {
return false, errors.New(strings.TrimSpace(string(out)))
}
return strings.Contains(string(out), "received"), nil
}}
g.d = Deps{Version: "v", Repos: NewRepos(map[string]*Repo{"test": repo}), Buckets: g.buckets, UV: g.db,
Events: &uvEvents{}, Mirrors: mirrors, MirrorUV: mirrors,
ProjectCode: func(string) (string, error) { return code, nil },
Private: func() map[string]bool { return map[string]bool{"test": g.private} },
CapsFor: func(_, _ string) CapSet { return oltp.CapSet("ay") }}
g.r = ops.NewRegistry()
if err := RegisterAll(g.r, g.d); err != nil {
t.Fatal(err)
}
return g
}
func (g *uvMirrorRig) env() []string {
return append(os.Environ(), "HOME="+g.home, "FOSSIL_HOME="+g.home, "FOSSIL_USER=admin")
}
func (g *uvMirrorRig) run(dir, bin string, args ...string) string {
g.t.Helper()
cmd := exec.Command(bin, args...)
cmd.Dir, cmd.Env = dir, g.env()
out, err := cmd.CombinedOutput()
if err != nil {
g.t.Fatalf("FAIL: %s %v: %v\n%s", bin, args, err, out)
}
return string(out)
}
func (g *uvMirrorRig) fossil(args ...string) string { return g.run(g.dir, "fossil", args...) }
// uvAdd publishes content upstream as name with the given mtime.
func (g *uvMirrorRig) uvAdd(name, content, mtime string) {
g.t.Helper()
src := filepath.Join(g.dir, "in.tmp")
if err := os.WriteFile(src, []byte(content), 0o644); err != nil {
g.t.Fatal(err)
}
g.fossil("uv", "add", src, "--as", name, "--mtime", mtime, "-R", g.up)
}
// uvSwitch calls mirror.uv as a repository admin.
func (g *uvMirrorRig) uvSwitch(on bool, user, pw string) (mirrorUVOut, error) {
out, err := g.r.Invoke(as("admin", "a"), "mirror.uv", mirrorUVIn{Repo: "test", On: on, User: user, Password: pw})
if err != nil {
return mirrorUVOut{}, err
}
return out.(mirrorUVOut), nil
}
// pull runs the mirror as the scheduler does and answers the pull it
// recorded.
func (g *uvMirrorRig) pull() MirrorPull {
g.t.Helper()
m, _, err := g.d.Mirrors.MirrorFor("test")
if err != nil {
g.t.Fatal(err)
}
if _, err := g.d.PullMirror(m); err != nil {
g.t.Fatalf("FAIL: the DAG pull failed: %v", err)
}
waitImportUV()
pulls, err := g.d.Mirrors.MirrorPulls("test", 1)
if err != nil || len(pulls) != 1 {
g.t.Fatalf("FAIL: no pull recorded: %v", err)
}
return pulls[0]
}
func (g *uvMirrorRig) row(name string) (uv.File, bool) {
g.t.Helper()
f, ok, err := g.db.UVFile("test", name)
if err != nil {
g.t.Fatal(err)
}
return f, ok
}
func TestMirrorImportsUnversionedFromStockFossil(t *testing.T) {
g := newUVMirrorRig(t, 1<<20)
g.uvAdd("hello.txt", "hello\n", "2026-01-01 00:00:00")
g.uvAdd("releases/1.0/notes.md", "# 1.0\n", "2026-01-02 00:00:00")
p := g.pull()
_, has := g.row("hello.txt")
check(t, !p.UVRan && !has, "the switch off: nothing imported, no round recorded",
"switch off: pull %+v, row %v", p, has)
if _, err := g.uvSwitch(true, "", ""); err != nil {
t.Fatalf("FAIL: turning the switch on: %v", err)
}
p = g.pull()
row, _ := g.row("hello.txt")
o, inBucket := g.objects.has(uv.Key("hello.txt"))
notes, _ := g.row("releases/1.0/notes.md")
check(t, p.UVRan && p.UVChanged == 2 && p.UVErr == "" && inBucket && string(o.body) == "hello\n" &&
row.Who == "mirror:127.0.0.1" && row.Mtime == 1767225600 && notes.Mtime == 1767312000,
"on: the upstream's files arrive in the bucket as mirror:<host>, as anonymous nobody (o, no y)",
"on: pull %+v, row %+v, object %v %q, notes %+v", p, row, inBucket, o.body, notes)
p = g.pull()
check(t, p.UVRan && p.UVChanged == 0 && p.UVErr == "",
"in step: a matching uv-hash draws silence, and the round still tells that from not shared",
"in step: pull %+v", p)
g.fossil("uv", "rm", "hello.txt", "-R", g.up)
p = g.pull()
row, _ = g.row("hello.txt")
_, stillThere := g.objects.has(uv.Key("hello.txt"))
check(t, p.UVRan && p.UVChanged == 1 && row.Deleted && !stillThere,
"an upstream `fossil uv rm` arrives as a tombstone on the next pull",
"tombstone: pull %+v, row %+v, content kept %v", p, row, stillThere)
pulls, _ := g.d.Mirrors.MirrorPulls("test", 0)
ran := 0
for _, q := range pulls {
if q.UVRan {
ran++
}
}
m, _, _ := g.d.Mirrors.MirrorFor("test")
check(t, len(pulls) == 4 && ran == 3 && m.UVAt != "" && m.UVChanged == 1,
"mirror_pull records each round beside its pull, and the mirror row the last",
"history: %+v, mirror %+v", pulls, m)
// A tracking mirror's own users still cannot write: the import is
// the only way in.
_, err := g.r.Invoke(as("admin", "ay"), "uv.put", uvPutIn{Repo: "test", Name: "local.txt", Mtime: 1, Content: "aGk="})
check(t, errors.Is(err, ops.ErrForbidden) && strings.Contains(fmtErr(err), "mirrors"),
"a local unversioned write on the tracking mirror stays refused",
"local write: %v", err)
// fossil sends at most max-download a reply and answers the rest
// with uvigot; the round asks again until all have come.
g.fossil("sql", "-R", g.up, "REPLACE INTO config(name,value,mtime) VALUES('max-download','1000',now())")
for i := 0; i < 3; i++ {
g.uvAdd("big"+strconv.Itoa(i)+".bin", strings.Repeat(strconv.Itoa(i), 3000), "2026-02-01 00:00:00")
}
p = g.pull()
all := true
for i := 0; i < 3; i++ {
f, ok := g.row("big" + strconv.Itoa(i) + ".bin")
all = all && ok && f.Size == 3000
}
check(t, p.UVChanged == 3 && p.UVErr == "" && all,
"files over the upstream's reply budget arrive over several exchanges",
"budget: pull %+v, all %v", p, all)
}
func fmtErr(err error) string {
if err == nil {
return ""
}
return err.Error()
}
func TestMirrorUVSwitchNeedsABucket(t *testing.T) {
g := newUVMirrorRig(t, 0)
_, err := g.uvSwitch(true, "", "")
check(t, errors.Is(err, ops.ErrInvalid) && strings.Contains(fmtErr(err), "no bucket"),
"no bucket bound: the switch is refused with the reason",
"no bucket: %v", err)
out, err := g.r.Invoke(as("admin", "a"), "mirror.list", mirrorListIn{})
if err != nil {
t.Fatal(err)
}
ms := out.(mirrorListOut).Mirrors
check(t, len(ms) == 1 && !ms[0].UV && strings.Contains(ms[0].UVBlocked, "no bucket"),
"the mirror's row says why the switch is disabled",
"list: %+v", ms)
g.buckets.rows["test"] = oltp.BucketBinding{Endpoint: g.bucket, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK",
LimitBytes: 1 << 20, PublicBase: "https://files.example"}
g.private = true
_, err = g.uvSwitch(true, "", "")
check(t, errors.Is(err, ops.ErrInvalid) && strings.Contains(fmtErr(err), "public"),
"a private repository with a public bucket base: refused, as UVWrite refuses it",
"private public: %v", err)
g.private = false
if _, err := g.uvSwitch(true, "", ""); err != nil {
t.Fatalf("FAIL: a public repository with a bucket: %v", err)
}
t.Log("ok: with a bucket the switch turns on")
}
// The round signs with the mirror's own credential or as nobody,
// whatever login the DAG pull uses. fossil answers uv-hash only to a
// login holding o, the letter its pull needs too (xfer.c), so "the pull
// succeeds and the round is not shared" is a pull made as a login the
// store remembers while the round goes as nobody.
func TestMirrorUVNotSharedThenCredential(t *testing.T) {
g := newUVMirrorRig(t, 1<<20)
g.uvAdd("hello.txt", "hello\n", "2026-01-01 00:00:00")
g.fossil("user", "capabilities", "nobody", "gjrz", "-R", g.up)
g.pullAs, g.pullPW = "puller", "pullerpw"
if _, err := g.uvSwitch(true, "", ""); err != nil {
t.Fatal(err)
}
p := g.pull()
_, has := g.row("hello.txt")
check(t, p.Err == "" && p.UVRan && p.UVErr == UVNotShared && !has,
"nobody without o: the DAG pull succeeds and the round records that the upstream does not share",
"not shared: pull %+v", p)
if _, err := g.uvSwitch(true, "puller", "wrong"); err != nil {
t.Fatal(err)
}
p = g.pull()
check(t, p.Err == "" && p.UVErr != "" && !strings.Contains(p.UVErr, "pullerpw") && !strings.Contains(p.UVErr, "wrong"),
"a wrong password is the round's failure, in words that carry no secret",
"wrong password: pull %+v", p)
t.Logf(" (a wrong password reads: %s)", p.UVErr)
out, err := g.uvSwitch(true, "puller", "pullerpw")
if err != nil {
t.Fatal(err)
}
check(t, out.Mirror.UVUser == "puller" && out.Mirror.UVCredential,
"the op answers the login and that a password is held, never the password",
"op: %+v", out.Mirror)
p = g.pull()
row, has := g.row("hello.txt")
check(t, p.UVChanged == 1 && p.UVErr == "" && has && row.Who == "mirror:127.0.0.1",
"a credential holding o (and no y): the login card verifies at stock fossil and the files arrive",
"credential: pull %+v row %+v", p, row)
// Keeping the login and leaving the password empty keeps it.
if _, err := g.uvSwitch(true, "puller", ""); err != nil {
t.Fatalf("FAIL: re-saving the form without the password: %v", err)
}
if _, err := g.uvSwitch(true, "someone", ""); !errors.Is(err, ops.ErrInvalid) {
t.Fatalf("FAIL: a new login without a password: %v", err)
}
t.Log("ok: an empty password keeps the held one for the same login, and a new login needs one")
}
func TestMirrorUVOverLimitIsSkipped(t *testing.T) {
g := newUVMirrorRig(t, 64)
g.uvAdd("small.txt", "small\n", "2026-01-01 00:00:00")
g.uvAdd("big.bin", strings.Repeat("x", 4096), "2026-01-01 00:00:00")
if _, err := g.uvSwitch(true, "", ""); err != nil {
t.Fatal(err)
}
p := g.pull()
_, small := g.row("small.txt")
_, big := g.row("big.bin")
check(t, p.Err == "" && p.UVChanged == 1 && small && !big &&
strings.Contains(p.UVErr, "big.bin: not carried (larger than this bucket's limit)"),
"a file over the bucket's limit is skipped and recorded, the rest arrive",
"over limit: pull %+v small %v big %v", p, small, big)
}
// The round runs after the pull's indexing and off the caller's
// goroutine, one at a time per mirror and two at a time in all, so a
// stalled upstream holds up neither the index nor the next mirror.
func TestMirrorUVRoundRunsOffTheSchedulerLoop(t *testing.T) {
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
srvS3, _ := newUVBucket(t)
buckets := newMemBuckets()
release := make(chan struct{})
arrived := make(chan string, 16)
var indexedFirst []bool
var mu sync.Mutex
indexed := map[string]bool{}
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
repo := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/"), "/xfer")
mu.Lock()
indexedFirst = append(indexedFirst, indexed[repo])
mu.Unlock()
arrived <- repo
<-release
w.Header().Set("Content-Type", "application/x-fossil")
w.Write(xferContainer("pragma uv-pull-only\n"))
}))
defer up.Close()
repos := map[string]*Repo{}
mirrors := oltpMirrors{db}
for _, name := range []string{"a", "b", "c"} {
name := name
buckets.rows[name] = oltp.BucketBinding{Endpoint: srvS3.URL, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK", LimitBytes: 1 << 20}
repos[name] = &Repo{Mirror: func(string) (bool, error) { return true, nil },
Sync: func() (int, error) { mu.Lock(); indexed[name] = true; mu.Unlock(); return 1, nil }}
if err := db.SetMirror(name, up.URL+"/"+name, time.Hour); err != nil {
t.Fatal(err)
}
if err := db.SetMirrorUV(name, true); err != nil {
t.Fatal(err)
}
}
d := Deps{Repos: NewRepos(repos), Buckets: buckets, UV: db, Events: &uvEvents{}, Mirrors: mirrors, MirrorUV: mirrors}
pull := func(name string) time.Duration {
m, _, _ := mirrors.MirrorFor(name)
start := time.Now()
if _, err := d.PullMirror(m); err != nil {
t.Fatal(err)
}
return time.Since(start)
}
took := pull("a") + pull("b") + pull("c") + pull("a")
check(t, took < 2*time.Second, "PullMirror returns while the upstream holds every round open",
"four pulls took %s", took)
time.Sleep(300 * time.Millisecond)
check(t, len(arrived) == 2, "two rounds run at once; the third waits, and a mirror never has two",
"%d rounds reached the upstream", len(arrived))
close(release)
waitImportUV()
mu.Lock()
first := append([]bool(nil), indexedFirst...)
mu.Unlock()
all := len(first) >= 3
for _, ok := range first {
all = all && ok
}
check(t, all, "each round starts after its pull was indexed", "indexed before the round: %v", first)
pa, _ := mirrors.MirrorPulls("a", 0)
check(t, len(pa) == 2 && !pa[0].UVRan && pa[1].UVRan,
"the round is recorded on the pull it followed; the pull skipped while it ran has none",
"a's pulls: %+v", pa)
}
func xferContainer(body string) []byte {
return compress([]byte(body))
}
// Words the upstream supplies are bounded and lose their control
// characters before they reach a row.
func TestMirrorUVNotesAreBounded(t *testing.T) {
long := "upstream refused: " + strings.Repeat("é\x1b[31m\n", 2000)
got := clipNote(long, uvNoteMax)
check(t, len(got) <= uvNoteMax && utf8.ValidString(got) && !strings.ContainsAny(got, "\x1b\n"),
"a long note full of escapes is cut to the bound, valid and printable",
"clipped to %d bytes: %q", len(got), got)
var notes []string
for i := 0; i < 50; i++ {
notes = append(notes, clipNote(long, uvNoteMax))
}
all := clipNote(joinNotes(notes), uvMsgMax)
check(t, len(all) <= uvMsgMax && strings.HasSuffix(all, "and 45 more"),
"a round's message names five notes and counts the rest, within its bound",
"message of %d bytes ends %q", len(all), all[max(0, len(all)-20):])
}