mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
fix(relay): delete the startup NIP-33 d_tag backfill (#8468)
## Summary
Delete the NIP-33 `d_tag` backfill that ran on every relay boot.
On every boot, before the listener opened, the relay ran:
```sql
SELECT DISTINCT community_id FROM events
WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL
```
No index covers `d_tag IS NULL`. The only `d_tag` index is the partial
`idx_events_parameterized`, which is `WHERE d_tag IS NOT NULL`. So this
query scanned every partition of `events`. In production it averaged
about 11s per boot, max 19s
(`buzz_startup_phase_seconds{phase="d_tag_backfill"}`), and found
nothing.
These rows shouldn't exist anymore:
- The backfill was the one-time migration for #246, the change that
added `d_tag`.
- In `crates/buzz-db/src`, every production write path sets the column
for kinds 30000–39999 through `extract_d_tag`: the main insert,
`replaceable.rs`, `channel_members.rs`, `relay_members.rs`, and
`push.rs`. A new source test,
`every_production_event_insert_names_d_tag`, now enforces that.
- The `events` inserts that omit `d_tag` are all test fixtures.
`scripts/backfill-d-tag.sql` is still there as the manual fallback for
any database that predates #246 and somehow never booted a relay since.
### What got simpler
- Removed `Db::backfill_d_tags` and its admission test.
- Removed the source-shape assertion in `observability_source.rs`.
- Removed the `d_tag_backfill` startup phase.
- `StartupStep::ALL` is now a slice, so removing a step no longer means
editing a hand-kept array length.
### Notes
- The `buzz_startup_phase_seconds{phase="d_tag_backfill"}` series stops
reporting.
- **Rebased onto main after #8467 merged.** Both remove a startup phase.
With the fixed-length `ALL` array the two merged cleanly in git into a
17-entry `[Self; 18]`, which doesn't compile. This PR makes `ALL` a
slice.
- **Before merging**, run this read-only query against production and
confirm it returns `0`: `SELECT count(*) FROM events WHERE kind BETWEEN
30000 AND 39999 AND d_tag IS NULL`. It scans every partition, so use a
read replica. If it isn't `0`, run `scripts/backfill-d-tag.sql` first.
The cutover script copies legacy `d_tag` values as-is, so only this
query proves historical data is clean.
- **Self-hosted installs** should run the same check before upgrading.
### Related issue
None found. Companion to #8467, which also targets relay startup time.
### Testing
- New `every_production_event_insert_names_d_tag`
(`crates/buzz-db/tests/observability_source.rs`): every production
`INSERT INTO events` must name `d_tag` in its column list. Dropping
`d_tag` from `push.rs` makes it fail and name the function.
- Runtime E2E against a real relay: boot has no `d_tag_backfill` phase;
kind:30023 replacement via WebSocket and HTTP keeps only the newer
event, an unrelated `d` survives, and an older replay can't win
(confirmed in SQL).
- Stacked on #8467 at `501dfd83`: a fresh two-replica boot runs all
enabled startup phases, with neither `nip43_reconcile` nor
`d_tag_backfill` present. Both replicas pass readiness and return the
same signed message over authenticated HTTP and WebSocket, and outsiders
are denied.
- Blox, at `501dfd83`: `just fmt-check clippy security-review-check
file-size-check` passes. Every `test-unit` step and `just test` pass
except the `buzz-acp` `git_bootstrap` native-git tests, which fail the
same way at base `b6aa5539` on Blox, and one `buzz-agent` catalog timing
test that fails only under full-suite load and passes when run alone.
- `e27139de` adds only a doc comment on
`every_production_event_insert_names_d_tag`, saying which SQL shapes the
text match covers. No code changed, so the results above still apply.
---------
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
@@ -1003,41 +1003,6 @@ async fn fleet_sweep_returns_committed_results_when_database_drops_mid_page() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn d_tag_backfill_admits_active_tenant_and_skips_quiescing_tenant() {
|
||||
let (db, communities) = fixture().await;
|
||||
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
|
||||
for (index, community) in communities[..2].iter().enumerate() {
|
||||
sqlx::query(
|
||||
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig) \
|
||||
VALUES ($1, $2, $3, to_timestamp(1), 30023, '[[\"d\",\"tag\"]]'::jsonb, '', $4)",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(vec![index as u8; 32])
|
||||
.bind(vec![2_u8; 32])
|
||||
.bind(vec![3_u8; 64])
|
||||
.execute(&db.pool)
|
||||
.await
|
||||
.expect("seed addressable event");
|
||||
}
|
||||
assert_eq!(db.backfill_d_tags().await.unwrap(), 1);
|
||||
let active: Option<String> =
|
||||
sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1")
|
||||
.bind(communities[0].as_uuid())
|
||||
.fetch_one(&db.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let quiescing: Option<String> =
|
||||
sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1")
|
||||
.bind(communities[1].as_uuid())
|
||||
.fetch_one(&db.pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(active.as_deref(), Some("tag"));
|
||||
assert_eq!(quiescing, None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn match_claim_admits_active_tenant_and_skips_quiescing_tenant() {
|
||||
|
||||
@@ -2776,53 +2776,6 @@ impl Db {
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`.
|
||||
///
|
||||
/// Idempotent — safe to call on every startup. No-ops when all rows are already populated.
|
||||
/// Admits each community before updating its NIP-33 rows with NULL d_tag.
|
||||
#[datastore_span(name = "backfill_d_tags", system = "postgresql")]
|
||||
pub async fn backfill_d_tags(&self) -> Result<u64> {
|
||||
let mut connection = crate::observability::acquire_writer(
|
||||
&self.pool,
|
||||
crate::observability::WriterOperation::Bootstrap,
|
||||
)
|
||||
.await?;
|
||||
let communities: Vec<Uuid> = sqlx::query_scalar(
|
||||
"SELECT DISTINCT community_id FROM events \
|
||||
WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL",
|
||||
)
|
||||
.fetch_all(&mut *connection)
|
||||
.await?;
|
||||
drop(connection);
|
||||
let mut updated = 0;
|
||||
for id in communities {
|
||||
let Some(mut tx) = crate::begin_community_sweep_transaction(
|
||||
&self.pool,
|
||||
CommunityId::from_uuid(id),
|
||||
crate::observability::WriterOperation::Bootstrap,
|
||||
)
|
||||
.await?
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let result = sqlx::query(
|
||||
"UPDATE events \
|
||||
SET d_tag = COALESCE( \
|
||||
(SELECT elem->>1 FROM jsonb_array_elements(tags) AS elem \
|
||||
WHERE elem->>0 = 'd' LIMIT 1), \
|
||||
'' \
|
||||
) \
|
||||
WHERE community_id = $1 AND kind BETWEEN 30000 AND 39999 AND d_tag IS NULL",
|
||||
)
|
||||
.bind(id)
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
updated += result.rows_affected();
|
||||
}
|
||||
Ok(updated)
|
||||
}
|
||||
|
||||
/// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey.
|
||||
#[datastore_span(name = "soft_delete_discovery_events", system = "postgresql")]
|
||||
pub async fn soft_delete_discovery_events(
|
||||
|
||||
@@ -156,14 +156,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() {
|
||||
assert!(event.contains("WriterOperation::Authorization"));
|
||||
assert!(event.contains("WriterOperation::SubscriptionHistory"));
|
||||
assert!(event.contains("ReaderOperation::SubscriptionHistory"));
|
||||
let backfill_d_tags = event
|
||||
.split_once("pub async fn backfill_d_tags")
|
||||
.expect("event store must expose the startup d-tag backfill")
|
||||
.1
|
||||
.split_once("/// Soft-delete NIP-29 discovery events")
|
||||
.expect("d-tag backfill must precede discovery deletion")
|
||||
.0;
|
||||
assert!(backfill_d_tags.contains("WriterOperation::Bootstrap"));
|
||||
let soft_delete_discovery = event
|
||||
.split_once("pub async fn soft_delete_discovery_events")
|
||||
.expect("event store must expose discovery-event deletion")
|
||||
@@ -1633,6 +1625,12 @@ fn masker_blanks_every_literal_form_and_keeps_code() {
|
||||
/// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.`
|
||||
/// schema prefix.
|
||||
fn inserts_events_row(source: &str) -> bool {
|
||||
!events_insert_tails(source).is_empty()
|
||||
}
|
||||
|
||||
/// `source` lowercased, with every run of whitespace, `\` line continuations,
|
||||
/// and `\n`/`\t`/`\r` escapes collapsed to one space.
|
||||
fn normalized_sql(source: &str) -> String {
|
||||
let mut normalized = String::with_capacity(source.len());
|
||||
let mut chars = source.chars().peekable();
|
||||
while let Some(ch) = chars.next() {
|
||||
@@ -1647,16 +1645,26 @@ fn inserts_events_row(source: &str) -> bool {
|
||||
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 == '_'))
|
||||
})
|
||||
})
|
||||
normalized
|
||||
}
|
||||
|
||||
/// The normalized SQL that follows each `INSERT INTO events` in `source`.
|
||||
fn events_insert_tails(source: &str) -> Vec<String> {
|
||||
let normalized = normalized_sql(source);
|
||||
let mut tails = Vec::new();
|
||||
for marker in ["insert into events", "insert into public.events"] {
|
||||
for (index, marker) in normalized.match_indices(marker) {
|
||||
let tail = &normalized[index + marker.len()..];
|
||||
if tail
|
||||
.chars()
|
||||
.next()
|
||||
.is_none_or(|next| !(next.is_ascii_alphanumeric() || next == '_'))
|
||||
{
|
||||
tails.push(tail.to_owned());
|
||||
}
|
||||
}
|
||||
}
|
||||
tails
|
||||
}
|
||||
|
||||
/// An events insert must run both follow-ups: the push enqueue and the
|
||||
@@ -2062,3 +2070,124 @@ fn every_production_event_insert_runs_the_roster_guard() {
|
||||
"expected at least nine event writers, found {checked_writers}"
|
||||
);
|
||||
}
|
||||
|
||||
// The rolled-back replica-floor probe writes a kind 9 row, which has no `d`.
|
||||
const EVENT_INSERT_D_TAG_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("];
|
||||
|
||||
/// Whether an `INSERT INTO events` tail names `d_tag` in an explicit column
|
||||
/// list. An insert without a column list does not count.
|
||||
fn events_insert_names_d_tag(tail: &str) -> bool {
|
||||
tail.trim_start()
|
||||
.strip_prefix('(')
|
||||
.and_then(|columns| columns.split_once(')'))
|
||||
.is_some_and(|(columns, _)| columns.split(',').any(|column| column.trim() == "d_tag"))
|
||||
}
|
||||
|
||||
fn event_insert_d_tag_violations(production_source: &str) -> Vec<String> {
|
||||
function_slices(production_source)
|
||||
.into_iter()
|
||||
.filter(|function_source| {
|
||||
let header = function_header(function_source);
|
||||
!EVENT_INSERT_D_TAG_EXCEPTIONS
|
||||
.iter()
|
||||
.any(|exception| header.starts_with(exception))
|
||||
&& !events_insert_tails(function_source)
|
||||
.iter()
|
||||
.all(|tail| events_insert_names_d_tag(tail))
|
||||
})
|
||||
.map(|function_source| function_header(function_source).to_owned())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_insert_d_tag_policy_rejects_writers_without_the_column() {
|
||||
let source = r##"
|
||||
pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, $3)");
|
||||
}
|
||||
pub(crate) async fn writer_with_d_tag(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events \
|
||||
(community_id, id, kind, d_tag) VALUES ($1, $2, $3, $4)");
|
||||
}
|
||||
pub(crate) async fn writer_with_compact_columns(tx: &mut AdmittedTx) {
|
||||
sqlx::query("insert into public.events (community_id,id,kind,d_tag) VALUES ($1,$2,$3,$4)");
|
||||
}
|
||||
pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id, kind) SELECT community_id, id, d_tag FROM x");
|
||||
}
|
||||
pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id, kind, d_tag_hash) VALUES ($1, $2, $3, $4)");
|
||||
}
|
||||
pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events SELECT * FROM staged_events");
|
||||
}
|
||||
pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO events (community_id, id, d_tag) VALUES ($1, $2, $3)");
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
||||
}
|
||||
pub(crate) async fn mention_writer(tx: &mut AdmittedTx) {
|
||||
sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)");
|
||||
}
|
||||
pub async fn verify_floor_guard_behavior(pool: &PgPool) {
|
||||
sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, 9)");
|
||||
}
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
fn fixture() {
|
||||
sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)");
|
||||
}
|
||||
}
|
||||
"##;
|
||||
assert_eq!(
|
||||
event_insert_d_tag_violations(&strip_cfg_test_items(source)),
|
||||
[
|
||||
"pub(crate) async fn writer_without_d_tag(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_d_tag_only_in_values(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_with_lookalike_column(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn writer_without_column_list(tx: &mut AdmittedTx) {",
|
||||
"pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) {",
|
||||
],
|
||||
"every production events insert must name `d_tag` in its column list"
|
||||
);
|
||||
}
|
||||
|
||||
/// Addressable events (kinds 30000-39999) replace by `d_tag`, and the only
|
||||
/// index that serves that lookup is partial on `d_tag IS NOT NULL`. A writer
|
||||
/// that leaves the column out stores NULL, so its rows never replace. Nothing
|
||||
/// repairs them after the fact, so every production writer must name the column.
|
||||
///
|
||||
/// This is a text match, not a proof. It checks the column list of literal
|
||||
/// `INSERT INTO [public.]events (...)` SQL inside function bodies under each
|
||||
/// crate's `src/`. It does not see `ONLY`, quoted or `format!`-built table
|
||||
/// names, a comment between the keywords, or SQL held in a constant outside a
|
||||
/// function, and it does not check that the bound value is non-NULL. Exempt
|
||||
/// probes are skipped by function header, so every insert inside them is too.
|
||||
#[test]
|
||||
fn every_production_event_insert_names_d_tag() {
|
||||
let mut checked_writers = 0;
|
||||
let mut violations = Vec::new();
|
||||
for path in production_event_writer_source_files() {
|
||||
let source = std::fs::read_to_string(&path).expect("read source file");
|
||||
let production = strip_cfg_test_items(&source);
|
||||
checked_writers += function_slices(&production)
|
||||
.into_iter()
|
||||
.filter(|function| inserts_events_row(function))
|
||||
.count();
|
||||
violations.extend(
|
||||
event_insert_d_tag_violations(&production)
|
||||
.into_iter()
|
||||
.map(|header| format!("{}: {header}", path.display())),
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
violations.is_empty(),
|
||||
"production `INSERT INTO events` writers must name `d_tag` in their column list: \
|
||||
{violations:?}"
|
||||
);
|
||||
// The same writer set as the roster-guard scan. A lower count means the
|
||||
// scan stopped seeing a writer, not that one was removed safely.
|
||||
assert!(
|
||||
checked_writers >= 9,
|
||||
"expected at least nine event writers, found {checked_writers}"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -704,19 +704,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
|
||||
}
|
||||
step.finish();
|
||||
|
||||
// NIP-33: backfill d_tag for any existing parameterized replaceable events
|
||||
// that predate the column addition. Idempotent — no-ops when fully populated.
|
||||
let mut step = StepTimer::start(StartupStep::DTagBackfill);
|
||||
match db.backfill_d_tags().await {
|
||||
Ok(0) => {}
|
||||
Ok(n) => info!("Backfilled d_tag for {n} NIP-33 events"),
|
||||
Err(e) => {
|
||||
step.degrade();
|
||||
error!("Failed to backfill d_tags: {e}");
|
||||
}
|
||||
}
|
||||
step.finish();
|
||||
|
||||
let (audit, audit_metrics_pool) = if config.audit_enabled {
|
||||
let step = StepTimer::start(StartupStep::AuditConnect);
|
||||
let audit_pool = connect_audit_pool(&db_config)
|
||||
|
||||
@@ -51,8 +51,6 @@ pub enum StartupStep {
|
||||
ReplicaFenceProbe,
|
||||
/// Ensure the deployment community, backfill the allowlist, and bootstrap the owner.
|
||||
MembershipBootstrap,
|
||||
/// Backfill NIP-33 `d_tag` values.
|
||||
DTagBackfill,
|
||||
/// Connect the audit database pool (only when audit is enabled).
|
||||
AuditConnect,
|
||||
/// Connect Redis commands and pub/sub.
|
||||
@@ -79,14 +77,13 @@ pub enum StartupStep {
|
||||
|
||||
impl StartupStep {
|
||||
/// The complete `phase` label vocabulary.
|
||||
pub const ALL: [Self; 18] = [
|
||||
pub const ALL: &[Self] = &[
|
||||
Self::DbConnect,
|
||||
Self::DbMigrate,
|
||||
Self::PartitionEnsure,
|
||||
Self::DeletionFenceVerify,
|
||||
Self::ReplicaFenceProbe,
|
||||
Self::MembershipBootstrap,
|
||||
Self::DTagBackfill,
|
||||
Self::AuditConnect,
|
||||
Self::RedisConnect,
|
||||
Self::SearchConnect,
|
||||
@@ -109,7 +106,6 @@ impl StartupStep {
|
||||
Self::DeletionFenceVerify => "deletion_fence_verify",
|
||||
Self::ReplicaFenceProbe => "replica_fence_probe",
|
||||
Self::MembershipBootstrap => "membership_bootstrap",
|
||||
Self::DTagBackfill => "d_tag_backfill",
|
||||
Self::AuditConnect => "audit_connect",
|
||||
Self::RedisConnect => "redis_connect",
|
||||
Self::SearchConnect => "search_connect",
|
||||
|
||||
Reference in New Issue
Block a user