Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971package 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 thatfunc 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 usestype 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 knowfunc (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 onefunc (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 syncedfunc (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 againfunc 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 readingfunc 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 thenfunc 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 goingfunc 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 awayfunc 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 backofffunc 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 onefunc 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 toofunc 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 cursorfunc 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 roundfunc 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()}