diff --git a/.env.example b/.env.example index 5de6e69a3b..dbb5636906 100644 --- a/.env.example +++ b/.env.example @@ -52,6 +52,9 @@ POSTGRES_SHM_SIZE=8gb # POSTGRES_TUNE_CPUS=auto # POSTGRES_TUNE_STORAGE=ssd # POSTGRES_TUNE_DB_SIZE=auto +# POSTGRES_TUNE_CONNECTIONS sets max_connections. With more than one Silo +# process (API, proxy and transcode nodes), cover database.max_connections +# (20 by default) for every one of them, plus room for admin sessions. # POSTGRES_TUNE_CONNECTIONS=100 # Optional host port overrides for docker-compose.yml. The container listeners diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 80f6b1fb0e..04c857b38b 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -132,6 +132,30 @@ than the Goose CLI: those targets copy legacy `schema_versions` rows into anything. Set `ENV_FILE=path/to/.env` to read the database URL from a different env file. +### Keep migrations online + +Migrations run at startup while other API replicas keep serving, against tables +that can hold millions of rows. Keep locks on existing tables short: + +- Create and drop indexes on existing tables with `CONCURRENTLY`, in a file + marked `-- +goose NO TRANSACTION`. Run `DROP INDEX CONCURRENTLY IF EXISTS` + before creating a new index: a copy left by an interrupted run is invalid, and + `CREATE INDEX CONCURRENTLY IF NOT EXISTS` would keep it, so the planner would + never use the index. +- Add foreign keys and `CHECK` constraints to existing tables as `NOT VALID`, + and run `VALIDATE CONSTRAINT` only after that commits: in a later migration, + or as a separate statement in a `NO TRANSACTION` file. Inside one + transaction, validation keeps the lock the `ADD` took. A `CHECK` attached to + `ADD COLUMN` scans the whole table under an `ACCESS EXCLUSIVE` lock. +- Backfill large tables in batches that commit as they go, through a procedure + or a Go migration, rather than one statement over every row. +- Scope session settings to the transaction with `SET LOCAL` or + `set_config(..., true)`. Compute partition bounds and dates in UTC, not in the + server's `TimeZone`. + +`20260914145000_index_artwork_reference_paths.sql` shows the index pattern, and +`20260930130041_store_media_search_fields.sql` the batched backfill. + ## Tests and lint While iterating: diff --git a/internal/database/invalid_index_rebuild.go b/internal/database/invalid_index_rebuild.go new file mode 100644 index 0000000000..bbea42ca1f --- /dev/null +++ b/internal/database/invalid_index_rebuild.go @@ -0,0 +1,79 @@ +package database + +import ( + "context" + "database/sql" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/pressly/goose/v3" +) + +// invalidIndexRebuildVersion sorts after every migration that built one of +// rebuiltConcurrentIndexes. +const invalidIndexRebuildVersion int64 = 20261008165913 + +// rebuiltConcurrentIndexes come from migrations that ran CREATE INDEX +// CONCURRENTLY IF NOT EXISTS without first removing an invalid copy. If a +// build was interrupted, the retry skipped the invalid leftover and goose +// recorded the migration as applied, so the planner has never used the index. +var rebuiltConcurrentIndexes = []struct{ name, definition string }{ + {"idx_media_files_folder_lower_file_path", "media_files (media_folder_id, lower(file_path))"}, + {"idx_media_files_folder_file_path_pattern", "media_files (media_folder_id, file_path text_pattern_ops)"}, + {"idx_media_items_content_type", "media_items USING btree (content_id, type)"}, + {"idx_uwp_profile_completed_cursor", "user_watch_progress USING btree (user_id, profile_id, updated_at DESC, media_item_id DESC) WHERE completed = TRUE"}, + {"idx_abs_playback_sessions_user_profile_started", "abs_playback_sessions USING btree (user_id, profile_id, started_at)"}, + {"idx_manga_enrichment_state_next_attempt", "manga_enrichment_state (next_attempt_at) WHERE next_attempt_at IS NOT NULL"}, +} + +// invalidIndexRebuildMigration rebuilds whichever of rebuiltConcurrentIndexes +// an interrupted build left invalid. +// +// A Go migration rather than SQL because valid indexes must be left alone, +// and that condition cannot live in SQL: DROP INDEX CONCURRENTLY refuses to run +// inside a DO block, and a plain DROP INDEX takes an ACCESS EXCLUSIVE lock +// that stalls every replica still serving the table. RunDB so each statement +// runs on its own. +func invalidIndexRebuildMigration() *goose.Migration { + m := goose.NewGoMigration( + invalidIndexRebuildVersion, + &goose.GoFunc{RunDB: func(ctx context.Context, db *sql.DB) error { + return rebuildInvalidIndexes(ctx, db, "public") + }}, + // The indexes belong to the migrations that first created them. + &goose.GoFunc{RunDB: func(context.Context, *sql.DB) error { return nil }}, + ) + m.Source = fmt.Sprintf("%d_rebuild_invalid_concurrent_indexes.go", invalidIndexRebuildVersion) + return m +} + +// rebuildInvalidIndexes drops each invalid index in rebuiltConcurrentIndexes +// and builds it again, both concurrently, leaving valid indexes alone. A +// missing index is built too: the migrations that create these run first, so +// one is missing only when an earlier attempt of this migration stopped +// between the drop and the build. +func rebuildInvalidIndexes(ctx context.Context, db *sql.DB, schema string) error { + quotedSchema := pgx.Identifier{schema}.Sanitize() + for _, index := range rebuiltConcurrentIndexes { + quotedName := pgx.Identifier{index.name}.Sanitize() + qualified := quotedSchema + "." + quotedName + var valid bool + if err := db.QueryRowContext(ctx, + `SELECT coalesce((SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass($1)), false)`, + qualified, + ).Scan(&valid); err != nil { + return fmt.Errorf("checking index %s: %w", index.name, err) + } + if valid { + continue + } + if _, err := db.ExecContext(ctx, "DROP INDEX CONCURRENTLY IF EXISTS "+qualified); err != nil { + return fmt.Errorf("dropping invalid index %s: %w", index.name, err) + } + if _, err := db.ExecContext(ctx, fmt.Sprintf("CREATE INDEX CONCURRENTLY IF NOT EXISTS %s ON %s.%s", + quotedName, quotedSchema, index.definition)); err != nil { + return fmt.Errorf("rebuilding index %s: %w", index.name, err) + } + } + return nil +} diff --git a/internal/database/invalid_index_rebuild_test.go b/internal/database/invalid_index_rebuild_test.go new file mode 100644 index 0000000000..b9342700e9 --- /dev/null +++ b/internal/database/invalid_index_rebuild_test.go @@ -0,0 +1,158 @@ +package database + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/stdlib" +) + +// The rebuild runs while other replicas keep serving the tables. With a reader +// holding a lock on the table, it must wait without blocking anyone else, then +// rebuild the invalid index and build missing ones, leaving valid ones alone. +func TestInvalidIndexRebuildKeepsTablesAvailablePostgres(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + ctx := t.Context() + config, err := pgx.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + db := stdlib.OpenDB(*config) + t.Cleanup(func() { _ = db.Close() }) + connect := func() *pgx.Conn { + t.Helper() + conn, err := pgx.ConnectConfig(ctx, config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close(context.Background()) }) + return conn + } + schema := fmt.Sprintf("invalid_index_rebuild_%d", time.Now().UnixNano()) + exec := func(query string) { + t.Helper() + if _, err := db.ExecContext(ctx, query); err != nil { + t.Fatal(err) + } + } + exec("CREATE SCHEMA " + schema) + t.Cleanup(func() { + if _, err := db.ExecContext(context.Background(), "DROP SCHEMA "+schema+" CASCADE"); err != nil { + t.Error(err) + } + }) + exec("CREATE TABLE " + schema + ".media_files (media_folder_id integer, file_path text)") + exec("CREATE TABLE " + schema + ".media_items (content_id text, type text)") + exec("CREATE TABLE " + schema + ".user_watch_progress (user_id integer, profile_id text, updated_at timestamptz, media_item_id text, completed boolean)") + exec("CREATE TABLE " + schema + ".abs_playback_sessions (user_id integer, profile_id text, started_at timestamptz)") + exec("CREATE TABLE " + schema + ".manga_enrichment_state (next_attempt_at timestamptz)") + exec("INSERT INTO " + schema + ".media_files VALUES (1, '/a.mkv'), (1, '/b.mkv')") + + exec("CREATE INDEX idx_media_items_content_type ON " + schema + ".media_items (content_id, type)") + indexOID := func(name string) uint32 { + t.Helper() + var oid uint32 + if err := db.QueryRowContext(ctx, "SELECT coalesce(to_regclass($1)::oid, 0)", schema+"."+name).Scan(&oid); err != nil { + t.Fatal(err) + } + return oid + } + validOID := indexOID("idx_media_items_content_type") + // A failed concurrent build leaves an invalid index under the name a later + // IF NOT EXISTS retry skips. + _, err = db.ExecContext(ctx, "CREATE UNIQUE INDEX CONCURRENTLY idx_media_files_folder_file_path_pattern ON "+schema+".media_files (media_folder_id)") + if pgErr, ok := errors.AsType[*pgconn.PgError](err); !ok || pgErr.Code != "23505" { + t.Fatalf("expected failed concurrent build, got %v", err) + } + + reader := connect() + readerTx, err := reader.Begin(ctx) + if err != nil { + t.Fatal(err) + } + if _, err := readerTx.Exec(ctx, "SELECT count(*) FROM "+schema+".media_files"); err != nil { + t.Fatal(err) + } + + done := make(chan error, 1) + go func() { done <- rebuildInvalidIndexes(ctx, db, schema) }() + + // Wait until the rebuild is queued behind the reader. + deadline := time.Now().Add(15 * time.Second) + for { + var waiting bool + if err := db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM pg_stat_activity + WHERE wait_event_type = 'Lock' AND pid <> pg_backend_pid() AND strpos(query, $1) > 0)`, schema).Scan(&waiting); err != nil { + t.Fatal(err) + } + if waiting { + break + } + select { + case err := <-done: + t.Fatalf("rebuild finished while a reader held the table: %v", err) + default: + } + if time.Now().After(deadline) { + t.Fatal("rebuild never waited for the reader") + } + time.Sleep(20 * time.Millisecond) + } + + writer := connect() + if _, err := writer.Exec(ctx, "SET lock_timeout = '2s'"); err != nil { + t.Fatal(err) + } + if _, err := writer.Exec(ctx, "INSERT INTO "+schema+".media_files VALUES (2, '/c.mkv')"); err != nil { + t.Fatalf("write blocked while the rebuild waited: %v", err) + } + if _, err := writer.Exec(ctx, "SELECT count(*) FROM "+schema+".media_files"); err != nil { + t.Fatalf("read blocked while the rebuild waited: %v", err) + } + + if err := readerTx.Commit(ctx); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(30 * time.Second): + t.Fatal("rebuild did not finish after the reader committed") + } + + // The invalid index is rebuilt, and the missing ones (as if an earlier + // attempt stopped right after dropping them) are built. + for _, index := range rebuiltConcurrentIndexes { + var valid, unique bool + var def string + if err := db.QueryRowContext(ctx, `SELECT indisvalid, indisunique, pg_get_indexdef(indexrelid) FROM pg_index WHERE indexrelid = to_regclass($1)`, + schema+"."+index.name).Scan(&valid, &unique, &def); err != nil { + t.Fatalf("index %s: %v", index.name, err) + } + if !valid || unique { + t.Errorf("index %s valid=%t unique=%t: %s", index.name, valid, unique, def) + } + } + var def string + if err := db.QueryRowContext(ctx, `SELECT pg_get_indexdef($1::regclass)`, schema+".idx_media_files_folder_file_path_pattern").Scan(&def); err != nil { + t.Fatal(err) + } + if !strings.Contains(def, "media_folder_id, file_path text_pattern_ops") { + t.Errorf("rebuilt index definition: %s", def) + } + if got := indexOID("idx_media_items_content_type"); got != validOID { + t.Error("valid index was rebuilt") + } +} diff --git a/internal/database/log_partition_timezone_test.go b/internal/database/log_partition_timezone_test.go new file mode 100644 index 0000000000..5e1799591d --- /dev/null +++ b/internal/database/log_partition_timezone_test.go @@ -0,0 +1,98 @@ +package database + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/partman" + "github.com/Silo-Server/silo-server/migrations" +) + +// Migration 028 creates the first log partitions and partman extends them +// later in UTC. On a server whose TimeZone is not UTC the two must still agree: +// otherwise partman's next partition overlaps or leaves a gap beside one 028 +// made, and writes fall into the default partition. Each zone gets its own +// database because 028 runs once; the test role needs CREATEDB. +func TestLogPartitionsMatchPartmanOnNonUTCServer(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + ctx := t.Context() + admin, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + + // Zones on both sides of UTC: a negative offset makes partman's partition + // overlap 028's, a positive one can leave a gap before it. + for _, zone := range []string{"America/New_York", "Asia/Tokyo"} { + t.Run(zone, func(t *testing.T) { + name := fmt.Sprintf("silo_log_partition_tz_%d", time.Now().UnixNano()) + if _, err := admin.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+pgx.Identifier{name}.Sanitize()+" WITH (FORCE)"); err != nil { + t.Error(err) + } + }() + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + cfg.ConnConfig.Database = name + cfg.ConnConfig.RuntimeParams["timezone"] = zone + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + provider, err := newMigrationProvider(pool, migrations.FS, "sql") + if err != nil { + t.Fatal(err) + } + defer func() { _ = provider.Close() }() + if _, err := provider.UpTo(ctx, 28); err != nil { + t.Fatal(err) + } + + // Reach past the periods 028 created, so partman has to add + // partitions next to them. + for table, granularity := range map[string]partman.Granularity{ + "operational_logs": partman.Daily, + "activity_log": partman.Weekly, + } { + if err := partman.NewManager(pool, table, granularity, 8).EnsureFuturePartitions(ctx); err != nil { + t.Errorf("%s: %v", table, err) + } + } + + if _, err := pool.Exec(ctx, ` +INSERT INTO operational_logs ("timestamp", level, component, message) +SELECT ts, 'info', 'test', 'partition coverage' +FROM generate_series(now(), now() + interval '8 days', interval '1 hour') ts; +INSERT INTO activity_log ("timestamp", client_ip, method, path) +SELECT ts, '127.0.0.1', 'GET', '/' +FROM generate_series(now(), now() + interval '8 weeks', interval '1 hour') ts;`); err != nil { + t.Fatal(err) + } + for _, table := range []string{"operational_logs_default", "activity_log_default"} { + var n int + if err := pool.QueryRow(ctx, "SELECT count(*) FROM public."+pgx.Identifier{table}.Sanitize()).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Errorf("%s holds %d rows inside the partitioned window", table, n) + } + } + }) + } +} diff --git a/internal/database/migrate.go b/internal/database/migrate.go index 08b32a1422..74d1fc66f4 100644 --- a/internal/database/migrate.go +++ b/internal/database/migrate.go @@ -191,11 +191,14 @@ func newMigrationProvider(pool *pgxpool.Pool, fsys fs.FS, dir string) (*goose.Pr // typed JSON, the displayprefs move parses the legacy jellycompat // keys, and the subtitle language backfill applies the scanner's // lang.CompatibleTag — none expressible in SQL without duplicating - // those rules. + // those rules. The invalid index rebuild decides per index whether to + // drop and rebuild it concurrently, which SQL cannot do outside a + // transaction. goose.WithGoMigrations( settingsBackfillMigration(), displayPrefsMoveMigration(), subtitleLanguageBackfillMigration(), + invalidIndexRebuildMigration(), ), ) if err != nil { diff --git a/internal/database/postgres.go b/internal/database/postgres.go index f6d0fe3944..cc406d8fef 100644 --- a/internal/database/postgres.go +++ b/internal/database/postgres.go @@ -6,9 +6,14 @@ import ( "github.com/Silo-Server/silo-server/internal/config" "github.com/Silo-Server/silo-server/internal/telemetry" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) +// jitOff is the jit setting Silo runs with, on its own connections and in the +// tuner's server recommendations. +const jitOff = "off" + // NewPool creates a new PostgreSQL connection pool using the provided // DatabaseConfig. It configures the pool with the specified maximum number of // connections and verifies the connection by issuing a ping. @@ -27,6 +32,7 @@ func NewPoolForRole(ctx context.Context, cfg config.DatabaseConfig, role string) if cfg.MaxConnections > 0 { poolCfg.MaxConns = int32(cfg.MaxConnections) } + poolCfg.AfterConnect = disableUnconfiguredJIT poolCfg.ConnConfig.Tracer = postgresTracer{role: role} pool, err := pgxpool.NewWithConfig(ctx, poolCfg) @@ -42,3 +48,18 @@ func NewPoolForRole(ctx context.Context, cfg config.DatabaseConfig, role string) postgresPools.add(pool, role) return pool, nil } + +// disableUnconfiguredJIT turns JIT off for the session unless something +// already chose it: the server configuration, ALTER DATABASE or ROLE, or +// DATABASE_URL, directly or through options. Silo's queries are short reads the +// planner often prices past jit_above_cost, and compiling them then takes +// longer than running them. +func disableUnconfiguredJIT(ctx context.Context, conn *pgx.Conn) error { + if _, err := conn.Exec(ctx, + `SELECT set_config('jit', $1, false) FROM pg_settings WHERE name = 'jit' AND source = 'default'`, + jitOff, + ); err != nil { + return fmt.Errorf("disabling JIT: %w", err) + } + return nil +} diff --git a/internal/database/postgres_jit_test.go b/internal/database/postgres_jit_test.go new file mode 100644 index 0000000000..8e504654e7 --- /dev/null +++ b/internal/database/postgres_jit_test.go @@ -0,0 +1,84 @@ +package database + +import ( + "context" + "fmt" + "net/url" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/config" +) + +// Pool connections run with JIT off unless something configured jit: here the +// URL, its options, or the database. The test role needs CREATEDB. +func TestNewPoolDisablesJITUnlessConfiguredPostgres(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + ctx := t.Context() + base, err := url.Parse(dsn) + if err != nil { + t.Fatal(err) + } + showJIT := func(t *testing.T, u url.URL) string { + t.Helper() + pool, err := NewPoolForRole(ctx, config.DatabaseConfig{URL: u.String(), MaxConnections: 1}, "application") + if err != nil { + t.Fatal(err) + } + defer ClosePool(pool) + var got string + if err := pool.QueryRow(ctx, `SHOW jit`).Scan(&got); err != nil { + t.Fatal(err) + } + return got + } + + for _, tt := range []struct{ param, value, want string }{ + {want: "off"}, + {param: "jit", value: "on", want: "on"}, + {param: "options", value: "-c jit=on", want: "on"}, + {param: "options", value: "-ecjit=on", want: "on"}, + {param: "options", value: "-c application_name=jitter", want: "off"}, + } { + u := *base + if tt.param != "" { + q := u.Query() + q.Set(tt.param, tt.value) + u.RawQuery = q.Encode() + } + if got := showJIT(t, u); got != tt.want { + t.Errorf("jit = %s, want %s (%s=%q)", got, tt.want, tt.param, tt.value) + } + } + + admin, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + name := fmt.Sprintf("silo_jit_%d", time.Now().UnixNano()) + quoted := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE DATABASE "+quoted); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted+" WITH (FORCE)"); err != nil { + t.Error(err) + } + }() + if _, err := admin.Exec(ctx, "ALTER DATABASE "+quoted+" SET jit = on"); err != nil { + t.Fatal(err) + } + u := *base + u.Path = "/" + name + if got := showJIT(t, u); got != "on" { + t.Errorf("jit = %s, want on from ALTER DATABASE", got) + } +} diff --git a/internal/database/postgres_tune.go b/internal/database/postgres_tune.go index 62dca0efa1..354d0f9e9a 100644 --- a/internal/database/postgres_tune.go +++ b/internal/database/postgres_tune.go @@ -226,7 +226,7 @@ func RecommendPostgresOLTPSettings(opts PostgresTuneOptions, postgresMajorVersio ) if postgresMajorVersion >= 12 { - settings = append(settings, PostgresTuneSetting{Name: "jit", Value: "off"}) + settings = append(settings, PostgresTuneSetting{Name: "jit", Value: jitOff}) } if postgresMajorVersion >= 15 { settings = append(settings, PostgresTuneSetting{Name: "wal_compression", Value: "lz4"}) diff --git a/internal/opslog/repo.go b/internal/opslog/repo.go index b282cf5514..42c8a0f6b2 100644 --- a/internal/opslog/repo.go +++ b/internal/opslog/repo.go @@ -208,7 +208,9 @@ func buildListQuery(opts ListOptions) (string, []any, int, error) { if err != nil { return "", nil, 0, err } - conditions = append(conditions, fmt.Sprintf("(timestamp, id) < ($%d, $%d)", argIdx, argIdx+1)) + // The redundant bare bound lets PostgreSQL prune partitions newer than + // the cursor; it cannot prune on the row comparison. + conditions = append(conditions, fmt.Sprintf("(timestamp, id) < ($%d, $%d) AND timestamp <= $%d", argIdx, argIdx+1, argIdx)) args = append(args, cursorTs, cursorID) argIdx += 2 } diff --git a/internal/opslog/repo_db_test.go b/internal/opslog/repo_db_test.go new file mode 100644 index 0000000000..de1acecee5 --- /dev/null +++ b/internal/opslog/repo_db_test.go @@ -0,0 +1,36 @@ +package opslog + +import ( + "context" + "strings" + "testing" + "time" +) + +// A cursor page must skip the daily partitions newer than the cursor instead +// of opening an index scan on every one of them. +func TestListCursorPrunesNewerPartitionsDB(t *testing.T) { + pool := testPool(t) + query, args, _, err := buildListQuery(ListOptions{Cursor: encodeCursor(time.Now().AddDate(0, 0, -30), 1)}) + if err != nil { + t.Fatal(err) + } + rows, err := pool.Query(context.Background(), "EXPLAIN (COSTS OFF) "+query, args...) + if err != nil { + t.Fatal(err) + } + var plan strings.Builder + for rows.Next() { + var line string + if err := rows.Scan(&line); err != nil { + t.Fatal(err) + } + plan.WriteString(line + "\n") + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if strings.Contains(plan.String(), "operational_logs_p_") { + t.Fatalf("cursor page scans partitions newer than the cursor:\n%s", plan.String()) + } +} diff --git a/internal/userstore/pgstore/progress.go b/internal/userstore/pgstore/progress.go index 0a8facd431..060b8106ac 100644 --- a/internal/userstore/pgstore/progress.go +++ b/internal/userstore/pgstore/progress.go @@ -745,7 +745,7 @@ func (s *PostgresUserStore) ListProgress(ctx context.Context, profileID, status } // ListCompletedProgressSince compares on the raw column so -// idx_uwp_profile_completed serves it as a range scan: a row's whole-second +// idx_uwp_profile_completed_cursor serves it as a range scan: a row's whole-second // updated_at is after a bound exactly when the column is at least the bound's // next whole second. func (s *PostgresUserStore) ListCompletedProgressSince(ctx context.Context, profileID string, since, until time.Time, limit int) ([]userstore.WatchProgress, error) { @@ -797,9 +797,9 @@ func (s *PostgresUserStore) ListProgressPage(ctx context.Context, profileID, sta // series_id → media_items (a plain media_items join would miss them); the // optional library predicate hits media_item_libraries. With // `ORDER BY updated_at DESC`, the completed branch's `completed = TRUE` keeps -// idx_uwp_profile_completed in play and the in_progress branch's +// idx_uwp_profile_completed_cursor in play and the in_progress branch's // `position_seconds > 0` keeps idx_uwp_profile_resumable in play, while the -// EXISTS sub-selects ride idx_item_libraries_content. The filter is coarse +// EXISTS sub-selects ride media_item_libraries_pkey. The filter is coarse // (callers re-check access/parental exclusions over the hydrated rows), and an // empty types slice with a nil libraryID degrades to the plain status listing. func (s *PostgresUserStore) ListProgressFiltered(ctx context.Context, profileID, status string, types []string, libraryID *int, limit, offset int) ([]userstore.WatchProgress, error) { diff --git a/migrations/search_number_token_test.go b/migrations/search_number_token_test.go new file mode 100644 index 0000000000..b6a233398e --- /dev/null +++ b/migrations/search_number_token_test.go @@ -0,0 +1,88 @@ +package migrations + +import ( + "context" + "os" + "strings" + "testing" + + "github.com/jackc/pgx/v5" +) + +// TestSearchNumberTokenInlinesPostgres pins normalize_search_number_token's +// output, which title_normalized and the search indexes store, and checks that +// PostgreSQL inlines it: a body PostgreSQL cannot inline plans a query for +// every token of every title write. +func TestSearchNumberTokenInlinesPostgres(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + connect := func(conforming string) *pgx.Conn { + t.Helper() + config, err := pgx.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + config.RuntimeParams["standard_conforming_strings"] = conforming + conn, err := pgx.ConnectConfig(t.Context(), config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close(context.Background()) }) + return conn + } + + // Each session parses the body once and keeps it, so its string literals + // must mean the same whichever way that session set + // standard_conforming_strings first. + for _, conforming := range []string{"on", "off"} { + conn := connect(conforming) + for input, want := range map[string]string{ + "Twentieth": "20", + "ONE": "1", + "second": "2", + "twentyone": "twentyone", + "101st": "101", + "21ST": "21", + "0th": "0", + "1stt": "1stt", + "a1st": "a1st", + "st": "st", + "": "", + } { + var got string + if err := conn.QueryRow(t.Context(), `SELECT public.normalize_search_number_token($1)`, input).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Errorf("standard_conforming_strings=%s: normalize_search_number_token(%q) = %q, want %q", conforming, input, got, want) + } + } + } + conn := connect("on") + var null string + if err := conn.QueryRow(t.Context(), `SELECT public.normalize_search_number_token(NULL)`).Scan(&null); err != nil || null != "" { + t.Errorf("normalize_search_number_token(NULL) = %q, %v; want empty", null, err) + } + + var plan strings.Builder + // A column input, so the planner cannot fold the call into a constant. + rows, err := conn.Query(t.Context(), `EXPLAIN (VERBOSE, COSTS OFF) SELECT public.normalize_search_number_token(n::text) FROM generate_series(1, 2) AS n`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var line string + if err := rows.Scan(&line); err != nil { + t.Fatal(err) + } + plan.WriteString(line + "\n") + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if strings.Contains(plan.String(), "normalize_search_number_token") { + t.Errorf("normalize_search_number_token is not inlined:\n%s", plan.String()) + } +} diff --git a/migrations/sql/028_log_partitioning.sql b/migrations/sql/028_log_partitioning.sql index 483ab768b7..85fe35f1fb 100644 --- a/migrations/sql/028_log_partitioning.sql +++ b/migrations/sql/028_log_partitioning.sql @@ -1,5 +1,9 @@ -- +goose Up -- +goose StatementBegin +-- Cut partitions at UTC boundaries, as internal/partman does when it extends +-- them; date_trunc and interval arithmetic otherwise follow the server TimeZone. +SET LOCAL TimeZone = 'UTC'; + CREATE TABLE public.operational_logs_partitioned ( id bigint GENERATED ALWAYS AS IDENTITY, "timestamp" timestamp with time zone DEFAULT now() NOT NULL, diff --git a/migrations/sql/20261008165511_inline_search_number_token.sql b/migrations/sql/20261008165511_inline_search_number_token.sql new file mode 100644 index 0000000000..6d58634e44 --- /dev/null +++ b/migrations/sql/20261008165511_inline_search_number_token.sql @@ -0,0 +1,129 @@ +-- +goose Up +-- +goose StatementBegin +-- Same output as migration 138's version, so title_normalized and the +-- expressions indexed over normalize_search_text stay valid without a rebuild. +-- A single-expression body lets PostgreSQL inline the function into its +-- callers instead of planning a CTE for every token of every title write. The +-- patterns avoid backslashes, whose meaning depends on standard_conforming_strings. +CREATE OR REPLACE FUNCTION public.normalize_search_number_token(token text) +RETURNS text +LANGUAGE sql +IMMUTABLE +PARALLEL SAFE +AS $$ + SELECT CASE LOWER(COALESCE(token, '')) + WHEN 'zero' THEN '0' + WHEN 'zeroth' THEN '0' + WHEN 'one' THEN '1' + WHEN 'first' THEN '1' + WHEN 'two' THEN '2' + WHEN 'second' THEN '2' + WHEN 'three' THEN '3' + WHEN 'third' THEN '3' + WHEN 'four' THEN '4' + WHEN 'fourth' THEN '4' + WHEN 'five' THEN '5' + WHEN 'fifth' THEN '5' + WHEN 'six' THEN '6' + WHEN 'sixth' THEN '6' + WHEN 'seven' THEN '7' + WHEN 'seventh' THEN '7' + WHEN 'eight' THEN '8' + WHEN 'eighth' THEN '8' + WHEN 'nine' THEN '9' + WHEN 'ninth' THEN '9' + WHEN 'ten' THEN '10' + WHEN 'tenth' THEN '10' + WHEN 'eleven' THEN '11' + WHEN 'eleventh' THEN '11' + WHEN 'twelve' THEN '12' + WHEN 'twelfth' THEN '12' + WHEN 'thirteen' THEN '13' + WHEN 'thirteenth' THEN '13' + WHEN 'fourteen' THEN '14' + WHEN 'fourteenth' THEN '14' + WHEN 'fifteen' THEN '15' + WHEN 'fifteenth' THEN '15' + WHEN 'sixteen' THEN '16' + WHEN 'sixteenth' THEN '16' + WHEN 'seventeen' THEN '17' + WHEN 'seventeenth' THEN '17' + WHEN 'eighteen' THEN '18' + WHEN 'eighteenth' THEN '18' + WHEN 'nineteen' THEN '19' + WHEN 'nineteenth' THEN '19' + WHEN 'twenty' THEN '20' + WHEN 'twentieth' THEN '20' + ELSE + CASE + WHEN LOWER(COALESCE(token, '')) ~ '^[0-9]+(st|nd|rd|th)$' + THEN REGEXP_REPLACE(LOWER(COALESCE(token, '')), '(st|nd|rd|th)$', '') + ELSE LOWER(COALESCE(token, '')) + END + END; +$$; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +CREATE OR REPLACE FUNCTION public.normalize_search_number_token(token text) +RETURNS text +LANGUAGE sql +IMMUTABLE +PARALLEL SAFE +AS $$ + WITH normalized AS ( + SELECT LOWER(COALESCE(token, '')) AS value + ) + SELECT CASE value + WHEN 'zero' THEN '0' + WHEN 'zeroth' THEN '0' + WHEN 'one' THEN '1' + WHEN 'first' THEN '1' + WHEN 'two' THEN '2' + WHEN 'second' THEN '2' + WHEN 'three' THEN '3' + WHEN 'third' THEN '3' + WHEN 'four' THEN '4' + WHEN 'fourth' THEN '4' + WHEN 'five' THEN '5' + WHEN 'fifth' THEN '5' + WHEN 'six' THEN '6' + WHEN 'sixth' THEN '6' + WHEN 'seven' THEN '7' + WHEN 'seventh' THEN '7' + WHEN 'eight' THEN '8' + WHEN 'eighth' THEN '8' + WHEN 'nine' THEN '9' + WHEN 'ninth' THEN '9' + WHEN 'ten' THEN '10' + WHEN 'tenth' THEN '10' + WHEN 'eleven' THEN '11' + WHEN 'eleventh' THEN '11' + WHEN 'twelve' THEN '12' + WHEN 'twelfth' THEN '12' + WHEN 'thirteen' THEN '13' + WHEN 'thirteenth' THEN '13' + WHEN 'fourteen' THEN '14' + WHEN 'fourteenth' THEN '14' + WHEN 'fifteen' THEN '15' + WHEN 'fifteenth' THEN '15' + WHEN 'sixteen' THEN '16' + WHEN 'sixteenth' THEN '16' + WHEN 'seventeen' THEN '17' + WHEN 'seventeenth' THEN '17' + WHEN 'eighteen' THEN '18' + WHEN 'eighteenth' THEN '18' + WHEN 'nineteen' THEN '19' + WHEN 'nineteenth' THEN '19' + WHEN 'twenty' THEN '20' + WHEN 'twentieth' THEN '20' + ELSE + CASE + WHEN value ~ '^[0-9]+(st|nd|rd|th)$' THEN REGEXP_REPLACE(value, '(st|nd|rd|th)$', '') + ELSE value + END + END + FROM normalized; +$$; +-- +goose StatementEnd diff --git a/migrations/sql/20261008170456_drop_redundant_indexes.sql b/migrations/sql/20261008170456_drop_redundant_indexes.sql new file mode 100644 index 0000000000..06b08fb818 --- /dev/null +++ b/migrations/sql/20261008170456_drop_redundant_indexes.sql @@ -0,0 +1,75 @@ +-- +goose NO TRANSACTION + +-- +goose Up +-- Every index dropped here is non-unique and backs no constraint, and the +-- planner gets the same access path from another index, so each one only costs +-- writes. Most sit on tables that library scans, metadata refreshes and +-- progress reports rewrite constantly. + +-- Each one's columns, opclasses, order and predicate are a leading prefix of +-- an index the table keeps (named after the arrow). idx_user_watch_progress_profile +-- and idx_media_files_folder stay although their supersets could serve them: +-- deduplication keeps them several times smaller, so reading a whole profile's +-- progress or counting a folder's files touches far fewer pages. +DROP INDEX CONCURRENTLY IF EXISTS public.idx_uwp_profile_completed; -- idx_uwp_profile_completed_cursor +DROP INDEX CONCURRENTLY IF EXISTS public.idx_item_libraries_content; -- media_item_libraries_pkey +DROP INDEX CONCURRENTLY IF EXISTS public.idx_item_libraries_folder; -- idx_item_libraries_folder_seen_content +DROP INDEX CONCURRENTLY IF EXISTS public.idx_item_people_content_id; -- idx_item_people_content_kind_person +DROP INDEX CONCURRENTLY IF EXISTS public.idx_item_people_person_id; -- idx_item_people_person_kind +-- Scanned backward, the descending index returns the ascending order. +DROP INDEX CONCURRENTLY IF EXISTS public.idx_episodes_air_date; -- idx_episodes_air_date_content + +-- Exact copies of a primary key or unique constraint. +DROP INDEX CONCURRENTLY IF EXISTS public.idx_series_match_queue_folder; +DROP INDEX CONCURRENTLY IF EXISTS public.idx_series_root_match_queue_folder; +DROP INDEX CONCURRENTLY IF EXISTS public.idx_literary_work_items_content; +DROP INDEX CONCURRENTLY IF EXISTS public.idx_watch_provider_connections_provider_enabled; + +-- No query reads these. Episode search matches the stored vectors on +-- episode_catalog_entries, and taste clusters are only read by profile, never +-- by distance, yet every cluster rewrite pays for the HNSW graph. +DROP INDEX CONCURRENTLY IF EXISTS public.idx_episodes_search_title; +DROP INDEX CONCURRENTLY IF EXISTS public.idx_episodes_search_overview; +DROP INDEX CONCURRENTLY IF EXISTS public.idx_taste_clusters_embedding; + +-- +goose Down +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_uwp_profile_completed + ON public.user_watch_progress USING btree (user_id, profile_id, updated_at DESC) + WHERE completed = TRUE; +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_libraries_content + ON public.media_item_libraries USING btree (content_id); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_libraries_folder + ON public.media_item_libraries USING btree (media_folder_id, first_seen_at DESC); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_people_content_id + ON public.item_people USING btree (content_id); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_people_person_id + ON public.item_people USING btree (person_id); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_episodes_air_date + ON public.episodes USING btree (air_date) + WHERE air_date IS NOT NULL; +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_series_match_queue_folder + ON public.series_match_queue USING btree (media_folder_id, group_key_version, content_group_key); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_series_root_match_queue_folder + ON public.series_root_match_queue USING btree (media_folder_id, observed_root_path); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_literary_work_items_content + ON public.literary_work_items USING btree (content_id); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_watch_provider_connections_provider_enabled + ON public.watch_provider_connections USING btree (provider, user_id, profile_id); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_episodes_search_title +ON public.episodes USING gin ( + setweight( + to_tsvector( + 'simple', + public.normalize_search_text( + COALESCE(NULLIF(BTRIM(title), ''), 'Episode ' || episode_number::text) + ) + ), + 'A' + ) +); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_episodes_search_overview +ON public.episodes USING gin ( + to_tsvector('english', COALESCE(overview, '')) +); +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_taste_clusters_embedding + ON public.user_taste_clusters USING hnsw ((embedding::halfvec(3072)) halfvec_cosine_ops); diff --git a/scripts/ci/db-pins.txt b/scripts/ci/db-pins.txt index cf1f745832..8c7ced878e 100644 --- a/scripts/ci/db-pins.txt +++ b/scripts/ci/db-pins.txt @@ -209,3 +209,22 @@ # Jellyfin password#PIN sign-ins share the per-profile PIN lockout. ./internal/jellycompat TestLoginPINAttemptsShareProfileLimitDB + +# Log partitions: migration 028 and partman cut the same UTC periods on a +# server whose TimeZone is not UTC, so no write falls into the default. +./internal/database TestLogPartitionsMatchPartmanOnNonUTCServer + +# Search normalization: the number-token function keeps its output and stays +# inlinable. +./migrations TestSearchNumberTokenInlinesPostgres + +# Indexes left invalid by an interrupted concurrent build are rebuilt without +# blocking reads or writes on their tables. +./internal/database TestInvalidIndexRebuildKeepsTablesAvailablePostgres + +# Pool connections run with JIT off unless the server, database, role or +# DATABASE_URL already chose jit. +./internal/database TestNewPoolDisablesJITUnlessConfiguredPostgres + +# Operational log pages: a cursor prunes the partitions newer than it. +./internal/opslog TestListCursorPrunesNewerPartitionsDB