566 lines
19 KiB
Go
566 lines
19 KiB
Go
|
package database
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"database/sql"
|
||
|
|
"fmt"
|
||
|
|
"os"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
"github.com/jackc/pgx/v5/stdlib"
|
||
|
|
|
||
|
|
"github.com/Silo-Server/silo-server/internal/jellycompat/displayprefs"
|
||
|
|
"github.com/Silo-Server/silo-server/migrations"
|
||
|
|
)
|
||
|
|
|
||
|
|
// TestPostgresDisplayPrefsMove runs the real goose provider — which registers
|
||
|
|
// the Go move migration — against a real database, then exercises the move
|
||
|
|
// directly over seeded legacy rows. The parsing rules are unit-tested in
|
||
|
|
// internal/jellycompat/displayprefs; this covers what only a live database
|
||
|
|
// shows: registration, the table's constraints, verbatim copy through real
|
||
|
|
// text columns, and that re-running or rolling back behaves.
|
||
|
|
func TestPostgresDisplayPrefsMove(t *testing.T) {
|
||
|
|
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
|
||
|
|
if dsn == "" {
|
||
|
|
t.Skip("SILO_TEST_DATABASE_URL is not set")
|
||
|
|
}
|
||
|
|
ctx := context.Background()
|
||
|
|
|
||
|
|
pool, err := pgxpool.New(ctx, dsn)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("connect test database: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(pool.Close)
|
||
|
|
|
||
|
|
// Migrate first, then seed legacy rows and run the move directly: the
|
||
|
|
// goose version gate has already consumed the registered migration, so
|
||
|
|
// calling the function is how the upgrade path is exercised against data.
|
||
|
|
if err := RunMigrations(ctx, pool, migrations.FS, "sql"); err != nil {
|
||
|
|
t.Fatalf("initial migration: %v", err)
|
||
|
|
}
|
||
|
|
userID := seedLegacyDisplayPrefsRows(ctx, t, pool)
|
||
|
|
|
||
|
|
sqlDB := stdlib.OpenDBFromPool(pool)
|
||
|
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||
|
|
runMove := func(fn func(context.Context, *sql.Tx) error, label string) {
|
||
|
|
t.Helper()
|
||
|
|
tx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("begin %s: %v", label, err)
|
||
|
|
}
|
||
|
|
if err := fn(ctx, tx); err != nil {
|
||
|
|
t.Fatalf("%s: %v", label, err)
|
||
|
|
}
|
||
|
|
if err := tx.Commit(); err != nil {
|
||
|
|
t.Fatalf("commit %s: %v", label, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
runMove(moveDisplayPrefs, "moveDisplayPrefs")
|
||
|
|
|
||
|
|
countLegacy := func() int {
|
||
|
|
var count int
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM user_settings
|
||
|
|
WHERE user_id = $1 AND key LIKE 'jellycompat:%'`, userID).Scan(&count); err != nil {
|
||
|
|
t.Fatalf("counting legacy rows: %v", err)
|
||
|
|
}
|
||
|
|
return count
|
||
|
|
}
|
||
|
|
|
||
|
|
t.Run("blobs move verbatim", func(t *testing.T) {
|
||
|
|
var value string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM jellycompat_displayprefs
|
||
|
|
WHERE user_id = $1 AND prefs_id = 'usersettings' AND client = 'emby'`, userID).
|
||
|
|
Scan(&value); err != nil {
|
||
|
|
t.Fatalf("reading moved blob: %v", err)
|
||
|
|
}
|
||
|
|
if value != `{"SortBy":"SortName", "CustomPrefs":{"b":"2","a":"1"}}` {
|
||
|
|
t.Errorf("blob = %q, want it byte-for-byte", value)
|
||
|
|
}
|
||
|
|
|
||
|
|
// The empty client is a real identity and must survive the key split.
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM jellycompat_displayprefs
|
||
|
|
WHERE user_id = $1 AND prefs_id = 'f137a2dd' AND client = ''`, userID).
|
||
|
|
Scan(&value); err != nil {
|
||
|
|
t.Fatalf("reading empty-client blob: %v", err)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
t.Run("user_settings keeps no jellycompat tenants", func(t *testing.T) {
|
||
|
|
if count := countLegacy(); count != 0 {
|
||
|
|
t.Errorf("%d jellycompat rows still ride user_settings", count)
|
||
|
|
}
|
||
|
|
var theme string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM user_settings WHERE user_id = $1 AND key = 'ui_theme'`, userID).
|
||
|
|
Scan(&theme); err != nil || theme != "cobalt-studio" {
|
||
|
|
t.Errorf("ui_theme = (%q, %v); the move touched a non-jellycompat row", theme, err)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
t.Run("unparseable rows are recorded, not silently deleted", func(t *testing.T) {
|
||
|
|
var value, reason string
|
||
|
|
err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value, reason FROM user_setting_migration_rejects
|
||
|
|
WHERE user_id = $1 AND source_table = 'user_settings' AND source_key = 'jellycompat:stray'`,
|
||
|
|
userID).Scan(&value, &reason)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("the stray row was dropped rather than recorded: %v", err)
|
||
|
|
}
|
||
|
|
if value != "not a displayprefs blob" || reason == "" {
|
||
|
|
t.Errorf("reject = (%q, %q); the original value and a reason must survive", value, reason)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
t.Run("a second run is a no-op", func(t *testing.T) {
|
||
|
|
counts := func() (blobs, rejects int) {
|
||
|
|
t.Helper()
|
||
|
|
if err := pool.QueryRow(ctx,
|
||
|
|
`SELECT COUNT(*) FROM jellycompat_displayprefs WHERE user_id = $1`, userID).
|
||
|
|
Scan(&blobs); err != nil {
|
||
|
|
t.Fatalf("counting blobs: %v", err)
|
||
|
|
}
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM user_setting_migration_rejects
|
||
|
|
WHERE user_id = $1 AND source_key LIKE 'jellycompat:%'`, userID).Scan(&rejects); err != nil {
|
||
|
|
t.Fatalf("counting rejects: %v", err)
|
||
|
|
}
|
||
|
|
return blobs, rejects
|
||
|
|
}
|
||
|
|
blobsBefore, rejectsBefore := counts()
|
||
|
|
|
||
|
|
runMove(moveDisplayPrefs, "moveDisplayPrefs re-run")
|
||
|
|
|
||
|
|
blobsAfter, rejectsAfter := counts()
|
||
|
|
if blobsAfter != blobsBefore || rejectsAfter != rejectsBefore {
|
||
|
|
t.Errorf("re-run changed counts: blobs %d→%d, rejects %d→%d",
|
||
|
|
blobsBefore, blobsAfter, rejectsBefore, rejectsAfter)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
t.Run("rollback restores the legacy rows", func(t *testing.T) {
|
||
|
|
runMove(unmoveDisplayPrefs, "unmoveDisplayPrefs")
|
||
|
|
|
||
|
|
if count := countLegacy(); count != 3 {
|
||
|
|
t.Errorf("rollback restored %d legacy rows, want 3", count)
|
||
|
|
}
|
||
|
|
var value string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM user_settings
|
||
|
|
WHERE user_id = $1 AND key = 'jellycompat:displayprefs:usersettings:emby'`, userID).
|
||
|
|
Scan(&value); err != nil {
|
||
|
|
t.Fatalf("reading restored blob row: %v", err)
|
||
|
|
}
|
||
|
|
if value != `{"SortBy":"SortName", "CustomPrefs":{"b":"2","a":"1"}}` {
|
||
|
|
t.Errorf("restored blob = %q, want it byte-for-byte", value)
|
||
|
|
}
|
||
|
|
|
||
|
|
var blobs int
|
||
|
|
if err := pool.QueryRow(ctx,
|
||
|
|
`SELECT COUNT(*) FROM jellycompat_displayprefs WHERE user_id = $1`, userID).
|
||
|
|
Scan(&blobs); err != nil {
|
||
|
|
t.Fatalf("counting blobs after rollback: %v", err)
|
||
|
|
}
|
||
|
|
if blobs != 0 {
|
||
|
|
t.Errorf("rollback left %d rows in jellycompat_displayprefs", blobs)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestPostgresDisplayPrefsMoveDoesNotDeleteConcurrentWrites reproduces the
|
||
|
|
// rolling-deploy window: the goose lock only excludes other migrators, so an
|
||
|
|
// old-binary app instance can commit a jellycompat row into user_settings
|
||
|
|
// while the move transaction sits between its SELECT and its deletes. Under
|
||
|
|
// READ COMMITTED each statement snapshots independently, so a pattern-based
|
||
|
|
// DELETE would see — and destroy — a row the SELECT never copied. The move
|
||
|
|
// must instead delete only the exact rows it read, leaving the late row
|
||
|
|
// stranded in user_settings for a re-run to pick up.
|
||
|
|
//
|
||
|
|
// The stall is real: an uncommitted conflicting insert on
|
||
|
|
// jellycompat_displayprefs blocks the move's ON CONFLICT insert, the late
|
||
|
|
// legacy row commits during the stall, and releasing the blocker lets the
|
||
|
|
// move finish.
|
||
|
|
func TestPostgresDisplayPrefsMoveDoesNotDeleteConcurrentWrites(t *testing.T) {
|
||
|
|
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
|
||
|
|
if dsn == "" {
|
||
|
|
t.Skip("SILO_TEST_DATABASE_URL is not set")
|
||
|
|
}
|
||
|
|
ctx := context.Background()
|
||
|
|
|
||
|
|
pool, err := pgxpool.New(ctx, dsn)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("connect test database: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(pool.Close)
|
||
|
|
if err := RunMigrations(ctx, pool, migrations.FS, "sql"); err != nil {
|
||
|
|
t.Fatalf("initial migration: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
var userID int
|
||
|
|
err = pool.QueryRow(ctx, `
|
||
|
|
INSERT INTO users (username, email, password_hash, role)
|
||
|
|
VALUES ('displayprefs-racetest', 'displayprefs-racetest@example.com', 'x', 'user')
|
||
|
|
ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email
|
||
|
|
RETURNING id`).Scan(&userID)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("seeding user: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(func() {
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
|
||
|
|
})
|
||
|
|
for _, stmt := range []string{
|
||
|
|
`DELETE FROM user_settings WHERE user_id = $1`,
|
||
|
|
`DELETE FROM jellycompat_displayprefs WHERE user_id = $1`,
|
||
|
|
`DELETE FROM user_setting_migration_rejects WHERE user_id = $1`,
|
||
|
|
} {
|
||
|
|
if _, err := pool.Exec(ctx, stmt, userID); err != nil {
|
||
|
|
t.Fatalf("clearing prior rows: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
INSERT INTO user_settings (user_id, key, value) VALUES ($1, $2, $3)`,
|
||
|
|
userID, "jellycompat:displayprefs:usersettings:emby", `{"SortBy":"SortName"}`); err != nil {
|
||
|
|
t.Fatalf("seeding legacy row: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
sqlDB := stdlib.OpenDBFromPool(pool)
|
||
|
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||
|
|
|
||
|
|
// The blocker: an uncommitted insert on the identity the seeded legacy row
|
||
|
|
// maps to, so the move's copy insert waits on this transaction.
|
||
|
|
blockerTx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("begin blocker: %v", err)
|
||
|
|
}
|
||
|
|
blockerReleased := false
|
||
|
|
defer func() {
|
||
|
|
if !blockerReleased {
|
||
|
|
_ = blockerTx.Rollback()
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
if _, err := blockerTx.ExecContext(ctx, `
|
||
|
|
INSERT INTO jellycompat_displayprefs (user_id, prefs_id, client, value)
|
||
|
|
VALUES ($1, 'usersettings', 'emby', 'blocker')`, userID); err != nil {
|
||
|
|
t.Fatalf("blocker insert: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
moveDone := make(chan error, 1)
|
||
|
|
go func() {
|
||
|
|
tx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
moveDone <- fmt.Errorf("begin move: %w", err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
if err := moveDisplayPrefs(ctx, tx); err != nil {
|
||
|
|
_ = tx.Rollback()
|
||
|
|
moveDone <- fmt.Errorf("moveDisplayPrefs: %w", err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
moveDone <- tx.Commit()
|
||
|
|
}()
|
||
|
|
|
||
|
|
// Wait until the move transaction is provably parked on the blocker's
|
||
|
|
// lock: its SELECT over user_settings has happened, its deletes have not.
|
||
|
|
waitDeadline := time.Now().Add(10 * time.Second)
|
||
|
|
for {
|
||
|
|
var waiting int
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM pg_stat_activity
|
||
|
|
WHERE wait_event_type = 'Lock' AND query LIKE '%jellycompat_displayprefs%'`).
|
||
|
|
Scan(&waiting); err != nil {
|
||
|
|
t.Fatalf("polling pg_stat_activity: %v", err)
|
||
|
|
}
|
||
|
|
if waiting > 0 {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if time.Now().After(waitDeadline) {
|
||
|
|
t.Fatal("the move never blocked on the conflicting insert")
|
||
|
|
}
|
||
|
|
time.Sleep(20 * time.Millisecond)
|
||
|
|
}
|
||
|
|
|
||
|
|
// The old binary's handler commits a fresh DisplayPreferences row now —
|
||
|
|
// after the move's SELECT, before its deletes.
|
||
|
|
const lateKey = "jellycompat:displayprefs:late:acme"
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
INSERT INTO user_settings (user_id, key, value) VALUES ($1, $2, $3)`,
|
||
|
|
userID, lateKey, `{"SortBy":"DateCreated"}`); err != nil {
|
||
|
|
t.Fatalf("committing the late legacy row: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// And it updates a row the move has already read: the delete predicate
|
||
|
|
// pins the value the SELECT saw, so this newer write must survive too.
|
||
|
|
const updatedKey = "jellycompat:displayprefs:usersettings:emby"
|
||
|
|
const updatedValue = `{"SortBy":"Runtime"}`
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
UPDATE user_settings SET value = $3 WHERE user_id = $1 AND key = $2`,
|
||
|
|
userID, updatedKey, updatedValue); err != nil {
|
||
|
|
t.Fatalf("committing the late update: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
blockerReleased = true
|
||
|
|
if err := blockerTx.Rollback(); err != nil {
|
||
|
|
t.Fatalf("releasing blocker: %v", err)
|
||
|
|
}
|
||
|
|
if err := <-moveDone; err != nil {
|
||
|
|
t.Fatalf("move under contention: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// The late row must not have been destroyed: it was never copied, so it
|
||
|
|
// must still ride user_settings.
|
||
|
|
var lateRows int
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM user_settings WHERE user_id = $1 AND key = $2`,
|
||
|
|
userID, lateKey).Scan(&lateRows); err != nil {
|
||
|
|
t.Fatalf("counting the late row: %v", err)
|
||
|
|
}
|
||
|
|
if lateRows != 1 {
|
||
|
|
var copied, rejected int
|
||
|
|
_ = pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM jellycompat_displayprefs
|
||
|
|
WHERE user_id = $1 AND prefs_id = 'late'`, userID).Scan(&copied)
|
||
|
|
_ = pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM user_setting_migration_rejects
|
||
|
|
WHERE user_id = $1 AND source_key = $2`, userID, lateKey).Scan(&rejected)
|
||
|
|
t.Fatalf("late row gone from user_settings (copied=%d rejected=%d): "+
|
||
|
|
"a concurrently committed row was deleted without being moved",
|
||
|
|
copied, rejected)
|
||
|
|
}
|
||
|
|
|
||
|
|
// The concurrently updated row must also still be in user_settings, with
|
||
|
|
// the newer value: the move copied the old value but its delete named that
|
||
|
|
// old value, so it must not have matched the updated row.
|
||
|
|
var survivingValue string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM user_settings WHERE user_id = $1 AND key = $2`,
|
||
|
|
userID, updatedKey).Scan(&survivingValue); err != nil {
|
||
|
|
t.Fatalf("updated legacy row gone from user_settings: %v — "+
|
||
|
|
"a concurrently updated row was deleted with only its old value moved", err)
|
||
|
|
}
|
||
|
|
if survivingValue != updatedValue {
|
||
|
|
t.Errorf("surviving legacy value = %q, want the late update %q",
|
||
|
|
survivingValue, updatedValue)
|
||
|
|
}
|
||
|
|
|
||
|
|
// A re-run — which the stranded row exists to allow — picks it up.
|
||
|
|
tx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("begin re-run: %v", err)
|
||
|
|
}
|
||
|
|
if err := moveDisplayPrefs(ctx, tx); err != nil {
|
||
|
|
t.Fatalf("re-run: %v", err)
|
||
|
|
}
|
||
|
|
if err := tx.Commit(); err != nil {
|
||
|
|
t.Fatalf("commit re-run: %v", err)
|
||
|
|
}
|
||
|
|
var copied int
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM jellycompat_displayprefs
|
||
|
|
WHERE user_id = $1 AND prefs_id = 'late' AND client = 'acme'`, userID).Scan(&copied); err != nil {
|
||
|
|
t.Fatalf("counting the re-run copy: %v", err)
|
||
|
|
}
|
||
|
|
if copied != 1 {
|
||
|
|
t.Errorf("re-run copied %d late rows, want 1", copied)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// TestPostgresDisplayPrefsRollbackDoesNotDeleteConcurrentUpdates reproduces
|
||
|
|
// the inverse rolling-deploy window: a new-binary app instance updates a blob
|
||
|
|
// after the rollback has read it but before the rollback deletes it. The
|
||
|
|
// rollback may restore its older snapshot to user_settings, but it must leave
|
||
|
|
// the newer canonical value in place rather than deleting a value it never
|
||
|
|
// restored.
|
||
|
|
func TestPostgresDisplayPrefsRollbackDoesNotDeleteConcurrentUpdates(t *testing.T) {
|
||
|
|
dsn := os.Getenv("SILO_TEST_DATABASE_URL")
|
||
|
|
if dsn == "" {
|
||
|
|
t.Skip("SILO_TEST_DATABASE_URL is not set")
|
||
|
|
}
|
||
|
|
ctx := context.Background()
|
||
|
|
|
||
|
|
pool, err := pgxpool.New(ctx, dsn)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("connect test database: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(pool.Close)
|
||
|
|
if err := RunMigrations(ctx, pool, migrations.FS, "sql"); err != nil {
|
||
|
|
t.Fatalf("initial migration: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
var userID int
|
||
|
|
err = pool.QueryRow(ctx, `
|
||
|
|
INSERT INTO users (username, email, password_hash, role)
|
||
|
|
VALUES ('displayprefs-rollback-racetest', 'displayprefs-rollback-racetest@example.com', 'x', 'user')
|
||
|
|
ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email
|
||
|
|
RETURNING id`).Scan(&userID)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("seeding user: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(func() {
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
|
||
|
|
})
|
||
|
|
for _, stmt := range []string{
|
||
|
|
`DELETE FROM user_settings WHERE user_id = $1`,
|
||
|
|
`DELETE FROM jellycompat_displayprefs WHERE user_id = $1`,
|
||
|
|
`DELETE FROM user_setting_migration_rejects WHERE user_id = $1`,
|
||
|
|
} {
|
||
|
|
if _, err := pool.Exec(ctx, stmt, userID); err != nil {
|
||
|
|
t.Fatalf("clearing prior rows: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
const (
|
||
|
|
prefsID = "usersettings"
|
||
|
|
client = "emby"
|
||
|
|
originalValue = `{"SortBy":"SortName"}`
|
||
|
|
updatedValue = `{"SortBy":"Runtime"}`
|
||
|
|
)
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
INSERT INTO jellycompat_displayprefs (user_id, prefs_id, client, value)
|
||
|
|
VALUES ($1, $2, $3, $4)`, userID, prefsID, client, originalValue); err != nil {
|
||
|
|
t.Fatalf("seeding moved row: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
sqlDB := stdlib.OpenDBFromPool(pool)
|
||
|
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||
|
|
|
||
|
|
// Hold the destination identity open so the rollback stalls after reading
|
||
|
|
// the canonical row but before its insert and delete.
|
||
|
|
blockerTx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("begin blocker: %v", err)
|
||
|
|
}
|
||
|
|
blockerReleased := false
|
||
|
|
defer func() {
|
||
|
|
if !blockerReleased {
|
||
|
|
_ = blockerTx.Rollback()
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
legacyKey := displayprefs.LegacyKey(prefsID, client)
|
||
|
|
if _, err := blockerTx.ExecContext(ctx, `
|
||
|
|
INSERT INTO user_settings (user_id, key, value)
|
||
|
|
VALUES ($1, $2, 'blocker')`, userID, legacyKey); err != nil {
|
||
|
|
t.Fatalf("blocker insert: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
rollbackDone := make(chan error, 1)
|
||
|
|
go func() {
|
||
|
|
tx, err := sqlDB.BeginTx(ctx, nil)
|
||
|
|
if err != nil {
|
||
|
|
rollbackDone <- fmt.Errorf("begin rollback: %w", err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
if err := unmoveDisplayPrefs(ctx, tx); err != nil {
|
||
|
|
_ = tx.Rollback()
|
||
|
|
rollbackDone <- fmt.Errorf("unmoveDisplayPrefs: %w", err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
rollbackDone <- tx.Commit()
|
||
|
|
}()
|
||
|
|
|
||
|
|
waitDeadline := time.Now().Add(10 * time.Second)
|
||
|
|
for {
|
||
|
|
var waiting int
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT COUNT(*) FROM pg_stat_activity
|
||
|
|
WHERE wait_event_type = 'Lock' AND query LIKE '%INSERT INTO user_settings%'`).
|
||
|
|
Scan(&waiting); err != nil {
|
||
|
|
t.Fatalf("polling pg_stat_activity: %v", err)
|
||
|
|
}
|
||
|
|
if waiting > 0 {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if time.Now().After(waitDeadline) {
|
||
|
|
t.Fatal("the rollback never blocked on the conflicting insert")
|
||
|
|
}
|
||
|
|
time.Sleep(20 * time.Millisecond)
|
||
|
|
}
|
||
|
|
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
UPDATE jellycompat_displayprefs SET value = $4
|
||
|
|
WHERE user_id = $1 AND prefs_id = $2 AND client = $3`,
|
||
|
|
userID, prefsID, client, updatedValue); err != nil {
|
||
|
|
t.Fatalf("committing the concurrent canonical update: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
blockerReleased = true
|
||
|
|
if err := blockerTx.Rollback(); err != nil {
|
||
|
|
t.Fatalf("releasing blocker: %v", err)
|
||
|
|
}
|
||
|
|
if err := <-rollbackDone; err != nil {
|
||
|
|
t.Fatalf("rollback under contention: %v", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
var survivingValue string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM jellycompat_displayprefs
|
||
|
|
WHERE user_id = $1 AND prefs_id = $2 AND client = $3`,
|
||
|
|
userID, prefsID, client).Scan(&survivingValue); err != nil {
|
||
|
|
t.Fatalf("concurrently updated canonical row was deleted: %v", err)
|
||
|
|
}
|
||
|
|
if survivingValue != updatedValue {
|
||
|
|
t.Errorf("surviving canonical value = %q, want %q", survivingValue, updatedValue)
|
||
|
|
}
|
||
|
|
|
||
|
|
var restoredValue string
|
||
|
|
if err := pool.QueryRow(ctx, `
|
||
|
|
SELECT value FROM user_settings WHERE user_id = $1 AND key = $2`,
|
||
|
|
userID, legacyKey).Scan(&restoredValue); err != nil {
|
||
|
|
t.Fatalf("reading restored legacy snapshot: %v", err)
|
||
|
|
}
|
||
|
|
if restoredValue != originalValue {
|
||
|
|
t.Errorf("restored legacy value = %q, want the rollback snapshot %q",
|
||
|
|
restoredValue, originalValue)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// seedLegacyDisplayPrefsRows writes the pre-cutover user_settings rows: two
|
||
|
|
// handler-written DisplayPreferences blobs, one jellycompat row only the legacy
|
||
|
|
// settings API's removed unknown-key carve-out could have produced, and a real
|
||
|
|
// user setting that must not move.
|
||
|
|
func seedLegacyDisplayPrefsRows(ctx context.Context, t *testing.T, pool *pgxpool.Pool) int {
|
||
|
|
t.Helper()
|
||
|
|
|
||
|
|
var userID int
|
||
|
|
err := pool.QueryRow(ctx, `
|
||
|
|
INSERT INTO users (username, email, password_hash, role)
|
||
|
|
VALUES ('displayprefs-migtest', 'displayprefs-migtest@example.com', 'x', 'user')
|
||
|
|
ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email
|
||
|
|
RETURNING id`).Scan(&userID)
|
||
|
|
if err != nil {
|
||
|
|
t.Fatalf("seeding user: %v", err)
|
||
|
|
}
|
||
|
|
t.Cleanup(func() {
|
||
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
|
||
|
|
})
|
||
|
|
|
||
|
|
// Clear anything a prior run left so the assertions see only this seed.
|
||
|
|
for _, stmt := range []string{
|
||
|
|
`DELETE FROM user_settings WHERE user_id = $1`,
|
||
|
|
`DELETE FROM jellycompat_displayprefs WHERE user_id = $1`,
|
||
|
|
`DELETE FROM user_setting_migration_rejects WHERE user_id = $1`,
|
||
|
|
} {
|
||
|
|
if _, err := pool.Exec(ctx, stmt, userID); err != nil {
|
||
|
|
t.Fatalf("clearing prior rows: %v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
for key, value := range map[string]string{
|
||
|
|
"jellycompat:displayprefs:usersettings:emby": `{"SortBy":"SortName", "CustomPrefs":{"b":"2","a":"1"}}`,
|
||
|
|
"jellycompat:displayprefs:f137a2dd:": `{"SortBy":"DateCreated"}`,
|
||
|
|
"jellycompat:stray": "not a displayprefs blob",
|
||
|
|
"ui_theme": "cobalt-studio",
|
||
|
|
} {
|
||
|
|
if _, err := pool.Exec(ctx, `
|
||
|
|
INSERT INTO user_settings (user_id, key, value) VALUES ($1, $2, $3)
|
||
|
|
ON CONFLICT (user_id, key) DO UPDATE SET value = EXCLUDED.value`,
|
||
|
|
userID, key, value); err != nil {
|
||
|
|
t.Fatalf("seeding user_settings %s: %v", key, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return userID
|
||
|
|
}
|