package main
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"os"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/s3"
)
// fakeS3 is an in-memory bucket speaking just enough of the S3 wire:
// PUT stores, GET returns or 404s. down makes every request 503.
type fakeS3 struct {
mu sync.Mutex
objs map[string][]byte
gets int
down bool
}
func newFakeS3() *fakeS3 { return &fakeS3{objs: map[string][]byte{}} }
func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
defer f.mu.Unlock()
if f.down {
w.WriteHeader(http.StatusServiceUnavailable)
return
}
switch r.Method {
case http.MethodPut:
b, _ := io.ReadAll(r.Body)
f.objs[r.URL.Path] = b
w.WriteHeader(http.StatusOK)
case http.MethodGet:
f.gets++
b, ok := f.objs[r.URL.Path]
if !ok {
w.WriteHeader(http.StatusNotFound)
return
}
_, _ = w.Write(b)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
}
func (f *fakeS3) has(path string) []byte {
f.mu.Lock()
defer f.mu.Unlock()
return f.objs[path]
}
func offloadFixture(t *testing.T, age time.Duration, max int64) (*fakeS3, ciLogStore, string) {
t.Helper()
fake := newFakeS3()
srv := httptest.NewServer(fake)
t.Cleanup(srv.Close)
b := s3.Bucket{Endpoint: srv.URL, Name: "forge", Region: "auto", AccessKey: "k", Secret: "s"}
dir := t.TempDir()
off := newCILogOffload(func() (s3.Bucket, bool) { return b, true }, srv.Client(), age, max)
return fake, ciLogStore{dir: dir, off: off}, dir
}
func writeDrained(t *testing.T, dir, repo, run, data string) []byte {
t.Helper()
b, _ := json.Marshal([]ci.LogChunk{{Seq: 1, Stream: "out", Data: data, Step: "s", Attempt: 1}})
if err := os.MkdirAll(dirOf(ciLogPath(dir, repo, run)), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(ciLogPath(dir, repo, run), b, 0o644); err != nil {
t.Fatal(err)
}
return b
}
func dirOf(p string) string { return p[:strings.LastIndex(p, "/")] }
func age(t *testing.T, p string, by time.Duration) {
t.Helper()
old := time.Now().Add(-by)
if err := os.Chtimes(p, old, old); err != nil {
t.Fatal(err)
}
}
func exists(p string) bool { _, err := os.Stat(p); return err == nil }
func TestOffloadUploadsDrainedLogUnderStableKeyAndKeepsCache(t *testing.T) {
fake, store, dir := offloadFixture(t, 0, 0)
want := writeDrained(t, dir, "alpha", "run-1", "hello")
store.Sweep(time.Now())
got := fake.has("/forge/ci-logs/alpha/run-1.json")
if string(got) != string(want) {
t.Fatalf("FAIL: bucket holds %q, want %q", got, want)
}
if !exists(ciLogPath(dir, "alpha", "run-1")) {
t.Fatal("FAIL: local cache copy was removed")
}
}
func TestOffloadReadsThroughOnLocalMissAndRepopulates(t *testing.T) {
_, store, dir := offloadFixture(t, 0, 0)
writeDrained(t, dir, "alpha", "run-1", "hello")
store.Sweep(time.Now())
if err := os.Remove(ciLogPath(dir, "alpha", "run-1")); err != nil {
t.Fatal(err)
}
chunks, ok := store.DrainedLogs("alpha", "run-1")
if !ok || len(chunks) != 1 || chunks[0].Data != "hello" {
t.Fatalf("FAIL: read-through = %+v ok=%v", chunks, ok)
}
if !exists(ciLogPath(dir, "alpha", "run-1")) {
t.Fatal("FAIL: cache not repopulated")
}
}
func TestOffloadMissIsRememberedBriefly(t *testing.T) {
fake, store, _ := offloadFixture(t, 0, 0)
for range 3 {
if _, ok := store.DrainedLogs("alpha", "live-run"); ok {
t.Fatal("FAIL: found a log nobody wrote")
}
}
if fake.gets != 1 {
t.Fatalf("FAIL: %d bucket reads for 3 misses, want 1", fake.gets)
}
}
func TestOffloadEvictsByAgeOnlyWhatIsUploaded(t *testing.T) {
fake, store, dir := offloadFixture(t, time.Hour, 0)
oldUp := ciLogPath(dir, "alpha", "old-up")
oldDown := ciLogPath(dir, "alpha", "old-not-uploaded")
fresh := ciLogPath(dir, "alpha", "fresh")
writeDrained(t, dir, "alpha", "old-up", "a")
writeDrained(t, dir, "alpha", "fresh", "c")
store.Sweep(time.Now()) // uploads both
writeDrained(t, dir, "alpha", "old-not-uploaded", "b")
fake.mu.Lock()
fake.down = true
fake.mu.Unlock()
age(t, oldUp, 2*time.Hour)
age(t, oldDown, 2*time.Hour)
store.Sweep(time.Now())
if exists(oldUp) {
t.Error("FAIL: uploaded old log kept")
}
if !exists(oldDown) {
t.Error("FAIL: log never uploaded was evicted")
}
if !exists(fresh) {
t.Error("FAIL: fresh log evicted")
}
}
func TestOffloadDoesNotEvictLogChangedAfterUpload(t *testing.T) {
fake, store, dir := offloadFixture(t, time.Hour, 0)
p := ciLogPath(dir, "alpha", "run-1")
writeDrained(t, dir, "alpha", "run-1", "first")
store.Sweep(time.Now())
writeDrained(t, dir, "alpha", "run-1", "second") // a re-drain
fake.mu.Lock()
fake.down = true
fake.mu.Unlock()
age(t, p, 2*time.Hour)
store.Sweep(time.Now())
if !exists(p) {
t.Fatal("FAIL: a log with unsent changes was evicted")
}
}
func TestOffloadEvictsOldestUploadedOverSizeBudget(t *testing.T) {
_, store, dir := offloadFixture(t, 0, 1)
a, b := ciLogPath(dir, "alpha", "a"), ciLogPath(dir, "alpha", "b")
writeDrained(t, dir, "alpha", "a", "x")
writeDrained(t, dir, "alpha", "b", "y")
age(t, a, 3*time.Hour)
age(t, b, 2*time.Hour)
store.Sweep(time.Now())
if exists(a) || exists(b) {
t.Fatalf("FAIL: kept a=%v b=%v with a 1-byte budget", exists(a), exists(b))
}
// the evicted logs are still readable from the bucket
if _, ok := store.DrainedLogs("alpha", "a"); !ok {
t.Fatal("FAIL: evicted log unreadable")
}
}
func TestOffloadDisabledChangesNothing(t *testing.T) {
dir := t.TempDir()
for name, store := range map[string]ciLogStore{
"nil": {dir: dir},
"unbound": {dir: dir, off: newCILogOffload(func() (s3.Bucket, bool) { return s3.Bucket{}, false }, nil, time.Nanosecond, 1)},
} {
p := writeDrainedPath(t, dir, name)
age(t, p, 1000*time.Hour)
store.Sweep(time.Now())
if !exists(p) {
t.Errorf("FAIL: %s: log removed with no bucket", name)
}
ents, _ := os.ReadDir(dirOf(p))
for _, e := range ents {
if strings.HasSuffix(e.Name(), ".sent") {
t.Errorf("FAIL: %s: marker %s written with no bucket", name, e.Name())
}
}
if _, ok := store.DrainedLogs("alpha", "absent"); ok {
t.Errorf("FAIL: %s: found a log from nowhere", name)
}
}
}
func writeDrainedPath(t *testing.T, dir, run string) string {
t.Helper()
writeDrained(t, dir, "alpha", run, "z")
return ciLogPath(dir, "alpha", run)
}