Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.6 kB · 220 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221package atproto
import ( "context" "fmt" "time"
"golang.org/x/time/rate" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log")
// deepenIdleInterval is how long the deepener waits before scanning again when// a pass achieved nothing -- either because every repo already has its whole// history, or because the ones that do not are failing. Long enough that a node// with nothing to fetch costs one query every ten minutes, short enough that a// repo discovered by the firehose starts getting its history within one.const deepenIdleInterval = 10 * time.Minute
// DeepenForever fetches this node's history, continuously and slowly, for as// long as ctx lives.//// History is the one part of the sync engine nothing waits for. A repo is// servable the moment its shallow sync lands; everything older is a nicety that// arrives when it arrives. Boot used to be when all of it arrived at once --// the sweep drains every repo's ladder in one pass, and restarts come round// more often than the six-hour sweep interval does, so in practice a serving// node spent its every boot replaying years of the network's chat at full lane// speed. So this is deliberately a trickle: one global rate limiter over the// whole node (see [config.DefaultDeepenRate]), and no relationship at all to// the pace of [ATProtoSynchronizer.SweepForever]'s reconciliation passes.//// It scans, ladders what it finds, and scans again. The scan is the entire// state machine: BackfillFloor and BackfillDone on the repo rows say where each// repo got to, so a pass that is cancelled, a process that is killed and a node// that is redeployed all resume from the same place, and repos discovered while// a pass was running are picked up by the next one.//// A pass and a sweep may run at once and may even touch the same repo. Nothing// guards against it because nothing needs to: the per-DID handleLock serializes// the walks, the per-host pdsLock serializes the requests, and the row updates// are compare-and-swap -- AdvanceRepoBackfill refuses a row somebody has// wedged for repair -- so the loser of any race leaves the row alone and// re-walks the window later.func (atsync *ATProtoSynchronizer) DeepenForever(ctx context.Context) { // The same hold the boot sweep takes, for the same reason: a warm index's // busiest minutes are the ones just after a restart, and history is in no // hurry. A fresh index waits for nothing, which is the initial build -- // the case where this loop IS the boot work. if !deepenSleep(ctx, atsync.sweepBootDelay(ctx)) { return } limiter := deepenLimiter(atsync.deepenRate()) for ctx.Err() == nil { windows, err := atsync.deepenPass(ctx, limiter) if err != nil && ctx.Err() == nil { log.Error(ctx, "history deepening pass failed", "err", err) } if ctx.Err() != nil { return } if windows == 0 { // Nothing to deepen, or nothing that can be: either way, asking // again immediately would be a hot loop over the same answer. if !deepenSleep(ctx, deepenIdleInterval) { return } } }}
// deepenSleep waits out d, reporting whether the context outlived it.func deepenSleep(ctx context.Context, d time.Duration) bool { if d <= 0 { return ctx.Err() == nil } select { case <-ctx.Done(): return false case <-time.After(d): return true }}
// deepenPass walks the history of every repo that owes any, and reports how// many windows that came to.//// It is the sweep's lane machinery with nothing but ladders in it: work is// sharded by host, one worker per host and at most --sweep-concurrency of them,// and within a host it is breadth-first, so every account a PDS serves reaches// a week of history before any of them reaches a month. What is different is// the pace -- every window waits on the node-wide limiter first -- and that// zero repos is an ordinary outcome rather than the end of anything.func (atsync *ATProtoSynchronizer) deepenPass(ctx context.Context, limiter *rate.Limiter) (int, error) { items, floors, err := atsync.deepenCandidates(ctx) if err != nil { return 0, err } if len(items) == 0 { return 0, nil } progress := &sweepProgress{mode: sweepModeDeepen, rate: deepenRateLabel(limiter)} progress.begin(0, 0, floors) stop := progress.start(ctx) defer stop()
log.Log(ctx, "deepening repo history", "repos", len(items), "hosts", laneCount(items), "rate", progress.rate)
// The lane cap here is the deepener's own --sweep-concurrency budget, not // one shared with a sweep that may be running: what keeps a host from being // talked to twice at once is its pdsLock, which both passes take. sched := newLaneScheduler(ctx, atsync.sweepConcurrency(), func(ctx context.Context, step sweepStep) bool { // The cap is on window walks rather than on requests: a window is one // range walk against one host and one burst of writes into the index, // which is the unit both of them feel. // // A lane waiting here is holding its slot, so a paced node runs fewer // hosts at once than its cap allows. That is the rate doing its job: // with a node-wide budget the slots stop being the scarce resource, and // a pass takes windows/rate however they are spread. if err := limiter.Wait(ctx); err != nil { return false } return atsync.sweepWindow(ctx, progress, step) }) for _, item := range items { sched.add(item) } lanes, err := sched.wait() if err != nil { return progress.windowsWalked(), err } windows := progress.windowsWalked() log.Log(ctx, "deepening pass complete", append([]any{"repos", len(items), "hosts", lanes}, progress.status()...)...) return windows, nil}
// deepenCandidates is every repo that is servable and still owes history,// laned by host, together with the watermarks they start from.//// A repo with a completed sync always has a host recorded on its row -- the// sync that completed is what wrote it -- so unlike a sweep this needs no// identity resolution to shard by. A row that somehow names no host still gets// a lane of its own rather than sharing one, exactly as [sweepLane] says.func (atsync *ATProtoSynchronizer) deepenCandidates(ctx context.Context) ([]sweepItem, map[string]time.Time, error) { dids, err := atsync.sweepCandidates(ctx) if err != nil { return nil, nil, err } var items []sweepItem floors := map[string]time.Time{} for _, did := range dids { repo, err := atsync.Model.GetRepo(did) if err != nil { return nil, nil, fmt.Errorf("failed to get repo for %s: %w", did, err) } switch { case repo == nil || repo.Version == "": // Never synced, or wedged for repair: the sweep's shallow queue // owns those, and deepening one would read the wrong ranges. continue case repo.TerminalStatus() || repo.BackfillDone: continue } items = append(items, sweepItem{ DID: did, Handle: repo.Handle, Lane: sweepLane(did, repo.PDS), Deepen: true, }) floors[did] = backfillFloorTime(repo.BackfillFloor) } return items, floors, nil}
// deepenRate is how many history windows a minute this node walks. 0 is// unlimited; anything negative is the default.func (atsync *ATProtoSynchronizer) deepenRate() int { if atsync.CLI == nil || atsync.CLI.DeepenRate < 0 { return config.DefaultDeepenRate } return atsync.CLI.DeepenRate}
// deepenBurstWindow is how much of the budget the limiter will hand out at// once: a few seconds' worth, so that a lane which has been waiting on a slow// host is not then made to wait on the clock, while a node still cannot empty a// minute of budget into one host in one go.const deepenBurstWindow = 3 * time.Second
// deepenLimiter turns a windows-per-minute cap into the limiter that paces// them. Zero means no cap, which is [rate.Inf]: a limiter that exists and never// waits, so there is nothing to nil-check at the call site.func deepenLimiter(windowsPerMinute int) *rate.Limiter { if windowsPerMinute <= 0 { return rate.NewLimiter(rate.Inf, 1) } perSecond := float64(windowsPerMinute) / 60 burst := int(perSecond * deepenBurstWindow.Seconds()) if burst < 1 { burst = 1 } return rate.NewLimiter(rate.Limit(perSecond), burst)}
// deepenRateLabel renders a limiter's pace for the status line.func deepenRateLabel(limiter *rate.Limiter) string { if limiter == nil || limiter.Limit() == rate.Inf { return "unlimited" } return fmt.Sprintf("%d/min", int(float64(limiter.Limit())*60))}
// windowsWalked is how many history windows this pass has landed.func (p *sweepProgress) windowsWalked() int { p.mu.Lock() defer p.mu.Unlock() return p.windows}