fix(database): fix non-UTC log partitions and cut index and JIT overhead

.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

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:

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

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

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

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 {