jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-06 01:11:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): refuse a second listed open channel with the same name

Browse Source 
Two open channels in a community could share a name, so the channel
directory showed two "square-ios" entries with nothing to tell them
apart. The relay now refuses a create, rename, private-to-open switch or
unlisted-to-listed switch that would give a listed channel (an open, not
unlisted stream or forum that is not deleted) a name another listed
channel in the community already has, ignoring case. An archived channel
keeps its name. Private channels, DMs and other types may still repeat
names: nobody browses them side by side, and refusing a name a private
channel holds would reveal that the channel exists.

The check runs in the same transaction as the write, after it, under a
per-name advisory lock, so two writers claiming one name take turns.
Kind 9007 and 9002 report it as "invalid: an open channel named ...
already exists". The store also reports a write that the follow-up
unique index idx_channels_listed_name refuses as the same error, so this
release stays correct beside that index during its rollout and after a
rollback.

scripts/maintenance/rename_duplicate_listed_channel_names.sql renames
the duplicates that already exist. It locks a batch's rows before any
name lock, checks each locked row again, and gives it a name that no
listed channel has. `buzz-admin reconcile-channels --channel <id>
--metadata` republishes a renamed channel's kind:39000 so clients see
the new name. A live edit can race it, as it can race the relay's own
publisher, so it runs with no relay serving. It signs later than the
stored event, as the relay does, and fails unless the stored event
matches the channel row after its write.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-10-30 13:21:05 -04:00
parent 0c82435ae0
commit c49c29124e
11 changed files with 1148 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
crates/buzz-admin/src/main.rs
View File

@@ -137,12 +137,18 @@ enum Command {
///
/// Without `--channel`, only channels missing discovery metadata are
/// reconciled. With `--channel`, only that channel's member snapshot is
/// replaced; canonical metadata and admin events remain untouched.
/// replaced; canonical metadata and admin events remain untouched unless
/// `--metadata` is given.
ReconcileChannels {
/// Optional channel UUID to force-republish.
#[arg(long)]
channel: Option<String>,
/// With `--channel`, also republish its kind:39000 metadata from the
/// channel row, for example after an operator rename.
#[arg(long, requires = "channel")]
metadata: bool,
/// Relay private key (hex) for signing events. Falls back to
/// BUZZ_RELAY_PRIVATE_KEY env var. If neither is set, generates
/// an ephemeral key (events will be unverifiable after restart).
@@ -211,8 +217,12 @@ async fn run(cli: Cli) -> Result<i32> {
Command::Deletions { command } => deletions::run(command).await,
Command::UserDeletions { command } => user_deletions::run(command).await,
Command::NipFi { command } => nip_fi::run(command).await,
Command::ReconcileChannels { channel, relay_key } => {
reconcile_channels(channel, relay_key).await?;
Command::ReconcileChannels {
channel,
metadata,
relay_key,
} => {
reconcile_channels(channel, metadata, relay_key).await?;
Ok(0)
}
}
@@ -721,6 +731,7 @@ async fn resolve_admin_tenant(db: &Db) -> Result<TenantContext> {
async fn reconcile_channels(
channel_arg: Option<String>,
republish_metadata: bool,
relay_key_arg: Option<String>,
) -> Result<()> {
use buzz_core::channel::group_state_identity_tags;
@@ -800,11 +811,16 @@ async fn reconcile_channels(
let members = db.get_members(tenant.community(), channel.id).await?;
// A targeted repair is deliberately roster-only. kind:39000 metadata
// is richer than this legacy backfill builder, and kind:39001 is not
// part of the stale-roster incident; replacing either can destroy
// canonical state. Full backfill still creates all three event kinds
// A targeted repair is roster-only unless `--metadata` asks for
// kind:39000 too; kind:39001 is never replaced by a targeted repair.
// kind:39000 is built from the channel row by the same function the
// relay uses, so an operator rename (for example
// scripts/maintenance/rename_duplicate_listed_channel_names.sql)
// reaches clients. Full backfill still creates all three event kinds
// for channels with no discovery metadata.
if republish_metadata {
republish_channel_metadata(&db, &tenant, channel.id, &relay_keys).await?;
}
if target_channel.is_none() {
// kind:39000 — channel metadata
{
@@ -874,6 +890,74 @@ async fn reconcile_channels(
Ok(())
}
/// Makes the newest relay-signed kind:39000 of a channel match its row, for
/// example after an operator rename.
///
/// Run it with no relay serving: the relay builds kind:39000 from a channel
/// read outside the replacement (`emit_group_discovery_events`), so a live
/// edit between this command's read and write could leave the older
/// snapshot. The command signs later than the stored event, as the relay
/// does, so it also replaces an event from the same second. It fails unless
/// a read after its write finds the stored event matching the row.
async fn republish_channel_metadata(
db: &Db,
tenant: &TenantContext,
channel_id: uuid::Uuid,
relay_keys: &Keys,
) -> Result<()> {
let latest = || async {
let stored = db
.query_events(&buzz_db::event::EventQuery {
kinds: Some(vec![39000]),
pubkey: Some(relay_keys.public_key().to_bytes().to_vec()),
d_tag: Some(channel_id.to_string()),
limit: Some(1),
..buzz_db::event::EventQuery::for_community(tenant.community())
})
.await?;
Ok::<_, anyhow::Error>(stored.into_iter().next().map(|stored| stored.event))
};
let row_tags = || async {
let channel = db.get_channel(tenant.community(), channel_id).await?;
let members = db.get_members(tenant.community(), channel_id).await?;
Ok::<_, anyhow::Error>(buzz_db::channel::group_metadata_tags(&channel, &members))
};
let matches = |event: &nostr::Event, tags: &[Vec<String>]| {
event
.tags
.iter()
.map(|tag| tag.as_slice())
.eq(tags.iter().map(Vec::as_slice))
};
let tags = row_tags().await?;
let stored = latest().await?;
if stored.as_ref().is_some_and(|event| matches(event, &tags)) {
return Ok(());
}
let created_at = stored
.map_or(0, |event| event.created_at.as_secs() + 1)
.max(nostr::Timestamp::now().as_secs());
let event = EventBuilder::new(Kind::Custom(39000), "")
.tags(
tags.into_iter()
.map(Tag::parse)
.collect::<Result<Vec<_>, _>>()?,
)
.custom_created_at(nostr::Timestamp::from(created_at))
.sign_with_keys(relay_keys)
.map_err(|e| anyhow::anyhow!("sign kind:39000: {e}"))?;
db.replace_addressable_event(tenant.community(), &event, Some(channel_id))
.await?;
let tags = row_tags().await?;
if !latest().await?.is_some_and(|event| matches(&event, &tags)) {
return Err(anyhow::anyhow!(
"channel {channel_id} changed while its kind:39000 was republished; stop every relay and run the command again"
));
}
Ok(())
}
#[cfg(test)]
mod storage_snapshot_tests {
use std::sync::atomic::{AtomicUsize, Ordering};

111
crates/buzz-admin/tests/postgres_reconcile_metadata.rs Normal file
View File

@@ -0,0 +1,111 @@
//! `buzz-admin reconcile-channels --channel <id> --metadata` leaves the newest
//! relay-signed kind:39000 matching the channel row, also when an older
//! snapshot of the row holds the replacement with a newer timestamp.
use buzz_core::CommunityId;
use buzz_db::event::EventQuery;
use buzz_db::{Db, DbConfig};
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
use tokio::process::Command;
use uuid::Uuid;
fn admin_binary() -> std::ffi::OsString {
// Nextest relocates binaries when extracting an archive. Cargo's build-time
// path remains the fallback for local cargo test runs.
std::env::var_os("NEXTEST_BIN_EXE_buzz_admin")
.unwrap_or_else(|| env!("CARGO_BIN_EXE_buzz-admin").into())
}
fn name_tag(event: &nostr::Event) -> Option<&str> {
event.tags.iter().find_map(|tag| match tag.as_slice() {
[key, value, ..] if key == "name" => Some(value.as_str()),
_ => None,
})
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn metadata_republish_replaces_a_stale_snapshot_that_holds_the_replacement() {
let url = std::env::var("BUZZ_TEST_DATABASE_URL")
.expect("run through scripts/postgres-test-run.sh for an isolated database");
let db = Db::new(&DbConfig {
database_url: url.clone(),
..DbConfig::default()
})
.await
.expect("connect to test DB");
let community = Uuid::new_v4();
let host = format!("reconcile-{}.example", community.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community)
.bind(&host)
.execute(db.pool())
.await
.expect("insert community");
let community = CommunityId::from_uuid(community);
let channel = Uuid::new_v4();
sqlx::query(
"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) \
VALUES ($1, $2, 'square-ios-79f8e9d4', 'stream', 'open', $3)",
)
.bind(channel)
.bind(community.as_uuid())
.bind(Keys::generate().public_key().to_bytes().to_vec())
.execute(db.pool())
.await
.expect("insert renamed channel");
// The row before the repair renamed it, dated ahead of this command's
// clock: a relay that read the old row published it, signing later than
// the event it replaced. The command must still replace it.
let relay = Keys::generate();
let mut before_rename = db.get_channel(community, channel).await.unwrap();
before_rename.name = "square-ios".to_owned();
let stale = EventBuilder::new(Kind::Custom(39000), "")
.tags(
buzz_db::channel::group_metadata_tags(&before_rename, &[])
.into_iter()
.map(Tag::parse)
.collect::<Result<Vec<_>, _>>()
.unwrap(),
)
.custom_created_at(Timestamp::from(Timestamp::now().as_secs() + 60))
.sign_with_keys(&relay)
.unwrap();
let (_, inserted) = db
.replace_addressable_event(community, &stale, Some(channel))
.await
.unwrap();
assert!(inserted);
let output = Command::new(admin_binary())
.args(["reconcile-channels", "--channel", &channel.to_string()])
.args([
"--metadata",
"--relay-key",
&relay.secret_key().to_secret_hex(),
])
.env("DATABASE_URL", &url)
.env("RELAY_URL", format!("wss://{host}"))
.output()
.await
.expect("run buzz-admin");
assert!(
output.status.success(),
"reconcile-channels failed:\n{}",
String::from_utf8_lossy(&output.stderr)
);
let newest = db
.query_events(&EventQuery {
kinds: Some(vec![39000]),
d_tag: Some(channel.to_string()),
limit: Some(1),
..EventQuery::for_community(community)
})
.await
.unwrap();
let newest = &newest.first().expect("kind:39000 stored").event;
assert_eq!(name_tag(newest), Some("square-ios-79f8e9d4"));
assert!(newest.created_at > stale.created_at);
}

5
crates/buzz-db/src/error.rs
View File

@@ -25,6 +25,11 @@ pub enum DbError {
#[error("channel not found: {0}")]
ChannelNotFound(uuid::Uuid),
/// Another listed open channel in the community already has this name
/// (see `channel::ensure_listed_channel_name_unique`).
#[error("an open channel named \"{0}\" already exists")]
ChannelNameTaken(String),
/// The requested member is not in the channel.
#[error("member not found in channel {0}")]
MemberNotFound(uuid::Uuid),

3
crates/buzz-db/src/runtime/tests.rs
View File

@@ -4277,6 +4277,9 @@ async fn e_tag_any_runs_with_real_binds_on_p_join_and_count() {
#[path = "tests/thread_window_postgres_tests.rs"]
mod thread_window_postgres_tests;
#[path = "tests/listed_channel_name_postgres_tests.rs"]
mod listed_channel_name_postgres_tests;
#[path = "tests/reader_pool_postgres_tests.rs"]
mod reader_pool_postgres_tests;

226
crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs Normal file
View File

@@ -0,0 +1,226 @@
//! scripts/maintenance/rename_duplicate_listed_channel_names.sql leaves no two
//! listed channels in a community with one name, ignoring case. It keeps the
//! name on the oldest unarchived channel, picks a new name that is free, and
//! leaves alone a row that a relay changed while the script waited for it.
use super::*;
const REPAIR_SCRIPT: &str =
include_str!("../../../../../scripts/maintenance/rename_duplicate_listed_channel_names.sql");
struct Seed {
community: Uuid,
created_by: Vec<u8>,
}
impl Seed {
async fn new(pool: &PgPool) -> Self {
let community = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community)
.bind(format!("listed-name-{}.example", community.simple()))
.execute(pool)
.await
.expect("insert community");
Self {
community,
created_by: nostr::Keys::generate().public_key().to_bytes().to_vec(),
}
}
/// Inserts an open stream `age` seconds old with plain SQL, as a relay
/// without the name rule did.
async fn open(&self, pool: &PgPool, id: Uuid, name: &str, age: i64) {
sqlx::query(
"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by, \
created_at) \
VALUES ($1, $2, $3, 'stream', 'open', $4, now() - make_interval(secs => $5))",
)
.bind(id)
.bind(self.community)
.bind(name)
.bind(&self.created_by)
.bind(age as f64)
.execute(pool)
.await
.expect("insert channel");
}
async fn names(&self, pool: &PgPool) -> Vec<(Uuid, String)> {
sqlx::query_as("SELECT id, name FROM channels WHERE community_id = $1 ORDER BY id")
.bind(self.community)
.fetch_all(pool)
.await
.expect("read channel names")
}
}
fn renamed(id: Uuid, name: &str) -> (Uuid, String) {
(id, format!("{name}-{}", &id.to_string()[..8]))
}
/// Runs the repair the way `psql -f` does: the `DO` block as its own
/// statement, so that its per-batch COMMITs run outside a transaction block.
/// A repair that stops making progress fails at the statement timeout.
/// Returns the number of rows the script's validation query reports.
pub(super) async fn run_repair(pool: &PgPool) -> i64 {
let (repair, validation) = REPAIR_SCRIPT
.split_once("\n-- Validation.")
.expect("script has a validation query");
let mut conn = pool.acquire().await.expect("acquire repair connection");
sqlx::query("SET statement_timeout = '30s'")
.execute(&mut *conn)
.await
.expect("bound the repair");
sqlx::raw_sql(sqlx::AssertSqlSafe(repair))
.execute(&mut *conn)
.await
.expect("run repair");
drop(conn);
let (_, query) = validation.split_once("\nSELECT").expect("validation query");
sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(format!(
"SELECT count(*) FROM (SELECT{}) AS remaining",
query.trim_end().trim_end_matches(';')
)))
.fetch_one(pool)
.await
.expect("run validation")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn repair_keeps_each_listed_name_on_its_oldest_unarchived_channel() {
let admin = PgPool::connect(&admin_url().await).await.unwrap();
let (pool, name) = create_scratch_db(&admin, "listed_name_repair").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;
seed.open(&pool, archived, "Square-iOS", 60).await;
seed.open(&pool, newer, "SQUARE-IOS", 10).await;
seed.open(&pool, private, "square-ios", 5).await;
seed.open(&pool, deleted, "square-ios", 90).await;
sqlx::query(
"UPDATE channels SET \
archived_at = CASE WHEN id = $2 THEN now() END, \
visibility = CASE WHEN id = $3 THEN 'private'::channel_visibility ELSE visibility END, \
deleted_at = CASE WHEN id = $4 THEN now() END \
WHERE community_id = $1",
)
.bind(seed.community)
.bind(archived)
.bind(private)
.bind(deleted)
.execute(&pool)
.await
.unwrap();
// A community with no duplicates is not touched.
let other = Seed::new(&pool).await;
other.open(&pool, Uuid::new_v4(), "square-ios", 0).await;
let other_before = other.names(&pool).await;
assert_eq!(run_repair(&pool).await, 0, "no duplicates remain");
let mut expected = vec![
(kept, "square-ios".to_owned()),
renamed(archived, "Square-iOS"),
renamed(newer, "SQUARE-IOS"),
(private, "square-ios".to_owned()),
(deleted, "square-ios".to_owned()),
];
expected.sort();
assert_eq!(seed.names(&pool).await, expected);
assert_eq!(other.names(&pool).await, other_before);
assert_eq!(run_repair(&pool).await, 0);
assert_eq!(
seed.names(&pool).await,
expected,
"a second run changes nothing"
);
drop_scratch_db(&admin, pool, &name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn repair_picks_a_free_name_when_the_short_one_is_taken() {
let admin = PgPool::connect(&admin_url().await).await.unwrap();
let (pool, name) = create_scratch_db(&admin, "listed_name_repair_taken").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);
let duplicate = Uuid::parse_str("20000000-0000-4000-8000-000000000000").unwrap();
let (kept, squatter) = (Uuid::new_v4(), Uuid::new_v4());
seed.open(&pool, kept, &long, 30).await;
seed.open(&pool, duplicate, &long, 10).await;
// Another listed channel already has the short new name, in another case.
let short = format!("{}-20000000", "A".repeat(246));
seed.open(&pool, squatter, &short, 60).await;
assert_eq!(run_repair(&pool).await, 0, "no duplicates remain");
let mut expected = vec![
(kept, long.clone()),
(duplicate, format!("{}-{duplicate}", "a".repeat(218))),
(squatter, short),
];
expected.sort();
assert_eq!(seed.names(&pool).await, expected);
assert!(expected.iter().all(|(_, name)| name.chars().count() <= 255));
drop_scratch_db(&admin, pool, &name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn repair_leaves_a_row_that_stopped_being_a_duplicate_while_it_waited() {
let admin = PgPool::connect(&admin_url().await).await.unwrap();
let (pool, name) = create_scratch_db(&admin, "listed_name_repair_wait").await;
let seed = Seed::new(&pool).await;
let (kept, duplicate) = (Uuid::new_v4(), Uuid::new_v4());
seed.open(&pool, kept, "payments", 30).await;
seed.open(&pool, duplicate, "payments", 10).await;
// A relay write holds the duplicate's row while the script starts.
let mut relay = pool.begin().await.unwrap();
sqlx::query("SELECT 1 FROM channels WHERE community_id = $1 AND id = $2 FOR NO KEY UPDATE")
.bind(seed.community)
.bind(duplicate)
.execute(&mut *relay)
.await
.unwrap();
let repair_pool = PgPool::connect_with(
(*pool.connect_options())
.clone()
.application_name("listed-name-repair"),
)
.await
.unwrap();
let repair = tokio::spawn({
let repair_pool = repair_pool.clone();
async move { run_repair(&repair_pool).await }
});
crate::test_support::wait_for_lock_wait(&pool, "listed-name-repair").await;
// The relay makes the channel private and renames it, then commits.
sqlx::query(
"UPDATE channels SET name = 'payments-private', visibility = 'private' \
WHERE community_id = $1 AND id = $2",
)
.bind(seed.community)
.bind(duplicate)
.execute(&mut *relay)
.await
.unwrap();
relay.commit().await.unwrap();
assert_eq!(repair.await.unwrap(), 0);
let mut expected = vec![
(kept, "payments".to_owned()),
(duplicate, "payments-private".to_owned()),
];
expected.sort();
assert_eq!(
seed.names(&pool).await,
expected,
"the private row is not renamed"
);
repair_pool.close().await;
drop_scratch_db(&admin, pool, &name).await;
}

182
crates/buzz-db/src/store/channel.rs
View File

@@ -155,7 +155,9 @@ pub async fn create_channel(
.bind(labels)
.bind(unlisted)
.execute(tx.conn())
.await?;
.await
.map_err(|e| listed_name_taken(e, Some(name)))?;
ensure_listed_channel_name_unique(&mut tx, id).await?;
sqlx::query(
r#"
@@ -392,12 +394,14 @@ async fn insert_channel_rows(
.bind(labels)
.bind(unlisted)
.execute(tx.conn())
.await?
.await
.map_err(|e| listed_name_taken(e, Some(name)))?
.rows_affected();
let was_created = rows_affected > 0;
if was_created {
ensure_listed_channel_name_unique(tx, channel_id).await?;
// Bootstrap the creator as owner.
sqlx::query(
r#"
@@ -446,6 +450,81 @@ async fn insert_channel_rows(
Ok(was_created)
}
/// SQL that is true for a channel the channel directory lists: an open,
/// listed stream or forum that is not deleted. An archived channel is still
/// listed, so it keeps its name. Within a community, no two of these share a
/// name, ignoring case ([`ensure_listed_channel_name_unique`]).
pub(crate) const LISTED_CHANNEL_SQL: &str =
"visibility = 'open' AND NOT unlisted AND channel_type IN ('stream', 'forum') \
AND deleted_at IS NULL";
/// Refuses, with [`DbError::ChannelNameTaken`], a write on `tx` that left
/// `channel_id` listed ([`LISTED_CHANNEL_SQL`]) under a name another listed
/// channel in the community already has, ignoring case. Private channels,
/// DMs and the other types may repeat names: nobody browses them side by
/// side, and refusing a name that a private channel holds would tell anyone
/// that the channel exists.
///
/// Call it after the write that creates the channel or changes its name,
/// visibility or unlisted flag. It first takes a transaction advisory lock
/// on the name, so two writers that claim one name take turns and the second
/// sees the first's committed row. A writer takes this lock last, after the
/// locks of [`lock_channel_for_settings_write`], and holds at most one.
async fn ensure_listed_channel_name_unique(
tx: &mut crate::AdmittedTx,
channel_id: Uuid,
) -> Result<()> {
let community_id = tx.community();
let name: Option<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT name FROM channels WHERE community_id = $1 AND id = $2 AND {LISTED_CHANNEL_SQL}"
)))
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(tx.conn())
.await?;
let Some(name) = name else {
return Ok(());
};
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1 || lower($2), 0))")
.bind(format!("buzz_channel_name:{}:", community_id.as_uuid()))
.bind(&name)
.execute(tx.conn())
.await?;
let taken: bool = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT EXISTS (SELECT 1 FROM channels WHERE community_id = $1 AND id <> $2 \
AND lower(name) = lower($3) AND {LISTED_CHANNEL_SQL})"
)))
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(&name)
.fetch_one(tx.conn())
.await?;
if taken {
return Err(DbError::ChannelNameTaken(name));
}
Ok(())
}
/// The unique index on `(community_id, lower(name))` over
/// [`LISTED_CHANNEL_SQL`] channels that migration 0079 adds. It refuses a
/// duplicate at the write itself, before
/// [`ensure_listed_channel_name_unique`] runs, so this release maps the
/// refusal too: it can run beside that index during the rollout of 0079 and
/// after a rollback.
const LISTED_NAME_INDEX: &str = "idx_channels_listed_name";
/// Reports a channel write that [`LISTED_NAME_INDEX`] refused as
/// [`DbError::ChannelNameTaken`] for `name`, the name the channel would have
/// had.
fn listed_name_taken(error: sqlx::Error, name: Option<&str>) -> DbError {
match (&error, name) {
(sqlx::Error::Database(db), Some(name)) if db.constraint() == Some(LISTED_NAME_INDEX) => {
DbError::ChannelNameTaken(name.to_owned())
}
_ => error.into(),
}
}
/// SQL that is true when `column` holds a legacy channel type
/// ([`buzz_core::channel::LEGACY_CHANNEL_TYPES`]): `<column> IN ('stream', …)`.
///
@@ -1090,21 +1169,30 @@ async fn update_channel_on(
q = q.bind(community_id.as_uuid());
q = q.bind(channel_id);
let policy_changed = if updates.visibility.is_some()
// The current name is what a visibility or unlisted change that
// [`LISTED_NAME_INDEX`] refuses would have listed.
let (policy_changed, current_name) = if updates.visibility.is_some()
|| updates.ttl_seconds.is_some()
|| updates.unlisted.is_some()
{
sqlx::query_scalar::<_,bool>("SELECT (($3::text IS NOT NULL AND visibility::text IS DISTINCT FROM $3) OR ($4 AND ttl_seconds IS DISTINCT FROM $5::int) OR ($6::bool IS NOT NULL AND unlisted IS DISTINCT FROM $6)) FROM channels WHERE community_id=$1 AND id=$2")
sqlx::query_as::<_,(bool, String)>("SELECT (($3::text IS NOT NULL AND visibility::text IS DISTINCT FROM $3) OR ($4 AND ttl_seconds IS DISTINCT FROM $5::int) OR ($6::bool IS NOT NULL AND unlisted IS DISTINCT FROM $6)), name FROM channels WHERE community_id=$1 AND id=$2")
.bind(community_id.as_uuid()).bind(channel_id).bind(updates.visibility.as_deref())
.bind(updates.ttl_seconds.is_some()).bind(updates.ttl_seconds.flatten()).bind(updates.unlisted)
.fetch_one(tx.conn()).await?
.fetch_one(tx.conn()).await
.map(|(changed, name)| (changed, Some(name)))?
} else {
false
(false, None)
};
let result = q.execute(tx.conn()).await?;
let result = q
.execute(tx.conn())
.await
.map_err(|e| listed_name_taken(e, updates.name.as_deref().or(current_name.as_deref())))?;
if result.rows_affected() == 0 {
return Err(DbError::ChannelNotFound(channel_id));
}
if updates.name.is_some() || updates.visibility.is_some() || updates.unlisted.is_some() {
ensure_listed_channel_name_unique(tx, channel_id).await?;
}
if policy_changed {
sqlx::query(
"UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1",
@@ -2401,6 +2489,86 @@ mod postgres_tests {
assert!(matches!(outcome, ChannelCreate::Created(_)), "{outcome:?}");
}
/// Holds an open create of the listed channel `race` on its own
/// transaction while a second create claims the same name beside it.
async fn second_listed_create_beside_open_create(commit_first: bool) -> Result<ChannelCreate> {
let pool = setup_pool().await;
let community = CommunityId::from_uuid(make_test_community(&pool).await);
let creator = Keys::generate();
let creator_bytes = creator.public_key().to_bytes().to_vec();
let listed = |name: &'static str| NewChannel {
name,
visibility: ChannelVisibility::Open,
..new_channel(&creator_bytes, &[])
};
let mut first = crate::begin_community_write_transaction(
&pool,
community,
crate::observability::WriterOperation::EventWrite,
)
.await
.expect("begin first create");
assert!(
insert_channel_rows(&mut first, Uuid::new_v4(), listed("race"))
.await
.expect("first create's rows")
);
let second_id = Uuid::new_v4();
let second = tokio::spawn({
let (creator, creator_bytes) = (creator.clone(), creator_bytes.clone());
async move {
let pool = crate::test_support::named_pool("listed-name-race").await;
create_channel_with_event(
&pool,
None,
community,
second_id,
NewChannel {
name: "RACE",
visibility: ChannelVisibility::Open,
..new_channel(&creator_bytes, &[])
},
&create_event(&creator, second_id),
)
.await
}
});
crate::test_support::wait_for_lock_wait(&pool, "listed-name-race").await;
assert!(
!second.is_finished(),
"the second create waits for the open create's name"
);
if commit_first {
first.commit().await.expect("commit first create");
} else {
drop(first);
}
second.await.expect("second create task")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_listed_create_beside_an_open_create_of_its_name_is_refused_once_committed() {
let outcome = second_listed_create_beside_open_create(true).await;
assert!(
matches!(&outcome, Err(DbError::ChannelNameTaken(name)) if name == "RACE"),
"{outcome:?}"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_listed_create_beside_a_failed_create_of_its_name_succeeds() {
let outcome = second_listed_create_beside_open_create(false).await;
assert!(
matches!(outcome, Ok(ChannelCreate::Created(_))),
"{outcome:?}"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_create_whose_event_fails_leaves_no_channel() {

270
crates/buzz-db/src/store/channel_name_postgres_tests.rs Normal file
View File

@@ -0,0 +1,270 @@
//! Two listed open channels in a community never share a name, ignoring
//! case. Every write that can make a channel claim a listed name goes through
//! the real store functions here: create, rename, private to open, and
//! unlisted to listed. An archived channel keeps its name; a deleted one
//! frees it.
use crate::channel::ChannelUpdate;
use crate::error::DbError;
use crate::Db;
use buzz_core::channel::{ChannelType, ChannelVisibility};
use buzz_core::CommunityId;
use nostr::Keys;
use sqlx::PgPool;
use uuid::Uuid;
struct Fixture {
db: Db,
community: CommunityId,
owner: Keys,
}
impl Fixture {
async fn new() -> Self {
let pool = PgPool::connect(&crate::test_support::database_url())
.await
.expect("connect to test DB");
if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") {
crate::migration::run_migrations(&pool)
.await
.expect("migrate test DB");
}
Self {
community: insert_community(&pool).await,
db: Db::from_pool(pool),
owner: Keys::generate(),
}
}
async fn create(
&self,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
) -> crate::error::Result<Uuid> {
self.create_in(self.community, name, channel_type, visibility)
.await
}
async fn create_in(
&self,
community: CommunityId,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
) -> crate::error::Result<Uuid> {
self.db
.create_channel(
community,
name,
channel_type,
visibility,
None,
&self.owner.public_key().to_bytes(),
None,
)
.await
.map(|channel| channel.id)
}
async fn open(&self, name: &str) -> Uuid {
self.create(name, ChannelType::Stream, ChannelVisibility::Open)
.await
.expect("create open channel")
}
async fn update(&self, channel: Uuid, update: ChannelUpdate) -> crate::error::Result<()> {
self.db
.update_channel(self.community, channel, update)
.await
.map(|_| ())
}
async fn listing_state(&self, channel: Uuid) -> (String, String, bool) {
let channel = self
.db
.get_channel(self.community, channel)
.await
.expect("reload channel");
(channel.name, channel.visibility, channel.unlisted)
}
}
async fn insert_community(pool: &PgPool) -> CommunityId {
let community = CommunityId::from_uuid(Uuid::new_v4());
sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)")
.bind(community.as_uuid())
.bind(format!("channel-name-{}.test", community.as_uuid()))
.execute(pool)
.await
.expect("insert community");
community
}
fn assert_taken(result: crate::error::Result<impl std::fmt::Debug>, name: &str) {
match result {
Err(DbError::ChannelNameTaken(taken)) => assert_eq!(taken, name),
other => panic!("expected {name:?} to be taken, got {other:?}"),
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_listed_name_is_taken_only_for_listed_channels() {
let f = Fixture::new().await;
f.open("Square-iOS").await;
// Another listed channel cannot take it, whatever its case or type, and
// the display prefix is not part of the name.
for (name, channel_type) in [
("square-ios", ChannelType::Stream),
("SQUARE-IOS", ChannelType::Forum),
("#square-ios ", ChannelType::Stream),
] {
assert_taken(
f.create(name, channel_type, ChannelVisibility::Open).await,
buzz_core::channel::canonical_channel_name(name),
);
}
// Channels the directory does not list may repeat it.
for (channel_type, visibility) in [
(ChannelType::Stream, ChannelVisibility::Private),
(ChannelType::Forum, ChannelVisibility::Private),
(ChannelType::Dm, ChannelVisibility::Private),
(ChannelType::Workflow, ChannelVisibility::Open),
(ChannelType::System, ChannelVisibility::Open),
] {
f.create("square-ios", channel_type, visibility)
.await
.unwrap_or_else(|e| panic!("{channel_type:?}/{visibility:?}: {e}"));
}
// Names are per community.
let other = insert_community(f.db.pool()).await;
f.create_in(
other,
"square-ios",
ChannelType::Stream,
ChannelVisibility::Open,
)
.await
.expect("another community may use the name");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_change_that_would_list_a_taken_name_changes_nothing() {
let f = Fixture::new().await;
f.open("payments").await;
let private = f
.create("payments", ChannelType::Stream, ChannelVisibility::Private)
.await
.expect("private channel");
let unlisted = f.open("payments-unlisted").await;
f.update(
unlisted,
ChannelUpdate {
name: Some("payments".into()),
unlisted: Some(true),
..Default::default()
},
)
.await
.expect("an unlisted channel may share the name");
let other = f.open("payments-team").await;
let cases = [
(
private,
ChannelUpdate {
visibility: Some("open".into()),
..Default::default()
},
),
(
unlisted,
ChannelUpdate {
unlisted: Some(false),
..Default::default()
},
),
(
other,
ChannelUpdate {
name: Some("PAYMENTS".into()),
description: Some("rolled back with the name".into()),
..Default::default()
},
),
];
for (channel, update) in cases {
let before = f.listing_state(channel).await;
let description =
f.db.get_channel(f.community, channel)
.await
.expect("reload")
.description;
let attempted = update.name.clone().unwrap_or_else(|| before.0.clone());
assert_taken(f.update(channel, update).await, &attempted);
assert_eq!(f.listing_state(channel).await, before);
assert_eq!(
f.db.get_channel(f.community, channel)
.await
.expect("reload")
.description,
description,
"the whole update rolls back"
);
}
// A rename to a free name, and the same name in another case, still work.
f.update(
other,
ChannelUpdate {
name: Some("payments-core".into()),
..Default::default()
},
)
.await
.expect("rename to a free name");
f.update(
other,
ChannelUpdate {
name: Some("Payments-Core".into()),
..Default::default()
},
)
.await
.expect("a channel may change the case of its own name");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn an_archived_channel_keeps_its_name_until_it_is_deleted() {
let f = Fixture::new().await;
let archived = f.open("square-ios").await;
f.db.archive_channel(f.community, archived)
.await
.expect("archive");
assert_taken(
f.create("square-ios", ChannelType::Stream, ChannelVisibility::Open)
.await,
"square-ios",
);
f.db.unarchive_channel(f.community, archived)
.await
.expect("unarchive never collides");
f.db.archive_channel(f.community, archived)
.await
.expect("archive again");
assert!(f
.db
.soft_delete_channel(f.community, archived)
.await
.expect("delete"));
f.open("square-ios").await;
}

3
crates/buzz-db/src/store/mod.rs
View File

@@ -96,5 +96,8 @@ mod legacy_channel_types_postgres_tests;
#[cfg(test)]
mod unlisted_channels_postgres_tests;
#[cfg(test)]
mod channel_name_postgres_tests;
#[cfg(test)]
mod admission_postgres_tests;

94
crates/buzz-relay/src/handlers/group_state_postgres_tests.rs
View File

@@ -1074,3 +1074,97 @@ async fn removed_creator_join_racing_a_private_switch_is_refused() {
"a removed creator must not rejoin a channel that became private"
);
}
/// A create or rename that would list a second open channel under a taken
/// name is refused with its reason as a client error, and stores nothing.
/// Mutation: let `ChannelNameTaken` fall through to the internal-error arm
/// of either kind → RED.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_taken_open_name_is_refused_as_a_client_error() {
use crate::handlers::ingest::IngestError;
let state = state().await;
let tenant = community(&state, "group-state-name-taken").await;
let owner = Keys::generate();
let create = |name: &str| {
let channel_id = Uuid::new_v4();
let event = EventBuilder::new(Kind::Custom(9007), "")
.tags([
Tag::parse(["h", &channel_id.to_string()]).unwrap(),
Tag::parse(["name", name]).unwrap(),
])
.sign_with_keys(&owner)
.expect("sign 9007");
(channel_id, event)
};
let (_, first) = create("launch");
assert!(
ingest_event(&state, &tenant, first, http_auth(&owner))
.await
.expect("first create")
.accepted
);
let (second, event) = create("Launch");
match ingest_event(&state, &tenant, event, http_auth(&owner)).await {
Err(IngestError::Rejected(message)) => assert_eq!(
message,
"invalid: an open channel named \"Launch\" already exists"
),
Err(other) => panic!("expected a refusal, got {other:?}"),
Ok(result) => panic!("expected a refusal, got {:?}", result.message),
}
assert!(
state
.db
.get_channel(tenant.community(), second)
.await
.is_err(),
"the refused create stores no channel"
);
let other = create_channel(&state, &tenant, &owner, &[]).await;
let before = state
.db
.get_channel(tenant.community(), other)
.await
.expect("channel")
.name;
let rename = EventBuilder::new(Kind::Custom(9002), "")
.tags([
Tag::parse(["h", &other.to_string()]).unwrap(),
Tag::parse(["name", "LAUNCH"]).unwrap(),
])
.sign_with_keys(&owner)
.expect("sign 9002");
let rename_id = rename.id;
match ingest_event(&state, &tenant, rename, http_auth(&owner)).await {
Err(IngestError::Rejected(message)) => assert_eq!(
message,
"invalid: an open channel named \"LAUNCH\" already exists"
),
Err(other) => panic!("expected a refusal, got {other:?}"),
Ok(result) => panic!("expected a refusal, got {:?}", result.message),
}
assert_eq!(
state
.db
.get_channel(tenant.community(), other)
.await
.expect("channel")
.name,
before,
"the refused rename keeps the name"
);
let stored: bool = sqlx::query_scalar(
"SELECT EXISTS (SELECT 1 FROM events WHERE community_id = $1 AND id = $2)",
)
.bind(tenant.community().as_uuid())
.bind(rename_id.as_bytes().as_slice())
.fetch_one(state.db.pool())
.await
.expect("look up the 9002");
assert!(!stored, "the refused 9002 is not stored");
}

11
crates/buzz-relay/src/handlers/ingest.rs
View File

@@ -3558,8 +3558,12 @@ async fn ingest_event_inner(
.db
.create_channel_with_event(tenant.community(), create.channel_id, channel, &event)
.await
.map_err(|e| IngestError::Internal(format!("error: {e}")))?
{
.map_err(|e| match e {
buzz_db::DbError::ChannelNameTaken(_) => {
IngestError::Rejected(format!("invalid: {e}"))
}
_ => IngestError::Internal(format!("error: {e}")),
})? {
buzz_db::channel::ChannelCreate::Created(stored) => {
metrics::counter!(
"buzz_channels_created_total",
@@ -3624,6 +3628,9 @@ async fn ingest_event_inner(
Some(buzz_db::DbError::AccessDenied(reason)) => {
IngestError::Rejected(format!("invalid: {reason}"))
}
Some(taken @ buzz_db::DbError::ChannelNameTaken(_)) => {
IngestError::Rejected(format!("invalid: {taken}"))
}
_ => IngestError::Internal(format!("error: channel update failed: {e}")),
})?
} else if buzz_core::kind::is_replaceable(kind_u32) {

161
scripts/maintenance/rename_duplicate_listed_channel_names.sql Normal file
View File

@@ -0,0 +1,161 @@
-- OUT-OF-BAND MAINTENANCE: do not run from relay startup migrations.
--
-- Gives every listed open channel a name that no other listed open channel in
-- its community has, ignoring case. A listed channel is an open, not unlisted
-- stream or forum that is not deleted; an archived one still counts. This is
-- the rule the relay enforces on create and rename (`LISTED_CHANNEL_SQL` in
-- crates/buzz-db/src/store/channel.rs). It repairs the duplicates created
-- before that rule, so the contract migration's unique index can build.
--
-- Run it only after every serving relay enforces the rule. An older relay can
-- create a new duplicate while the script runs.
--
-- In each group of channels that share a name, one channel keeps it: the
-- oldest unarchived one, or the oldest archived one when all are archived.
-- Every other channel in the group becomes "<name>-<first 8 characters of its
-- id>", or "<name>-<id>" when another listed channel already has that name.
-- The script prints a NOTICE for each rename. When both names are taken, it
-- leaves the channel unchanged with a NOTICE, and the validation below
-- reports it.
--
-- The rename writes only the channel row. Clients read the name from the
-- relay-signed kind:39000, so republish it for each renamed channel:
--
-- buzz-admin reconcile-channels --channel <id> --metadata
--
-- Each batch renames up to 100 channels in one community and commits. It
-- takes the community deletion lock first and skips a community that is not
-- active. It locks every row of the batch before it takes the relay's name
-- lock for any new name, in the order the relay takes them. After it locks a
-- row, it checks again that the row is still a listed duplicate, because a
-- relay may have changed it meanwhile. The script is idempotent: a second run
-- finds nothing to rename.
--
-- Run it outside a transaction, so that each batch can commit:
--
-- psql -v ON_ERROR_STOP=1 -f scripts/maintenance/rename_duplicate_listed_channel_names.sql
DO $$
DECLARE
target UUID;
skipped UUID[] := '{}';
batch UUID[];
duplicate UUID;
live RECORD;
new_name TEXT;
renamed INT;
BEGIN
LOOP
SELECT ch.community_id INTO target
FROM channels ch
WHERE ch.visibility = 'open' AND NOT ch.unlisted
AND ch.channel_type IN ('stream', 'forum') AND ch.deleted_at IS NULL
AND ch.community_id <> ALL (skipped)
GROUP BY ch.community_id, lower(ch.name)
HAVING count(*) > 1
LIMIT 1;
EXIT WHEN target IS NULL;
PERFORM set_config('lock_timeout', '5s', true);
PERFORM pg_advisory_xact_lock_shared(community_deletion_lock_key(target));
IF NOT EXISTS (
SELECT 1 FROM communities
WHERE id = target AND deletion_state = 'active' AND deleted_at IS NULL
) THEN
RAISE NOTICE 'skipped community %: it is not active', target;
skipped := skipped || target;
COMMIT;
CONTINUE;
END IF;
-- Lock every row of the batch before any name lock, in the order
-- the relay takes them.
SELECT array_agg(locked.id ORDER BY locked.id) INTO batch
FROM (
WITH ranked AS (
SELECT id, row_number() OVER (
PARTITION BY lower(name)
ORDER BY archived_at IS NOT NULL, created_at, id
) AS place
FROM channels
WHERE community_id = target
AND visibility = 'open' AND NOT unlisted
AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL
)
SELECT ch.id
FROM channels ch
JOIN ranked ON ranked.id = ch.id
WHERE ch.community_id = target AND ranked.place > 1
ORDER BY ch.id
LIMIT 100
FOR NO KEY UPDATE OF ch
) locked;
renamed := 0;
FOREACH duplicate IN ARRAY coalesce(batch, '{}')
LOOP
-- The ranking used the statement's snapshot. Now that the row is
-- locked, read it again: a relay may have renamed it, made it
-- private, or renamed the channel that kept the name.
SELECT ch.name, ch.archived_at, ch.created_at INTO live
FROM channels ch
WHERE ch.community_id = target AND ch.id = duplicate
AND ch.visibility = 'open' AND NOT ch.unlisted
AND ch.channel_type IN ('stream', 'forum') AND ch.deleted_at IS NULL;
CONTINUE WHEN NOT FOUND OR NOT EXISTS (
SELECT 1 FROM channels other
WHERE other.community_id = target AND other.id <> duplicate
AND lower(other.name) = lower(live.name)
AND other.visibility = 'open' AND NOT other.unlisted
AND other.channel_type IN ('stream', 'forum') AND other.deleted_at IS NULL
AND (other.archived_at IS NOT NULL, other.created_at, other.id)
< (live.archived_at IS NOT NULL, live.created_at, duplicate)
);
new_name := NULL;
FOREACH new_name IN ARRAY ARRAY[
left(live.name, 246) || '-' || left(duplicate::text, 8),
left(live.name, 218) || '-' || duplicate::text
] LOOP
PERFORM pg_advisory_xact_lock(hashtextextended(
'buzz_channel_name:' || target::text || ':' || lower(new_name), 0));
EXIT WHEN NOT EXISTS (
SELECT 1 FROM channels other
WHERE other.community_id = target AND other.id <> duplicate
AND lower(other.name) = lower(new_name)
AND other.visibility = 'open' AND NOT other.unlisted
AND other.channel_type IN ('stream', 'forum') AND other.deleted_at IS NULL
);
new_name := NULL;
END LOOP;
IF new_name IS NULL THEN
RAISE NOTICE 'left channel % in community % as "%": both new names are taken',
duplicate, target, live.name;
CONTINUE;
END IF;
UPDATE channels SET name = new_name, updated_at = NOW()
WHERE community_id = target AND id = duplicate;
RAISE NOTICE 'renamed channel % in community % from "%" to "%"',
duplicate, target, live.name, new_name;
renamed := renamed + 1;
END LOOP;
-- A batch that found nothing to rename would find the same community
-- again; skip it so the loop ends. The validation below reports it.
IF renamed = 0 THEN
skipped := skipped || target;
END IF;
COMMIT;
END LOOP;
END
$$;
-- Validation. Expected result: zero rows. A row names a community the script
-- skipped (not active) or a duplicate created while it ran; rerun the script.
SELECT community_id, lower(name) AS name, count(*) AS channels
FROM channels
WHERE visibility = 'open' AND NOT unlisted
AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL
GROUP BY community_id, lower(name)
HAVING count(*) > 1;