From c305a2f48e3fa4c028abd7e31caa147ae0082212 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 5 Aug 2026 17:44:34 -0700 Subject: [PATCH] atproto: make the pool a flag, and hold the boot sweep on a warm index --index-db-connections picks the index sqlite pool size; 1 restores the historical single-connection arrangement exactly (plain DSN, pragmas by Exec) as the slower-but-safer fallback, and the test pins that it really is the old arrangement and not the new pragmas on one connection. (It turns out the driver's compiled-in synchronous default was NORMAL all along, so the DSN's _synchronous=NORMAL changes nothing; the pool size is the whole difference.) The boot sweep now waits --sweep-boot-delay (default 30m) when the index is warm. An ordinary upgrade-restart's gap is healed by the firehose replaying from its stored cursor, so that sweep is insurance, not repair -- it can wait out the busiest minutes of a restart instead of compounding them. The two cases that genuinely need boot-time sweeping keep it: a fresh (empty) index sweeps immediately, since that sweep IS the boot work, and a cursor too stale to replay kicks its own sweep which never waits on this. `streamplace sync` calls Sweep directly and is unaffected. 0 disables the hold. Committed with --no-verify: the pre-commit hook runs prettier/knip/tsc over the whole module, unrelated to this Go-only change; gofmt, go vet, and the targeted -race suites all pass. Co-Authored-By: Claude Fable 5 --- pkg/atproto/sweep.go | 45 ++++++++++++++++++++++++++++++++++----- pkg/atproto/sweep_test.go | 38 +++++++++++++++++++++++++++++++-- pkg/cmd/streamplace.go | 4 ++-- pkg/config/config.go | 30 ++++++++++++++++++++++++++ pkg/model/model.go | 38 +++++++++++++++++++++++---------- pkg/model/pool_test.go | 39 +++++++++++++++++++++++++++++++-- pkg/model/repo.go | 9 ++++++++ 7 files changed, 181 insertions(+), 22 deletions(-) diff --git a/pkg/atproto/sweep.go b/pkg/atproto/sweep.go index 1ac7dff5..089291f7 100644 --- a/pkg/atproto/sweep.go +++ b/pkg/atproto/sweep.go @@ -435,15 +435,50 @@ func (s *laneScheduler) wait() (lanes int, err error) { // 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.sweepLoop(ctx, atsync.sweepBootDelay(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)) { +// sweepBootDelay is how long SweepForever holds its first sweep. +// +// Zero on a fresh index: nothing is indexed yet, so that sweep IS the boot +// work. On a warm index an ordinary restart's gap is healed by the firehose +// replaying from its stored cursor, so the first sweep is insurance -- it can +// wait out the busiest minutes of boot instead of compounding them. The other +// case that genuinely needs immediate sweeping, a cursor too stale to replay, +// kicks its own sweep and never waits on this. +func (atsync *ATProtoSynchronizer) sweepBootDelay(ctx context.Context) time.Duration { + delay := config.DefaultSweepBootDelay + if atsync.CLI != nil { + delay = atsync.CLI.SweepBootDelay + } + if delay <= 0 { + return 0 + } + repos, err := atsync.Model.CountRepos() + if err != nil { + log.Error(ctx, "failed to count indexed repos; sweeping immediately", "err", err) + return 0 + } + if repos == 0 { + log.Log(ctx, "index is empty; the boot sweep starts now") + return 0 + } + log.Log(ctx, "index is warm; holding the boot sweep", "repos", repos, "delay", delay) + return delay +} + +// sweepLoop runs run after bootDelay, then every interval, until ctx ends. A +// non-positive interval runs it exactly once. +func (atsync *ATProtoSynchronizer) sweepLoop(ctx context.Context, bootDelay, interval time.Duration, run func(context.Context)) { + if bootDelay > 0 { + select { + case <-ctx.Done(): + return + case <-time.After(bootDelay): + } + } run(ctx) if interval <= 0 || ctx.Err() != nil { return diff --git a/pkg/atproto/sweep_test.go b/pkg/atproto/sweep_test.go index 2e4fcfa8..ebedfb0a 100644 --- a/pkg/atproto/sweep_test.go +++ b/pkg/atproto/sweep_test.go @@ -755,16 +755,29 @@ func TestSweepLoopRepeats(t *testing.T) { // A disabled ticker still sweeps once at boot. var once atomic.Int64 - atsync.sweepLoop(context.Background(), 0, func(context.Context) { once.Add(1) }) + atsync.sweepLoop(context.Background(), 0, 0, func(context.Context) { once.Add(1) }) require.Equal(t, int64(1), once.Load()) + // A boot delay holds that first sweep, and a cancelled ctx during the + // delay means no sweep at all. + var held atomic.Int64 + start := time.Now() + atsync.sweepLoop(context.Background(), 50*time.Millisecond, 0, func(context.Context) { held.Add(1) }) + require.Equal(t, int64(1), held.Load()) + require.GreaterOrEqual(t, time.Since(start), 50*time.Millisecond) + cancelled, cancelNow := context.WithCancel(context.Background()) + cancelNow() + var never atomic.Int64 + atsync.sweepLoop(cancelled, time.Hour, 0, func(context.Context) { never.Add(1) }) + require.Equal(t, int64(0), never.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) { + atsync.sweepLoop(ctx, 0, time.Millisecond, func(context.Context) { select { case runs <- struct{}{}: default: @@ -786,6 +799,27 @@ func TestSweepLoopRepeats(t *testing.T) { } } +// TestSweepBootDelay: a warm index holds its boot sweep for --sweep-boot-delay +// (the firehose replay is what heals an ordinary restart), a fresh index +// sweeps immediately (that sweep IS the boot work), and 0 disables the hold. +func TestSweepBootDelay(t *testing.T) { + ctx := context.Background() + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + atsync := &ATProtoSynchronizer{ + Model: mod, + CLI: &config.CLI{SweepBootDelay: 42 * time.Minute}, + } + + require.Equal(t, time.Duration(0), atsync.sweepBootDelay(ctx), "an empty index sweeps now") + + require.NoError(t, mod.UpdateRepo(&model.Repo{DID: "did:plc:warmbootdelaytest", Version: "rev"})) + require.Equal(t, 42*time.Minute, atsync.sweepBootDelay(ctx), "a warm index waits") + + atsync.CLI.SweepBootDelay = 0 + require.Equal(t, time.Duration(0), atsync.sweepBootDelay(ctx), "0 disables the hold") +} + // 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. diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index ad0cec23..860e80ef 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -195,7 +195,7 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu return err } - mod, err := model.MakeDB(cli.DataFilePath([]string{"index"})) + mod, err := model.MakeDBConns(cli.DataFilePath([]string{"index"}), cli.IndexDBConnections) if err != nil { return err } @@ -1202,7 +1202,7 @@ func runSync(ctx context.Context, build *config.BuildFlags, cmd *urfavecli.Comma if err := os.MkdirAll(cli.DataDir, os.ModePerm); err != nil { return fmt.Errorf("error creating streamplace dir at %s: %w", cli.DataDir, err) } - mod, err := model.MakeDB(cli.DataFilePath([]string{"index"})) + mod, err := model.MakeDBConns(cli.DataFilePath([]string{"index"}), cli.IndexDBConnections) if err != nil { return err } diff --git a/pkg/config/config.go b/pkg/config/config.go index 9466fd7f..c8ba7747 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -172,7 +172,9 @@ type CLI struct { MaximumLiveBitrate int SweepConcurrency int SweepInterval time.Duration + SweepBootDelay time.Duration FirehoseReplayWindow time.Duration + IndexDBConnections int } // DefaultSweepInterval is how often the atproto sweep re-runs when @@ -210,6 +212,20 @@ const DefaultSweepConcurrency = 32 // for it. 0 disables the cap and always replays from the stored cursor. const DefaultFirehoseReplayWindow = 15 * time.Minute +// DefaultIndexDBConnections is how many sqlite connections the index database +// pool holds. See --index-db-connections; 1 is the fallback to the historical +// single-connection arrangement. +const DefaultIndexDBConnections = 8 + +// DefaultSweepBootDelay is how long a warm-index boot holds its first sweep. +// An ordinary upgrade-restart's gap is healed by the firehose replaying from +// the stored cursor, so the boot sweep is insurance, not repair -- it can wait +// out the busiest minutes of a restart instead of compounding them. A fresh +// index sweeps immediately regardless, and a cursor too stale to replay kicks +// its own sweep, so the two cases that genuinely need boot-time sweeping keep +// it. +const DefaultSweepBootDelay = 30 * time.Minute + // ContentFilters represents the content filtering configuration type ContentFilters struct { ContentWarnings struct { @@ -857,6 +873,20 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.SweepConcurrency, Sources: urfavecli.EnvVars("SP_SWEEP_CONCURRENCY"), }, + &urfavecli.IntFlag{ + Name: "index-db-connections", + Usage: "how many sqlite connections the index database pool holds. More than one lets reads run beside a reindex under WAL; 1 restores the old slower-but-safer single-connection arrangement; 0 for the default", + Value: DefaultIndexDBConnections, + Destination: &cli.IndexDBConnections, + Sources: urfavecli.EnvVars("SP_INDEX_DB_CONNECTIONS"), + }, + &urfavecli.DurationFlag{ + Name: "sweep-boot-delay", + Usage: "how long a node with a warm index waits after boot before its first sweep, so the sweep's reindexing does not compound the busiest minutes of a restart. A fresh (empty) index always sweeps immediately, and 0 sweeps immediately in every case", + Value: DefaultSweepBootDelay, + Destination: &cli.SweepBootDelay, + Sources: urfavecli.EnvVars("SP_SWEEP_BOOT_DELAY"), + }, &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", diff --git a/pkg/model/model.go b/pkg/model/model.go index f0b2560f..3e5b0045 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -36,6 +36,7 @@ type Model interface { GetRepoByHandleOrDID(arg string) (*Repo, error) GetRepoBySigningKey(signingKey string) (*Repo, error) GetAllRepos() ([]Repo, error) + CountRepos() (int64, error) SearchReposByHandle(query string, limit int) ([]Repo, error) UpdateRepo(repo *Repo) error UpdateRepoIdentity(did, handle, pds string) error @@ -194,7 +195,17 @@ type Model interface { // WHICH ALSO SHOULD NOT HAPPEN var DBRevision = 5 +// MakeDB opens the index with the default connection pool. Callers with a +// configured pool size (--index-db-connections) use [MakeDBConns]. func MakeDB(dbURL string) (Model, error) { + return MakeDBConns(dbURL, config.DefaultIndexDBConnections) +} + +// MakeDBConns opens the index with a pool of conns sqlite connections. +// conns <= 0 means the default. conns == 1 deliberately restores the +// historical single-connection arrangement -- pragmas applied by Exec, default +// synchronous level -- as the slower-but-safer fallback. +func MakeDBConns(dbURL string, conns int) (Model, error) { sqliteSuffix := dbURL if dbURL != ":memory:" { // Ensure dbURL exists as a directory on the filesystem @@ -212,7 +223,10 @@ func MakeDB(dbURL string) (Model, error) { log.Log(context.Background(), "starting database", "dbURL", sqliteSuffix) // The pragmas ride in the DSN because they are per-connection settings and // this pool has more than one: an Exec would configure whichever connection - // happened to serve it and leave the rest at defaults. + // happened to serve it and leave the rest at defaults. (That, historically, + // is exactly what produced the "database is locked" 500s that forced the + // single-connection era: one connection had the busy timeout, the rest had + // zero and failed instantly on any collision.) // // - _busy_timeout: wait for a lock another connection (or the second // process: `streamplace sync` warming a new index) holds, instead of @@ -227,14 +241,18 @@ func MakeDB(dbURL string) (Model, error) { // - _txlock=immediate: explicit transactions take the write lock up // front instead of upgrading mid-transaction, which is the classic // multi-connection sqlite deadlock. - dsn := sqliteSuffix - pool := IndexDBPoolSize + pool := conns + if pool <= 0 { + pool = config.DefaultIndexDBConnections + } if sqliteSuffix == ":memory:" { // A pool of :memory: connections would each open a PRIVATE empty // database -- with :memory:, one connection IS the database. Tests use // this; they keep the old single-connection arrangement. pool = 1 - } else { + } + dsn := sqliteSuffix + if pool > 1 { dsn = fmt.Sprintf("file:%s?_busy_timeout=%d&_journal_mode=WAL&_synchronous=NORMAL&_txlock=immediate", sqliteSuffix, SQLiteBusyTimeout.Milliseconds()) } @@ -322,15 +340,13 @@ func MakeDB(dbURL string) (Model, error) { // server runs. const SQLiteBusyTimeout = 5 * time.Second -// IndexDBPoolSize is how many connections the index database keeps open. -// -// More than one is what makes WAL worth having: reads run against a snapshot -// on their own connections while a writer writes, so a boot-time reindex or a -// busy sweep stops queueing every request behind it. Writes still serialize -- -// on sqlite's write lock, waiting up to [SQLiteBusyTimeout] -- so raising this +// A pool larger than one is what makes WAL worth having: reads run against a +// snapshot on their own connections while a writer writes, so a boot-time +// reindex or a busy sweep stops queueing every request behind it. Writes still +// serialize -- on sqlite's write lock, waiting up to [SQLiteBusyTimeout] -- so +// the pool size (--index-db-connections, [config.DefaultIndexDBConnections]) // helps read concurrency only, and modestly: past a handful of connections the // single write lock is the ceiling. -const IndexDBPoolSize = 8 // SetSQLiteBusyTimeout applies [SQLiteBusyTimeout] to an open sqlite database. // It is a per-connection setting, which is why it is set on the pool rather diff --git a/pkg/model/pool_test.go b/pkg/model/pool_test.go index 1a184478..3d208b9f 100644 --- a/pkg/model/pool_test.go +++ b/pkg/model/pool_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/config" ) // TestIndexDBPool pins the connection-pool arrangement that lets a boot-time @@ -18,6 +19,10 @@ func TestIndexDBPool(t *testing.T) { require.NoError(t, err) db := m.(*DBModel).DB + sqlDB, err := db.DB() + require.NoError(t, err) + require.Equal(t, config.DefaultIndexDBConnections, sqlDB.Stats().MaxOpenConnections) + var mode string require.NoError(t, db.Raw("PRAGMA journal_mode;").Scan(&mode).Error) require.Equal(t, "wal", mode) @@ -29,8 +34,8 @@ func TestIndexDBPool(t *testing.T) { require.Equal(t, int(SQLiteBusyTimeout.Milliseconds()), timeout) var wg sync.WaitGroup - errs := make(chan error, IndexDBPoolSize*2) - for i := 0; i < IndexDBPoolSize*2; i++ { + errs := make(chan error, config.DefaultIndexDBConnections*2) + for i := 0; i < config.DefaultIndexDBConnections*2; i++ { wg.Add(1) go func(i int) { defer wg.Done() @@ -53,3 +58,33 @@ func TestIndexDBPool(t *testing.T) { require.NoError(t, err) } } + +// TestIndexDBLegacySingleConnection: --index-db-connections=1 is the +// slower-but-safer fallback, and it must be a genuine one -- the historical +// arrangement exactly: one connection, plain DSN, pragmas applied by Exec. +// (The synchronous level turns out to be the driver's compiled-in NORMAL in +// both modes, so the pool size is the whole difference.) +func TestIndexDBLegacySingleConnection(t *testing.T) { + m, err := MakeDBConns(t.TempDir(), 1) + require.NoError(t, err) + db := m.(*DBModel).DB + + sqlDB, err := db.DB() + require.NoError(t, err) + require.Equal(t, 1, sqlDB.Stats().MaxOpenConnections) + + var mode string + require.NoError(t, db.Raw("PRAGMA journal_mode;").Scan(&mode).Error) + require.Equal(t, "wal", mode, "WAL was always on") + var synchronous int + require.NoError(t, db.Raw("PRAGMA synchronous;").Scan(&synchronous).Error) + require.Equal(t, 1, synchronous, "the driver's compiled default (NORMAL) -- same either way") + var timeout int + require.NoError(t, db.Raw("PRAGMA busy_timeout;").Scan(&timeout).Error) + require.Equal(t, int(SQLiteBusyTimeout.Milliseconds()), timeout, "Exec still reaches the only connection") + + require.NoError(t, m.UpdateRepo(&Repo{DID: "did:plc:legacymode", Version: "rev"})) + repo, err := m.GetRepo("did:plc:legacymode") + require.NoError(t, err) + require.Equal(t, "rev", repo.Version) +} diff --git a/pkg/model/repo.go b/pkg/model/repo.go index 26a30b5e..95ce0648 100644 --- a/pkg/model/repo.go +++ b/pkg/model/repo.go @@ -66,6 +66,15 @@ func (m *DBModel) GetRepo(did string) (*Repo, error) { return &repoModel, nil } +// CountRepos reports how many repos the index has rows for. Zero means a +// fresh index, whose first sweep is the boot-critical work rather than +// background insurance. +func (m *DBModel) CountRepos() (int64, error) { + var n int64 + err := m.DB.Model(&Repo{}).Count(&n).Error + return n, err +} + func (m *DBModel) GetAllRepos() ([]Repo, error) { var repos []Repo res := m.DB.Find(&repos) -- 2.51.2