package spindle
import (
"context"
"testing"
"time"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/tapc"
)
func ticketEvent(rkey string) tapc.Event {
return recordEvent(testRepoDid, org_tangled.TrackTicketNSID, rkey, tapc.RecordCreateAction, map[string]string{})
}
func dueRecords(t *testing.T, env *ensureEnv) []db.RecordRetry {
t.Helper()
due, err := env.s.db.DueRecordRetries(time.Now())
if err != nil {
t.Fatal(err)
}
return due
}
func TestAPullRecordUnderMarkOnlyCostsNoReadsAndWaitsForItsJob(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
ctx := withMarkOnly(context.Background())
before := env.knot.count()
if err := env.s.processEvent(ctx, ticketEvent("3kticket00001")); err != nil {
t.Fatal(err)
}
if got := env.knot.count(); got != before {
t.Fatalf("marking a pull record cost %d knot requests", got-before)
}
if got := dueRecords(t, env); len(got) != 1 || got[0].Rkey != "3kticket00001" {
t.Fatalf("record marks = %+v, want the ticket", got)
}
env.s.runRepoJob(context.Background(), testRepoDid)
if env.knot.count() == before {
t.Fatal("the job never read the ticket from the knot")
}
if got := dueRecords(t, env); len(got) != 0 {
t.Fatalf("a finished job left record marks %+v", got)
}
}
func TestAPullRecordWaitsForItsReposPendingRefresh(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
ctx := withMarkOnly(context.Background())
if err := env.s.processEvent(ctx, markEvent(testRepoDid, env.spindleDid(), testOwner)); err != nil {
t.Fatal(err)
}
if err := env.s.processEvent(ctx, ticketEvent("3kticket00001")); err != nil {
t.Fatal(err)
}
if got := dueRecords(t, env); len(got) != 1 {
t.Fatalf("a record for a repo that is about to be served was dropped, marks = %+v", got)
}
env.s.runRepoJob(context.Background(), testRepoDid)
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("the job didn't serve the repo, rows = %+v", rows)
}
if got := dueRecords(t, env); len(got) != 0 {
t.Fatalf("the record mark outlived the repo's job: %+v", got)
}
}
func TestAPullRecordForAnUnknownRepoLeavesNoMark(t *testing.T) {
env := newEnsureEnv(t)
if err := env.s.processEvent(withMarkOnly(context.Background()), ticketEvent("3kticket00001")); err != nil {
t.Fatal(err)
}
if got := dueRecords(t, env); len(got) != 0 {
t.Fatalf("an unknown repo's record left marks %+v", got)
}
}
func TestAFailingRecordIsBackedOffInsteadOfSpinningTheJob(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
rec := ticketEvent("3kticket00001").Record
if err := env.s.markRecord(rec); err != nil {
t.Fatal(err)
}
env.knot.failing = true
env.s.runRecordMarks(context.Background(), testRepoDid)
r := db.RecordRetry{RepoDid: testRepoDid, Collection: org_tangled.TrackTicketNSID, Rkey: "3kticket00001"}
if attempts, err := env.s.db.RecordRetryAttempts(r); err != nil || attempts != 1 {
t.Fatalf("attempts = %d, %v, want one failure on the mark", attempts, err)
}
if got := dueRecords(t, env); len(got) != 0 {
t.Fatalf("a failed record is due again at once: %+v", got)
}
}
func TestAFloodOfRecordsForAnotherRepoLeavesAServedReposMarkAlone(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
if _, err := env.s.db.Exec(`insert into record_retries (repo_did, collection, rkey, attempts, due, epoch)
with recursive n(i) as (select 0 union all select i + 1 from n where i < 30000)
select 'did:plc:flood' || (i % 40), 'c', 'k' || i, 0, 0, 1 from n`); err != nil {
t.Fatal(err)
}
if err := env.s.processEvent(withMarkOnly(context.Background()), ticketEvent("3kticket00001")); err != nil {
t.Fatal(err)
}
if marks, err := env.s.db.DueRecordMarksOf(testRepoDid, time.Now()); err != nil || len(marks) != 1 {
t.Fatalf("marks = %+v, %v, want the served repo's event kept while other repos fill the table", marks, err)
}
}