package main
import (
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/errlog"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
)
// A completed run raises ci.completed on the event plane, carrying the
// runId — the one place a run's terminal transition is learned, so
// every live page can wait on the event instead of asking dagnats on
// its own timer (RFC 0005).
func TestPollCIRaisesCICompletedWithRunID(t *testing.T) {
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{
RunID: "r1",
Status: "completed",
CompletedAt: "2026-08-29T12:00:00Z",
})
}))
defer dagnats.Close()
engine := ci.New(dagnats.URL)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err := db.AddCIRun(oltp.CIRun{RunID: "r1", Repo: "alpha", Checkin: "c1",
Workflow: "w", Pipeline: "ci/p.json", Status: "running"}); err != nil {
t.Fatal(err)
}
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
sub, err := ev.Conn().SubscribeSync("quarry.evt.alpha.ci.completed")
if err != nil {
t.Fatal(err)
}
defer sub.Unsubscribe()
ev.Conn().Flush()
pollCI(engine, db, nil, false, t.TempDir(), ev)
msg, err := sub.NextMsg(3 * time.Second)
if err != nil {
t.Fatalf("ci.completed not raised: %v", err)
}
var got struct {
Kind string `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
if err := json.Unmarshal(msg.Data, &got); err != nil {
t.Fatal(err)
}
if got.Kind != "ci.completed" || got.Repo != "alpha" {
t.Fatalf("event = %+v, want kind ci.completed repo alpha", got)
}
if got.Payload["runId"] != "r1" {
t.Fatalf("payload = %+v, want runId r1", got.Payload)
}
rows, err := db.CIRuns("alpha", 10)
if err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].Status != "completed" || rows[0].Completed == "" {
t.Fatalf("run not recorded terminal: %+v", rows)
}
}
// A verdict that could not be written is written on the next pass.
//
// The run used to be marked complete first, so a failed tag write left
// a finished row the poller never asks about again, and the landing
// waited on a result that would never arrive.
func TestAVerdictThatFailedToWriteIsWrittenOnTheNextPass(t *testing.T) {
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{RunID: "r1", Status: "completed",
CompletedAt: "2026-08-29T12:00:00Z"})
}))
defer dagnats.Close()
engine := ci.New(dagnats.URL)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err := db.AddCIRun(oltp.CIRun{RunID: "r1", Repo: "alpha", Checkin: "c1",
Workflow: "w", Pipeline: "ci/p.json", Status: "running"}); err != nil {
t.Fatal(err)
}
// The lease holder is the seam every tag write passes through: the
// first attempt fails, the next one lands.
attempts, written := 0, 0
repos := api.NewRepos(map[string]*api.Repo{"alpha": {
Guard: func(func() error) error {
attempts++
if attempts == 1 {
return errors.New("lease lost")
}
written++
return nil
},
}})
unfinished := func() bool {
runs, err := db.UnfinishedCIRuns()
if err != nil {
t.Fatal(err)
}
return len(runs) == 1 && runs[0].RunID == "r1"
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
if attempts == 1 && written == 0 && unfinished() {
t.Log("ok: a failed verdict write leaves the run on the poller's work list")
} else {
t.Fatalf("FAIL: after a failed write attempts=%d written=%d unfinished=%v",
attempts, written, unfinished())
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
if written == 1 && !unfinished() {
t.Log("ok: the next pass writes the verdict and only then completes the run")
} else {
t.Fatalf("FAIL: after the retry attempts=%d written=%d unfinished=%v",
attempts, written, unfinished())
}
}
// ciFixture is a finished run waiting on its verdict: dagnats reports
// it complete, the OLTP row still says running.
func ciFixture(t *testing.T, runID string) (*ci.Engine, *oltp.DB) {
t.Helper()
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{RunID: runID, Status: "completed",
CompletedAt: "2026-08-29T12:00:00Z"})
}))
t.Cleanup(dagnats.Close)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
if err := db.AddCIRun(oltp.CIRun{RunID: runID, Repo: "alpha", Checkin: "c1",
Workflow: "w", Pipeline: "ci/p.json", Status: "running"}); err != nil {
t.Fatal(err)
}
return ci.New(dagnats.URL), db
}
func runFinished(t *testing.T, db *oltp.DB, runID string) bool {
t.Helper()
runs, err := db.UnfinishedCIRuns()
if err != nil {
t.Fatal(err)
}
for _, r := range runs {
if r.RunID == runID {
return false
}
}
return true
}
// A tag that was written but not yet indexed IS written. Fossil's tag
// log is append-only, so retrying it would append a second control
// artifact that clones with the repository and raises tag.created to
// every webhook, once per pass.
func TestAVerdictWrittenButNotIndexedIsNeverWrittenAgain(t *testing.T) {
engine, db := ciFixture(t, "r-sync")
tags, syncs := 0, 0
repos := api.NewRepos(map[string]*api.Repo{"alpha": {
Guard: func(func() error) error { tags++; return nil },
Sync: func() (int, error) {
syncs++
if syncs == 1 {
return 0, errors.New("ingest round failed")
}
return 0, nil
},
}})
pollCI(engine, db, repos, true, t.TempDir(), nil)
pollCI(engine, db, repos, true, t.TempDir(), nil)
if tags == 1 && runFinished(t, db, "r-sync") {
t.Log("ok: one tag after two passes, and the run is complete")
} else {
t.Fatalf("FAIL: tags=%d finished=%v", tags, runFinished(t, db, "r-sync"))
}
}
// A node that does not hold the store's lease cannot write the verdict,
// and asking again every pass does not change that. It waits for the
// holder's verdict to arrive in its index and completes the run then,
// without writing.
func TestANodeWithoutTheLeaseCompletesOnceTheHolderRecordsTheVerdict(t *testing.T) {
engine, db := ciFixture(t, "r-holder")
ix, err := index.Open(filepath.Join(t.TempDir(), "ix.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
attempts := 0
repos := api.NewRepos(map[string]*api.Repo{"alpha": {
Index: ix,
Guard: func(func() error) error {
attempts++
return leaseRefusal(lease.ErrNotHolder)
},
}})
pollCI(engine, db, repos, true, t.TempDir(), nil)
if attempts == 1 && !runFinished(t, db, "r-holder") {
t.Log("ok: without the lease the run waits for the holder")
} else {
t.Fatalf("FAIL: attempts=%d finished=%v", attempts, runFinished(t, db, "r-holder"))
}
// The holder's verdict arrives the way any tag does: folded in.
if err := ix.FoldNow([]index.Parsed{{Hash: strings.Repeat("d4", 32), Art: &artifact.Artifact{
Type: artifact.Control, Date: "2026-08-29T12:00:05", User: "quarry",
Tags: []artifact.Tag{{Name: "+ci-pass", Target: "c1", Value: "r-holder"}}}}}); err != nil {
t.Fatal(err)
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
if attempts == 1 && runFinished(t, db, "r-holder") {
t.Log("ok: the holder's verdict completes the run with no second write")
} else {
t.Fatalf("FAIL: attempts=%d finished=%v", attempts, runFinished(t, db, "r-holder"))
}
}
// A verdict that can never be written — the check-in is gone, the
// store refuses — stops being retried after a bound, and says why.
func TestAVerdictThatCanNeverBeWrittenIsAbandonedAfterABound(t *testing.T) {
engine, db := ciFixture(t, "r-fault")
attempts := 0
repos := api.NewRepos(map[string]*api.Repo{"alpha": {
Guard: func(func() error) error {
attempts++
return errors.New("zit tag add: no such check-in")
},
}})
for i := 1; i < ciVerdictAttempts; i++ {
pollCI(engine, db, repos, true, t.TempDir(), nil)
}
if runFinished(t, db, "r-fault") {
t.Fatalf("FAIL: abandoned before the bound, after %d attempts", attempts)
}
pollCI(engine, db, repos, true, t.TempDir(), nil)
pollCI(engine, db, repos, true, t.TempDir(), nil)
if attempts == ciVerdictAttempts && runFinished(t, db, "r-fault") {
t.Logf("ok: abandoned after %d attempts and never tried again", attempts)
} else {
t.Fatalf("FAIL: attempts=%d finished=%v", attempts, runFinished(t, db, "r-fault"))
}
logged := false
for _, e := range errlog.Default.Recent(50) {
if e.Surface == "ci" && e.Path == "verdict-abandoned" && e.Repo == "alpha" {
logged = true
}
}
if logged {
t.Log("ok: the abandonment is on the error log with its repository")
} else {
t.Fatal("FAIL: no verdict-abandoned entry on the error log")
}
}