From 2515f25c9c1dddbf56bf986cf650008f411a35c7 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Sun, 19 Jul 2026 11:08:28 -0400 Subject: [PATCH 1/2] relay: skip TTL deadline bump for known-permanent channels (T1a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every stored channel-scoped event currently executes bump_ttl_deadline, whose predicate (ttl_seconds IS NOT NULL) matches zero rows on permanent channels — a wasted write-shaped round trip and implicit transaction on the hot ingest path (~1 of the ~5 write txns/message measured in the write-amplification baseline). The ingest path already fetches the channel row once per request (E1 §4.8). Use it: skip the bump only when the row positively shows ttl_seconds IS NULL. A missing row (kind:9007 pre-create, transient get_channel error) still bumps, so ephemeral channels can never archive early due to a failed prefetch. A concurrent update_channel that sets a TTL resets ttl_deadline itself, making the skipped bump redundant by construction. Ships with bench/t1a_evidence.sh (runnable-from-tip gate): asserts zero TTL UPDATE statements via pg_stat_statements on a permanent channel, deadline still advances on an ephemeral channel, and the TTL-set-during-ingest race leaves a future deadline. wamp_bench is the shared load generator used by the evidence script and the baseline benchmark. Part of the relay write-amplification plan (rev 3), lane T1a. Base: 8908bd6b. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- bench/t1a_evidence.sh | 75 +++++++++++ crates/buzz-relay/src/handlers/ingest.rs | 13 +- crates/buzz-test-client/src/bin/wamp_bench.rs | 118 ++++++++++++++++++ 3 files changed, 205 insertions(+), 1 deletion(-) create mode 100755 bench/t1a_evidence.sh create mode 100644 crates/buzz-test-client/src/bin/wamp_bench.rs diff --git a/bench/t1a_evidence.sh b/bench/t1a_evidence.sh new file mode 100755 index 0000000000..e2d903854a --- /dev/null +++ b/bench/t1a_evidence.sh @@ -0,0 +1,75 @@ +#!/usr/bin/env bash +# T1a correctness evidence (runnable from this lane tip). +# +# Proves, against a live relay built from this tree: +# 1. permanent channel: kind:9 ingest emits ZERO `UPDATE channels ... ttl` statements +# (pg_stat_statements assertion, not absence-of-log); +# 2. ephemeral channel: the TTL bump is still observed (ttl_deadline strictly advances +# across a message); +# 3. TTL-set-during-ingest race: setting a TTL on a permanent channel concurrent with +# a message burst leaves ttl_deadline set and in the future (update_channel's own +# deadline reset covers the skipped bump). +# +# Usage: bench/t1a_evidence.sh +# Requires: relay running FROM THIS TREE against ; psql via docker exec; +# BENCH_PRIVATE_KEY env (member secret key hex); wamp_bench built --release. +set -euo pipefail + +PG="$1"; DBURL="$2"; RELAY="$3"; COMMUNITY="$4" +PSQL=(docker exec "$PG" psql "$DBURL" -tA) +sql() { "${PSQL[@]}" -c "$1"; } +BIN="./target/release/wamp_bench" +# BENCH_PUB = x-only pubkey hex for BENCH_PRIVATE_KEY (both required). +PUB="${BENCH_PUB:?set BENCH_PUB (x-only pubkey hex matching BENCH_PRIVATE_KEY)}" + +mkchan() { # $1 name, $2 ttl_seconds or NULL -> echoes uuid + local id; id=$(python3 -c "import uuid;print(uuid.uuid4())") + sql "insert into channels (id, community_id, name, channel_type, visibility, created_by, ttl_seconds, ttl_deadline) + values ('$id','$COMMUNITY','$1','stream','private',decode('$PUB','hex'),$2, + case when $2::int is null then null else now() + ($2::int || ' seconds')::interval end)" >/dev/null + sql "insert into channel_members (community_id, channel_id, pubkey, role) + values ('$COMMUNITY','$id',decode('$PUB','hex'),'owner')" >/dev/null + echo "$id" +} + +FAIL=0 + +echo "== 1. permanent channel: zero TTL UPDATE statements ==" +PERM=$(mkchan t1a-perm NULL) +sql "select pg_stat_statements_reset()" >/dev/null +env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$PERM" 20 10 2 /tmp/t1a-perm.lat >/tmp/t1a-perm.json +ACC=$(python3 -c "import json;print(json.load(open('/tmp/t1a-perm.json'))['accepted'])") +TTL_CALLS=$(sql "select coalesce(sum(calls),0) from pg_stat_statements where query ilike '%UPDATE channels SET ttl_deadline%'") +echo "accepted=$ACC ttl_update_calls=$TTL_CALLS" +[[ "$ACC" -gt 0 && "$TTL_CALLS" == "0" ]] || { echo "FAIL: expected >0 accepted and 0 TTL updates"; FAIL=1; } + +echo "== 2. ephemeral channel: bump still observed ==" +EPH=$(mkchan t1a-eph 3600) +D0=$(sql "select extract(epoch from ttl_deadline) from channels where id='$EPH'") +sleep 2 +env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$EPH" 5 3 1 /tmp/t1a-eph.lat >/tmp/t1a-eph.json +D1=$(sql "select extract(epoch from ttl_deadline) from channels where id='$EPH'") +echo "deadline before=$D0 after=$D1" +python3 -c "import sys; sys.exit(0 if float('$D1') > float('$D0') else 1)" \ + || { echo "FAIL: ephemeral ttl_deadline did not advance"; FAIL=1; } + +echo "== 3. TTL-set-during-ingest race ==" +RACE=$(mkchan t1a-race NULL) +env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$RACE" 50 6 4 /tmp/t1a-race.lat >/tmp/t1a-race.json & +BPID=$! +sleep 2 +# update_channel-equivalent: set TTL and reset deadline in one statement, mid-burst. +sql "update channels set ttl_seconds=600, ttl_deadline=now() + interval '600 seconds', updated_at=now() where id='$RACE' and deleted_at is null" +wait "$BPID" +ROW=$(sql "select ttl_seconds, (ttl_deadline > now()) from channels where id='$RACE'") +echo "post-race: $ROW" +[[ "$ROW" == "600|t" ]] || { echo "FAIL: expected ttl=600 with future deadline"; FAIL=1; } +# and subsequent messages now bump it (channel is ephemeral now) +D0=$(sql "select extract(epoch from ttl_deadline) from channels where id='$RACE'") +sleep 2 +env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$RACE" 5 3 1 /tmp/t1a-race2.lat >/tmp/t1a-race2.json +D1=$(sql "select extract(epoch from ttl_deadline) from channels where id='$RACE'") +python3 -c "import sys; sys.exit(0 if float('$D1') > float('$D0') else 1)" \ + || { echo "FAIL: post-race ephemeral bump not observed"; FAIL=1; } + +[[ "$FAIL" == 0 ]] && echo "T1A EVIDENCE: ALL PASS" || { echo "T1A EVIDENCE: FAILURES"; exit 1; } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 737396209e..a342fa63e0 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2353,8 +2353,19 @@ async fn ingest_event_inner( // Any successfully stored channel-scoped event keeps the channel alive. // Skip kind:9007 (create) — the deadline was just set during creation. + // Skip permanent channels: the fresh per-request `channel_row` already + // tells us `ttl_seconds` is NULL, so the UPDATE would match zero rows by + // its own predicate — pure wasted round trip on the hot path. Only a + // *positive* "no TTL" skips; a missing/unfetched row (kind:9007 pre-create, + // transient `get_channel` error swallowed by `.ok()`) still bumps, so an + // ephemeral channel can never archive early because of a failed prefetch. + // Race note: a concurrent `update_channel` that sets a TTL resets + // `ttl_deadline` itself (NOW() + ttl), making the skipped bump redundant. if let Some(ch_id) = channel_id { - if kind_u32 != KIND_NIP29_CREATE_GROUP { + let known_permanent = channel_row + .as_ref() + .is_some_and(|row| row.ttl_seconds.is_none()); + if kind_u32 != KIND_NIP29_CREATE_GROUP && !known_permanent { if let Err(e) = state.db.bump_ttl_deadline(tenant.community(), ch_id).await { warn!(channel = %ch_id, "TTL deadline bump failed: {e}"); } diff --git a/crates/buzz-test-client/src/bin/wamp_bench.rs b/crates/buzz-test-client/src/bin/wamp_bench.rs new file mode 100644 index 0000000000..b1bcad2104 --- /dev/null +++ b/crates/buzz-test-client/src/bin/wamp_bench.rs @@ -0,0 +1,118 @@ +//! Paced kind:9 load generator for relay write-amplification benchmarking. +//! +//! Opens `conns` authenticated WebSocket connections (one shared identity) +//! and sends kind:9 text events to `channel_uuid` at a total rate of `qps` +//! for `duration_secs`. Each connection is synchronous (send -> await OK), +//! paced by a per-connection tokio interval, so OK latency is measured +//! end-to-end. Emits a JSON summary on stdout and one raw latency sample +//! (milliseconds, f64) per line to `latency_out`. +//! +//! Usage: wamp-bench +//! Env: BUZZ_RELAY_URL (default ws://localhost:3000), BENCH_PRIVATE_KEY (hex) + +use std::time::{Duration, Instant}; + +use buzz_test_client::BuzzTestClient; +use nostr::Keys; +use tokio::time::MissedTickBehavior; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let _ = rustls::crypto::CryptoProvider::install_default( + rustls::crypto::aws_lc_rs::default_provider(), + ); + let args: Vec = std::env::args().collect(); + if args.len() != 6 { + eprintln!("Usage: wamp-bench "); + std::process::exit(1); + } + let channel_id = args[1].clone(); + let qps: f64 = args[2].parse()?; + let duration_secs: u64 = args[3].parse()?; + let conns: usize = args[4].parse()?; + let latency_out = args[5].clone(); + + let url = std::env::var("BUZZ_RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".into()); + let keys = match std::env::var("BENCH_PRIVATE_KEY") { + Ok(hex) => Keys::parse(&hex)?, + Err(_) => anyhow::bail!("BENCH_PRIVATE_KEY is required (channel member secret key)"), + }; + + let per_conn_interval = Duration::from_secs_f64(conns as f64 / qps); + let deadline = Instant::now() + Duration::from_secs(duration_secs); + + let mut tasks = Vec::new(); + for conn_idx in 0..conns { + let url = url.clone(); + let keys = keys.clone(); + let channel_id = channel_id.clone(); + tasks.push(tokio::spawn(async move { + let mut client = BuzzTestClient::connect(&url, &keys).await?; + let mut interval = tokio::time::interval(per_conn_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + let mut latencies: Vec = Vec::new(); + let mut sent: u64 = 0; + let mut rejected: u64 = 0; + let mut seq: u64 = 0; + while Instant::now() < deadline { + interval.tick().await; + seq += 1; + let content = format!( + "wamp-bench c{conn_idx} m{seq} payload: the quick brown fox jumps over the lazy dog 0123456789" + ); + let start = Instant::now(); + let ok = client + .send_text_message(&keys, &channel_id, &content, 9) + .await?; + let elapsed_ms = start.elapsed().as_secs_f64() * 1e3; + sent += 1; + if ok.accepted { + latencies.push(elapsed_ms); + } else { + rejected += 1; + eprintln!("REJECTED conn={conn_idx} seq={seq}: {}", ok.message); + } + } + client.disconnect().await?; + Ok::<_, anyhow::Error>((sent, rejected, latencies)) + })); + } + + let mut sent = 0u64; + let mut rejected = 0u64; + let mut latencies: Vec = Vec::new(); + for task in tasks { + let (s, r, l) = task.await??; + sent += s; + rejected += r; + latencies.extend(l); + } + latencies.sort_by(|a, b| a.partial_cmp(b).expect("finite latencies")); + let pct = |p: f64| -> f64 { + if latencies.is_empty() { + return f64::NAN; + } + let idx = ((latencies.len() as f64 - 1.0) * p).round() as usize; + latencies[idx] + }; + let raw: String = latencies.iter().map(|l| format!("{l:.3}\n")).collect(); + std::fs::write(&latency_out, raw)?; + println!( + "{}", + serde_json::json!({ + "sent": sent, + "accepted": sent - rejected, + "rejected": rejected, + "qps_target": qps, + "duration_secs": duration_secs, + "conns": conns, + "ok_latency_ms": { + "p50": pct(0.50), + "p95": pct(0.95), + "p99": pct(0.99), + "max": latencies.last().copied().unwrap_or(f64::NAN), + }, + }) + ); + Ok(()) +} From 598b80cce6b262ef1e731790182b679d047c8a49 Mon Sep 17 00:00:00 2001 From: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Date: Sun, 19 Jul 2026 14:02:41 -0400 Subject: [PATCH 2/2] db: refresh channel TTL in event transaction Move ephemeral-channel TTL refresh into a deferred event trigger so a concurrent permanent-to-ephemeral transition cannot be missed by a stale relay prefetch. The trigger locks by channel identity before evaluating TTL state and preserves the existing best-effort failure contract. Remove the obsolete post-insert application transaction and add Postgres coverage for permanent, ephemeral, duplicate, and forced activation-race behavior. Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> --- bench/t1a_evidence.sh | 22 ++-- crates/buzz-db/src/channel.rs | 19 --- crates/buzz-db/src/event.rs | 151 +++++++++++++++++++++++ crates/buzz-db/src/lib.rs | 9 -- crates/buzz-db/src/migration.rs | 21 +++- crates/buzz-relay/src/handlers/ingest.rs | 21 ---- migrations/0022_event_ttl_refresh.sql | 40 ++++++ 7 files changed, 220 insertions(+), 63 deletions(-) create mode 100644 migrations/0022_event_ttl_refresh.sql diff --git a/bench/t1a_evidence.sh b/bench/t1a_evidence.sh index e2d903854a..1f23f2dc42 100755 --- a/bench/t1a_evidence.sh +++ b/bench/t1a_evidence.sh @@ -2,13 +2,13 @@ # T1a correctness evidence (runnable from this lane tip). # # Proves, against a live relay built from this tree: -# 1. permanent channel: kind:9 ingest emits ZERO `UPDATE channels ... ttl` statements -# (pg_stat_statements assertion, not absence-of-log); +# 1. permanent channel: kind:9 ingest emits no separate top-level TTL UPDATE +# transaction (the deferred trigger's conditional statement stays inside the +# event transaction and affects zero rows); # 2. ephemeral channel: the TTL bump is still observed (ttl_deadline strictly advances # across a message); -# 3. TTL-set-during-ingest race: setting a TTL on a permanent channel concurrent with -# a message burst leaves ttl_deadline set and in the future (update_channel's own -# deadline reset covers the skipped bump). +# 3. TTL-set-during-ingest race: messages committed after concurrent TTL activation +# extend the deadline beyond the activation update's own deadline. # # Usage: bench/t1a_evidence.sh # Requires: relay running FROM THIS TREE against ; psql via docker exec; @@ -34,7 +34,7 @@ mkchan() { # $1 name, $2 ttl_seconds or NULL -> echoes uuid FAIL=0 -echo "== 1. permanent channel: zero TTL UPDATE statements ==" +echo "== 1. permanent channel: zero separate top-level TTL UPDATE statements ==" PERM=$(mkchan t1a-perm NULL) sql "select pg_stat_statements_reset()" >/dev/null env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$PERM" 20 10 2 /tmp/t1a-perm.lat >/tmp/t1a-perm.json @@ -59,11 +59,13 @@ env -u BUZZ_AUTH_TAG BUZZ_RELAY_URL="$RELAY" "$BIN" "$RACE" 50 6 4 /tmp/t1a-race BPID=$! sleep 2 # update_channel-equivalent: set TTL and reset deadline in one statement, mid-burst. -sql "update channels set ttl_seconds=600, ttl_deadline=now() + interval '600 seconds', updated_at=now() where id='$RACE' and deleted_at is null" +ACTIVATION_DEADLINE=$(sql "update channels set ttl_seconds=600, ttl_deadline=clock_timestamp() + interval '600 seconds', updated_at=now() where id='$RACE' and deleted_at is null returning extract(epoch from ttl_deadline)") wait "$BPID" -ROW=$(sql "select ttl_seconds, (ttl_deadline > now()) from channels where id='$RACE'") -echo "post-race: $ROW" -[[ "$ROW" == "600|t" ]] || { echo "FAIL: expected ttl=600 with future deadline"; FAIL=1; } +ROW=$(sql "select ttl_seconds, extract(epoch from ttl_deadline) from channels where id='$RACE'") +echo "activation_deadline=$ACTIVATION_DEADLINE post-race=$ROW" +FINAL_DEADLINE="${ROW#*|}" +python3 -c "import sys; sys.exit(0 if float('$FINAL_DEADLINE') > float('$ACTIVATION_DEADLINE') else 1)" \ + || { echo "FAIL: later message did not extend TTL beyond activation deadline"; FAIL=1; } # and subsequent messages now bump it (channel is ephemeral now) D0=$(sql "select extract(epoch from ttl_deadline) from channels where id='$RACE'") sleep 2 diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index c522d27bc2..5c3bd2538c 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -1312,25 +1312,6 @@ pub async fn get_member_role( Ok(row.map(|r| r.try_get("role")).transpose()?) } -/// Bump the TTL deadline for an ephemeral channel after a new message. -/// -/// No-op for permanent channels or channels that are already archived/deleted. -pub async fn bump_ttl_deadline( - pool: &PgPool, - community_id: CommunityId, - channel_id: Uuid, -) -> Result<()> { - sqlx::query( - "UPDATE channels SET ttl_deadline = NOW() + (ttl_seconds || ' seconds')::interval \ - WHERE community_id = $1 AND id = $2 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .execute(pool) - .await?; - Ok(()) -} - /// Archive ephemeral channels whose TTL deadline has passed. /// /// Returns the `(community_id, host, channel_id)` list that was archived. Idempotent — the diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 9a1cc5b664..352a1fe404 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1454,6 +1454,157 @@ mod tests { id } + async fn make_test_channel( + pool: &PgPool, + community_id: Uuid, + ttl_seconds: Option, + ) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query( + "INSERT INTO channels \ + (id, community_id, name, created_by, ttl_seconds, ttl_deadline) \ + VALUES ($1, $2, $3, $4, $5, \ + CASE WHEN $5 IS NULL THEN NULL \ + ELSE clock_timestamp() + make_interval(secs => $5) END)", + ) + .bind(id) + .bind(community_id) + .bind(format!("event-ttl-test-{}", id.simple())) + .bind(vec![7_u8; 32]) + .bind(ttl_seconds) + .execute(pool) + .await + .expect("insert test channel"); + id + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn event_insert_ttl_trigger_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); + + let permanent = make_test_channel(&pool, community_uuid, None).await; + let permanent_event = make_text_event("permanent channel event"); + assert!( + insert_event(&pool, community, &permanent_event, Some(permanent)) + .await + .expect("insert permanent event") + .1 + ); + let permanent_deadline: Option> = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(permanent) + .fetch_one(&pool) + .await + .expect("read permanent deadline"); + assert_eq!(permanent_deadline, None); + + let ephemeral = make_test_channel(&pool, community_uuid, Some(60)).await; + let initial_deadline: DateTime = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(ephemeral) + .fetch_one(&pool) + .await + .expect("read initial ephemeral deadline"); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let ephemeral_event = make_text_event("ephemeral channel event"); + assert!( + insert_event(&pool, community, &ephemeral_event, Some(ephemeral)) + .await + .expect("insert ephemeral event") + .1 + ); + let bumped_deadline: DateTime = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(ephemeral) + .fetch_one(&pool) + .await + .expect("read bumped ephemeral deadline"); + assert!(bumped_deadline > initial_deadline); + assert!( + !insert_event(&pool, community, &ephemeral_event, Some(ephemeral)) + .await + .expect("insert duplicate event") + .1 + ); + let duplicate_deadline: DateTime = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(ephemeral) + .fetch_one(&pool) + .await + .expect("read deadline after duplicate"); + assert_eq!(duplicate_deadline, bumped_deadline); + + // 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. + 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", + ) + .bind(community_uuid) + .bind(racing) + .fetch_one(&pool) + .await + .expect("prefetch permanent channel"); + assert_eq!(stale_ttl, None); + + let mut activation = pool.begin().await.expect("begin TTL activation"); + let activation_deadline: DateTime = sqlx::query_scalar( + "UPDATE channels \ + SET ttl_seconds = 60, ttl_deadline = clock_timestamp() + interval '60 seconds' \ + WHERE community_id = $1 AND id = $2 RETURNING ttl_deadline", + ) + .bind(community_uuid) + .bind(racing) + .fetch_one(&mut *activation) + .await + .expect("activate TTL while holding channel row lock"); + + let race_pool = pool.clone(); + let racing_event = make_text_event("event after stale permanent prefetch"); + let insert = tokio::spawn(async move { + insert_event(&race_pool, community, &racing_event, Some(racing)).await + }); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + !insert.is_finished(), + "event trigger must wait on TTL activation" + ); + activation.commit().await.expect("commit TTL activation"); + assert!( + insert + .await + .expect("join racing insert") + .expect("racing insert") + .1 + ); + + let final_deadline: DateTime = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(racing) + .fetch_one(&pool) + .await + .expect("read deadline after racing event"); + assert!( + final_deadline > activation_deadline + chrono::Duration::milliseconds(50), + "later event must extend TTL beyond activation deadline: activation={activation_deadline}, final={final_deadline}" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() { diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 5585833096..552150e132 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1688,15 +1688,6 @@ impl Db { channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await } - /// Bump the TTL deadline for an ephemeral channel after a new message. - pub async fn bump_ttl_deadline( - &self, - community_id: CommunityId, - channel_id: Uuid, - ) -> Result<()> { - channel::bump_ttl_deadline(&self.pool, community_id, channel_id).await - } - /// Archive ephemeral channels whose TTL deadline has passed. pub async fn reap_expired_ephemeral_channels( &self, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index c28fa3682f..dce6ca6ad4 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -532,9 +532,12 @@ mod tests { let normalized = normalize_sql(statement); let updates_channels = identifier_after_keyword(statement, "update").as_deref() == Some("channels"); + let update_assignments = normalized + .split_once(" set ") + .map(|(_, tail)| tail.split_once(" where ").map_or(tail, |(set, _)| set)); let mutates_with_update = updates_channels - && normalized.contains(" set ") - && normalized.contains("community_id"); + && update_assignments + .is_some_and(|assignments| assignments.contains("community_id")); let alters_channels = identifier_after_keyword(statement, "alter table").as_deref() == Some("channels"); let drops_channels = identifier_after_keyword(statement, "drop table").as_deref() @@ -557,7 +560,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 21); + assert_eq!(migrations.len(), 22); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -839,6 +842,16 @@ mod tests { .sql .as_str() .contains("join_policy_acceptances")); + + // Channel TTL refresh belongs to the event insertion transaction so a + // concurrent permanent -> ephemeral transition cannot be missed. + assert_eq!(migrations[21].version, 22); + let ttl_refresh = migrations[21].sql.as_str(); + assert!(ttl_refresh.contains("CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl")); + assert!(ttl_refresh.contains("AFTER INSERT ON events")); + assert!(ttl_refresh.contains("DEFERRABLE INITIALLY DEFERRED")); + assert!(ttl_refresh.contains("clock_timestamp()")); + assert!(ttl_refresh.contains("NEW.kind <> 9007")); } #[test] @@ -1081,7 +1094,7 @@ mod tests { run_migrations(&pool) .await .expect("retry succeeds after operator repair"); - assert_eq!(applied_versions(&pool).await.last().copied(), Some(21)); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(22)); } #[tokio::test] diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index a342fa63e0..707bf37efd 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2351,27 +2351,6 @@ async fn ingest_event_inner( }); } - // Any successfully stored channel-scoped event keeps the channel alive. - // Skip kind:9007 (create) — the deadline was just set during creation. - // Skip permanent channels: the fresh per-request `channel_row` already - // tells us `ttl_seconds` is NULL, so the UPDATE would match zero rows by - // its own predicate — pure wasted round trip on the hot path. Only a - // *positive* "no TTL" skips; a missing/unfetched row (kind:9007 pre-create, - // transient `get_channel` error swallowed by `.ok()`) still bumps, so an - // ephemeral channel can never archive early because of a failed prefetch. - // Race note: a concurrent `update_channel` that sets a TTL resets - // `ttl_deadline` itself (NOW() + ttl), making the skipped bump redundant. - if let Some(ch_id) = channel_id { - let known_permanent = channel_row - .as_ref() - .is_some_and(|row| row.ttl_seconds.is_none()); - if kind_u32 != KIND_NIP29_CREATE_GROUP && !known_permanent { - if let Err(e) = state.db.bump_ttl_deadline(tenant.community(), ch_id).await { - warn!(channel = %ch_id, "TTL deadline bump failed: {e}"); - } - } - } - if crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) diff --git a/migrations/0022_event_ttl_refresh.sql b/migrations/0022_event_ttl_refresh.sql new file mode 100644 index 0000000000..8b1632fa3f --- /dev/null +++ b/migrations/0022_event_ttl_refresh.sql @@ -0,0 +1,40 @@ +-- Refresh ephemeral-channel expiry in the transaction that makes a +-- channel-scoped event durable. Deferring to COMMIT closes the stale-prefetch +-- race: the UPDATE sees a TTL transition committed while ingest was in flight, +-- or waits on its row lock and rechecks after it commits, without restoring a +-- separate hot-path transaction. +CREATE FUNCTION refresh_channel_ttl_after_event_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + -- Kind 9007 creates the channel and initializes its deadline itself. + IF NEW.channel_id IS NOT NULL AND NEW.kind <> 9007 THEN + BEGIN + -- Lock by identity before testing ttl_seconds. If a concurrent TTL + -- transition is uncommitted, this waits and follows its updated row + -- version instead of treating the old permanent version as final. + PERFORM 1 FROM channels + WHERE community_id = NEW.community_id AND id = NEW.channel_id + FOR UPDATE; + + 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; + 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 +$$; + +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();