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) } }