Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/buzz-admin/tests/postgres_user_inventory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() {
for target in [community, participant_community] {
for _ in 0..2 {
let channel = Uuid::new_v4();
sqlx::query("INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'inventory-test', decode($3, 'hex'))")
sqlx::query("INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'inventory-test-' || $2::uuid::text, decode($3, 'hex'))")
.bind(target).bind(channel).bind(pubkey).execute(&mut *tx).await.unwrap();
sqlx::query("INSERT INTO channel_members (community_id, channel_id, pubkey) VALUES ($1, $2, decode($3, 'hex'))")
.bind(target).bind(channel).bind(pubkey).execute(&mut *tx).await.unwrap();
Expand Down
2 changes: 1 addition & 1 deletion crates/buzz-db/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ pub enum DbError {
ChannelNotFound(uuid::Uuid),

/// Another listed open channel in the community already has this name
/// (see `channel::ensure_listed_channel_name_unique`).
/// (the `idx_channels_listed_name` index, migration 0083).
#[error("an open channel named \"{0}\" already exists")]
ChannelNameTaken(String),

Expand Down
29 changes: 28 additions & 1 deletion crates/buzz-db/src/runtime/migration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -778,14 +778,26 @@ mod postgres_tests {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);

assert_eq!(migrations.len(), 82);
assert_eq!(migrations.len(), 83);
for (index, migration) in migrations.iter().enumerate() {
assert_eq!(migration.version, (index + 1) as i64);
}
assert!(migrations[79]
.sql
.as_str()
.contains("ADD COLUMN auto_join BOOLEAN NOT NULL DEFAULT false"));
// Listed channel names (0083): a bounded contract index that refuses
// to build over unrepaired duplicates. The desired schema carries the
// same plain index.
assert_eq!(migrations[82].version, 83);
let listed_names = strip_sql_comments(migrations[82].sql.as_str());
assert!(listed_names.contains("SET LOCAL lock_timeout = '1s';"));
assert!(listed_names.contains("SET LOCAL statement_timeout = '5s';"));
assert!(listed_names.contains("CREATE UNIQUE INDEX idx_channels_listed_name"));
assert!(!listed_names.contains("CONCURRENTLY"));
assert!(crate::test_support::desired_state_schema_sql().contains(
"CREATE UNIQUE INDEX idx_channels_listed_name ON channels (community_id, lower(name))"
));
assert!(migrations[70]
.sql
.as_str()
Expand Down Expand Up @@ -2924,6 +2936,21 @@ mod postgres_tests {
"column parity mismatch for channels: schema.sql desired state has drifted \
from the migrations; update schema/tables/public/channels.sql to match"
);
// The listed-name index (0083) is compared on its own for that reason.
let listed_name = |shapes: Vec<(String, String, String)>| {
shapes
.into_iter()
.filter(|(name, _, _)| name == "idx_channels_listed_name")
.collect::<Vec<_>>()
};
let desired_listed_name = listed_name(index_shapes(&desired, "channels").await);
assert_eq!(desired_listed_name.len(), 1, "desired state has the index");
assert_eq!(
desired_listed_name,
listed_name(index_shapes(&migrated, "channels").await),
"idx_channels_listed_name parity mismatch: update \
schema/tables/public/channels.sql to match migration 0083"
);
assert_eq!(
events_key_defs(&desired).await,
events_key_defs(&migrated).await,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,18 @@
//! renames that leave no two listed channels in a community with one name,
//! ignoring case. It keeps the name on the oldest unarchived channel and picks
//! a new name that no other listed channel has.
//!
//! Migration 0083 then adds the unique index. It refuses to run over
//! duplicates that were not renamed. Otherwise it builds the index itself on a
//! populated table under its lock budget, accepts a valid concurrent prebuild
//! without replacing it, and rejects an invalid one.

use super::*;

/// The last migration before 0083. Duplicates can exist, and the plan has
/// work to do, only before the index does.
const BEFORE_INDEX: i64 = 82;

const PLAN_SQL: &str =
include_str!("../../../../../scripts/maintenance/plan_duplicate_listed_channel_renames.sql");

Expand Down Expand Up @@ -64,7 +73,7 @@ fn renamed(id: Uuid, name: &str) -> (Uuid, String) {

/// The plan for `host`: (channel, current name, new name), as the rename
/// script reads it.
pub(super) async fn rename_plan(pool: &PgPool, host: &str) -> Vec<(Uuid, String, String)> {
async fn rename_plan(pool: &PgPool, host: &str) -> Vec<(Uuid, String, String)> {
rename_plan_with_starters(pool, host, &[]).await
}

Expand Down Expand Up @@ -92,7 +101,7 @@ async fn rename_plan_with_starters(

/// Applies the plan the way the relay applies each rename the script sends,
/// and returns how many channels it renamed.
pub(super) async fn rename_duplicates(pool: &PgPool, host: &str) -> usize {
async fn rename_duplicates(pool: &PgPool, host: &str) -> usize {
let plan = rename_plan(pool, host).await;
for (channel, _, new_name) in &plan {
sqlx::query("UPDATE channels SET name = $2 WHERE id = $1")
Expand All @@ -108,7 +117,8 @@ pub(super) async fn rename_duplicates(pool: &PgPool, host: &str) -> usize {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn plan_keeps_each_listed_name_on_its_oldest_unarchived_channel() {
let (pool, _scratch) = ScratchDb::migrated(&admin_url().await, "listed_name_plan").await;
let (pool, _scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_plan", BEFORE_INDEX).await;
let seed = Seed::new(&pool).await;
let [kept, archived, newer, private, deleted] = [(); 5].map(|_| Uuid::new_v4());
seed.open(&pool, kept, "square-ios", 30).await;
Expand Down Expand Up @@ -187,7 +197,8 @@ async fn plan_keeps_each_listed_name_on_its_oldest_unarchived_channel() {
#[ignore = "requires Postgres"]
async fn plan_frees_a_starter_name_that_only_archived_channels_hold() {
let (pool, _scratch) =
ScratchDb::migrated(&admin_url().await, "listed_name_plan_starter").await;
ScratchDb::migrated_through(&admin_url().await, "listed_name_plan_starter", BEFORE_INDEX)
.await;
let seed = Seed::new(&pool).await;
let [general, welcome_old, welcome_new, random] = [(); 4].map(|_| Uuid::new_v4());
seed.open(&pool, general, "General", 30).await;
Expand Down Expand Up @@ -299,7 +310,9 @@ async fn plan_frees_a_starter_name_that_only_archived_channels_hold() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn plan_picks_a_free_name_when_the_short_one_is_taken() {
let (pool, _scratch) = ScratchDb::migrated(&admin_url().await, "listed_name_plan_taken").await;
let (pool, _scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_plan_taken", BEFORE_INDEX)
.await;
let seed = Seed::new(&pool).await;
// The longest name the column holds, so the new name must truncate it.
let long = "a".repeat(255);
Expand Down Expand Up @@ -381,7 +394,8 @@ async fn run_rename_script(
#[tokio::test]
#[ignore = "requires Postgres"]
async fn script_renames_each_planned_channel_once_whatever_its_name() {
let (pool, scratch) = ScratchDb::migrated(&admin_url().await, "listed_name_script").await;
let (pool, scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_script", BEFORE_INDEX).await;
let seed = Seed::new(&pool).await;
let [kept, duplicate, other] = [(); 3].map(|_| Uuid::new_v4());
let tricky = format!("-ops\x1e{other}\x1fowned\x1f\n\t\"$(x)");
Expand Down Expand Up @@ -428,7 +442,9 @@ async fn script_renames_each_planned_channel_once_whatever_its_name() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn script_stops_at_the_first_auth_refusal() {
let (pool, scratch) = ScratchDb::migrated(&admin_url().await, "listed_name_script_auth").await;
let (pool, scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_script_auth", BEFORE_INDEX)
.await;
let seed = Seed::new(&pool).await;
for (group, age) in [("first", 30), ("first", 10), ("second", 30), ("second", 10)] {
seed.open(&pool, Uuid::new_v4(), group, age).await;
Expand All @@ -449,7 +465,9 @@ async fn script_stops_at_the_first_auth_refusal() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn script_stops_for_a_community_the_relay_does_not_serve() {
let (pool, scratch) = ScratchDb::migrated(&admin_url().await, "listed_name_script_state").await;
let (pool, scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_script_state", BEFORE_INDEX)
.await;
let archived = Seed::new(&pool).await;
let deleting = Seed::new(&pool).await;
for seed in [&archived, &deleting] {
Expand Down Expand Up @@ -479,3 +497,142 @@ async fn script_stops_for_a_community_the_relay_does_not_serve() {
assert!(stderr.contains(message), "{host}: {stderr}");
}
}

async fn ledger_version(pool: &PgPool) -> i64 {
sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations")
.fetch_one(pool)
.await
.expect("read migration ledger")
}

async fn index_oid(pool: &PgPool) -> Option<i64> {
sqlx::query_scalar("SELECT to_regclass('public.idx_channels_listed_name')::oid::bigint")
.fetch_one(pool)
.await
.expect("look up index")
}

#[tokio::test]
#[ignore = "requires Postgres"]
async fn migration_schema_listed_name_index_needs_the_renames_then_builds_inline() {
let (pool, _scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_index", BEFORE_INDEX).await;
let seed = Seed::new(&pool).await;
let (kept, newer) = (Uuid::new_v4(), Uuid::new_v4());
seed.open(&pool, kept, "square-ios", 30).await;
seed.open(&pool, newer, "Square-iOS", 10).await;
let before = seed.names(&pool).await;

let error = migration::run_migrations(&pool).await.unwrap_err();
assert!(
error.to_string().contains("duplicate names remain"),
"must name the duplicate rows, not merely fail: {error}"
);
assert_eq!(ledger_version(&pool).await, BEFORE_INDEX);
assert_eq!(index_oid(&pool).await, None);
assert_eq!(seed.names(&pool).await, before, "no data changes");

assert_eq!(rename_duplicates(&pool, &seed.host).await, 1);
assert!(rename_plan(&pool, &seed.host).await.is_empty());

// A channel writer that holds its lock past the 1s lock budget makes the
// inline build fail closed: nothing is recorded and startup can retry.
let mut writer = pool.begin().await.unwrap();
sqlx::query("LOCK TABLE channels IN ROW EXCLUSIVE MODE")
.execute(&mut *writer)
.await
.unwrap();
let error = migration::run_migrations(&pool).await.unwrap_err();
assert!(
error.to_string().contains("lock timeout"),
"the build must not queue behind a writer: {error}"
);
writer.rollback().await.unwrap();
assert_eq!(ledger_version(&pool).await, BEFORE_INDEX);
assert_eq!(index_oid(&pool).await, None);

// With no writer in the way, the populated table migrates with no
// operator step: the migration builds the index itself.
migration::run_migrations(&pool)
.await
.expect("migrate a populated table after the renames");
assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1);
let mut expected = vec![
(kept, "square-ios".to_owned()),
renamed(newer, "Square-iOS"),
];
expected.sort();
assert_eq!(seed.names(&pool).await, expected);

// The database now refuses what an older relay could still write.
let duplicate = sqlx::query(
"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) \
VALUES ($1, $2, 'SQUARE-IOS', 'forum', 'open', $3)",
)
.bind(Uuid::new_v4())
.bind(seed.community)
.bind(&seed.created_by)
.execute(&pool)
.await
.unwrap_err();
assert_eq!(
duplicate
.as_database_error()
.and_then(|e| e.constraint())
.map(str::to_owned),
Some("idx_channels_listed_name".to_owned())
);
}

#[tokio::test]
#[ignore = "requires Postgres"]
async fn migration_schema_listed_name_index_validates_a_prebuild() {
let (pool, _scratch) =
ScratchDb::migrated_through(&admin_url().await, "listed_name_prebuild", BEFORE_INDEX).await;
for remnant in [
// A different predicate: archived channels left out.
"CREATE UNIQUE INDEX idx_channels_listed_name ON channels (community_id, lower(name)) \
WHERE visibility = 'open' AND NOT unlisted AND archived_at IS NULL",
// A failed concurrent build. Simulated only in this disposable
// superuser database.
"CREATE UNIQUE INDEX idx_channels_listed_name ON channels (community_id, lower(name)) \
WHERE visibility = 'open' AND NOT unlisted \
AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL; \
UPDATE pg_index SET indisvalid = false \
WHERE indexrelid = 'idx_channels_listed_name'::regclass",
] {
sqlx::raw_sql(sqlx::AssertSqlSafe(remnant))
.execute(&pool)
.await
.unwrap();
let error = migration::run_migrations(&pool).await.unwrap_err();
assert!(
error
.to_string()
.contains("is invalid or has a different definition"),
"must reject the catalog shape: {error}"
);
assert_eq!(ledger_version(&pool).await, BEFORE_INDEX);
sqlx::query("DROP INDEX CONCURRENTLY idx_channels_listed_name")
.execute(&pool)
.await
.unwrap();
}

crate::test_support::prebuild_listed_channel_name_index(&pool).await;
let oid = index_oid(&pool).await;
// A channel writer holds its lock through the migration. Over a prebuild,
// 0083 must not request the ShareLock a CREATE INDEX takes, even with
// IF NOT EXISTS: that waits behind writers and queues new ones.
let mut writer = pool.begin().await.unwrap();
sqlx::query("LOCK TABLE channels IN ROW EXCLUSIVE MODE")
.execute(&mut *writer)
.await
.unwrap();
migration::run_migrations(&pool)
.await
.expect("accept the prebuild without waiting for channel writers");
writer.rollback().await.unwrap();
assert_eq!(ledger_version(&pool).await, BEFORE_INDEX + 1);
assert_eq!(index_oid(&pool).await, oid, "the prebuild is not replaced");
}
Loading
Loading