diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index 9ae81f3a83..5f48693a98 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -1003,41 +1003,6 @@ async fn fleet_sweep_returns_committed_results_when_database_drops_mid_page() { ); } -#[tokio::test] -#[ignore = "requires Postgres"] -async fn d_tag_backfill_admits_active_tenant_and_skips_quiescing_tenant() { - let (db, communities) = fixture().await; - crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; - for (index, community) in communities[..2].iter().enumerate() { - sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig) \ - VALUES ($1, $2, $3, to_timestamp(1), 30023, '[[\"d\",\"tag\"]]'::jsonb, '', $4)", - ) - .bind(community.as_uuid()) - .bind(vec![index as u8; 32]) - .bind(vec![2_u8; 32]) - .bind(vec![3_u8; 64]) - .execute(&db.pool) - .await - .expect("seed addressable event"); - } - assert_eq!(db.backfill_d_tags().await.unwrap(), 1); - let active: Option = - sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1") - .bind(communities[0].as_uuid()) - .fetch_one(&db.pool) - .await - .unwrap(); - let quiescing: Option = - sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1") - .bind(communities[1].as_uuid()) - .fetch_one(&db.pool) - .await - .unwrap(); - assert_eq!(active.as_deref(), Some("tag")); - assert_eq!(quiescing, None); -} - #[tokio::test] #[ignore = "requires Postgres"] async fn match_claim_admits_active_tenant_and_skips_quiescing_tenant() { diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 0a7d87118c..a53b82944b 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -2776,53 +2776,6 @@ impl Db { Ok(result) } - /// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`. - /// - /// Idempotent — safe to call on every startup. No-ops when all rows are already populated. - /// Admits each community before updating its NIP-33 rows with NULL d_tag. - #[datastore_span(name = "backfill_d_tags", system = "postgresql")] - pub async fn backfill_d_tags(&self) -> Result { - let mut connection = crate::observability::acquire_writer( - &self.pool, - crate::observability::WriterOperation::Bootstrap, - ) - .await?; - let communities: Vec = sqlx::query_scalar( - "SELECT DISTINCT community_id FROM events \ - WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL", - ) - .fetch_all(&mut *connection) - .await?; - drop(connection); - let mut updated = 0; - for id in communities { - let Some(mut tx) = crate::begin_community_sweep_transaction( - &self.pool, - CommunityId::from_uuid(id), - crate::observability::WriterOperation::Bootstrap, - ) - .await? - else { - continue; - }; - let result = sqlx::query( - "UPDATE events \ - SET d_tag = COALESCE( \ - (SELECT elem->>1 FROM jsonb_array_elements(tags) AS elem \ - WHERE elem->>0 = 'd' LIMIT 1), \ - '' \ - ) \ - WHERE community_id = $1 AND kind BETWEEN 30000 AND 39999 AND d_tag IS NULL", - ) - .bind(id) - .execute(tx.conn()) - .await?; - tx.commit().await?; - updated += result.rows_affected(); - } - Ok(updated) - } - /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey. #[datastore_span(name = "soft_delete_discovery_events", system = "postgresql")] pub async fn soft_delete_discovery_events( diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 87a35e2d92..380187bd1d 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -156,14 +156,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { assert!(event.contains("WriterOperation::Authorization")); assert!(event.contains("WriterOperation::SubscriptionHistory")); assert!(event.contains("ReaderOperation::SubscriptionHistory")); - let backfill_d_tags = event - .split_once("pub async fn backfill_d_tags") - .expect("event store must expose the startup d-tag backfill") - .1 - .split_once("/// Soft-delete NIP-29 discovery events") - .expect("d-tag backfill must precede discovery deletion") - .0; - assert!(backfill_d_tags.contains("WriterOperation::Bootstrap")); let soft_delete_discovery = event .split_once("pub async fn soft_delete_discovery_events") .expect("event store must expose discovery-event deletion") @@ -1633,6 +1625,12 @@ fn masker_blanks_every_literal_form_and_keeps_code() { /// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.` /// schema prefix. fn inserts_events_row(source: &str) -> bool { + !events_insert_tails(source).is_empty() +} + +/// `source` lowercased, with every run of whitespace, `\` line continuations, +/// and `\n`/`\t`/`\r` escapes collapsed to one space. +fn normalized_sql(source: &str) -> String { let mut normalized = String::with_capacity(source.len()); let mut chars = source.chars().peekable(); while let Some(ch) = chars.next() { @@ -1647,16 +1645,26 @@ fn inserts_events_row(source: &str) -> bool { normalized.push(ch.to_ascii_lowercase()); } } - ["insert into events", "insert into public.events"] - .iter() - .any(|marker| { - normalized.match_indices(marker).any(|(index, marker)| { - normalized[index + marker.len()..] - .chars() - .next() - .is_none_or(|next| !(next.is_ascii_alphanumeric() || next == '_')) - }) - }) + normalized +} + +/// The normalized SQL that follows each `INSERT INTO events` in `source`. +fn events_insert_tails(source: &str) -> Vec { + let normalized = normalized_sql(source); + let mut tails = Vec::new(); + for marker in ["insert into events", "insert into public.events"] { + for (index, marker) in normalized.match_indices(marker) { + let tail = &normalized[index + marker.len()..]; + if tail + .chars() + .next() + .is_none_or(|next| !(next.is_ascii_alphanumeric() || next == '_')) + { + tails.push(tail.to_owned()); + } + } + } + tails } /// An events insert must run both follow-ups: the push enqueue and the @@ -2062,3 +2070,124 @@ fn every_production_event_insert_runs_the_roster_guard() { "expected at least nine event writers, found {checked_writers}" ); } + +// The rolled-back replica-floor probe writes a kind 9 row, which has no `d`. +const EVENT_INSERT_D_TAG_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; + +/// Whether an `INSERT INTO events` tail names `d_tag` in an explicit column +/// list. An insert without a column list does not count. +fn events_insert_names_d_tag(tail: &str) -> bool { + tail.trim_start() + .strip_prefix('(') + .and_then(|columns| columns.split_once(')')) + .is_some_and(|(columns, _)| columns.split(',').any(|column| column.trim() == "d_tag")) +} + +fn event_insert_d_tag_violations(production_source: &str) -> Vec { + function_slices(production_source) + .into_iter() + .filter(|function_source| { + let header = function_header(function_source); + !EVENT_INSERT_D_TAG_EXCEPTIONS + .iter() + .any(|exception| header.starts_with(exception)) + && !events_insert_tails(function_source) + .iter() + .all(|tail| events_insert_names_d_tag(tail)) + }) + .map(|function_source| function_header(function_source).to_owned()) + .collect() +} + +#[test] +fn event_insert_d_tag_policy_rejects_writers_without_the_column() { + let source = r##" +pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, $3)"); +} +pub(crate) async fn writer_with_d_tag(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events \ + (community_id, id, kind, d_tag) VALUES ($1, $2, $3, $4)"); +} +pub(crate) async fn writer_with_compact_columns(tx: &mut AdmittedTx) { + sqlx::query("insert into public.events (community_id,id,kind,d_tag) VALUES ($1,$2,$3,$4)"); +} +pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id, kind) SELECT community_id, id, d_tag FROM x"); +} +pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id, kind, d_tag_hash) VALUES ($1, $2, $3, $4)"); +} +pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events SELECT * FROM staged_events"); +} +pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id, d_tag) VALUES ($1, $2, $3)"); + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); +} +pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)"); +} +pub async fn verify_floor_guard_behavior(pool: &PgPool) { + sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, 9)"); +} +#[cfg(test)] +mod tests { + fn fixture() { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); + } +} +"##; + assert_eq!( + event_insert_d_tag_violations(&strip_cfg_test_items(source)), + [ + "pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) {", + "pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) {", + ], + "every production events insert must name `d_tag` in its column list" + ); +} + +/// Addressable events (kinds 30000-39999) replace by `d_tag`, and the only +/// index that serves that lookup is partial on `d_tag IS NOT NULL`. A writer +/// that leaves the column out stores NULL, so its rows never replace. Nothing +/// repairs them after the fact, so every production writer must name the column. +/// +/// This is a text match, not a proof. It checks the column list of literal +/// `INSERT INTO [public.]events (...)` SQL inside function bodies under each +/// crate's `src/`. It does not see `ONLY`, quoted or `format!`-built table +/// names, a comment between the keywords, or SQL held in a constant outside a +/// function, and it does not check that the bound value is non-NULL. Exempt +/// probes are skipped by function header, so every insert inside them is too. +#[test] +fn every_production_event_insert_names_d_tag() { + let mut checked_writers = 0; + let mut violations = Vec::new(); + for path in production_event_writer_source_files() { + let source = std::fs::read_to_string(&path).expect("read source file"); + let production = strip_cfg_test_items(&source); + checked_writers += function_slices(&production) + .into_iter() + .filter(|function| inserts_events_row(function)) + .count(); + violations.extend( + event_insert_d_tag_violations(&production) + .into_iter() + .map(|header| format!("{}: {header}", path.display())), + ); + } + assert!( + violations.is_empty(), + "production `INSERT INTO events` writers must name `d_tag` in their column list: \ + {violations:?}" + ); + // The same writer set as the roster-guard scan. A lower count means the + // scan stopped seeing a writer, not that one was removed safely. + assert!( + checked_writers >= 9, + "expected at least nine event writers, found {checked_writers}" + ); +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 40a3d059bc..7ada2da66d 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -704,19 +704,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { } step.finish(); - // NIP-33: backfill d_tag for any existing parameterized replaceable events - // that predate the column addition. Idempotent — no-ops when fully populated. - let mut step = StepTimer::start(StartupStep::DTagBackfill); - match db.backfill_d_tags().await { - Ok(0) => {} - Ok(n) => info!("Backfilled d_tag for {n} NIP-33 events"), - Err(e) => { - step.degrade(); - error!("Failed to backfill d_tags: {e}"); - } - } - step.finish(); - let (audit, audit_metrics_pool) = if config.audit_enabled { let step = StepTimer::start(StartupStep::AuditConnect); let audit_pool = connect_audit_pool(&db_config) diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index 0bc8308fd4..6d95057414 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -51,8 +51,6 @@ pub enum StartupStep { ReplicaFenceProbe, /// Ensure the deployment community, backfill the allowlist, and bootstrap the owner. MembershipBootstrap, - /// Backfill NIP-33 `d_tag` values. - DTagBackfill, /// Connect the audit database pool (only when audit is enabled). AuditConnect, /// Connect Redis commands and pub/sub. @@ -79,14 +77,13 @@ pub enum StartupStep { impl StartupStep { /// The complete `phase` label vocabulary. - pub const ALL: [Self; 18] = [ + pub const ALL: &[Self] = &[ Self::DbConnect, Self::DbMigrate, Self::PartitionEnsure, Self::DeletionFenceVerify, Self::ReplicaFenceProbe, Self::MembershipBootstrap, - Self::DTagBackfill, Self::AuditConnect, Self::RedisConnect, Self::SearchConnect, @@ -109,7 +106,6 @@ impl StartupStep { Self::DeletionFenceVerify => "deletion_fence_verify", Self::ReplicaFenceProbe => "replica_fence_probe", Self::MembershipBootstrap => "membership_bootstrap", - Self::DTagBackfill => "d_tag_backfill", Self::AuditConnect => "audit_connect", Self::RedisConnect => "redis_connect", Self::SearchConnect => "search_connect",