Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110package spindle
import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log/slog" "net" "net/http" "net/url" "os" "path/filepath" "slices" "strconv" "strings" "sync" "sync/atomic" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket" hydrant "tangled.org/ptr.pet/hydrant/ffi/go"
"tangled.org/core/api/org_tangled" "tangled.org/core/spindle/db" "tangled.org/core/tapc")
const ( backfillPollInterval = 3 * time.Second backfillStableChecks = 2 backfillEventAttempts = 3 backfillRetryDelay = time.Minute hydrantShutdownTimeout = 30 * time.Second
// owners whose backfill failed get another round after this, doubling // each time it fails again. the cap is what dead accounts cost forever failedOwnerRetryMin = 2 * time.Minute failedOwnerRetryMax = time.Hour
// six doublings from a minute add up to about an hour before giving up parkedRetryMin = time.Minute parkedRetries = 6
pendingCatchUpMaxAge = 7 * 24 * time.Hour
hydrantRequestTimeout = 30 * time.Second hydrantEraseTimeout = 10 * time.Minute // hydrant caps a request body at 8 MiB, which is about 195k dids hydrantDidBatch = 1000)
// jetstream has the live changes, so these are only the gaps it missedvar backfillCollections = []string{ org_tangled.RepoDeclarationNSID, org_tangled.RepoManifestNSID, org_tangled.GraphCollaboratorNSID,}
type embeddedHydrant interface { Done() <-chan struct{} Err() error Shutdown(timeout time.Duration) error}
func startHydrant(settings map[string]string) (embeddedHydrant, error) { return hydrant.Start(settings)}
// backfill runs an embedded hydrant while owners have records to backfill and// shuts it down once spindle has processed all of them, so its memory only// gets used when there is worktype backfill struct { s *Spindle l *slog.Logger dir string start func(settings map[string]string) (embeddedHydrant, error) handle func(ctx context.Context, evt tapc.Event) error // a manifest deleted during a gap never streams, so only the knot shows it refresh func(ctx context.Context, repoDid syntax.DID) error wake chan struct{}
poll time.Duration retryBase time.Duration retryFailed time.Duration retryParked time.Duration
mu sync.Mutex tracked map[syntax.DID]struct{} add map[syntax.DID]struct{} remove map[syntax.DID]struct{}
// owners hydrant couldn't backfill. hydrant only retries them while it // runs, so without this they'd wait for a new owner or a restart. only // rounds touch it, and those never overlap failing map[syntax.DID]failure
// id of the last stream event spindle finished with, -1 before the first last atomic.Int64
// a gap during catch-up only sets catchUpAgain, the running loop goes once more catchingUp bool catchUpAgain bool catchUps sync.WaitGroup closed bool}
func newBackfill(s *Spindle, l *slog.Logger, dir string) *backfill { return &backfill{ s: s, l: l, dir: dir, start: startHydrant, handle: s.markBackfilled, refresh: func(_ context.Context, repoDid syntax.DID) error { return s.markRepo(repoDid) }, wake: make(chan struct{}, 1), poll: backfillPollInterval, retryBase: time.Second, retryFailed: failedOwnerRetryMin, retryParked: parkedRetryMin, tracked: make(map[syntax.DID]struct{}), add: make(map[syntax.DID]struct{}), remove: make(map[syntax.DID]struct{}), failing: make(map[syntax.DID]failure), }}
type failure struct { rounds int retryAt time.Time}
func (b *backfill) failedOwnerRetry(rounds int) time.Duration { return min(b.retryFailed<<min(rounds-1, 10), failedOwnerRetryMax)}
// nextRetry is when the earliest failed owner is due again, zero if none isfunc (b *backfill) nextRetry() time.Time { var next time.Time for _, f := range b.failing { if next.IsZero() || f.retryAt.Before(next) { next = f.retryAt } } return next}
// Track queues owners for backfill and wakes hydrant up if any are new to this// process. hydrant skips owners it already backfilled, so repeats are cheap.func (b *backfill) Track(dids ...syntax.DID) { b.mu.Lock() queued := false for _, did := range dids { if _, ok := b.tracked[did]; ok || did == "" { continue } // an owner untracked earlier stays queued for removal, since nothing // followed their repo meanwhile. erasing then adding them again is // what makes hydrant backfill them from scratch instead of skipping b.tracked[did] = struct{}{} b.add[did] = struct{}{} queued = true } b.mu.Unlock() if queued { nudge(b.wake) }}
// hydrant skips dids it already has, so the catch-up erases and re-tracks them// saved first so a restart can't lose the gap, and a did leaves the list once backfilledfunc (b *backfill) CatchUp() error { owners, err := b.s.backfillOwners() if err != nil { return err } repos, err := b.s.db.AllRepos() if err != nil { return fmt.Errorf("loading repos to catch up: %w", err) } dids := owners for _, repo := range repos { dids = append(dids, repo.RepoDid) } dids = slices.DeleteFunc(dids, func(did syntax.DID) bool { return did == "" }) saveErr := b.s.db.AddPendingCatchUp(dids, time.Now()) b.queueCatchUp(dids) if saveErr != nil { return fmt.Errorf("saving the pending catch-up: %w", saveErr) } return nil}
// called from the jetstream reader, which can't block on the catch-upfunc (b *backfill) CatchUpLater() { b.mu.Lock() defer b.mu.Unlock() if b.catchingUp { b.catchUpAgain = true return } // close waits on catchUps, and nothing may start once closed is set if b.closed { return } b.catchingUp = true b.catchUps.Go(func() { for { if err := b.CatchUp(); err != nil { b.l.Error("jetstream missed events and the catch-up couldn't start", "err", err) } b.mu.Lock() again := b.catchUpAgain b.catchUpAgain = false b.catchingUp = again b.mu.Unlock() if !again { return } } })}
func (s *Spindle) markBackfilled(ctx context.Context, evt tapc.Event) error { if evt.Type == tapc.EvtRecord && evt.Record != nil { switch evt.Record.Collection.String() { case org_tangled.RepoManifestNSID, org_tangled.GraphCollaboratorNSID: if served, err := s.knowsRepo(evt.Record.Did); err != nil || served { return err } case org_tangled.RepoDeclarationNSID: if served, err := s.declaresServedRepo(evt.Record); err != nil || served { return err } } } return s.processEvent(withMarkOnly(ctx), evt)}
func (s *Spindle) declaresServedRepo(rec *tapc.RecordEventData) (bool, error) { if rec.Action == tapc.RecordDeleteAction { return false, nil } var declaration org_tangled.RepoDeclaration if err := json.Unmarshal(rec.Record, &declaration); err != nil { return false, nil } repoDid, err := syntax.ParseDID(declaration.Repo) if err != nil { return false, nil } rows, err := s.db.ReposByDid(repoDid) if err != nil { return false, err } return slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Owner == rec.Did && r.Rkey == rec.Rkey }), nil}
func (b *backfill) Resume() error { b.dropStalePending() pending, err := b.s.db.PendingCatchUp() if err != nil { return fmt.Errorf("loading the pending catch-up: %w", err) } if len(pending) > 0 { b.l.Info("resuming an unfinished catch-up", "dids", len(pending)) b.queueCatchUp(pending) } return nil}
func (b *backfill) dropStalePending() { dropped, err := b.s.db.DropStalePendingCatchUp(time.Now().Add(-pendingCatchUpMaxAge)) if err != nil { b.l.Error("dropping stale pending catch-up", "err", err) return } for _, did := range dropped { b.l.Warn("giving up on catching up a did that never backfilled", "did", did) }}
func (b *backfill) queueCatchUp(dids []syntax.DID) { b.mu.Lock() for _, did := range dids { b.tracked[did] = struct{}{} b.add[did] = struct{}{} b.remove[did] = struct{}{} } b.mu.Unlock() nudge(b.wake)}
// Untrack waits for the next time hydrant runs anyway, since forgetting an// owner isn't worth starting it for. a restart loses the queue, but each round// also drops whatever hydrant tracks that spindle no longer wantsfunc (b *backfill) Untrack(did syntax.DID) { b.mu.Lock() delete(b.tracked, did) delete(b.add, did) b.remove[did] = struct{}{} b.mu.Unlock() if err := b.s.db.DeletePendingCatchUp(did); err != nil { b.l.Error("forgetting a pending catch-up", "did", did, "err", err) }}
func (b *backfill) take() (add, remove []syntax.DID) { b.mu.Lock() defer b.mu.Unlock() for did := range b.add { add = append(add, did) } for did := range b.remove { remove = append(remove, did) } clear(b.add) clear(b.remove) return add, remove}
// requeue puts back work a failed round didn't get to, unless newer calls// already changed their mind about those ownersfunc (b *backfill) requeue(add, remove []syntax.DID) { b.mu.Lock() defer b.mu.Unlock() for _, did := range add { if _, ok := b.tracked[did]; ok { b.add[did] = struct{}{} } } for _, did := range remove { b.remove[did] = struct{}{} }}
// unwanted is every did in dids that spindle isn't trackingfunc (b *backfill) unwanted(dids []syntax.DID) []syntax.DID { b.mu.Lock() defer b.mu.Unlock() var out []syntax.DID for _, did := range dids { if _, ok := b.tracked[did]; !ok { out = append(out, did) } } return out}
func (b *backfill) run(ctx context.Context) { defer func() { b.mu.Lock() b.closed = true b.mu.Unlock() b.catchUps.Wait() }() // due far enough ahead that every parked record counts as due on startup due := time.Now().Add(b.retryParked << parkedRetries) for { b.retryParkedRecords(ctx, due) err := b.round(ctx) if ctx.Err() != nil { return } if err != nil { b.l.Error("backfill failed, retrying", "err", err, "in", backfillRetryDelay) select { case <-ctx.Done(): return case <-time.After(backfillRetryDelay): } due = time.Now() continue } var retry <-chan time.Time if next := b.nextWake(); !next.IsZero() { retry = time.After(time.Until(next)) } select { case <-ctx.Done(): return case <-b.wake: case <-retry: } due = time.Now() }}
func (b *backfill) nextWake() time.Time { next := b.nextRetry() parked, err := b.s.db.NextParkedRetry() if err != nil { b.l.Error("reading when parked records are due", "err", err) return next } if next.IsZero() || (!parked.IsZero() && parked.Before(next)) { next = parked } return next}
func (b *backfill) socketPath() string { return filepath.Join(b.dir, "api.sock")}
func (b *backfill) cursorPath() string { return filepath.Join(b.dir, "cursor")}
func (b *backfill) settings() map[string]string { settings := map[string]string{ "HYDRANT_DATABASE_PATH": filepath.Join(b.dir, "db"), "HYDRANT_API_BIND": "unix:" + b.socketPath(), "HYDRANT_PLC_URL": b.s.cfg.Server.PlcUrl, "HYDRANT_FILTER_COLLECTIONS": strings.Join(backfillCollections, ","), // backfill only, owners are added explicitly and live events come from jetstream "HYDRANT_ENABLE_FIREHOSE": "false", "HYDRANT_RELAY_HOSTS": "", "HYDRANT_SEED_HOSTS": "", "HYDRANT_ENABLE_CRAWLER": "false", "HYDRANT_CRAWLER_URLS": "", "HYDRANT_IDENTITY_CACHE_SIZE": "50000", "RUST_LOG": "warn", } // dev hosts resolve to loopback, so hydrant needs the private network flag if b.s.cfg.Server.Dev { settings["HYDRANT_ALLOW_PRIVATE_NETWORKS"] = "true" } return settings}
// round starts hydrant, hands it the queued owners and due retries, and// feeds its stream to spindle until every owner is settled and processed,// then shuts it downfunc (b *backfill) round(ctx context.Context) (err error) { add, remove := b.take() if len(add) == 0 && !b.retryDue(time.Now()) { b.requeue(nil, remove) return nil } defer func() { if err != nil { b.requeue(add, remove) b.s.metrics.RecordBackfillRound("failed") } else { b.s.metrics.RecordBackfillRound("done") } }()
if err := os.MkdirAll(b.dir, 0o700); err != nil { return fmt.Errorf("creating hydrant dir: %w", err) } last, err := b.loadCursor() if err != nil { return err }
started := time.Now() h, err := b.start(b.settings()) if err != nil { return fmt.Errorf("starting hydrant: %w", err) } b.s.metrics.SetBackfillRunning(true) defer func() { if err := h.Shutdown(hydrantShutdownTimeout); err != nil { b.l.Error("hydrant did not shut down", "err", err) } b.s.metrics.SetBackfillRunning(false) }()
api := newHydrantAPI(b.socketPath()) if err := api.waitReady(ctx, h); err != nil { return err } head, err := api.head(ctx) if err != nil { return err } // past the head means the db was wiped or swapped under the cursor if last >= 0 && (head == nil || last > *head) { b.l.Warn("hydrant cursor is past the end of its stream, replaying the stream from the start", "cursor", last) last = -1 } b.last.Store(last)
known, err := api.trackedRepos(ctx) if err != nil { return err } var dids []syntax.DID for _, repo := range known { dids = append(dids, repo.Did) } remove = append(remove, b.unwanted(dids)...) slices.Sort(remove) remove = slices.Compact(remove) if err := api.erase(ctx, remove); err != nil { return err } if err := api.track(ctx, add); err != nil { return err } known = slices.DeleteFunc(known, func(repo hydrantRepo) bool { _, gone := slices.BinarySearch(remove, repo.Did) return gone }) retried := b.dueRetries(known, started) if err := api.resync(ctx, retried); err != nil { return err } b.l.Info("backfilling owners", "owners", len(add), "retrying", len(retried), "untracked", len(remove), "cursor", last)
streamCtx, stopStream := context.WithCancel(ctx) streamDone := make(chan struct{}) go func() { defer close(streamDone) b.follow(streamCtx, api) }() defer func() { stopStream() <-streamDone if err := b.saveCursor(b.last.Load()); err != nil { b.l.Warn("saving hydrant cursor", "err", err) } }()
// an outcome sticks for the round, because hydrant keeps moving failed // owners back into its queue on its own backoff, so a round that waited for // that to calm down would never end settled := make(map[syntax.DID]outcome) ticker := time.NewTicker(b.poll) defer ticker.Stop() ready := 0 for { select { case <-ctx.Done(): return ctx.Err() case <-h.Done(): return fmt.Errorf("hydrant stopped: %w", h.Err()) case <-b.wake: more, less := b.take() add, remove = append(add, more...), append(remove, less...) if err := api.erase(ctx, less); err != nil { return err } if err := api.track(ctx, more); err != nil { return err } // a catch-up can re-track a did this round already settled, so forget its outcome for _, did := range more { delete(settled, did) } ready = 0 case <-ticker.C: repos, err := api.queuedRepos(ctx) if err != nil { return err } head, err := api.head(ctx) if err != nil { return err } left := 0 for _, repo := range repos { if settled[repo.Did] == waiting { settled[repo.Did] = repo.outcome() } if settled[repo.Did] == waiting { left++ } } if left == 0 && (head == nil || b.last.Load() >= *head) { ready++ } else { ready = 0 } if ready >= backfillStableChecks { // the queue only shows what hydrant is still working on, so an // owner it let go of without syncing only turns up in the full list final, err := api.trackedRepos(ctx) if err != nil { return err } for _, repo := range final { if settled[repo.Did] == waiting { settled[repo.Did] = repo.outcome() } } failing := b.recordFailures(final, settled, retried, time.Now()) b.markFailedServed(settled) b.settlePending(ctx, add, settled) b.dropStalePending() b.l.Info("backfill done, shutting hydrant down", "owners", len(add), "failed", failing, "retry_at", b.nextRetry(), "took", time.Since(started).Round(time.Second).String()) return nil } } }}
func (b *backfill) retryDue(now time.Time) bool { next := b.nextRetry() return !next.IsZero() && !next.After(now)}
// dueRetries picks the failed owners whose retry is due. one spindle has no// record of is due right away, since it either just failed for the first time// or this process restarted and forgotfunc (b *backfill) dueRetries(repos []hydrantRepo, now time.Time) []syntax.DID { var due []syntax.DID for _, repo := range repos { if repo.outcome() != failed { continue } if f, ok := b.failing[repo.Did]; !ok || !f.retryAt.After(now) { due = append(due, repo.Did) } } return due}
// recordFailures pushes back the next retry of every owner that failed again// after being retried, and forgets the ones that made it. an owner that wasn't// due keeps its time even if hydrant retried and failed it on its ownfunc (b *backfill) recordFailures(repos []hydrantRepo, settled map[syntax.DID]outcome, retried []syntax.DID, now time.Time) int { wasRetried := make(map[syntax.DID]struct{}, len(retried)) for _, did := range retried { wasRetried[did] = struct{}{} } failing := make(map[syntax.DID]failure) for _, repo := range repos { if settled[repo.Did] != failed { continue } f, known := b.failing[repo.Did] if _, ok := wasRetried[repo.Did]; !known || ok { f.rounds++ f.retryAt = now.Add(b.failedOwnerRetry(f.rounds)) } failing[repo.Did] = f } b.failing = failing b.s.metrics.SetBackfillFailingOwners(len(failing)) return len(failing)}
// follow feeds hydrant's stream to spindle from the saved cursor, reconnecting// at the last processed event if the connection dropsfunc (b *backfill) follow(ctx context.Context, api hydrantAPI) { for ctx.Err() == nil { err := b.followOnce(ctx, api) if ctx.Err() != nil { return } b.l.Warn("hydrant stream dropped, reconnecting", "err", err, "cursor", b.last.Load()) select { case <-ctx.Done(): case <-time.After(time.Second): } }}
func (b *backfill) followOnce(ctx context.Context, api hydrantAPI) error { conn, err := api.subscribe(ctx, b.last.Load()+1) if err != nil { return err } defer conn.Close() // ReadMessage doesn't watch ctx, so closing the conn is what unblocks it stop := context.AfterFunc(ctx, func() { conn.Close() }) defer stop()
lastSaved := time.Now() for { _, msg, err := conn.ReadMessage() if err != nil { return err } var evt tapc.Event if err := json.Unmarshal(msg, &evt); err != nil { return fmt.Errorf("parsing hydrant event: %w", err) } if evt.Type == "error" { return fmt.Errorf("hydrant closed the stream: %s", msg) } if err := b.process(ctx, evt); err != nil { return err } // only records move the cursor, other event ids run past the stream's head if evt.Type != tapc.EvtRecord { continue } b.last.Store(evt.ID) // a crash replays at most a second of events, which is fine since // processing a record twice changes nothing if time.Since(lastSaved) >= time.Second { if err := b.saveCursor(evt.ID); err != nil { b.l.Warn("saving hydrant cursor", "err", err) } lastSaved = time.Now() } }}
// a failed park returns an error, which stops the stream so the event replaysfunc (b *backfill) process(ctx context.Context, evt tapc.Event) error { if evt.Type != tapc.EvtRecord || evt.Record == nil { return nil } var err error for attempt := 1; ; attempt++ { if err = b.handle(ctx, evt); err == nil { return nil } if ctx.Err() != nil { return ctx.Err() } if attempt == backfillEventAttempts { break } select { case <-ctx.Done(): return ctx.Err() case <-time.After(time.Duration(attempt) * b.retryBase): } } return b.park(evt, err)}
func (b *backfill) park(evt tapc.Event, cause error) error { raw, err := json.Marshal(evt) if err != nil { return err } uri := evt.Record.AtUri().String() if err := b.s.db.ParkRecord(db.ParkedRecord{ URI: uri, Rev: evt.Record.Rev, Event: raw, RetryAt: time.Now().Add(b.retryParked), Err: cause.Error(), }); err != nil { return fmt.Errorf("parking %s: %w", uri, err) } b.l.Warn("parked a backfilled record that kept failing", "uri", uri, "err", cause, "retry_in", b.retryParked.String()) b.s.metrics.RecordBackfillParked("parked") b.countParked() return nil}
// retryParkedRecords runs off the stored event, so hydrant doesn't have to be upfunc (b *backfill) retryParkedRecords(ctx context.Context, due time.Time) { parked, err := b.s.db.DueParkedRecords(due) if err != nil { b.l.Error("listing parked records", "err", err) return } defer b.countParked() for _, p := range parked { var evt tapc.Event err := json.Unmarshal(p.Event, &evt) if err == nil { err = b.handle(ctx, evt) } if ctx.Err() != nil { return } attempts := p.Attempts + 1 var saveErr error switch { case err == nil: b.l.Info("parked record went through", "uri", p.URI, "attempts", attempts) b.s.metrics.RecordBackfillParked("recovered") saveErr = b.s.db.UnparkRecord(p.URI, p.Rev) case attempts >= parkedRetries: b.l.Error("gave up on a parked record", "uri", p.URI, "attempts", attempts, "err", err) b.s.metrics.RecordBackfillParked("gave_up") saveErr = b.s.db.UnparkRecord(p.URI, p.Rev) default: wait := b.retryParked << attempts b.l.Warn("parked record failed again", "uri", p.URI, "attempts", attempts, "retry_in", wait.String(), "err", err) saveErr = b.s.db.RescheduleParked(p.URI, p.Rev, attempts, time.Now().Add(wait), err.Error()) } if saveErr != nil { b.l.Error("updating parked record", "uri", p.URI, "err", saveErr) } }}
func (b *backfill) countParked() { n, err := b.s.db.CountParked() if err != nil { b.l.Error("counting parked records", "err", err) return } b.s.metrics.SetBackfillParkedRecords(n)}
// the cursor lives next to hydrant's db rather than in spindle's, so wiping// that dir also starts the stream over instead of skipping a new db's eventsfunc (b *backfill) loadCursor() (int64, error) { raw, err := os.ReadFile(b.cursorPath()) if errors.Is(err, os.ErrNotExist) { return -1, nil } if err != nil { return 0, fmt.Errorf("reading hydrant cursor: %w", err) } last, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64) if err != nil { return 0, fmt.Errorf("parsing hydrant cursor: %w", err) } return last, nil}
func (b *backfill) saveCursor(last int64) error { if last < 0 { return nil } tmp := b.cursorPath() + ".tmp" if err := os.WriteFile(tmp, []byte(strconv.FormatInt(last, 10)), 0o600); err != nil { return err } return os.Rename(tmp, b.cursorPath())}
// hydrantAPI talks to hydrant's http api over its unix sockettype hydrantAPI struct { socket string http *http.Client}
func newHydrantAPI(socket string) hydrantAPI { dial := func(ctx context.Context, _, _ string) (net.Conn, error) { var d net.Dialer return d.DialContext(ctx, "unix", socket) } return hydrantAPI{ socket: socket, http: &http.Client{Transport: &http.Transport{DialContext: dial}}, }}
func (a hydrantAPI) do(ctx context.Context, method, path string, body any, out any) error { return a.doWithin(ctx, hydrantRequestTimeout, method, path, body, out)}
func (a hydrantAPI) doWithin(ctx context.Context, timeout time.Duration, method, path string, body any, out any) error { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() var reqBody io.Reader if body != nil { raw, err := json.Marshal(body) if err != nil { return err } reqBody = bytes.NewReader(raw) } req, err := http.NewRequestWithContext(ctx, method, "http://hydrant"+path, reqBody) if err != nil { return err } req.Header.Set("Accept", "application/json") if body != nil { req.Header.Set("Content-Type", "application/json") } resp, err := a.http.Do(req) if err != nil { return fmt.Errorf("hydrant %s %s: %w", method, path, err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return fmt.Errorf("hydrant %s %s: %s: %s", method, path, resp.Status, bytes.TrimSpace(msg)) } if out == nil { return nil } return json.NewDecoder(resp.Body).Decode(out)}
// the api binds after start, so the first probes are expected to failfunc (a hydrantAPI) waitReady(ctx context.Context, h embeddedHydrant) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() for { err := a.doWithin(ctx, time.Second, http.MethodGet, "/stream/head", nil, nil) if err == nil { return nil } select { case <-ctx.Done(): return fmt.Errorf("hydrant api never came up: %w", err) case <-h.Done(): return fmt.Errorf("hydrant stopped before its api came up: %w", h.Err()) case <-time.After(100 * time.Millisecond): } }}
func (a hydrantAPI) track(ctx context.Context, dids []syntax.DID) error { return a.sendDids(ctx, http.MethodPut, "/repos", dids)}
// erase makes hydrant forget these owners entirely. a plain untrack keeps// their records, and a later backfill would skip every one it still hasfunc (a hydrantAPI) erase(ctx context.Context, dids []syntax.DID) error { for _, did := range dids { if err := a.doWithin(ctx, hydrantEraseTimeout, http.MethodPost, "/redactions/"+did.String(), nil, nil); err != nil { return err } } return nil}
// resync makes hydrant retry these owners right away instead of on its own backofffunc (a hydrantAPI) resync(ctx context.Context, dids []syntax.DID) error { return a.sendDids(ctx, http.MethodPost, "/repos/resync", dids)}
// both endpoints are idempotent, so a failed round can resend the whole listfunc (a hydrantAPI) sendDids(ctx context.Context, method, path string, dids []syntax.DID) error { type didRef struct { Did syntax.DID `json:"did"` } for batch := range slices.Chunk(dids, hydrantDidBatch) { body := make([]didRef, len(batch)) for i, did := range batch { body[i] = didRef{did} } if err := a.do(ctx, method, path, body, nil); err != nil { return err } } return nil}
// hydrantRepo is what GET /repos says about one ownertype hydrantRepo struct { Did syntax.DID `json:"did"` Tracked bool `json:"tracked"` Status string `json:"status"` Pending bool `json:"pending"` // set while the owner sits in hydrant's resync queue Resync *hydrantResync `json:"resync,omitempty"`}
type hydrantResync struct { State string `json:"state"` Kind string `json:"kind,omitempty"` HostFailures int `json:"host_failures,omitempty"`}
type outcome int
const ( waiting outcome = iota done failed)
// outcome says whether waiting longer this round can still get this owner// backfilledfunc (r hydrantRepo) outcome() outcome { switch { case r.Resync != nil && r.Resync.State == "gone": return done case r.Resync != nil: // a live pds pacing hydrant gets through soon. a pds that keeps // failing makes hydrant file its other owners as ratelimited too if r.Resync.Kind == "ratelimited" && r.Resync.HostFailures == 0 { return waiting } return failed case r.Pending: return waiting } switch r.Status { case "synced", "deactivated", "takendown", "suspended", "deleted", "throttled": return done } // hydrant let go of it without calling it synced, so retrying later is // all that's left return failed}
// trackedRepos lists every owner hydrant keeps up to datefunc (a hydrantAPI) trackedRepos(ctx context.Context) ([]hydrantRepo, error) { return a.listRepos(ctx, false)}
// queuedRepos lists the owners hydrant still has a backfill or a retry queued// for. an owner missing from it is done with for now, so a round can watch// this instead of every owner it knowsfunc (a hydrantAPI) queuedRepos(ctx context.Context) ([]hydrantRepo, error) { return a.listRepos(ctx, true)}
func (a hydrantAPI) listRepos(ctx context.Context, queued bool) ([]hydrantRepo, error) { const limit = 1000 var repos []hydrantRepo cursor := "" for { var page []hydrantRepo query := url.Values{"limit": {strconv.Itoa(limit)}} if queued { query.Set("queued", "true") } if cursor != "" { query.Set("cursor", cursor) } if err := a.do(ctx, http.MethodGet, "/repos?"+query.Encode(), nil, &page); err != nil { return nil, err } for _, repo := range page { if repo.Tracked { repos = append(repos, repo) } } if len(page) < limit { return repos, nil } cursor = page[len(page)-1].Did.String() }}
// head is the id of the newest event a replay can deliver, nil while there are nonefunc (a hydrantAPI) head(ctx context.Context) (*int64, error) { var out struct { ID *int64 `json:"id"` } if err := a.do(ctx, http.MethodGet, "/stream/head", nil, &out); err != nil { return nil, err } return out.ID, nil}
func (a hydrantAPI) subscribe(ctx context.Context, cursor int64) (*websocket.Conn, error) { dialer := websocket.Dialer{ NetDialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { var d net.Dialer return d.DialContext(ctx, "unix", a.socket) }, HandshakeTimeout: 10 * time.Second, } url := "ws://hydrant/stream?cursor=" + strconv.FormatInt(max(cursor, 0), 10) conn, _, err := dialer.DialContext(ctx, url, nil) if err != nil { return nil, fmt.Errorf("subscribing to hydrant stream: %w", err) } return conn, nil}
func (b *backfill) markFailedServed(settled map[syntax.DID]outcome) { for did, o := range settled { if o != failed { continue } if served, err := b.s.knowsRepo(did); err != nil || !served { continue } if err := b.s.markRepo(did); err != nil { b.l.Error("marking a repo whose backfill failed", "repo", did, "err", err) } }}
// only this round's outcomes count, an older one can predate a later catch-upfunc (b *backfill) settlePending(ctx context.Context, roundDids []syntax.DID, settled map[syntax.DID]outcome) { var backfilled []syntax.DID for _, did := range roundDids { if settled[did] != done { continue } if ctx.Err() != nil { break } if b.refreshServed(ctx, did) { backfilled = append(backfilled, did) } } if err := b.s.db.DeletePendingCatchUp(backfilled...); err != nil { b.l.Error("clearing the pending catch-up", "err", err) }}
// false keeps the did pending, a failed refresh doesn't because the retry set owns itfunc (b *backfill) refreshServed(ctx context.Context, did syntax.DID) bool { served, err := b.s.knowsRepo(did) if err != nil { b.l.Error("checking whether a backfilled did is a served repo", "did", did, "err", err) return false } if !served { return true } if err := b.refresh(ctx, did); err != nil { if ctx.Err() != nil { return false } b.l.Warn("refreshing a repo after the catch-up failed", "repo", did, "err", err) b.s.failRepoRefresh(did) } return true}