chore(push): retire push notification database trigger

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

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<i32>,
+    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<u8>,
+    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<GateBarrier>,
+) -> 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.

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<u8>)> = 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<u8>)> = 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;
+}

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;

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();