From df9bcb4b84ae1e53aedb30743f5d17cc0b596ad8 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Thu, 30 Jul 2026 16:27:39 -0700 Subject: [PATCH] atproto: check the firehose for holes, and ask the hosts about the rest The engine could repair any repo cheaply, but nothing ever noticed that a repo needed repairing. #commit events were applied without checking that they followed the commit we last saw, the stored rev went stale the moment a backfill finished, and a repo with BackfillDone was never looked at again -- so a span of commits missed while this node was down, or written before a fresh index started listening, stayed missing forever and looked exactly like an account that had gone quiet. Two things close that loop. Contiguity: every commit says which rev it follows and which it creates, so after its ops are indexed it either advances our stored rev -- proving an unbroken chain from the backfill to now -- or it cannot, and the repo is marked for repair. Events are handled a goroutine each, with no ordering between two commits on one repo, so the advance is a DB-level compare-and-swap: the happy path is one statement and no read, and a losing race is resolved by re-reading the row rather than by whoever wrote last. Marking clears Version -- the wedge every repair path already keys on -- and nothing else, keeping the backfill watermark, the root CID and the status a repair must not lose. Head checks: a sweep now starts each servable repo with one getLatestCommit and compares it with the rev we hold. Same rev, nothing to do. Different rev, and the repo is marked for repair exactly as a firehose gap marks it, and repaired by the machinery that was already there. That runs through the phase-9 lane scheduler as a step class of its own, ahead of a lane's shallow and deepening work, so the one-worker-per-host invariant is untouched; a check that finds drift hands the repair back to the lane it is already running on. Sweeps now repeat on --sweep-interval (default 6h, 0 disables), with a guard so a long sweep cannot have a second one started on top of it. A repair is not a first sync: the missed span starts at the rev we were last good at, so it reads from an hour before that rev's own timestamp (never less than the usual day), and it merges what it learns into the row instead of overwriting it -- walking one recent window cannot un-complete a repo's history. What it still cannot see is records deleted during a gap, which needs a diff rather than a walk; that is future work and is written down where it matters. Also here, from the rollout audit: a backfilled teleport no longer schedules an arrival notification (a fresh index would have announced every teleport an account ever did, at once), and both sqlite databases get a busy timeout, so the background `streamplace sync` that warms a new index revision cannot fail the server's writes. Co-Authored-By: Claude Fable 5 --- pkg/atproto/atproto.go | 40 +++++- pkg/atproto/backfill_walk.go | 13 +- pkg/atproto/contiguity.go | 162 +++++++++++++++++++++++ pkg/atproto/contiguity_test.go | 228 +++++++++++++++++++++++++++++++++ pkg/atproto/firehose.go | 25 +++- pkg/atproto/headcheck.go | 93 ++++++++++++++ pkg/atproto/headcheck_test.go | 99 ++++++++++++++ pkg/atproto/sweep.go | 197 +++++++++++++++++++++++----- pkg/atproto/sweep_test.go | 160 ++++++++++++++++++++++- pkg/atproto/sync.go | 8 ++ pkg/atproto/teleport_test.go | 90 +++++++++++++ pkg/cmd/streamplace.go | 18 ++- pkg/config/config.go | 19 +++ pkg/config/config_test.go | 25 ++++ pkg/model/model.go | 25 ++++ pkg/model/repo.go | 58 +++++++++ pkg/model/repo_test.go | 161 +++++++++++++++++++++++ pkg/reposync/head.go | 41 +++++- pkg/statedb/statedb.go | 17 ++- pkg/statedb/statedb_test.go | 17 +++ 20 files changed, 1422 insertions(+), 74 deletions(-) create mode 100644 pkg/atproto/contiguity.go create mode 100644 pkg/atproto/contiguity_test.go create mode 100644 pkg/atproto/headcheck.go create mode 100644 pkg/atproto/headcheck_test.go create mode 100644 pkg/atproto/teleport_test.go create mode 100644 pkg/model/repo_test.go diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go index 5c88bf7c..f5b050ef 100644 --- a/pkg/atproto/atproto.go +++ b/pkg/atproto/atproto.go @@ -17,7 +17,6 @@ import ( "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/model" - "stream.place/streamplace/pkg/reposync" ) var SyncGetRepo = comatproto.SyncGetRepo @@ -116,7 +115,13 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s // First contact is shallow: everything this node indexes, but only the last // [InitialWindow] of the collections that can hold years of records. The // account is servable in seconds; the sweep deepens its history afterwards. - floor := reposync.TIDForTime(time.Now().Add(-InitialWindow)) + // A repo marked for repair instead reads from where its index was last + // known good, which is where the span it missed begins -- see [repairFloor]. + repairFrom := "" + if oldRepo != nil { + repairFrom = oldRepo.RepairFrom + } + floor := repairFloor(repairFrom, time.Now()) result, err := atsync.backfillRepo(ctx, ident, &xrpcc, floor) if err != nil { if parked := parkTerminalRepo(ctx, mod, ident.DID.String(), err); parked != nil { @@ -127,6 +132,11 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s // A completed backfill proves the account is fine, so Status goes back to // empty -- UpdateRepo writes every column, so this happens by construction. + // The same property is why the history state is merged rather than assigned: + // a repair walks one recent window and would otherwise report a repo with + // five years indexed as having a day. RepairFrom is left zero on purpose -- + // the repair it asked for is the one that just finished. + floor, done := mergeBackfillState(oldRepo, result) newRepo := model.Repo{ DID: ident.DID.String(), PDS: ident.PDSEndpoint(), @@ -134,8 +144,8 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepo(ctx context.Context, handle s RootCID: result.RootCID, Handle: ident.Handle.String(), Status: model.RepoStatusOK, - BackfillFloor: result.Floor, - BackfillDone: result.Done, + BackfillFloor: floor, + BackfillDone: done, } err = mod.UpdateRepo(&newRepo) if err != nil { @@ -289,6 +299,9 @@ func (atsync *ATProtoSynchronizer) RefreshIdentity(ctx context.Context, did stri // this repo's whole history again from the top of the ladder. newRepo.BackfillFloor = oldRepo.BackfillFloor newRepo.BackfillDone = oldRepo.BackfillDone + // And for the repair watermark, which is how a pending repair knows + // which span of history it is there to re-read. + newRepo.RepairFrom = oldRepo.RepairFrom } err = atsync.Model.UpdateRepo(&newRepo) if err != nil { @@ -313,7 +326,16 @@ func (atsync *ATProtoSynchronizer) ResolveAuthorHandle(ctx context.Context, did return handle } -func (atsync *ATProtoSynchronizer) resolveIdent(ctx context.Context, arg string, cached bool) (*identity.Identity, error) { +// directory hands back the identity directory to resolve with, building the +// pair on first use. +// +// Under a lock because a sweep resolves identities from dozens of goroutines at +// once -- lane workers and the sharding resolver, at the same instant -- and +// two of them racing to install the lazily built directory would each end up +// using a different cache, if the race detector let them get that far. +func (atsync *ATProtoSynchronizer) directory(cached bool) identity.Directory { + atsync.dirMu.Lock() + defer atsync.dirMu.Unlock() if atsync.PLCDirectory == nil { atsync.PLCDirectory = CustomDirectory(atsync.CLI.PLCURL) } @@ -321,10 +343,14 @@ func (atsync *ATProtoSynchronizer) resolveIdent(ctx context.Context, arg string, cachedDir := identity.NewCacheDirectory(atsync.PLCDirectory, 250_000, time.Hour*24, time.Minute*2, time.Minute*5) atsync.CachedPLCDirectory = &cachedDir } - dir := atsync.PLCDirectory if cached { - dir = atsync.CachedPLCDirectory + return atsync.CachedPLCDirectory } + return atsync.PLCDirectory +} + +func (atsync *ATProtoSynchronizer) resolveIdent(ctx context.Context, arg string, cached bool) (*identity.Identity, error) { + dir := atsync.directory(cached) id, err := syntax.ParseAtIdentifier(arg) if err != nil { return nil, err diff --git a/pkg/atproto/backfill_walk.go b/pkg/atproto/backfill_walk.go index c4615e8e..0ec6ea6b 100644 --- a/pkg/atproto/backfill_walk.go +++ b/pkg/atproto/backfill_walk.go @@ -271,15 +271,10 @@ func (atsync *ATProtoSynchronizer) backfillRepo(ctx context.Context, ident *iden func (atsync *ATProtoSynchronizer) walkBackfill(ctx context.Context, ident *identity.Identity, xrpcc *xrpc.Client, ranges []reposync.KeyRange) (string, string, error) { did := ident.DID.String() - dir := atsync.PLCDirectory - if dir == nil { - // resolveIdent initializes this lazily, and every caller goes through - // it first; be defensive rather than nil-panic. Note this is the - // *uncached* directory on purpose: a signing key cached from before a - // rotation would fail commit verification, and backfills are rare - // enough that the extra lookup does not matter. - dir = CustomDirectory(atsync.CLI.PLCURL) - } + // The *uncached* directory on purpose: a signing key cached from before a + // rotation would fail commit verification, and backfills are rare enough + // that the extra lookup does not matter. + dir := atsync.directory(false) // Every retry in this walk consults what the host has been telling us about // backing off; see [pdsBackoffHints]. It only works if the calls go through diff --git a/pkg/atproto/contiguity.go b/pkg/atproto/contiguity.go new file mode 100644 index 00000000..f07aba14 --- /dev/null +++ b/pkg/atproto/contiguity.go @@ -0,0 +1,162 @@ +package 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. They do not: +// events are handled one goroutine each, so two commits on one repo 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 +} diff --git a/pkg/atproto/contiguity_test.go b/pkg/atproto/contiguity_test.go new file mode 100644 index 00000000..8c3b3901 --- /dev/null +++ b/pkg/atproto/contiguity_test.go @@ -0,0 +1,228 @@ +package atproto + +import ( + "context" + "sync" + "testing" + "time" + + indigoatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/reposync" +) + +// contiguityTestSync is a synchronizer with nothing but an index: the rev +// tracking runs off the repo row and touches no network. +func contiguityTestSync(t *testing.T) (*ATProtoSynchronizer, model.Model) { + t.Helper() + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + return &ATProtoSynchronizer{Model: mod}, mod +} + +func commitEvent(did, since, rev string) *indigoatproto.SyncSubscribeRepos_Commit { + evt := &indigoatproto.SyncSubscribeRepos_Commit{Repo: did, Rev: rev} + if since != "" { + evt.Since = &since + } + return evt +} + +// syncedRepo is a repo with a completed sync and a history behind it: exactly +// the row a firehose gap must not damage. +func syncedRepo(did, version string) *model.Repo { + return &model.Repo{ + DID: did, + Handle: "someone.example", + PDS: "https://pds.example", + Version: version, + RootCID: "bafyreiabc", + BackfillFloor: "3lpfloor00000", + BackfillDone: true, + } +} + +// TestTrackCommitRev is the contiguity check itself: the three things a commit +// can be relative to what we hold, and the two kinds of repo it must not touch. +// +// The ops of an event are indexed before this runs in every case -- gap +// included, since fresh data now beats correct data later -- so what is under +// test here is only what the event does to the row. +func TestTrackCommitRev(t *testing.T) { + ctx := context.Background() + atsync, mod := contiguityTestSync(t) + + // Contiguous: the commit follows the rev we hold, so we hold its rev now + // and the chain from our backfill to here is unbroken. + require.NoError(t, mod.UpdateRepo(syncedRepo("did:plc:chain", "3lprev0000000"))) + atsync.trackCommitRev(ctx, commitEvent("did:plc:chain", "3lprev0000000", "3lprev0000001")) + got, err := mod.GetRepo("did:plc:chain") + require.NoError(t, err) + require.Equal(t, "3lprev0000001", got.Version) + require.Equal(t, "bafyreiabc", got.RootCID, "only the rev moves") + + // Stale: a redelivery from a second relay, or a commit a backfill already + // read. The rev must not regress. + atsync.trackCommitRev(ctx, commitEvent("did:plc:chain", "3lprev0000000", "3lprev0000001")) + atsync.trackCommitRev(ctx, commitEvent("did:plc:chain", "3lpolder00000", "3lpold0000000")) + got, err = mod.GetRepo("did:plc:chain") + require.NoError(t, err) + require.Equal(t, "3lprev0000001", got.Version, "old news does not move the rev backwards") + require.Empty(t, got.RepairFrom, "and is not a gap") + + // Gap: a commit from ahead of us that does not follow what we hold. The + // repo is wedged for repair, with its history intact and the rev the + // repair has to start from written down. + require.NoError(t, mod.UpdateRepo(syncedRepo("did:plc:gap", "3lprev0000000"))) + atsync.trackCommitRev(ctx, commitEvent("did:plc:gap", "3lpmissed00000", "3lprev0000009")) + got, err = mod.GetRepo("did:plc:gap") + require.NoError(t, err) + require.Empty(t, got.Version, "a gap wedges the repo so the repair path finds it") + require.Equal(t, "3lprev0000000", got.RepairFrom) + require.Equal(t, "bafyreiabc", got.RootCID) + require.Equal(t, "3lpfloor00000", got.BackfillFloor) + require.True(t, got.BackfillDone, "an hour of missed commits does not un-index a history") + + // A commit with no Since at all -- the first commit of a repo, or a relay + // that does not send one -- cannot be proven contiguous, so it is a gap. + require.NoError(t, mod.UpdateRepo(syncedRepo("did:plc:nosince", "3lprev0000000"))) + atsync.trackCommitRev(ctx, commitEvent("did:plc:nosince", "", "3lprev0000009")) + got, err = mod.GetRepo("did:plc:nosince") + require.NoError(t, err) + require.Empty(t, got.Version) + + // A stranger stays a stranger: the firehose does not create rows. + atsync.trackCommitRev(ctx, commitEvent("did:plc:stranger", "3lprev0000000", "3lprev0000001")) + got, err = mod.GetRepo("did:plc:stranger") + require.NoError(t, err) + require.Nil(t, got) + + // A repo whose backfill is owed or running is left alone: an empty Version + // already means "sync me", and the sync will write a rev of its own. + require.NoError(t, mod.UpdateRepo(&model.Repo{DID: "did:plc:wedged", PDS: "https://pds.example"})) + atsync.trackCommitRev(ctx, commitEvent("did:plc:wedged", "3lprev0000000", "3lprev0000001")) + got, err = mod.GetRepo("did:plc:wedged") + require.NoError(t, err) + require.Empty(t, got.Version) + require.Empty(t, got.RepairFrom, "an unsynced repo has no gap to repair") + + // An event with no rev is not evidence of anything. + require.NoError(t, mod.UpdateRepo(syncedRepo("did:plc:norev", "3lprev0000000"))) + atsync.trackCommitRev(ctx, commitEvent("did:plc:norev", "3lpsomething0", "")) + got, err = mod.GetRepo("did:plc:norev") + require.NoError(t, err) + require.Equal(t, "3lprev0000000", got.Version) +} + +// TestTrackCommitRevOutOfOrder is the race the CAS exists for: events are +// handled a goroutine each, so a repo's commits arrive in whatever order the +// scheduler feels like. However they interleave, the chain must end at the +// newest rev, and commits that really are contiguous must not be mistaken for a +// gap. +func TestTrackCommitRevOutOfOrder(t *testing.T) { + ctx := context.Background() + atsync, mod := contiguityTestSync(t) + require.NoError(t, mod.UpdateRepo(syncedRepo("did:plc:race", "3lprev0000000"))) + + // Three consecutive commits, delivered at once and in no order. + events := []*indigoatproto.SyncSubscribeRepos_Commit{ + commitEvent("did:plc:race", "3lprev0000000", "3lprev0000001"), + commitEvent("did:plc:race", "3lprev0000001", "3lprev0000002"), + commitEvent("did:plc:race", "3lprev0000002", "3lprev0000003"), + } + var wg sync.WaitGroup + for _, evt := range events { + wg.Add(1) + go func() { + defer wg.Done() + atsync.trackCommitRev(ctx, evt) + }() + } + wg.Wait() + + got, err := mod.GetRepo("did:plc:race") + require.NoError(t, err) + if got.Version == "" { + // A losing interleaving costs a repair, never a record: the repair + // starts from the rev we did have. + require.NotEmpty(t, got.RepairFrom) + return + } + require.Equal(t, "3lprev0000003", got.Version) + require.True(t, got.BackfillDone) +} + +// TestRepairFloor: how far back a repair reads. The missed span starts at the +// rev we were last good at, so that -- not "one day ago" -- is where the walk +// has to start, and a repair is never shallower than a first sync. +func TestRepairFloor(t *testing.T) { + now := time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC) + at := func(d time.Duration) string { return reposync.TIDForTime(now.Add(d)) } + floorTime := func(t *testing.T, tid string) time.Time { + t.Helper() + parsed, err := reposync.TimeForTID(tid) + require.NoError(t, err) + return parsed + } + + // First contact: one day, exactly as before. + require.Equal(t, at(-InitialWindow), repairFloor("", now)) + // A rev that is not a TID tells us nothing about when it was. + require.Equal(t, at(-InitialWindow), repairFloor("not-a-tid", now)) + // A rev from inside the last day: still one day, because a repair must not + // read less than a first sync would. + require.Equal(t, at(-InitialWindow), repairFloor(at(-time.Hour), now)) + require.Equal(t, at(-InitialWindow), repairFloor(at(-23*time.Hour), now)) + + // A node that was down for a week reads from a week ago, plus the slack, + // rather than from yesterday -- everything written during those six days + // carries an rkey from those six days. + week := repairFloor(at(-7*24*time.Hour), now) + require.Equal(t, now.Add(-7*24*time.Hour-repairSlack), floorTime(t, week)) + require.True(t, floorTime(t, week).Before(now.Add(-InitialWindow))) +} + +// TestMergeBackfillState: a repair is a shallow sync of a repo that may have +// years indexed. Walking a recent window says something true about the last +// day and nothing at all about the years, so the row's history has to survive +// it -- otherwise every gap would send a completed repo back to the top of the +// deepening ladder. +func TestMergeBackfillState(t *testing.T) { + // Real TIDs, because "deeper" means "sorts earlier" and a made-up string + // would let the test agree with itself about the wrong order. + now := time.Now() + day := reposync.TIDForTime(now.Add(-InitialWindow)) + month := reposync.TIDForTime(now.Add(-30 * 24 * time.Hour)) + hour := reposync.TIDForTime(now.Add(-time.Hour)) + shallow := backfillResult{Floor: day, Done: false} + + // First contact: whatever the walk found. + floor, done := mergeBackfillState(nil, shallow) + require.Equal(t, day, floor) + require.False(t, done) + + // A repo with its whole history, repaired: still complete. + floor, done = mergeBackfillState(&model.Repo{BackfillDone: true}, shallow) + require.True(t, done) + require.Empty(t, floor, "a complete history has no watermark left to hold") + + // A repo mid-ladder keeps the deeper of the two watermarks: it really is + // indexed from the older one forward. + floor, done = mergeBackfillState(&model.Repo{BackfillFloor: month}, shallow) + require.Equal(t, month, floor) + require.False(t, done) + + // A row whose watermark is shallower than what we just walked keeps the + // fresh one; a row with no watermark at all contributes nothing. + floor, _ = mergeBackfillState(&model.Repo{BackfillFloor: hour}, shallow) + require.Equal(t, day, floor) + floor, _ = mergeBackfillState(&model.Repo{}, shallow) + require.Equal(t, day, floor) + + // The full-CAR fallback reads everything, so it completes a repo outright. + floor, done = mergeBackfillState(&model.Repo{BackfillFloor: month}, + backfillResult{Done: true}) + require.True(t, done) + require.Empty(t, floor) +} diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index 9179090d..8937623f 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -9,6 +9,7 @@ import ( "runtime" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -44,13 +45,17 @@ import ( const dedupWindow = 5 * time.Minute type ATProtoSynchronizer struct { - CLI *config.CLI - Model model.Model - StatefulDB *statedb.StatefulDB - Noter notificationpkg.Notifier - Bus *bus.Bus + CLI *config.CLI + Model model.Model + StatefulDB *statedb.StatefulDB + Noter notificationpkg.Notifier + Bus *bus.Bus + // The identity directories, built on first use behind dirMu; read them + // through [ATProtoSynchronizer.directory] rather than directly. Set them + // before the synchronizer is used and they are taken as given. PLCDirectory identity.Directory CachedPLCDirectory identity.Directory + dirMu sync.Mutex OATProxy *oatproxy.OATProxy // firehose liveness, written from every relay consumer concurrently @@ -64,6 +69,10 @@ type ATProtoSynchronizer struct { // top of StartFirehose. commitDedup *firehoseDeduper identityDedup *firehoseDeduper + + // sweeping is held for the length of a sweep, so the periodic ticker + // cannot start a second one on top of the first. + sweeping atomic.Bool } func (atsync *ATProtoSynchronizer) markSeen() { @@ -695,6 +704,12 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt log.Error(ctx, "unexpected record op kind") } } + + // Every op in this commit is indexed, so the index can claim this commit. + // Only reached on a clean pass: an event we bailed out of half-applied + // leaves the stored rev where it was, and the next commit for that repo + // notices the hole and orders a repair. + atsync.trackCommitRev(ctx, evt) } // reviveRepo un-parks a repo we had written off. A commit event is proof the diff --git a/pkg/atproto/headcheck.go b/pkg/atproto/headcheck.go new file mode 100644 index 00000000..47c2f807 --- /dev/null +++ b/pkg/atproto/headcheck.go @@ -0,0 +1,93 @@ +package atproto + +import ( + "context" + "fmt" + + "github.com/bluesky-social/indigo/xrpc" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/reposync" +) + +// headRev asks a repo's host which revision it is on. +// +// One request, nothing verified: see [reposync.LatestCommit] for why that is +// the right trade for a drift check. It goes through the same per-host lock and +// the same backoff memory as every other sync request, so a pass over thousands +// of repos is as polite to a host as a backfill is. +func (atsync *ATProtoSynchronizer) headRev(ctx context.Context, did string) (string, error) { + ident, err := atsync.resolveIdent(ctx, did, true) + if err != nil { + return "", fmt.Errorf("failed to resolve %s: %w", did, err) + } + host := ident.PDSEndpoint() + if host == "" { + return "", fmt.Errorf("no PDS endpoint found for %s", did) + } + xrpcc := &xrpc.Client{Host: host, Client: SyncHTTPClient} + + lock := pdsLocks.GetLock(host) + lock.Lock() + defer lock.Unlock() + latest, err := reposync.LatestCommit(ctx, xrpcc, did, reposync.RetryPolicy{Hints: pdsBackoffHints}) + if err != nil { + return "", err + } + return latest.Rev, nil +} + +// sweepCheck is the step that closes the reconciliation loop: it asks one +// repo's host whether the rev we hold is still its rev. +// +// Without it, a repo that finished its backfill is never looked at again, and a +// span of commits missed while this node was down -- or written before a fresh +// index started listening -- is indistinguishable from an account that has been +// quiet. With it, silence is checked once per sweep for the price of one +// request, and drift is turned into the ordinary repair the rest of the engine +// already knows how to do. +// +// It reports whether the repo should go on to its lane's ladder, which for a +// repo that is current means "if it still owes history". A repo that has +// drifted goes back to the lane's shallow queue instead, via enqueue: the +// repair has to happen before deepening means anything. +func (atsync *ATProtoSynchronizer) sweepCheck(ctx context.Context, progress *sweepProgress, enqueue func(sweepItem), step sweepStep) bool { + defer progress.checked() + + repo, err := atsync.Model.GetRepo(step.DID) + if err != nil { + log.Error(ctx, "failed to get repo", "did", step.DID, "err", err) + return false + } + if repo == nil || repo.Version == "" || repo.TerminalStatus() { + // The row moved since the plan was made -- the firehose marked it for + // repair, or it got parked. Either way the row is now right and this + // check would only ask a question somebody already answered. + return false + } + + rev, err := atsync.headRev(ctx, step.DID) + if err != nil { + if parked := parkTerminalRepo(ctx, atsync.Model, step.DID, err); parked == nil { + log.Warn(ctx, "failed to check repo head", "did", step.DID, "err", err) + } + return false + } + if rev == repo.Version { + return !repo.BackfillDone + } + + log.Log(ctx, "repo has drifted from its host", "did", step.DID, + "ourRev", repo.Version, "hostRev", rev) + marked, err := atsync.Model.MarkRepoForRepair(ctx, step.DID, repo.Version) + if err != nil { + log.Error(ctx, "failed to mark repo for repair", "did", step.DID, "err", err) + return !repo.BackfillDone + } + if !marked { + // Somebody else wedged it first; it is already on its way to a repair. + return false + } + progress.repairing() + enqueue(sweepItem{DID: step.DID, Lane: step.Lane}) + return false +} diff --git a/pkg/atproto/headcheck_test.go b/pkg/atproto/headcheck_test.go new file mode 100644 index 00000000..1736fc72 --- /dev/null +++ b/pkg/atproto/headcheck_test.go @@ -0,0 +1,99 @@ +package atproto + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/devenv" + "stream.place/streamplace/pkg/placestream" +) + +// TestHeadCheckHealsSilentGap is the test the whole reconciliation loop exists +// for. +// +// An account is indexed to completion. Then a record is written to its repo +// with nobody listening -- no firehose, no event, nothing that would ever tell +// this node the repo moved. That is not a contrived situation: it is a node +// that was down longer than a relay's replay window, and it is every account a +// freshly built index inherits. +// +// Before the head check, the record was invisible forever: the repo had a +// completed backfill, so no sweep would look at it again. After it, one +// getLatestCommit per sweep notices the disagreement, the repo repairs itself +// through the ordinary path, and the record lands -- with the history the repo +// already had still intact. +func TestHeadCheckHealsSilentGap(t *testing.T) { + dev := devenv.WithDevEnv(t) + ctx := context.Background() + atsync, mod := backfillTestSynchronizer(t, dev) + + user := dev.CreateAccount(t) + createBackfillRecord(t, user, "place.stream.chat.profile", "self", &placestream.ChatProfile{}) + createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "before")) + require.NoError(t, atsync.StatefulDB.AddRepo(user.DID)) + require.NoError(t, untilNoErrors(t, func() error { + paths, err := walkAll(ctx, dev, user.DID, backfillRanges("")) + if err != nil { + return err + } + if len(paths) != 2 { + return fmt.Errorf("PDS has %d records, want 2", len(paths)) + } + return nil + }), "waiting for the repo to settle") + + require.NoError(t, atsync.Sweep(ctx)) + indexed, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.NotEmpty(t, indexed.Version) + require.True(t, indexed.BackfillDone, "the sweep read the whole repo") + messages, err := mod.MostRecentChatMessages(user.DID) + require.NoError(t, err) + require.Len(t, messages, 1) + + // Behind our back: no firehose is running in this test, so nothing at all + // tells the index that this happened. + createBackfillRecord(t, user, "place.stream.chat.message", "", chatMessageRecord(user.DID, "after the gap")) + require.NoError(t, untilNoErrors(t, func() error { + paths, err := walkAll(ctx, dev, user.DID, backfillRanges("")) + if err != nil { + return err + } + if len(paths) != 3 { + return fmt.Errorf("PDS has %d records, want 3", len(paths)) + } + return nil + }), "waiting for the new record to commit") + + // Proof that the gap is real before we heal it. + stale, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.Equal(t, indexed.Version, stale.Version, "nothing has told the index anything") + hostRev, err := atsync.headRev(ctx, user.DID) + require.NoError(t, err) + require.NotEqual(t, stale.Version, hostRev, "the repo really did move") + + // The sweep's head-check pass finds the drift and repairs it. + require.NoError(t, atsync.Sweep(ctx)) + + healed, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.Equal(t, hostRev, healed.Version, "the repair caught the index up to the host") + require.True(t, healed.BackfillDone, "repairing a day of history does not un-index the rest") + require.Empty(t, healed.RepairFrom, "the repair it asked for is the one that ran") + messages, err = mod.MostRecentChatMessages(user.DID) + require.NoError(t, err) + require.Len(t, messages, 2, "the record written during the gap is indexed") + + // And a sweep of a node that is genuinely current is one request per repo + // and nothing else: no repair, no duplicates. + require.NoError(t, atsync.Sweep(ctx)) + current, err := mod.GetRepo(user.DID) + require.NoError(t, err) + require.Equal(t, hostRev, current.Version) + messages, err = mod.MostRecentChatMessages(user.DID) + require.NoError(t, err) + require.Len(t, messages, 2) +} diff --git a/pkg/atproto/sweep.go b/pkg/atproto/sweep.go index f8e7e079..0333bc7d 100644 --- a/pkg/atproto/sweep.go +++ b/pkg/atproto/sweep.go @@ -36,6 +36,9 @@ type sweepItem struct { // nothing but history: it starts in its lane's ladder rather than in its // lane's shallow queue. Deepen bool + // Check is set for a repo that is servable and believed current, and so + // starts with one request that asks its host whether that belief is true. + Check bool } // sweepStep is one unit of work a lane does: either the shallow sync a repo @@ -154,11 +157,16 @@ func (atsync *ATProtoSynchronizer) feedUnresolved(ctx context.Context, items []s // has finished its shallow work, a lane runs this program to completion by // itself and then gives its slot to the next host. // -// Shallow work always comes first, because a repo with no completed sync cannot -// be deepened at all, and because a repo that has just been discovered is not -// servable until it has one. Deepening is round-robin within the host, which is -// what the ladder buckets are for. +// Head checks come first, because each is one request that turns a repo we +// believe is current into one we know is current -- or into shallow work this +// lane did not know it had. Shallow work is next, because a repo with no +// completed sync cannot be deepened at all, and because a repo that has just +// been discovered is not servable until it has one. Deepening is round-robin +// within the host, which is what the ladder buckets are for. type laneProgram struct { + // check is the repos on this host whose head has not been verified against + // ours this sweep. + check []sweepItem // shallow is the repos on this host with no completed sync, oldest first. shallow []sweepItem // ladder holds the repos with history left to fetch, bucketed by how many @@ -175,9 +183,14 @@ type laneProgram struct { live bool } -// next takes the lane's next step: the oldest waiting shallow sync if there is -// one, otherwise the least-deepened repo's next window. +// next takes the lane's next step: an unchecked head if there is one, then the +// oldest waiting shallow sync, then the least-deepened repo's next window. func (p *laneProgram) next() (sweepStep, bool) { + if len(p.check) > 0 { + item := p.check[0] + p.check = p.check[1:] + return sweepStep{sweepItem: item}, true + } if len(p.shallow) > 0 { item := p.shallow[0] p.shallow = p.shallow[1:] @@ -194,13 +207,16 @@ func (p *laneProgram) next() (sweepStep, bool) { return sweepStep{}, false } -// add puts a repo into the half of the program it belongs in. +// add puts a repo into the part of the program it belongs in. func (p *laneProgram) add(item sweepItem) { - if item.Deepen { + switch { + case item.Check: + p.check = append(p.check, item) + case item.Deepen: p.push(item, 0) - return + default: + p.shallow = append(p.shallow, item) } - p.shallow = append(p.shallow, item) } // push queues a repo for its next window, having had windows of them already. @@ -211,6 +227,9 @@ func (p *laneProgram) push(item sweepItem, windows int) { return } item.Deepen = true + // Whatever this repo was doing, it is deepening now: a checked or freshly + // synced repo must not be handed back its old step class. + item.Check = false for len(p.ladder) <= windows { p.ladder = append(p.ladder, nil) } @@ -386,6 +405,7 @@ func (s *laneScheduler) abandon(prog *laneProgram) { s.mu.Lock() defer s.mu.Unlock() prog.live = false + prog.check = nil prog.shallow = nil prog.ladder = nil } @@ -400,6 +420,67 @@ func (s *laneScheduler) wait() (lanes int, err error) { return s.seen, s.ctx.Err() } +// SweepForever runs a sweep at boot and another every [config.CLI.SweepInterval] +// after that, for as long as ctx lives. +// +// Repeating is what makes the head check worth having: a repo that is current +// costs one request per interval, and one that has drifted -- because this node +// was down longer than the relay's replay window, or because a fresh index +// started listening after the accounts it inherited had already moved -- is +// found and repaired within an interval instead of never. +func (atsync *ATProtoSynchronizer) SweepForever(ctx context.Context) { + atsync.sweepLoop(ctx, atsync.sweepInterval(), func(ctx context.Context) { + atsync.sweepOnce(ctx, atsync.Sweep) + }) +} + +// sweepLoop runs run now and every interval after, until ctx ends. A +// non-positive interval runs it exactly once: the boot sweep is not optional, +// only repeating it is. +func (atsync *ATProtoSynchronizer) sweepLoop(ctx context.Context, interval time.Duration, run func(context.Context)) { + run(ctx) + if interval <= 0 || ctx.Err() != nil { + return + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run(ctx) + } + } +} + +// sweepOnce runs a sweep unless one is already running. +// +// A sweep of a large index can take longer than the interval -- a fresh node's +// first one takes hours -- and two of them at once would double every host's +// request rate while doing the same work twice. The tick is dropped rather than +// queued: the next one is another interval away, which is exactly when a sweep +// that just finished should run again. +func (atsync *ATProtoSynchronizer) sweepOnce(ctx context.Context, sweep func(context.Context) error) { + if !atsync.sweeping.CompareAndSwap(false, true) { + log.Log(ctx, "skipping scheduled sweep; the previous one is still running") + return + } + defer atsync.sweeping.Store(false) + if err := sweep(ctx); err != nil && ctx.Err() == nil { + log.Error(ctx, "backfill sweep failed", "err", err) + } +} + +// sweepInterval is how often this node re-sweeps. Zero (or negative) disables +// the ticker, leaving the boot sweep on its own. +func (atsync *ATProtoSynchronizer) sweepInterval() time.Duration { + if atsync.CLI == nil { + return config.DefaultSweepInterval + } + return atsync.CLI.SweepInterval +} + // sweepConcurrency is how many host lanes this node runs at once. func (atsync *ATProtoSynchronizer) sweepConcurrency() int { if atsync.CLI != nil && atsync.CLI.SweepConcurrency > 0 { @@ -440,19 +521,28 @@ func (atsync *ATProtoSynchronizer) Sweep(ctx context.Context) error { return err } progress := &sweepProgress{} - progress.begin(plan.shallow, plan.floors) + progress.begin(plan.shallow, plan.checks, plan.floors) stop := progress.start(ctx) defer stop() - log.Log(ctx, "sweeping repos", "shallow", plan.shallow, "deepen", len(plan.floors), - "knownHosts", laneCount(plan.ready), "unresolved", len(plan.unresolved)) + log.Log(ctx, "sweeping repos", "shallow", plan.shallow, "check", plan.checks, + "deepen", len(plan.floors), "knownHosts", laneCount(plan.ready), + "unresolved", len(plan.unresolved)) - var failed atomic.Int64 - sched := newLaneScheduler(ctx, atsync.sweepConcurrency(), func(ctx context.Context, step sweepStep) bool { - if !step.Deepen { - return atsync.sweepSync(ctx, progress, &failed, step) + var attempted, failed atomic.Int64 + // The scheduler is captured by the work it runs: a head check that finds + // drift has repair work to hand back, and hands it to the lane it is + // already running on. + var sched *laneScheduler + sched = newLaneScheduler(ctx, atsync.sweepConcurrency(), func(ctx context.Context, step sweepStep) bool { + switch { + case step.Check: + return atsync.sweepCheck(ctx, progress, sched.add, step) + case !step.Deepen: + return atsync.sweepSync(ctx, progress, &attempted, &failed, step) + default: + return atsync.sweepWindow(ctx, progress, step) } - return atsync.sweepWindow(ctx, progress, step) }) // Lanes whose host is already known start working immediately, in priority // order (own DIDs first); the rest stream in as the resolver finds them. @@ -464,8 +554,11 @@ func (atsync *ATProtoSynchronizer) Sweep(ctx context.Context) error { if err != nil { return err } - if plan.shallow > 0 && int(failed.Load()) == plan.shallow { - return fmt.Errorf("all %d repos failed to sync", plan.shallow) + // Counted rather than compared against the plan: head checks add shallow + // work as they find it, so the number of syncs a sweep tries is not known + // when it starts. + if tried := attempted.Load(); tried > 0 && failed.Load() == tried { + return fmt.Errorf("all %d repos failed to sync", tried) } log.Log(ctx, "backfill sweep complete", append([]any{"totalRepos", len(dids), "hosts", lanes}, progress.status()...)...) @@ -474,7 +567,8 @@ func (atsync *ATProtoSynchronizer) Sweep(ctx context.Context) error { // sweepSync gives a repo the shallow sync it has never had, and reports whether // it now has history to deepen. -func (atsync *ATProtoSynchronizer) sweepSync(ctx context.Context, progress *sweepProgress, failed *atomic.Int64, step sweepStep) bool { +func (atsync *ATProtoSynchronizer) sweepSync(ctx context.Context, progress *sweepProgress, attempted, failed *atomic.Int64, step sweepStep) bool { + attempted.Add(1) repo, err := atsync.SyncBlueskyRepoCached(ctx, step.DID) if err != nil { log.Error(ctx, "failed to sync repo", "did", step.DID, "err", err) @@ -607,6 +701,9 @@ type sweepPlan struct { // shallow is how many repos in total need a shallow sync, ready and // unresolved together. shallow int + // checks is how many repos are servable and believed current, and so get a + // head check before anything else happens to them. + checks int // floors is the backfill watermark of every repo that starts in a ladder, // for the status line's horizon. floors map[string]time.Time @@ -616,9 +713,11 @@ type sweepPlan struct { // // A repo row with an empty Version has never completed a sync: either brand new, // or left half-indexed by a run that died, which is the same thing as far as -// anyone reading the index is concerned. One with a Version and no BackfillDone -// has some history and wants the rest. Anything parked or complete is left -// alone. +// anyone reading the index is concerned. Anything parked is left alone. Every +// other repo -- servable, and as far as this node knows current -- starts with +// a head check, including the ones with history left to fetch: a repo whose +// recent records are wrong is not made righter by deepening it, and the check +// costs one request against the several its first window will. func (atsync *ATProtoSynchronizer) sweepPlan(dids []string) (*sweepPlan, error) { plan := &sweepPlan{floors: map[string]time.Time{}} for _, did := range dids { @@ -641,10 +740,15 @@ func (atsync *ATProtoSynchronizer) sweepPlan(dids []string) (*sweepPlan, error) } else { plan.unresolved = append(plan.unresolved, sweepItem{DID: did}) } - case repo.TerminalStatus() || repo.BackfillDone: + case repo.TerminalStatus(): default: - plan.ready = append(plan.ready, sweepItem{DID: did, Lane: sweepLane(did, repo.PDS), Deepen: true}) - plan.floors[did] = backfillFloorTime(repo.BackfillFloor) + plan.checks++ + plan.ready = append(plan.ready, sweepItem{DID: did, Lane: sweepLane(did, repo.PDS), Check: true}) + if !repo.BackfillDone { + // It joins its lane's ladder once its head checks out, but the + // horizon it holds is true from the moment the sweep starts. + plan.floors[did] = backfillFloorTime(repo.BackfillFloor) + } } } return plan, nil @@ -668,6 +772,8 @@ func backfillFloorTime(tid string) time.Time { type sweepProgress struct { mu sync.Mutex started bool + checkTotal int + checkDone int shallowTotal int shallowDone int deepenTotal int @@ -681,10 +787,12 @@ type sweepProgress struct { } // begin starts a sweep with the work its plan found. -func (p *sweepProgress) begin(shallow int, floors map[string]time.Time) { +func (p *sweepProgress) begin(shallow, checks int, floors map[string]time.Time) { p.mu.Lock() defer p.mu.Unlock() p.started = true + p.checkTotal = checks + p.checkDone = 0 p.shallowTotal = shallow p.shallowDone = 0 p.deepenTotal = len(floors) @@ -696,6 +804,22 @@ func (p *sweepProgress) begin(shallow int, floors map[string]time.Time) { } } +// checked records one repo's head having been compared with ours, however that +// went: the fraction is of checks made, so that it finishes. +func (p *sweepProgress) checked() { + p.mu.Lock() + defer p.mu.Unlock() + p.checkDone++ +} + +// repairing records a head check finding drift, which is a shallow sync this +// sweep did not know it had. +func (p *sweepProgress) repairing() { + p.mu.Lock() + defer p.mu.Unlock() + p.shallowTotal++ +} + // synced records one repo's shallow sync completing. func (p *sweepProgress) synced() { p.mu.Lock() @@ -709,7 +833,12 @@ func (p *sweepProgress) synced() { func (p *sweepProgress) laddered(did string, floor time.Time) { p.mu.Lock() defer p.mu.Unlock() - p.deepenTotal++ + if _, counted := p.floors[did]; !counted { + // A repo the plan already expected to deepen -- one that drifted, got + // repaired, and is on its way back to the ladder it never left -- is + // not a second repo. + p.deepenTotal++ + } p.floors[did] = floor } @@ -756,12 +885,18 @@ func (p *sweepProgress) horizon() int64 { func (p *sweepProgress) status() []any { p.mu.Lock() defer p.mu.Unlock() - return []any{ + var out []any + // Only a sweep with repos to check has anything to say about checking + // them, which is every sweep but a fresh node's first. + if p.checkTotal > 0 { + out = append(out, "checked", fmt.Sprintf("%d/%d", p.checkDone, p.checkTotal)) + } + return append(out, "shallow", fmt.Sprintf("%d/%d", p.shallowDone, p.shallowTotal), "deepened", fmt.Sprintf("%d/%d", p.deepenDone, p.deepenTotal), "windows", p.windows, "horizon", p.horizon(), - } + ) } // start runs the status ticker until the returned function is called, which diff --git a/pkg/atproto/sweep_test.go b/pkg/atproto/sweep_test.go index 1fdfae86..2e4fcfa8 100644 --- a/pkg/atproto/sweep_test.go +++ b/pkg/atproto/sweep_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -379,12 +380,62 @@ func indexOf(xs []string, x string) int { // stepLabel renders a step the way the lane-program tests compare them: which // repo, and whether it is the shallow sync or the nth window. func stepLabel(step sweepStep) string { + if step.Check { + return step.DID + "/check" + } if !step.Deepen { return step.DID + "/shallow" } return fmt.Sprintf("%s/window%d", step.DID, step.Windows+1) } +// TestSweepLaneProgramChecksFirst: a lane's head checks come before its other +// work, because each is one request that says whether the rest of the work on +// that repo is the right work -- a repo whose recent records are wrong is not +// made righter by deepening it. A check that finds drift adds a shallow sync +// the sweep did not know it had, and that sync still preempts the ladder. +func TestSweepLaneProgramChecksFirst(t *testing.T) { + ready := make(chan struct{}) + var mu sync.Mutex + var steps []string + var sched *laneScheduler + windows := map[string]int{} + + sched = newLaneScheduler(context.Background(), 4, func(_ context.Context, step sweepStep) bool { + <-ready + mu.Lock() + defer mu.Unlock() + steps = append(steps, stepLabel(step)) + switch { + case step.Check: + if step.DID == "drifted" { + // What sweepCheck does with drift: hand the repair back to + // this same lane, where it goes ahead of the ladder. + sched.add(sweepItem{DID: step.DID, Lane: step.Lane}) + return false + } + return true // current, and still owes history + case !step.Deepen: + return true + default: + windows[step.DID]++ + return false + } + }) + sched.add(sweepItem{DID: "current", Lane: "pds.example", Check: true}) + sched.add(sweepItem{DID: "drifted", Lane: "pds.example", Check: true}) + sched.add(sweepItem{DID: "new", Lane: "pds.example"}) + close(ready) + _, err := sched.wait() + require.NoError(t, err) + + require.Equal(t, []string{ + "current/check", "drifted/check", + "new/shallow", "drifted/shallow", + "current/window1", "new/window1", "drifted/window1", + }, steps) +} + // TestSweepLaneProgramShallowFirst: a host's repos are all made servable before // any of them is deepened, and each repo's ladder starts at the bottom rung. A // sweep that deepened one repo's history while another on the same host had @@ -696,6 +747,95 @@ func TestSweepLanesStopOnCancel(t *testing.T) { // TestSweepConcurrencyFlag: the cap comes from --sweep-concurrency, and an unset // or nonsense value is the documented default. +// TestSweepLoopRepeats: the boot sweep always runs, and after it the ticker +// keeps running them until the node goes away. That repetition is what makes +// the head check a reconciliation loop rather than a one-off. +func TestSweepLoopRepeats(t *testing.T) { + atsync := &ATProtoSynchronizer{} + + // A disabled ticker still sweeps once at boot. + var once atomic.Int64 + atsync.sweepLoop(context.Background(), 0, func(context.Context) { once.Add(1) }) + require.Equal(t, int64(1), once.Load()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runs := make(chan struct{}, 8) + done := make(chan struct{}) + go func() { + defer close(done) + atsync.sweepLoop(ctx, time.Millisecond, func(context.Context) { + select { + case runs <- struct{}{}: + default: + } + }) + }() + for i := 0; i < 3; i++ { + select { + case <-runs: + case <-time.After(10 * time.Second): + t.Fatalf("only %d sweeps ran", i) + } + } + cancel() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("the sweep loop outlived its context") + } +} + +// TestSweepOnceSkipsWhileRunning: a sweep of a large index can take longer than +// the interval, and two at once would double every host's request rate to do +// the same work twice. The tick is dropped, not queued. +func TestSweepOnceSkipsWhileRunning(t *testing.T) { + atsync := &ATProtoSynchronizer{} + ctx := context.Background() + + started := make(chan struct{}) + release := make(chan struct{}) + finished := make(chan struct{}) + go func() { + defer close(finished) + atsync.sweepOnce(ctx, func(context.Context) error { + close(started) + <-release + return nil + }) + }() + <-started + + var skipped atomic.Bool + skipped.Store(true) + atsync.sweepOnce(ctx, func(context.Context) error { + skipped.Store(false) + return nil + }) + require.True(t, skipped.Load(), "a second sweep must not start on top of the first") + + close(release) + <-finished + + // And the slot is handed back, so the next tick sweeps. + var ran atomic.Bool + atsync.sweepOnce(ctx, func(context.Context) error { + ran.Store(true) + return fmt.Errorf("a sweep that fails is logged, not fatal") + }) + require.True(t, ran.Load()) +} + +// TestSweepIntervalConfig: how often a node re-checks the repos it indexes. +func TestSweepIntervalConfig(t *testing.T) { + require.Equal(t, config.DefaultSweepInterval, (&ATProtoSynchronizer{}).sweepInterval(), + "a synchronizer without a CLI still re-sweeps") + require.Equal(t, 90*time.Minute, + (&ATProtoSynchronizer{CLI: &config.CLI{SweepInterval: 90 * time.Minute}}).sweepInterval()) + require.Equal(t, time.Duration(0), + (&ATProtoSynchronizer{CLI: &config.CLI{SweepInterval: 0}}).sweepInterval(), "0 disables the ticker") +} + func TestSweepConcurrencyFlag(t *testing.T) { require.Equal(t, config.DefaultSweepConcurrency, (&ATProtoSynchronizer{}).sweepConcurrency(), "a synchronizer without a CLI still sweeps") @@ -725,8 +865,9 @@ func TestSweepProgressStatusLine(t *testing.T) { month := time.Now().Add(-30 * 24 * time.Hour) // Three repos to make servable, one already servable and mid-ladder: the - // horizon is that one's watermark. - progress.begin(3, map[string]time.Time{"did:plc:old": week}) + // horizon is that one's watermark. Nothing to head-check, so the line does + // not mention checking -- which is a fresh node's first sweep exactly. + progress.begin(3, 0, map[string]time.Time{"did:plc:old": week}) require.Equal(t, []any{"shallow", "0/3", "deepened", "0/1", "windows", 0, "horizon", week.Unix()}, progress.status()) @@ -766,6 +907,21 @@ func TestSweepProgressStatusLine(t *testing.T) { // The ticker stops when told to, without leaking a goroutine. stop := progress.start(context.Background()) stop() + + // A warm node's sweep starts with a head check per servable repo, and says + // so until it has made all of them. A check that finds drift is a shallow + // sync this sweep did not know it had, so the denominator grows. + var warm sweepProgress + warm.begin(1, 2, nil) + require.Equal(t, + []any{"checked", "0/2", "shallow", "0/1", "deepened", "0/0", "windows", 0, "horizon", int64(0)}, + warm.status()) + warm.checked() + warm.checked() + warm.repairing() + require.Equal(t, + []any{"checked", "2/2", "shallow", "0/2", "deepened", "0/0", "windows", 0, "horizon", int64(0)}, + warm.status()) } // walkAll walks a repo's ranges against the dev PDS and returns the paths, so diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 429ec82d..760cb75f 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -553,6 +553,14 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } go atsync.Bus.Publish(userDID, rec) + if isFirstSync { + // A backfill reads history, and a teleport out of history has + // already happened: announcing it would tell a streamer somebody is + // arriving who arrived last year. The record is indexed either way; + // only the announcement is a live-only thing. + return nil + } + // schedule arrival notification 10 seconds after startsAt arrivalTime := startsAt.Add(10 * time.Second) waitDuration := time.Until(arrivalTime) diff --git a/pkg/atproto/teleport_test.go b/pkg/atproto/teleport_test.go new file mode 100644 index 00000000..9eb90b71 --- /dev/null +++ b/pkg/atproto/teleport_test.go @@ -0,0 +1,90 @@ +package atproto + +import ( + "bytes" + "context" + "sync" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/bus" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/spid" +) + +// TestTeleportArrivalNotFromBackfill: a teleport record indexed by a backfill +// is history, and history does not arrive. +// +// The arrival notification is scheduled for ten seconds after the teleport +// starts, and a teleport from last week is already past that, so a fresh index +// reading an account's repo would announce every teleport it has ever done, all +// at once, to the streamers they pointed at. Indexing the record is right; +// announcing it is not. +func TestTeleportArrivalNotFromBackfill(t *testing.T) { + ctx := context.Background() + atsync, mod, b := offlineSynchronizer(t) + + traveller := "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" + streamer := "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb" + require.NoError(t, mod.UpdateRepo(&model.Repo{ + DID: traveller, + Handle: "traveller.test", + PDS: "http://127.0.0.1:1", + Version: "3lrev00000000", + })) + + // Watch the streamer's topic, which is where an arrival is announced. + ch := b.Subscribe(streamer) + defer b.Unsubscribe(streamer, ch) + var mu sync.Mutex + var arrivals []bus.Message + go func() { + for msg := range ch { + mu.Lock() + arrivals = append(arrivals, msg) + mu.Unlock() + } + }() + countArrivals := func() int { + mu.Lock() + defer mu.Unlock() + return len(arrivals) + } + + duration := int64(600) + index := func(rkey, startsAt string, isFirstSync bool) { + t.Helper() + rec := &placestream.LiveTeleport{ + LexiconTypeID: "place.stream.live.teleport", + Streamer: streamer, + StartsAt: startsAt, + DurationSeconds: &duration, + } + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + recCBOR := buf.Bytes() + rcid, err := spid.GetCID(rec) + require.NoError(t, err) + require.NoError(t, atsync.handleCreateUpdate(ctx, traveller, syntax.RecordKey(rkey), + &recCBOR, rcid.String(), syntax.NSID("place.stream.live.teleport"), false, isFirstSync)) + } + + // A teleport from last week, met during a backfill. The notification is + // scheduled with no wait at all, so if it were scheduled we would see it. + past := time.Now().Add(-7 * 24 * time.Hour).UTC().Format(time.RFC3339) + index("3lteleportold0", past, true) + time.Sleep(250 * time.Millisecond) + require.Equal(t, 0, countArrivals(), "a backfilled teleport must not announce an arrival") + stored, err := mod.GetTeleportByURI("at://" + traveller + "/place.stream.live.teleport/3lteleportold0") + require.NoError(t, err) + require.NotNil(t, stored, "the record is still indexed; only the announcement is live-only") + + // The same record arriving live still announces: this is a guard on + // backfills, not a change to what a teleport does. + index("3lteleportnew0", past, false) + require.Eventually(t, func() bool { return countArrivals() == 1 }, 5*time.Second, 10*time.Millisecond, + "a live teleport still announces an arrival") +} diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index cfaf21c3..d1ee2c2e 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -268,16 +268,14 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu Noter: noter, Bus: b, } - // Sync every repo we know about, once per boot: a repair pass for repos left - // half-indexed by a previous run, and then history deepening, which on a - // fresh node runs for as long as the network is big. Nothing below depends - // on it, so it runs in the background off the serve context -- shutdown - // cancels it -- and the node is up and serving in the meantime. - go func() { - if err := atsync.Sweep(ctx); err != nil && ctx.Err() == nil { - log.Error(ctx, "backfill sweep failed", "err", err) - } - }() + // Sync every repo we know about, at boot and every --sweep-interval after: + // a repair pass for repos left half-indexed by a previous run, a head check + // that finds the ones that drifted while we were not listening, and history + // deepening, which on a fresh node runs for as long as the network is big. + // Nothing below depends on it, so it runs in the background off the serve + // context -- shutdown cancels it -- and the node is up and serving in the + // meantime. + go atsync.SweepForever(ctx) mm, err := media.MakeMediaManager(ctx, cli, signer, mod, b, atsync, ldb) if err != nil { diff --git a/pkg/config/config.go b/pkg/config/config.go index 6bc0134c..8df34271 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -171,8 +171,20 @@ type CLI struct { VODConcurrency int MaximumLiveBitrate int SweepConcurrency int + SweepInterval time.Duration } +// DefaultSweepInterval is how often the atproto sweep re-runs when +// --sweep-interval is unset. +// +// The sweep's first pass over a repo that is up to date is a single +// getLatestCommit, so this is a per-repo request budget: six hours means an +// indexed account is asked about four times a day, and drift -- a gap in the +// firehose, a span missed while this node was down -- is found and repaired +// within that. Any lower buys hours of detection latency for a proportional +// increase in traffic against every PDS on the network. +const DefaultSweepInterval = 6 * time.Hour + // DefaultSweepConcurrency is how many PDS hosts the atproto backfill sweep // works on at once when --sweep-concurrency is unset or zero. // @@ -829,6 +841,13 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.SweepConcurrency, Sources: urfavecli.EnvVars("SP_SWEEP_CONCURRENCY"), }, + &urfavecli.DurationFlag{ + Name: "sweep-interval", + Usage: "how often to re-run the atproto sweep, which asks every indexed repo's host whether our copy is still current and repairs the ones that are not. 0 disables re-running; the sweep at startup always happens", + Value: DefaultSweepInterval, + Destination: &cli.SweepInterval, + Sources: urfavecli.EnvVars("SP_SWEEP_INTERVAL"), + }, &urfavecli.StringFlag{ Name: "maximum-live-bitrate", Usage: "maximum allowed live ingest bitrate, measured per emitted segment. Accepts a bits-per-second number or a decimal SI suffix — e.g. 30M, 30000k, or 30000000 (all 30 Mbps). A stream whose bitrate exceeds this (plus a 10% margin) is disconnected and the streamer is shown a problem. 0 = unlimited", diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 9771eabc..97f02c0d 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -3,11 +3,36 @@ package config import ( "context" "testing" + "time" "github.com/stretchr/testify/require" urfavecli "github.com/urfave/cli/v3" ) +// flagRun builds a command the way `streamplace sync` is built -- the command +// an operator uses to warm an index -- and runs it with the given arguments, +// handing back the CLI the flags landed in. +func flagRun(t *testing.T, args ...string) *CLI { + t.Helper() + cli := &CLI{} + cmd := cli.NewCommand("sync") + cmd.Action = func(context.Context, *urfavecli.Command) error { return nil } + require.NoError(t, cmd.Run(context.Background(), append([]string{"sync"}, args...))) + return cli +} + +// TestSweepIntervalFlag: how often a node re-checks every repo it indexes is an +// operator's decision -- and setting it to zero, which turns the periodic sweep +// off entirely, has to be expressible. +func TestSweepIntervalFlag(t *testing.T) { + require.Equal(t, DefaultSweepInterval, flagRun(t).SweepInterval, "unset is the default") + require.Equal(t, 90*time.Minute, flagRun(t, "--sweep-interval", "90m").SweepInterval) + require.Equal(t, time.Duration(0), flagRun(t, "--sweep-interval=0").SweepInterval) + + t.Setenv("SP_SWEEP_INTERVAL", "2h") + require.Equal(t, 2*time.Hour, flagRun(t).SweepInterval) +} + // TestSweepConcurrencyFlag: the sweep's host-lane cap is settable from the // command line and the environment, and every command built from NewCommand -- // including `streamplace sync`, which is the one an operator uses to warm an diff --git a/pkg/model/model.go b/pkg/model/model.go index 63b8429d..49e59646 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -39,6 +39,8 @@ type Model interface { SearchReposByHandle(query string, limit int) ([]Repo, error) UpdateRepo(repo *Repo) error AdvanceRepoBackfill(ctx context.Context, did, version, rootCID, floor string, done bool) error + AdvanceRepoVersion(ctx context.Context, did, from, to string) (bool, error) + MarkRepoForRepair(ctx context.Context, did, from string) (bool, error) SetRepoStatus(ctx context.Context, did string, status string) error TerminalRepoDIDs(ctx context.Context) ([]string, error) @@ -220,6 +222,9 @@ func MakeDB(dbURL string) (Model, error) { if err != nil { return nil, fmt.Errorf("error setting journal mode: %w", err) } + if err := SetSQLiteBusyTimeout(db); err != nil { + return nil, err + } err = db.Use(prometheus.New(prometheus.Config{ DBName: "index", @@ -277,3 +282,23 @@ func MakeDB(dbURL string) (Model, error) { } return &DBModel{DB: db}, nil } + +// SQLiteBusyTimeout is how long a sqlite connection waits for a lock another +// process holds before giving up with SQLITE_BUSY. +// +// Within one process the single connection (SetMaxOpenConns(1)) serializes +// everything, so this is entirely about the second process: `streamplace sync` +// warms a new index revision while the server runs, and a writer that meets a +// checkpointing writer must wait rather than fail the query. +const SQLiteBusyTimeout = 5 * time.Second + +// SetSQLiteBusyTimeout applies [SQLiteBusyTimeout] to an open sqlite database. +// It is a per-connection setting, which is why it is set on the pool rather +// than being part of the DSN nothing else in here uses. +func SetSQLiteBusyTimeout(db *gorm.DB) error { + ms := SQLiteBusyTimeout.Milliseconds() + if err := db.Exec(fmt.Sprintf("PRAGMA busy_timeout = %d;", ms)).Error; err != nil { + return fmt.Errorf("error setting busy timeout: %w", err) + } + return nil +} diff --git a/pkg/model/repo.go b/pkg/model/repo.go index e5d7c988..64ef1f5b 100644 --- a/pkg/model/repo.go +++ b/pkg/model/repo.go @@ -35,6 +35,13 @@ type Repo struct { // BackfillDone reports that those windowed collections are indexed all the // way back to the start of the repo, so there is no history left to fetch. BackfillDone bool `gorm:"column:backfill_done" json:"backfillDone,omitempty"` + // RepairFrom is the revision this repo was known good at when drift was + // detected -- a firehose commit that did not follow our rev, or a head + // check that disagreed with it. Marking a repo for repair clears Version + // (the wedge every repair path already keys on), which would otherwise + // throw away the one fact the repair needs: where the missed span starts. + // Empty for a repo that has never been marked. + RepairFrom string `gorm:"column:repair_from" json:"repairFrom,omitempty"` } // TerminalStatus reports whether this repo is in an account state no amount of @@ -133,6 +140,57 @@ func (m *DBModel) AdvanceRepoBackfill(ctx context.Context, did, version, rootCID }).Error } +// AdvanceRepoVersion moves a repo's revision from one value to another, and +// only from that value: it is a compare-and-swap, and it reports whether it +// applied. +// +// The firehose hands events to a goroutine each, so nothing orders two commits +// on one repo. A CAS makes that harmless -- the event whose Since matches the +// stored rev is by definition the next one, and every other outcome is decided +// by re-reading the row rather than by whichever write landed last. +// +// An empty from is refused rather than executed: an empty Version is the wedge +// that means "this repo is being backfilled, or needs to be", and quietly +// filling it in from an event would un-wedge a repair nobody has done yet. +func (m *DBModel) AdvanceRepoVersion(ctx context.Context, did, from, to string) (bool, error) { + if from == "" || to == "" { + return false, nil + } + res := m.DB.WithContext(ctx).Model(&Repo{}). + Where("did = ? AND version = ?", did, from). + Select("Version").Updates(Repo{Version: to}) + if res.Error != nil { + return false, res.Error + } + return res.RowsAffected > 0, nil +} + +// MarkRepoForRepair records that this repo's index no longer matches its host: +// it clears Version -- the wedge that makes every existing repair path (the +// cached-sync fall-through, the sweep's plan) pick the repo up -- and remembers +// the rev it was last known good at in RepairFrom. +// +// Only Version and RepairFrom are written. The rest of the row is history the +// repair must not lose: the backfill watermark says how far back this repo is +// indexed, and a repair walks a recent window, so blanking it would send a +// completed repo back to the top of the deepening ladder. +// +// It is a compare-and-swap on from, so a repo somebody else has already wedged +// (or has since advanced past) is left alone, and it reports whether it applied. +func (m *DBModel) MarkRepoForRepair(ctx context.Context, did, from string) (bool, error) { + if from == "" { + return false, nil + } + res := m.DB.WithContext(ctx).Model(&Repo{}). + Where("did = ? AND version = ?", did, from). + Select("Version", "RepairFrom"). + Updates(Repo{Version: "", RepairFrom: from}) + if res.Error != nil { + return false, res.Error + } + return res.RowsAffected > 0, nil +} + // TerminalRepoDIDs lists the repos parked in a terminal account state, so the // boot-time sync sweep can skip them in one query instead of failing on each. func (m *DBModel) TerminalRepoDIDs(ctx context.Context) ([]string, error) { diff --git a/pkg/model/repo_test.go b/pkg/model/repo_test.go new file mode 100644 index 00000000..4869155c --- /dev/null +++ b/pkg/model/repo_test.go @@ -0,0 +1,161 @@ +package model + +import ( + "context" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +// fullRepo is a row with something in every column the sync engine cares +// about, so that a test asserting "only this column moved" means it. +func fullRepo(did string) *Repo { + return &Repo{ + DID: did, + Handle: "someone.example", + PDS: "https://pds.example", + Version: "3lprev0000000", + RootCID: "bafyreiabc", + BackfillFloor: "3lpfloor00000", + BackfillDone: true, + } +} + +// TestAdvanceRepoVersion is the compare-and-swap the firehose's contiguity +// check rests on: it moves the rev only from the value the caller saw, and it +// says whether it did. +func TestAdvanceRepoVersion(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + require.NoError(t, db.UpdateRepo(fullRepo("did:plc:a"))) + + // The rev this event follows is not ours: nothing happens, and the caller + // is told so rather than left to assume. + applied, err := db.AdvanceRepoVersion(ctx, "did:plc:a", "3lpsomethingelse", "3lpnext000000") + require.NoError(t, err) + require.False(t, applied) + got, err := db.GetRepo("did:plc:a") + require.NoError(t, err) + require.Equal(t, "3lprev0000000", got.Version) + + applied, err = db.AdvanceRepoVersion(ctx, "did:plc:a", "3lprev0000000", "3lpnext000000") + require.NoError(t, err) + require.True(t, applied) + + // Only the rev moved. Everything else is the sync state a repair would + // otherwise have to rebuild. + got, err = db.GetRepo("did:plc:a") + require.NoError(t, err) + require.Equal(t, "3lpnext000000", got.Version) + require.Equal(t, "bafyreiabc", got.RootCID, "root_c_id is not a column to lose") + require.Equal(t, "3lpfloor00000", got.BackfillFloor) + require.True(t, got.BackfillDone) + require.Equal(t, "someone.example", got.Handle) + + // The same event again -- a redelivery from a second relay -- is a no-op. + applied, err = db.AdvanceRepoVersion(ctx, "did:plc:a", "3lprev0000000", "3lpnext000000") + require.NoError(t, err) + require.False(t, applied) + + // An empty from is the wedge that means "this repo needs a backfill". + // Filling it in from an event would cancel a repair nobody has done. + require.NoError(t, db.UpdateRepo(&Repo{DID: "did:plc:wedged", Version: ""})) + applied, err = db.AdvanceRepoVersion(ctx, "did:plc:wedged", "", "3lpnext000000") + require.NoError(t, err) + require.False(t, applied) + got, err = db.GetRepo("did:plc:wedged") + require.NoError(t, err) + require.Empty(t, got.Version, "a wedged repo stays wedged") + + // A repo we have never heard of is not created by an event. + applied, err = db.AdvanceRepoVersion(ctx, "did:plc:stranger", "3lprev0000000", "3lpnext000000") + require.NoError(t, err) + require.False(t, applied) + got, err = db.GetRepo("did:plc:stranger") + require.NoError(t, err) + require.Nil(t, got) +} + +// TestAdvanceRepoVersionRace: firehose events are handled a goroutine each, so +// commits on one repo race. Exactly one of them may win each hop, and the row +// must end up on the chain rather than wherever the last writer happened to be. +func TestAdvanceRepoVersionRace(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + require.NoError(t, db.UpdateRepo(&Repo{DID: "did:plc:a", Version: "3lprev0000000"})) + + const racers = 16 + var wg sync.WaitGroup + var mu sync.Mutex + winners := 0 + start := make(chan struct{}) + for i := 0; i < racers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + applied, err := db.AdvanceRepoVersion(ctx, "did:plc:a", "3lprev0000000", "3lpnext000000") + if err != nil { + t.Error(err) + return + } + mu.Lock() + defer mu.Unlock() + if applied { + winners++ + } + }() + } + close(start) + wg.Wait() + + require.Equal(t, 1, winners, "one commit follows a given rev, so one CAS applies") + got, err := db.GetRepo("did:plc:a") + require.NoError(t, err) + require.Equal(t, "3lpnext000000", got.Version) +} + +// TestMarkRepoForRepair: the mark reuses the wedge every repair path already +// keys on, and must not take the repo's history down with it. +func TestMarkRepoForRepair(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + require.NoError(t, db.UpdateRepo(fullRepo("did:plc:a"))) + + marked, err := db.MarkRepoForRepair(ctx, "did:plc:a", "3lprev0000000") + require.NoError(t, err) + require.True(t, marked) + + got, err := db.GetRepo("did:plc:a") + require.NoError(t, err) + require.Empty(t, got.Version, "the wedge is what makes the sweep pick it up") + require.Equal(t, "3lprev0000000", got.RepairFrom, "where the missed span starts") + require.Equal(t, "bafyreiabc", got.RootCID) + require.Equal(t, "3lpfloor00000", got.BackfillFloor) + require.True(t, got.BackfillDone, "a gap in the last hour does not un-index five years") + require.Equal(t, "someone.example", got.Handle) + require.Equal(t, RepoStatusOK, got.Status) + + // It is a CAS too: a row somebody already wedged, or already moved past, + // is left exactly as they left it. + marked, err = db.MarkRepoForRepair(ctx, "did:plc:a", "3lprev0000000") + require.NoError(t, err) + require.False(t, marked) + got, err = db.GetRepo("did:plc:a") + require.NoError(t, err) + require.Equal(t, "3lprev0000000", got.RepairFrom) + + marked, err = db.MarkRepoForRepair(ctx, "did:plc:a", "") + require.NoError(t, err) + require.False(t, marked, "there is nothing to repair from") +} + +// TestSQLiteBusyTimeout: the pragma is per-connection, so the only proof it +// took is asking the connection. +func TestSQLiteBusyTimeout(t *testing.T) { + db := indexedTestDB(t) + var timeout int + require.NoError(t, db.DB.Raw("PRAGMA busy_timeout").Scan(&timeout).Error) + require.Equal(t, int(SQLiteBusyTimeout.Milliseconds()), timeout) +} diff --git a/pkg/reposync/head.go b/pkg/reposync/head.go index 000228ac..0348d8d2 100644 --- a/pkg/reposync/head.go +++ b/pkg/reposync/head.go @@ -26,6 +26,38 @@ type Head struct { Rev string } +// LatestCommit asks a host which commit a repo is on, and nothing more: one +// request, no blocks fetched, no signature checked. +// +// The answer is therefore the host's word rather than proof. That is enough to +// tell "our index is at the same rev as the host" from "it is not", which is +// all a drift check needs -- and a check that finds drift hands the repo to the +// fully verified walk in [FetchVerifiedHead], so nothing gets indexed on the +// strength of this call. +// +// At most one retry policy may be given; omitting it uses the package defaults. +func LatestCommit(ctx context.Context, client *xrpc.Client, did string, retry ...RetryPolicy) (*indigoat.SyncGetLatestCommit_Output, error) { + if len(retry) > 1 { + return nil, fmt.Errorf("at most one retry policy, got %d", len(retry)) + } + var policy RetryPolicy + if len(retry) == 1 { + policy = retry[0] + } + policy = policy.forHost(client.Host) + + var latest *indigoat.SyncGetLatestCommit_Output + err := policy.do(ctx, "com.atproto.sync.getLatestCommit "+did, func() error { + var err error + latest, err = indigoat.SyncGetLatestCommit(ctx, client, did) + return err + }) + if err != nil { + return nil, fmt.Errorf("com.atproto.sync.getLatestCommit for %s: %w", did, err) + } + return latest, nil +} + // FetchVerifiedHead resolves a repo's current commit and proves it belongs to // did. // @@ -52,14 +84,9 @@ func FetchVerifiedHead(ctx context.Context, client *xrpc.Client, f BlockFetcher, return nil, fmt.Errorf("invalid did %q: %w", did, err) } - var latest *indigoat.SyncGetLatestCommit_Output - err = policy.do(ctx, "com.atproto.sync.getLatestCommit "+did, func() error { - var err error - latest, err = indigoat.SyncGetLatestCommit(ctx, client, did) - return err - }) + latest, err := LatestCommit(ctx, client, did, policy) if err != nil { - return nil, fmt.Errorf("com.atproto.sync.getLatestCommit for %s: %w", did, err) + return nil, err } commitCID, err := cid.Decode(latest.Cid) if err != nil { diff --git a/pkg/statedb/statedb.go b/pkg/statedb/statedb.go index ea1d8f94..a3cbda04 100644 --- a/pkg/statedb/statedb.go +++ b/pkg/statedb/statedb.go @@ -107,9 +107,8 @@ func MakeDB(ctx context.Context, cli *config.CLI, noter notificationpkg.Notifier } } if dbType == DBTypeSQLite { - err = db.Exec("PRAGMA journal_mode=WAL;").Error - if err != nil { - return nil, fmt.Errorf("error setting journal mode: %w", err) + if err := sqlitePragmas(db); err != nil { + return nil, err } sqlDB, err := db.DB() if err != nil { @@ -151,6 +150,18 @@ func MakeDB(ctx context.Context, cli *config.CLI, noter notificationpkg.Notifier return state, nil } +// sqlitePragmas applies the two settings a sqlite state database needs: WAL, so +// readers do not block the writer, and a busy timeout, so a writer in another +// process (`streamplace sync`, warming a new index) is waited for instead of +// erroring out. It is a function rather than two lines in MakeDB because +// MakeDB's `model` parameter shadows the package the timeout lives in. +func sqlitePragmas(db *gorm.DB) error { + if err := db.Exec("PRAGMA journal_mode=WAL;").Error; err != nil { + return fmt.Errorf("error setting journal mode: %w", err) + } + return model.SetSQLiteBusyTimeout(db) +} + func openDB(dial gorm.Dialector) (*gorm.DB, error) { return gorm.Open(dial, &gorm.Config{ SkipDefaultTransaction: true, diff --git a/pkg/statedb/statedb_test.go b/pkg/statedb/statedb_test.go index 4f0fb942..a7752d73 100644 --- a/pkg/statedb/statedb_test.go +++ b/pkg/statedb/statedb_test.go @@ -42,3 +42,20 @@ func WithAllDatabases(t *testing.T, f func(*StatefulDB)) { }) } } + +// TestSQLiteBusyTimeout: the state database is the one two streamplace +// processes share -- the server, and a `streamplace sync` warming a new index +// revision -- so a writer that meets the other's lock has to wait rather than +// fail. The pragma is per-connection, so the only proof it took is asking the +// connection. +func TestSQLiteBusyTimeout(t *testing.T) { + cli := config.CLI{DBURL: ":memory:"} + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := MakeDB(t.Context(), &cli, nil, mod) + require.NoError(t, err) + + var timeout int + require.NoError(t, state.DB.Raw("PRAGMA busy_timeout").Scan(&timeout).Error) + require.Equal(t, int(model.SQLiteBusyTimeout.Milliseconds()), timeout) +} -- 2.51.2