Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
6.9 kB · 165 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166package atproto
import ( "context" "time"
indigoatproto "github.com/bluesky-social/indigo/api/atproto" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/reposync")
// revCASAttempts is how many times a commit tries to place itself on the repo// row before concluding it found a gap.//// One attempt is the whole story when commits arrive in order. Within a single// relay they now do -- the parallel scheduler chains same-DID events through a// per-repo queue -- but this node consumes several relays at once, each with a// scheduler of its own, and the deduper only guarantees that a given commit is// handled once, not that two commits on one repo are handled in order. So they// still race, and the older one can win the CAS after the newer one has already// missed it. A second look then finds the row exactly where the newer commit// expected it. Three is one more than that story needs; a lost race past it// costs one unnecessary repair, never a missed record.const revCASAttempts = 3
// repairSlack is how far before the last known rev a repair starts reading.//// The rev is a TID stamped by the repo's PDS, and the floor it becomes is// compared against rkeys stamped by that same PDS, so this is not correcting// for a clock difference between us and them. It covers the gap between when a// record's rkey was minted and when the commit carrying it was stamped, plus// any host whose clock has been stepped backwards since.const repairSlack = time.Hour
// trackCommitRev keeps this node's idea of a repo's revision honest, and is// what makes the firehose a checkable stream rather than a hope.//// A #commit says which rev it follows (Since) and which rev it creates (Rev).// Called after the event's ops have been indexed, this either advances the// stored rev -- proving that we have applied every commit for this repo in an// unbroken chain -- or discovers that we cannot prove it, and marks the repo// for repair.//// Repos we do not track, and repos in the middle of a backfill, are left// entirely alone: an empty stored Version already means "sync me", and the// backfill about to finish will write a rev of its own.func (atsync *ATProtoSynchronizer) trackCommitRev(ctx context.Context, evt *indigoatproto.SyncSubscribeRepos_Commit) { if evt.Rev == "" { return } since := "" if evt.Since != nil { since = *evt.Since }
for attempt := 0; attempt < revCASAttempts; attempt++ { // The happy path is one statement and no read: if the row still holds // the rev this commit follows, this commit is the next one. applied, err := atsync.Model.AdvanceRepoVersion(ctx, evt.Repo, since, evt.Rev) if err != nil { log.Error(ctx, "failed to advance repo rev", "did", evt.Repo, "err", err) return } if applied { return }
// It did not apply. Find out which of the four reasons it was. row, err := atsync.Model.GetRepo(evt.Repo) if err != nil { log.Error(ctx, "failed to read repo rev", "did", evt.Repo, "err", err) return } switch { case row == nil || row.Version == "" || syncInFlight(evt.Repo): // A stranger, or a repo whose backfill is running or owed. Nothing // here is better than what that backfill will write. return case row.Version == since: // The rev this commit follows arrived while we were looking: an // out-of-order sibling won the CAS after ours missed it. Try again. continue case evt.Rev <= row.Version: // Old news -- a redelivery, or a commit we already have by way of // a backfill. The ops were indexed idempotently; the rev stays put // rather than regressing. return }
// evt.Rev is ahead of us and does not follow what we have: commits for // this repo went missing. The ops from this one are indexed either way // -- fresh data now beats correct data later -- but the span between // our rev and this one has to be re-read. log.Log(ctx, "firehose gap detected", "did", evt.Repo, "ourRev", row.Version, "evtSince", since, "evtRev", evt.Rev) marked, err := atsync.Model.MarkRepoForRepair(ctx, evt.Repo, row.Version) if err != nil { log.Error(ctx, "failed to mark repo for repair", "did", evt.Repo, "err", err) return } if !marked { // Somebody moved the row between the read and the mark. Whatever // they wrote, this commit still has to place itself against it. continue } return }}
// repairFloor is how far back a sync reads the windowed collections.//// First contact reads [InitialWindow] and nothing more: the account is servable// in seconds and the deepening ladder fills in its history afterwards.//// A repair is different. lastRev is where our index was known good, so that is// where the missed span starts, and everything written during the gap has an// rkey from inside it. Reading from just before that rev covers the whole gap// for a fraction of what re-reading the ladder would cost -- and the result is// still never shallower than a first sync, so a repair of a repo we saw a// minute ago still refreshes the last day.//// Known limitation, deliberately not solved here: this finds records created// during the gap, not records DELETED during it, and not a record backdated// into a window this walk does not cover. Both need a diff of what we hold// against what the repo holds, which is future work.func repairFloor(lastRev string, now time.Time) string { standard := now.Add(-InitialWindow) if lastRev == "" { return reposync.TIDForTime(standard) } revTime, err := reposync.TimeForTID(lastRev) if err != nil { // Not a TID we can place in time -- an old hand-written row, or a host // with its own idea of revs. The standard window is the safe answer. return reposync.TIDForTime(standard) } if from := revTime.Add(-repairSlack); from.Before(standard) { return reposync.TIDForTime(from) } return reposync.TIDForTime(standard)}
// mergeBackfillState folds what a sync just learned into what the row already// knew about this repo's history.//// It exists because a repair is a shallow sync of a repo that may have years of// history indexed: the walk it just did says "the last day is indexed", which// is true and is not the whole truth. Walking a recent window cannot un-complete// history, and cannot raise a watermark that reaches further back than it does.func mergeBackfillState(old *model.Repo, res backfillResult) (floor string, done bool) { floor, done = res.Floor, res.Done if old != nil && old.BackfillDone { done = true } if done { // Nothing left to fetch, so there is no watermark to keep: an empty // floor is what a completed history reads as everywhere else. return "", true } if old != nil && old.BackfillFloor != "" && (floor == "" || old.BackfillFloor < floor) { floor = old.BackfillFloor } return floor, false}