From c0ab789499bde118a468178d23b726fd109a8e1c Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 16 Oct 2026 11:22:21 -0400 Subject: [PATCH] feat(buzz-db): retire the channel-TTL trigger (#8210) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary This drops the channel-TTL event trigger. The application already refreshes ephemeral-channel deadlines in the event transaction (`event_follow_up::refresh_channel_ttls`, from #8135), so the trigger only repeats that work. This PR originally retired both follow-up triggers. #8352 has since retired the push-match trigger in migration 0063, so after the rebase this PR covers only the TTL half. Migration `0065_retire_channel_ttl_trigger.sql`: - `SET LOCAL lock_timeout = '5s'` (the same bound as 0063) - drops `events_refresh_channel_ttl` and `refresh_channel_ttl_after_event_insert()` Dropping the trigger on the partitioned parent also drops its partition clones. Published migrations, including 0022 and 0024, stay checksum-frozen. The same objects are removed from `schema/` (`schema/functions/…`, `schema/triggers/events.sql`, `schema/schema.sql`) and from the pgschema reconcile script. After this, migrated and desired-state databases have no event follow-up triggers. Tests: - `follow_up_retirement_tests.rs` (renamed from `push_retirement_tests.rs`): the push-retirement upgrade test now stops at 0063. A new test upgrades 0064 → 0065 and checks that the TTL trigger is gone from the parent and every partition, that the function is gone, and that the published checksums haven't changed. - `event_follow_up_postgres_tests.rs`: the contract now has two arms. `AppOnly` is the shipped schema, where both triggers are absent. `Dual` reconstructs the historical overlap of both triggers. The `PushRetired` arm (the push trigger dropped, the TTL trigger kept) was the shipped state before this PR. Its tests are removed because `Dual` and `AppOnly` together cover it. - `embedded_migrator_contains_consolidated_initial_schema` pins the 0065 statements and checks that both objects are absent from the desired schema. **Deploy gate:** do not run 0065 against a database still served by a relay older than #8135. After 0065, raw SQL writers no longer refresh TTL deadlines. `docs/push-gateway-deployment.md` is updated to match. ### What got simpler The trigger and the app no longer both refresh TTL deadlines; the app is now the only producer. One plpgsql function and one trigger are gone from both schema paths, and 8 per-partition drops are gone from the reconcile script. The follow-up contract drops from three arms to two. ### Related issue BUZZ-313 (Linear), part of the trigger-retirement work. ### Testing Head is `4f7d0baf0`. Results below are at `678927730` unless noted. `678927730..4f7d0baf0` changes only one test in `crates/buzz-db/src/store/event.rs`, described in the first bullet. - **Fixed test, `permanent_channel_event_commits_do_not_lock_the_channel_row`** (`4f7d0baf0`). - **Why it changed:** after 0065, its raw-pool inserts no longer ran any TTL refresh, so the test passed whatever `refresh_channel_ttls` did. This was raised in review. - **What it does now:** it inserts through the production `AdmittedTx` path and keeps the barrier and the tuple-lock probe. It commits all N transactions while an external transaction holds `FOR SHARE` on the channel row. It also asserts that no `buzz_db_channel_ttl_refresh_failures_total` is counted. - **Mutation check:** with `SELECT … FOR UPDATE` injected into `refresh_channel_ttl`, the new test fails (`must not wait on the channel row`, timeout at 10s). The old test passes under the same mutation. - **Blox:** fmt and `buzz-db` clippy pass. - **GitHub CI at `4f7d0baf0`:** every required check passes (56 pass, 0 fail). CI builds the merge ref with `main` `93ffc37cb`. - **New and renamed tests, checked by name in the CI logs at `4f7d0baf0`.** These pass in PostgreSQL Domain / PostgreSQL Tests ([117033725006](https://github.com/block/buzz/actions/runs/38915998692/job/117033725006)): - `permanent_channel_event_commits_do_not_lock_the_channel_row` - `migration_schema_ttl_retirement_drops_every_partition_trigger_and_keeps_checksums` - `migration_schema_push_retirement_preserves_ttl_and_published_checksums` - `event_insert_ttl_refresh_handles_permanent_ephemeral_duplicate_and_activation_race` - `{desired,migration_schema}_event_follow_up_contract_{dual,app_only}` (four tests) `embedded_migrator_contains_consolidated_initial_schema` passes in Rust / Unit Tests ([117024125423](https://github.com/block/buzz/actions/runs/38915998692/job/117024125423)). No `*_push_retired` test runs anywhere. - **Blox gates at `678927730`** (private PG16 + Redis, on a workstation shared with other sessions): - `cargo fmt --check`, workspace `clippy -D warnings` and `buzz-db --lib` all pass. - The `buzz-admin`, `buzz-audit`, `buzz-deletion`, `buzz-push-gateway`, `buzz-search` and `buzz-workflow` Postgres packages pass. - `buzz-db`: 517 of 519 tests passed. The 2 that hung, both `writer_pool_*` tests, were killed. `buzz-relay`: 413 of 426 passed. The 13 failures are covered below. - **Uncontended reruns at `4f7d0baf0`** on a fresh, idle workstation (PG16.15, `competing=0` for every arm): - **`writer_pool_rejects_non_read_committed_database_default` predates this PR.** It timed out 3/3 at head and 3/3 at `main` `93ffc37cb`, with a 300s timeout per run. - **`writer_pool_metrics_stop_after_session_timeout_setup_failure` also predates this PR.** It timed out 2/3 at head and 2/3 at `main`, and passed in the remaining run on each side. - Both tests pass in CI. The gate accepts them as a named exception for this PR, as a pre-existing hang on the PG16 test environment. - **`buzz-db`:** 518 of 519 tests passed. The single failure is `writer_pool_rejects_non_read_committed_database_default`, killed after it hung (the exception above). - **`buzz-relay`: still open.** 413 of 426 tests passed. The same 13 `handlers::group_state_postgres_tests` failed as on the shared box, each panicking at once in its fixture's `ensure community` step with `Sqlx(PoolTimedOut)`. Contention does not explain them. The fixture's community setup is in `buzz-db`, which this PR changes. A head/`main` comparison of the full `buzz-relay` package on the same idle box is running to find out whether this predates the PR. All 13 pass in CI at this head. - **E2E (Gimli, Blox)** at `050e593af`. `050e593af..678927730` changes only tests and docs; the migration, schema and runtime code are byte-identical. Gimli's audited verdict: **EXERCISED AND WORKS**, for DB and relay runtime only. Gimli ran it at `050e593af`; this head's runtime is byte-identical but was not freshly executed. - **Upgrade:** a real relay startup took the schema from 0064 to 0065. That removed the TTL function and the trigger from the parent table and from all 8 partitions. Checksums for migrations 1–64 are unchanged. - **Lock-timeout rollback:** with a real lock held, the migration failed with `55P03` at 5.0007s and the schema rolled back fully to 64. Once the lock was released, a retry applied 65. - **Fresh desired schema:** after pgschema plus reconcile, the retired objects are absent. - **Real signed kind-9 sends:** on both schema paths, these refreshed ephemeral-channel deadlines to about +600s. Permanent channels stayed NULL. - **Raw SQL insert:** at 64 it refreshed the deadline. After 65 it left the deadline unchanged, on both the upgraded and the fresh schema. - **Not covered:** native or device testing, and deploy. - **Review:** Legolas found 0 P0 and 0 P1. His three optional P2s are fixed in `678927730`. - **No human app test:** this is a DB-only change. - **Migration numbering:** #8237 and #8372 also add a `0065_*` migration. Whichever PR merges second must renumber. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- ...tests.rs => follow_up_retirement_tests.rs} | 141 +++++++++++++----- crates/buzz-db/src/runtime/migration.rs | 16 +- crates/buzz-db/src/store/channel.rs | 19 +-- crates/buzz-db/src/store/event.rs | 126 ++++++++++------ crates/buzz-db/src/store/event_follow_up.rs | 7 +- .../store/event_follow_up_postgres_tests.rs | 99 ++++++------ .../src/store/personal_read/postgres_tests.rs | 2 +- crates/buzz-db/tests/observability_source.rs | 4 +- docs/push-gateway-deployment.md | 15 +- .../0065_retire_channel_ttl_trigger.sql | 14 ++ ...refresh_channel_ttl_after_event_insert.sql | 41 ----- schema/schema.sql | 1 - schema/triggers/events.sql | 4 - scripts/reconcile-schema-after-pgschema.sql | 8 - 14 files changed, 280 insertions(+), 217 deletions(-) rename crates/buzz-db/src/runtime/{push_retirement_tests.rs => follow_up_retirement_tests.rs} (54%) create mode 100644 migrations/0065_retire_channel_ttl_trigger.sql delete mode 100644 schema/functions/refresh_channel_ttl_after_event_insert.sql diff --git a/crates/buzz-db/src/runtime/push_retirement_tests.rs b/crates/buzz-db/src/runtime/follow_up_retirement_tests.rs similarity index 54% rename from crates/buzz-db/src/runtime/push_retirement_tests.rs rename to crates/buzz-db/src/runtime/follow_up_retirement_tests.rs index a8601dbe26..cd71dbcab4 100644 --- a/crates/buzz-db/src/runtime/push_retirement_tests.rs +++ b/crates/buzz-db/src/runtime/follow_up_retirement_tests.rs @@ -1,6 +1,7 @@ -//! Upgrade contract for push-only trigger retirement. +//! Upgrade contracts for the event follow-up trigger retirements: push +//! (0063), then channel TTL (0065). use super::postgres_tests::{connect_test_pool, reset_public_schema}; -use super::{run_migrations, run_migrations_through}; +use super::run_migrations_through; use sqlx::PgPool; async fn ttl_definitions(pool: &PgPool) -> Vec<(String, String)> { @@ -20,6 +21,51 @@ async fn push_trigger_count(pool: &PgPool) -> i64 { .expect("count push triggers") } +/// Insert a channel with an expired TTL deadline, then a raw `INSERT INTO +/// events` on it that bypasses the application follow-up. Returns whether the +/// database itself pushed the deadline forward. +async fn raw_insert_refreshes_expired_ttl(pool: &PgPool, content: &str) -> bool { + 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, '[]', $4, $5, $6)", + ) + .bind(community) + .bind([12_u8; 32]) + .bind([11_u8; 32]) + .bind(content) + .bind([13_u8; 64]) + .bind(channel) + .execute(pool) + .await + .unwrap(); + 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() +} + #[tokio::test] #[ignore = "requires Postgres"] async fn migration_schema_push_retirement_preserves_ttl_and_published_checksums() { @@ -47,7 +93,9 @@ async fn migration_schema_push_retirement_preserves_ttl_and_published_checksums( .await .unwrap(); - run_migrations(&pool).await.expect("retire push trigger"); + run_migrations_through(&pool, 63) + .await + .expect("retire push trigger"); assert_eq!( push_trigger_count(&pool).await, 0, @@ -77,39 +125,64 @@ async fn migration_schema_push_retirement_preserves_ttl_and_published_checksums( // 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_insert_refreshes_expired_ttl(&pool, "TTL survives push retirement").await, "raw inserts still refresh TTL after push retirement" ); pool.close().await; } + +async fn ttl_function_present(pool: &PgPool) -> bool { + sqlx::query_scalar( + "SELECT to_regprocedure('refresh_channel_ttl_after_event_insert()') IS NOT NULL", + ) + .fetch_one(pool) + .await + .expect("look up TTL function") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_ttl_retirement_drops_every_partition_trigger_and_keeps_checksums() { + let pool = connect_test_pool().await; + reset_public_schema(&pool).await; + run_migrations_through(&pool, 64) + .await + .expect("apply published migrations"); + assert!( + ttl_definitions(&pool).await.len() > 1, + "parent and partition TTL triggers exist before upgrade" + ); + assert!(ttl_function_present(&pool).await); + let checksums_before: Vec<(i64, Vec)> = sqlx::query_as( + "SELECT version, checksum FROM _sqlx_migrations WHERE version <= 64 ORDER BY version", + ) + .fetch_all(&pool) + .await + .unwrap(); + + run_migrations_through(&pool, 65) + .await + .expect("retire TTL trigger"); + assert!( + ttl_definitions(&pool).await.is_empty(), + "retired on parent and every partition" + ); + assert!(!ttl_function_present(&pool).await); + assert_eq!(push_trigger_count(&pool).await, 0); + let checksums_after: Vec<(i64, Vec)> = sqlx::query_as( + "SELECT version, checksum FROM _sqlx_migrations WHERE version <= 64 ORDER BY version", + ) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(checksums_after, checksums_before); + + // With both triggers gone, a raw writer that skips the application + // follow-up no longer extends an ephemeral channel's deadline. + assert!( + !raw_insert_refreshes_expired_ttl(&pool, "TTL no longer refreshed by the database").await, + "raw inserts must not refresh TTL after 0065" + ); + pool.close().await; +} diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 74d6560e19..b373d37b4d 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -186,7 +186,7 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(conn: &mut PgConnection) -> } #[cfg(test)] -#[path = "push_retirement_tests.rs"] +#[path = "follow_up_retirement_tests.rs"] mod retirement_postgres_tests; #[cfg(test)] @@ -709,7 +709,19 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 64); + assert_eq!(migrations.len(), 65); + assert_eq!(migrations[64].version, 65); + let retire_ttl = strip_sql_comments(migrations[64].sql.as_str()); + for statement in [ + "SET LOCAL lock_timeout = '5s';", + "DROP TRIGGER events_refresh_channel_ttl ON events;", + "DROP FUNCTION refresh_channel_ttl_after_event_insert();", + ] { + assert!(retire_ttl.contains(statement), "0065 must run {statement}"); + } + let desired_schema = crate::test_support::desired_state_schema_sql(); + assert!(!desired_schema.contains("events_refresh_channel_ttl")); + assert!(!desired_schema.contains("refresh_channel_ttl_after_event_insert")); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index d8ff4dad9c..ffc60694ee 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -601,12 +601,13 @@ pub async fn update_channel( q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); - // T1a repair: a TTL change can flip this channel's event-trigger fast - // path (migration 0024 reads ttl_seconds under a SHARED per-channel - // advisory lock). Take the same key EXCLUSIVE before the UPDATE so a - // concurrent event either sees the committed TTL or strictly precedes - // this transition — whose own deadline reset is then the latest word. - // Non-TTL updates don't touch the fast path and skip the lock. + // T1a repair: a TTL change can flip this channel's event-commit fast + // path (`event_follow_up::refresh_channel_ttls` reads ttl_seconds under + // a SHARED per-channel advisory lock). Take the same key EXCLUSIVE + // before the UPDATE so a concurrent event either sees the committed TTL + // or strictly precedes this transition — whose own deadline reset is + // then the latest word. Non-TTL updates don't touch the fast path and + // skip the lock. if updates.ttl_seconds.is_some() { let mut tx = crate::begin_community_write_transaction( pool, @@ -1138,9 +1139,9 @@ 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. +/// (`event_follow_up::refresh_channel_ttls`); 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()) } diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 10078bde3e..3e76d75190 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -3259,7 +3259,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn event_insert_ttl_trigger_handles_permanent_ephemeral_duplicate_and_activation_race() { + async fn event_insert_ttl_refresh_handles_permanent_ephemeral_duplicate_and_activation_race() { let pool = setup_pool().await; let community_uuid = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_uuid); @@ -3326,8 +3326,8 @@ mod postgres_tests { // Reproduce the blocked stale-prefetch ordering: ingest has already // observed a permanent channel, then TTL activation locks/updates the - // row before the event INSERT reaches its trigger. The trigger must - // wait and refresh from the later event after activation commits. + // row before the event INSERT reaches its pre-commit TTL refresh. The + // refresh must wait, then use the later event once activation commits. let racing = make_test_channel(&pool, community_uuid, None).await; let stale_ttl: Option = sqlx::query_scalar( "SELECT ttl_seconds FROM channels WHERE community_id = $1 AND id = $2", @@ -3342,7 +3342,7 @@ mod postgres_tests { let mut activation = pool.begin().await.expect("begin TTL activation"); // Model the repaired update_channel protocol (migration 0024): the // TTL transition holds the per-channel advisory key EXCLUSIVE, which - // is what the event trigger's shared acquisition now waits on. + // is what the event TTL refresh's shared acquisition waits on. sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") .bind(format!("buzz_channel_ttl:{community_uuid}:{racing}")) .execute(&mut *activation) @@ -3367,7 +3367,7 @@ mod postgres_tests { tokio::time::sleep(std::time::Duration::from_millis(100)).await; assert!( !insert.is_finished(), - "event trigger must wait on TTL activation" + "event TTL refresh must wait on TTL activation" ); activation.commit().await.expect("commit TTL activation"); assert!( @@ -3392,53 +3392,59 @@ mod postgres_tests { ); } - /// T1a repair regression test (migration 0024): permanent-channel event - /// commits must not serialize on the channel row. The 0022 trigger took - /// `FOR UPDATE` on the channel tuple before testing `ttl_seconds`, so - /// concurrent commits into one hot permanent channel queued at commit - /// time (deferred trigger) — invisible to any single-connection test. - /// This holds N insert transactions at a barrier past their INSERTs, - /// then proves (a) while all N sit pre-commit, no transaction holds a - /// row-level lock on the channel tuple, and (b) all N commits succeed - /// with the channel row untouched (permanent ⇒ no deadline write). + /// T1a repair regression test (migration 0024, now app-side): commits of + /// permanent-channel events must not lock the channel row. The 0022 + /// trigger took `FOR UPDATE` on the channel tuple before testing + /// `ttl_seconds`, so concurrent commits into one hot permanent channel + /// queued at commit time, which no single-connection test can see. The + /// refresh now runs in [`AdmittedTx::commit`] as the last statement before + /// COMMIT. + /// + /// This inserts N events through the production admitted-transaction + /// path and holds every transaction at a barrier past its INSERT. It then + /// proves three things: + /// - (a) while all N are uncommitted, no transaction holds a row-level + /// lock on the channel tuple; + /// - (b) while an external transaction holds `FOR SHARE` on the channel + /// row, all N commits, and therefore their TTL refreshes, finish + /// without blocking and without a counted refresh failure; + /// - (c) the permanent channel's deadline stays NULL. + /// + /// Mutation oracle: if the refresh takes `FOR UPDATE` on the channel row, + /// the commits queue behind the `FOR SHARE` holder. The join times out, + /// or a `lock_timeout` is counted as a refresh failure. #[tokio::test] #[ignore = "requires Postgres"] async fn permanent_channel_event_commits_do_not_lock_the_channel_row() { const N: usize = 8; - // setup_pool's default cap (10) covers N held transactions plus the - // pg_locks inspector connection. + // setup_pool's default cap (10) covers N held transactions, the + // FOR SHARE holder, and the pg_locks inspector connection. let pool = setup_pool().await; let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); let channel = make_test_channel(&pool, community_uuid, None).await; - // Open N transactions, run the full event INSERT in each (the deferred - // trigger fires at COMMIT), and park them at a barrier. + // Open N admitted transactions, run the production event insert in + // each (the TTL refresh runs in `commit`), and park them at a barrier. let mut txs = Vec::new(); for i in 0..N { - let mut tx = pool.begin().await.expect("begin insert txn"); - let event = make_text_event(&format!("hot channel event {i}")); - sqlx::query( - "INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id) \ - VALUES ($1,$2,$3,$4,9,$5,$6,$7,now(),$8)", + let mut tx = crate::begin_community_event_write_transaction( + &pool, + community, + crate::observability::WriterOperation::EventWrite, ) - .bind(community_uuid) - .bind(event.id.as_bytes().as_slice()) - .bind(event.pubkey.as_bytes().as_slice()) - .bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).unwrap()) - .bind(serde_json::to_value(&event.tags).unwrap()) - .bind(&event.content) - .bind(event.sig.serialize().as_slice()) - .bind(channel) - .execute(&mut *tx) .await - .expect("insert event inside held txn"); + .expect("begin admitted insert txn"); + let event = make_text_event(&format!("hot channel event {i}")); + let (_, inserted) = insert_event_in_transaction(&mut tx, &event, Some(channel)) + .await + .expect("insert event inside held txn"); + assert!(inserted, "held txn {i} must insert its event"); txs.push(tx); } // With all N transactions holding completed INSERTs, none may hold a - // row-level lock on the channels tuple. (The 0022 trigger would not - // have taken it yet either — it locks at COMMIT — so also verify the - // commit phase below completes without mutual blocking.) + // row-level lock on the channels tuple. let tuple_locks: i64 = sqlx::query_scalar( "SELECT count(*) FROM pg_locks l \ JOIN pg_class c ON c.oid = l.relation \ @@ -3449,21 +3455,43 @@ mod postgres_tests { .expect("inspect pg_locks"); assert_eq!(tuple_locks, 0, "no channel tuple locks while txns are held"); - // Release all commits concurrently. Under 0022 these serialized on the - // channel row (each holding it across its WAL flush); under 0024 the - // shared advisory key admits them all. Join with a timeout so a - // regression fails fast instead of hanging the suite. - let commits = txs - .into_iter() - .map(|tx| tokio::spawn(async move { tx.commit().await })) - .collect::>(); - for c in commits { - tokio::time::timeout(std::time::Duration::from_secs(10), c) + // An external FOR SHARE holder conflicts with any FOR UPDATE / UPDATE + // row lock on the channel, so a refresh that locks the permanent row + // cannot complete while it is held. + let mut holder = pool.begin().await.expect("begin FOR SHARE holder"); + sqlx::query("SELECT 1 FROM channels WHERE community_id = $1 AND id = $2 FOR SHARE") + .bind(community_uuid) + .bind(channel) + .execute(&mut *holder) + .await + .expect("holder takes FOR SHARE on the channel row"); + + // Release all commits concurrently while the holder is open. Join with + // a timeout so a regression fails fast instead of hanging the suite. + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + { + let _guard = metrics::set_default_local_recorder(&recorder); + let commits = futures_util::future::join_all(txs.into_iter().map(AdmittedTx::commit)); + for result in tokio::time::timeout(std::time::Duration::from_secs(10), commits) .await - .expect("concurrent permanent-channel commits must not block") - .expect("join commit task") - .expect("commit succeeds"); + .expect("permanent-channel commits must not wait on the channel row") + { + result.expect("commit succeeds"); + } } + holder.rollback().await.expect("release FOR SHARE holder"); + + let refresh_failures: Vec<_> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_db_channel_ttl_refresh_failures_total") + .collect(); + assert!( + refresh_failures.is_empty(), + "permanent-channel refreshes must not fail: {refresh_failures:?}" + ); let deadline: Option> = sqlx::query_scalar( "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index be0d55902c..d2853b4d89 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -11,10 +11,9 @@ //! [`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). -//! Migration 0063 retires the push trigger; the TTL trigger remains installed. -//! Push requires application enqueueing on every supported writer. Every production +//! This replaced the `events_enqueue_push_match` and `events_refresh_channel_ttl` +//! triggers (migrations 0023, 0024, 0040; retired by 0063 and 0065), so nothing +//! in the database does this work any more. 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 ae47ff2e14..68b68e20e2 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 historical overlap, push retired, and both follow-ups app-only. +//! app-only as shipped and with the historical trigger overlap reconstructed. use crate::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus}; use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError}; use buzz_core::CommunityId; @@ -214,67 +214,66 @@ async fn activate_lease( ); } -/// Preserve the overlap and fully app-only compatibility contracts alongside -/// the shipped push-only retirement (the TTL trigger remains installed). +/// Which producers are live. `AppOnly` is the shipped schema: 0063 and 0065 +/// retired both follow-up triggers. `Dual` reconstructs the historical overlap +/// (app follow-ups plus both triggers) in this disposable database only. #[derive(Clone, Copy, Debug)] enum Arm { Dual, AppOnly, - PushRetired, } 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')", + WHERE tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')", ) .fetch_one(pool) .await - .expect("count follow-up triggers"); + .expect("count follow-up triggers, including partition clones"); assert_eq!( - triggers, 1, - "only the TTL trigger remains in the shipped schema" + triggers, 0, + "no follow-up trigger remains on any partition of 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) { + let absent: bool = sqlx::query_scalar( + "SELECT to_regprocedure('enqueue_push_match_job()') IS NULL \ + AND to_regprocedure('refresh_channel_ttl_after_event_insert()') IS NULL", + ) + .fetch_one(pool) + .await + .unwrap(); + assert!(absent, "no follow-up trigger function remains"); + if let Arm::Dual = arm { sqlx::raw_sql(include_str!( - "../../../../migrations/0059_push_rollback_gate.sql" + "../../../../migrations/0040_push_message_kinds.sql" )) .execute(pool) .await .unwrap(); + sqlx::raw_sql(include_str!( + "../../../../migrations/0024_event_ttl_refresh_shared_lock.sql" + )) + .execute(pool) + .await + .unwrap(); + } + sqlx::raw_sql(include_str!( + "../../../../migrations/0059_push_rollback_gate.sql" + )) + .execute(pool) + .await + .unwrap(); + if let Arm::Dual = arm { + for statement in [ + "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 \ + FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert()", + ] { + sqlx::query(statement).execute(pool).await.unwrap(); + } + } else { let absent: bool = sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL") .fetch_one(pool) @@ -512,7 +511,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { assert!( deadline(pool, community, channel).await >= database_time_before_commit + chrono::Duration::seconds(60), - "the deferred trigger must refresh TTL at commit" + "the TTL refresh must set the deadline at commit time" ); assert_eq!(match_count(&db, community, &committed).await, 1); @@ -824,18 +823,6 @@ 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/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs index cba0d992a4..99da1ff848 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -283,7 +283,7 @@ async fn personal_read_intent_does_not_lock_shared_conversation_rows() { .await .unwrap(); assert!(matches!(result, IntentOutcome::Applied)); - // Exercise the actual event-insert TTL trigger while private progress is + // Exercise the actual event-insert TTL refresh while private progress is // uncommitted, then verify event deletion can update the observed row. let incoming = EventBuilder::new(Kind::Custom(9), "concurrent ephemeral ingest") .sign_with_keys(&Keys::generate()) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 4d4edcf8e9..06aa82c8e6 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1925,8 +1925,8 @@ fn every_production_event_insert_runs_the_follow_up_hook() { 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:?}" + "production `INSERT INTO events` writers must call `event_follow_up` because no \ + trigger does the push enqueue or channel TTL refresh: {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. diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 331d49708f..30af6cbc56 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -236,13 +236,16 @@ 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 +After migration 0063, verify that `events_enqueue_push_match` is absent from +the parent and every partition and `enqueue_push_match_job()` is absent. +Migration 0065 then retires `events_refresh_channel_ttl` and +`refresh_channel_ttl_after_event_insert()` the same way; after it, raw SQL +writers no longer refresh ephemeral-channel deadlines. 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 +their messages may commit without producing notifications or extending +ephemeral-channel deadlines (so those channels can expire early). Do not restore +either 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. diff --git a/migrations/0065_retire_channel_ttl_trigger.sql b/migrations/0065_retire_channel_ttl_trigger.sql new file mode 100644 index 0000000000..f92ef7cc90 --- /dev/null +++ b/migrations/0065_retire_channel_ttl_trigger.sql @@ -0,0 +1,14 @@ +-- Channel-TTL refresh is done by the application in the event transaction +-- (crates/buzz-db/src/store/event_follow_up.rs). Retire the trigger that +-- repeats that work, completing the follow-up trigger retirement begun by +-- 0063. All event writers and rollback targets must support application TTL +-- refresh 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_refresh_channel_ttl ON events; +DROP FUNCTION refresh_channel_ttl_after_event_insert(); diff --git a/schema/functions/refresh_channel_ttl_after_event_insert.sql b/schema/functions/refresh_channel_ttl_after_event_insert.sql deleted file mode 100644 index 398c4c0db6..0000000000 --- a/schema/functions/refresh_channel_ttl_after_event_insert.sql +++ /dev/null @@ -1,41 +0,0 @@ --- Channel TTL refresh (keep in sync with migrations/0024). Runs deferred, in --- the transaction that makes a channel-scoped event durable, so a TTL --- transition committed while ingest was in flight is never missed. The --- per-channel advisory lock is SHARED here — permanent-channel commits admit --- each other — and taken EXCLUSIVE by TTL transitions (update_channel in --- crates/buzz-db/src/channel.rs), which forces the same total order the --- 0022 row lock provided without serializing the hot path. -CREATE FUNCTION refresh_channel_ttl_after_event_insert() RETURNS trigger -LANGUAGE plpgsql AS $$ -DECLARE - channel_ttl INTEGER; -BEGIN - -- Kind 9007 creates the channel and initializes its deadline itself. - IF NEW.channel_id IS NOT NULL AND NEW.kind <> 9007 THEN - BEGIN - PERFORM pg_advisory_xact_lock_shared(hashtextextended( - 'buzz_channel_ttl:' || NEW.community_id::text || ':' || NEW.channel_id::text, 0)); - - SELECT ttl_seconds INTO channel_ttl - FROM channels - WHERE community_id = NEW.community_id AND id = NEW.channel_id; - - IF channel_ttl IS NOT NULL THEN - UPDATE channels - SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds) - WHERE community_id = NEW.community_id - AND id = NEW.channel_id - AND ttl_seconds IS NOT NULL - AND archived_at IS NULL - AND deleted_at IS NULL; - END IF; - EXCEPTION WHEN OTHERS THEN - -- Preserve the existing best-effort contract: a TTL refresh failure - -- must not reject an otherwise valid durable event. - RAISE WARNING 'channel TTL refresh failed for community %, channel %: %', - NEW.community_id, NEW.channel_id, SQLERRM; - END; - END IF; - RETURN NULL; -END -$$; diff --git a/schema/schema.sql b/schema/schema.sql index b6d355cff9..f65943a362 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/refresh_channel_ttl_after_event_insert.sql \i functions/guard_channel_roster_snapshot.sql \i functions/events_created_at_floor_guard.sql \i functions/prevent_community_deletion_request_retargeting.sql diff --git a/schema/triggers/events.sql b/schema/triggers/events.sql index 4091f4fdc6..64d9bbdd5d 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -1,7 +1,3 @@ -CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl -AFTER INSERT ON events -DEFERRABLE INITIALLY DEFERRED -FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); CREATE TRIGGER trg_events_guard_channel_roster_snapshot BEFORE INSERT ON events FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 2f6d8f5402..06d6258643 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_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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_past; @@ -29,7 +28,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_01'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_01; @@ -42,7 +40,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_02'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_02; @@ -55,7 +52,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_03'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_03; @@ -68,7 +64,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_04'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_04; @@ -81,7 +76,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_05'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_05; @@ -94,7 +88,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_06'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_06; @@ -107,7 +100,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p_future'::regclass ) THEN - 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; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_future;