mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(buzz-db): run push enqueue and channel TTL refresh in app code (#8135)
## Summary Moves the producer side of the two event follow-up triggers into application code. The triggers stay in place, so both producers run until a later change drops the triggers after this ships to every pod. Builds on #8130 (`AdmittedTx`). Refs BUZZ-176. ### Push match enqueue New `store::event_follow_up::enqueue_push_match` replaces the `events_enqueue_push_match` trigger (`schema/functions/enqueue_push_match_job.sql`, migration 0023) and matches it exactly: - Same kind allowlist (9, 40002, 45001, 45003), now `PUSH_MATCH_KINDS`. - Takes the same SHARED `buzz_push_gate:<community>` advisory lock as its own statement, so under READ COMMITTED the lease check gets a snapshot taken after the lock is granted. Lease activations still take it EXCLUSIVE in `push.rs`. - Same lease predicate (`active AND endpoint_enabled AND expires_at > now`), then `INSERT push_match_queue ... ON CONFLICT DO NOTHING`. - Fails closed: any error rejects the event, as the trigger did. It runs right after the INSERT, only when the insert affected a row. The consumer side (claims, `lease_until`, attempts, reaper) is already app-owned and is unchanged. ### Channel TTL refresh `AdmittedTx` now records every channel that received an event (except kind 9007 and channel-less events). `AdmittedTx::commit()` refreshes their `ttl_deadline` as the last statement before COMMIT, which keeps the commit-time behavior of the deferred `events_refresh_channel_ttl` trigger (migration 0024): - One savepoint per channel, in sorted order, under the same SHARED `buzz_channel_ttl:<community>:<channel>` lock. - Runs the same `UPDATE` with the same `archived_at`/`deleted_at` filters as its own statement after the lock, so its snapshot follows the lock grant. Its `ttl_seconds IS NOT NULL` qual filters a permanent channel before any tuple lock, so a permanent channel's row is never locked. An ephemeral channel's row lock is held through the remaining refreshes and the COMMIT round trip (the trigger held it only inside COMMIT). - Lock keys come from one builder per protocol (`channel::channel_ttl_lock_key`, `push::push_gate_lock_key`), shared by the SHARED and EXCLUSIVE sides. The shared push-gate wait is observed through `observe_advisory_lock(LockType::PushGate, …)`. - On failure it rolls back the savepoint, logs a warning, and increments `buzz_db_channel_ttl_refresh_failures_total{reason}`, where `reason` is `lock_timeout` or `error`. The event still commits, matching the trigger's `EXCEPTION WHEN OTHERS`. A cancelled statement (`57014`) still rejects the transaction, because plpgsql never caught that either. - While both producers run, an app refresh that hits `lock_timeout` is counted, and then the trigger retries it inside COMMIT and may succeed. So during the dual window the failure counter can over-report and the worst-case wait doubles. Both go away when the retirement PR drops the triggers, after which the counter is exact. ### Call sites Every production `INSERT INTO events` calls `event_follow_up::after_admitted_insert`: - `insert_event_on` - `insert_event_with_thread_metadata_tx` - `LockedMemberSnapshot::replace_member_event` - `replace_addressable_event` - `publish_nip43_membership_locked` - `accept_lease_event` (kind 30350, a no-op today, kept so the policy has no exceptions) `replace_parameterized_event_in_transaction_impl` inserts inside the `parameterized_replace` savepoint. It enqueues the push job inside the savepoint, so a rollback to that savepoint also drops the job, and records the TTL channel only after the savepoint is released. The two rolled-back verification probes (`verify_floor_guard_behavior`, `verify_channel_roster_fence_behavior`) are named exceptions, because they never commit. ### Source policy `tests/observability_source.rs` adds `every_production_event_insert_runs_the_follow_up_hook`. It scans every `crates/*/src` and fails on any function that inserts into `events` without the hook. Matching is normalized (letter case, whitespace, `\` continuations, `\n`/`\t`/`\r` escapes, optional `public.`). The file is masked once first: comments (`//`, nested `/* */`) and string, raw-string (including `br"`/`cr"`), and char literals become spaces. `#[cfg(test)]` markers, `fn` signatures and the hook are all found in that masked view, so a marker in a comment (even a multiline block comment) strips nothing, an `fn` line inside a multiline string doesn't split a function, and a commented-out or quoted hook doesn't count. SQL is still matched in the original text. A table test pins the masker's output for each literal form. A fixture test pins what the scan rejects (no hook, push without TTL, a hook in a line comment, block comment, or string, lowercase or line-broken or escaped SQL, writers after test-only items or under comment mentions) and what it ignores (writers in `#[cfg(test)] mod`). A floor of seven hooked writers catches a scan that silently stops seeing writers. ### Tests `src/store/event_follow_up_postgres_tests.rs` is the contract suite from the BUZZ-176 discovery work, ported to the `AdmittedTx` API. It is a crate-internal `#[cfg(test)]` module because it runs `SET LOCAL` timeouts and in-transaction reads through the crate-private `AdmittedTx::conn()`. It runs on both the desired and migration schemas, in two arms: - **dual**: production during the migration window, with the triggers and the app code both running. - **app-only**: drops both triggers in the disposable database (and asserts no partition clones remain), the way the retirement migration will. The trigger-only baseline is the base branch, where the same suite passed before this change. The suite also pins the two branches that differ from "swallow everything": a `lock_timeout` failure increments `buzz_db_channel_ttl_refresh_failures_total{reason="lock_timeout"}` exactly once, and a statement cancel (`57014`) during the refresh rejects the event with no row, job, or deadline change. ## What got simpler Nothing yet. This adds a parallel producer during the migration window. It is what lets the follow-up PR delete two plpgsql functions, two triggers, and the "keep in sync with migrations/00xx" duplication between `schema/functions/` and `push.rs`. ## Verification Rebased onto `main` @ `745501962`. All runs used a Linux cloud workstation with the pinned toolchain (rustc/clippy 1.95.0, hermit), and the tree was clean before and after each run. The final head, `c5a06a82`, changes only `crates/buzz-db/tests/observability_source.rs` compared with `1eb11bde` (via `fbc9a598`, `7c0baa41`, `9199992c` and `d405b2f6`). Production source is identical between the two. | Check | SHA | Result | |---|---|---| | `cargo fmt --check`, `cargo clippy -p buzz-db --all-targets -D warnings` | `fbc9a598` | pass | | `cargo clippy` (workspace) | `1eb11bde` | pass | | source policy (`--test observability_source`), fmt, buzz-db clippy | `c5a06a82` | 17/17, pass, pass | | `cargo test --locked -p buzz-db` (non-PG) | `c5a06a82` | 148 lib + 17 policy + 5 doctests | | source policy (`--test observability_source`) | `fbc9a598` | 16/16 | | `cargo test --locked -p buzz-db` (non-PG) | `fbc9a598` | 148 lib + 16 policy + 5 doctests | | contract suite (`store::event_follow_up_postgres_tests`, desired/migration × dual/app-only) | `1eb11bde` | 4/4 | | buzz-db PG full | `1eb11bde` | 464/464 | | buzz-relay PG | `1eb11bde` | 378/378 | | GitHub CI | `fbc9a598` | 53 pass, 0 fail (Codex security review pending) | **Falsifications** (each mutation made the named test fail, then was reverted): - dropping the `relay_members` hook → policy names `publish_nip43_membership_locked` - no-op push producer → app-only arms fail, dual arms pass (trigger covers) - skipping the TTL refresh → app-only fails on the deadline; dual fails on the failure-metric assertion - swallowing `57014` → all four arms fail - dropping the metric increment → all four arms fail - unanchored `#[cfg(test)]` match, or no escape handling → fixture test fails - a `#[cfg(test)]` constructor that skips admission, added inside `impl AdmittedTx` (indented or as a column-0 `mod`) → `admitted_tx_is_constructed_only_by_admitting_constructors` fails; putting the strip back into the constructor scan → its fixture test fails - at `7c0baa41`: `accept_lease_event`'s hook commented out and the real call made through an alias → policy names `accept_lease_event` (the same mutation passes under the `fbc9a598` policy); hook matching on unmasked text → the comment/string fixtures fail - at `9199992c`/`d405b2f6`: no nested block-comment depth, no string-escape skip, no `br"`/`cr"` raw-literal prefix, or no raw-string branch at all in the masker → the follow-up fixture test fails - at `c5a06a82`: finding `#[cfg(test)]` in the unmasked text → `writer_after_block_commented_cfg_test` is no longer reported; finding `fn` signatures in the unmasked text → `writer_after_raw_string_example` is no longer reported; no `cr` raw prefix → the masker table test fails on `cr"` - unhooked writers appended to `side_effects.rs` (after test-only items, under a doc-comment mention, with an escaped newline) → policy names each **Live relay E2E**: an independent reviewer ran this at `1eb11bde` against a real relay and CLI, with PostgreSQL 17 and Redis 7, the follow-up triggers absent, and no partition clones. - A kind 9 on a TTL channel advanced the shortened deadline by about 580 s. On a permanent channel, the deadline stayed NULL. - Each accepted event produced exactly one pending `push_match_queue` row with the same event id. - An exact signed replay was a no-op. - While the exclusive TTL lock was held, the event and its job stayed invisible and the deadline didn't change. After release, both persisted once. - Roster create/join/add/remove kept SQL and wire state in step (1→2→3→2), with one current 39002 head. - NIP-33 replace, duplicate, stale, failure and retry cases all passed. Limits: the push lease was a SQL fixture, and push consumers were disabled. Kind 30023 is global, so channel-scoped TTL on the parameterized path is covered only by the PG contracts. No live relay rerun was done at `fbc9a598`, because that delta is test-only. **Latency** (measured at pre-rebase `f8380927`, same production delta; debug build on loopback, 1,000 samples per cell, EVENT→OK): head app-only p50/p99 3.80/4.58 ms (permanent) and 3.87/4.64 ms (TTL), against base trigger-only 3.49–3.53/4.00–4.03 ms and 3.56–3.62/4.30–4.46 ms. About +0.3 ms p50, from the extra client round trips (lock, UPDATE, savepoints, push lock/insert) that the triggers ran server-side. One debug loopback run, not a production p99. Not measured: contention from many writers on one hot TTL channel, where each writer now waits on the channel row lock for about two more client round trips (RELEASE SAVEPOINT and COMMIT) than with the trigger. In review, a burst of 8 concurrent posts to one TTL channel committed with no deadlock or lock timeout. Watch send p99 and `pg_locks` waits on `channels` during rollout. **Known pre-existing, not touched here:** `api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo` is flaky when Redis is present. It returns a 504 in 4 of 5 runs at base `745501962` and 2 of 5 at `fbc9a598`, always at the same assertion. No CI lane selects it. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Luke Tornquist
@@ -1,7 +1,10 @@
|
||||
//! Community-admitted event-write transactions.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use buzz_core::CommunityId;
|
||||
use sqlx::{PgConnection, Postgres, Transaction};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::deletion::{DeletionStore, ServingWriteLease};
|
||||
use crate::Result;
|
||||
@@ -68,6 +71,8 @@ use crate::Result;
|
||||
pub struct AdmittedTx {
|
||||
tx: Transaction<'static, Postgres>,
|
||||
community: CommunityId,
|
||||
/// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes.
|
||||
ttl_refresh_channels: BTreeSet<Uuid>,
|
||||
}
|
||||
|
||||
impl AdmittedTx {
|
||||
@@ -78,7 +83,11 @@ impl AdmittedTx {
|
||||
community: CommunityId,
|
||||
) -> Result<Self> {
|
||||
store.guard_transaction(&mut tx, community).await?;
|
||||
Ok(Self { tx, community })
|
||||
Ok(Self {
|
||||
tx,
|
||||
community,
|
||||
ttl_refresh_channels: BTreeSet::new(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Validate `lease` on `tx` under the community admission lock and wrap
|
||||
@@ -94,6 +103,7 @@ impl AdmittedTx {
|
||||
Ok(Self {
|
||||
tx,
|
||||
community: lease.community_id,
|
||||
ttl_refresh_channels: BTreeSet::new(),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -111,9 +121,29 @@ impl AdmittedTx {
|
||||
&mut self.tx
|
||||
}
|
||||
|
||||
/// Record that an event committed with this transaction belongs to
|
||||
/// `channel_id`, so [`AdmittedTx::commit`] refreshes the channel's TTL.
|
||||
pub(crate) fn record_channel_event(&mut self, channel_id: Option<Uuid>, kind: i32) {
|
||||
if let Some(channel) = crate::store::event_follow_up::ttl_refresh_channel(channel_id, kind)
|
||||
{
|
||||
self.ttl_refresh_channels.insert(channel);
|
||||
}
|
||||
}
|
||||
|
||||
/// Commit the transaction. This is the only commit path for admitted
|
||||
/// event writes.
|
||||
pub async fn commit(self) -> Result<()> {
|
||||
///
|
||||
/// Refreshes the TTL of every channel that received an event first, as
|
||||
/// the last statement before COMMIT.
|
||||
pub async fn commit(mut self) -> Result<()> {
|
||||
if !self.ttl_refresh_channels.is_empty() {
|
||||
crate::store::event_follow_up::refresh_channel_ttls(
|
||||
&mut self.tx,
|
||||
self.community,
|
||||
&self.ttl_refresh_channels,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
self.tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -585,11 +585,7 @@ pub async fn update_channel(
|
||||
if updates.ttl_seconds.is_some() {
|
||||
let mut tx = begin_event_write_transaction(pool).await?;
|
||||
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
|
||||
.bind(format!(
|
||||
"buzz_channel_ttl:{}:{}",
|
||||
community_id.as_uuid(),
|
||||
channel_id
|
||||
))
|
||||
.bind(channel_ttl_lock_key(community_id, channel_id))
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
let result = q.execute(&mut *tx).await?;
|
||||
@@ -1012,6 +1008,14 @@ impl Db {
|
||||
}
|
||||
}
|
||||
|
||||
/// The per-channel TTL advisory lock key. Event commits take it SHARED
|
||||
/// (`event_follow_up::refresh_channel_ttls`, and the 0024 trigger until it is
|
||||
/// retired); TTL transitions in [`update_channel`] take it EXCLUSIVE. Both
|
||||
/// sides build it here, so the keys cannot drift.
|
||||
pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String {
|
||||
format!("buzz_channel_ttl:{}:{channel}", community.as_uuid())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod postgres_tests {
|
||||
use super::*;
|
||||
|
||||
@@ -398,6 +398,13 @@ impl LockedMemberSnapshot {
|
||||
"member snapshot event id already exists".into(),
|
||||
));
|
||||
}
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
&mut self.tx,
|
||||
event.id.as_bytes().as_slice(),
|
||||
kind,
|
||||
Some(channel_id),
|
||||
)
|
||||
.await?;
|
||||
crate::insert_mentions_in_transaction(&mut self.tx, event, Some(channel_id)).await?;
|
||||
Ok((
|
||||
buzz_core::StoredEvent::with_received_at(
|
||||
|
||||
@@ -500,6 +500,15 @@ async fn insert_event_on(
|
||||
.await?;
|
||||
|
||||
let was_inserted = result.rows_affected() > 0;
|
||||
if was_inserted {
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
tx,
|
||||
id_bytes.as_slice(),
|
||||
kind_i32,
|
||||
channel_id,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
Ok((
|
||||
StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
|
||||
@@ -1744,6 +1753,14 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
|
||||
let was_inserted = result.rows_affected() > 0;
|
||||
|
||||
if was_inserted {
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
tx,
|
||||
id_bytes.as_slice(),
|
||||
kind_i32,
|
||||
channel_id,
|
||||
)
|
||||
.await?;
|
||||
|
||||
if let Some(ref meta) = thread_meta {
|
||||
let broadcast_val: bool = meta.broadcast;
|
||||
|
||||
|
||||
208
crates/buzz-db/src/store/event_follow_up.rs
Normal file
208
crates/buzz-db/src/store/event_follow_up.rs
Normal file
@@ -0,0 +1,208 @@
|
||||
//! App-owned follow-up work for a newly inserted `events` row.
|
||||
//!
|
||||
//! Two pieces of work follow every durable event insert:
|
||||
//!
|
||||
//! - **Push match enqueue.** Push-eligible kinds get a `push_match_queue` row
|
||||
//! when the community has an eligible lease. This runs right after the
|
||||
//! INSERT, in the same transaction, and any failure rejects the event.
|
||||
//! - **Channel TTL refresh.** A channel-scoped event pushes its ephemeral
|
||||
//! channel's `ttl_deadline` forward. [`AdmittedTx`] records the channel and
|
||||
//! runs the refresh as the last statement before COMMIT (see
|
||||
//! [`refresh_channel_ttls`]). A refresh failure is logged and counted, and
|
||||
//! the event still commits.
|
||||
//!
|
||||
//! This is the application-side replacement for the `events_enqueue_push_match`
|
||||
//! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040).
|
||||
//! Both run during the migration window; the push enqueue is idempotent
|
||||
//! (`ON CONFLICT DO NOTHING`) and a second TTL refresh is harmless. Every
|
||||
//! production `INSERT INTO events` must call this module, which
|
||||
//! `tests/observability_source.rs` enforces.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use buzz_core::CommunityId;
|
||||
use sqlx::PgConnection;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{AdmittedTx, DbError, Result};
|
||||
|
||||
/// Event kinds that wake push subscribers. Keep identical to the relay's
|
||||
/// validated NIP-PL descriptor.
|
||||
pub(crate) const PUSH_MATCH_KINDS: [i32; 4] = [9, 40002, 45001, 45003];
|
||||
|
||||
/// Kind 9007 creates the channel and initializes its deadline itself.
|
||||
const KIND_CHANNEL_CREATE: i32 = 9007;
|
||||
|
||||
/// SQLSTATE `lock_not_available`, raised when `lock_timeout` expires.
|
||||
const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03";
|
||||
|
||||
/// Follow-up for an event row inserted in an admitted transaction.
|
||||
///
|
||||
/// Call only when the INSERT affected a row. Enqueues the push match job now
|
||||
/// and records the channel for the pre-commit TTL refresh.
|
||||
///
|
||||
/// If the insert ran inside a savepoint that may still roll back, call
|
||||
/// [`enqueue_push_match`] inside the savepoint and
|
||||
/// [`AdmittedTx::record_channel_event`] after it is released instead.
|
||||
pub(crate) async fn after_admitted_insert(
|
||||
tx: &mut AdmittedTx,
|
||||
event_id: &[u8],
|
||||
kind: i32,
|
||||
channel_id: Option<Uuid>,
|
||||
) -> Result<()> {
|
||||
let community = tx.community();
|
||||
enqueue_push_match(tx.conn(), community, event_id, kind).await?;
|
||||
tx.record_channel_event(channel_id, kind);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Enqueue a push match job for a just-inserted event, if its kind is
|
||||
/// push-eligible and the community has an eligible lease.
|
||||
///
|
||||
/// Takes the push gate SHARED and holds it to transaction end. Lease
|
||||
/// activations take it EXCLUSIVE (`acquire_push_gate_lock` in `push.rs`), so
|
||||
/// an event either sees the committed lease or strictly precedes the
|
||||
/// activation, in which case no wake was owed. The lock must be its own
|
||||
/// statement: under READ COMMITTED the eligibility check needs a snapshot
|
||||
/// taken after the lock is granted.
|
||||
///
|
||||
/// Errors propagate and reject the event, as the trigger did.
|
||||
pub(crate) async fn enqueue_push_match(
|
||||
conn: &mut PgConnection,
|
||||
community: CommunityId,
|
||||
event_id: &[u8],
|
||||
kind: i32,
|
||||
) -> Result<()> {
|
||||
if !PUSH_MATCH_KINDS.contains(&kind) {
|
||||
return Ok(());
|
||||
}
|
||||
crate::observability::observe_advisory_lock(
|
||||
crate::observability::LockType::PushGate,
|
||||
sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
|
||||
.bind(crate::push::push_gate_lock_key(community))
|
||||
.execute(&mut *conn),
|
||||
)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO push_match_queue (community_id, event_id) \
|
||||
SELECT $1, $2 \
|
||||
WHERE EXISTS ( \
|
||||
SELECT 1 FROM push_leases \
|
||||
WHERE community_id = $1 \
|
||||
AND active \
|
||||
AND endpoint_enabled \
|
||||
AND expires_at > EXTRACT(EPOCH FROM now())::bigint) \
|
||||
ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(event_id)
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any.
|
||||
pub(crate) fn ttl_refresh_channel(channel_id: Option<Uuid>, kind: i32) -> Option<Uuid> {
|
||||
channel_id.filter(|_| kind != KIND_CHANNEL_CREATE)
|
||||
}
|
||||
|
||||
/// Refresh the TTL deadline of every channel that received an event in this
|
||||
/// transaction. [`AdmittedTx::commit`] calls this as its last work before
|
||||
/// COMMIT, which keeps the deferred-trigger timing from migration 0024: the
|
||||
/// deadline is computed at commit. An ephemeral channel's row lock, taken by
|
||||
/// the UPDATE, is held through the remaining refreshes and the COMMIT round
|
||||
/// trip; the trigger held it only inside COMMIT. A permanent channel's row is
|
||||
/// never locked.
|
||||
///
|
||||
/// Channels are refreshed in sorted order, so two transactions that touch the
|
||||
/// same channels cannot deadlock on the shared locks against an exclusive
|
||||
/// waiter. Each refresh runs in its own savepoint. A failure is rolled back,
|
||||
/// logged, and counted, and the event still commits, as with the trigger's
|
||||
/// `EXCEPTION WHEN OTHERS`. A cancelled statement (`57014`) still rejects the
|
||||
/// transaction, because plpgsql never caught it either.
|
||||
pub(crate) async fn refresh_channel_ttls(
|
||||
conn: &mut PgConnection,
|
||||
community: CommunityId,
|
||||
channels: &BTreeSet<Uuid>,
|
||||
) -> Result<()> {
|
||||
for &channel in channels {
|
||||
sqlx::query("SAVEPOINT channel_ttl_refresh")
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
match refresh_channel_ttl(conn, community, channel).await {
|
||||
Ok(()) => {
|
||||
sqlx::query("RELEASE SAVEPOINT channel_ttl_refresh")
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
}
|
||||
Err(error) => {
|
||||
// plpgsql `WHEN OTHERS` never caught `query_canceled`, so a
|
||||
// cancelled refresh has always rejected the event.
|
||||
if error.is_statement_cancelled() {
|
||||
return Err(error);
|
||||
}
|
||||
sqlx::query("ROLLBACK TO SAVEPOINT channel_ttl_refresh")
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
sqlx::query("RELEASE SAVEPOINT channel_ttl_refresh")
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
let reason = if sqlstate(&error).as_deref() == Some(SQLSTATE_LOCK_NOT_AVAILABLE) {
|
||||
"lock_timeout"
|
||||
} else {
|
||||
"error"
|
||||
};
|
||||
metrics::counter!(
|
||||
"buzz_db_channel_ttl_refresh_failures_total",
|
||||
"reason" => reason,
|
||||
)
|
||||
.increment(1);
|
||||
tracing::warn!(
|
||||
%error,
|
||||
channel_id = %channel,
|
||||
reason,
|
||||
"channel TTL refresh failed; event committed without refreshing the deadline"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn refresh_channel_ttl(
|
||||
conn: &mut PgConnection,
|
||||
community: CommunityId,
|
||||
channel: Uuid,
|
||||
) -> Result<()> {
|
||||
// SHARED here, EXCLUSIVE in `update_channel` for TTL transitions: the same
|
||||
// total order the 0022 row lock gave, without serializing hot channels.
|
||||
sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
|
||||
.bind(crate::channel::channel_ttl_lock_key(community, channel))
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
// A separate statement, so its snapshot is taken after the lock is
|
||||
// granted. `ttl_seconds IS NOT NULL` filters a permanent channel before
|
||||
// any tuple lock is taken, so its row is never locked (migration 0024).
|
||||
sqlx::query(
|
||||
"UPDATE channels \
|
||||
SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds) \
|
||||
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.as_uuid())
|
||||
.bind(channel)
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn sqlstate(error: &DbError) -> Option<String> {
|
||||
match error {
|
||||
DbError::Sqlx(sqlx::Error::Database(database_error)) => {
|
||||
database_error.code().map(|code| code.into_owned())
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
773
crates/buzz-db/src/store/event_follow_up_postgres_tests.rs
Normal file
773
crates/buzz-db/src/store/event_follow_up_postgres_tests.rs
Normal file
@@ -0,0 +1,773 @@
|
||||
//! PostgreSQL contract for the event follow-up producers (push match
|
||||
//! enqueue and channel TTL refresh), on the desired and migration schemas,
|
||||
//! with the follow-up triggers present (dual) and dropped (app-only).
|
||||
use crate::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus};
|
||||
use crate::{event, migration, push, AdmittedTx, Db, DbConfig, DbError};
|
||||
use buzz_core::CommunityId;
|
||||
use chrono::{DateTime, Utc};
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter};
|
||||
use nostr::{Event, EventBuilder, Keys, Kind, Tag};
|
||||
use sqlx::PgPool;
|
||||
use std::time::Duration;
|
||||
use tokio::time::timeout;
|
||||
use uuid::Uuid;
|
||||
|
||||
async fn connect(migration_schema: bool) -> Db {
|
||||
let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
|
||||
.expect("Postgres wrapper must provide an isolated database URL");
|
||||
let expected_mode = if migration_schema {
|
||||
"migration"
|
||||
} else {
|
||||
"desired"
|
||||
};
|
||||
assert_eq!(
|
||||
std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(),
|
||||
Ok(expected_mode)
|
||||
);
|
||||
let config = DbConfig {
|
||||
database_url,
|
||||
..DbConfig::default()
|
||||
};
|
||||
let db = Db::new(&config)
|
||||
.await
|
||||
.expect("connect to isolated database");
|
||||
if migration_schema {
|
||||
migration::run_migrations(db.pool())
|
||||
.await
|
||||
.expect("apply embedded migrations to empty database");
|
||||
}
|
||||
db
|
||||
}
|
||||
|
||||
async fn create_community(pool: &PgPool) -> CommunityId {
|
||||
let id = Uuid::new_v4();
|
||||
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
|
||||
.bind(id)
|
||||
.bind(format!("follow-up-{}.example", id.simple()))
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("create disposable community");
|
||||
CommunityId::from_uuid(id)
|
||||
}
|
||||
|
||||
async fn channel_with_ttl(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
author: &[u8],
|
||||
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::INT IS NULL THEN NULL \
|
||||
ELSE clock_timestamp() - interval '1 day' END)",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(community.as_uuid())
|
||||
.bind(format!("follow-up-{}", id.simple()))
|
||||
.bind(author)
|
||||
.bind(ttl_seconds)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("create disposable channel");
|
||||
id
|
||||
}
|
||||
|
||||
async fn create_channel(pool: &PgPool, community: CommunityId, author: &[u8]) -> Uuid {
|
||||
channel_with_ttl(pool, community, author, Some(60)).await
|
||||
}
|
||||
|
||||
fn signed_event(keys: &Keys, kind: u16, content: &str) -> Event {
|
||||
EventBuilder::new(Kind::Custom(kind), content)
|
||||
.sign_with_keys(keys)
|
||||
.expect("sign test event")
|
||||
}
|
||||
|
||||
async fn match_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 {
|
||||
sqlx::query_scalar(
|
||||
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(event.id.as_bytes().as_slice())
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("count push match jobs")
|
||||
}
|
||||
|
||||
async fn event_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 {
|
||||
sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
.bind(event.id.as_bytes().as_slice())
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("count stored events")
|
||||
}
|
||||
|
||||
async fn deadline(pool: &PgPool, community: CommunityId, channel: Uuid) -> DateTime<Utc> {
|
||||
sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
.bind(channel)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("read channel deadline")
|
||||
}
|
||||
|
||||
async fn optional_deadline(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
channel: Uuid,
|
||||
) -> Option<DateTime<Utc>> {
|
||||
sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
.bind(channel)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("read optional channel deadline")
|
||||
}
|
||||
|
||||
async fn reset_deadline(pool: &PgPool, community: CommunityId, channel: Uuid) -> DateTime<Utc> {
|
||||
sqlx::query_scalar(
|
||||
"UPDATE channels SET ttl_deadline=clock_timestamp()-interval '1 day' \
|
||||
WHERE community_id=$1 AND id=$2 RETURNING ttl_deadline",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(channel)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("reset disposable channel deadline")
|
||||
}
|
||||
|
||||
/// `buzz_db_channel_ttl_refresh_failures_total` by `reason` label.
|
||||
fn ttl_refresh_failures(snapshotter: &Snapshotter) -> Vec<(String, u64)> {
|
||||
snapshotter
|
||||
.snapshot()
|
||||
.into_vec()
|
||||
.into_iter()
|
||||
.filter(|(key, ..)| key.key().name() == "buzz_db_channel_ttl_refresh_failures_total")
|
||||
.map(|(key, _, _, value)| {
|
||||
let reason = key
|
||||
.key()
|
||||
.labels()
|
||||
.find(|label| label.key() == "reason")
|
||||
.map(|label| label.value().to_owned())
|
||||
.unwrap_or_default();
|
||||
let DebugValue::Counter(count) = value else {
|
||||
panic!("TTL refresh failures must be a counter");
|
||||
};
|
||||
(reason, count)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn begin_caller_owned_event_transaction(db: &Db, community: CommunityId) -> AdmittedTx {
|
||||
db.begin_event_write_transaction(community)
|
||||
.await
|
||||
.expect("begin admitted event transaction")
|
||||
}
|
||||
|
||||
fn assert_sqlstate(error: &DbError, expected: &str) {
|
||||
let DbError::Sqlx(sqlx::Error::Database(database_error)) = error else {
|
||||
panic!("expected a Postgres database error, got {error:?}");
|
||||
};
|
||||
assert_eq!(database_error.code().as_deref(), Some(expected));
|
||||
}
|
||||
|
||||
async fn activate_lease(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
keys: &Keys,
|
||||
marker: u8,
|
||||
expires_at: i64,
|
||||
) {
|
||||
let source_event_id = [marker; 32];
|
||||
let endpoint_hash = [marker.wrapping_add(64); 32];
|
||||
let subscriptions = serde_json::json!([]);
|
||||
assert_eq!(
|
||||
push::replace_active_lease(
|
||||
pool,
|
||||
community,
|
||||
&keys.public_key().to_bytes(),
|
||||
"follow-up-installation",
|
||||
push::LeaseVersion {
|
||||
source_event_id: &source_event_id,
|
||||
source_created_at: 1,
|
||||
generation: 1,
|
||||
expires_at,
|
||||
},
|
||||
push::ActiveLease {
|
||||
app_profile: "ios-production",
|
||||
endpoint_hash: &endpoint_hash,
|
||||
endpoint_grant: "test-grant",
|
||||
max_class: "default",
|
||||
subscriptions: &subscriptions,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("activate push lease"),
|
||||
push::ReplaceLeaseOutcome::Accepted
|
||||
);
|
||||
}
|
||||
|
||||
/// Which producers are live. `Dual` is production during the migration
|
||||
/// window: the app follow-ups plus the 0023/0024 triggers. `AppOnly` drops
|
||||
/// both triggers in this disposable database, as the retirement migration
|
||||
/// will.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum Arm {
|
||||
Dual,
|
||||
AppOnly,
|
||||
}
|
||||
|
||||
async fn apply_arm(pool: &PgPool, arm: Arm) {
|
||||
let triggers: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM pg_trigger \
|
||||
WHERE tgrelid = 'events'::regclass \
|
||||
AND tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')",
|
||||
)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("count follow-up triggers");
|
||||
assert_eq!(triggers, 2, "both follow-up triggers exist before the arm");
|
||||
if let Arm::AppOnly = arm {
|
||||
for statement in [
|
||||
"DROP TRIGGER events_enqueue_push_match ON events",
|
||||
"DROP TRIGGER events_refresh_channel_ttl ON events",
|
||||
] {
|
||||
sqlx::query(statement)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("drop follow-up trigger in disposable database");
|
||||
}
|
||||
let remaining: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM pg_trigger \
|
||||
WHERE tgname IN ('events_enqueue_push_match', 'events_refresh_channel_ttl')",
|
||||
)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.expect("count remaining follow-up triggers, including partition clones");
|
||||
assert_eq!(
|
||||
remaining, 0,
|
||||
"app-only arm must leave no trigger on any partition"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn assert_contract(migration_schema: bool, arm: Arm) {
|
||||
let db = connect(migration_schema).await;
|
||||
let pool = db.pool();
|
||||
apply_arm(pool, arm).await;
|
||||
let keys = Keys::generate();
|
||||
let community = create_community(pool).await;
|
||||
let channel = create_channel(pool, community, &keys.public_key().to_bytes()).await;
|
||||
let now = Utc::now().timestamp();
|
||||
activate_lease(pool, community, &keys, 11, now + 3600).await;
|
||||
|
||||
for kind in [9, 40002, 45001, 45003] {
|
||||
let deadline_before_first_insert = reset_deadline(pool, community, channel).await;
|
||||
let message = signed_event(&keys, kind, &format!("allowlisted-{kind}"));
|
||||
assert!(
|
||||
db.insert_event(community, &message, Some(channel))
|
||||
.await
|
||||
.expect("insert allowlisted event")
|
||||
.1
|
||||
);
|
||||
assert_eq!(event_count(pool, community, &message).await, 1);
|
||||
assert_eq!(match_count(pool, community, &message).await, 1);
|
||||
let deadline_after_first_insert = deadline(pool, community, channel).await;
|
||||
assert!(
|
||||
deadline_after_first_insert > deadline_before_first_insert,
|
||||
"first insertion of kind {kind} must refresh the channel TTL"
|
||||
);
|
||||
assert!(
|
||||
!db.insert_event(community, &message, Some(channel))
|
||||
.await
|
||||
.expect("deduplicate event")
|
||||
.1
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &message).await, 1);
|
||||
assert_eq!(
|
||||
deadline(pool, community, channel).await,
|
||||
deadline_after_first_insert,
|
||||
"replaying kind {kind} must not refresh the channel TTL"
|
||||
);
|
||||
}
|
||||
|
||||
let reaction = signed_event(&keys, 7, "not-push-eligible");
|
||||
assert!(
|
||||
db.insert_event(community, &reaction, Some(channel))
|
||||
.await
|
||||
.expect("insert reaction")
|
||||
.1
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &reaction).await, 0);
|
||||
|
||||
let no_lease_community = create_community(pool).await;
|
||||
let no_lease_message = signed_event(&keys, 9, "no-lease");
|
||||
assert!(
|
||||
db.insert_event(no_lease_community, &no_lease_message, None)
|
||||
.await
|
||||
.expect("insert without lease")
|
||||
.1
|
||||
);
|
||||
assert_eq!(
|
||||
match_count(pool, no_lease_community, &no_lease_message).await,
|
||||
0
|
||||
);
|
||||
|
||||
for (marker, state) in [(21_u8, "expired"), (22, "inactive"), (23, "disabled")] {
|
||||
let predicate_community = create_community(pool).await;
|
||||
activate_lease(
|
||||
pool,
|
||||
predicate_community,
|
||||
&keys,
|
||||
marker,
|
||||
if state == "expired" {
|
||||
now - 1
|
||||
} else {
|
||||
now + 3600
|
||||
},
|
||||
)
|
||||
.await;
|
||||
if state == "inactive" {
|
||||
let source_event_id = [marker.wrapping_add(32); 32];
|
||||
assert_eq!(
|
||||
push::revoke_lease(
|
||||
pool,
|
||||
predicate_community,
|
||||
&keys.public_key().to_bytes(),
|
||||
"follow-up-installation",
|
||||
push::LeaseVersion {
|
||||
source_event_id: &source_event_id,
|
||||
source_created_at: 2,
|
||||
generation: 2,
|
||||
expires_at: now + 3600,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("revoke disposable push lease"),
|
||||
push::ReplaceLeaseOutcome::Accepted
|
||||
);
|
||||
} else if state == "disabled" {
|
||||
sqlx::query("UPDATE push_leases SET endpoint_enabled=false WHERE community_id=$1")
|
||||
.bind(predicate_community.as_uuid())
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("disable disposable push endpoint");
|
||||
}
|
||||
let message = signed_event(&keys, 9, &format!("{state}-lease"));
|
||||
assert!(
|
||||
db.insert_event(predicate_community, &message, None)
|
||||
.await
|
||||
.expect("insert event for ineligible lease")
|
||||
.1
|
||||
);
|
||||
assert_eq!(event_count(pool, predicate_community, &message).await, 1);
|
||||
assert_eq!(match_count(pool, predicate_community, &message).await, 0);
|
||||
}
|
||||
|
||||
let mut push_gate_holder = pool.begin().await.expect("begin push-gate lock holder");
|
||||
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
|
||||
.bind(format!("buzz_push_gate:{}", community.as_uuid()))
|
||||
.execute(&mut *push_gate_holder)
|
||||
.await
|
||||
.expect("hold exclusive push-gate lock");
|
||||
let gated = signed_event(&keys, 9, "push-gate-timeout");
|
||||
let mut gated_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
sqlx::query("SET LOCAL lock_timeout = '100ms'")
|
||||
.execute(gated_tx.conn())
|
||||
.await
|
||||
.expect("bound push-gate lock wait");
|
||||
let gated_error = event::insert_event_in_transaction(&mut gated_tx, &gated, None)
|
||||
.await
|
||||
.expect_err("push-eligible insert must fail closed when its gate cannot be acquired");
|
||||
assert_sqlstate(&gated_error, "55P03");
|
||||
gated_tx
|
||||
.rollback()
|
||||
.await
|
||||
.expect("rollback failed push-gate transaction");
|
||||
assert_eq!(event_count(pool, community, &gated).await, 0);
|
||||
assert_eq!(match_count(pool, community, &gated).await, 0);
|
||||
|
||||
let ungated = signed_event(&keys, 7, "push-gate-non-eligible-control");
|
||||
let mut ungated_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
sqlx::query("SET LOCAL lock_timeout = '100ms'")
|
||||
.execute(ungated_tx.conn())
|
||||
.await
|
||||
.expect("bound unrelated lock wait");
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut ungated_tx, &ungated, None)
|
||||
.await
|
||||
.expect("non-push event must not use the push gate")
|
||||
.1
|
||||
);
|
||||
ungated_tx
|
||||
.commit()
|
||||
.await
|
||||
.expect("commit non-push control while gate held");
|
||||
assert_eq!(event_count(pool, community, &ungated).await, 1);
|
||||
assert_eq!(match_count(pool, community, &ungated).await, 0);
|
||||
push_gate_holder
|
||||
.rollback()
|
||||
.await
|
||||
.expect("release exclusive push-gate lock");
|
||||
let gated_control = signed_event(&keys, 9, "push-gate-released-control");
|
||||
assert!(
|
||||
db.insert_event(community, &gated_control, None)
|
||||
.await
|
||||
.expect("insert after push gate release")
|
||||
.1
|
||||
);
|
||||
assert_eq!(event_count(pool, community, &gated_control).await, 1);
|
||||
assert_eq!(match_count(pool, community, &gated_control).await, 1);
|
||||
|
||||
let rollback_deadline = reset_deadline(pool, community, channel).await;
|
||||
let rolled_back = signed_event(&keys, 9, "rolled-back");
|
||||
let mut rollback_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut rollback_tx, &rolled_back, Some(channel))
|
||||
.await
|
||||
.expect("insert before rollback")
|
||||
.1
|
||||
);
|
||||
let in_transaction: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(rolled_back.id.as_bytes().as_slice())
|
||||
.fetch_one(rollback_tx.conn())
|
||||
.await
|
||||
.expect("read uncommitted follow-up");
|
||||
assert_eq!(in_transaction, 1);
|
||||
rollback_tx
|
||||
.rollback()
|
||||
.await
|
||||
.expect("rollback event and follow-up");
|
||||
assert_eq!(event_count(pool, community, &rolled_back).await, 0);
|
||||
assert_eq!(match_count(pool, community, &rolled_back).await, 0);
|
||||
assert_eq!(deadline(pool, community, channel).await, rollback_deadline);
|
||||
|
||||
let before_commit = reset_deadline(pool, community, channel).await;
|
||||
let committed = signed_event(&keys, 9, "deferred-ttl");
|
||||
let mut commit_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut commit_tx, &committed, Some(channel))
|
||||
.await
|
||||
.expect("insert before commit")
|
||||
.1
|
||||
);
|
||||
let deadline_before_commit: DateTime<Utc> =
|
||||
sqlx::query_scalar("SELECT ttl_deadline FROM channels WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
.bind(channel)
|
||||
.fetch_one(commit_tx.conn())
|
||||
.await
|
||||
.expect("read deadline inside event transaction");
|
||||
assert_eq!(deadline_before_commit, before_commit);
|
||||
let database_time_before_commit: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
|
||||
.fetch_one(commit_tx.conn())
|
||||
.await
|
||||
.expect("sample database clock immediately before commit");
|
||||
commit_tx
|
||||
.commit()
|
||||
.await
|
||||
.expect("commit deferred TTL refresh");
|
||||
assert!(
|
||||
deadline(pool, community, channel).await
|
||||
>= database_time_before_commit + chrono::Duration::seconds(60),
|
||||
"the deferred trigger must refresh TTL at commit"
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &committed).await, 1);
|
||||
|
||||
let before_skips = reset_deadline(pool, community, channel).await;
|
||||
let channel_create = signed_event(&keys, 9007, "channel-create");
|
||||
assert!(
|
||||
db.insert_event(community, &channel_create, Some(channel))
|
||||
.await
|
||||
.expect("insert kind 9007")
|
||||
.1
|
||||
);
|
||||
assert_eq!(deadline(pool, community, channel).await, before_skips);
|
||||
assert_eq!(match_count(pool, community, &channel_create).await, 0);
|
||||
let channelless = signed_event(&keys, 9, "channel-null");
|
||||
assert!(
|
||||
db.insert_event(community, &channelless, None)
|
||||
.await
|
||||
.expect("insert channel-less event")
|
||||
.1
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &channelless).await, 1);
|
||||
|
||||
let no_ttl_channel =
|
||||
channel_with_ttl(pool, community, &keys.public_key().to_bytes(), None).await;
|
||||
let no_ttl_message = signed_event(&keys, 7, "ttl-null");
|
||||
assert!(
|
||||
db.insert_event(community, &no_ttl_message, Some(no_ttl_channel))
|
||||
.await
|
||||
.expect("insert into channel without TTL")
|
||||
.1
|
||||
);
|
||||
assert_eq!(
|
||||
optional_deadline(pool, community, no_ttl_channel).await,
|
||||
None
|
||||
);
|
||||
|
||||
for (column, marker) in [("archived_at", "archived"), ("deleted_at", "deleted")] {
|
||||
let inactive_channel = create_channel(pool, community, &keys.public_key().to_bytes()).await;
|
||||
let inactive_deadline = deadline(pool, community, inactive_channel).await;
|
||||
let update = match column {
|
||||
"archived_at" => {
|
||||
"UPDATE channels SET archived_at=clock_timestamp() \
|
||||
WHERE community_id=$1 AND id=$2"
|
||||
}
|
||||
"deleted_at" => {
|
||||
"UPDATE channels SET deleted_at=clock_timestamp() \
|
||||
WHERE community_id=$1 AND id=$2"
|
||||
}
|
||||
_ => unreachable!("fixed test column"),
|
||||
};
|
||||
sqlx::query(update)
|
||||
.bind(community.as_uuid())
|
||||
.bind(inactive_channel)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("mark disposable channel inactive");
|
||||
let message = signed_event(&keys, 7, &format!("{marker}-channel"));
|
||||
assert!(
|
||||
db.insert_event(community, &message, Some(inactive_channel))
|
||||
.await
|
||||
.expect("insert against inactive disposable channel")
|
||||
.1
|
||||
);
|
||||
assert_eq!(event_count(pool, community, &message).await, 1);
|
||||
assert_eq!(
|
||||
deadline(pool, community, inactive_channel).await,
|
||||
inactive_deadline
|
||||
);
|
||||
}
|
||||
|
||||
let before_lock_timeout = reset_deadline(pool, community, channel).await;
|
||||
let mut lock_holder = pool.begin().await.expect("begin lock holder");
|
||||
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
|
||||
.bind(format!(
|
||||
"buzz_channel_ttl:{}:{channel}",
|
||||
community.as_uuid()
|
||||
))
|
||||
.execute(&mut *lock_holder)
|
||||
.await
|
||||
.expect("hold exclusive TTL lock");
|
||||
let locked_event = signed_event(&keys, 9, "ttl-lock-timeout");
|
||||
let mut locked_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
sqlx::query("SET LOCAL lock_timeout = '100ms'")
|
||||
.execute(locked_tx.conn())
|
||||
.await
|
||||
.expect("bound TTL lock wait");
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut locked_tx, &locked_event, Some(channel))
|
||||
.await
|
||||
.expect("insert while TTL lock held")
|
||||
.1
|
||||
);
|
||||
let recorder = DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
{
|
||||
let _local = metrics::set_default_local_recorder(&recorder);
|
||||
timeout(Duration::from_secs(5), locked_tx.commit())
|
||||
.await
|
||||
.expect("commit must not hang on TTL lock")
|
||||
.expect("TTL warning must not abort event commit");
|
||||
}
|
||||
assert_eq!(
|
||||
ttl_refresh_failures(&snapshotter),
|
||||
[("lock_timeout".to_owned(), 1)],
|
||||
"a swallowed TTL refresh failure must be counted once, by reason"
|
||||
);
|
||||
|
||||
// A cancelled statement is not swallowed: plpgsql `WHEN OTHERS` never
|
||||
// caught 57014, so the event is rejected. Both arms discriminate: if
|
||||
// the app swallowed the cancel, the app-only arm would commit the
|
||||
// event, and in the dual arm the trigger's own wait at COMMIT is not
|
||||
// cut short by the 200ms statement timeout, so the commit outlives the
|
||||
// 5s guard.
|
||||
let cancelled_event = signed_event(&keys, 9, "ttl-statement-cancel");
|
||||
let mut cancelled_tx = begin_caller_owned_event_transaction(&db, community).await;
|
||||
sqlx::query("SET LOCAL statement_timeout = '200ms'")
|
||||
.execute(cancelled_tx.conn())
|
||||
.await
|
||||
.expect("bound TTL statement");
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut cancelled_tx, &cancelled_event, Some(channel))
|
||||
.await
|
||||
.expect("insert while TTL lock held")
|
||||
.1
|
||||
);
|
||||
let cancel_error = timeout(Duration::from_secs(5), cancelled_tx.commit())
|
||||
.await
|
||||
.expect("commit must not hang on TTL lock")
|
||||
.expect_err("a cancelled TTL refresh must reject the event");
|
||||
assert_sqlstate(&cancel_error, "57014");
|
||||
assert_eq!(event_count(pool, community, &cancelled_event).await, 0);
|
||||
assert_eq!(match_count(pool, community, &cancelled_event).await, 0);
|
||||
assert_eq!(
|
||||
deadline(pool, community, channel).await,
|
||||
before_lock_timeout
|
||||
);
|
||||
lock_holder
|
||||
.rollback()
|
||||
.await
|
||||
.expect("release exclusive TTL lock");
|
||||
assert_eq!(event_count(pool, community, &locked_event).await, 1);
|
||||
assert_eq!(match_count(pool, community, &locked_event).await, 1);
|
||||
|
||||
assert_entry_points(&db, community, channel, &keys).await;
|
||||
}
|
||||
|
||||
/// Every production writer that can carry a channel runs both follow-ups,
|
||||
/// not just `insert_event`. In the app-only arm this is what catches a
|
||||
/// writer that skips the hook.
|
||||
async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) {
|
||||
let pool = db.pool();
|
||||
|
||||
// Thread-metadata writer (relay ingest, reactions, serving-lease writes).
|
||||
let before = reset_deadline(pool, community, channel).await;
|
||||
let threaded = signed_event(keys, 9, "thread-metadata-writer");
|
||||
assert!(
|
||||
db.insert_event_with_thread_metadata(community, &threaded, Some(channel), None)
|
||||
.await
|
||||
.expect("insert through the thread-metadata writer")
|
||||
.1
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &threaded).await, 1);
|
||||
assert!(deadline(pool, community, channel).await > before);
|
||||
|
||||
// Replaceable writer (NIP-16 and NIP-29 discovery state).
|
||||
let before = reset_deadline(pool, community, channel).await;
|
||||
let replaceable = signed_event(keys, 39000, "replaceable-writer");
|
||||
assert!(
|
||||
db.replace_addressable_event(community, &replaceable, Some(channel))
|
||||
.await
|
||||
.expect("insert through the replaceable writer")
|
||||
.1
|
||||
);
|
||||
assert_eq!(match_count(pool, community, &replaceable).await, 0);
|
||||
assert!(deadline(pool, community, channel).await > before);
|
||||
|
||||
// Parameterized writer: the insert runs inside a savepoint, and the
|
||||
// TTL refresh must follow only an insert that survives it.
|
||||
let d_tag = format!("follow-up-{}", Uuid::new_v4().simple());
|
||||
let parameterized = EventBuilder::new(Kind::Custom(30023), "parameterized-writer")
|
||||
.tags([Tag::identifier(d_tag.clone())])
|
||||
.sign_with_keys(keys)
|
||||
.expect("sign parameterized event");
|
||||
let before = reset_deadline(pool, community, channel).await;
|
||||
let mut tx = begin_caller_owned_event_transaction(db, community).await;
|
||||
let result = db
|
||||
.replace_parameterized_event_in_transaction(
|
||||
&mut tx,
|
||||
¶meterized,
|
||||
&d_tag,
|
||||
Some(channel),
|
||||
ParameterizedReplacePrecondition::Unconditional,
|
||||
)
|
||||
.await
|
||||
.expect("insert through the parameterized writer");
|
||||
assert_eq!(result.status, ParameterizedReplaceStatus::Inserted);
|
||||
tx.commit().await.expect("commit parameterized insert");
|
||||
assert!(deadline(pool, community, channel).await > before);
|
||||
|
||||
let before = reset_deadline(pool, community, channel).await;
|
||||
let mut tx = begin_caller_owned_event_transaction(db, community).await;
|
||||
let replay = db
|
||||
.replace_parameterized_event_in_transaction(
|
||||
&mut tx,
|
||||
¶meterized,
|
||||
&d_tag,
|
||||
Some(channel),
|
||||
ParameterizedReplacePrecondition::Unconditional,
|
||||
)
|
||||
.await
|
||||
.expect("replay through the parameterized writer");
|
||||
assert_eq!(replay.status, ParameterizedReplaceStatus::Duplicate);
|
||||
tx.commit().await.expect("commit parameterized replay");
|
||||
assert_eq!(
|
||||
deadline(pool, community, channel).await,
|
||||
before,
|
||||
"a replay rolled back to its savepoint must not refresh the channel TTL"
|
||||
);
|
||||
|
||||
// Member-snapshot writer (kind 39002 on the roster-locked transaction).
|
||||
let relay_keys = Keys::generate();
|
||||
let before = reset_deadline(pool, community, channel).await;
|
||||
let mut snapshot = db
|
||||
.lock_member_snapshot(community, channel, &relay_keys.public_key().to_bytes())
|
||||
.await
|
||||
.expect("lock member snapshot");
|
||||
let roster = EventBuilder::new(Kind::Custom(39002), "")
|
||||
.tags([Tag::identifier(channel.to_string())])
|
||||
.sign_with_keys(&relay_keys)
|
||||
.expect("sign member snapshot");
|
||||
assert!(
|
||||
snapshot
|
||||
.replace_member_event(&roster)
|
||||
.await
|
||||
.expect("replace member snapshot")
|
||||
.1
|
||||
);
|
||||
snapshot.release().await.expect("commit member snapshot");
|
||||
assert!(deadline(pool, community, channel).await > before);
|
||||
|
||||
// One transaction, two channels, one of them twice: each channel is
|
||||
// refreshed once, at commit.
|
||||
let second = create_channel(pool, community, &keys.public_key().to_bytes()).await;
|
||||
let before_first = reset_deadline(pool, community, channel).await;
|
||||
let before_second = reset_deadline(pool, community, second).await;
|
||||
let mut tx = begin_caller_owned_event_transaction(db, community).await;
|
||||
for (target, content) in [
|
||||
(channel, "multi-a"),
|
||||
(second, "multi-b"),
|
||||
(channel, "multi-c"),
|
||||
] {
|
||||
let message = signed_event(keys, 7, content);
|
||||
assert!(
|
||||
event::insert_event_in_transaction(&mut tx, &message, Some(target))
|
||||
.await
|
||||
.expect("insert multi-channel event")
|
||||
.1
|
||||
);
|
||||
}
|
||||
let database_time_before_commit: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
|
||||
.fetch_one(tx.conn())
|
||||
.await
|
||||
.expect("sample database clock before multi-channel commit");
|
||||
tx.commit().await.expect("commit multi-channel transaction");
|
||||
for (target, before) in [(channel, before_first), (second, before_second)] {
|
||||
let after = deadline(pool, community, target).await;
|
||||
assert!(after > before);
|
||||
assert!(after >= database_time_before_commit + chrono::Duration::seconds(60));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn desired_event_follow_up_contract_dual() {
|
||||
assert_contract(false, Arm::Dual).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn desired_event_follow_up_contract_app_only() {
|
||||
assert_contract(false, Arm::AppOnly).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn migration_schema_event_follow_up_contract_dual() {
|
||||
assert_contract(true, Arm::Dual).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn migration_schema_event_follow_up_contract_app_only() {
|
||||
assert_contract(true, Arm::AppOnly).await;
|
||||
}
|
||||
@@ -20,6 +20,12 @@ pub mod deletion;
|
||||
pub mod dm;
|
||||
/// Event storage and retrieval.
|
||||
pub mod event;
|
||||
/// App-owned push enqueue and channel TTL refresh for inserted events.
|
||||
pub(crate) mod event_follow_up;
|
||||
|
||||
#[cfg(test)]
|
||||
mod event_follow_up_postgres_tests;
|
||||
|
||||
/// Home feed queries.
|
||||
pub mod feed;
|
||||
/// Git repository name registry (NIP-34 kind:30617).
|
||||
|
||||
@@ -29,20 +29,27 @@ async fn begin_operation_transaction(
|
||||
Ok(sqlx::Transaction::begin(connection, None).await?)
|
||||
}
|
||||
|
||||
/// Namespace for the per-community push-gate advisory lock. Must match the
|
||||
/// key built inside the `enqueue_push_match_job` trigger (migration 0023):
|
||||
/// event inserts take it SHARED there; every lease transition that can make
|
||||
/// match eligibility true takes it EXCLUSIVE here, forcing a total order so
|
||||
/// a concurrent event insert either sees the committed lease or strictly
|
||||
/// precedes the activation (in which case no wake was owed). Distinct key
|
||||
/// domain from the audit lock and the lease address/author locks.
|
||||
/// Namespace for the per-community push-gate advisory lock. Event inserts
|
||||
/// take it SHARED in `event_follow_up::enqueue_push_match` (and in the
|
||||
/// `enqueue_push_match_job` trigger from migration 0023 until it is retired);
|
||||
/// every lease transition that can make match eligibility true takes it
|
||||
/// EXCLUSIVE here, forcing a total order so a concurrent event insert either
|
||||
/// sees the committed lease or strictly precedes the activation (in which case
|
||||
/// no wake was owed). Distinct key domain from the audit lock and the lease
|
||||
/// address/author locks.
|
||||
const PUSH_GATE_LOCK_NAMESPACE: &str = "buzz_push_gate:";
|
||||
|
||||
/// The push-gate advisory lock key for `community`. Both sides of the lock
|
||||
/// protocol build it here, so the SHARED and EXCLUSIVE keys cannot drift.
|
||||
pub(crate) fn push_gate_lock_key(community: CommunityId) -> String {
|
||||
format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid())
|
||||
}
|
||||
|
||||
async fn acquire_push_gate_lock(tx: &mut sqlx::PgConnection, community: CommunityId) -> Result<()> {
|
||||
crate::observability::observe_advisory_lock(
|
||||
crate::observability::LockType::PushGate,
|
||||
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
|
||||
.bind(format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid()))
|
||||
.bind(push_gate_lock_key(community))
|
||||
.execute(&mut *tx),
|
||||
)
|
||||
.await?;
|
||||
@@ -246,7 +253,8 @@ pub async fn accept_lease_event(
|
||||
)
|
||||
.await?;
|
||||
// T1b: an activation can flip the community from "no eligible lease" to
|
||||
// "eligible", so it must serialize against the trigger's shared gate lock.
|
||||
// "eligible", so it must serialize against the event producer's shared
|
||||
// gate lock.
|
||||
// Acquired after the address/author locks to keep one global lock order.
|
||||
if active.is_some() {
|
||||
acquire_push_gate_lock(tx.conn(), community).await?;
|
||||
@@ -361,6 +369,14 @@ pub async fn accept_lease_event(
|
||||
}
|
||||
return Err(error.into());
|
||||
}
|
||||
// The row above is always kind 30350 with no channel.
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
&mut tx,
|
||||
event.id.as_bytes().as_slice(),
|
||||
30350,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let (is_active, app_profile, endpoint_hash, endpoint_grant, max_class, subscriptions) = active
|
||||
.map_or((false, None, None, None, None, None), |active| {
|
||||
@@ -475,8 +491,8 @@ async fn replace_lease(
|
||||
};
|
||||
|
||||
// T1b: an activating replacement can flip the community from "no eligible
|
||||
// lease" to "eligible"; serialize it against the trigger's shared gate
|
||||
// lock (gate → lease row, matching accept_lease_event's global order).
|
||||
// lease" to "eligible"; serialize it against the event producer's shared
|
||||
// gate lock (gate → lease row, matching accept_lease_event's global order).
|
||||
// Revocations (is_active = false) never make eligibility true and skip it.
|
||||
let mut tx =
|
||||
begin_operation_transaction(pool, crate::observability::WriterOperation::EventWrite)
|
||||
|
||||
@@ -1279,6 +1279,13 @@ impl Db {
|
||||
|
||||
let was_inserted = insert_result.rows_affected() > 0;
|
||||
if was_inserted {
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
&mut tx,
|
||||
event.id.as_bytes().as_slice(),
|
||||
kind_i32,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
} else {
|
||||
tx.rollback().await?;
|
||||
|
||||
@@ -333,6 +333,15 @@ async fn replace_parameterized_event_in_transaction_impl(
|
||||
if insert_result.rows_affected() == 0 {
|
||||
return Ok(false);
|
||||
}
|
||||
// Inside the savepoint, so a later failure here also drops the job.
|
||||
// The TTL refresh is recorded only once the savepoint is released.
|
||||
crate::store::event_follow_up::enqueue_push_match(
|
||||
tx.conn(),
|
||||
community_id,
|
||||
incoming_id,
|
||||
kind_i32,
|
||||
)
|
||||
.await?;
|
||||
|
||||
if is_nip_rs {
|
||||
sqlx::query(
|
||||
@@ -380,6 +389,9 @@ async fn replace_parameterized_event_in_transaction_impl(
|
||||
sqlx::query("RELEASE SAVEPOINT parameterized_replace")
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
if status == ParameterizedReplaceStatus::Inserted {
|
||||
tx.record_channel_event(channel_id, kind_i32);
|
||||
}
|
||||
|
||||
Ok(ParameterizedReplaceResult::new(
|
||||
event,
|
||||
@@ -535,6 +547,14 @@ impl Db {
|
||||
));
|
||||
}
|
||||
|
||||
crate::store::event_follow_up::after_admitted_insert(
|
||||
&mut tx,
|
||||
event.id.as_bytes().as_slice(),
|
||||
kind_i32,
|
||||
channel_id,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// The replaceable event and its denormalized mention index are one
|
||||
// authoritative discovery write. An indexing error must roll back the
|
||||
// new event and restore the previously-live event.
|
||||
|
||||
@@ -950,19 +950,28 @@ fn fn_signature_starts_here(trimmed_line: &str) -> bool {
|
||||
/// signature to the doc comment or attributes of the next function, so a
|
||||
/// following function's docs (which may quote SQL) are never attributed to the
|
||||
/// function before it.
|
||||
///
|
||||
/// Signatures and attributes are found in the masked view (see
|
||||
/// [`mask_comments_and_literals`]), so an `fn` line inside a comment or a
|
||||
/// multiline string never splits a function. Doc comments are blank in that
|
||||
/// view, so `///` lead-ins are read from the original line.
|
||||
fn function_slices(production_source: &str) -> Vec<&str> {
|
||||
let masked = mask_comments_and_literals(production_source);
|
||||
// (signature offset, offset where the next function's lead-in begins)
|
||||
let mut starts = Vec::new();
|
||||
let mut lead_ins = Vec::new();
|
||||
let mut lead_in: Option<usize> = None;
|
||||
let mut offset = 0usize;
|
||||
for line in production_source.split_inclusive('\n') {
|
||||
let trimmed = line.trim_start();
|
||||
let indent = line.len() - trimmed.len();
|
||||
for (line, code) in production_source
|
||||
.split_inclusive('\n')
|
||||
.zip(masked.split_inclusive('\n'))
|
||||
{
|
||||
let trimmed = code.trim_start();
|
||||
let indent = code.len() - trimmed.len();
|
||||
if fn_signature_starts_here(trimmed) {
|
||||
starts.push(offset + indent);
|
||||
lead_ins.push(lead_in.take().unwrap_or(offset + indent));
|
||||
} else if trimmed.starts_with("///") || trimmed.starts_with("#[") {
|
||||
} else if line.trim_start().starts_with("///") || trimmed.starts_with("#[") {
|
||||
lead_in.get_or_insert(offset);
|
||||
} else {
|
||||
lead_in = None;
|
||||
@@ -1130,7 +1139,7 @@ fn serving_table_policy_requires_admitted_tx_not_raw_transactions() {
|
||||
let source = r#"
|
||||
pub(crate) async fn raw_transaction_writer(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(&mut **tx)
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
@@ -1143,7 +1152,7 @@ pub(crate) async fn raw_connection_writer(conn: &mut sqlx::PgConnection) {
|
||||
pub(crate) async fn admitted_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.bind(tx.community().as_uuid())
|
||||
.execute(&mut **tx)
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
@@ -1161,76 +1170,80 @@ pub(crate) async fn admitted_writer(tx: &mut AdmittedTx) {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admitted_tx_is_constructed_only_by_admitting_constructors() {
|
||||
// The fields are private to `runtime/admitted_tx.rs`, so only that file can
|
||||
// build the value. Pin that each function there that builds it also admits
|
||||
// the transaction it wraps, so a new unguarded constructor cannot slip in.
|
||||
let source = include_str!("../src/runtime/admitted_tx.rs");
|
||||
let production: String = strip_cfg_test_items(source)
|
||||
/// Every function in `source` that builds an `AdmittedTx`, paired with whether
|
||||
/// it admits the transaction it wraps. The scan deliberately does not strip
|
||||
/// `#[cfg(test)]` items: a test-only forge would let crate tests run follow-ups
|
||||
/// on an unadmitted transaction, so it must admit like any other constructor.
|
||||
fn admitted_tx_constructors(source: &str) -> Vec<(String, bool)> {
|
||||
let production: String = source
|
||||
.lines()
|
||||
.filter(|line| !line.trim_start().starts_with("//"))
|
||||
// `impl … for AdmittedTx {` headers open a block, not a value.
|
||||
.filter(|line| !line.trim_start().starts_with("impl"))
|
||||
.map(|line| format!("{line}\n"))
|
||||
.collect();
|
||||
let constructors: Vec<&str> = function_slices(&production)
|
||||
function_slices(&production)
|
||||
.into_iter()
|
||||
.filter(|function| function.contains("Self {") || function.contains("AdmittedTx {"))
|
||||
.collect();
|
||||
.map(|function| {
|
||||
let admits = function.contains(".guard_transaction(&mut tx, community)")
|
||||
|| function.contains(".guard_transaction_with_serving_lease(&mut tx, lease)");
|
||||
(function.to_string(), admits)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admitted_tx_is_constructed_only_by_admitting_constructors() {
|
||||
// The fields are private to `runtime/admitted_tx.rs`, so only that file can
|
||||
// build the value. Pin that each function there that builds it also admits
|
||||
// the transaction it wraps, so a new unguarded constructor cannot slip in.
|
||||
let constructors = admitted_tx_constructors(include_str!("../src/runtime/admitted_tx.rs"));
|
||||
assert_eq!(
|
||||
constructors.len(),
|
||||
2,
|
||||
"AdmittedTx must have exactly the two admitting constructors"
|
||||
);
|
||||
for constructor in constructors {
|
||||
for (constructor, admits) in constructors {
|
||||
assert!(
|
||||
constructor.contains(".guard_transaction(&mut tx, community)")
|
||||
|| constructor.contains(".guard_transaction_with_serving_lease(&mut tx, lease)"),
|
||||
admits,
|
||||
"AdmittedTx constructor must admit the transaction it wraps: {constructor}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove each top-level `#[cfg(test)]` item and keep the production code
|
||||
/// around it. Truncating at the first `#[cfg(test)]` would hide production
|
||||
/// functions that follow a test-only item. Tracks brace depth: the item ends
|
||||
/// on the first line that leaves depth at or below zero and ends with `;` or
|
||||
/// `}`, ignoring a trailing `//` comment. A column-0 `}` line (rustfmt's
|
||||
/// top-level close) always ends it. Braces inside string or char literals are
|
||||
/// still counted, so a test-only item with unbalanced literal braces (for
|
||||
/// example `"{"`) runs on to the next column-0 `}`; no such item exists today.
|
||||
fn strip_cfg_test_items(source: &str) -> String {
|
||||
let mut kept = String::with_capacity(source.len());
|
||||
let mut lines = source.lines();
|
||||
while let Some(line) = lines.next() {
|
||||
if line != "#[cfg(test)]" {
|
||||
kept.push_str(line);
|
||||
kept.push('\n');
|
||||
continue;
|
||||
}
|
||||
let mut depth = 0_isize;
|
||||
for line in lines.by_ref().skip_while(|line| line.starts_with("#[")) {
|
||||
depth += line.matches('{').count() as isize - line.matches('}').count() as isize;
|
||||
// Any `//` may start the trailing comment (an earlier one can sit
|
||||
// inside a string such as `"https://…"`), so try every prefix. A
|
||||
// false match inside a string only ends the item early, which
|
||||
// scans more lines and can never hide production code.
|
||||
let item_end = depth <= 0
|
||||
&& line
|
||||
.match_indices("//")
|
||||
.map(|(at, _)| &line[..at])
|
||||
.chain([line])
|
||||
.any(|text| {
|
||||
let text = text.trim_end();
|
||||
text.ends_with(';') || text.ends_with('}')
|
||||
});
|
||||
if item_end || line == "}" || line == "};" {
|
||||
break;
|
||||
}
|
||||
}
|
||||
#[test]
|
||||
fn admitted_tx_constructor_scan_reports_test_only_forges() {
|
||||
let source = r#"
|
||||
impl AdmittedTx {
|
||||
pub(crate) async fn admit(mut tx: Tx, community: CommunityId) -> Result<Self> {
|
||||
guard.guard_transaction(&mut tx, community).await?;
|
||||
Ok(Self { tx, community })
|
||||
}
|
||||
kept
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self {
|
||||
Self { tx, community }
|
||||
}
|
||||
}
|
||||
"#;
|
||||
let unadmitted: Vec<String> = admitted_tx_constructors(source)
|
||||
.into_iter()
|
||||
.filter(|(_, admits)| !admits)
|
||||
.map(|(function, _)| {
|
||||
function
|
||||
.lines()
|
||||
.next()
|
||||
.unwrap_or_default()
|
||||
.trim()
|
||||
.to_string()
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(
|
||||
unadmitted,
|
||||
["pub(crate) fn forge_for_tests(tx: Tx, community: CommunityId) -> Self {"],
|
||||
"an indented `#[cfg(test)]` forge must still be reported as unadmitted"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1405,3 +1418,519 @@ fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() {
|
||||
"guarded-table scan must exercise at least one file that writes a fenced table"
|
||||
);
|
||||
}
|
||||
|
||||
// Rolled-back verification probes write `events` rows that never commit, so
|
||||
// they owe no push job or TTL refresh.
|
||||
const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 2] = [
|
||||
"pub async fn verify_floor_guard_behavior(",
|
||||
"pub async fn verify_channel_roster_fence_behavior(",
|
||||
];
|
||||
|
||||
/// `source` with every `#[cfg(test)]` item removed, wherever it sits in the
|
||||
/// file. Each item runs to the `;` that ends it or to the brace that closes
|
||||
/// its body. Braces inside comments and literals are not counted (see
|
||||
/// [`mask_comments_and_literals`]). Unlike cutting the file at the first
|
||||
/// `#[cfg(test)]`, this keeps production code that follows a test-only item.
|
||||
/// The whole file is masked once and the marker is found in that view, so a
|
||||
/// mention in a comment or string (even a multiline one) strips nothing. The
|
||||
/// marker also counts only when it is the first code on its line.
|
||||
fn strip_cfg_test_items(source: &str) -> String {
|
||||
const MARKER: &str = "#[cfg(test)]";
|
||||
let masked = mask_comments_and_literals(source);
|
||||
let mut kept = String::with_capacity(source.len());
|
||||
let mut copied = 0usize;
|
||||
let mut search = 0usize;
|
||||
while let Some(found) = masked[search..].find(MARKER) {
|
||||
let at = search + found;
|
||||
let line_start = masked[..at].rfind('\n').map_or(0, |newline| newline + 1);
|
||||
search = at + MARKER.len();
|
||||
if !masked[line_start..at].trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
kept.push_str(&source[copied..at]);
|
||||
copied = search + cfg_test_item_len(&masked[search..]);
|
||||
search = copied;
|
||||
}
|
||||
kept.push_str(&source[copied..]);
|
||||
kept
|
||||
}
|
||||
|
||||
/// Length of the item that starts `masked_item`, which must already be masked
|
||||
/// so braces in comments and literals are not counted.
|
||||
fn cfg_test_item_len(masked_item: &str) -> usize {
|
||||
let mut depth = 0usize;
|
||||
for (index, byte) in masked_item.bytes().enumerate() {
|
||||
match byte {
|
||||
b';' if depth == 0 => return index + 1,
|
||||
b'{' => depth += 1,
|
||||
b'}' => {
|
||||
depth = depth.saturating_sub(1);
|
||||
if depth == 0 {
|
||||
return index + 1;
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
masked_item.len()
|
||||
}
|
||||
|
||||
/// Whether the `r` at `index` begins a raw literal token: `r"`, or the `br"` /
|
||||
/// `cr"` raw byte and C strings, rather than ending an identifier such as `for"`.
|
||||
fn raw_literal_prefix_starts_token(bytes: &[u8], index: usize) -> bool {
|
||||
let is_ident = |at: Option<usize>| {
|
||||
at.and_then(|at| bytes.get(at))
|
||||
.is_some_and(|byte| byte.is_ascii_alphanumeric() || *byte == b'_')
|
||||
};
|
||||
let before = index.checked_sub(1);
|
||||
!is_ident(before)
|
||||
|| (matches!(before.map(|at| bytes[at]), Some(b'b' | b'c'))
|
||||
&& !is_ident(index.checked_sub(2)))
|
||||
}
|
||||
|
||||
/// `source` with every comment (`//`, `///`, nested `/* */`) and every string,
|
||||
/// raw string, and char literal replaced by spaces, byte for byte. Newlines are
|
||||
/// kept, so offsets and line structure match `source`. Scans that look for
|
||||
/// code (braces, calls) read the masked text; scans that look for SQL read
|
||||
/// the original, because the SQL lives inside string literals.
|
||||
fn mask_comments_and_literals(source: &str) -> String {
|
||||
let bytes = source.as_bytes();
|
||||
let mut masked = bytes.to_vec();
|
||||
let mut blank = |from: usize, to: usize| {
|
||||
for byte in &mut masked[from..to.min(bytes.len())] {
|
||||
if *byte != b'\n' {
|
||||
*byte = b' ';
|
||||
}
|
||||
}
|
||||
};
|
||||
let mut index = 0usize;
|
||||
while index < bytes.len() {
|
||||
let start = index;
|
||||
match bytes[index] {
|
||||
b'/' if bytes.get(index + 1) == Some(&b'/') => {
|
||||
while index < bytes.len() && bytes[index] != b'\n' {
|
||||
index += 1;
|
||||
}
|
||||
}
|
||||
b'/' if bytes.get(index + 1) == Some(&b'*') => {
|
||||
let mut depth = 0usize;
|
||||
while index < bytes.len() {
|
||||
if bytes[index] == b'/' && bytes.get(index + 1) == Some(&b'*') {
|
||||
depth += 1;
|
||||
index += 2;
|
||||
} else if bytes[index] == b'*' && bytes.get(index + 1) == Some(&b'/') {
|
||||
depth -= 1;
|
||||
index += 2;
|
||||
if depth == 0 {
|
||||
break;
|
||||
}
|
||||
} else {
|
||||
index += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
b'r' if matches!(bytes.get(index + 1), Some(b'"' | b'#'))
|
||||
&& raw_literal_prefix_starts_token(bytes, index)
|
||||
&& bytes.get(
|
||||
index
|
||||
+ 1
|
||||
+ bytes[index + 1..]
|
||||
.iter()
|
||||
.take_while(|byte| **byte == b'#')
|
||||
.count(),
|
||||
) == Some(&b'"') =>
|
||||
{
|
||||
let hashes = bytes[index + 1..]
|
||||
.iter()
|
||||
.take_while(|byte| **byte == b'#')
|
||||
.count();
|
||||
let close = format!("\"{}", "#".repeat(hashes));
|
||||
let body = index + 2 + hashes;
|
||||
index = source[body..]
|
||||
.find(&close)
|
||||
.map_or(bytes.len(), |offset| body + offset + close.len());
|
||||
}
|
||||
b'"' => {
|
||||
index += 1;
|
||||
while index < bytes.len() && bytes[index] != b'"' {
|
||||
index += if bytes[index] == b'\\' { 2 } else { 1 };
|
||||
}
|
||||
index += 1;
|
||||
}
|
||||
b'\'' if char_literal_end(source, index).is_some() => {
|
||||
index = char_literal_end(source, index).expect("guard matched");
|
||||
}
|
||||
_ => {
|
||||
index += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
blank(start, index);
|
||||
}
|
||||
String::from_utf8(masked).expect("masking replaces whole ASCII-delimited spans")
|
||||
}
|
||||
|
||||
/// The end (exclusive) of the char literal whose opening quote is at `start`,
|
||||
/// such as `'{'`, `'é'`, `'\''`, `'\x7f'`, or `'\u{1F600}'`. `None` for a
|
||||
/// lifetime, which has no closing quote.
|
||||
fn char_literal_end(source: &str, start: usize) -> Option<usize> {
|
||||
// `start` holds an ASCII quote, so `start + 1` is a char boundary.
|
||||
let rest = &source[start + 1..];
|
||||
let body = if let Some(escape) = rest.strip_prefix('\\') {
|
||||
// The longest escape is `\u{10FFFF}`: a backslash and 9 more bytes.
|
||||
// Skip the escaped byte so `'\''` does not close on its own quote.
|
||||
let close = escape.get(1..)?.find('\'')? + 1;
|
||||
(close <= 9).then_some(1 + close)?
|
||||
} else {
|
||||
rest.chars().next()?.len_utf8()
|
||||
};
|
||||
rest[body..]
|
||||
.starts_with('\'')
|
||||
.then_some(start + 1 + body + 1)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn masker_blanks_every_literal_form_and_keeps_code() {
|
||||
for (input, expected) in [
|
||||
// Raw strings with zero, one, or several `#`, and the `b`/`c` raw
|
||||
// prefixes. Each body ends in `\`, which would escape the closing quote
|
||||
// if the literal were scanned as a plain string.
|
||||
(r#"a(r"\") {}"#, "a( ) {}"),
|
||||
(r###"a(r##"x"#{"##) {}"###, "a( ) {}"),
|
||||
(r##"a(br#"\"#) {}"##, "a(b ) {}"),
|
||||
(r#"a(br"\") {}"#, "a(b ) {}"),
|
||||
(r#"a(cr"\") {}"#, "a(c ) {}"),
|
||||
// Plain byte and C strings.
|
||||
(r#"a(b"{") {}"#, "a(b ) {}"),
|
||||
(r#"a(c"{") {}"#, "a(c ) {}"),
|
||||
// Plain and escaped char literals.
|
||||
(
|
||||
r"a('\'', '\\', '\n', '{') {}",
|
||||
"a( , , , ) {}",
|
||||
),
|
||||
// Multi-byte and long-escape char literals. Each holds a brace, or
|
||||
// hides one from a scan that misreads the literal's length.
|
||||
("a('é', '{') {}", "a( , ) {}"),
|
||||
(r"a('\u{1F600}', '{') {}", "a( , ) {}"),
|
||||
(r"a('\x7f', '{') {}", "a( , ) {}"),
|
||||
// Code that only looks like a literal stays as it is.
|
||||
("let r#type = 1; {}", "let r#type = 1; {}"),
|
||||
("fn f<'a>(x: &'a str) {}", "fn f<'a>(x: &'a str) {}"),
|
||||
] {
|
||||
assert_eq!(
|
||||
mask_comments_and_literals(input),
|
||||
expected,
|
||||
"masking `{input}`"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `source` inserts into `events` itself (not `event_mentions` or any
|
||||
/// other `events_*` table), in any letter case, across line breaks and string
|
||||
/// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.`
|
||||
/// schema prefix.
|
||||
fn inserts_events_row(source: &str) -> bool {
|
||||
let mut normalized = String::with_capacity(source.len());
|
||||
let mut chars = source.chars().peekable();
|
||||
while let Some(ch) = chars.next() {
|
||||
if ch.is_whitespace() || ch == '\\' {
|
||||
if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) {
|
||||
chars.next();
|
||||
}
|
||||
if !normalized.ends_with(' ') {
|
||||
normalized.push(' ');
|
||||
}
|
||||
} else {
|
||||
normalized.push(ch.to_ascii_lowercase());
|
||||
}
|
||||
}
|
||||
["insert into events", "insert into public.events"]
|
||||
.iter()
|
||||
.any(|marker| {
|
||||
normalized.match_indices(marker).any(|(index, marker)| {
|
||||
normalized[index + marker.len()..]
|
||||
.chars()
|
||||
.next()
|
||||
.is_none_or(|next| !(next.is_ascii_alphanumeric() || next == '_'))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/// An events insert must run both follow-ups: the push enqueue and the
|
||||
/// pre-commit TTL refresh. `after_admitted_insert` does both; a writer that
|
||||
/// inserts inside a savepoint calls the two halves separately.
|
||||
///
|
||||
/// Only real calls count: a hook in a comment or a string literal does not.
|
||||
fn runs_event_follow_ups(function_source: &str) -> bool {
|
||||
let code = mask_comments_and_literals(function_source);
|
||||
code.contains("event_follow_up::after_admitted_insert(")
|
||||
|| (code.contains("event_follow_up::enqueue_push_match(")
|
||||
&& code.contains(".record_channel_event("))
|
||||
}
|
||||
|
||||
fn event_insert_follow_up_violations(production_source: &str) -> Vec<String> {
|
||||
function_slices(production_source)
|
||||
.into_iter()
|
||||
.filter(|function_source| {
|
||||
let header = function_header(function_source);
|
||||
inserts_events_row(function_source)
|
||||
&& !EVENT_INSERT_FOLLOW_UP_EXCEPTIONS
|
||||
.iter()
|
||||
.any(|exception| header.starts_with(exception))
|
||||
&& !runs_event_follow_ups(function_source)
|
||||
})
|
||||
.map(|function_source| function_header(function_source).to_owned())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_insert_follow_up_policy_rejects_writers_without_the_hook() {
|
||||
let source = r##"
|
||||
pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events \
|
||||
(community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
||||
.await
|
||||
.expect("enqueue");
|
||||
}
|
||||
pub(crate) async fn hooked_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
||||
.await
|
||||
.expect("follow up");
|
||||
}
|
||||
pub(crate) async fn savepoint_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
||||
.await
|
||||
.expect("enqueue");
|
||||
tx.record_channel_event(channel, kind);
|
||||
}
|
||||
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("probe");
|
||||
}
|
||||
pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("insert into
|
||||
public.events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO \
|
||||
events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
/// Kept at module scope, not `#[cfg(test)]`, because production calls it.
|
||||
pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
// see `#[cfg(test)] mod x {`
|
||||
pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO\nevents (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
// crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
||||
}
|
||||
pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
/* crate::store::event_follow_up::enqueue_push_match(tx.conn(), community, id, kind)
|
||||
/* nested */ tx.record_channel_event(channel, kind); */
|
||||
}
|
||||
pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
tracing::debug!("skipped event_follow_up::after_admitted_insert(tx, ..)");
|
||||
}
|
||||
pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
/* /* inner */ crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel) */
|
||||
}
|
||||
pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
tracing::debug!("\"event_follow_up::after_admitted_insert(\"");
|
||||
}
|
||||
pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
tracing::debug!(r#""event_follow_up::after_admitted_insert(""#);
|
||||
}
|
||||
/*
|
||||
#[cfg(test)]
|
||||
*/
|
||||
pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) {
|
||||
let _example = r#"
|
||||
fn example() {
|
||||
crate::store::event_follow_up::after_admitted_insert(tx, id, kind, channel)
|
||||
}
|
||||
"#;
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
#[cfg(test)]
|
||||
fn test_only_helper() -> (&'static str, char, &'static str, char, &'static [u8]) {
|
||||
// an unbalanced { in a comment
|
||||
("{", '{', r#"}"} {"#, '"', br#"\"#)
|
||||
}
|
||||
pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(tx.conn())
|
||||
.await
|
||||
.expect("write");
|
||||
}
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
async fn test_writer(pool: &PgPool) {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)")
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("fixture");
|
||||
}
|
||||
}
|
||||
"##;
|
||||
|
||||
assert_eq!(
|
||||
event_insert_follow_up_violations(&strip_cfg_test_items(source)),
|
||||
[
|
||||
"pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_under_doc_mention(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_under_brace_mention(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_escaped_newline(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_commented_out_hook(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_block_commented_hook(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_hook_in_string(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_nested_block_commented_hook(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_hook_after_escaped_quote(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_hook_in_raw_string(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_after_block_commented_cfg_test(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_after_raw_string_example(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_after_test_helper(tx: &mut AdmittedTx) {",
|
||||
],
|
||||
"an events insert must run the push enqueue and record the TTL refresh"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_production_event_insert_runs_the_follow_up_hook() {
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
fn collect_rs_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
||||
for entry in std::fs::read_dir(dir).expect("read source directory") {
|
||||
let path = entry.expect("read directory entry").path();
|
||||
if path.is_dir() {
|
||||
collect_rs_files(&path, out);
|
||||
} else if should_scan_guarded_write_source_file(&path) {
|
||||
out.push(path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let crates_root = Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.parent()
|
||||
.expect("crate lives under crates/");
|
||||
let mut files = Vec::new();
|
||||
for entry in std::fs::read_dir(crates_root).expect("read crates directory") {
|
||||
let src = entry.expect("read crates entry").path().join("src");
|
||||
if src.is_dir() {
|
||||
collect_rs_files(&src, &mut files);
|
||||
}
|
||||
}
|
||||
|
||||
let mut hooked_writers = 0usize;
|
||||
let mut violations = Vec::new();
|
||||
for path in files {
|
||||
let source = std::fs::read_to_string(&path).expect("read source file");
|
||||
let production = strip_cfg_test_items(&source);
|
||||
if !inserts_events_row(&production) {
|
||||
continue;
|
||||
}
|
||||
for function_source in function_slices(&production) {
|
||||
if inserts_events_row(function_source) && runs_event_follow_ups(function_source) {
|
||||
hooked_writers += 1;
|
||||
}
|
||||
}
|
||||
violations.extend(
|
||||
event_insert_follow_up_violations(&production)
|
||||
.into_iter()
|
||||
.map(|header| format!("{}: {header}", path.display())),
|
||||
);
|
||||
}
|
||||
|
||||
assert!(
|
||||
violations.is_empty(),
|
||||
"production `INSERT INTO events` writers must call `event_follow_up` so the push \
|
||||
enqueue and channel TTL refresh survive the trigger retirement: {violations:?}"
|
||||
);
|
||||
// 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.
|
||||
assert!(
|
||||
hooked_writers >= 7,
|
||||
"expected at least seven hooked event writers, found {hooked_writers}"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user