package spindle
import (
"context"
"sync"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/tapc"
)
type jobLog struct {
mu sync.Mutex
runs map[syntax.DID]int
current map[syntax.DID]int
overlap bool
}
func (l *jobLog) started(did syntax.DID) int {
l.mu.Lock()
defer l.mu.Unlock()
if l.runs == nil {
l.runs, l.current = map[syntax.DID]int{}, map[syntax.DID]int{}
}
l.runs[did]++
l.current[did]++
if l.current[did] > 1 {
l.overlap = true
}
return l.runs[did]
}
func (l *jobLog) ended(did syntax.DID) {
l.mu.Lock()
defer l.mu.Unlock()
l.current[did]--
}
func (l *jobLog) count(did syntax.DID) int {
l.mu.Lock()
defer l.mu.Unlock()
return l.runs[did]
}
func runWorkers(t *testing.T, jobs *repoJobs, n int, run func(context.Context, syntax.DID)) {
ctx, cancel := context.WithCancel(context.Background())
var workers sync.WaitGroup
for range n {
workers.Go(func() { jobs.work(ctx, run) })
}
t.Cleanup(func() { cancel(); workers.Wait() })
}
func TestRepoJobsRunOneJobPerRepoAndOneRerunForKicksDuringIt(t *testing.T) {
var jobs repoJobs
var log jobLog
gate := make(chan struct{})
stuck, other := syntax.DID("did:plc:stuck"), syntax.DID("did:plc:other")
runWorkers(t, &jobs, 4, func(_ context.Context, did syntax.DID) {
defer log.ended(did)
if log.started(did) == 1 && did == stuck {
<-gate
}
})
jobs.kick(stuck)
eventually(t, 2*time.Second, "the stuck repo's job to start", func() bool { return log.count(stuck) == 1 })
jobs.kick(other)
eventually(t, 2*time.Second, "another repo to run while the first is stuck", func() bool { return log.count(other) == 1 })
for range 5 {
jobs.kick(stuck)
}
time.Sleep(50 * time.Millisecond)
if got := log.count(stuck); got != 1 {
t.Fatalf("the stuck repo ran %d times while its job was still running", got)
}
close(gate)
eventually(t, 2*time.Second, "the kicks during the job to cause a rerun", func() bool { return log.count(stuck) == 2 })
time.Sleep(100 * time.Millisecond)
if got := log.count(stuck); got != 2 {
t.Fatalf("five kicks during one job ran it %d times, want one rerun", got)
}
if log.overlap {
t.Fatal("two jobs for one repo ran at once")
}
}
func markEvent(did syntax.DID, spindle string, owner syntax.DID) tapc.Event {
manifest := org_tangled.RepoManifest{
Spindle: ptr(spindle),
Declaration: &org_tangled.RepoManifest_Declaration{Owner: owner.String(), Name: "repo"},
}
return recordEvent(did, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, manifest)
}
func TestAnEventUnderMarkOnlyCostsNoReadsAndLeavesAMarkForTheJob(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo", testBob)
ctx := withMarkOnly(context.Background())
if err := env.s.processEvent(ctx, markEvent(testRepoDid, env.spindleDid(), testOwner)); err != nil {
t.Fatal(err)
}
if got := env.knot.count(); got != 0 {
t.Fatalf("marking a repo cost %d knot requests", got)
}
if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 0 {
t.Fatalf("mark = attempts %d, exists %v, %v, want a fresh row", attempts, marked, err)
}
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("the repo was served before its job ran, rows = %+v", rows)
}
env.s.runRepoJob(context.Background(), testRepoDid)
if rows := env.rows(t); len(rows) != 1 || !env.collaborator(t, testBob) {
t.Fatalf("the job didn't serve the repo, rows = %+v", rows)
}
requireMarked(t, env.s.db, testRepoDid, false, "a finished job left its mark")
before := env.knot.count()
env.knot.erase(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String())
collaborator := recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordDeleteAction, nil)
if err := env.s.processEvent(ctx, collaborator); err != nil {
t.Fatal(err)
}
if got := env.knot.count(); got != before {
t.Fatalf("a collaborator event cost %d knot requests under mark-only", got-before)
}
requireMarked(t, env.s.db, testRepoDid, true, "the collaborator event left no mark")
}
func TestAFailedJobKeepsItsMarkAndBacksOff(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
if err := env.s.markRepo(testRepoDid); err != nil {
t.Fatal(err)
}
env.knot.failing = true
env.s.runRepoJob(context.Background(), testRepoDid)
if attempts, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || !marked || attempts != 1 {
t.Fatalf("after a failed job: attempts %d, exists %v, %v, want one failure on the row", attempts, marked, err)
}
if due := pendingRetriesAt(t, env, time.Now()); len(due) != 0 {
t.Fatalf("a failed job is due again at once: %v", due)
}
}
func pendingRetriesAt(t *testing.T, env *ensureEnv, at time.Time) []syntax.DID {
t.Helper()
due, err := env.s.db.DueRepoRetries(at)
if err != nil {
t.Fatal(err)
}
return due
}
func TestAJobForADidWithNoKnotDropsItsMark(t *testing.T) {
env, did := noKnotEnv(t)
ctx := withMarkOnly(context.Background())
if err := env.s.processEvent(ctx, markEvent(did, env.spindleDid(), testOwner)); err != nil {
t.Fatal(err)
}
requireMarked(t, env.s.db, did, true, "setup: no mark")
env.s.runRepoJob(context.Background(), did)
requireMarked(t, env.s.db, did, false, "a did with no knot kept its mark")
}
func TestRetriesAfterOneOutageDontAllFallDueInTheSameSecond(t *testing.T) {
d, _ := newTestSpindleDB(t)
now := time.Unix(100000, 0)
r := &repoRetries{now: func() time.Time { return now }}
for i := range 60 {
r.fail(d, syntax.DID("did:plc:out"+string(rune('a'+i%26))+string(rune('a'+i/26))))
}
rows, err := d.Query(`select due from repo_retries`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
seconds := map[int64]bool{}
for rows.Next() {
var due int64
if err := rows.Scan(&due); err != nil {
t.Fatal(err)
}
if lo, hi := now.Add(repoRetryBase*3/4).Unix(), now.Add(repoRetryBase).Unix(); due < lo || due > hi {
t.Fatalf("due %d is outside the jitter window %d to %d", due, lo, hi)
}
seconds[due] = true
}
if len(seconds) < 4 {
t.Fatalf("60 retries share %d due seconds, want them spread out", len(seconds))
}
}
func TestAPanickingRepoJobLeavesItsWorkerRunning(t *testing.T) {
var jobs repoJobs
var log jobLog
poison, fine := syntax.DID("did:plc:poison"), syntax.DID("did:plc:fine")
runWorkers(t, &jobs, 1, func(_ context.Context, did syntax.DID) {
log.started(did)
if did == poison {
panic("boom")
}
})
jobs.kick(poison)
jobs.kick(fine)
eventually(t, 5*time.Second, "the only worker to run the next repo after the panic", func() bool { return log.count(fine) == 1 })
jobs.kick(poison)
eventually(t, 5*time.Second, "the poisoned repo to be kickable again", func() bool { return log.count(poison) == 2 })
}
func TestTheTickKicksEveryRepoWithWorkDueOnceAndSkipsOnesBackingOff(t *testing.T) {
env := newEnsureEnv(t)
now := time.Now()
withMarks, backingOff := syntax.DID("did:plc:withmarks"), syntax.DID("did:plc:backingoff")
for _, did := range []syntax.DID{withMarks, backingOff} {
for _, rkey := range []string{"k1", "k2"} {
r := db.RecordRetry{RepoDid: did, Collection: "org.tangled.track.ticket", Rkey: rkey}
if err := env.s.db.MarkRecordDirty(r, now, 100); err != nil {
t.Fatal(err)
}
}
}
if err := env.s.db.PutRepoRetry(backingOff, 1, now.Add(time.Hour)); err != nil {
t.Fatal(err)
}
env.s.kickDueRepos()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
did, ok := env.s.retries.jobs.next(ctx)
if !ok || did != withMarks {
t.Fatalf("first kicked repo = %q, %v, want %s", did, ok, withMarks)
}
short, stop := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer stop()
if did, ok := env.s.retries.jobs.next(short); ok {
t.Fatalf("a second repo %q was kicked, want only the one with marks that isn't backing off", did)
}
}