package spindle
import (
"context"
"errors"
"log/slog"
"slices"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/spindle/db"
)
const (
pendingOwner = syntax.DID("did:plc:owner")
pendingRepo = syntax.DID("did:plc:repo")
)
func catchUpBackfill(t *testing.T, f *fakeHydrant) *backfill {
t.Helper()
b, _ := newTestBackfill(t, f)
b.s.l = slog.Default()
b.refresh = func(context.Context, syntax.DID) error { return nil }
if err := b.s.db.AddRepo(db.Repo{Knot: "knot.test", Owner: pendingOwner, Rkey: "repo", RepoDid: pendingRepo}); err != nil {
t.Fatal(err)
}
return b
}
func pendingDids(t *testing.T, b *backfill) []syntax.DID {
t.Helper()
dids, err := b.s.db.PendingCatchUp()
if err != nil {
t.Fatal(err)
}
return dids
}
// the pending list lives in the spindle db, so the restart has to reuse it
func restartedBackfill(t *testing.T, old *backfill, f *fakeHydrant) *backfill {
t.Helper()
b := newBackfill(old.s, slog.Default(), shortTempDir(t))
b.start = f.start
b.poll = old.poll
b.retryBase = old.retryBase
b.handle = old.handle
b.refresh = old.refresh
return b
}
func TestCatchUpSavesItsDidsBeforeAnyRoundRuns(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
if got, want := pendingDids(t, b), []syntax.DID{pendingOwner, pendingRepo}; !slices.Equal(got, want) {
t.Fatalf("pending = %v, want %v", got, want)
}
}
func failOn(f *fakeHydrant, failing syntax.DID) {
f.onTrack = func(did syntax.DID) {
if did == failing {
f.setRepo(failedWith(did, "transport", 0))
} else {
f.setRepo(synced(did))
}
}
}
func TestPendingCatchUpKeepsADidHydrantCouldNotBackfill(t *testing.T) {
f := &fakeHydrant{}
failOn(f, pendingOwner)
b := catchUpBackfill(t, f)
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
runRound(t, b)
if got, want := pendingDids(t, b), []syntax.DID{pendingOwner}; !slices.Equal(got, want) {
t.Fatalf("pending = %v, want only the failed did %v", got, want)
}
}
// after a restart the pending list is the only thing that knows these dids are owed
func TestPendingCatchUpSurvivesARestartMidCatchUp(t *testing.T) {
interrupted := &fakeHydrant{}
b := catchUpBackfill(t, interrupted)
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
interrupted.failNext = errors.New("db is locked")
if err := b.round(context.Background()); err == nil {
t.Fatal("round succeeded although hydrant didn't start")
}
if len(pendingDids(t, b)) != 2 {
t.Fatalf("pending = %v, want both dids still owed", pendingDids(t, b))
}
f := &fakeHydrant{}
b2 := restartedBackfill(t, b, f)
if err := b2.Resume(); err != nil {
t.Fatal(err)
}
runRound(t, b2)
if want := []syntax.DID{pendingOwner, pendingRepo}; !slices.Equal(slices.Sorted(slices.Values(f.erased)), want) {
t.Fatalf("erased = %v, want the resumed catch-up to erase %v", f.erased, want)
}
if got := pendingDids(t, b2); len(got) != 0 {
t.Fatalf("pending = %v after the resumed catch-up finished", got)
}
}
func TestResumeWithNothingPendingStartsNoRound(t *testing.T) {
f := &fakeHydrant{}
b := catchUpBackfill(t, f)
if err := b.Resume(); err != nil {
t.Fatal(err)
}
runRound(t, b)
if f.starts != 0 {
t.Fatalf("hydrant started %d times with nothing to catch up", f.starts)
}
}
func TestResumeDropsDidsThatNeverBackfilledInAWeek(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
if err := b.s.db.AddPendingCatchUp([]syntax.DID{"did:plc:dead"}, time.Now().Add(-pendingCatchUpMaxAge-time.Hour)); err != nil {
t.Fatal(err)
}
if err := b.s.db.AddPendingCatchUp([]syntax.DID{"did:plc:slow"}, time.Now().Add(-time.Hour)); err != nil {
t.Fatal(err)
}
if err := b.Resume(); err != nil {
t.Fatal(err)
}
if got, want := pendingDids(t, b), []syntax.DID{"did:plc:slow"}; !slices.Equal(got, want) {
t.Fatalf("pending = %v, want %v", got, want)
}
add, _ := b.take()
if !slices.Equal(add, []syntax.DID{"did:plc:slow"}) {
t.Fatalf("queued %v, want only the did that is still pending", add)
}
}
func TestUntrackForgetsAPendingCatchUp(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
b.Untrack(pendingOwner)
if got, want := pendingDids(t, b), []syntax.DID{pendingRepo}; !slices.Equal(got, want) {
t.Fatalf("pending = %v, want %v", got, want)
}
}
// outcomes only count for dids in the round list, even a done one
func TestSettlePendingOnlyTouchesDidsTheRoundBackfilled(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
if err := b.s.db.AddPendingCatchUp([]syntax.DID{"did:plc:a", "did:plc:b", "did:plc:c"}, time.Now()); err != nil {
t.Fatal(err)
}
b.settlePending(
context.Background(),
[]syntax.DID{"did:plc:a", "did:plc:c"},
map[syntax.DID]outcome{"did:plc:a": done, "did:plc:b": done, "did:plc:c": failed},
)
if got, want := pendingDids(t, b), []syntax.DID{"did:plc:b", "did:plc:c"}; !slices.Equal(got, want) {
t.Fatalf("pending = %v, want %v", got, want)
}
}
func recordRefreshes(b *backfill, fail error) *[]syntax.DID {
var refreshed []syntax.DID
b.refresh = func(_ context.Context, did syntax.DID) error {
refreshed = append(refreshed, did)
return fail
}
return &refreshed
}
// hydrant never sends deleted manifests, so only the knot can tell us a repo died in the gap
func TestCatchUpRefreshesServedReposButNotOwners(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
refreshed := recordRefreshes(b, nil)
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
runRound(t, b)
if !slices.Equal(*refreshed, []syntax.DID{pendingRepo}) {
t.Fatalf("refreshed = %v, want only the served repo did", *refreshed)
}
if got := pendingDids(t, b); len(got) != 0 {
t.Fatalf("pending = %v", got)
}
}
func TestCatchUpRefreshFailureGoesToTheRetrySetAndClearsThePending(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
recordRefreshes(b, errors.New("knot is down"))
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
runRound(t, b)
if got := pendingRetries(t, b.s.db); !slices.Equal(got, []syntax.DID{pendingRepo}) {
t.Fatalf("retry set = %v, want the repo whose refresh failed", got)
}
if got := pendingDids(t, b); len(got) != 0 {
t.Fatalf("pending = %v, the retry set owns it now", got)
}
}
func TestCancelledRefreshKeepsTheDidPending(t *testing.T) {
b := catchUpBackfill(t, &fakeHydrant{})
ctx, cancel := context.WithCancel(context.Background())
b.refresh = func(context.Context, syntax.DID) error {
cancel()
return ctx.Err()
}
if err := b.s.db.AddPendingCatchUp([]syntax.DID{pendingRepo}, time.Now()); err != nil {
t.Fatal(err)
}
b.settlePending(ctx, []syntax.DID{pendingRepo}, map[syntax.DID]outcome{pendingRepo: done})
if got := pendingDids(t, b); !slices.Equal(got, []syntax.DID{pendingRepo}) {
t.Fatalf("pending = %v, want the did kept for the next start", got)
}
if got := pendingRetries(t, b.s.db); len(got) != 0 {
t.Fatalf("retry set = %v, a shutdown isn't a refresh failure", got)
}
}
func TestCatchUpDoesNotRefreshARepoHydrantCouldNotBackfill(t *testing.T) {
f := &fakeHydrant{}
failOn(f, pendingRepo)
b := catchUpBackfill(t, f)
refreshed := recordRefreshes(b, nil)
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
runRound(t, b)
if len(*refreshed) != 0 {
t.Fatalf("refreshed %v although hydrant failed it, it has to stay pending for the next round", *refreshed)
}
if got := pendingDids(t, b); !slices.Equal(got, []syntax.DID{pendingRepo}) {
t.Fatalf("pending = %v", got)
}
}