package main
import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strconv"
"sync"
"testing"
"craftdesign.group/quarry/internal/ci"
)
// laneEngine is dagnats' #624 tail API over a map keyed the way the
// engine keys its hot lane: step, then attempt. A read without
// ?attempt= answers the step's latest attempt, as the engine does.
type laneEngine struct {
mu sync.Mutex
lanes map[string]map[int][]ci.LogChunk
// reads counts the reads of each "step/attempt"; broken names the
// lanes that answer 500.
reads map[string]int
broken map[string]bool
}
func newLane() *laneEngine {
return &laneEngine{lanes: map[string]map[int][]ci.LogChunk{},
reads: map[string]int{}, broken: map[string]bool{}}
}
func (l *laneEngine) readsOf(key string) int {
l.mu.Lock()
defer l.mu.Unlock()
return l.reads[key]
}
func (l *laneEngine) breakLane(key string) {
l.mu.Lock()
defer l.mu.Unlock()
l.broken[key] = true
}
func (l *laneEngine) set(step string, attempt int, chunks []ci.LogChunk) {
l.mu.Lock()
defer l.mu.Unlock()
if l.lanes[step] == nil {
l.lanes[step] = map[int][]ci.LogChunk{}
}
l.lanes[step][attempt] = chunks
}
func (l *laneEngine) ServeHTTP(w http.ResponseWriter, r *http.Request) {
l.mu.Lock()
defer l.mu.Unlock()
step := r.URL.Query().Get("step")
attempt, _ := strconv.Atoi(r.URL.Query().Get("attempt"))
if attempt == 0 {
for n := range l.lanes[step] {
attempt = max(attempt, n)
}
}
key := step + "/" + strconv.Itoa(attempt)
l.reads[key]++
if l.broken[key] {
http.Error(w, "engine hiccup", http.StatusInternalServerError)
return
}
type wire struct {
Seq int `json:"seq"`
Stream string `json:"stream"`
Data []byte `json:"data"`
}
page := struct {
Chunks []wire `json:"chunks"`
EOF bool `json:"eof"`
}{Chunks: []wire{}, EOF: true}
for _, c := range l.lanes[step][attempt] {
page.Chunks = append(page.Chunks, wire{Seq: c.Seq, Stream: c.Stream, Data: []byte(c.Data)})
}
_ = json.NewEncoder(w).Encode(page)
}
func drainedAttempts(t *testing.T, dir, step string) map[int][]ci.LogChunk {
t.Helper()
all, ok := ciLogStore{dir: dir}.DrainedLogs("alpha", "run-1")
if !ok {
t.Fatal("nothing drained")
}
out := map[int][]ci.LogChunk{}
for _, c := range all {
if c.Step == step {
out[c.Attempt] = append(out[c.Attempt], c)
}
}
return out
}
func hasChunk(chunks []ci.LogChunk, stream, data string) bool {
for _, c := range chunks {
if c.Stream == stream && c.Data == data {
return true
}
}
return false
}
// #269: a step that fails its first attempt and passes its second keeps
// both logs, each stamped with its attempt; the failing one is intact.
func TestDrainKeepsEveryAttempt(t *testing.T) {
lane := newLane()
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- FAIL: TestFlake"},
{Seq: 2, Stream: "marker", Data: "failed"}})
lane.set("quarry.test", 2, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "ok"},
{Seq: 2, Stream: "marker", Data: "completed"}})
srv := httptest.NewServer(lane)
defer srv.Close()
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "completed", Attempts: 2}}
if err := drainLogs(ci.New(srv.URL), dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got := drainedAttempts(t, dir, "quarry.test")
if len(got) != 2 {
t.Fatalf("attempts drained = %d, want 2: %+v", len(got), got)
}
if !hasChunk(got[1], "out", "--- FAIL: TestFlake") || !hasChunk(got[1], "marker", "failed") {
t.Fatalf("attempt 1 lost its failure: %+v", got[1])
}
if !hasChunk(got[2], "marker", "completed") {
t.Fatalf("attempt 2 missing: %+v", got[2])
}
}
// #269, run 79670815: the run is marked failed (another step ran out of
// attempts) while this step is mid-retry. The drain keeps finished
// attempt 1 beside running attempt 2, and a later drain — the verdict
// retrying, the hot lane expiring — never replaces the finished
// attempt with anything, while the running one may still grow.
func TestDrainNeverOverwritesAFinishedAttempt(t *testing.T) {
lane := newLane()
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- FAIL: TestFlake"},
{Seq: 2, Stream: "marker", Data: "failed"}})
lane.set("quarry.test", 2, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
srv := httptest.NewServer(lane)
defer srv.Close()
e := ci.New(srv.URL)
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "running", Attempts: 2}}
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got := drainedAttempts(t, dir, "quarry.test")
if !hasChunk(got[1], "marker", "failed") || !hasChunk(got[2], "out", "=== RUN TestFlake") {
t.Fatalf("first drain: %+v", got)
}
// The hot lane moves on: attempt 1's subject now answers with other
// output, attempt 2 has grown.
lane.set("quarry.test", 1, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
lane.set("quarry.test", 2, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"},
{Seq: 2, Stream: "out", Data: "--- PASS: TestFlake"}})
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
if n := lane.readsOf("quarry.test/1"); n != 1 {
t.Errorf("ended attempt 1 was read %d times, want once", n)
}
got = drainedAttempts(t, dir, "quarry.test")
if len(got[1]) != 2 || !hasChunk(got[1], "out", "--- FAIL: TestFlake") || !hasChunk(got[1], "marker", "failed") {
t.Fatalf("finished attempt 1 was overwritten: %+v", got[1])
}
if !hasChunk(got[2], "out", "--- PASS: TestFlake") {
t.Fatalf("running attempt 2 did not grow: %+v", got[2])
}
// And the lane expiring entirely leaves both as they were.
lane.set("quarry.test", 1, nil)
lane.set("quarry.test", 2, nil)
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got = drainedAttempts(t, dir, "quarry.test")
if len(got[1]) != 2 || len(got[2]) != 2 {
t.Fatalf("an expired lane erased the record: %+v", got)
}
}
// A running attempt already holding more than the engine now answers —
// a lane trimmed, a read cut short — keeps what it held.
func TestDrainKeepsTheLongerRunningAttempt(t *testing.T) {
lane := newLane()
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"},
{Seq: 2, Stream: "out", Data: "still going"}})
srv := httptest.NewServer(lane)
defer srv.Close()
e := ci.New(srv.URL)
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "running", Attempts: 1}}
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
lane.set("quarry.test", 1, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
if err := drainLogs(e, dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
if got := drainedAttempts(t, dir, "quarry.test"); len(got[1]) != 2 || !hasChunk(got[1], "out", "still going") {
t.Fatalf("a shorter read replaced the running attempt: %+v", got[1])
}
}
// A record drained before attempts were kept holds the engine's default
// read, which is the step's LATEST attempt; re-draining labels it so,
// not as attempt 1.
func TestDrainLabelsAnOldRecordAsTheLatestAttempt(t *testing.T) {
dir := t.TempDir()
old := []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- PASS: TestFlake", Step: "quarry.test"},
{Seq: 2, Stream: "marker", Data: "completed", Step: "quarry.test"}}
p := ciLogPath(dir, "alpha", "run-1")
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
t.Fatal(err)
}
b, _ := json.Marshal(old)
if err := os.WriteFile(p, b, 0o644); err != nil {
t.Fatal(err)
}
lane := newLane() // the hot lane has expired
srv := httptest.NewServer(lane)
defer srv.Close()
steps := map[string]ci.Step{"quarry.test": {Status: "completed", Attempts: 3}}
if err := drainLogs(ci.New(srv.URL), dir, "alpha", "run-1", steps); err != nil {
t.Fatal(err)
}
got := drainedAttempts(t, dir, "quarry.test")
if len(got) != 1 || len(got[3]) != 2 {
t.Fatalf("old record relabelled as %+v, want attempt 3", got)
}
}
// A transient engine error on one lane still keeps what the other lanes
// answered, reports the error, and leaves no half-written file behind.
func TestDrainKeepsWhatItReadWhenALaneErrs(t *testing.T) {
lane := newLane()
lane.set("quarry.test", 1, []ci.LogChunk{
{Seq: 1, Stream: "out", Data: "--- FAIL: TestFlake"},
{Seq: 2, Stream: "marker", Data: "failed"}})
lane.set("quarry.test", 2, []ci.LogChunk{{Seq: 1, Stream: "out", Data: "=== RUN TestFlake"}})
lane.breakLane("quarry.test/2")
srv := httptest.NewServer(lane)
defer srv.Close()
dir := t.TempDir()
steps := map[string]ci.Step{"quarry.test": {Status: "running", Attempts: 2}}
if err := drainLogs(ci.New(srv.URL), dir, "alpha", "run-1", steps); err == nil {
t.Fatal("a lane error was swallowed")
}
got := drainedAttempts(t, dir, "quarry.test")
if !hasChunk(got[1], "marker", "failed") {
t.Fatalf("attempt 1 was read and then discarded: %+v", got)
}
ents, err := os.ReadDir(filepath.Dir(ciLogPath(dir, "alpha", "run-1")))
if err != nil {
t.Fatal(err)
}
if len(ents) != 1 {
t.Fatalf("record directory holds %d entries, want only the record", len(ents))
}
}