package spindle
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"slices"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/gorilla/websocket"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/tapc"
)
// a short dir, because unix socket paths are capped at ~104 bytes on darwin
// and t.TempDir() alone gets close to that
func shortTempDir(t *testing.T) string {
t.Helper()
dir, err := os.MkdirTemp("/tmp", "bf-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// fakeHydrant serves the parts of hydrant's api the backfill uses
type fakeHydrant struct {
mu sync.Mutex
events []tapc.Event
puts [][]syntax.DID
erased []syntax.DID
tracked map[syntax.DID]bool
repos map[syntax.DID]hydrantRepo
resynced []syntax.DID
onResync func(did syntax.DID)
onTrack func(did syntax.DID)
// owners hydrant's retry worker requeues every other time spindle looks
flapping map[syntax.DID]bool
listings int
cursors []string
starts int
failNext error
// every start fails, not just the next
failStarts error
ops []string
srv *httptest.Server
done chan struct{}
shutdowns atomic.Int32
}
func (f *fakeHydrant) subscriptions() int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.cursors)
}
func (f *fakeHydrant) Done() <-chan struct{} { return f.done }
func (f *fakeHydrant) Err() error { return nil }
func (f *fakeHydrant) Shutdown(time.Duration) error { f.srv.Close(); f.shutdowns.Add(1); return nil }
func (f *fakeHydrant) addRecord(id int64, did, rkey string) {
f.mu.Lock()
defer f.mu.Unlock()
f.events = append(f.events, tapc.Event{
ID: id,
Type: tapc.EvtRecord,
Record: &tapc.RecordEventData{
Did: syntax.DID(did),
Collection: syntax.NSID(org_tangled.RepoDeclarationNSID),
Rkey: syntax.RecordKey(rkey),
Action: tapc.RecordCreateAction,
Record: json.RawMessage(`{}`),
},
})
}
func (f *fakeHydrant) addEvent(rec tapc.RecordEventData) {
f.mu.Lock()
defer f.mu.Unlock()
f.events = append(f.events, tapc.Event{ID: int64(len(f.events)), Type: tapc.EvtRecord, Record: &rec})
}
func (f *fakeHydrant) erasedDids() []syntax.DID {
f.mu.Lock()
defer f.mu.Unlock()
return slices.Clone(f.erased)
}
// empty for an owner hydrant doesn't know
func (f *fakeHydrant) repoStatus(did syntax.DID) string {
f.mu.Lock()
defer f.mu.Unlock()
return f.repos[did].Status
}
func (f *fakeHydrant) failStartsWith(err error) {
f.mu.Lock()
defer f.mu.Unlock()
f.failStarts = err
}
// identity events aren't replayed, so only live connections get this one
func (f *fakeHydrant) addLiveOnly(id int64, did string) {
f.mu.Lock()
defer f.mu.Unlock()
f.events = append(f.events, tapc.Event{
ID: id,
Type: tapc.EvtIdentity,
Identity: &tapc.IdentityEventData{DID: syntax.DID(did), IsActive: true},
})
}
// setRepo is what hydrant reports for repo.Did, which it then already
// tracks. owners nobody set are synced
func (f *fakeHydrant) setRepo(repo hydrantRepo) {
f.mu.Lock()
defer f.mu.Unlock()
if f.repos == nil {
f.repos = make(map[syntax.DID]hydrantRepo)
f.tracked = make(map[syntax.DID]bool)
}
f.repos[repo.Did] = repo
f.tracked[repo.Did] = true
}
func synced(did syntax.DID) hydrantRepo { return hydrantRepo{Did: did, Status: "synced"} }
// retrying is a failed owner hydrant queued again
func retrying(did syntax.DID) hydrantRepo {
return hydrantRepo{Did: did, Status: "error(transport error)", Pending: true}
}
func failedWith(did syntax.DID, kind string, hostFailures int) hydrantRepo {
return hydrantRepo{
Did: did,
Status: "error(" + kind + ")",
Resync: &hydrantResync{State: "error", Kind: kind, HostFailures: hostFailures},
}
}
func (f *fakeHydrant) start(settings map[string]string) (embeddedHydrant, error) {
f.mu.Lock()
f.starts++
err := f.failNext
f.failNext = nil
if err == nil {
err = f.failStarts
}
f.mu.Unlock()
if err != nil {
return nil, err
}
l, err := net.Listen("unix", strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:"))
if err != nil {
return nil, err
}
f.srv = httptest.NewUnstartedServer(http.HandlerFunc(f.serve))
f.srv.Listener = l
f.srv.Start()
f.done = make(chan struct{})
return f, nil
}
func (f *fakeHydrant) serve(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/stream/head":
f.mu.Lock()
var head *int64
for _, evt := range f.events {
if evt.Type == tapc.EvtRecord {
head = &evt.ID
}
}
f.mu.Unlock()
json.NewEncoder(w).Encode(map[string]any{"id": head})
case strings.HasPrefix(r.URL.Path, "/redactions/"):
did := syntax.DID(strings.TrimPrefix(r.URL.Path, "/redactions/"))
f.mu.Lock()
f.erased = append(f.erased, did)
f.ops = append(f.ops, "erase "+did.String())
delete(f.tracked, did)
delete(f.repos, did)
// the event stays so its id still counts, only the body goes
for i, evt := range f.events {
if evt.Record != nil && evt.Record.Did == did {
f.events[i].Record = nil
}
}
f.mu.Unlock()
w.Write([]byte("{}"))
case r.URL.Path == "/repos" && r.Method == http.MethodGet:
f.mu.Lock()
var repos []hydrantRepo
f.listings++
for did := range f.tracked {
repo, ok := f.repos[did]
if !ok {
repo = hydrantRepo{Did: did, Status: "synced"}
}
if f.flapping[did] && f.listings%2 == 0 {
repo.Resync, repo.Pending = nil, true
}
if r.URL.Query().Get("queued") == "true" && !repo.Pending && repo.Resync == nil {
continue
}
repo.Tracked = true
repos = append(repos, repo)
}
f.mu.Unlock()
slices.SortFunc(repos, func(a, b hydrantRepo) int { return strings.Compare(string(a.Did), string(b.Did)) })
if cursor := syntax.DID(r.URL.Query().Get("cursor")); cursor != "" {
repos = slices.DeleteFunc(repos, func(repo hydrantRepo) bool { return repo.Did <= cursor })
}
if limit, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && len(repos) > limit {
repos = repos[:limit]
}
json.NewEncoder(w).Encode(repos)
case r.URL.Path == "/repos/resync":
var body []struct {
Did syntax.DID `json:"did"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
f.mu.Lock()
onResync := f.onResync
for _, b := range body {
f.resynced = append(f.resynced, b.Did)
}
f.mu.Unlock()
for _, b := range body {
if onResync != nil {
onResync(b.Did)
}
}
w.Write([]byte("[]"))
case r.URL.Path == "/repos":
var body []struct {
Did syntax.DID `json:"did"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
var dids []syntax.DID
for _, b := range body {
dids = append(dids, b.Did)
}
slices.Sort(dids)
f.mu.Lock()
if f.tracked == nil {
f.tracked = make(map[syntax.DID]bool)
}
f.puts = append(f.puts, dids)
f.ops = append(f.ops, fmt.Sprintf("put %d", len(dids)))
for _, did := range dids {
f.tracked[did] = true
}
onTrack := f.onTrack
f.mu.Unlock()
for _, did := range dids {
if onTrack != nil {
onTrack(did)
}
}
w.Write([]byte("[]"))
case r.URL.Path == "/stream":
f.stream(w, r)
default:
http.NotFound(w, r)
}
}
func (f *fakeHydrant) stream(w http.ResponseWriter, r *http.Request) {
conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
if err != nil {
return
}
defer conn.Close()
next, _ := strconv.ParseInt(r.URL.Query().Get("cursor"), 10, 64)
f.mu.Lock()
f.cursors = append(f.cursors, r.URL.Query().Get("cursor"))
connectedAt := len(f.events)
f.mu.Unlock()
gone := closedWhenDone(conn)
for {
f.mu.Lock()
var events []tapc.Event
for i, evt := range f.events {
if evt.ID >= next && (evt.Type == tapc.EvtRecord || i >= connectedAt) {
events = append(events, evt)
}
}
f.mu.Unlock()
for _, evt := range events {
if err := conn.WriteJSON(evt); err != nil {
return
}
next = evt.ID + 1
}
select {
case <-gone:
return
case <-time.After(5 * time.Millisecond):
}
}
}
// the server side of a websocket only learns the client left by reading
func closedWhenDone(conn *websocket.Conn) <-chan struct{} {
gone := make(chan struct{})
go func() {
defer close(gone)
for {
if _, _, err := conn.ReadMessage(); err != nil {
return
}
}
}()
return gone
}
func newTestBackfill(t *testing.T, f *fakeHydrant) (*backfill, *[]int64) {
t.Helper()
d, _ := newTestSpindleDB(t)
b := newBackfill(&Spindle{cfg: &config.Config{}, db: d}, slog.Default(), shortTempDir(t))
b.start = f.start
b.poll = 10 * time.Millisecond
b.retryBase = time.Millisecond
var mu sync.Mutex
handled := []int64{}
b.handle = func(_ context.Context, evt tapc.Event) error {
mu.Lock()
defer mu.Unlock()
handled = append(handled, evt.ID)
return nil
}
return b, &handled
}
func runRound(t *testing.T, b *backfill) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := b.round(ctx); err != nil {
t.Fatalf("round: %v", err)
}
}
func TestBackfillRoundFeedsTheStreamThenShutsDown(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
f.addRecord(1, "did:plc:a", "two")
f.addRecord(2, "did:plc:b", "three")
b, handled := newTestBackfill(t, f)
b.Track("did:plc:b", "did:plc:a")
runRound(t, b)
if !slices.Equal(*handled, []int64{0, 1, 2}) {
t.Fatalf("handled %v, want [0 1 2]", *handled)
}
if len(f.puts) != 1 || !slices.Equal(f.puts[0], []syntax.DID{"did:plc:a", "did:plc:b"}) {
t.Fatalf("puts = %v, want one put of both owners", f.puts)
}
if f.shutdowns.Load() != 1 {
t.Fatalf("shutdowns = %d, want 1", f.shutdowns.Load())
}
if last, err := b.loadCursor(); err != nil || last != 2 {
t.Fatalf("saved cursor = %d, %v, want 2", last, err)
}
// a known owner doesn't start hydrant again, and a new one resumes after the cursor
b.Track("did:plc:a")
runRound(t, b)
if f.starts != 1 {
t.Fatalf("starts = %d after tracking a known owner, want 1", f.starts)
}
f.addRecord(3, "did:plc:c", "four")
b.Track("did:plc:c")
runRound(t, b)
if !slices.Equal(*handled, []int64{0, 1, 2, 3}) {
t.Fatalf("handled %v, want [0 1 2 3]", *handled)
}
if !slices.Equal(f.cursors, []string{"0", "3"}) {
t.Fatalf("stream cursors = %v, want [0 3]", f.cursors)
}
}
// untracking only remembers the wish, so nothing happens until a round has
// something to add. a restart forgets it, and hydrant's copy goes then
func TestBackfillUntrackIsLazy(t *testing.T) {
f := &fakeHydrant{}
b, _ := newTestBackfill(t, f)
b.Track("did:plc:a", "did:plc:b")
runRound(t, b)
b.Untrack("did:plc:b")
runRound(t, b)
if f.starts != 1 {
t.Fatalf("starts = %d, untracking alone must not start hydrant", f.starts)
}
b.Track("did:plc:c")
runRound(t, b)
if !slices.Equal(f.erased, []syntax.DID{"did:plc:b"}) {
t.Fatalf("erased = %v, want [did:plc:b]", f.erased)
}
if !slices.Equal(f.puts[len(f.puts)-1], []syntax.DID{"did:plc:c"}) {
t.Fatalf("last put = %v, want [did:plc:c]", f.puts[len(f.puts)-1])
}
// nothing followed a while it was untracked, so tracking it again has to
// delete it first, or hydrant would call it backfilled and skip it
b.Untrack("did:plc:a")
b.Track("did:plc:a")
add, remove := b.take()
if !slices.Equal(add, []syntax.DID{"did:plc:a"}) || !slices.Equal(remove, []syntax.DID{"did:plc:a"}) {
t.Fatalf("take = %v, %v, want did:plc:a in both", add, remove)
}
// a new process only knows who it wants, so c goes
b, _ = newTestBackfill(t, f)
b.Track("did:plc:a")
runRound(t, b)
if !slices.Equal(f.erased, []syntax.DID{"did:plc:b", "did:plc:c"}) {
t.Fatalf("erased = %v, want b and c", f.erased)
}
}
func TestBackfillWaitsForOwnersItCanStillGet(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
// a first backfill still running, and a live pds asking hydrant to slow down
f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
f.setRepo(failedWith("did:plc:b", "ratelimited", 0))
b, _ := newTestBackfill(t, f)
b.Track("did:plc:a", "did:plc:b")
go func() {
time.Sleep(200 * time.Millisecond)
f.setRepo(synced("did:plc:a"))
f.setRepo(synced("did:plc:b"))
}()
started := time.Now()
runRound(t, b)
if time.Since(started) < 200*time.Millisecond {
t.Fatalf("round ended after %v while owners could still be backfilled", time.Since(started))
}
if len(b.failing) != 0 {
t.Fatalf("failing = %v, want none", b.failing)
}
}
// hydrant keeps moving owners it failed back into its queue on its own
// backoff, so pending hardly ever reads 0. none of that can gain anything this
// round, so it mustn't keep the round going
func TestBackfillEndsWhileFailedOwnersKeepRetrying(t *testing.T) {
f := &fakeHydrant{flapping: map[syntax.DID]bool{"did:plc:b": true}}
f.addRecord(0, "did:plc:a", "one")
f.setRepo(failedWith("did:plc:b", "generic", 0))
// ratelimited only because its pds is down and hydrant skips it
f.setRepo(failedWith("did:plc:c", "ratelimited", 4))
f.setRepo(hydrantRepo{Did: "did:plc:d", Status: "takendown", Resync: &hydrantResync{State: "gone"}})
// hydrant dropped it from its queues without syncing it
f.setRepo(hydrantRepo{Did: "did:plc:e", Status: "desynchronized"})
b, handled := newTestBackfill(t, f)
b.Track("did:plc:a", "did:plc:b", "did:plc:c", "did:plc:d", "did:plc:e")
runRound(t, b)
if !slices.Equal(*handled, []int64{0}) {
t.Fatalf("handled %v, want [0]", *handled)
}
// spindle didn't know they failed before, so they got retried straight away
if !slices.Equal(f.resynced, []syntax.DID{"did:plc:b", "did:plc:c", "did:plc:e"}) {
t.Fatalf("resynced = %v, want b, c and e", f.resynced)
}
if len(b.failing) != 3 || b.failing["did:plc:b"].rounds != 1 || b.failing["did:plc:e"].rounds != 1 {
t.Fatalf("failing = %v, want b, c and e after one round", b.failing)
}
if wait := time.Until(b.nextRetry()); wait < time.Minute || wait > failedOwnerRetryMin {
t.Fatalf("next retry in %v, want about %v", wait, failedOwnerRetryMin)
}
}
// the first round leaves did:plc:a failed, with its retry a backoff away
func failedOnce(t *testing.T) (*fakeHydrant, *backfill) {
t.Helper()
f := &fakeHydrant{}
f.setRepo(failedWith("did:plc:a", "transport", 0))
b, _ := newTestBackfill(t, f)
b.Track("did:plc:a", "did:plc:b")
runRound(t, b)
return f, b
}
// stands in for waiting out the backoff
func retryNow(b *backfill) {
for did, f := range b.failing {
f.retryAt = time.Now()
b.failing[did] = f
}
}
func TestBackfillRetriesAFailedOwnerAfterABackoff(t *testing.T) {
f, b := failedOnce(t)
if wait := time.Until(b.nextRetry()); wait < time.Minute || wait > failedOwnerRetryMin {
t.Fatalf("next retry in %v, want about %v", wait, failedOwnerRetryMin)
}
runRound(t, b)
if f.starts != 1 {
t.Fatalf("starts = %d before the retry was due, want 1", f.starts)
}
retryNow(b)
before := len(f.resynced)
f.onResync = func(did syntax.DID) { f.setRepo(synced(did)) }
runRound(t, b)
if got := f.resynced[before:]; !slices.Equal(got, []syntax.DID{"did:plc:a"}) {
t.Fatalf("resynced %v in the retry round, want did:plc:a", got)
}
}
func TestBackfillBacksOffFurtherWhenTheRetryFails(t *testing.T) {
f, b := failedOnce(t)
retryNow(b)
// hydrant shows the retry queued first, so the old failure can't count
// as the new outcome
f.onResync = func(did syntax.DID) {
f.setRepo(retrying(did))
go func() {
time.Sleep(50 * time.Millisecond)
f.setRepo(failedWith(did, "transport", 0))
}()
}
started := time.Now()
runRound(t, b)
if time.Since(started) < 50*time.Millisecond {
t.Fatalf("round ended after %v before the retry finished", time.Since(started))
}
if wait := time.Until(b.nextRetry()); wait <= failedOwnerRetryMin {
t.Fatalf("next retry in %v, want it backed off past %v", wait, failedOwnerRetryMin)
}
}
func TestBackfillForgetsAFailedOwnerOnceItBackfills(t *testing.T) {
f, b := failedOnce(t)
retryNow(b)
f.onResync = func(did syntax.DID) { f.setRepo(synced(did)) }
runRound(t, b)
if !b.nextRetry().IsZero() {
t.Fatalf("next retry at %v after the owner backfilled, want none", b.nextRetry())
}
}
// with no new owners coming, run itself has to come back for a failed one
func TestBackfillRunComesBackForFailedOwners(t *testing.T) {
f := &fakeHydrant{}
f.setRepo(failedWith("did:plc:a", "transport", 0))
f.onResync = func(did syntax.DID) {
f.mu.Lock()
retries := len(f.resynced)
f.mu.Unlock()
// the first round's retry fails again, and the next one gets through
if retries > 1 {
f.setRepo(synced(did))
}
}
b, _ := newTestBackfill(t, f)
b.retryFailed = 50 * time.Millisecond
b.Track("did:plc:a")
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
done := make(chan struct{})
go func() {
defer close(done)
b.run(ctx)
}()
for {
f.mu.Lock()
retries := len(f.resynced)
f.mu.Unlock()
if retries >= 2 {
break
}
if ctx.Err() != nil {
t.Fatal("the failed owner was never retried")
}
time.Sleep(10 * time.Millisecond)
}
cancel()
<-done
if f.starts < 2 {
t.Fatalf("starts = %d, want a retry round", f.starts)
}
}
func TestBackfillParksARecordThatKeepsFailing(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
f.addRecord(1, "did:plc:a", "bad")
f.addRecord(2, "did:plc:a", "three")
b, _ := newTestBackfill(t, f)
attempts := map[int64]int{}
b.handle = func(_ context.Context, evt tapc.Event) error {
attempts[evt.ID]++
if evt.ID == 1 {
return errors.New("database is locked")
}
return nil
}
b.Track("did:plc:a")
runRound(t, b)
want := map[int64]int{0: 1, 1: backfillEventAttempts, 2: 1}
if fmt.Sprint(attempts) != fmt.Sprint(want) {
t.Fatalf("attempts = %v, want %v", attempts, want)
}
if last, err := b.loadCursor(); err != nil || last != 2 {
t.Fatalf("saved cursor = %d, %v, want 2", last, err)
}
parked := dueParked(t, b)
if len(parked) != 1 || parked[0].URI != "at://did:plc:a/"+org_tangled.RepoDeclarationNSID+"/bad" || parked[0].Err != "database is locked" {
t.Fatalf("parked = %+v, want the bad record with its error", parked)
}
}
// looks a day ahead, so records not due yet come back too
func dueParked(t *testing.T, b *backfill) []db.ParkedRecord {
t.Helper()
parked, err := b.s.db.DueParkedRecords(time.Now().Add(24 * time.Hour))
if err != nil {
t.Fatal(err)
}
return parked
}
func parkFailing(t *testing.T, b *backfill, f *fakeHydrant) {
t.Helper()
f.addRecord(0, "did:plc:a", "bad")
handle := b.handle
b.handle = func(context.Context, tapc.Event) error { return errors.New("database is locked") }
b.Track("did:plc:a")
runRound(t, b)
b.handle = handle
if len(dueParked(t, b)) != 1 {
t.Fatal("nothing got parked")
}
}
func TestBackfillRetriesAParkedRecordUntilItGoesThrough(t *testing.T) {
f := &fakeHydrant{}
b, handled := newTestBackfill(t, f)
parkFailing(t, b, f)
b.retryParkedRecords(context.Background(), time.Now())
if len(*handled) != 0 {
t.Fatalf("handled %v before the retry was due", *handled)
}
fails := 2
handle := b.handle
b.handle = func(ctx context.Context, evt tapc.Event) error {
if fails > 0 {
fails--
return errors.New("still locked")
}
return handle(ctx, evt)
}
for range 3 {
b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
}
if !slices.Equal(*handled, []int64{0}) {
t.Fatalf("handled %v, want the parked record once it worked", *handled)
}
if parked := dueParked(t, b); len(parked) != 0 {
t.Fatalf("parked = %+v after it went through, want none", parked)
}
}
func TestBackfillGivesUpOnAParkedRecordAfterTheLastRetry(t *testing.T) {
f := &fakeHydrant{}
b, _ := newTestBackfill(t, f)
b.retryParked = time.Minute
parkFailing(t, b, f)
b.handle = func(context.Context, tapc.Event) error { return errors.New("broken for good") }
var waits []time.Duration
for range parkedRetries - 1 {
b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
parked := dueParked(t, b)
if len(parked) != 1 {
t.Fatalf("parked = %+v before the last retry", parked)
}
waits = append(waits, time.Until(parked[0].RetryAt).Round(time.Minute))
}
want := []time.Duration{2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 16 * time.Minute, 32 * time.Minute}
if !slices.Equal(waits, want) {
t.Fatalf("waits = %v, want %v", waits, want)
}
b.retryParkedRecords(context.Background(), time.Now().Add(24*time.Hour))
if parked := dueParked(t, b); len(parked) != 0 {
t.Fatalf("parked = %+v after the last retry, want it given up", parked)
}
}
func TestBackfillRetriesEveryParkedRecordAtStartup(t *testing.T) {
f := &fakeHydrant{}
b, handled := newTestBackfill(t, f)
b.retryParked = time.Hour
parkFailing(t, b, f)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
done := make(chan struct{})
go func() {
defer close(done)
b.run(ctx)
}()
for len(dueParked(t, b)) != 0 {
if ctx.Err() != nil {
t.Fatal("the parked record wasn't retried at startup")
}
time.Sleep(10 * time.Millisecond)
}
cancel()
<-done
if !slices.Equal(*handled, []int64{0}) {
t.Fatalf("handled %v, want the parked record", *handled)
}
}
func TestBackfillReplaysTheStreamOfANewDb(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
b, handled := newTestBackfill(t, f)
if err := b.saveCursor(100); err != nil {
t.Fatal(err)
}
b.Track("did:plc:a")
runRound(t, b)
if !slices.Equal(*handled, []int64{0}) {
t.Fatalf("handled %v, want the new db's event", *handled)
}
}
// head only counts records, so identity ids can't move the cursor
func TestBackfillCursorSkipsLiveOnlyEvents(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
b, handled := newTestBackfill(t, f)
b.Track("did:plc:a")
go func() {
for f.subscriptions() == 0 {
time.Sleep(5 * time.Millisecond)
}
f.addLiveOnly(1, "did:plc:a")
time.Sleep(50 * time.Millisecond)
f.setRepo(synced("did:plc:a"))
}()
runRound(t, b)
if last, err := b.loadCursor(); err != nil || last != 0 {
t.Fatalf("saved cursor = %d, %v, want the last record's id", last, err)
}
f.addRecord(2, "did:plc:b", "two")
b.Track("did:plc:b")
runRound(t, b)
if !slices.Equal(*handled, []int64{0, 2}) {
t.Fatalf("handled %v, want the new record without replaying the old one", *handled)
}
}
func TestBackfillTracksOwnersInBatches(t *testing.T) {
f := &fakeHydrant{}
b, _ := newTestBackfill(t, f)
var dids []syntax.DID
for i := range 2*hydrantDidBatch + 1 {
dids = append(dids, syntax.DID(fmt.Sprintf("did:plc:%024d", i)))
}
b.Track(dids...)
runRound(t, b)
var sizes []int
for _, put := range f.puts {
sizes = append(sizes, len(put))
}
if !slices.Equal(sizes, []int{hydrantDidBatch, hydrantDidBatch, 1}) {
t.Fatalf("put sizes = %v, want two full batches and one more", sizes)
}
h, err := f.start(b.settings())
if err != nil {
t.Fatal(err)
}
defer h.Shutdown(0)
repos, err := newHydrantAPI(b.socketPath()).trackedRepos(context.Background())
if err != nil {
t.Fatal(err)
}
if len(repos) != len(dids) {
t.Fatalf("listed %d owners over every page, want %d", len(repos), len(dids))
}
}
func TestBackfillTakesOwnersAddedDuringARound(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
f.setRepo(hydrantRepo{Did: "did:plc:a", Status: "desynchronized", Pending: true})
b, handled := newTestBackfill(t, f)
b.Track("did:plc:a")
go func() {
for f.subscriptions() == 0 {
time.Sleep(5 * time.Millisecond)
}
b.Track("did:plc:b")
time.Sleep(50 * time.Millisecond)
f.addRecord(1, "did:plc:b", "two")
f.setRepo(synced("did:plc:a"))
}()
runRound(t, b)
if !slices.Equal(*handled, []int64{0, 1}) {
t.Fatalf("handled %v, want the record of the owner that joined", *handled)
}
if f.starts != 1 {
t.Fatalf("starts = %d, want the owner taken by the running round", f.starts)
}
}
func TestBackfillErasesAnOwnerBeforeTrackingItAgain(t *testing.T) {
f := &fakeHydrant{}
f.addRecord(0, "did:plc:a", "one")
b, handled := newTestBackfill(t, f)
b.Track("did:plc:a")
runRound(t, b)
b.Untrack("did:plc:a")
b.Track("did:plc:a")
f.onTrack = func(did syntax.DID) { f.addRecord(1, did.String(), "one") }
// a fresh cursor replays the erased event, which has no body left
os.Remove(b.cursorPath())
runRound(t, b)
if !slices.Equal(f.ops, []string{"put 1", "erase did:plc:a", "put 1"}) {
t.Fatalf("ops = %v, want the erase before the second put", f.ops)
}
if !slices.Equal(*handled, []int64{0, 1}) {
t.Fatalf("handled %v, want the first round's record and the fresh one", *handled)
}
}
func TestBackfillCatchUpErasesAndTracksOwnersAndRepoDids(t *testing.T) {
f := &fakeHydrant{}
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: "did:plc:owner", Rkey: "repo", RepoDid: "did:plc:repo"}); err != nil {
t.Fatal(err)
}
b.Track("did:plc:owner")
runRound(t, b)
if err := b.CatchUp(); err != nil {
t.Fatal(err)
}
runRound(t, b)
want := []string{"put 1", "erase did:plc:owner", "erase did:plc:repo", "put 2"}
if !slices.Equal(f.ops, want) {
t.Fatalf("ops = %v, want %v", f.ops, want)
}
}
func TestHydrantReadinessNoticesHydrantExit(t *testing.T) {
socket := shortTempDir(t) + "/api.sock"
l, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() }))
srv.Listener = l
srv.Start()
defer srv.Close()
h := &fakeHydrant{done: make(chan struct{})}
close(h.done)
started := time.Now()
err = newHydrantAPI(socket).waitReady(context.Background(), h)
if err == nil || time.Since(started) > 10*time.Second {
t.Fatalf("waitReady = %v after %v, want it to notice hydrant exited", err, time.Since(started))
}
}
func TestBackfillRequeuesOwnersWhenHydrantFailsToStart(t *testing.T) {
f := &fakeHydrant{failNext: errors.New("db is locked")}
b, _ := newTestBackfill(t, f)
b.Track("did:plc:a")
if err := b.round(context.Background()); err == nil {
t.Fatal("round succeeded although hydrant didn't start")
}
runRound(t, b)
if len(f.puts) != 1 || !slices.Equal(f.puts[0], []syntax.DID{"did:plc:a"}) {
t.Fatalf("puts = %v, the owner should have been retried", f.puts)
}
}
func TestBackfillSettingsAllowPrivateNetworksOnlyInDev(t *testing.T) {
for _, dev := range []bool{false, true} {
cfg := &config.Config{}
cfg.Server.Dev = dev
b := newBackfill(&Spindle{cfg: cfg}, slog.Default(), shortTempDir(t))
got, ok := b.settings()["HYDRANT_ALLOW_PRIVATE_NETWORKS"]
if ok != dev || (dev && got != "true") {
t.Errorf("dev=%v: HYDRANT_ALLOW_PRIVATE_NETWORKS = %q (set: %v)", dev, got, ok)
}
}
}
// real hydrant, offline. nothing resolves so nothing backfills, but it has to
// start, answer, shut down and let go of its db for the next round
func TestBackfillRunsEmbeddedHydrant(t *testing.T) {
cfg := &config.Config{}
cfg.Server.PlcUrl = "http://127.0.0.1:1"
d, _ := newTestSpindleDB(t)
b := newBackfill(&Spindle{cfg: cfg, db: d}, slog.Default(), shortTempDir(t))
b.poll = 100 * time.Millisecond
b.handle = func(context.Context, tapc.Event) error { return nil }
round := func() {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
if err := b.round(ctx); err != nil {
t.Fatalf("round: %v", err)
}
}
b.Track("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")
round()
if _, err := os.Stat(filepath.Join(b.dir, "db")); err != nil {
t.Fatalf("hydrant didn't put its db under the backfill dir: %v", err)
}
// the second start has to reopen the db, and erase the first owner through hydrant's api
b.Untrack("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa")
b.Track("did:plc:bbbbbbbbbbbbbbbbbbbbbbbb")
round()
}