relay: skip TTL deadline bump for known-permanent channels (T1a write-amp)

bench/t1a_evidence.sh

@@ -0,0 +1,77 @@
+#!/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 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: messages committed after concurrent TTL activation
+#      extend the deadline beyond the activation update's own deadline.
+#
+# Usage: bench/t1a_evidence.sh <pg_container> <db_url> <relay_ws_url> <community_uuid>
+# Requires: relay running FROM THIS TREE against <db_url>; 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 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
+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.
+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, 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
+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; }

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

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<i32>,
+    ) -> 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<DateTime<Utc>> = 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<Utc> = 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<Utc> = 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<Utc> = 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<i32> = 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<Utc> = 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<Utc> = 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() {

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,

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]

crates/buzz-relay/src/handlers/ingest.rs

@@ -2351,16 +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.
-    if let Some(ch_id) = channel_id {
-        if kind_u32 != KIND_NIP29_CREATE_GROUP {
-            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)