From a31c0aad561d63ace772a20161ac78f8423cd11e Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 14:25:17 -0400 Subject: [PATCH] buzz-db: check the floor only on newly inserted event rows Review rework for the insert-time created_at floor (BUZZ-316 phase 1). - Ordinary and thread event writers now run the floor check after their ON CONFLICT DO NOTHING INSERT, only when a row was inserted and before follow-ups. Replaying an already-stored id stays Duplicate even when its created_at has aged below the floor, matching the trigger. - Source policy: UPDATE rule catches ONLY, aliases and row-constructor column lists; quoted identifiers are normalized; the floor policy scans every crate's src via the shared workspace walk. - NIP-43 membership insert writes a literal NULL channel_id, pinned positionally; the bind-based exemption is gone. - Canvas CAS fixtures take a per-test base instead of a process-wide OnceLock. - seed-admin-dashboard.sh also refuses PGHOSTADDR, PGSERVICE and multi-host PGHOST. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/replica_fence.rs | 14 +- crates/buzz-db/src/store/event.rs | 95 ++++++--- .../src/store/event_floor_postgres_tests.rs | 70 ++++++ crates/buzz-db/src/store/relay_members.rs | 4 +- crates/buzz-db/tests/observability_source.rs | 200 ++++++++++++++---- scripts/seed-admin-dashboard.sh | 4 + 6 files changed, 302 insertions(+), 85 deletions(-) diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 1f85f20c5f..06401a5a10 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -4,10 +4,11 @@ //! contain is provably present on the replica. The proof rests on two parts: //! //! 1. **Insert-time floor**: supported channel-event writers check `created_at` -//! against database `clock_timestamp()` on their admitted transaction before -//! INSERT. This is the primary guard. Migration 0021's deferred trigger and -//! writer-pool `buzz.created_at_floor` GUC remain the mixed-version backstop -//! until phase 2; they also reject rows that age below the floor before COMMIT. +//! against database `clock_timestamp()` on their admitted transaction for +//! each new row. Duplicate INSERTs need no floor check. This is the primary +//! guard. Migration 0021's deferred trigger and the writer-pool +//! `buzz.created_at_floor` GUC remain the mixed-version backstop until phase 2; +//! they also reject rows that age below the floor before COMMIT. //! 2. **Ordered heartbeat handshake** (this module): on one pinned writer //! connection, one transaction first takes the exclusive replica-floor //! advisory lock (draining compliant shared-lock channel writers), then @@ -76,7 +77,10 @@ pub const CREATED_AT_FLOOR_SECS: i64 = 960; /// Enforce the channel-event floor on the admitted INSERT connection. /// -/// One preceding SELECT keeps the five existing INSERT shapes intact. Its +/// One SELECT keeps the five existing INSERT shapes intact. Ordinary/thread +/// writers check after INSERT only when it inserted a row, before follow-ups; +/// replacements check after replay exits, before superseding the old head. +/// The caller must propagate failure and roll back the write transaction. Its /// database clock is sampled after BEGIN: transactions open at the probe scan /// are bounded by xact_start; transactions starting later sample after S. No /// commit hook, candidate tracking or ordering against TTL refresh is needed. diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 2ac53ad8b9..f582247d1d 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -441,7 +441,8 @@ pub async fn insert_event( /// /// This is the transaction-composition seam for callers that must keep the /// event insert open while performing related work. The caller owns commit or -/// rollback. +/// rollback. On error, roll back or drop the transaction: the event row may +/// already have been inserted before the app floor check rejected it. pub async fn insert_event_in_transaction( tx: &mut AdmittedTx, event: &Event, @@ -484,7 +485,6 @@ async fn insert_event_on( let not_before = extract_not_before(event); crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) .await?; - crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?; let result = sqlx::query( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) @@ -509,6 +509,7 @@ async fn insert_event_on( let was_inserted = result.rows_affected() > 0; if was_inserted { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?; crate::store::event_follow_up::after_admitted_insert( tx, id_bytes.as_slice(), @@ -1790,7 +1791,6 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) .await?; - crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?; let result = sqlx::query( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) @@ -1816,6 +1816,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let was_inserted = result.rows_affected() > 0; if was_inserted { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?; crate::store::event_follow_up::after_admitted_insert( tx, id_bytes.as_slice(), @@ -4157,11 +4158,8 @@ mod postgres_tests { // ─── canvas CAS tests ───────────────────────────────────────────────────── - fn make_canvas_event_at(content: &str, created_at: u64) -> nostr::Event { - // Keep the CAS fixtures' logical seconds and ties, anchored at a fresh - // timestamp so they exercise the serving writer's floor admission. - static BASE: std::sync::OnceLock = std::sync::OnceLock::new(); - let base = BASE.get_or_init(|| nostr::Timestamp::now().as_secs() - 1000); + fn make_canvas_event_at(base: u64, content: &str, created_at: u64) -> nostr::Event { + // Each test owns its fresh base; logical seconds and ties stay explicit. make_event_at( buzz_core::kind::KIND_CANVAS as u16, content, @@ -4172,9 +4170,9 @@ mod postgres_tests { /// Build two canvas events sharing `created_at`, returned `(lower, higher)` /// by event id. Regenerates until the ids differ (always, since keys differ) /// so tests can assert deterministic head ordering on the id tiebreak. - fn same_second_ordered_pair(created_at: u64) -> (nostr::Event, nostr::Event) { - let a = make_canvas_event_at("# A", created_at); - let b = make_canvas_event_at("# B", created_at); + fn same_second_ordered_pair(base: u64, created_at: u64) -> (nostr::Event, nostr::Event) { + let a = make_canvas_event_at(base, "# A", created_at); + let b = make_canvas_event_at(base, "# B", created_at); if a.id.as_bytes() <= b.id.as_bytes() { (a, b) } else { @@ -4182,6 +4180,20 @@ mod postgres_tests { } } + #[test] + fn canvas_fixture_base_is_owned_by_each_test() { + let first_base = 1_700_000_000; + let second_base = first_base + crate::replica_fence::CREATED_AT_FLOOR_SECS as u64 * 2; + let first = make_canvas_event_at(first_base, "first test", 1000); + let second = make_canvas_event_at(second_base, "later test", 1000); + assert_eq!(first.created_at.as_secs(), first_base + 1000); + assert_eq!(second.created_at.as_secs(), second_base + 1000); + let (lower, higher) = same_second_ordered_pair(second_base, 1000); + assert_eq!(lower.created_at, higher.created_at); + assert_eq!(lower.created_at.as_secs(), second_base + 1000); + assert!(lower.id.as_bytes() < higher.id.as_bytes()); + } + #[tokio::test] async fn canvas_head_checked_rejects_non_canvas_kinds_before_opening_a_transaction() { // A lazy pool that is never connected: the kind check must reject @@ -4209,10 +4221,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_expect_no_head_creates_first_canvas() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let event = make_canvas_event_at("# First", 1000); + let event = make_canvas_event_at(canvas_base, "# First", 1000); let (_, status) = insert_canvas_head_checked( &pool, @@ -4229,10 +4242,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_expect_no_head_rejects_when_head_exists() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let first = make_canvas_event_at("# First", 1000); + let first = make_canvas_event_at(canvas_base, "# First", 1000); insert_canvas_head_checked( &pool, community, @@ -4243,7 +4257,7 @@ mod postgres_tests { .await .expect("first canvas"); - let second = make_canvas_event_at("# Racing create", 1001); + let second = make_canvas_event_at(canvas_base, "# Racing create", 1001); let (_, status) = insert_canvas_head_checked( &pool, community, @@ -4269,10 +4283,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_expected_head_matches_and_advances() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let first = make_canvas_event_at("# First", 1000); + let first = make_canvas_event_at(canvas_base, "# First", 1000); insert_canvas_head_checked( &pool, community, @@ -4283,7 +4298,7 @@ mod postgres_tests { .await .expect("first canvas"); - let second = make_canvas_event_at("# Second", 1001); + let second = make_canvas_event_at(canvas_base, "# Second", 1001); let head = first.id.as_bytes(); let (_, status) = insert_canvas_head_checked( &pool, @@ -4300,10 +4315,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_expected_head_mismatch_rejects() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let first = make_canvas_event_at("# First", 1000); + let first = make_canvas_event_at(canvas_base, "# First", 1000); insert_canvas_head_checked( &pool, community, @@ -4314,7 +4330,7 @@ mod postgres_tests { .await .expect("first canvas"); - let stale = make_canvas_event_at("# Stale edit", 1002); + let stale = make_canvas_event_at(canvas_base, "# Stale edit", 1002); let wrong_head = [0u8; 32]; let (_, status) = insert_canvas_head_checked( &pool, @@ -4331,10 +4347,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_expected_head_missing_rejects() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let event = make_canvas_event_at("# Edit with no head", 1000); + let event = make_canvas_event_at(canvas_base, "# Edit with no head", 1000); let some_head = [1u8; 32]; let (_, status) = insert_canvas_head_checked( @@ -4352,10 +4369,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_replay_of_head_is_duplicate() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let event = make_canvas_event_at("# First", 1000); + let event = make_canvas_event_at(canvas_base, "# First", 1000); insert_canvas_head_checked( &pool, community, @@ -4387,10 +4405,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_replay_skips_stale_precondition() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let event = make_canvas_event_at("# First", 1000); + let event = make_canvas_event_at(canvas_base, "# First", 1000); insert_canvas_head_checked( &pool, community, @@ -4422,11 +4441,12 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_rejects_same_second_higher_id() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let (lower, higher) = same_second_ordered_pair(1000); + let (lower, higher) = same_second_ordered_pair(canvas_base, 1000); insert_canvas_head_checked( &pool, community, @@ -4465,11 +4485,12 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_accepts_same_second_lower_id() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let (lower, higher) = same_second_ordered_pair(1000); + let (lower, higher) = same_second_ordered_pair(canvas_base, 1000); // Seed the higher-id event as the head first. insert_canvas_head_checked( &pool, @@ -4499,10 +4520,11 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_rejects_behind_clock_writer() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let head = make_canvas_event_at("# Head", 2000); + let head = make_canvas_event_at(canvas_base, "# Head", 2000); insert_canvas_head_checked( &pool, community, @@ -4514,7 +4536,7 @@ mod postgres_tests { .expect("head at t=2000"); // Writer's clock is behind — created_at earlier than head. - let behind = make_canvas_event_at("# Behind clock", 1100); + let behind = make_canvas_event_at(canvas_base, "# Behind clock", 1100); let (_, status) = insert_canvas_head_checked( &pool, community, @@ -4598,12 +4620,13 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_deleted_replay_returns_revision_mismatch() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; // H: seed the initial head. - let head = make_canvas_event_at("# Head", 1000); + let head = make_canvas_event_at(canvas_base, "# Head", 1000); insert_canvas_head_checked( &pool, community, @@ -4616,7 +4639,7 @@ mod postgres_tests { let head_id = head.id.to_bytes().to_vec(); // A: insert a second revision. - let a = make_canvas_event_at("# A", 1001); + let a = make_canvas_event_at(canvas_base, "# A", 1001); insert_canvas_head_checked( &pool, community, @@ -4721,11 +4744,12 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_concurrent_authors_only_one_advances() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; - let base = make_canvas_event_at("# Base", 1000); + let base = make_canvas_event_at(canvas_base, "# Base", 1000); insert_canvas_head_checked( &pool, community, @@ -4750,8 +4774,8 @@ mod postgres_tests { // Both edits assert `base` as their head and are timestamped strictly // ahead of it (writer discipline), so neither trips SupersedeFailed — // only the advisory-lock serialization decides the winner. - let a = make_canvas_event_at("# Author A", 1001); - let b = make_canvas_event_at("# Author B", 1002); + let a = make_canvas_event_at(canvas_base, "# Author A", 1001); + let b = make_canvas_event_at(canvas_base, "# Author B", 1002); let (pool_a, pool_b) = (pool.clone(), pool.clone()); let (id_a, id_b) = (base_id.clone(), base_id.clone()); @@ -4862,6 +4886,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_checked_concurrent_first_create_only_one_wins() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; @@ -4875,8 +4900,8 @@ mod postgres_tests { .await .expect("blocker acquires key"); - let a = make_canvas_event_at("# First A", 1000); - let b = make_canvas_event_at("# First B", 1001); + let a = make_canvas_event_at(canvas_base, "# First A", 1000); + let b = make_canvas_event_at(canvas_base, "# First B", 1001); let (pool_a, pool_b) = (pool.clone(), pool.clone()); let ta = tokio::spawn(async move { @@ -4983,6 +5008,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn db_untagged_canvas_append_serializes_on_advisory_key() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let db = crate::Db::from_pool(pool.clone()); let community = CommunityId::from_uuid(make_test_community(&pool).await); @@ -4996,7 +5022,7 @@ mod postgres_tests { .await .expect("holder acquires key"); - let event = make_canvas_event_at("# Untagged", 1000); + let event = make_canvas_event_at(canvas_base, "# Untagged", 1000); let write_task = tokio::spawn(async move { db.insert_event_with_thread_metadata(community, &event, Some(channel), None) .await @@ -5038,13 +5064,14 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn channel_head_canvas_deletion_serializes_on_advisory_key() { + let canvas_base = nostr::Timestamp::now().as_secs() - 1000; let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await; let lock_key = canvas_lock_key(community, channel); // Insert a canvas event to delete. - let event = make_canvas_event_at("# To delete", 1000); + let event = make_canvas_event_at(canvas_base, "# To delete", 1000); insert_canvas_head_checked( &pool, community, diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs index 857d91ba68..741e3b7d52 100644 --- a/crates/buzz-db/src/store/event_floor_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs @@ -120,6 +120,62 @@ async fn insert_event_floor_rejects_old_and_accepts_fresh() { ); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn insert_event_floor_accepts_expired_duplicate() { + let (db, community, channel, keys) = fixture().await; + let old = signed( + &keys, + 9, + channel, + database_clock(&db).await - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60), + ); + crate::test_support::insert_historical_event(db.pool(), community, &old, channel).await; + let (_, inserted) = event::insert_event(db.pool(), community, &old, Some(channel)) + .await + .expect("expired duplicate is idempotent"); + assert!(!inserted); + assert_eq!(count(&db, community, &old).await, 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn thread_metadata_event_floor_accepts_expired_duplicate() { + let (db, community, channel, keys) = fixture().await; + let time = database_clock(&db).await - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60); + let old = signed(&keys, 9, channel, time); + crate::test_support::insert_historical_event(db.pool(), community, &old, channel).await; + let mut tx = begin(&db, community).await; + let meta = event::ThreadMetadataParams { + event_id: old.id.as_bytes(), + event_created_at: DateTime::from_timestamp(old.created_at.as_secs() as i64, 0) + .expect("timestamp"), + channel_id: channel, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: true, + }; + let (_, inserted) = + event::insert_event_with_thread_metadata_tx(&mut tx, &old, Some(channel), Some(meta)) + .await + .expect("expired duplicate is idempotent"); + assert!(!inserted); + tx.commit().await.expect("commit duplicate result"); + assert_eq!(count(&db, community, &old).await, 1); + let metadata: i64 = sqlx::query_scalar( + "SELECT count(*) FROM thread_metadata WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(old.id.as_bytes().as_slice()) + .fetch_one(db.pool()) + .await + .expect("metadata count"); + assert_eq!(metadata, 0, "a duplicate must not run thread follow-ups"); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn thread_metadata_event_floor_rejects_old_and_accepts_fresh() { @@ -264,7 +320,21 @@ async fn cluster_global_event_floor_held_transaction_cannot_write_below_probe_wa channel, entry.fence_wall - chrono::Duration::seconds(1), ); + sqlx::query("SAVEPOINT rejected_floor_insert") + .execute(tx.conn()) + .await + .expect("isolate rejected insert"); rejected(event::insert_event_in_transaction(&mut tx, &below_wall, Some(channel)).await); + // The check runs after INSERT. Roll back that write before continuing in + // the same held transaction, just as a caller must on any insert failure. + sqlx::query("ROLLBACK TO SAVEPOINT rejected_floor_insert") + .execute(tx.conn()) + .await + .expect("roll back rejected row"); + sqlx::query("RELEASE SAVEPOINT rejected_floor_insert") + .execute(tx.conn()) + .await + .expect("release rejection savepoint"); // Sample just after the probe; a near-floor accepted insert is above its // wall even when this transaction commits after the heartbeat. let fresh = signed( diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 6b21c7ce01..61d7104da1 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -10,7 +10,6 @@ use buzz_core::StoredEvent; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; use sqlx::{PgPool, Row as _}; -use uuid::Uuid; use crate::error::{DbError, Result}; use crate::{observability, replaceable, CommunityId, Db, RouteDecision, RoutePredicate}; @@ -1372,7 +1371,7 @@ impl Db { let insert_result = sqlx::query( "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL, $10) \ ON CONFLICT DO NOTHING", ) .bind(community_id.as_uuid()) @@ -1384,7 +1383,6 @@ impl Db { .bind(&event.content) .bind(sig_bytes.as_slice()) .bind(received_at) - .bind::>(None) .bind(d_tag.as_deref()) .execute(tx.conn()) .await?; diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 40e40e8db4..b491107fc5 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1626,12 +1626,19 @@ fn masker_blanks_every_literal_form_and_keeps_code() { } } -/// Normalize SQL case, whitespace and Rust string continuations/escapes for +/// Normalize SQL case, quoted identifiers, whitespace and Rust string escapes for /// the event INSERT and UPDATE policies. fn normalized_sql_source(source: &str) -> String { let mut normalized = String::with_capacity(source.len()); let mut chars = source.chars().peekable(); while let Some(ch) = chars.next() { + if ch == '"' { + continue; + } + if ch == '\\' && chars.peek() == Some(&'"') { + chars.next(); + continue; + } if ch.is_whitespace() || ch == '\\' { if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) { chars.next(); @@ -1698,6 +1705,9 @@ pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) { .await .expect("write"); } +pub(crate) async fn quoted_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO \"public\".\"events\" (id) VALUES ($1)"); +} pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO events \ (community_id, id) VALUES ($1, $2)") @@ -1862,6 +1872,7 @@ mod tests { event_insert_follow_up_violations(&strip_cfg_test_items(source)), [ "pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {", + "pub(crate) async fn quoted_writer(tx: &mut AdmittedTx) {", "pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {", "pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {", "pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {", @@ -2076,34 +2087,60 @@ const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 2] = [ fn is_channel_null_event_writer(function: &str) -> bool { let header = function_header(function); + let sql = normalized_sql_source(function).replace(' ', ""); if header.starts_with("pub async fn publish_nip43_membership_locked(") { - // Kind 13534: the channel bind is always NULL. - return function.contains(".bind::>(None)"); + // Kind 13534: channel_id is the tenth value, a literal NULL. + return sql.contains("values($1,$2,$3,$4,$5,$6,$7,$8,$9,null,$10)"); } if header.starts_with("pub async fn accept_lease_event(") { - // Kind 30350: channel_id is the literal NULL in VALUES. - return normalized_sql_source(function).contains("now(),null,$8)"); + // Kind 30350: channel_id is the literal NULL after received_at. + return sql.contains("now(),null,$8)"); } false } fn updates_event_floor_columns(function: &str) -> bool { let normalized = normalized_sql_source(function); - ["update events set ", "update public.events set "] - .iter() - .any(|marker| { - normalized.match_indices(marker).any(|(index, marker)| { - let assignments = normalized[index + marker.len()..] - .split(" where ") - .next() - .expect("split has first part"); - assignments.split(',').any(|assignment| { - assignment.split_once('=').is_some_and(|(column, _)| { - matches!(column.trim(), "created_at" | "channel_id") - }) + normalized.match_indices("update ").any(|(index, marker)| { + let Some((target, assignments)) = normalized[index + marker.len()..].split_once(" set ") + else { + return false; + }; + let mut target = target.split_whitespace(); + let mut table = target.next(); + if table == Some("only") { + table = target.next(); + } + if !matches!(table, Some("events" | "public.events")) { + return false; + } + // The target may have an alias, with or without AS. + let alias = target.collect::>(); + if !matches!(alias.as_slice(), [] | [_] | ["as", _]) { + return false; + } + let assignments = assignments.split(" where ").next().expect("first part"); + let mut depth = 0usize; + assignments + .split(|ch| { + match ch { + '(' => depth += 1, + ')' => depth = depth.saturating_sub(1), + _ => {} + } + ch == ',' && depth == 0 + }) + .any(|assignment| { + assignment.split_once('=').is_some_and(|(columns, _)| { + columns + .trim() + .trim_start_matches('(') + .trim_end_matches(')') + .split(',') + .any(|column| matches!(column.trim(), "created_at" | "channel_id")) }) }) - }) + }) } fn event_floor_policy_violations(production: &str) -> Vec { @@ -2144,6 +2181,36 @@ async fn rewrite_time(tx: &mut AdmittedTx) { async fn attach_channel(tx: &mut AdmittedTx) { sqlx::query("uPdAtE public.events\nSET channel_id=$1 WHERE id=$2"); } +async fn alias_time(tx: &mut AdmittedTx) { + sqlx::query("UPDATE events e SET created_at = $1 WHERE e.id = $2"); +} +async fn as_alias_time(tx: &mut AdmittedTx) { + sqlx::query("UPDATE events AS e SET created_at = $1 WHERE e.id = $2"); +} +async fn only_channel(tx: &mut AdmittedTx) { + sqlx::query("UPDATE ONLY events SET channel_id = $1 WHERE id = $2"); +} +async fn only_public_alias(tx: &mut AdmittedTx) { + sqlx::query("UPDATE ONLY public.events AS e SET channel_id = $1 WHERE e.id = $2"); +} +async fn row_time(tx: &mut AdmittedTx) { + sqlx::query("UPDATE events SET (created_at, kind) = ($1, 1) WHERE id = $2"); +} +async fn row_channel(tx: &mut AdmittedTx) { + sqlx::query("UPDATE events SET (kind, channel_id) = (1, $1) WHERE id = $2"); +} +async fn quoted_insert(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO \"events\" (id) VALUES ($1)"); +} +async fn quoted_schema_insert(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO \"public\".\"events\" (id) VALUES ($1)"); +} +async fn quoted_update(tx: &mut AdmittedTx) { + sqlx::query("UPDATE \"events\" SET created_at = $1 WHERE id = $2"); +} +async fn quoted_schema_update(tx: &mut AdmittedTx) { + sqlx::query("UPDATE \"public\".\"events\" SET channel_id = $1 WHERE id = $2"); +} async fn delete_only(tx: &mut AdmittedTx) { sqlx::query("UPDATE events SET deleted_at=now() WHERE channel_id=$1 AND created_at<$2"); } @@ -2154,26 +2221,48 @@ async fn delete_only(tx: &mut AdmittedTx) { "async fn missing_guard(tx: &mut AdmittedTx) {", "async fn rewrite_time(tx: &mut AdmittedTx) {", "async fn attach_channel(tx: &mut AdmittedTx) {", + "async fn alias_time(tx: &mut AdmittedTx) {", + "async fn as_alias_time(tx: &mut AdmittedTx) {", + "async fn only_channel(tx: &mut AdmittedTx) {", + "async fn only_public_alias(tx: &mut AdmittedTx) {", + "async fn row_time(tx: &mut AdmittedTx) {", + "async fn row_channel(tx: &mut AdmittedTx) {", + "async fn quoted_insert(tx: &mut AdmittedTx) {", + "async fn quoted_schema_insert(tx: &mut AdmittedTx) {", + "async fn quoted_update(tx: &mut AdmittedTx) {", + "async fn quoted_schema_update(tx: &mut AdmittedTx) {", ] ); } +#[test] +fn event_floor_policy_requires_nip43_channel_null_at_its_insert_position() { + let literal_null = r#" +pub async fn publish_nip43_membership_locked(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id,id,pubkey,kind,content,tags,sig,created_at,received_at,channel_id,d_tag) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,NULL,$10)"); +} +"#; + assert!(event_floor_policy_violations(literal_null).is_empty()); + let unrelated_null_bind = literal_null.replace("$9,NULL,$10", "$9,$10,$11").replace( + "sqlx::query(", + "let unrelated = query.bind::>(None); sqlx::query(", + ); + assert_eq!( + event_floor_policy_violations(&unrelated_null_bind).len(), + 1, + "an unrelated NULL bind must not exempt a channel-bearing INSERT" + ); +} + #[test] fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it() { - fn collect(dir: &std::path::Path, files: &mut Vec) { - for entry in std::fs::read_dir(dir).expect("read source directory") { - let path = entry.expect("source entry").path(); - if path.is_dir() { - collect(&path, files); - } else if should_scan_guarded_write_source_file(&path) { - files.push(path); - } - } - } - let mut files = Vec::new(); - collect( - &std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"), - &mut files, + let files = production_event_writer_source_files(); + let crates_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("crates root"); + assert!( + files.contains(&crates_root.join("buzz-relay/src/state.rs")), + "floor policy must scan the workspace, including relay writers" ); let mut guarded_writers = 0; let mut violations = Vec::new(); @@ -2238,16 +2327,41 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() { std::fs::set_permissions(&psql, std::fs::Permissions::from_mode(0o755)) .expect("executable stub"); let witness = root.join("psql-was-invoked"); - let output = std::process::Command::new("/bin/bash") - .arg(script) - .env("PGHOST", "shared-db.example.invalid") - .env("PATH", format!("{}:/usr/bin:/bin", root.display())) - .env("SEED_PSQL_WITNESS", &witness) - .output() - .expect("run local seed guard"); - let invoked = witness.exists(); + let cases = [ + ("remote host", "shared-db.example.invalid", "", ""), + ("address override", "localhost", "192.0.2.1", ""), + ("service override", "localhost", "", "production"), + ( + "socket-prefixed host list", + "/tmp,shared-db.example.invalid", + "", + "", + ), + ("host list", "localhost,shared-db.example.invalid", "", ""), + ]; + let results = cases.map(|(name, host, address, service)| { + let output = std::process::Command::new("/bin/bash") + .arg(&script) + .env("PGHOST", host) + .env("PGHOSTADDR", address) + .env("PGSERVICE", service) + .env("PATH", format!("{}:/usr/bin:/bin", root.display())) + .env("SEED_PSQL_WITNESS", &witness) + .output() + .expect("run local seed guard"); + let invoked = witness.exists(); + if invoked { + std::fs::remove_file(&witness).expect("clear witness"); + } + (name, output, invoked) + }); std::fs::remove_dir_all(root).expect("remove scratch fixture"); - assert!(!invoked, "a non-local target reached psql"); - assert!(!output.status.success()); - assert!(String::from_utf8_lossy(&output.stderr).contains("requires a local PostgreSQL target")); + for (name, output, invoked) in results { + assert!(!invoked, "{name} reached psql"); + assert!(!output.status.success(), "{name} accepted"); + assert!( + String::from_utf8_lossy(&output.stderr).contains("requires a local PostgreSQL target"), + "{name}: missing refusal" + ); + } } diff --git a/scripts/seed-admin-dashboard.sh b/scripts/seed-admin-dashboard.sh index 5372e1d014..e015560481 100755 --- a/scripts/seed-admin-dashboard.sh +++ b/scripts/seed-admin-dashboard.sh @@ -20,6 +20,10 @@ export PGPASSWORD="${PGPASSWORD:-buzz_dev}" export PGDATABASE="${PGDATABASE:-buzz}" # This raw fixture writer is unarmed and must never target a live replica fence. +if [[ -n "${PGHOSTADDR:-}" || -n "${PGSERVICE:-}" || "${PGHOST}" == *,* ]]; then + echo "error: admin seed requires a local PostgreSQL target without PGHOSTADDR, PGSERVICE or multiple hosts" >&2 + exit 1 +fi case "${PGHOST}" in localhost|127.0.0.1|::1|/*) ;; *)