package 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 missed
var 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 work
type 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 is
func (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 backfilled
func (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-up
func (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 wants
func (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 owners
func (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 tracking
func (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 down
func (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 forgot
func (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 own
func (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 drops
func (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 replays
func (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 up
func (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 events
func (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 socket
type 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 fail
func (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 has
func (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 backoff
func (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 list
func (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 owner
type 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
// backfilled
func (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 date
func (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 knows
func (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 none
func (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-up
func (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 it
func (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
}