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)
+}