From 2c65438cff718905342b02f473a3b6c6831f9a1b Mon Sep 17 00:00:00 2001 From: Rian Date: Fri, 31 Jul 2026 18:03:38 -0300 Subject: [PATCH] fix(db): label routes replica_is_writer when READ_DATABASE_URL is not a standby Pointing READ_DATABASE_URL at the writer was accepted silently, and route telemetry then reported decision="replica" for pages the writer served. The freshness fence is trivially satisfied in that configuration -- the "replica" has zero lag because it IS the writer -- so nothing is unsafe, but buzz_db_route_decision becomes semantically misleading: operators watching the rollout dashboards read decision="replica" as the offload percentage and would believe replica offload is working while every byte still comes off the writer. The boot ping now reads cluster identity on the reader connection it already holds and settles a cached verdict. Pages the reader serves are labeled decision="replica_is_writer" while that verdict stands, so decision="replica" stays a truthful offload signal. The reason label (fresh/covered) is unchanged. Warn-only by design. Refusing the reader pool would trade a reporting defect for lost capacity, and it would latch the wrong answer on Aurora, where cluster-ro legitimately resolves to the writer while a cluster has no readers and starts resolving to a real reader as soon as one is added. The gate is pg_is_in_recovery() on the reader rather than an identifier match: system_identifier comes from pg_control_system(), which is superuser-only by default, so requiring a match would silently no-op wherever the relay's role is not a superuser. The identifier is read when permitted and reported in the warning to separate "the same node as the writer" from "some other primary", both of which offload the writer just as little. Refs #3644 Signed-off-by: Rian --- crates/buzz-db/src/lib.rs | 240 ++++++++++++++++++++++++++-- crates/buzz-db/src/replica_fence.rs | 139 ++++++++++++++++ 2 files changed, 369 insertions(+), 10 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 245e49bb2d..621ef7c67f 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -209,6 +209,25 @@ pub struct Db { /// not yet probed (or the probe hit a transient error and will retry). /// Shared across `Db` clones. pub(crate) reader_aurora_identity: std::sync::Arc>, + /// Whether the configured reader endpoint is not a hot standby — probed + /// once at boot ([`Db::spawn_read_pool_boot_ping`]) and cached. Unset + /// means not yet probed (no replica configured, the boot ping could not + /// reach the reader, or identity could not be read). + /// + /// Named for the `replica_is_writer` route label it drives, which is the + /// overwhelmingly common cause (a reader URL that resolves to the writer, + /// including Aurora `cluster-ro` on a cluster with no readers). The gate + /// is strictly "not replaying WAL", so an unrelated primary lands in the + /// same bucket — also correctly, since it offloads the writer just as + /// little. + /// + /// `Some(true)` only relabels route telemetry + /// (`buzz_db_route_decision{decision="replica_is_writer"}`); it never + /// gates routing. A reader that is really the writer serves correct rows + /// with a trivially satisfied fence — the defect is that + /// `decision="replica"` would report offload that is not happening. See + /// [`Db::replica_decision`]. Shared across `Db` clones. + pub(crate) reader_is_writer: std::sync::Arc>, } /// The session that served (or will serve) a routed read, so follow-up @@ -666,6 +685,7 @@ impl Db { fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), replica_read_max_age, reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()), + reader_is_writer: std::sync::Arc::new(std::sync::OnceLock::new()), }) } @@ -745,11 +765,16 @@ impl Db { /// [`Db::reader_aurora_capability_on`]. Prime failure is fine: the routed /// path re-probes on the connection it already holds, so a failed prime /// costs a round trip rather than a second acquire budget. + /// + /// It also settles [`Db::reader_is_writer`] on that same connection, so a + /// reader endpoint that is really the writer is called out in the boot log + /// and in route telemetry (see [`Db::prime_reader_is_writer`]). pub fn spawn_read_pool_boot_ping(&self) { let Some(read_pool) = self.read_pool.clone() else { return; }; let aurora_identity = self.reader_aurora_identity.clone(); + let db = self.clone(); tokio::spawn(async move { match read_pool.acquire().await { Ok(mut conn) => { @@ -763,6 +788,7 @@ impl Db { "aurora identity boot prime failed; first routed read will probe" ), } + db.prime_reader_is_writer(&mut conn).await; } Err(e) => tracing::warn!( "read replica unreachable at boot; serving all-writer until it recovers: {e}" @@ -781,6 +807,7 @@ impl Db { fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), replica_read_max_age: None, reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()), + reader_is_writer: std::sync::Arc::new(std::sync::OnceLock::new()), } } @@ -800,6 +827,7 @@ impl Db { fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), replica_read_max_age: None, reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()), + reader_is_writer: std::sync::Arc::new(std::sync::OnceLock::new()), } } @@ -996,6 +1024,86 @@ impl Db { } } + /// Settle [`Db::reader_is_writer`] from the cluster identity of a reader + /// connection the caller already holds, and WARN when the reader endpoint + /// is not a hot standby. + /// + /// The gate is `pg_is_in_recovery()` on the reader, not an identifier + /// match: a reader that is not replaying WAL is not a replica, and that is + /// true whether it is the writer itself or an unrelated primary. The + /// identifier comparison only sharpens the log line, and is `None` for a + /// role without `EXECUTE` on `pg_control_system()` (see + /// [`replica_fence::ClusterIdentity`]). + /// + /// Deliberately warn-only. `READ_DATABASE_URL` pointing at the writer + /// serves correct rows — the freshness fence is trivially satisfied + /// because the "replica" *is* the writer — so refusing the reader pool + /// would trade a reporting defect for lost capacity. It would also latch + /// the wrong answer for Aurora, where `cluster-ro` legitimately resolves + /// to the writer while a cluster has no readers and starts resolving to a + /// real reader the moment one is added. + /// + /// Boot-only, and only on the boot ping's successful path: a reader that + /// is unreachable at boot leaves the verdict unset, and routes are + /// labeled `replica` as before. Re-probing per request would put a + /// privileged catalog call on the read path to serve a dashboard. + async fn prime_reader_is_writer(&self, conn: &mut sqlx::PgConnection) { + if self.reader_is_writer.get().is_some() { + return; + } + let reader = match replica_fence::cluster_identity(conn).await { + Ok(identity) => identity, + Err(e) => { + tracing::debug!( + error = %e, + "reader cluster identity unreadable; route labels stay unqualified" + ); + return; + } + }; + if reader.in_recovery { + let _ = self.reader_is_writer.set(false); + return; + } + let writer = match self.pool.acquire().await { + Ok(mut writer_conn) => replica_fence::cluster_identity(&mut writer_conn).await.ok(), + Err(e) => { + tracing::debug!(error = %e, "writer cluster identity unreadable for comparison"); + None + } + }; + // Not in recovery ⇒ not a standby, whatever the identifiers say. The + // comparison distinguishes "the same node the writer uses" from "some + // other primary", which is the difference between a copy-pasted URL + // and a wrong host. + let same_cluster = writer.as_ref().and_then(|w| reader.same_cluster_as(w)); + let _ = self.reader_is_writer.set(true); + tracing::warn!( + reader_system_identifier = ?reader.system_identifier, + writer_system_identifier = ?writer.as_ref().and_then(|w| w.system_identifier.clone()), + same_cluster = ?same_cluster, + "READ_DATABASE_URL is not a hot standby (pg_is_in_recovery() = false); \ + reads routed to it are NOT offloading the writer. Route telemetry reports \ + decision=\"replica_is_writer\" instead of \"replica\" until this is fixed." + ); + } + + /// The `decision` label for a page the reader pool actually served. + /// + /// `replica_is_writer` once the boot identity probe proved the reader + /// endpoint is not a standby ([`Db::prime_reader_is_writer`]), so + /// `decision="replica"` stays a truthful offload signal: the misleading + /// state this separates out is a "replica" whose every byte still comes + /// off the writer. The `reason` label is unchanged — the predicate verdict + /// (`fresh`/`covered`) is orthogonal to where the page came from. + fn replica_decision(&self) -> &'static str { + if self.reader_is_writer.get() == Some(&true) { + "replica_is_writer" + } else { + "replica" + } + } + /// Record one route decision (Rev 2 observability): which path, where it /// went, and why. fn record_route(path: &'static str, decision: &'static str, reason: &'static str) { @@ -1667,7 +1775,7 @@ impl Db { RouteDecision::Replica(mut tx, _entry, reason) => { match event::query_events_on(&mut tx, q).await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -1700,7 +1808,7 @@ impl Db { RouteDecision::Replica(mut tx, _entry, reason) => { match event::query_events_on(&mut tx, q).await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -1737,7 +1845,7 @@ impl Db { RouteDecision::Replica(mut tx, _entry, reason) => { match event::count_events_on(&mut tx, q).await { Ok(count) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(count) } Err(e) => { @@ -1898,7 +2006,7 @@ impl Db { RouteDecision::Replica(mut tx, _entry, reason) => { match event::get_events_by_ids_on(&mut tx, community_id, ids).await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -2732,7 +2840,7 @@ impl Db { Ok(replies) => { if cursor.is_none() { // Predicate A: bounded-stale head page, served as proved. - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); return Ok(replies); } let full = replies.len() >= limit as usize; @@ -2740,7 +2848,7 @@ impl Db { .last() .is_some_and(|tail| tail.created_at <= entry.fence_wall); if full && below_fence { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); return Ok(replies); } // Candidate terminal page, or page reaching above the @@ -2856,7 +2964,7 @@ impl Db { .await { Ok(window) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); return Ok(( window, ReadSession { @@ -3179,7 +3287,7 @@ impl Db { .await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -3256,7 +3364,7 @@ impl Db { .await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -3322,7 +3430,7 @@ impl Db { .await { Ok(events) => { - Self::record_route(path, "replica", reason); + Self::record_route(path, self.replica_decision(), reason); Ok(events) } Err(e) => { @@ -7568,6 +7676,118 @@ mod tests { drop_scratch_db(&admin, db.pool.clone(), &wname).await; } + /// A `READ_DATABASE_URL` that resolves to the writer must be labeled + /// `replica_is_writer`, never plain `replica`. + /// + /// This is the reporting defect the identity probe exists for: the reader + /// is the writer, so the freshness fence is trivially satisfied (zero lag + /// by construction) and every routed read reports as served-by-replica + /// while every byte still comes off the writer. Nothing serves wrong rows + /// — the damage is that `decision="replica"` is the offload percentage on + /// the rollout dashboards. + /// + /// Both URLs point at the SAME scratch database, which *is* the + /// misconfiguration rather than a stand-in for it. The probe is called + /// directly instead of through [`Db::spawn_read_pool_boot_ping`] so the + /// assertions cannot race a spawned task, and the label is asserted + /// BEFORE probing too: a test that only checked the post-probe label + /// would still pass if the constructor prejudged every reader. + #[tokio::test(flavor = "current_thread")] + #[ignore = "requires Postgres"] + async fn reader_pointing_at_the_writer_is_labeled_replica_is_writer() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (seed, wname) = create_scratch_db(&admin, "reader_is_writer").await; + seed.close().await; + let base = admin_url().await; + let scratch_url = { + let idx = base.rfind('/').expect("db url has a path segment"); + format!("{}/{}", &base[..idx], wname) + }; + + let mut db = Db::new(&DbConfig { + database_url: scratch_url.clone(), + // The misconfiguration under test: the "replica" is the writer. + read_database_url: Some(scratch_url), + max_connections: 4, + read_max_connections: Some(4), + ..DbConfig::default() + }) + .await + .expect("connect Db whose reader URL is the writer"); + db.fence().force_open_for_tests(chrono::Utc::now()); + db.set_replica_read_max_age_for_tests(Some(Duration::from_secs(5))); + + assert_eq!( + db.replica_decision(), + "replica", + "an unprobed reader must not be prejudged as the writer" + ); + + let read_pool = db.read_pool.clone().expect("reader pool configured"); + let mut conn = read_pool.acquire().await.expect("reader connection"); + db.prime_reader_is_writer(&mut conn).await; + drop(conn); + + assert_eq!( + db.reader_is_writer.get(), + Some(&true), + "a reader that is not in recovery must settle the verdict as true" + ); + assert_eq!( + db.replica_decision(), + "replica_is_writer", + "the probed verdict must reach the route label" + ); + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let query = EventQuery::for_community(CommunityId::from_uuid(Uuid::new_v4())); + let count = { + let _guard = metrics::set_default_local_recorder(&recorder); + db.count_events_routed("reader_is_writer_probe", &query) + .await + } + .expect("the reader still answers the read"); + assert_eq!(count, 0, "empty scratch database"); + + let labels: std::collections::HashMap<(String, String), u64> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_db_route_decision") + .map(|(key, _, _, value)| { + let metrics_util::debugging::DebugValue::Counter(n) = value else { + panic!("buzz_db_route_decision must be a counter"); + }; + let keyed: Vec<_> = key.key().labels().collect(); + let get = |name: &str| { + keyed + .iter() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + .unwrap_or_default() + }; + ((get("decision"), get("reason")), n) + }) + .collect(); + + assert_eq!( + labels.get(&("replica_is_writer".to_owned(), "fresh".to_owned())), + Some(&1), + "a page served by a writer-backed reader must be labeled \ + replica_is_writer/fresh; got {labels:?}" + ); + assert!( + !labels.keys().any(|(decision, _)| decision == "replica"), + "no plain replica label may be emitted, or the offload percentage \ + still counts writer-served pages; got {labels:?}" + ); + + drop_scratch_db(&admin, db.pool.clone(), &wname).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn routed_reads_are_confined_to_the_requested_community() { diff --git a/crates/buzz-db/src/replica_fence.rs b/crates/buzz-db/src/replica_fence.rs index c840db393e..c0e132ab09 100644 --- a/crates/buzz-db/src/replica_fence.rs +++ b/crates/buzz-db/src/replica_fence.rs @@ -711,6 +711,75 @@ pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result< } } +/// Cluster identity of one session: the evidence that separates a real hot +/// standby from a reader endpoint that is really the writer. +/// +/// `in_recovery` is the load-bearing half — a session that is not replaying +/// WAL is not on a replica, whatever else it is. `system_identifier` adds the +/// diagnosis an operator needs to act: an identifier equal to the writer's +/// means `READ_DATABASE_URL` resolves into the same cluster as +/// `DATABASE_URL` (the copy-paste misconfiguration, and the Aurora +/// `cluster-ro`-with-no-readers case), while a differing one means the reader +/// is a *separate* primary — same misleading `replica` telemetry, entirely +/// different remediation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterIdentity { + /// `pg_control_system().system_identifier` rendered as text: it is only + /// ever compared or logged, never arithmetic. + /// + /// `None` when the connected role lacks `EXECUTE` on the control + /// function — it is superuser-only by default and Buzz does not require + /// the relay's role to be a superuser. Identity is evidence, never a + /// routing gate, so an unreadable identifier degrades the diagnosis + /// rather than failing the probe. + pub system_identifier: Option, + /// `pg_is_in_recovery()`: `true` on a hot standby, `false` on a primary. + /// Readable by every role, so this half is always present. + pub in_recovery: bool, +} + +impl ClusterIdentity { + /// Whether this session is on the same cluster as `writer`. + /// + /// `None` when either side's identifier is unreadable: absence of the + /// privileged identifier is not evidence of a *distinct* cluster, and + /// reporting "different" there would invent a diagnosis. Callers gate on + /// [`ClusterIdentity::in_recovery`] and treat this purely as message + /// detail. + pub fn same_cluster_as(&self, writer: &ClusterIdentity) -> Option { + match (&self.system_identifier, &writer.system_identifier) { + (Some(reader), Some(writer)) => Some(reader == writer), + _ => None, + } + } +} + +/// Read the [`ClusterIdentity`] of the session on `conn`. +/// +/// Two statements rather than one: a single SELECT of both values would fail +/// whole on `insufficient_privilege` (42501), losing the recovery flag that +/// every role can read — and the recovery flag is the half the caller +/// actually gates on. +pub async fn cluster_identity(conn: &mut PgConnection) -> Result { + let in_recovery: bool = sqlx::query_scalar("SELECT pg_is_in_recovery()") + .fetch_one(&mut *conn) + .await?; + let system_identifier = match sqlx::query_scalar::<_, String>( + "SELECT system_identifier::text FROM pg_control_system()", + ) + .fetch_one(&mut *conn) + .await + { + Ok(identifier) => Some(identifier), + Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("42501") => None, + Err(e) => return Err(e), + }; + Ok(ClusterIdentity { + system_identifier, + in_recovery, + }) +} + /// Observe the heartbeat on a specific reader session — the /// connection-local half of the proof. Returns the observed token/epoch /// plus the backend identity of the session for route-decision evidence: @@ -1082,6 +1151,76 @@ mod tests { assert_eq!(one, 1); } + /// Identifier comparison is a diagnosis aid, so an unreadable identifier + /// must report "unknown", never "different" — a role without `EXECUTE` on + /// `pg_control_system()` would otherwise be reported as a distinct + /// cluster on every deployment that grants the relay less than superuser. + #[test] + fn cluster_identity_compares_only_when_both_identifiers_are_readable() { + let with = |identifier: Option<&str>| ClusterIdentity { + system_identifier: identifier.map(str::to_owned), + in_recovery: false, + }; + assert_eq!( + with(Some("7")).same_cluster_as(&with(Some("7"))), + Some(true), + "equal identifiers are the same cluster" + ); + assert_eq!( + with(Some("7")).same_cluster_as(&with(Some("8"))), + Some(false), + "differing identifiers are different clusters" + ); + assert_eq!( + with(None).same_cluster_as(&with(Some("7"))), + None, + "an unreadable reader identifier is unknown, not different" + ); + assert_eq!( + with(Some("7")).same_cluster_as(&with(None)), + None, + "an unreadable writer identifier is unknown, not different" + ); + } + + /// A primary must report `in_recovery = false`, and reading identity must + /// not depend on the role holding `EXECUTE` on the privileged control + /// function: the probe returns `Ok` either way, with the identifier + /// present or `None`. Comparing a session against itself is the + /// same-cluster case the boot probe warns on. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cluster_identity_reads_a_primary_with_or_without_control_privileges() { + let pool = PgPool::connect(&test_db_url()).await.expect("connect"); + let mut conn = pool.acquire().await.expect("conn"); + let identity = cluster_identity(&mut conn) + .await + .expect("identity must be readable on any role"); + assert!( + !identity.in_recovery, + "the dev database is a primary, not a standby" + ); + let self_comparison = identity.same_cluster_as(&identity); + match &identity.system_identifier { + Some(_) => assert_eq!( + self_comparison, + Some(true), + "a readable identifier must match itself" + ), + None => assert_eq!( + self_comparison, None, + "an unreadable identifier must stay unknown rather than guess" + ), + } + // A privilege error must be swallowed as `None`, not left poisoning + // the session for the statements the boot ping runs next. + let one: i32 = sqlx::query_scalar("SELECT 1") + .fetch_one(&mut *conn) + .await + .expect("connection usable after identity probe"); + assert_eq!(one, 1); + } + /// End-to-end probe against a real database: each probe commits a /// strictly newer token, records a retained entry, and a session on the /// same database observes a token/epoch that resolves that entry.