diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 0e023340a3..e63a1143d9 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -185,6 +185,10 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(conn: &mut PgConnection) -> Ok(()) } +#[cfg(test)] +#[path = "push_retirement_tests.rs"] +mod retirement_postgres_tests; + #[cfg(test)] mod postgres_tests { use super::*; @@ -705,7 +709,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 62); + assert_eq!(migrations.len(), 63); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); @@ -1674,7 +1678,8 @@ mod postgres_tests { assert!(!sql.contains("NEW.kind IN (7, 9, 1059, 40007, 46010)")); let desired_schema = crate::test_support::desired_state_schema_sql(); - assert!(desired_schema.contains("NEW.kind IN (9, 40002, 45001, 45003)")); + assert!(!desired_schema.contains("enqueue_push_match_job")); + assert!(!desired_schema.contains("events_enqueue_push_match")); assert!(!desired_schema.contains("NEW.kind IN (7, 9, 1059, 40007, 46010)")); } @@ -2539,7 +2544,7 @@ mod postgres_tests { .expect("drop probe database"); } - async fn connect_test_pool() -> PgPool { + pub(super) async fn connect_test_pool() -> PgPool { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| TEST_DB_URL.to_owned()); @@ -2549,7 +2554,7 @@ mod postgres_tests { .expect("connect to test DB") } - async fn reset_public_schema(pool: &PgPool) { + pub(super) async fn reset_public_schema(pool: &PgPool) { sqlx::query("DROP SCHEMA IF EXISTS public CASCADE") .execute(pool) .await diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index 40b1dcb52f..ac9c6e2f5d 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -64,7 +64,7 @@ impl PushEnqueue { record_drop("shutdown"); break; }, - result = tokio::time::timeout(ENQUEUE_TIMEOUT, enqueue(&pool, &job)) => result, + result = tokio::time::timeout(ENQUEUE_TIMEOUT, enqueue(&pool, &job, #[cfg(test)] None)) => result, }; match result { Ok(Ok(())) => { @@ -145,7 +145,37 @@ pub(crate) fn record_drop(reason: &'static str) { metrics::counter!("buzz_push_enqueue_total", "result" => reason).increment(1); } -async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> { +#[cfg(test)] +type GateBarrier = ( + tokio::sync::oneshot::Sender, + tokio::sync::oneshot::Receiver<()>, +); + +/// Pause the real enqueue transaction after it acquires the production gate. +#[cfg(test)] +pub(crate) async fn enqueue_with_gate_barrier( + pool: &PgPool, + community: CommunityId, + event_id: Vec, + barrier: GateBarrier, +) -> crate::Result<()> { + enqueue( + pool, + &Job { + community, + event_id, + barrier: None, + }, + Some(barrier), + ) + .await +} + +async fn enqueue( + pool: &PgPool, + job: &Job, + #[cfg(test)] gate_barrier: Option, +) -> crate::Result<()> { let mut tx = pool.begin().await?; // Server-side deadlines also bound work if the client disconnects or the // task is cancelled. These apply only to the dedicated push connection. @@ -158,6 +188,17 @@ async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> { .bind(crate::push::push_gate_lock_key(job.community)) .execute(&mut *tx) .await?; + #[cfg(test)] + if let Some((held, release)) = gate_barrier { + let pid = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *tx) + .await?; + held.send(pid) + .expect("race test observes enqueue transaction"); + release + .await + .expect("race test releases enqueue transaction"); + } // Compare recorded receipt/update times without imposing commit ordering. // Registration overlapping message arrival may send or suppress a wake. // This bounded job never scans historical messages on lease activation. diff --git a/crates/buzz-db/src/runtime/push_retirement_tests.rs b/crates/buzz-db/src/runtime/push_retirement_tests.rs new file mode 100644 index 0000000000..a8601dbe26 --- /dev/null +++ b/crates/buzz-db/src/runtime/push_retirement_tests.rs @@ -0,0 +1,115 @@ +//! Upgrade contract for push-only trigger retirement. +use super::postgres_tests::{connect_test_pool, reset_public_schema}; +use super::{run_migrations, run_migrations_through}; +use sqlx::PgPool; + +async fn ttl_definitions(pool: &PgPool) -> Vec<(String, String)> { + sqlx::query_as( + "SELECT tgrelid::regclass::text, pg_get_triggerdef(oid) FROM pg_trigger \ + WHERE tgname = 'events_refresh_channel_ttl' ORDER BY 1", + ) + .fetch_all(pool) + .await + .expect("read TTL triggers") +} + +async fn push_trigger_count(pool: &PgPool) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM pg_trigger WHERE tgname = 'events_enqueue_push_match'") + .fetch_one(pool) + .await + .expect("count push triggers") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_push_retirement_preserves_ttl_and_published_checksums() { + let pool = connect_test_pool().await; + reset_public_schema(&pool).await; + run_migrations_through(&pool, 62) + .await + .expect("apply published migrations"); + assert!( + push_trigger_count(&pool).await > 1, + "parent and partition triggers exist before upgrade" + ); + let ttl_before = ttl_definitions(&pool).await; + assert!(ttl_before.len() > 1); + let function_before: String = sqlx::query_scalar( + "SELECT pg_get_functiondef('refresh_channel_ttl_after_event_insert()'::regprocedure)", + ) + .fetch_one(&pool) + .await + .unwrap(); + let checksums_before: Vec<(i64, Vec)> = sqlx::query_as( + "SELECT version, checksum FROM _sqlx_migrations WHERE version <= 62 ORDER BY version", + ) + .fetch_all(&pool) + .await + .unwrap(); + + run_migrations(&pool).await.expect("retire push trigger"); + assert_eq!( + push_trigger_count(&pool).await, + 0, + "retired on parent and every partition" + ); + let function_absent: bool = + sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL") + .fetch_one(&pool) + .await + .unwrap(); + assert!(function_absent); + assert_eq!(ttl_definitions(&pool).await, ttl_before); + let function_after: String = sqlx::query_scalar( + "SELECT pg_get_functiondef('refresh_channel_ttl_after_event_insert()'::regprocedure)", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(function_after, function_before); + let checksums_after: Vec<(i64, Vec)> = sqlx::query_as( + "SELECT version, checksum FROM _sqlx_migrations WHERE version <= 62 ORDER BY version", + ) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(checksums_after, checksums_before); + + // A raw writer has no application TTL follow-up. The preserved deferred + // trigger must still refresh its channel when the event commits. + let community = uuid::Uuid::new_v4(); + let channel = uuid::Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("retirement-{community}.example")) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO channels (id, community_id, name, created_by, ttl_seconds, ttl_deadline) \ + VALUES ($1, $2, 'ttl', $3, 60, clock_timestamp() - interval '1 day')", + ) + .bind(channel) + .bind(community) + .bind([11_u8; 32]) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id) \ + VALUES ($1, $2, $3, now(), 9, '[]', 'TTL survives push retirement', $4, $5)") + .bind(community).bind([12_u8; 32]).bind([11_u8; 32]).bind([13_u8; 64]).bind(channel) + .execute(&pool).await.unwrap(); + let refreshed: bool = sqlx::query_scalar( + "SELECT ttl_deadline > clock_timestamp() FROM channels WHERE id=$1 AND community_id=$2", + ) + .bind(channel) + .bind(community) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + refreshed, + "raw inserts still refresh TTL after push retirement" + ); + pool.close().await; +} diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index 5219af6062..be0d55902c 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -13,8 +13,8 @@ //! //! This is the application-side replacement for the `events_enqueue_push_match` //! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040). -//! Updated writer connections suppress the surviving push trigger. Legacy -//! writers retain transactional enqueue until replaced. Every production +//! Migration 0063 retires the push trigger; the TTL trigger remains installed. +//! Push requires application enqueueing on every supported writer. Every production //! `INSERT INTO events` must call this module, enforced by source-policy tests. use std::collections::BTreeSet; 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 index b32b8bfbd4..ae47ff2e14 100644 --- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -1,6 +1,6 @@ //! 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). +//! with historical overlap, push retired, and both follow-ups app-only. use crate::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus}; use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError}; use buzz_core::CommunityId; @@ -214,14 +214,13 @@ async fn activate_lease( ); } -/// 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. +/// Preserve the overlap and fully app-only compatibility contracts alongside +/// the shipped push-only retirement (the TTL trigger remains installed). #[derive(Clone, Copy, Debug)] enum Arm { Dual, AppOnly, + PushRetired, } async fn apply_arm(pool: &PgPool, arm: Arm) { @@ -233,32 +232,43 @@ async fn apply_arm(pool: &PgPool, arm: Arm) { .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" - ); - sqlx::query("DROP FUNCTION enqueue_push_match_job()") + assert_eq!( + triggers, 1, + "only the TTL trigger remains in the shipped schema" + ); + let absent: bool = + sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL") + .fetch_one(pool) + .await + .unwrap(); + assert!(absent); + match arm { + Arm::Dual => { + // Reconstruct the historical overlap explicitly in this test only. + sqlx::raw_sql(include_str!( + "../../../../migrations/0040_push_message_kinds.sql" + )) .execute(pool) .await .unwrap(); + sqlx::raw_sql(include_str!( + "../../../../migrations/0059_push_rollback_gate.sql" + )) + .execute(pool) + .await + .unwrap(); + sqlx::query("CREATE TRIGGER events_enqueue_push_match AFTER INSERT ON events FOR EACH ROW EXECUTE FUNCTION enqueue_push_match_job()") + .execute(pool).await.unwrap(); + } + Arm::AppOnly => { + sqlx::query("DROP TRIGGER events_refresh_channel_ttl ON events") + .execute(pool) + .await + .unwrap(); + } + Arm::PushRetired => {} + } + if !matches!(arm, Arm::Dual) { sqlx::raw_sql(include_str!( "../../../../migrations/0059_push_rollback_gate.sql" )) @@ -272,7 +282,7 @@ async fn apply_arm(pool: &PgPool, arm: Arm) { .unwrap(); assert!( absent, - "rollback migration must not resurrect a retired function" + "rollback migration must not resurrect the retired function" ); } } @@ -814,6 +824,18 @@ async fn migration_schema_event_follow_up_contract_app_only() { assert_contract(true, Arm::AppOnly).await; } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_event_follow_up_contract_push_retired() { + assert_contract(false, Arm::PushRetired).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_event_follow_up_contract_push_retired() { + assert_contract(true, Arm::PushRetired).await; +} + // Runs with both schema sources and with/without the overlap trigger. async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) { let pool = db.pool(); diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 4269c3f7e6..26eaee65e4 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -1558,6 +1558,30 @@ mod postgres_tests { pool } + async fn insert_push_event( + community: CommunityId, + event: &nostr::Event, + channel: Option, + ) -> crate::Result<(buzz_core::StoredEvent, bool)> { + let db = crate::Db::new(&crate::DbConfig { + database_url: std::env::var("BUZZ_TEST_DATABASE_URL").expect("isolated test database"), + push_enabled: true, + ..crate::DbConfig::default() + }) + .await + .expect("connect application writer"); + let result = db.insert_event(community, event, channel).await; + db.push_enqueue + .as_ref() + .expect("enabled producer") + .flush() + .await; + db.cancel_push_enqueue(); + db.join_push_enqueue().await; + db.pool().close().await; + result + } + fn lease_event(keys: &nostr::Keys, installation: &str, created_at: u64) -> nostr::Event { nostr::EventBuilder::new(nostr::Kind::Custom(30_350), "ciphertext") .tag(nostr::Tag::parse(["d", installation]).expect("d tag")) @@ -2204,7 +2228,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn matcher_trigger_is_allowlisted_and_deleted_events_are_discarded() { + async fn matcher_application_is_allowlisted_and_deleted_events_are_discarded() { let pool = setup_pool().await; // Global claim assertions below require a queue free of other tests' // leftovers. @@ -2223,10 +2247,10 @@ mod postgres_tests { let read_state = nostr::EventBuilder::new(nostr::Kind::Custom(30_078), "read") .sign_with_keys(&keys) .expect("sign read state"); - crate::event::insert_event(&pool, community, &push_event, None) + insert_push_event(community, &push_event, None) .await .expect("insert push event"); - crate::event::insert_event(&pool, community, &read_state, None) + insert_push_event(community, &read_state, None) .await .expect("insert non-push event"); @@ -2267,12 +2291,12 @@ mod postgres_tests { async fn matcher_load_error_preserves_claimed_job_for_recovery() { let pool = setup_pool().await; let community = make_community(&pool).await; - // Eligible lease required for the T1b-gated trigger to enqueue. + // Eligible lease required for the T1b-gated application producer to enqueue. activate(&pool, community, &[79; 32], "install", &[80; 32], 1).await; let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "retry me") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&pool, community, &event, None) + insert_push_event(community, &event, None) .await .expect("insert event"); @@ -2317,7 +2341,7 @@ mod postgres_tests { let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "size one") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&setup, community, &event, None) + insert_push_event(community, &event, None) .await .expect("insert event"); setup.close().await; @@ -2361,12 +2385,12 @@ mod postgres_tests { .await .expect("drain matcher queue"); let community = make_community(&pool).await; - // Eligible lease required for the T1b-gated trigger to enqueue. + // Eligible lease required for the T1b-gated application producer to enqueue. activate(&pool, community, &[81; 32], "install", &[82; 32], 1).await; let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "one job") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&pool, community, &event, None) + insert_push_event(community, &event, None) .await .expect("insert event"); let barrier = Arc::new(Barrier::new(8)); @@ -2397,6 +2421,12 @@ mod postgres_tests { let event_id = [23; 32]; activate(&pool, community, &author, "install", &[24; 32], 1).await; let wake_id = enqueue_one(&pool, community, &author, &event_id, 1).await; + sqlx::query("INSERT INTO push_match_queue (community_id, event_id) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(event_id) + .execute(&pool) + .await + .expect("queue a rematch independently of the delivered wake"); sqlx::query( "UPDATE push_wake_outbox SET state='delivered', created_at=now()-interval '2 days' \ WHERE community_id=$1 AND id=$2", @@ -2440,7 +2470,7 @@ mod postgres_tests { let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "poison") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&pool, community, &event, None) + insert_push_event(community, &event, None) .await .expect("insert event"); let wake_id = match enqueue_wake( @@ -2661,7 +2691,7 @@ mod postgres_tests { let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), format!("a{i}")) .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&pool, community_a, &event, None) + insert_push_event(community_a, &event, None) .await .expect("insert community-a event"); a_ids.push(event.id.as_bytes().to_vec()); @@ -2669,7 +2699,7 @@ mod postgres_tests { let b_event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "b0") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); - crate::event::insert_event(&pool, community_b, &b_event, None) + insert_push_event(community_b, &b_event, None) .await .expect("insert community-b event"); @@ -2770,37 +2800,41 @@ mod postgres_tests { ); } - // Lease activation must wait for an event insert holding the shared gate lock. + // Lease activation must wait for an enqueue transaction holding the shared gate lock. #[tokio::test] #[ignore = "requires Postgres"] - async fn gate_orders_lease_activation_after_in_flight_event() { + async fn gate_orders_lease_activation_after_in_flight_enqueue() { let pool = setup_pool().await; let community = make_community(&pool).await; - // Hold the insert transaction open after its trigger takes the shared gate lock. - let raced = nostr::EventBuilder::new(nostr::Kind::Custom(9), "raced wake") + let raced = nostr::EventBuilder::new(nostr::Kind::Custom(9), "raced enqueue") .sign_with_keys(&nostr::Keys::generate()) .expect("sign raced event"); - let mut insert_tx = pool.begin().await.expect("begin raced insert"); - sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \ - VALUES ($1, $2, $3, now(), 9, '[]', 'raced wake', $4, now())", - ) - .bind(community.as_uuid()) - .bind(raced.id.as_bytes().as_slice()) - .bind(raced.pubkey.as_bytes().as_slice()) - .bind(raced.sig.serialize().as_slice()) - .execute(&mut *insert_tx) - .await - .expect("insert raced event inside held txn"); - - let insert_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") - .fetch_one(&mut *insert_tx) + crate::event::insert_event(&pool, community, &raced, None) .await - .expect("read insert backend pid"); + .expect("store event before post-commit enqueue"); + let (held_tx, held_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let enqueue = { + let pool = pool.clone(); + tokio::spawn(async move { + crate::runtime::push_enqueue::enqueue_with_gate_barrier( + &pool, + community, + raced.id.as_bytes().to_vec(), + (held_tx, release_rx), + ) + .await + .expect("real enqueue transaction succeeds"); + }) + }; + let enqueue_pid = tokio::time::timeout(Duration::from_secs(5), held_rx) + .await + .expect("enqueue reaches its gate") + .expect("enqueue backend pid"); // A concurrent activation must block on the exclusive gate lock until - // the insert transaction resolves. + // the enqueue transaction resolves. let activation = { let pool = pool.clone(); tokio::spawn(async move { @@ -2830,7 +2864,7 @@ mod postgres_tests { "SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted \ AND $1 = ANY(pg_blocking_pids(pid))", ) - .bind(insert_pid) + .bind(enqueue_pid) .fetch_one(&pool) .await .expect("inspect advisory waiters"); @@ -2843,8 +2877,14 @@ mod postgres_tests { assert!(parked, "activation must block on the exclusive gate lock"); assert!(!activation.is_finished()); - // Committing the insert releases the shared lock so activation can proceed. - insert_tx.commit().await.expect("commit raced insert"); + // Committing the enqueue releases the shared lock so activation can proceed. + release_tx + .send(()) + .expect("release real enqueue transaction"); + tokio::time::timeout(Duration::from_secs(5), enqueue) + .await + .expect("enqueue commits") + .expect("join enqueue"); assert_eq!( tokio::time::timeout(std::time::Duration::from_secs(10), activation) .await @@ -2856,7 +2896,7 @@ mod postgres_tests { let direct = nostr::EventBuilder::new(nostr::Kind::Custom(9), "direct enqueue") .sign_with_keys(&nostr::Keys::generate()) .expect("sign direct event"); - crate::event::insert_event(&pool, community, &direct, None) + insert_push_event(community, &direct, None) .await .expect("insert post-activation event"); let queued: i64 = sqlx::query_scalar( diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 8fce29028b..331d49708f 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -212,6 +212,40 @@ triggering message. Keep fallback-to-channel and placeholder/failure cases as explicit counts in the manual sample until privacy-preserving client telemetry is designed. +### Relay push-trigger retirement gate + +Migration 0063 removes the push trigger for supported relay message paths. +Before merge, verify that the deployed primary and profiler relays use the +bounded post-commit application producer and immediate worker-stop behavior from +[#8216](https://github.com/block/buzz/pull/8216), or a verified equivalent. +Investigate any concrete supported path that creates notification-eligible +messages outside those relays. Hypothetical external SQL clients and unattributed +database sessions do not require a privileged audit or block retirement. + +Rollback disables push while retaining the qualified image and schema; image +downgrades are excluded after retirement. Pending migrations must also be +considered: migration 0062 adds system channels, which older images cannot read +or filter correctly. The push producer floor alone is not sufficient to justify +an image downgrade. + +Before separately authorized migration or deployment, bind the qualified release +and profiler image digests to the reviewed source and rollback policy. Pull +requests build images without publishing them; final published digest binding is +a release gate, not a prerequisite for merging the source change. Keep automatic +migration disabled during the transition. Resolve overlapping retirement +migrations and verify the published migration ledger before execution. Migration +0063's lock timeout bounds acquisition waits, not lock-hold or statement duration. + +After migration, verify that `events_enqueue_push_match` is absent from the +parent and every partition, `enqueue_push_match_job()` is absent, and the +channel-TTL trigger/function remain intact. Verify application enqueueing and +flag-only rollback on the exact deployed candidate. Legacy images and raw SQL +writers without the application producer are unsupported after this point: +their messages may commit without producing notifications. Do not restore the +trigger or downgrade an image as an urgent rollback step. Reintroducing legacy +writer support requires a separate reviewed forward migration and compatibility +plan; never edit published migrations or their checksums. + ### Relay rollback and reactivation Set `BUZZ_PUSH_ENABLED=false` in the approved configuration for **every writer @@ -224,14 +258,16 @@ The relay copies the flag into `DbConfig::push_enabled`. Every physical writer connection, including the audit pool and replacement connections, sets the `buzz.push_enabled` PostgreSQL session setting. Other `DbConfig` callers default to disabled. Updated writers also set `buzz.push_async_enqueue=on`, which makes -the surviving trigger skip enqueueing regardless of the enabled flag. The +any surviving pre-0063 trigger skip enqueueing regardless of the enabled flag. The application producer is created only when enabled. Disabled writes perform no push locks, lease scans or queue writes. -The overlap migration updates only an existing trigger function; it does not -recreate a retired trigger or function. Deploy that migration before relying on -the flag with a surviving trigger. Older images and raw SQL writers with a -missing setting retain legacy enqueue behavior; inventory and replace or -explicitly disable all such connections before declaring rollback complete. +Before retirement, the overlap migration updates only an existing trigger +function; it does not recreate a retired trigger or function. Deploy it before +relying on the flag with a surviving trigger. On these pre-0063 schemas, older +images and raw SQL writers with a missing setting retain legacy enqueue behavior; +inventory and replace or explicitly disable all such connections before declaring +rollback complete. After 0063, only the application producer can enqueue push +work, and the retirement gate above excludes those legacy writers and images. On SIGTERM/Ctrl-C, the relay cancels enqueue, matcher and delivery futures immediately, before its listener drain. Uncommitted transactions roll back; committed claims @@ -264,8 +300,9 @@ update, including renewal of an active lease, can also suppress a notification under this best-effort contract. Activation never scans historical messages or deliberately backfills older notifications. -This isolation requires the overlap migration and updated writers. Old images -and raw SQL writers can still run the legacy transactional trigger until replaced. +Before 0063, this isolation requires the overlap migration and updated writers; +legacy writers can still execute the transactional trigger. After 0063, the +trigger is absent and all supported writers must use the application producer. Pending matcher jobs and pending/claimed delivery jobs remain dormant. Do not process or delete them during urgent rollback. **Before re-enabling**, inspect diff --git a/migrations/0063_retire_push_match_trigger.sql b/migrations/0063_retire_push_match_trigger.sql new file mode 100644 index 0000000000..162f7e17bc --- /dev/null +++ b/migrations/0063_retire_push_match_trigger.sql @@ -0,0 +1,12 @@ +-- Push matching is produced by the application after event commit. Retire +-- only its compatibility trigger; channel-TTL refresh remains unchanged. +-- All event writers and rollback targets must support application enqueueing +-- before this migration runs. Published migrations remain checksum-frozen. +-- +-- DROP TRIGGER locks the partitioned parent and its partitions. Bound the +-- lock acquisition wait. This does not bound statement duration or the time +-- ingestion can be blocked while these locks are held. +SET LOCAL lock_timeout = '5s'; + +DROP TRIGGER events_enqueue_push_match ON events; +DROP FUNCTION enqueue_push_match_job(); diff --git a/schema/functions/enqueue_push_match_job.sql b/schema/functions/enqueue_push_match_job.sql deleted file mode 100644 index ae085e76ea..0000000000 --- a/schema/functions/enqueue_push_match_job.sql +++ /dev/null @@ -1,36 +0,0 @@ --- T1b push gate (migrations 0023 and 0059). Enqueue only when the --- community has an active, endpoint-enabled, unexpired lease; the shared --- advisory lock pairs with the exclusive lock taken by lease activations --- (crates/buzz-db/src/push.rs) to close the lost-wake race. -CREATE FUNCTION enqueue_push_match_job() RETURNS trigger -LANGUAGE plpgsql AS $$ -BEGIN - -- Legacy connections without the setting preserve their existing behavior. - -- Updated writer pools always set it, including on reconnect. Check before - -- taking push locks or touching leases/queues so disabled chat stays healthy. - IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN - RETURN NEW; - END IF; - IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN - RETURN NEW; - END IF; - -- Keep this allowlist identical to the relay's validated NIP-PL descriptor. - -- This compatibility path covers legacy writers until they are replaced. - IF NEW.kind IN (9, 40002, 45001, 45003) THEN - PERFORM pg_advisory_xact_lock_shared( - hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0)); - IF EXISTS ( - SELECT 1 FROM push_leases - WHERE community_id = NEW.community_id - AND active - AND endpoint_enabled - AND expires_at > EXTRACT(EPOCH FROM now())::bigint - ) THEN - INSERT INTO push_match_queue (community_id, event_id) - VALUES (NEW.community_id, NEW.id) - ON CONFLICT DO NOTHING; - END IF; - END IF; - RETURN NEW; -END -$$; diff --git a/schema/schema.sql b/schema/schema.sql index 5ec8bd5829..bb71025f2c 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -64,7 +64,6 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; -- functions \i functions/channels_community_id_immutable.sql -\i functions/enqueue_push_match_job.sql \i functions/refresh_channel_ttl_after_event_insert.sql \i functions/guard_channel_roster_snapshot.sql \i functions/events_created_at_floor_guard.sql diff --git a/schema/triggers/events.sql b/schema/triggers/events.sql index 5491d1631d..4091f4fdc6 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -1,7 +1,3 @@ -CREATE TRIGGER events_enqueue_push_match -AFTER INSERT ON events -FOR EACH ROW EXECUTE FUNCTION enqueue_push_match_job(); - CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl AFTER INSERT ON events DEFERRABLE INITIALLY DEFERRED diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 7eef8c4bbe..2f6d8f5402 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -16,7 +16,6 @@ BEGIN -- pgschema may copy parent triggers onto standalone children. Drop -- those copies before ATTACH; PostgreSQL recreates inherited parent -- triggers while attaching and rejects same-named child triggers. - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_past; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_past; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_past; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_past; @@ -30,7 +29,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_01'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_01; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_01; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_01; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_01; @@ -44,7 +42,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_02'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_02; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_02; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_02; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_02; @@ -58,7 +55,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_03'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_03; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_03; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_03; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_03; @@ -72,7 +68,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_04'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_04; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_04; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_04; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_04; @@ -86,7 +81,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_05'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_05; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_05; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_05; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_05; @@ -100,7 +94,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_06'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_06; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_06; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_06; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_06; @@ -114,7 +107,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p_future'::regclass ) THEN - DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_future; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_future; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_future; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_future;