package api
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/archive"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/work"
)
// Five check-ins: v1 is tagged, mid is neither a tip nor tagged (the
// expensive, crawlable case), tip is trunk's tip, dev an open branch's
// tip with a landing open onto trunk, and old a closed branch's tip.
var (
arcV1 = strings.Repeat("a1", 32)
arcMid = strings.Repeat("b2", 32)
arcTip = strings.Repeat("c3", 32)
arcDev = strings.Repeat("d4", 32)
arcOld = strings.Repeat("e5", 32)
)
func archiveIndex(t *testing.T) *index.Index {
t.Helper()
ix, err := index.Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
if err := ix.FoldNow([]index.Parsed{
{Hash: arcV1, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:00:00", User: "dan", Comment: "root",
Tags: []artifact.Tag{{Name: "*branch", Target: "*", Value: "trunk"}, {Name: "*sym-trunk", Target: "*"}, {Name: "+sym-v1.0", Target: "*"}}}},
{Hash: arcMid, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:10:00", User: "dan", Comment: "mid",
Parents: []string{arcV1}}},
{Hash: arcTip, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:20:00", User: "dan", Comment: "tip",
Parents: []string{arcMid}}},
{Hash: arcDev, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:30:00", User: "dan", Comment: "dev",
Parents: []string{arcMid}, Tags: []artifact.Tag{{Name: "*branch", Target: "*", Value: "dev"}, {Name: "*sym-dev", Target: "*"}, {Name: "-sym-trunk", Target: "*"}}}},
{Hash: arcOld, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:40:00", User: "dan", Comment: "old",
Parents: []string{arcMid}, Tags: []artifact.Tag{{Name: "*branch", Target: "*", Value: "old"}, {Name: "*sym-old", Target: "*"},
{Name: "-sym-trunk", Target: "*"}, {Name: "+closed", Target: "*"}}}},
{Hash: "f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6f6", Art: &artifact.Artifact{Type: artifact.Forum, ThreadTitle: "land dev",
Date: "2026-08-15T10:00:00", User: "dan",
Body: []byte("zit-landing: 1\nbranch: dev\ntarget: trunk\n\nready\n")}},
}); err != nil {
t.Fatal(err)
}
return ix
}
type archiveRig struct {
r *ops.Registry
d Deps
bucket *uvBucket
builds atomic.Int32
// gate, when set, holds every build until it is closed.
gate chan struct{}
// started hears each build as it begins.
started chan string
}
// arcProducer is zitcli.ProbeVersion's answer: the bare version.
const arcProducer = "0.3.30"
func newArchiveRig(t *testing.T, mutate func(*Deps, *memBuckets, string)) *archiveRig {
t.Helper()
srv, bucket := newUVBucket(t)
rig := &archiveRig{bucket: bucket, started: make(chan string, 64)}
store := newMemBuckets()
store.BindBucket("test", oltp.BucketBinding{Endpoint: srv.URL, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "s3cret"})
repo := &Repo{Index: archiveIndex(t), Archive: func(ctx context.Context, format, version, name, dst string, max int64) error {
rig.builds.Add(1)
rig.started <- version
if rig.gate != nil {
<-rig.gate
}
return os.WriteFile(dst, []byte(format+" of "+version), 0o644)
}, MergeArchive: func(ctx context.Context, branchTip, targetTip, name, dst string, max int64) error {
rig.builds.Add(1)
rig.started <- "pair:" + branchTip + ":" + targetTip
if rig.gate != nil {
<-rig.gate
}
return os.WriteFile(dst, []byte("pair"), 0o644)
}}
rig.d = Deps{Version: "v", ZitVersion: arcProducer, Repos: NewRepos(map[string]*Repo{"test": repo}), Buckets: store,
ArchiveReaderLane: work.NewLane("reader archive builds", 2, 8), ArchiveLane: work.NewLane("archive builds", 1, 4),
ArchiveColdLane: work.NewLane("cold archive builds", 1, 0), ArchiveObjects: archive.NewBucketed()}
if mutate != nil {
mutate(&rig.d, store, srv.URL)
}
rig.r = ops.NewRegistry()
if err := RegisterAll(rig.r, rig.d); err != nil {
t.Fatal(err)
}
return rig
}
func signedIn(user string) context.Context { return ops.WithUser(context.Background(), user) }
// get serves archive.get the way both the web route and the API do:
// the op, then ServeFile.
func (rig *archiveRig) get(ctx context.Context, version, format string) *httptest.ResponseRecorder {
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/r/test/archive/"+version, nil).WithContext(ctx)
out, err := rig.r.Invoke(ctx, "archive.get", archiveIn{Repo: "test", Version: version, Format: format})
if err != nil {
writeOpError(w, req, "archive.get", err)
return w
}
ServeFile(w, req, "archive.get", out.(ops.File))
return w
}
// put stores an object under the fake bucket's lock: a sweep may be
// running in the background.
func (rig *archiveRig) put(key string) {
rig.bucket.mu.Lock()
defer rig.bucket.mu.Unlock()
rig.bucket.objects[key] = uvObject{body: []byte("x")}
}
func (rig *archiveRig) objectKeys() []string {
rig.bucket.mu.Lock()
defer rig.bucket.mu.Unlock()
var keys []string
for k := range rig.bucket.objects {
keys = append(keys, k)
}
return keys
}
// The first request for an archive builds it once however many readers
// arrive together, puts it in the bucket, and every one of them is sent
// there; a later request is redirected without building.
func TestArchiveIsBuiltOnceIntoTheBucketAndRedirected(t *testing.T) {
rig := newArchiveRig(t, nil)
rig.gate = make(chan struct{})
const n = 8
codes := make(chan *httptest.ResponseRecorder, n)
var wg sync.WaitGroup
for range n {
wg.Add(1)
go func() { defer wg.Done(); codes <- rig.get(context.Background(), arcTip, "tar") }()
}
<-rig.started
close(rig.gate)
wg.Wait()
close(codes)
for w := range codes {
if w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d, want 302: %s", w.Code, w.Body.String())
}
}
if got := rig.builds.Load(); got != 1 {
t.Fatalf("FAIL: %d builds for %d concurrent readers, want 1", got, n)
}
want := "files/archive/test/0.3.30/" + arcTip + ".tar.gz"
if keys := rig.objectKeys(); len(keys) != 1 || keys[0] != want {
t.Fatalf("FAIL: bucket holds %v, want [%s]", keys, want)
}
if body := string(rig.bucket.objects[want].body); body != "tar of "+arcTip {
t.Fatalf("FAIL: object body %q", body)
}
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusFound || rig.builds.Load() != 1 {
t.Fatalf("FAIL: second request: status %d, builds %d", w.Code, rig.builds.Load())
}
loc, err := url.Parse(w.Header().Get("Location"))
if err != nil {
t.Fatal(err)
}
q := loc.Query()
if !strings.HasSuffix(loc.Path, "/archive/test/0.3.30/"+arcTip+".tar.gz") ||
q.Get("X-Amz-Expires") != "300" ||
q.Get("response-content-disposition") != "attachment; filename*=UTF-8''test-"+arcTip[:12]+".tar.gz" ||
q.Get("response-content-type") != "application/gzip" {
t.Fatalf("FAIL: Location %s", loc)
}
if cc := w.Header().Get("Cache-Control"); !strings.Contains(cc, "no-store") {
t.Fatalf("FAIL: a redirect to a five-minute link was cacheable: %q", cc)
}
if w.Body.Len() > 512 {
t.Fatalf("FAIL: the redirect carried %d bytes; archive bytes must not pass through the forge", w.Body.Len())
}
t.Logf("ok: %d readers, 1 build, then 302 to %s", n, loc.Path)
}
// Keys carry the repository, so two repositories sharing one bucket
// never answer each other's archives.
func TestArchiveKeyIncludesTheRepository(t *testing.T) {
a := archiveObjectKey("alpha", arcProducer, arcTip, "zip")
b := archiveObjectKey("beta", arcProducer, arcTip, "zip")
if a == b || !strings.HasPrefix(a, "archive/alpha/") || !strings.HasSuffix(a, arcTip+".zip") {
t.Fatalf("FAIL: %q vs %q", a, b)
}
if archiveObjectKey("alpha", "0.3.31", arcTip, "zip") == a {
t.Fatal("FAIL: a new producer reuses the old producer's object")
}
t.Logf("ok: %s", a)
}
// Three lanes. Signed-in readers have their own; anonymous requests for
// an open branch's tip or a tagged check-in use the normal one; every
// other anonymous request, a closed branch's tip included, the cold
// one, which sheds with 503 and Retry-After when full.
func TestArchiveLanesSplitReadersTipsAndCold(t *testing.T) {
rig := newArchiveRig(t, nil)
rig.gate = make(chan struct{})
first := make(chan *httptest.ResponseRecorder, 1)
go func() { first <- rig.get(context.Background(), arcMid, "tar") }()
<-rig.started
for _, v := range []string{arcMid, arcOld} {
w := rig.get(context.Background(), v, "zip")
if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") == "" {
t.Fatalf("FAIL: cold %.8s while the cold lane is full: status %d Retry-After %q, want 503 with one",
v, w.Code, w.Header().Get("Retry-After"))
}
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 || st.Shed != 2 {
t.Fatalf("FAIL: cold lane %+v, want 1 admitted and 2 shed (mid and the closed tip)", st)
}
served := make(chan *httptest.ResponseRecorder, 5)
go func() { served <- rig.get(context.Background(), arcTip, "zip") }() // trunk's tip
go func() { served <- rig.get(context.Background(), arcDev, "zip") }() // an open branch's tip
go func() { served <- rig.get(context.Background(), arcV1, "zip") }() // tagged
go func() { served <- rig.get(context.Background(), "tag:v1.0", "tar") }() // tag by name
go func() { served <- rig.get(signedIn("dan"), arcMid, "zip") }() // signed in
// One runs on the normal lane and one on the readers' while the
// cold one is still held; the rest wait on the normal lane.
for range 2 {
<-rig.started
}
close(rig.gate)
for range 5 {
if w := <-served; w.Code != http.StatusFound {
t.Fatalf("FAIL: a warm request: status %d %s", w.Code, w.Body.String())
}
}
if w := <-first; w.Code != http.StatusFound {
t.Fatalf("FAIL: the first cold request: status %d", w.Code)
}
if st := rig.d.ArchiveLane.Stats(); st.Admitted != 4 || st.Shed != 0 {
t.Fatalf("FAIL: normal lane %+v, want 4 admitted", st)
}
if st := rig.d.ArchiveReaderLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: reader lane %+v, want 1 admitted", st)
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: cold lane %+v after the warm requests, want still 1 admitted", st)
}
t.Log("ok: readers, open tips and tags, and cold each on their own lane")
}
// An archive the bucket already holds is a redirect even while the
// lane that would build it is full: serving it takes no lane.
func TestArchiveInTheBucketIsServedWhileTheColdLaneIsFull(t *testing.T) {
rig := newArchiveRig(t, nil)
rig.put("files/" + archiveObjectKey("test", arcProducer, arcOld, "zip"))
rig.gate = make(chan struct{})
first := make(chan *httptest.ResponseRecorder, 1)
go func() { first <- rig.get(context.Background(), arcMid, "tar") }()
<-rig.started
if w := rig.get(context.Background(), arcOld, "zip"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d, want 302 without a lane", w.Code)
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 || st.Shed != 0 {
t.Fatalf("FAIL: cold lane %+v: serving a held archive took or was refused a lane", st)
}
close(rig.gate)
<-first
t.Log("ok")
}
// A pair archive (RFC 0009) from an anonymous reader is a cold build
// unless the pair is an open landing's branch tip and target tip; a
// signed-in reader's uses the readers' lane.
func TestPairArchivesAreColdUnlessAnOpenLandingsPair(t *testing.T) {
rig := newArchiveRig(t, nil)
cases := []struct {
ctx context.Context
pair string
lane func() work.Stats
}{
{context.Background(), "pair:" + arcDev + ":" + arcTip, rig.d.ArchiveLane.Stats},
{context.Background(), "pair:" + arcDev + ":" + arcV1, rig.d.ArchiveColdLane.Stats},
{context.Background(), "pair:" + arcOld + ":" + arcTip, rig.d.ArchiveColdLane.Stats},
{signedIn("dan"), "pair:" + arcOld + ":" + arcTip, rig.d.ArchiveReaderLane.Stats},
}
for _, c := range cases {
before := c.lane().Admitted
if w := rig.get(c.ctx, c.pair, "tar"); w.Code != http.StatusOK {
t.Fatalf("FAIL: %s: status %d %s", c.pair, w.Code, w.Body.String())
}
if got := c.lane(); got.Admitted != before+1 {
t.Fatalf("FAIL: %s did not run on %s", c.pair, got.Name)
}
}
t.Log("ok: an open landing's pair is shared work; any other anonymous pair is cold")
}
// With no zit version known, the producer cannot be named, so no key is
// trustworthy: the bucket is not used at all.
func TestArchiveWithAnUnknownZitVersionSkipsTheBucket(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, _ *memBuckets, _ string) { d.ZitVersion = "" })
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusOK || w.Body.String() != "tar of "+arcTip {
t.Fatalf("FAIL: status %d body %q", w.Code, w.Body.String())
}
if keys := rig.objectKeys(); len(keys) != 0 {
t.Fatalf("FAIL: the bucket was written with no producer known: %v", keys)
}
t.Log("ok: served from the forge")
}
// A bucket that refuses the PUT costs one build per check-in, not two
// per request: the built file is served (through the disk cache) and
// the bucket is left alone for a while.
func TestArchiveUploadFailureServesTheBuiltFile(t *testing.T) {
dir := t.TempDir()
rig := newArchiveRig(t, func(d *Deps, _ *memBuckets, _ string) {
c, err := archive.New(dir, 1<<20)
if err != nil {
t.Fatal(err)
}
d.Archives = c
})
rig.bucket.mu.Lock()
rig.bucket.refusePut = true
rig.bucket.mu.Unlock()
for range 3 {
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusOK || w.Body.String() != "tar of "+arcTip {
t.Fatalf("FAIL: status %d body %q", w.Code, w.Body.String())
}
}
if got := rig.builds.Load(); got != 1 {
t.Fatalf("FAIL: %d builds for three requests of one check-in, want 1", got)
}
if w := rig.get(context.Background(), arcV1, "tar"); w.Code != http.StatusOK || rig.builds.Load() != 2 {
t.Fatalf("FAIL: another check-in: status %d builds %d, want one more build", w.Code, rig.builds.Load())
}
t.Log("ok: one build per check-in while the bucket refuses")
}
// repo.delete starts removing the repository's archives from its own
// bucket and the forge's, whatever their last probe said.
func TestRepoDeleteDropsArchivesFromBothBuckets(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, store *memBuckets, endpoint string) {
store.BindBucket(oltp.ForgeBucket, oltp.BucketBinding{Endpoint: endpoint, Bucket: "forge", Region: "auto", AccessKey: "AK", Secret: "s3cret"})
store.SetBucketProbe("test", time.Now(), "403 on PUT", false)
d.Registry = &fakeRegistry{}
})
rig.put("files/archive/test/0.3.30/" + arcTip + ".zip")
rig.put("forge/archive/test/0.3.30/" + arcTip + ".zip")
rig.put("forge/archive/other/0.3.30/" + arcTip + ".zip")
if _, err := rig.r.Invoke(signedIn("dan"), "repo.delete", repoDeleteIn{Name: "test", Confirm: "test"}); err != nil {
t.Fatalf("FAIL: %v", err)
}
deadline := time.Now().Add(5 * time.Second)
for {
keys := rig.objectKeys()
if len(keys) == 1 && keys[0] == "forge/archive/other/0.3.30/"+arcTip+".zip" {
break
}
if time.Now().After(deadline) {
t.Fatalf("FAIL: after repo.delete the buckets hold %v", keys)
}
time.Sleep(20 * time.Millisecond)
}
t.Log("ok: both buckets swept of the deleted repository, nothing else touched")
}
// A private repository refuses a reader who may not see it before
// anything is built or asked of the bucket.
func TestPrivateArchiveRefusesBeforeAnyBuild(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, _ *memBuckets, _ string) {
d.CanRead = func(ctx context.Context, repo string) bool { return ops.User(ctx) == "dan" }
})
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusNotFound {
t.Fatalf("FAIL: status %d, want 404", w.Code)
}
if rig.builds.Load() != 0 || len(rig.objectKeys()) != 0 {
t.Fatalf("FAIL: builds=%d objects=%v for a refused reader", rig.builds.Load(), rig.objectKeys())
}
if w := rig.get(signedIn("dan"), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: a permitted reader: status %d", w.Code)
}
t.Log("ok: refused with nothing built; the permitted reader is redirected")
}
// With no bucket anywhere the archive is built and served as before,
// through the same bounded queue.
func TestArchiveWithoutABucketIsServedThroughTheQueue(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, store *memBuckets, _ string) { store.UnbindBucket("test") })
w := rig.get(context.Background(), arcTip, "zip")
if w.Code != http.StatusOK || w.Body.String() != "zip of "+arcTip {
t.Fatalf("FAIL: status %d body %q", w.Code, w.Body.String())
}
if st := rig.d.ArchiveLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: normal lane %+v, want the build admitted through it", st)
}
w = rig.get(context.Background(), arcMid, "zip")
if w.Code != http.StatusOK {
t.Fatalf("FAIL: cold, no bucket: status %d", w.Code)
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: cold lane %+v, want the cold build admitted through it", st)
}
t.Log("ok: built per request, through the lanes")
}
// A repository with no bucket of its own keeps its archives in the
// forge's.
func TestArchiveFallsBackToTheForgeBucket(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, store *memBuckets, endpoint string) {
store.UnbindBucket("test")
store.BindBucket(oltp.ForgeBucket, oltp.BucketBinding{Endpoint: endpoint, Bucket: "forge", Region: "auto", AccessKey: "AK", Secret: "s3cret"})
})
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
if keys := rig.objectKeys(); len(keys) != 1 || !strings.HasPrefix(keys[0], "forge/archive/test/") {
t.Fatalf("FAIL: objects %v", keys)
}
t.Log("ok: built into the forge's bucket")
}
// A check-in this repository does not hold is not served from the
// bucket, even when an object for that hash is there: the bucket is
// asked only about check-ins the index has.
func TestArchiveOfAnUnknownCheckinIsNotRedirected(t *testing.T) {
rig := newArchiveRig(t, nil)
ghost := strings.Repeat("ee", 32)
rig.put("files/" + archiveObjectKey("test", arcProducer, ghost, "tar"))
w := rig.get(context.Background(), ghost, "tar")
if w.Code == http.StatusFound {
t.Fatalf("FAIL: a check-in the repository does not hold was redirected to %s", w.Header().Get("Location"))
}
t.Logf("ok: status %d", w.Code)
}
// Deleting a repository removes its archives from the bucket, and only
// its own.
func TestDroppingArchivesRemovesOnlyThatRepository(t *testing.T) {
rig := newArchiveRig(t, nil)
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
rig.put("files/archive/test2/0.3.30/" + arcTip + ".tar.gz")
b, ok := rig.d.archiveBucket("test")
if !ok {
t.Fatal("FAIL: no archive bucket")
}
if err := rig.d.dropArchives(context.Background(), "test", b); err != nil {
t.Fatalf("FAIL: %v", err)
}
if keys := rig.objectKeys(); len(keys) != 1 || !strings.Contains(keys[0], "/test2/") {
t.Fatalf("FAIL: objects after drop %v", keys)
}
// The process forgot it too: the next request builds again.
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound || rig.builds.Load() != 2 {
t.Fatalf("FAIL: after drop: status %d builds %d", w.Code, rig.builds.Load())
}
t.Log("ok")
}
// Archives an older zit built are swept from a repository's prefix in
// the background, after the first build for it in this process: the
// current producer's stay, another repository's are never touched, and
// a run deletes at most its cap.
func TestOldProducersArchivesAreSwept(t *testing.T) {
rig := newArchiveRig(t, nil)
put := func(key string) { rig.put("files/" + key) }
put("archive/test/0.3.29/" + arcMid + ".zip")
put("archive/test/0.3.28/" + arcMid + ".tar.gz")
put("archive/test/" + arcMid + ".zip") // written with no producer
put("archive/test/0.3.30/" + arcV1 + ".zip")
put("archive/test2/0.3.29/" + arcMid + ".zip")
put("unversioned/test/0.3.29/file")
b, _ := rig.d.archiveBucket("test")
n, err := rig.d.sweepArchives(context.Background(), "test", b, 2)
if err != nil || n != 2 {
t.Fatalf("FAIL: first run deleted %d (err %v), want the cap of 2", n, err)
}
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
// The build starts the sweep; it runs in the background.
want := map[string]bool{
"files/archive/test/0.3.30/" + arcV1 + ".zip": true,
"files/archive/test/0.3.30/" + arcTip + ".tar.gz": true,
"files/archive/test2/0.3.29/" + arcMid + ".zip": true,
"files/unversioned/test/0.3.29/file": true,
}
deadline := time.Now().Add(5 * time.Second)
for {
keys := rig.objectKeys()
ok := len(keys) == len(want)
for _, k := range keys {
ok = ok && want[k]
}
if ok {
break
}
if time.Now().After(deadline) {
t.Fatalf("FAIL: bucket holds %v, want %v", keys, want)
}
time.Sleep(20 * time.Millisecond)
}
t.Log("ok: old producers swept, current kept, other prefixes untouched")
}