mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(buzz-db): retire the channel-TTL trigger (#8210)
## 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>
This commit is contained in:
@@ -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::postgres_tests::{connect_test_pool, reset_public_schema};
|
||||||
use super::{run_migrations, run_migrations_through};
|
use super::run_migrations_through;
|
||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
|
|
||||||
async fn ttl_definitions(pool: &PgPool) -> Vec<(String, String)> {
|
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")
|
.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]
|
#[tokio::test]
|
||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn migration_schema_push_retirement_preserves_ttl_and_published_checksums() {
|
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
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
run_migrations(&pool).await.expect("retire push trigger");
|
run_migrations_through(&pool, 63)
|
||||||
|
.await
|
||||||
|
.expect("retire push trigger");
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
push_trigger_count(&pool).await,
|
push_trigger_count(&pool).await,
|
||||||
0,
|
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
|
// A raw writer has no application TTL follow-up. The preserved deferred
|
||||||
// trigger must still refresh its channel when the event commits.
|
// 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!(
|
assert!(
|
||||||
refreshed,
|
raw_insert_refreshes_expired_ttl(&pool, "TTL survives push retirement").await,
|
||||||
"raw inserts still refresh TTL after push retirement"
|
"raw inserts still refresh TTL after push retirement"
|
||||||
);
|
);
|
||||||
pool.close().await;
|
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<u8>)> = 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<u8>)> = 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;
|
||||||
|
}
|
||||||
@@ -186,7 +186,7 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(conn: &mut PgConnection) ->
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
#[path = "push_retirement_tests.rs"]
|
#[path = "follow_up_retirement_tests.rs"]
|
||||||
mod retirement_postgres_tests;
|
mod retirement_postgres_tests;
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
@@ -709,7 +709,19 @@ mod postgres_tests {
|
|||||||
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
|
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
|
||||||
migrations.sort_by_key(|migration| migration.version);
|
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_eq!(migrations[58].version, 59);
|
||||||
assert!(migrations[58].sql.as_str().contains("buzz.push_enabled"));
|
assert!(migrations[58].sql.as_str().contains("buzz.push_enabled"));
|
||||||
assert_eq!(migrations[59].version, 60);
|
assert_eq!(migrations[59].version, 60);
|
||||||
|
|||||||
@@ -601,12 +601,13 @@ pub async fn update_channel(
|
|||||||
q = q.bind(community_id.as_uuid());
|
q = q.bind(community_id.as_uuid());
|
||||||
q = q.bind(channel_id);
|
q = q.bind(channel_id);
|
||||||
|
|
||||||
// T1a repair: a TTL change can flip this channel's event-trigger fast
|
// T1a repair: a TTL change can flip this channel's event-commit fast
|
||||||
// path (migration 0024 reads ttl_seconds under a SHARED per-channel
|
// path (`event_follow_up::refresh_channel_ttls` reads ttl_seconds under
|
||||||
// advisory lock). Take the same key EXCLUSIVE before the UPDATE so a
|
// a SHARED per-channel advisory lock). Take the same key EXCLUSIVE
|
||||||
// concurrent event either sees the committed TTL or strictly precedes
|
// before the UPDATE so a concurrent event either sees the committed TTL
|
||||||
// this transition — whose own deadline reset is then the latest word.
|
// or strictly precedes this transition — whose own deadline reset is
|
||||||
// Non-TTL updates don't touch the fast path and skip the lock.
|
// then the latest word. Non-TTL updates don't touch the fast path and
|
||||||
|
// skip the lock.
|
||||||
if updates.ttl_seconds.is_some() {
|
if updates.ttl_seconds.is_some() {
|
||||||
let mut tx = crate::begin_community_write_transaction(
|
let mut tx = crate::begin_community_write_transaction(
|
||||||
pool,
|
pool,
|
||||||
@@ -1138,9 +1139,9 @@ impl Db {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// The per-channel TTL advisory lock key. Event commits take it SHARED
|
/// 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
|
/// (`event_follow_up::refresh_channel_ttls`); TTL transitions in
|
||||||
/// retired); TTL transitions in [`update_channel`] take it EXCLUSIVE. Both
|
/// [`update_channel`] take it EXCLUSIVE. Both sides build it here, so the keys
|
||||||
/// sides build it here, so the keys cannot drift.
|
/// cannot drift.
|
||||||
pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String {
|
pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String {
|
||||||
format!("buzz_channel_ttl:{}:{channel}", community.as_uuid())
|
format!("buzz_channel_ttl:{}:{channel}", community.as_uuid())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3259,7 +3259,7 @@ mod postgres_tests {
|
|||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[ignore = "requires Postgres"]
|
#[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 pool = setup_pool().await;
|
||||||
let community_uuid = make_test_community(&pool).await;
|
let community_uuid = make_test_community(&pool).await;
|
||||||
let community = CommunityId::from_uuid(community_uuid);
|
let community = CommunityId::from_uuid(community_uuid);
|
||||||
@@ -3326,8 +3326,8 @@ mod postgres_tests {
|
|||||||
|
|
||||||
// Reproduce the blocked stale-prefetch ordering: ingest has already
|
// Reproduce the blocked stale-prefetch ordering: ingest has already
|
||||||
// observed a permanent channel, then TTL activation locks/updates the
|
// observed a permanent channel, then TTL activation locks/updates the
|
||||||
// row before the event INSERT reaches its trigger. The trigger must
|
// row before the event INSERT reaches its pre-commit TTL refresh. The
|
||||||
// wait and refresh from the later event after activation commits.
|
// refresh must wait, then use the later event once activation commits.
|
||||||
let racing = make_test_channel(&pool, community_uuid, None).await;
|
let racing = make_test_channel(&pool, community_uuid, None).await;
|
||||||
let stale_ttl: Option<i32> = sqlx::query_scalar(
|
let stale_ttl: Option<i32> = sqlx::query_scalar(
|
||||||
"SELECT ttl_seconds FROM channels WHERE community_id = $1 AND id = $2",
|
"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");
|
let mut activation = pool.begin().await.expect("begin TTL activation");
|
||||||
// Model the repaired update_channel protocol (migration 0024): the
|
// Model the repaired update_channel protocol (migration 0024): the
|
||||||
// TTL transition holds the per-channel advisory key EXCLUSIVE, which
|
// 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))")
|
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
|
||||||
.bind(format!("buzz_channel_ttl:{community_uuid}:{racing}"))
|
.bind(format!("buzz_channel_ttl:{community_uuid}:{racing}"))
|
||||||
.execute(&mut *activation)
|
.execute(&mut *activation)
|
||||||
@@ -3367,7 +3367,7 @@ mod postgres_tests {
|
|||||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||||
assert!(
|
assert!(
|
||||||
!insert.is_finished(),
|
!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");
|
activation.commit().await.expect("commit TTL activation");
|
||||||
assert!(
|
assert!(
|
||||||
@@ -3392,53 +3392,59 @@ mod postgres_tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// T1a repair regression test (migration 0024): permanent-channel event
|
/// T1a repair regression test (migration 0024, now app-side): commits of
|
||||||
/// commits must not serialize on the channel row. The 0022 trigger took
|
/// permanent-channel events must not lock the channel row. The 0022
|
||||||
/// `FOR UPDATE` on the channel tuple before testing `ttl_seconds`, so
|
/// trigger took `FOR UPDATE` on the channel tuple before testing
|
||||||
/// concurrent commits into one hot permanent channel queued at commit
|
/// `ttl_seconds`, so concurrent commits into one hot permanent channel
|
||||||
/// time (deferred trigger) — invisible to any single-connection test.
|
/// queued at commit time, which no single-connection test can see. The
|
||||||
/// This holds N insert transactions at a barrier past their INSERTs,
|
/// refresh now runs in [`AdmittedTx::commit`] as the last statement before
|
||||||
/// then proves (a) while all N sit pre-commit, no transaction holds a
|
/// COMMIT.
|
||||||
/// row-level lock on the channel tuple, and (b) all N commits succeed
|
///
|
||||||
/// with the channel row untouched (permanent ⇒ no deadline write).
|
/// 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]
|
#[tokio::test]
|
||||||
#[ignore = "requires Postgres"]
|
#[ignore = "requires Postgres"]
|
||||||
async fn permanent_channel_event_commits_do_not_lock_the_channel_row() {
|
async fn permanent_channel_event_commits_do_not_lock_the_channel_row() {
|
||||||
const N: usize = 8;
|
const N: usize = 8;
|
||||||
// setup_pool's default cap (10) covers N held transactions plus the
|
// setup_pool's default cap (10) covers N held transactions, the
|
||||||
// pg_locks inspector connection.
|
// FOR SHARE holder, and the pg_locks inspector connection.
|
||||||
let pool = setup_pool().await;
|
let pool = setup_pool().await;
|
||||||
let community_uuid = make_test_community(&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;
|
let channel = make_test_channel(&pool, community_uuid, None).await;
|
||||||
|
|
||||||
// Open N transactions, run the full event INSERT in each (the deferred
|
// Open N admitted transactions, run the production event insert in
|
||||||
// trigger fires at COMMIT), and park them at a barrier.
|
// each (the TTL refresh runs in `commit`), and park them at a barrier.
|
||||||
let mut txs = Vec::new();
|
let mut txs = Vec::new();
|
||||||
for i in 0..N {
|
for i in 0..N {
|
||||||
let mut tx = pool.begin().await.expect("begin insert txn");
|
let mut tx = crate::begin_community_event_write_transaction(
|
||||||
let event = make_text_event(&format!("hot channel event {i}"));
|
&pool,
|
||||||
sqlx::query(
|
community,
|
||||||
"INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id) \
|
crate::observability::WriterOperation::EventWrite,
|
||||||
VALUES ($1,$2,$3,$4,9,$5,$6,$7,now(),$8)",
|
|
||||||
)
|
)
|
||||||
.bind(community_uuid)
|
.await
|
||||||
.bind(event.id.as_bytes().as_slice())
|
.expect("begin admitted insert txn");
|
||||||
.bind(event.pubkey.as_bytes().as_slice())
|
let event = make_text_event(&format!("hot channel event {i}"));
|
||||||
.bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).unwrap())
|
let (_, inserted) = insert_event_in_transaction(&mut tx, &event, Some(channel))
|
||||||
.bind(serde_json::to_value(&event.tags).unwrap())
|
|
||||||
.bind(&event.content)
|
|
||||||
.bind(event.sig.serialize().as_slice())
|
|
||||||
.bind(channel)
|
|
||||||
.execute(&mut *tx)
|
|
||||||
.await
|
.await
|
||||||
.expect("insert event inside held txn");
|
.expect("insert event inside held txn");
|
||||||
|
assert!(inserted, "held txn {i} must insert its event");
|
||||||
txs.push(tx);
|
txs.push(tx);
|
||||||
}
|
}
|
||||||
|
|
||||||
// With all N transactions holding completed INSERTs, none may hold a
|
// With all N transactions holding completed INSERTs, none may hold a
|
||||||
// row-level lock on the channels tuple. (The 0022 trigger would not
|
// row-level lock on the channels tuple.
|
||||||
// have taken it yet either — it locks at COMMIT — so also verify the
|
|
||||||
// commit phase below completes without mutual blocking.)
|
|
||||||
let tuple_locks: i64 = sqlx::query_scalar(
|
let tuple_locks: i64 = sqlx::query_scalar(
|
||||||
"SELECT count(*) FROM pg_locks l \
|
"SELECT count(*) FROM pg_locks l \
|
||||||
JOIN pg_class c ON c.oid = l.relation \
|
JOIN pg_class c ON c.oid = l.relation \
|
||||||
@@ -3449,21 +3455,43 @@ mod postgres_tests {
|
|||||||
.expect("inspect pg_locks");
|
.expect("inspect pg_locks");
|
||||||
assert_eq!(tuple_locks, 0, "no channel tuple locks while txns are held");
|
assert_eq!(tuple_locks, 0, "no channel tuple locks while txns are held");
|
||||||
|
|
||||||
// Release all commits concurrently. Under 0022 these serialized on the
|
// An external FOR SHARE holder conflicts with any FOR UPDATE / UPDATE
|
||||||
// channel row (each holding it across its WAL flush); under 0024 the
|
// row lock on the channel, so a refresh that locks the permanent row
|
||||||
// shared advisory key admits them all. Join with a timeout so a
|
// cannot complete while it is held.
|
||||||
// regression fails fast instead of hanging the suite.
|
let mut holder = pool.begin().await.expect("begin FOR SHARE holder");
|
||||||
let commits = txs
|
sqlx::query("SELECT 1 FROM channels WHERE community_id = $1 AND id = $2 FOR SHARE")
|
||||||
.into_iter()
|
.bind(community_uuid)
|
||||||
.map(|tx| tokio::spawn(async move { tx.commit().await }))
|
.bind(channel)
|
||||||
.collect::<Vec<_>>();
|
.execute(&mut *holder)
|
||||||
for c in commits {
|
|
||||||
tokio::time::timeout(std::time::Duration::from_secs(10), c)
|
|
||||||
.await
|
.await
|
||||||
.expect("concurrent permanent-channel commits must not block")
|
.expect("holder takes FOR SHARE on the channel row");
|
||||||
.expect("join commit task")
|
|
||||||
.expect("commit succeeds");
|
// 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("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<DateTime<Utc>> = sqlx::query_scalar(
|
let deadline: Option<DateTime<Utc>> = sqlx::query_scalar(
|
||||||
"SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
|
"SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
|
||||||
|
|||||||
@@ -11,10 +11,9 @@
|
|||||||
//! [`refresh_channel_ttls`]). A refresh failure is logged and counted, and
|
//! [`refresh_channel_ttls`]). A refresh failure is logged and counted, and
|
||||||
//! the event still commits.
|
//! the event still commits.
|
||||||
//!
|
//!
|
||||||
//! This is the application-side replacement for the `events_enqueue_push_match`
|
//! This replaced the `events_enqueue_push_match` and `events_refresh_channel_ttl`
|
||||||
//! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040).
|
//! triggers (migrations 0023, 0024, 0040; retired by 0063 and 0065), so nothing
|
||||||
//! Migration 0063 retires the push trigger; the TTL trigger remains installed.
|
//! in the database does this work any more. Every production
|
||||||
//! Push requires application enqueueing on every supported writer. Every production
|
|
||||||
//! `INSERT INTO events` must call this module, enforced by source-policy tests.
|
//! `INSERT INTO events` must call this module, enforced by source-policy tests.
|
||||||
|
|
||||||
use std::collections::BTreeSet;
|
use std::collections::BTreeSet;
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
//! PostgreSQL contract for the event follow-up producers (push match
|
//! PostgreSQL contract for the event follow-up producers (push match
|
||||||
//! enqueue and channel TTL refresh), on the desired and migration schemas,
|
//! 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::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus};
|
||||||
use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError};
|
use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError};
|
||||||
use buzz_core::CommunityId;
|
use buzz_core::CommunityId;
|
||||||
@@ -214,37 +214,36 @@ async fn activate_lease(
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Preserve the overlap and fully app-only compatibility contracts alongside
|
/// Which producers are live. `AppOnly` is the shipped schema: 0063 and 0065
|
||||||
/// the shipped push-only retirement (the TTL trigger remains installed).
|
/// 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)]
|
#[derive(Clone, Copy, Debug)]
|
||||||
enum Arm {
|
enum Arm {
|
||||||
Dual,
|
Dual,
|
||||||
AppOnly,
|
AppOnly,
|
||||||
PushRetired,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn apply_arm(pool: &PgPool, arm: Arm) {
|
async fn apply_arm(pool: &PgPool, arm: Arm) {
|
||||||
let triggers: i64 = sqlx::query_scalar(
|
let triggers: i64 = sqlx::query_scalar(
|
||||||
"SELECT count(*) FROM pg_trigger \
|
"SELECT count(*) FROM pg_trigger \
|
||||||
WHERE tgrelid = 'events'::regclass \
|
WHERE tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')",
|
||||||
AND tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')",
|
|
||||||
)
|
)
|
||||||
.fetch_one(pool)
|
.fetch_one(pool)
|
||||||
.await
|
.await
|
||||||
.expect("count follow-up triggers");
|
.expect("count follow-up triggers, including partition clones");
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
triggers, 1,
|
triggers, 0,
|
||||||
"only the TTL trigger remains in the shipped schema"
|
"no follow-up trigger remains on any partition of the shipped schema"
|
||||||
);
|
);
|
||||||
let absent: bool =
|
let absent: bool = sqlx::query_scalar(
|
||||||
sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL")
|
"SELECT to_regprocedure('enqueue_push_match_job()') IS NULL \
|
||||||
|
AND to_regprocedure('refresh_channel_ttl_after_event_insert()') IS NULL",
|
||||||
|
)
|
||||||
.fetch_one(pool)
|
.fetch_one(pool)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert!(absent);
|
assert!(absent, "no follow-up trigger function remains");
|
||||||
match arm {
|
if let Arm::Dual = arm {
|
||||||
Arm::Dual => {
|
|
||||||
// Reconstruct the historical overlap explicitly in this test only.
|
|
||||||
sqlx::raw_sql(include_str!(
|
sqlx::raw_sql(include_str!(
|
||||||
"../../../../migrations/0040_push_message_kinds.sql"
|
"../../../../migrations/0040_push_message_kinds.sql"
|
||||||
))
|
))
|
||||||
@@ -252,29 +251,29 @@ async fn apply_arm(pool: &PgPool, arm: Arm) {
|
|||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
sqlx::raw_sql(include_str!(
|
sqlx::raw_sql(include_str!(
|
||||||
"../../../../migrations/0059_push_rollback_gate.sql"
|
"../../../../migrations/0024_event_ttl_refresh_shared_lock.sql"
|
||||||
))
|
))
|
||||||
.execute(pool)
|
.execute(pool)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.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!(
|
sqlx::raw_sql(include_str!(
|
||||||
"../../../../migrations/0059_push_rollback_gate.sql"
|
"../../../../migrations/0059_push_rollback_gate.sql"
|
||||||
))
|
))
|
||||||
.execute(pool)
|
.execute(pool)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.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 =
|
let absent: bool =
|
||||||
sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL")
|
sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL")
|
||||||
.fetch_one(pool)
|
.fetch_one(pool)
|
||||||
@@ -512,7 +511,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
|
|||||||
assert!(
|
assert!(
|
||||||
deadline(pool, community, channel).await
|
deadline(pool, community, channel).await
|
||||||
>= database_time_before_commit + chrono::Duration::seconds(60),
|
>= 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);
|
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;
|
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.
|
// Runs with both schema sources and with/without the overlap trigger.
|
||||||
async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) {
|
async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) {
|
||||||
let pool = db.pool();
|
let pool = db.pool();
|
||||||
|
|||||||
@@ -283,7 +283,7 @@ async fn personal_read_intent_does_not_lock_shared_conversation_rows() {
|
|||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert!(matches!(result, IntentOutcome::Applied));
|
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.
|
// uncommitted, then verify event deletion can update the observed row.
|
||||||
let incoming = EventBuilder::new(Kind::Custom(9), "concurrent ephemeral ingest")
|
let incoming = EventBuilder::new(Kind::Custom(9), "concurrent ephemeral ingest")
|
||||||
.sign_with_keys(&Keys::generate())
|
.sign_with_keys(&Keys::generate())
|
||||||
|
|||||||
@@ -1925,8 +1925,8 @@ fn every_production_event_insert_runs_the_follow_up_hook() {
|
|||||||
|
|
||||||
assert!(
|
assert!(
|
||||||
violations.is_empty(),
|
violations.is_empty(),
|
||||||
"production `INSERT INTO events` writers must call `event_follow_up` so the push \
|
"production `INSERT INTO events` writers must call `event_follow_up` because no \
|
||||||
enqueue and channel TTL refresh survive the trigger retirement: {violations:?}"
|
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
|
// 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.
|
// stopped seeing a writer, not that one was removed safely.
|
||||||
|
|||||||
@@ -236,13 +236,16 @@ migration disabled during the transition. Resolve overlapping retirement
|
|||||||
migrations and verify the published migration ledger before execution. Migration
|
migrations and verify the published migration ledger before execution. Migration
|
||||||
0063's lock timeout bounds acquisition waits, not lock-hold or statement duration.
|
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
|
After migration 0063, verify that `events_enqueue_push_match` is absent from
|
||||||
parent and every partition, `enqueue_push_match_job()` is absent, and the
|
the parent and every partition and `enqueue_push_match_job()` is absent.
|
||||||
channel-TTL trigger/function remain intact. Verify application enqueueing and
|
Migration 0065 then retires `events_refresh_channel_ttl` and
|
||||||
flag-only rollback on the exact deployed candidate. Legacy images and raw SQL
|
`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:
|
writers without the application producer are unsupported after this point:
|
||||||
their messages may commit without producing notifications. Do not restore the
|
their messages may commit without producing notifications or extending
|
||||||
trigger or downgrade an image as an urgent rollback step. Reintroducing legacy
|
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
|
writer support requires a separate reviewed forward migration and compatibility
|
||||||
plan; never edit published migrations or their checksums.
|
plan; never edit published migrations or their checksums.
|
||||||
|
|
||||||
|
|||||||
14
migrations/0065_retire_channel_ttl_trigger.sql
Normal file
14
migrations/0065_retire_channel_ttl_trigger.sql
Normal file
@@ -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();
|
||||||
@@ -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
|
|
||||||
$$;
|
|
||||||
@@ -64,7 +64,6 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
|||||||
|
|
||||||
-- functions
|
-- functions
|
||||||
\i functions/channels_community_id_immutable.sql
|
\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/guard_channel_roster_snapshot.sql
|
||||||
\i functions/events_created_at_floor_guard.sql
|
\i functions/events_created_at_floor_guard.sql
|
||||||
\i functions/prevent_community_deletion_request_retargeting.sql
|
\i functions/prevent_community_deletion_request_retargeting.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
|
CREATE TRIGGER trg_events_guard_channel_roster_snapshot
|
||||||
BEFORE INSERT ON events
|
BEFORE INSERT ON events
|
||||||
FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot();
|
FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot();
|
||||||
|
|||||||
@@ -16,7 +16,6 @@ BEGIN
|
|||||||
-- pgschema may copy parent triggers onto standalone children. Drop
|
-- pgschema may copy parent triggers onto standalone children. Drop
|
||||||
-- those copies before ATTACH; PostgreSQL recreates inherited parent
|
-- those copies before ATTACH; PostgreSQL recreates inherited parent
|
||||||
-- triggers while attaching and rejects same-named child triggers.
|
-- 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 events_created_at_floor ON events_p_past;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_past;
|
||||||
@@ -29,7 +28,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_01'::regclass
|
AND inhrelid = 'events_p2026_01'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_01;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_01;
|
||||||
@@ -42,7 +40,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_02'::regclass
|
AND inhrelid = 'events_p2026_02'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_02;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_02;
|
||||||
@@ -55,7 +52,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_03'::regclass
|
AND inhrelid = 'events_p2026_03'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_03;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_03;
|
||||||
@@ -68,7 +64,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_04'::regclass
|
AND inhrelid = 'events_p2026_04'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_04;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_04;
|
||||||
@@ -81,7 +76,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_05'::regclass
|
AND inhrelid = 'events_p2026_05'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_05;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_05;
|
||||||
@@ -94,7 +88,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p2026_06'::regclass
|
AND inhrelid = 'events_p2026_06'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p2026_06;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_06;
|
||||||
@@ -107,7 +100,6 @@ BEGIN
|
|||||||
WHERE inhparent = 'events'::regclass
|
WHERE inhparent = 'events'::regclass
|
||||||
AND inhrelid = 'events_p_future'::regclass
|
AND inhrelid = 'events_p_future'::regclass
|
||||||
) THEN
|
) 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 events_created_at_floor ON events_p_future;
|
||||||
DROP TRIGGER IF EXISTS community_write_fence_events 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;
|
DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_future;
|
||||||
|
|||||||
Reference in New Issue
Block a user