diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index 09ea6dd89c..d26981804f 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -1,7 +1,10 @@ //! Community-admitted event-write transactions. +use std::collections::BTreeSet; + use buzz_core::CommunityId; use sqlx::{PgConnection, Postgres, Transaction}; +use uuid::Uuid; use crate::deletion::{DeletionStore, ServingWriteLease}; use crate::Result; @@ -68,6 +71,8 @@ use crate::Result; pub struct AdmittedTx { tx: Transaction<'static, Postgres>, community: CommunityId, + /// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes. + ttl_refresh_channels: BTreeSet, } impl AdmittedTx { @@ -78,7 +83,11 @@ impl AdmittedTx { community: CommunityId, ) -> Result { store.guard_transaction(&mut tx, community).await?; - Ok(Self { tx, community }) + Ok(Self { + tx, + community, + ttl_refresh_channels: BTreeSet::new(), + }) } /// Validate `lease` on `tx` under the community admission lock and wrap @@ -94,6 +103,7 @@ impl AdmittedTx { Ok(Self { tx, community: lease.community_id, + ttl_refresh_channels: BTreeSet::new(), }) } @@ -111,9 +121,29 @@ impl AdmittedTx { &mut self.tx } + /// Record that an event committed with this transaction belongs to + /// `channel_id`, so [`AdmittedTx::commit`] refreshes the channel's TTL. + pub(crate) fn record_channel_event(&mut self, channel_id: Option, kind: i32) { + if let Some(channel) = crate::store::event_follow_up::ttl_refresh_channel(channel_id, kind) + { + self.ttl_refresh_channels.insert(channel); + } + } + /// Commit the transaction. This is the only commit path for admitted /// event writes. - pub async fn commit(self) -> Result<()> { + /// + /// Refreshes the TTL of every channel that received an event first, as + /// the last statement before COMMIT. + pub async fn commit(mut self) -> Result<()> { + if !self.ttl_refresh_channels.is_empty() { + crate::store::event_follow_up::refresh_channel_ttls( + &mut self.tx, + self.community, + &self.ttl_refresh_channels, + ) + .await?; + } self.tx.commit().await?; Ok(()) } diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 5e89b1101b..8d43473d50 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -585,11 +585,7 @@ pub async fn update_channel( if updates.ttl_seconds.is_some() { let mut tx = begin_event_write_transaction(pool).await?; sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") - .bind(format!( - "buzz_channel_ttl:{}:{}", - community_id.as_uuid(), - channel_id - )) + .bind(channel_ttl_lock_key(community_id, channel_id)) .execute(&mut *tx) .await?; let result = q.execute(&mut *tx).await?; @@ -1012,6 +1008,14 @@ impl Db { } } +/// The per-channel TTL advisory lock key. Event commits take it SHARED +/// (`event_follow_up::refresh_channel_ttls`, and the 0024 trigger until it is +/// retired); TTL transitions in [`update_channel`] take it EXCLUSIVE. Both +/// sides build it here, so the keys cannot drift. +pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String { + format!("buzz_channel_ttl:{}:{channel}", community.as_uuid()) +} + #[cfg(test)] mod postgres_tests { use super::*; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index a75f2e0483..5cf3ed40dc 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -398,6 +398,13 @@ impl LockedMemberSnapshot { "member snapshot event id already exists".into(), )); } + crate::store::event_follow_up::after_admitted_insert( + &mut self.tx, + event.id.as_bytes().as_slice(), + kind, + Some(channel_id), + ) + .await?; crate::insert_mentions_in_transaction(&mut self.tx, event, Some(channel_id)).await?; Ok(( buzz_core::StoredEvent::with_received_at( diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index c35610c1cf..9dd4793d0d 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -500,6 +500,15 @@ async fn insert_event_on( .await?; let was_inserted = result.rows_affected() > 0; + if was_inserted { + crate::store::event_follow_up::after_admitted_insert( + tx, + id_bytes.as_slice(), + kind_i32, + channel_id, + ) + .await?; + } Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), @@ -1744,6 +1753,14 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let was_inserted = result.rows_affected() > 0; if was_inserted { + crate::store::event_follow_up::after_admitted_insert( + tx, + id_bytes.as_slice(), + kind_i32, + channel_id, + ) + .await?; + if let Some(ref meta) = thread_meta { let broadcast_val: bool = meta.broadcast; diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs new file mode 100644 index 0000000000..98fc08c29a --- /dev/null +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -0,0 +1,208 @@ +//! App-owned follow-up work for a newly inserted `events` row. +//! +//! Two pieces of work follow every durable event insert: +//! +//! - **Push match enqueue.** Push-eligible kinds get a `push_match_queue` row +//! when the community has an eligible lease. This runs right after the +//! INSERT, in the same transaction, and any failure rejects the event. +//! - **Channel TTL refresh.** A channel-scoped event pushes its ephemeral +//! channel's `ttl_deadline` forward. [`AdmittedTx`] records the channel and +//! runs the refresh as the last statement before COMMIT (see +//! [`refresh_channel_ttls`]). A refresh failure is logged and counted, and +//! the event still commits. +//! +//! This is the application-side replacement for the `events_enqueue_push_match` +//! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040). +//! Both run during the migration window; the push enqueue is idempotent +//! (`ON CONFLICT DO NOTHING`) and a second TTL refresh is harmless. Every +//! production `INSERT INTO events` must call this module, which +//! `tests/observability_source.rs` enforces. + +use std::collections::BTreeSet; + +use buzz_core::CommunityId; +use sqlx::PgConnection; +use uuid::Uuid; + +use crate::{AdmittedTx, DbError, Result}; + +/// Event kinds that wake push subscribers. Keep identical to the relay's +/// validated NIP-PL descriptor. +pub(crate) const PUSH_MATCH_KINDS: [i32; 4] = [9, 40002, 45001, 45003]; + +/// Kind 9007 creates the channel and initializes its deadline itself. +const KIND_CHANNEL_CREATE: i32 = 9007; + +/// SQLSTATE `lock_not_available`, raised when `lock_timeout` expires. +const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; + +/// Follow-up for an event row inserted in an admitted transaction. +/// +/// Call only when the INSERT affected a row. Enqueues the push match job now +/// and records the channel for the pre-commit TTL refresh. +/// +/// If the insert ran inside a savepoint that may still roll back, call +/// [`enqueue_push_match`] inside the savepoint and +/// [`AdmittedTx::record_channel_event`] after it is released instead. +pub(crate) async fn after_admitted_insert( + tx: &mut AdmittedTx, + event_id: &[u8], + kind: i32, + channel_id: Option, +) -> Result<()> { + let community = tx.community(); + enqueue_push_match(tx.conn(), community, event_id, kind).await?; + tx.record_channel_event(channel_id, kind); + Ok(()) +} + +/// Enqueue a push match job for a just-inserted event, if its kind is +/// push-eligible and the community has an eligible lease. +/// +/// Takes the push gate SHARED and holds it to transaction end. Lease +/// activations take it EXCLUSIVE (`acquire_push_gate_lock` in `push.rs`), so +/// an event either sees the committed lease or strictly precedes the +/// activation, in which case no wake was owed. The lock must be its own +/// statement: under READ COMMITTED the eligibility check needs a snapshot +/// taken after the lock is granted. +/// +/// Errors propagate and reject the event, as the trigger did. +pub(crate) async fn enqueue_push_match( + conn: &mut PgConnection, + community: CommunityId, + event_id: &[u8], + kind: i32, +) -> Result<()> { + if !PUSH_MATCH_KINDS.contains(&kind) { + return Ok(()); + } + crate::observability::observe_advisory_lock( + crate::observability::LockType::PushGate, + sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))") + .bind(crate::push::push_gate_lock_key(community)) + .execute(&mut *conn), + ) + .await?; + sqlx::query( + "INSERT INTO push_match_queue (community_id, event_id) \ + SELECT $1, $2 \ + WHERE EXISTS ( \ + SELECT 1 FROM push_leases \ + WHERE community_id = $1 \ + AND active \ + AND endpoint_enabled \ + AND expires_at > EXTRACT(EPOCH FROM now())::bigint) \ + ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(event_id) + .execute(&mut *conn) + .await?; + Ok(()) +} + +/// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any. +pub(crate) fn ttl_refresh_channel(channel_id: Option, kind: i32) -> Option { + channel_id.filter(|_| kind != KIND_CHANNEL_CREATE) +} + +/// Refresh the TTL deadline of every channel that received an event in this +/// transaction. [`AdmittedTx::commit`] calls this as its last work before +/// COMMIT, which keeps the deferred-trigger timing from migration 0024: the +/// deadline is computed at commit. An ephemeral channel's row lock, taken by +/// the UPDATE, is held through the remaining refreshes and the COMMIT round +/// trip; the trigger held it only inside COMMIT. A permanent channel's row is +/// never locked. +/// +/// Channels are refreshed in sorted order, so two transactions that touch the +/// same channels cannot deadlock on the shared locks against an exclusive +/// waiter. Each refresh runs in its own savepoint. A failure is rolled back, +/// logged, and counted, and the event still commits, as with the trigger's +/// `EXCEPTION WHEN OTHERS`. A cancelled statement (`57014`) still rejects the +/// transaction, because plpgsql never caught it either. +pub(crate) async fn refresh_channel_ttls( + conn: &mut PgConnection, + community: CommunityId, + channels: &BTreeSet, +) -> Result<()> { + for &channel in channels { + sqlx::query("SAVEPOINT channel_ttl_refresh") + .execute(&mut *conn) + .await?; + match refresh_channel_ttl(conn, community, channel).await { + Ok(()) => { + sqlx::query("RELEASE SAVEPOINT channel_ttl_refresh") + .execute(&mut *conn) + .await?; + } + Err(error) => { + // plpgsql `WHEN OTHERS` never caught `query_canceled`, so a + // cancelled refresh has always rejected the event. + if error.is_statement_cancelled() { + return Err(error); + } + sqlx::query("ROLLBACK TO SAVEPOINT channel_ttl_refresh") + .execute(&mut *conn) + .await?; + sqlx::query("RELEASE SAVEPOINT channel_ttl_refresh") + .execute(&mut *conn) + .await?; + let reason = if sqlstate(&error).as_deref() == Some(SQLSTATE_LOCK_NOT_AVAILABLE) { + "lock_timeout" + } else { + "error" + }; + metrics::counter!( + "buzz_db_channel_ttl_refresh_failures_total", + "reason" => reason, + ) + .increment(1); + tracing::warn!( + %error, + channel_id = %channel, + reason, + "channel TTL refresh failed; event committed without refreshing the deadline" + ); + } + } + } + Ok(()) +} + +async fn refresh_channel_ttl( + conn: &mut PgConnection, + community: CommunityId, + channel: Uuid, +) -> Result<()> { + // SHARED here, EXCLUSIVE in `update_channel` for TTL transitions: the same + // total order the 0022 row lock gave, without serializing hot channels. + sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))") + .bind(crate::channel::channel_ttl_lock_key(community, channel)) + .execute(&mut *conn) + .await?; + // A separate statement, so its snapshot is taken after the lock is + // granted. `ttl_seconds IS NOT NULL` filters a permanent channel before + // any tuple lock is taken, so its row is never locked (migration 0024). + sqlx::query( + "UPDATE channels \ + SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds) \ + WHERE community_id = $1 AND id = $2 \ + AND ttl_seconds IS NOT NULL \ + AND archived_at IS NULL \ + AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(channel) + .execute(&mut *conn) + .await?; + Ok(()) +} + +fn sqlstate(error: &DbError) -> Option { + match error { + DbError::Sqlx(sqlx::Error::Database(database_error)) => { + database_error.code().map(|code| code.into_owned()) + } + _ => None, + } +} diff --git a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs new file mode 100644 index 0000000000..10c396dfb7 --- /dev/null +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -0,0 +1,773 @@ +//! PostgreSQL contract for the event follow-up producers (push match +//! enqueue and channel TTL refresh), on the desired and migration schemas, +//! with the follow-up triggers present (dual) and dropped (app-only). +use crate::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus}; +use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError}; +use buzz_core::CommunityId; +use chrono::{DateTime, Utc}; +use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use sqlx::PgPool; +use std::time::Duration; +use tokio::time::timeout; +use uuid::Uuid; + +async fn connect(migration_schema: bool) -> Db { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .expect("Postgres wrapper must provide an isolated database URL"); + let expected_mode = if migration_schema { + "migration" + } else { + "desired" + }; + assert_eq!( + std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), + Ok(expected_mode) + ); + let config = DbConfig { + database_url, + ..DbConfig::default() + }; + let db = Db::new(&config) + .await + .expect("connect to isolated database"); + if migration_schema { + migration::run_migrations(db.pool()) + .await + .expect("apply embedded migrations to empty database"); + } + db +} + +async fn create_community(pool: &PgPool) -> CommunityId { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("follow-up-{}.example", id.simple())) + .execute(pool) + .await + .expect("create disposable community"); + CommunityId::from_uuid(id) +} + +async fn channel_with_ttl( + pool: &PgPool, + community: CommunityId, + author: &[u8], + ttl_seconds: Option, +) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query( + "INSERT INTO channels \ + (id, community_id, name, created_by, ttl_seconds, ttl_deadline) \ + VALUES ($1, $2, $3, $4, $5, \ + CASE WHEN $5::INT IS NULL THEN NULL \ + ELSE clock_timestamp() - interval '1 day' END)", + ) + .bind(id) + .bind(community.as_uuid()) + .bind(format!("follow-up-{}", id.simple())) + .bind(author) + .bind(ttl_seconds) + .execute(pool) + .await + .expect("create disposable channel"); + id +} + +async fn create_channel(pool: &PgPool, community: CommunityId, author: &[u8]) -> Uuid { + channel_with_ttl(pool, community, author, Some(60)).await +} + +fn signed_event(keys: &Keys, kind: u16, content: &str) -> Event { + EventBuilder::new(Kind::Custom(kind), content) + .sign_with_keys(keys) + .expect("sign test event") +} + +async fn match_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(pool) + .await + .expect("count push match jobs") +} + +async fn event_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(pool) + .await + .expect("count stored events") +} + +async fn deadline(pool: &PgPool, community: CommunityId, channel: Uuid) -> DateTime { + sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(pool) + .await + .expect("read channel deadline") +} + +async fn optional_deadline( + pool: &PgPool, + community: CommunityId, + channel: Uuid, +) -> Option> { + sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(pool) + .await + .expect("read optional channel deadline") +} + +async fn reset_deadline(pool: &PgPool, community: CommunityId, channel: Uuid) -> DateTime { + sqlx::query_scalar( + "UPDATE channels SET ttl_deadline=clock_timestamp()-interval '1 day' \ + WHERE community_id=$1 AND id=$2 RETURNING ttl_deadline", + ) + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(pool) + .await + .expect("reset disposable channel deadline") +} + +/// `buzz_db_channel_ttl_refresh_failures_total` by `reason` label. +fn ttl_refresh_failures(snapshotter: &Snapshotter) -> Vec<(String, u64)> { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_db_channel_ttl_refresh_failures_total") + .map(|(key, _, _, value)| { + let reason = key + .key() + .labels() + .find(|label| label.key() == "reason") + .map(|label| label.value().to_owned()) + .unwrap_or_default(); + let DebugValue::Counter(count) = value else { + panic!("TTL refresh failures must be a counter"); + }; + (reason, count) + }) + .collect() +} + +async fn begin_caller_owned_event_transaction(db: &Db, community: CommunityId) -> AdmittedTx { + db.begin_event_write_transaction(community) + .await + .expect("begin admitted event transaction") +} + +fn assert_sqlstate(error: &DbError, expected: &str) { + let DbError::Sqlx(sqlx::Error::Database(database_error)) = error else { + panic!("expected a Postgres database error, got {error:?}"); + }; + assert_eq!(database_error.code().as_deref(), Some(expected)); +} + +async fn activate_lease( + pool: &PgPool, + community: CommunityId, + keys: &Keys, + marker: u8, + expires_at: i64, +) { + let source_event_id = [marker; 32]; + let endpoint_hash = [marker.wrapping_add(64); 32]; + let subscriptions = serde_json::json!([]); + assert_eq!( + push::replace_active_lease( + pool, + community, + &keys.public_key().to_bytes(), + "follow-up-installation", + push::LeaseVersion { + source_event_id: &source_event_id, + source_created_at: 1, + generation: 1, + expires_at, + }, + push::ActiveLease { + app_profile: "ios-production", + endpoint_hash: &endpoint_hash, + endpoint_grant: "test-grant", + max_class: "default", + subscriptions: &subscriptions, + }, + ) + .await + .expect("activate push lease"), + push::ReplaceLeaseOutcome::Accepted + ); +} + +/// Which producers are live. `Dual` is production during the migration +/// window: the app follow-ups plus the 0023/0024 triggers. `AppOnly` drops +/// both triggers in this disposable database, as the retirement migration +/// will. +#[derive(Clone, Copy, Debug)] +enum Arm { + Dual, + AppOnly, +} + +async fn apply_arm(pool: &PgPool, arm: Arm) { + let triggers: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger \ + WHERE tgrelid = 'events'::regclass \ + AND tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')", + ) + .fetch_one(pool) + .await + .expect("count follow-up triggers"); + assert_eq!(triggers, 2, "both follow-up triggers exist before the arm"); + if let Arm::AppOnly = arm { + for statement in [ + "DROP TRIGGER events_enqueue_push_match ON events", + "DROP TRIGGER events_refresh_channel_ttl ON events", + ] { + sqlx::query(statement) + .execute(pool) + .await + .expect("drop follow-up trigger in disposable database"); + } + let remaining: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger \ + WHERE tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')", + ) + .fetch_one(pool) + .await + .expect("count remaining follow-up triggers, including partition clones"); + assert_eq!( + remaining, 0, + "app-only arm must leave no trigger on any partition" + ); + } +} + +async fn assert_contract(migration_schema: bool, arm: Arm) { + let db = connect(migration_schema).await; + let pool = db.pool(); + apply_arm(pool, arm).await; + let keys = Keys::generate(); + let community = create_community(pool).await; + let channel = create_channel(pool, community, &keys.public_key().to_bytes()).await; + let now = Utc::now().timestamp(); + activate_lease(pool, community, &keys, 11, now + 3600).await; + + for kind in [9, 40002, 45001, 45003] { + let deadline_before_first_insert = reset_deadline(pool, community, channel).await; + let message = signed_event(&keys, kind, &format!("allowlisted-{kind}")); + assert!( + db.insert_event(community, &message, Some(channel)) + .await + .expect("insert allowlisted event") + .1 + ); + assert_eq!(event_count(pool, community, &message).await, 1); + assert_eq!(match_count(pool, community, &message).await, 1); + let deadline_after_first_insert = deadline(pool, community, channel).await; + assert!( + deadline_after_first_insert > deadline_before_first_insert, + "first insertion of kind {kind} must refresh the channel TTL" + ); + assert!( + !db.insert_event(community, &message, Some(channel)) + .await + .expect("deduplicate event") + .1 + ); + assert_eq!(match_count(pool, community, &message).await, 1); + assert_eq!( + deadline(pool, community, channel).await, + deadline_after_first_insert, + "replaying kind {kind} must not refresh the channel TTL" + ); + } + + let reaction = signed_event(&keys, 7, "not-push-eligible"); + assert!( + db.insert_event(community, &reaction, Some(channel)) + .await + .expect("insert reaction") + .1 + ); + assert_eq!(match_count(pool, community, &reaction).await, 0); + + let no_lease_community = create_community(pool).await; + let no_lease_message = signed_event(&keys, 9, "no-lease"); + assert!( + db.insert_event(no_lease_community, &no_lease_message, None) + .await + .expect("insert without lease") + .1 + ); + assert_eq!( + match_count(pool, no_lease_community, &no_lease_message).await, + 0 + ); + + for (marker, state) in [(21_u8, "expired"), (22, "inactive"), (23, "disabled")] { + let predicate_community = create_community(pool).await; + activate_lease( + pool, + predicate_community, + &keys, + marker, + if state == "expired" { + now - 1 + } else { + now + 3600 + }, + ) + .await; + if state == "inactive" { + let source_event_id = [marker.wrapping_add(32); 32]; + assert_eq!( + push::revoke_lease( + pool, + predicate_community, + &keys.public_key().to_bytes(), + "follow-up-installation", + push::LeaseVersion { + source_event_id: &source_event_id, + source_created_at: 2, + generation: 2, + expires_at: now + 3600, + }, + ) + .await + .expect("revoke disposable push lease"), + push::ReplaceLeaseOutcome::Accepted + ); + } else if state == "disabled" { + sqlx::query("UPDATE push_leases SET endpoint_enabled=false WHERE community_id=$1") + .bind(predicate_community.as_uuid()) + .execute(pool) + .await + .expect("disable disposable push endpoint"); + } + let message = signed_event(&keys, 9, &format!("{state}-lease")); + assert!( + db.insert_event(predicate_community, &message, None) + .await + .expect("insert event for ineligible lease") + .1 + ); + assert_eq!(event_count(pool, predicate_community, &message).await, 1); + assert_eq!(match_count(pool, predicate_community, &message).await, 0); + } + + let mut push_gate_holder = pool.begin().await.expect("begin push-gate lock holder"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!("buzz_push_gate:{}", community.as_uuid())) + .execute(&mut *push_gate_holder) + .await + .expect("hold exclusive push-gate lock"); + let gated = signed_event(&keys, 9, "push-gate-timeout"); + let mut gated_tx = begin_caller_owned_event_transaction(&db, community).await; + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(gated_tx.conn()) + .await + .expect("bound push-gate lock wait"); + let gated_error = event::insert_event_in_transaction(&mut gated_tx, &gated, None) + .await + .expect_err("push-eligible insert must fail closed when its gate cannot be acquired"); + assert_sqlstate(&gated_error, "55P03"); + gated_tx + .rollback() + .await + .expect("rollback failed push-gate transaction"); + assert_eq!(event_count(pool, community, &gated).await, 0); + assert_eq!(match_count(pool, community, &gated).await, 0); + + let ungated = signed_event(&keys, 7, "push-gate-non-eligible-control"); + let mut ungated_tx = begin_caller_owned_event_transaction(&db, community).await; + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(ungated_tx.conn()) + .await + .expect("bound unrelated lock wait"); + assert!( + event::insert_event_in_transaction(&mut ungated_tx, &ungated, None) + .await + .expect("non-push event must not use the push gate") + .1 + ); + ungated_tx + .commit() + .await + .expect("commit non-push control while gate held"); + assert_eq!(event_count(pool, community, &ungated).await, 1); + assert_eq!(match_count(pool, community, &ungated).await, 0); + push_gate_holder + .rollback() + .await + .expect("release exclusive push-gate lock"); + let gated_control = signed_event(&keys, 9, "push-gate-released-control"); + assert!( + db.insert_event(community, &gated_control, None) + .await + .expect("insert after push gate release") + .1 + ); + assert_eq!(event_count(pool, community, &gated_control).await, 1); + assert_eq!(match_count(pool, community, &gated_control).await, 1); + + let rollback_deadline = reset_deadline(pool, community, channel).await; + let rolled_back = signed_event(&keys, 9, "rolled-back"); + let mut rollback_tx = begin_caller_owned_event_transaction(&db, community).await; + assert!( + event::insert_event_in_transaction(&mut rollback_tx, &rolled_back, Some(channel)) + .await + .expect("insert before rollback") + .1 + ); + let in_transaction: i64 = sqlx::query_scalar( + "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(rolled_back.id.as_bytes().as_slice()) + .fetch_one(rollback_tx.conn()) + .await + .expect("read uncommitted follow-up"); + assert_eq!(in_transaction, 1); + rollback_tx + .rollback() + .await + .expect("rollback event and follow-up"); + assert_eq!(event_count(pool, community, &rolled_back).await, 0); + assert_eq!(match_count(pool, community, &rolled_back).await, 0); + assert_eq!(deadline(pool, community, channel).await, rollback_deadline); + + let before_commit = reset_deadline(pool, community, channel).await; + let committed = signed_event(&keys, 9, "deferred-ttl"); + let mut commit_tx = begin_caller_owned_event_transaction(&db, community).await; + assert!( + event::insert_event_in_transaction(&mut commit_tx, &committed, Some(channel)) + .await + .expect("insert before commit") + .1 + ); + let deadline_before_commit: DateTime = + sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(commit_tx.conn()) + .await + .expect("read deadline inside event transaction"); + assert_eq!(deadline_before_commit, before_commit); + let database_time_before_commit: DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(commit_tx.conn()) + .await + .expect("sample database clock immediately before commit"); + commit_tx + .commit() + .await + .expect("commit deferred TTL refresh"); + assert!( + deadline(pool, community, channel).await + >= database_time_before_commit + chrono::Duration::seconds(60), + "the deferred trigger must refresh TTL at commit" + ); + assert_eq!(match_count(pool, community, &committed).await, 1); + + let before_skips = reset_deadline(pool, community, channel).await; + let channel_create = signed_event(&keys, 9007, "channel-create"); + assert!( + db.insert_event(community, &channel_create, Some(channel)) + .await + .expect("insert kind 9007") + .1 + ); + assert_eq!(deadline(pool, community, channel).await, before_skips); + assert_eq!(match_count(pool, community, &channel_create).await, 0); + let channelless = signed_event(&keys, 9, "channel-null"); + assert!( + db.insert_event(community, &channelless, None) + .await + .expect("insert channel-less event") + .1 + ); + assert_eq!(match_count(pool, community, &channelless).await, 1); + + let no_ttl_channel = + channel_with_ttl(pool, community, &keys.public_key().to_bytes(), None).await; + let no_ttl_message = signed_event(&keys, 7, "ttl-null"); + assert!( + db.insert_event(community, &no_ttl_message, Some(no_ttl_channel)) + .await + .expect("insert into channel without TTL") + .1 + ); + assert_eq!( + optional_deadline(pool, community, no_ttl_channel).await, + None + ); + + for (column, marker) in [("archived_at", "archived"), ("deleted_at", "deleted")] { + let inactive_channel = create_channel(pool, community, &keys.public_key().to_bytes()).await; + let inactive_deadline = deadline(pool, community, inactive_channel).await; + let update = match column { + "archived_at" => { + "UPDATE channels SET archived_at=clock_timestamp() \ + WHERE community_id=$1 AND id=$2" + } + "deleted_at" => { + "UPDATE channels SET deleted_at=clock_timestamp() \ + WHERE community_id=$1 AND id=$2" + } + _ => unreachable!("fixed test column"), + }; + sqlx::query(update) + .bind(community.as_uuid()) + .bind(inactive_channel) + .execute(pool) + .await + .expect("mark disposable channel inactive"); + let message = signed_event(&keys, 7, &format!("{marker}-channel")); + assert!( + db.insert_event(community, &message, Some(inactive_channel)) + .await + .expect("insert against inactive disposable channel") + .1 + ); + assert_eq!(event_count(pool, community, &message).await, 1); + assert_eq!( + deadline(pool, community, inactive_channel).await, + inactive_deadline + ); + } + + let before_lock_timeout = reset_deadline(pool, community, channel).await; + let mut lock_holder = pool.begin().await.expect("begin lock holder"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!( + "buzz_channel_ttl:{}:{channel}", + community.as_uuid() + )) + .execute(&mut *lock_holder) + .await + .expect("hold exclusive TTL lock"); + let locked_event = signed_event(&keys, 9, "ttl-lock-timeout"); + let mut locked_tx = begin_caller_owned_event_transaction(&db, community).await; + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(locked_tx.conn()) + .await + .expect("bound TTL lock wait"); + assert!( + event::insert_event_in_transaction(&mut locked_tx, &locked_event, Some(channel)) + .await + .expect("insert while TTL lock held") + .1 + ); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + { + let _local = metrics::set_default_local_recorder(&recorder); + timeout(Duration::from_secs(5), locked_tx.commit()) + .await + .expect("commit must not hang on TTL lock") + .expect("TTL warning must not abort event commit"); + } + assert_eq!( + ttl_refresh_failures(&snapshotter), + [("lock_timeout".to_owned(), 1)], + "a swallowed TTL refresh failure must be counted once, by reason" + ); + + // A cancelled statement is not swallowed: plpgsql `WHEN OTHERS` never + // caught 57014, so the event is rejected. Both arms discriminate: if + // the app swallowed the cancel, the app-only arm would commit the + // event, and in the dual arm the trigger's own wait at COMMIT is not + // cut short by the 200ms statement timeout, so the commit outlives the + // 5s guard. + let cancelled_event = signed_event(&keys, 9, "ttl-statement-cancel"); + let mut cancelled_tx = begin_caller_owned_event_transaction(&db, community).await; + sqlx::query("SET LOCAL statement_timeout = '200ms'") + .execute(cancelled_tx.conn()) + .await + .expect("bound TTL statement"); + assert!( + event::insert_event_in_transaction(&mut cancelled_tx, &cancelled_event, Some(channel)) + .await + .expect("insert while TTL lock held") + .1 + ); + let cancel_error = timeout(Duration::from_secs(5), cancelled_tx.commit()) + .await + .expect("commit must not hang on TTL lock") + .expect_err("a cancelled TTL refresh must reject the event"); + assert_sqlstate(&cancel_error, "57014"); + assert_eq!(event_count(pool, community, &cancelled_event).await, 0); + assert_eq!(match_count(pool, community, &cancelled_event).await, 0); + assert_eq!( + deadline(pool, community, channel).await, + before_lock_timeout + ); + lock_holder + .rollback() + .await + .expect("release exclusive TTL lock"); + assert_eq!(event_count(pool, community, &locked_event).await, 1); + assert_eq!(match_count(pool, community, &locked_event).await, 1); + + assert_entry_points(&db, community, channel, &keys).await; +} + +/// Every production writer that can carry a channel runs both follow-ups, +/// not just `insert_event`. In the app-only arm this is what catches a +/// writer that skips the hook. +async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) { + let pool = db.pool(); + + // Thread-metadata writer (relay ingest, reactions, serving-lease writes). + let before = reset_deadline(pool, community, channel).await; + let threaded = signed_event(keys, 9, "thread-metadata-writer"); + assert!( + db.insert_event_with_thread_metadata(community, &threaded, Some(channel), None) + .await + .expect("insert through the thread-metadata writer") + .1 + ); + assert_eq!(match_count(pool, community, &threaded).await, 1); + assert!(deadline(pool, community, channel).await > before); + + // Replaceable writer (NIP-16 and NIP-29 discovery state). + let before = reset_deadline(pool, community, channel).await; + let replaceable = signed_event(keys, 39000, "replaceable-writer"); + assert!( + db.replace_addressable_event(community, &replaceable, Some(channel)) + .await + .expect("insert through the replaceable writer") + .1 + ); + assert_eq!(match_count(pool, community, &replaceable).await, 0); + assert!(deadline(pool, community, channel).await > before); + + // Parameterized writer: the insert runs inside a savepoint, and the + // TTL refresh must follow only an insert that survives it. + let d_tag = format!("follow-up-{}", Uuid::new_v4().simple()); + let parameterized = EventBuilder::new(Kind::Custom(30023), "parameterized-writer") + .tags([Tag::identifier(d_tag.clone())]) + .sign_with_keys(keys) + .expect("sign parameterized event"); + let before = reset_deadline(pool, community, channel).await; + let mut tx = begin_caller_owned_event_transaction(db, community).await; + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + ¶meterized, + &d_tag, + Some(channel), + ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("insert through the parameterized writer"); + assert_eq!(result.status, ParameterizedReplaceStatus::Inserted); + tx.commit().await.expect("commit parameterized insert"); + assert!(deadline(pool, community, channel).await > before); + + let before = reset_deadline(pool, community, channel).await; + let mut tx = begin_caller_owned_event_transaction(db, community).await; + let replay = db + .replace_parameterized_event_in_transaction( + &mut tx, + ¶meterized, + &d_tag, + Some(channel), + ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("replay through the parameterized writer"); + assert_eq!(replay.status, ParameterizedReplaceStatus::Duplicate); + tx.commit().await.expect("commit parameterized replay"); + assert_eq!( + deadline(pool, community, channel).await, + before, + "a replay rolled back to its savepoint must not refresh the channel TTL" + ); + + // Member-snapshot writer (kind 39002 on the roster-locked transaction). + let relay_keys = Keys::generate(); + let before = reset_deadline(pool, community, channel).await; + let mut snapshot = db + .lock_member_snapshot(community, channel, &relay_keys.public_key().to_bytes()) + .await + .expect("lock member snapshot"); + let roster = EventBuilder::new(Kind::Custom(39002), "") + .tags([Tag::identifier(channel.to_string())]) + .sign_with_keys(&relay_keys) + .expect("sign member snapshot"); + assert!( + snapshot + .replace_member_event(&roster) + .await + .expect("replace member snapshot") + .1 + ); + snapshot.release().await.expect("commit member snapshot"); + assert!(deadline(pool, community, channel).await > before); + + // One transaction, two channels, one of them twice: each channel is + // refreshed once, at commit. + let second = create_channel(pool, community, &keys.public_key().to_bytes()).await; + let before_first = reset_deadline(pool, community, channel).await; + let before_second = reset_deadline(pool, community, second).await; + let mut tx = begin_caller_owned_event_transaction(db, community).await; + for (target, content) in [ + (channel, "multi-a"), + (second, "multi-b"), + (channel, "multi-c"), + ] { + let message = signed_event(keys, 7, content); + assert!( + event::insert_event_in_transaction(&mut tx, &message, Some(target)) + .await + .expect("insert multi-channel event") + .1 + ); + } + let database_time_before_commit: DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(tx.conn()) + .await + .expect("sample database clock before multi-channel commit"); + tx.commit().await.expect("commit multi-channel transaction"); + for (target, before) in [(channel, before_first), (second, before_second)] { + let after = deadline(pool, community, target).await; + assert!(after > before); + assert!(after >= database_time_before_commit + chrono::Duration::seconds(60)); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_event_follow_up_contract_dual() { + assert_contract(false, Arm::Dual).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_event_follow_up_contract_app_only() { + assert_contract(false, Arm::AppOnly).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_event_follow_up_contract_dual() { + assert_contract(true, Arm::Dual).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_event_follow_up_contract_app_only() { + assert_contract(true, Arm::AppOnly).await; +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 8e43f9af3e..91c4a6e282 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -20,6 +20,12 @@ pub mod deletion; pub mod dm; /// Event storage and retrieval. pub mod event; +/// App-owned push enqueue and channel TTL refresh for inserted events. +pub(crate) mod event_follow_up; + +#[cfg(test)] +mod event_follow_up_postgres_tests; + /// Home feed queries. pub mod feed; /// Git repository name registry (NIP-34 kind:30617). diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 06f5cc9fd5..f8a5ce15ee 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -29,20 +29,27 @@ async fn begin_operation_transaction( Ok(sqlx::Transaction::begin(connection, None).await?) } -/// Namespace for the per-community push-gate advisory lock. Must match the -/// key built inside the `enqueue_push_match_job` trigger (migration 0023): -/// event inserts take it SHARED there; every lease transition that can make -/// match eligibility true takes it EXCLUSIVE here, forcing a total order so -/// a concurrent event insert either sees the committed lease or strictly -/// precedes the activation (in which case no wake was owed). Distinct key -/// domain from the audit lock and the lease address/author locks. +/// Namespace for the per-community push-gate advisory lock. Event inserts +/// take it SHARED in `event_follow_up::enqueue_push_match` (and in the +/// `enqueue_push_match_job` trigger from migration 0023 until it is retired); +/// every lease transition that can make match eligibility true takes it +/// EXCLUSIVE here, forcing a total order so a concurrent event insert either +/// sees the committed lease or strictly precedes the activation (in which case +/// no wake was owed). Distinct key domain from the audit lock and the lease +/// address/author locks. const PUSH_GATE_LOCK_NAMESPACE: &str = "buzz_push_gate:"; +/// The push-gate advisory lock key for `community`. Both sides of the lock +/// protocol build it here, so the SHARED and EXCLUSIVE keys cannot drift. +pub(crate) fn push_gate_lock_key(community: CommunityId) -> String { + format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid()) +} + async fn acquire_push_gate_lock(tx: &mut sqlx::PgConnection, community: CommunityId) -> Result<()> { crate::observability::observe_advisory_lock( crate::observability::LockType::PushGate, sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") - .bind(format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid())) + .bind(push_gate_lock_key(community)) .execute(&mut *tx), ) .await?; @@ -246,7 +253,8 @@ pub async fn accept_lease_event( ) .await?; // T1b: an activation can flip the community from "no eligible lease" to - // "eligible", so it must serialize against the trigger's shared gate lock. + // "eligible", so it must serialize against the event producer's shared + // gate lock. // Acquired after the address/author locks to keep one global lock order. if active.is_some() { acquire_push_gate_lock(tx.conn(), community).await?; @@ -361,6 +369,14 @@ pub async fn accept_lease_event( } return Err(error.into()); } + // The row above is always kind 30350 with no channel. + crate::store::event_follow_up::after_admitted_insert( + &mut tx, + event.id.as_bytes().as_slice(), + 30350, + None, + ) + .await?; let (is_active, app_profile, endpoint_hash, endpoint_grant, max_class, subscriptions) = active .map_or((false, None, None, None, None, None), |active| { @@ -475,8 +491,8 @@ async fn replace_lease( }; // T1b: an activating replacement can flip the community from "no eligible - // lease" to "eligible"; serialize it against the trigger's shared gate - // lock (gate → lease row, matching accept_lease_event's global order). + // lease" to "eligible"; serialize it against the event producer's shared + // gate lock (gate → lease row, matching accept_lease_event's global order). // Revocations (is_active = false) never make eligibility true and skip it. let mut tx = begin_operation_transaction(pool, crate::observability::WriterOperation::EventWrite) diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 9a13e5aa47..8db2082014 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1279,6 +1279,13 @@ impl Db { let was_inserted = insert_result.rows_affected() > 0; if was_inserted { + crate::store::event_follow_up::after_admitted_insert( + &mut tx, + event.id.as_bytes().as_slice(), + kind_i32, + None, + ) + .await?; tx.commit().await?; } else { tx.rollback().await?; diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 1fb53c50a0..200d44cb65 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -333,6 +333,15 @@ async fn replace_parameterized_event_in_transaction_impl( if insert_result.rows_affected() == 0 { return Ok(false); } + // Inside the savepoint, so a later failure here also drops the job. + // The TTL refresh is recorded only once the savepoint is released. + crate::store::event_follow_up::enqueue_push_match( + tx.conn(), + community_id, + incoming_id, + kind_i32, + ) + .await?; if is_nip_rs { sqlx::query( @@ -380,6 +389,9 @@ async fn replace_parameterized_event_in_transaction_impl( sqlx::query("RELEASE SAVEPOINT parameterized_replace") .execute(tx.conn()) .await?; + if status == ParameterizedReplaceStatus::Inserted { + tx.record_channel_event(channel_id, kind_i32); + } Ok(ParameterizedReplaceResult::new( event, @@ -535,6 +547,14 @@ impl Db { )); } + crate::store::event_follow_up::after_admitted_insert( + &mut tx, + event.id.as_bytes().as_slice(), + kind_i32, + channel_id, + ) + .await?; + // The replaceable event and its denormalized mention index are one // authoritative discovery write. An indexing error must roll back the // new event and restore the previously-live event. diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index dbe5d2d149..1608779952 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -950,19 +950,28 @@ fn fn_signature_starts_here(trimmed_line: &str) -> bool { /// signature to the doc comment or attributes of the next function, so a /// following function's docs (which may quote SQL) are never attributed to the /// function before it. +/// +/// Signatures and attributes are found in the masked view (see +/// [`mask_comments_and_literals`]), so an `fn` line inside a comment or a +/// multiline string never splits a function. Doc comments are blank in that +/// view, so `///` lead-ins are read from the original line. fn function_slices(production_source: &str) -> Vec<&str> { + let masked = mask_comments_and_literals(production_source); // (signature offset, offset where the next function's lead-in begins) let mut starts = Vec::new(); let mut lead_ins = Vec::new(); let mut lead_in: Option = None; let mut offset = 0usize; - for line in production_source.split_inclusive('\n') { - let trimmed = line.trim_start(); - let indent = line.len() - trimmed.len(); + for (line, code) in production_source + .split_inclusive('\n') + .zip(masked.split_inclusive('\n')) + { + let trimmed = code.trim_start(); + let indent = code.len() - trimmed.len(); if fn_signature_starts_here(trimmed) { starts.push(offset + indent); lead_ins.push(lead_in.take().unwrap_or(offset + indent)); - } else if trimmed.starts_with("///") || trimmed.starts_with("#[") { + } else if line.trim_start().starts_with("///") || trimmed.starts_with("#[") { lead_in.get_or_insert(offset); } else { lead_in = None; @@ -1130,7 +1139,7 @@ fn serving_table_policy_requires_admitted_tx_not_raw_transactions() { let source = r#" pub(crate) async fn raw_transaction_writer(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") - .execute(&mut **tx) + .execute(tx.conn()) .await .expect("write"); } @@ -1143,7 +1152,7 @@ pub(crate) async fn raw_connection_writer(conn: &mut sqlx::PgConnection) { pub(crate) async fn admitted_writer(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") .bind(tx.community().as_uuid()) - .execute(&mut **tx) + .execute(tx.conn()) .await .expect("write"); } @@ -1161,76 +1170,80 @@ pub(crate) async fn admitted_writer(tx: &mut AdmittedTx) { ); } -#[test] -fn admitted_tx_is_constructed_only_by_admitting_constructors() { - // The fields are private to `runtime/admitted_tx.rs`, so only that file can - // build the value. Pin that each function there that builds it also admits - // the transaction it wraps, so a new unguarded constructor cannot slip in. - let source = include_str!("../src/runtime/admitted_tx.rs"); - let production: String = strip_cfg_test_items(source) +/// Every function in `source` that builds an `AdmittedTx`, paired with whether +/// it admits the transaction it wraps. The scan deliberately does not strip +/// `#[cfg(test)]` items: a test-only forge would let crate tests run follow-ups +/// on an unadmitted transaction, so it must admit like any other constructor. +fn admitted_tx_constructors(source: &str) -> Vec<(String, bool)> { + let production: String = source .lines() .filter(|line| !line.trim_start().starts_with("//")) // `impl … for AdmittedTx {` headers open a block, not a value. .filter(|line| !line.trim_start().starts_with("impl")) .map(|line| format!("{line}\n")) .collect(); - let constructors: Vec<&str> = function_slices(&production) + function_slices(&production) .into_iter() .filter(|function| function.contains("Self {") || function.contains("AdmittedTx {")) - .collect(); + .map(|function| { + let admits = function.contains(".guard_transaction(&mut tx, community)") + || function.contains(".guard_transaction_with_serving_lease(&mut tx, lease)"); + (function.to_string(), admits) + }) + .collect() +} + +#[test] +fn admitted_tx_is_constructed_only_by_admitting_constructors() { + // The fields are private to `runtime/admitted_tx.rs`, so only that file can + // build the value. Pin that each function there that builds it also admits + // the transaction it wraps, so a new unguarded constructor cannot slip in. + let constructors = admitted_tx_constructors(include_str!("../src/runtime/admitted_tx.rs")); assert_eq!( constructors.len(), 2, "AdmittedTx must have exactly the two admitting constructors" ); - for constructor in constructors { + for (constructor, admits) in constructors { assert!( - constructor.contains(".guard_transaction(&mut tx, community)") - || constructor.contains(".guard_transaction_with_serving_lease(&mut tx, lease)"), + admits, "AdmittedTx constructor must admit the transaction it wraps: {constructor}" ); } } -/// Remove each top-level `#[cfg(test)]` item and keep the production code -/// around it. Truncating at the first `#[cfg(test)]` would hide production -/// functions that follow a test-only item. Tracks brace depth: the item ends -/// on the first line that leaves depth at or below zero and ends with `;` or -/// `}`, ignoring a trailing `//` comment. A column-0 `}` line (rustfmt's -/// top-level close) always ends it. Braces inside string or char literals are -/// still counted, so a test-only item with unbalanced literal braces (for -/// example `"{"`) runs on to the next column-0 `}`; no such item exists today. -fn strip_cfg_test_items(source: &str) -> String { - let mut kept = String::with_capacity(source.len()); - let mut lines = source.lines(); - while let Some(line) = lines.next() { - if line != "#[cfg(test)]" { - kept.push_str(line); - kept.push('\n'); - continue; - } - let mut depth = 0_isize; - for line in lines.by_ref().skip_while(|line| line.starts_with("#[")) { - depth += line.matches('{').count() as isize - line.matches('}').count() as isize; - // Any `//` may start the trailing comment (an earlier one can sit - // inside a string such as `"https://…"`), so try every prefix. A - // false match inside a string only ends the item early, which - // scans more lines and can never hide production code. - let item_end = depth <= 0 - && line - .match_indices("//") - .map(|(at, _)| &line[..at]) - .chain([line]) - .any(|text| { - let text = text.trim_end(); - text.ends_with(';') || text.ends_with('}') - }); - if item_end || line == "}" || line == "};" { - break; - } - } +#[test] +fn admitted_tx_constructor_scan_reports_test_only_forges() { + let source = r#" +impl AdmittedTx { + pub(crate) async fn admit(mut tx: Tx, community: CommunityId) -> Result { + guard.guard_transaction(&mut tx, community).await?; + Ok(Self { tx, community }) } - kept + + #[cfg(test)] + pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self { + Self { tx, community } + } +} +"#; + let unadmitted: Vec = admitted_tx_constructors(source) + .into_iter() + .filter(|(_, admits)| !admits) + .map(|(function, _)| { + function + .lines() + .next() + .unwrap_or_default() + .trim() + .to_string() + }) + .collect(); + assert_eq!( + unadmitted, + ["pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self {"], + "an indented `#[cfg(test)]` forge must still be reported as unadmitted" + ); } #[test] @@ -1405,3 +1418,519 @@ fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() { "guarded-table scan must exercise at least one file that writes a fenced table" ); } + +// Rolled-back verification probes write `events` rows that never commit, so +// they owe no push job or TTL refresh. +const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 2] = [ + "pub async fn verify_floor_guard_behavior(", + "pub async fn verify_channel_roster_fence_behavior(", +]; + +/// `source` with every `#[cfg(test)]` item removed, wherever it sits in the +/// file. Each item runs to the `;` that ends it or to the brace that closes +/// its body. Braces inside comments and literals are not counted (see +/// [`mask_comments_and_literals`]). Unlike cutting the file at the first +/// `#[cfg(test)]`, this keeps production code that follows a test-only item. +/// The whole file is masked once and the marker is found in that view, so a +/// mention in a comment or string (even a multiline one) strips nothing. The +/// marker also counts only when it is the first code on its line. +fn strip_cfg_test_items(source: &str) -> String { + const MARKER: &str = "#[cfg(test)]"; + let masked = mask_comments_and_literals(source); + let mut kept = String::with_capacity(source.len()); + let mut copied = 0usize; + let mut search = 0usize; + while let Some(found) = masked[search..].find(MARKER) { + let at = search + found; + let line_start = masked[..at].rfind('\n').map_or(0, |newline| newline + 1); + search = at + MARKER.len(); + if !masked[line_start..at].trim().is_empty() { + continue; + } + kept.push_str(&source[copied..at]); + copied = search + cfg_test_item_len(&masked[search..]); + search = copied; + } + kept.push_str(&source[copied..]); + kept +} + +/// Length of the item that starts `masked_item`, which must already be masked +/// so braces in comments and literals are not counted. +fn cfg_test_item_len(masked_item: &str) -> usize { + let mut depth = 0usize; + for (index, byte) in masked_item.bytes().enumerate() { + match byte { + b';' if depth == 0 => return index + 1, + b'{' => depth += 1, + b'}' => { + depth = depth.saturating_sub(1); + if depth == 0 { + return index + 1; + } + } + _ => {} + } + } + masked_item.len() +} + +/// Whether the `r` at `index` begins a raw literal token: `r"`, or the `br"` / +/// `cr"` raw byte and C strings, rather than ending an identifier such as `for"`. +fn raw_literal_prefix_starts_token(bytes: &[u8], index: usize) -> bool { + let is_ident = |at: Option| { + at.and_then(|at| bytes.get(at)) + .is_some_and(|byte| byte.is_ascii_alphanumeric() || *byte == b'_') + }; + let before = index.checked_sub(1); + !is_ident(before) + || (matches!(before.map(|at| bytes[at]), Some(b'b' | b'c')) + && !is_ident(index.checked_sub(2))) +} + +/// `source` with every comment (`//`, `///`, nested `/* */`) and every string, +/// raw string, and char literal replaced by spaces, byte for byte. Newlines are +/// kept, so offsets and line structure match `source`. Scans that look for +/// code (braces, calls) read the masked text; scans that look for SQL read +/// the original, because the SQL lives inside string literals. +fn mask_comments_and_literals(source: &str) -> String { + let bytes = source.as_bytes(); + let mut masked = bytes.to_vec(); + let mut blank = |from: usize, to: usize| { + for byte in &mut masked[from..to.min(bytes.len())] { + if *byte != b'\n' { + *byte = b' '; + } + } + }; + let mut index = 0usize; + while index < bytes.len() { + let start = index; + match bytes[index] { + b'/' if bytes.get(index + 1) == Some(&b'/') => { + while index < bytes.len() && bytes[index] != b'\n' { + index += 1; + } + } + b'/' if bytes.get(index + 1) == Some(&b'*') => { + let mut depth = 0usize; + while index < bytes.len() { + if bytes[index] == b'/' && bytes.get(index + 1) == Some(&b'*') { + depth += 1; + index += 2; + } else if bytes[index] == b'*' && bytes.get(index + 1) == Some(&b'/') { + depth -= 1; + index += 2; + if depth == 0 { + break; + } + } else { + index += 1; + } + } + } + b'r' if matches!(bytes.get(index + 1), Some(b'"' | b'#')) + && raw_literal_prefix_starts_token(bytes, index) + && bytes.get( + index + + 1 + + bytes[index + 1..] + .iter() + .take_while(|byte| **byte == b'#') + .count(), + ) == Some(&b'"') => + { + let hashes = bytes[index + 1..] + .iter() + .take_while(|byte| **byte == b'#') + .count(); + let close = format!("\"{}", "#".repeat(hashes)); + let body = index + 2 + hashes; + index = source[body..] + .find(&close) + .map_or(bytes.len(), |offset| body + offset + close.len()); + } + b'"' => { + index += 1; + while index < bytes.len() && bytes[index] != b'"' { + index += if bytes[index] == b'\\' { 2 } else { 1 }; + } + index += 1; + } + b'\'' if char_literal_end(source, index).is_some() => { + index = char_literal_end(source, index).expect("guard matched"); + } + _ => { + index += 1; + continue; + } + } + blank(start, index); + } + String::from_utf8(masked).expect("masking replaces whole ASCII-delimited spans") +} + +/// The end (exclusive) of the char literal whose opening quote is at `start`, +/// such as `'{'`, `'é'`, `'\''`, `'\x7f'`, or `'\u{1F600}'`. `None` for a +/// lifetime, which has no closing quote. +fn char_literal_end(source: &str, start: usize) -> Option { + // `start` holds an ASCII quote, so `start + 1` is a char boundary. + let rest = &source[start + 1..]; + let body = if let Some(escape) = rest.strip_prefix('\\') { + // The longest escape is `\u{10FFFF}`: a backslash and 9 more bytes. + // Skip the escaped byte so `'\''` does not close on its own quote. + let close = escape.get(1..)?.find('\'')? + 1; + (close <= 9).then_some(1 + close)? + } else { + rest.chars().next()?.len_utf8() + }; + rest[body..] + .starts_with('\'') + .then_some(start + 1 + body + 1) +} + +#[test] +fn masker_blanks_every_literal_form_and_keeps_code() { + for (input, expected) in [ + // Raw strings with zero, one, or several `#`, and the `b`/`c` raw + // prefixes. Each body ends in `\`, which would escape the closing quote + // if the literal were scanned as a plain string. + (r#"a(r"\") {}"#, "a( ) {}"), + (r###"a(r##"x"#{"##) {}"###, "a( ) {}"), + (r##"a(br#"\"#) {}"##, "a(b ) {}"), + (r#"a(br"\") {}"#, "a(b ) {}"), + (r#"a(cr"\") {}"#, "a(c ) {}"), + // Plain byte and C strings. + (r#"a(b"{") {}"#, "a(b ) {}"), + (r#"a(c"{") {}"#, "a(c ) {}"), + // Plain and escaped char literals. + ( + r"a('\'', '\\', '\n', '{') {}", + "a( , , , ) {}", + ), + // Multi-byte and long-escape char literals. Each holds a brace, or + // hides one from a scan that misreads the literal's length. + ("a('é', '{') {}", "a( , ) {}"), + (r"a('\u{1F600}', '{') {}", "a( , ) {}"), + (r"a('\x7f', '{') {}", "a( , ) {}"), + // Code that only looks like a literal stays as it is. + ("let r#type = 1; {}", "let r#type = 1; {}"), + ("fn f<'a>(x: &'a str) {}", "fn f<'a>(x: &'a str) {}"), + ] { + assert_eq!( + mask_comments_and_literals(input), + expected, + "masking `{input}`" + ); + } +} + +/// Whether `source` inserts into `events` itself (not `event_mentions` or any +/// other `events_*` table), in any letter case, across line breaks and string +/// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.` +/// schema prefix. +fn inserts_events_row(source: &str) -> bool { + let mut normalized = String::with_capacity(source.len()); + let mut chars = source.chars().peekable(); + while let Some(ch) = chars.next() { + if ch.is_whitespace() || ch == '\\' { + if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) { + chars.next(); + } + if !normalized.ends_with(' ') { + normalized.push(' '); + } + } else { + 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 == '_')) + }) + }) +} + +/// An events insert must run both follow-ups: the push enqueue and the +/// pre-commit TTL refresh. `after_admitted_insert` does both; a writer that +/// inserts inside a savepoint calls the two halves separately. +/// +/// Only real calls count: a hook in a comment or a string literal does not. +fn runs_event_follow_ups(function_source: &str) -> bool { + let code = mask_comments_and_literals(function_source); + code.contains("event_follow_up::after_admitted_insert(") + || (code.contains("event_follow_up::enqueue_push_match(") + && code.contains(".record_channel_event(")) +} + +fn event_insert_follow_up_violations(production_source: &str) -> Vec { + function_slices(production_source) + .into_iter() + .filter(|function_source| { + let header = function_header(function_source); + inserts_events_row(function_source) + && !EVENT_INSERT_FOLLOW_UP_EXCEPTIONS + .iter() + .any(|exception| header.starts_with(exception)) + && !runs_event_follow_ups(function_source) + }) + .map(|function_source| function_header(function_source).to_owned()) + .collect() +} + +#[test] +fn event_insert_follow_up_policy_rejects_writers_without_the_hook() { + let source = r##" +pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events \ + (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind) + .await + .expect("enqueue"); +} +pub(crate) async fn hooked_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) + .await + .expect("follow up"); +} +pub(crate) async fn savepoint_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind) + .await + .expect("enqueue"); + tx.record_channel_event(channel, kind); +} +pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub async fn verify_floor_guard_behavior(pool: &PgPool) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(pool) + .await + .expect("probe"); +} +pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) { + sqlx::query("insert into + public.events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub(crate) async fn continued_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO \ + events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +/// Kept at module scope, not `#[cfg(test)]`, because production calls it. +pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +// see `#[cfg(test)] mod x {` +pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO\nevents (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + // crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) +} +pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + /* crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind) + /* nested */ tx.record_channel_event(channel, kind); */ +} +pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + tracing::debug!("skipped event_follow_up::after_admitted_insert(tx, ..)"); +} +pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + /* /* inner */ crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) */ +} +pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + tracing::debug!("\"event_follow_up::after_admitted_insert(\""); +} +pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); + tracing::debug!(r#""event_follow_up::after_admitted_insert(""#); +} +/* +#[cfg(test)] +*/ +pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) { + let _example = r#" +fn example() { + crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) +} +"#; + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +#[cfg(test)] +fn test_only_helper() -> (&'static str, char, &'static str, char, &'static [u8]) { + // an unbalanced { in a comment + ("{", '{', r#"}"} {"#, '"', br#"\"#) +} +pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()) + .await + .expect("write"); +} +#[cfg(test)] +mod tests { + async fn test_writer(pool: &PgPool) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(pool) + .await + .expect("fixture"); + } +} +"##; + + assert_eq!( + event_insert_follow_up_violations(&strip_cfg_test_items(source)), + [ + "pub(crate) async fn unhooked_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) {", + "pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) {", + "pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) {", + ], + "an events insert must run the push enqueue and record the TTL refresh" + ); +} + +#[test] +fn every_production_event_insert_runs_the_follow_up_hook() { + use std::path::{Path, PathBuf}; + + fn collect_rs_files(dir: &Path, out: &mut Vec) { + for entry in std::fs::read_dir(dir).expect("read source directory") { + let path = entry.expect("read directory entry").path(); + if path.is_dir() { + collect_rs_files(&path, out); + } else if should_scan_guarded_write_source_file(&path) { + out.push(path); + } + } + } + + let crates_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("crate lives under crates/"); + let mut files = Vec::new(); + for entry in std::fs::read_dir(crates_root).expect("read crates directory") { + let src = entry.expect("read crates entry").path().join("src"); + if src.is_dir() { + collect_rs_files(&src, &mut files); + } + } + + let mut hooked_writers = 0usize; + let mut violations = Vec::new(); + for path in files { + let source = std::fs::read_to_string(&path).expect("read source file"); + let production = strip_cfg_test_items(&source); + if !inserts_events_row(&production) { + continue; + } + for function_source in function_slices(&production) { + if inserts_events_row(function_source) && runs_event_follow_ups(function_source) { + hooked_writers += 1; + } + } + violations.extend( + event_insert_follow_up_violations(&production) + .into_iter() + .map(|header| format!("{}: {header}", path.display())), + ); + } + + assert!( + violations.is_empty(), + "production `INSERT INTO events` writers must call `event_follow_up` so the push \ + enqueue and channel TTL refresh survive the trigger retirement: {violations:?}" + ); + // The seven writers at the time of BUZZ-176. A lower count means the scan + // stopped seeing a writer, not that one was removed safely. + assert!( + hooked_writers >= 7, + "expected at least seven hooked event writers, found {hooked_writers}" + ); +}