package spindle
import (
"context"
"fmt"
"net/http"
"slices"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
)
func TestARepoJobOnAHangingKnotStopsAtItsBudgetAndBacksTheRepoOff(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
if err := env.s.markRepo(testRepoDid); err != nil {
t.Fatal(err)
}
defer func(old time.Duration) { repoJobBudget = old }(repoJobBudget)
repoJobBudget = 100 * time.Millisecond
hang := make(chan struct{})
t.Cleanup(func() { close(hang) })
env.knot.beforeAnswer = func(*http.Request) { <-hang }
finished := make(chan struct{})
go func() {
env.s.runRepoJob(context.Background(), testRepoDid)
close(finished)
}()
select {
case <-finished:
case <-time.After(5 * time.Second):
t.Fatal("a repo job on a hanging knot kept its worker far past the budget")
}
if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 1 {
t.Fatalf("attempts = %d, exists %v, %v, want the repo backed off after running out of time", attempts, marked, err)
}
if _, due, err := env.s.db.RepoRetryDue(testRepoDid, env.s.retries.clock()); err != nil || due {
t.Fatalf("due = %v, %v, want the repo to wait out a backoff before the next job", due, err)
}
}
func TestARepoJobWorksThroughALimitedNumberOfMarksThenYieldsItsWorker(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
const marks = recordMarksPerJob + 50
for i := range marks {
r := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.TrackTicketNSID, Rkey: fmt.Sprintf("3kticket%05d", i)}
if err := env.s.db.MarkRecordDirty(r, time.Now(), recordMarksPerRepo); err != nil {
t.Fatal(err)
}
}
env.s.runRepoJob(context.Background(), testRepoDid)
if left := dueRecords(t, env); len(left) != 50 {
t.Fatalf("%d record marks left, want the %d past the per-job limit", len(left), 50)
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if did, ok := env.s.retries.jobs.next(ctx); !ok || did != testRepoDid {
t.Fatalf("next job = %q, %v, want the repo queued again for the rest of its marks", did, ok)
}
}
func fairJobs(pool int, priority ...string) (*repoJobs, map[syntax.DID]string) {
knots := map[syntax.DID]string{}
set := map[string]bool{}
for _, knot := range priority {
set[knot] = true
}
j := &repoJobs{}
j.configure(pool, func(did syntax.DID) string { return knots[did] }, set)
return j, knots
}
func kickRepos(j *repoJobs, knots map[syntax.DID]string, knot string, n int) []syntax.DID {
var dids []syntax.DID
for i := range n {
did := syntax.DID(fmt.Sprintf("did:plc:%s%03d", knot, i))
knots[did] = knot
j.kick(did)
dids = append(dids, did)
}
return dids
}
func within(d time.Duration, next func(context.Context) (syntax.DID, bool)) (syntax.DID, bool) {
ctx, cancel := context.WithTimeout(context.Background(), d)
defer cancel()
return next(ctx)
}
func startAll(j *repoJobs, knots map[syntax.DID]string) (started []syntax.DID, perKnot map[string]int) {
perKnot = map[string]int{}
for {
did, ok := within(100*time.Millisecond, j.next)
if !ok {
return started, perKnot
}
started = append(started, did)
perKnot[knots[did]]++
}
}
func TestAnIdlePoolKeepsAReserveAndUnservedReposShareAHalf(t *testing.T) {
for _, tc := range []struct {
name string
pool, repos int
knot string
want int
}{
{"an eighth of a large pool", 64, 80, "a", 56},
{"unserved repos get half the pool", 8, 10, "", 4},
} {
t.Run(tc.name, func(t *testing.T) {
j, knots := fairJobs(tc.pool)
kickRepos(j, knots, tc.knot, tc.repos)
if started, _ := startAll(j, knots); len(started) != tc.want {
t.Fatalf("%d jobs started on a pool of %d, want %d", len(started), tc.pool, tc.want)
}
})
}
}
func TestOneKnotOnAnIdlePoolLeavesTheReserveForANewcomer(t *testing.T) {
j, knots := fairJobs(16)
kickRepos(j, knots, "a", 30)
if started, _ := startAll(j, knots); len(started) != 14 {
t.Fatalf("%d jobs started on a pool of 16 with only one knot waiting, want all but the reserve of 2", len(started))
}
newcomer := kickRepos(j, knots, "b", 1)[0]
if did, ok := within(time.Second, j.next); !ok || did != newcomer {
t.Fatalf("next = %q, %v, want a new knot to get a worker at once while another is busy", did, ok)
}
}
func TestKnotsWithJobsWaitingEachGetAnEqualShare(t *testing.T) {
j, knots := fairJobs(8)
kickRepos(j, knots, "a", 8)
kickRepos(j, knots, "b", 8)
_, perKnot := startAll(j, knots)
if perKnot["a"] != 4 || perKnot["b"] != 4 {
t.Fatalf("running per knot = %v, want 4 each while both have jobs waiting", perKnot)
}
}
func TestASlowKnotWithManyReposCantHoldMoreThanItsShareWhileAnotherKnotWaits(t *testing.T) {
j, knots := fairJobs(8)
kickRepos(j, knots, "hostile", 30)
healthy := kickRepos(j, knots, "healthy", 1)[0]
var firstFive []syntax.DID
var hostile int
for range 5 {
did, ok := within(time.Second, j.next)
if !ok {
t.Fatalf("only %d jobs started, want the healthy knot's repo to get a worker", len(firstFive))
}
firstFive = append(firstFive, did)
if knots[did] == "hostile" {
hostile++
}
}
if hostile != 4 || !slices.Contains(firstFive, healthy) {
t.Fatalf("hostile knot started %d of the first 5 (%v), want 4 and the healthy repo among them", hostile, firstFive)
}
}
func TestAFinishedJobGivesItsSlotToAWaitingOne(t *testing.T) {
j, knots := fairJobs(8)
kickRepos(j, knots, "a", 5)
kickRepos(j, knots, "b", 5)
started, perKnot := startAll(j, knots)
if perKnot["a"] != 4 {
t.Fatalf("running per knot = %v, want knot a at its share of 4", perKnot)
}
j.finish(started[0])
if _, ok := within(time.Second, j.next); !ok {
t.Fatal("a finished job didn't free a slot for a waiting one")
}
}
func TestTheRepoJobWorkerCountComesFromTheConfig(t *testing.T) {
env := newEnsureEnv(t)
if got := env.s.repoJobWorkerCount(); got != defaultRepoJobWorkers {
t.Fatalf("workers = %d with nothing configured, want the default %d", got, defaultRepoJobWorkers)
}
env.s.cfg.Server.RepoJobWorkers = 7
if got := env.s.repoJobWorkerCount(); got != 7 {
t.Fatalf("workers = %d, want the configured 7", got)
}
}
func TestAPriorityKnotsJobRunsAtOnceWhileAHostileKnotHoldsTheSharedPool(t *testing.T) {
j, knots := fairJobs(8, "priority.example.com")
kickRepos(j, knots, "hostile", 30)
if started, _ := startAll(j, knots); len(started) != 6 {
t.Fatalf("%d hostile jobs started, want them held to the shared pool less its reserve", len(started))
}
vip := kickRepos(j, knots, "priority.example.com", 1)[0]
if did, ok := within(time.Second, j.nextPriority); !ok || did != vip {
t.Fatalf("priority worker got %q, %v, want the priority knot's job at once", did, ok)
}
}
func TestANonPriorityKnotCantUseThePriorityWorkers(t *testing.T) {
j, knots := fairJobs(8, "priority.example.com")
kickRepos(j, knots, "other", 5)
if did, ok := within(200*time.Millisecond, j.nextPriority); ok {
t.Fatalf("a priority worker took %q of a knot that isn't a priority one", did)
}
}
func TestAPriorityKnotsJobCanAlsoUseASharedWorker(t *testing.T) {
j, knots := fairJobs(8, "priority.example.com")
vip := kickRepos(j, knots, "priority.example.com", 1)[0]
if did, ok := within(time.Second, j.next); !ok || did != vip {
t.Fatalf("shared worker got %q, %v, want it to take a priority knot's job under the usual rules", did, ok)
}
}
func TestPriorityKnotsMatchAsDidsAndAsHosts(t *testing.T) {
env := newEnsureEnv(t)
env.s.cfg.Server.PriorityKnots = []string{"did:web:knot1.tangled.sh", " KNOT2.example.com ", "did:web:knot3.example.com%3A8443"}
got := env.s.priorityKnots()
for _, want := range []string{"knot1.tangled.sh", "knot2.example.com", "knot3.example.com:8443"} {
if !got[want] {
t.Fatalf("priority knots = %v, want %s to match however it was written", got, want)
}
}
byDid, byHost := priorityKnotHosts([]string{"did:web:knot1.tangled.sh"}), priorityKnotHosts([]string{"knot1.tangled.sh"})
if len(byDid) != 1 || !byDid["knot1.tangled.sh"] || !byHost["knot1.tangled.sh"] {
t.Fatalf("a knot's did gave %v and its host %v, want the same key", byDid, byHost)
}
}
func TestPriorityWorkersOnlyExistWithPriorityKnotsAndDefaultToAQuarter(t *testing.T) {
env := newEnsureEnv(t)
env.s.cfg.Server.RepoJobWorkers = 64
if got := env.s.priorityJobWorkerCount(); got != 0 {
t.Fatalf("priority workers = %d with no priority knots, want none", got)
}
env.s.cfg.Server.PriorityKnots = []string{"knot1.tangled.sh"}
if got := env.s.priorityJobWorkerCount(); got != 16 {
t.Fatalf("priority workers = %d, want a quarter of 64", got)
}
env.s.cfg.Server.PriorityRepoJobWorkers = 5
if got := env.s.priorityJobWorkerCount(); got != 5 {
t.Fatalf("priority workers = %d, want the configured 5", got)
}
}