package archive
import (
"context"
"errors"
"os"
"sync"
"sync/atomic"
"testing"
"time"
)
// memObjects is a bucket in memory that counts what it was asked.
type memObjects struct {
where string
mu sync.Mutex
objects map[string][]byte
heads atomic.Int32
uploads atomic.Int32
}
func newMemObjects(where string) *memObjects {
return &memObjects{where: where, objects: map[string][]byte{}}
}
func (m *memObjects) Where() string { return m.where }
func (m *memObjects) Has(_ context.Context, key string) (bool, error) {
m.heads.Add(1)
m.mu.Lock()
defer m.mu.Unlock()
_, ok := m.objects[key]
return ok, nil
}
func (m *memObjects) Upload(_ context.Context, key, path string) error {
m.uploads.Add(1)
b, err := os.ReadFile(path)
if err != nil {
return err
}
m.mu.Lock()
defer m.mu.Unlock()
m.objects[key] = b
return nil
}
// Many readers asking for the same archive at once cost one build and
// one upload; the rest wait for it and are answered by it.
func TestBucketedBuildsOnceUnderConcurrency(t *testing.T) {
b := NewBucketed()
objs := newMemObjects("e/b")
var builds atomic.Int32
release := make(chan struct{})
build := func(dst string) error {
builds.Add(1)
<-release
return os.WriteFile(dst, []byte("tarball"), 0o644)
}
const n = 16
var wg sync.WaitGroup
var ran atomic.Int32
errs := make(chan error, n)
for range n {
wg.Add(1)
go func() {
defer wg.Done()
res, err := b.Ensure(context.Background(), objs, "archive/r/h.tar.gz", build)
if res.Built {
ran.Add(1)
}
errs <- err
}()
}
time.Sleep(50 * time.Millisecond)
close(release)
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
t.Fatalf("FAIL: %v", err)
}
}
if builds.Load() != 1 || objs.uploads.Load() != 1 || ran.Load() != 1 {
t.Fatalf("FAIL: builds=%d uploads=%d built-reports=%d, want 1 each", builds.Load(), objs.uploads.Load(), ran.Load())
}
if string(objs.objects["archive/r/h.tar.gz"]) != "tarball" {
t.Fatalf("FAIL: object = %q", objs.objects["archive/r/h.tar.gz"])
}
t.Logf("ok: %d readers, 1 build, 1 upload", n)
}
// A key once seen in the bucket is not asked about again, and one the
// bucket already holds is never built.
func TestBucketedRemembersAndSkipsWhatTheBucketHolds(t *testing.T) {
b := NewBucketed()
objs := newMemObjects("e/b")
objs.objects["archive/r/h.zip"] = []byte("old")
build := func(string) error { t.Fatal("FAIL: built an archive the bucket holds"); return nil }
for range 3 {
res, err := b.Ensure(context.Background(), objs, "archive/r/h.zip", build)
if err != nil || res.Built {
t.Fatalf("FAIL: res=%+v err=%v", res, err)
}
}
if objs.heads.Load() != 1 {
t.Fatalf("FAIL: %d HEADs, want 1: a positive answer is remembered", objs.heads.Load())
}
// Another bucket under the same key knows nothing of the first.
other := newMemObjects("e/other")
var built atomic.Int32
if _, err := b.Ensure(context.Background(), other, "archive/r/h.zip", func(dst string) error {
built.Add(1)
return os.WriteFile(dst, []byte("x"), 0o644)
}); err != nil || built.Load() != 1 {
t.Fatalf("FAIL: rebinding to another bucket did not build there: built=%d err=%v", built.Load(), err)
}
t.Log("ok")
}
// Forget drops what is remembered under a prefix, so a deleted
// repository's archives are asked about again rather than assumed.
func TestBucketedForgetsAPrefix(t *testing.T) {
b := NewBucketed()
objs := newMemObjects("e/b")
objs.objects["archive/a/h.zip"] = []byte("a")
objs.objects["archive/ab/h.zip"] = []byte("ab")
ctx := context.Background()
none := func(string) error { return errors.New("no build") }
b.Ensure(ctx, objs, "archive/a/h.zip", none)
b.Ensure(ctx, objs, "archive/ab/h.zip", none)
b.Forget("archive/a/")
delete(objs.objects, "archive/a/h.zip")
if _, err := b.Ensure(ctx, objs, "archive/a/h.zip", none); err == nil {
t.Fatal("FAIL: a forgotten key was still assumed present")
}
before := objs.heads.Load()
if _, err := b.Ensure(ctx, objs, "archive/ab/h.zip", none); err != nil || objs.heads.Load() != before {
t.Fatalf("FAIL: Forget(archive/a/) touched archive/ab/: err=%v", err)
}
t.Log("ok")
}
// A failed build uploads nothing and is not remembered; the next
// reader builds again.
func TestBucketedFailedBuildIsNotRemembered(t *testing.T) {
b := NewBucketed()
objs := newMemObjects("e/b")
if _, err := b.Ensure(context.Background(), objs, "k", func(string) error { return errors.New("boom") }); err == nil {
t.Fatal("FAIL: a failed build reported success")
}
if objs.uploads.Load() != 0 {
t.Fatal("FAIL: a failed build was uploaded")
}
res, err := b.Ensure(context.Background(), objs, "k", func(dst string) error { return os.WriteFile(dst, []byte("ok"), 0o644) })
if err != nil || !res.Built || res.Size != 2 {
t.Fatalf("FAIL: res=%+v err=%v", res, err)
}
t.Log("ok")
}