fix(mirror): make read-consistency staleness check deterministic

Thread now_ms in from the caller instead of reading the wall clock
inside check_mirror_read_consistency, so the BoundedStaleness lag
comparison is a pure function of its inputs. This removes a race
between lag injection and the staleness check that could flake
boundary-condition tests on a slow runner.

Author: farhan-syah

nodedb/src/control/server/pgwire/ddl/database/mirror/read.rs

@@ -20,8 +20,6 @@
 //!   This is the lowest-latency option and is correct for use cases that
 //!   accept CRDT-style monotonic convergence (e.g. mobile/edge workloads).
 
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
-
 use nodedb_types::error::sqlstate;
 use nodedb_types::{DatabaseId, MirrorOrigin, MirrorStatus};
 
@@ -52,11 +50,17 @@ pub enum MirrorReadOutcome {
 /// `promoted` is true when the mirror has been promoted and the database
 /// is now a writable primary — in that case all consistency levels are
 /// served locally (no longer a mirror).
+/// `now_ms` is the current wall-clock time (ms since epoch), supplied by the
+/// caller rather than read internally so the staleness comparison is a pure
+/// function of its inputs — this keeps `BoundedStaleness` boundary tests
+/// deterministic instead of racing the wall clock between lag injection and
+/// the check.
 pub fn check_mirror_read_consistency(
     catalog: &SystemCatalog,
     mirror_db_id: DatabaseId,
     mirror_origin: &MirrorOrigin,
     consistency: ReadConsistency,
+    now_ms: u64,
 ) -> MirrorReadOutcome {
     // A promoted mirror is a normal writable database — serve all reads locally.
     if matches!(mirror_origin.status, MirrorStatus::Promoted) {
@@ -106,11 +110,6 @@ pub fn check_mirror_read_consistency(
                 }
             };
 
-            let now_ms = SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap_or(Duration::ZERO)
-                .as_millis() as u64;
-
             let actual_lag_ms = now_ms.saturating_sub(last_apply_ms);
             let max_lag_ms = max_lag.as_millis() as u64;
 
@@ -176,7 +175,13 @@ mod tests {
         let db_id = DatabaseId::new(1024);
         let origin = sample_origin(MirrorStatus::Following);
 
-        match check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong) {
+        match check_mirror_read_consistency(
+            &catalog,
+            db_id,
+            &origin,
+            ReadConsistency::Strong,
+            now_ms(),
+        ) {
             MirrorReadOutcome::Reject {
                 sqlstate_code,
                 message,
@@ -198,7 +203,13 @@ mod tests {
         let db_id = DatabaseId::new(1025);
         let origin = sample_origin(MirrorStatus::Following);
 
-        match check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Eventual) {
+        match check_mirror_read_consistency(
+            &catalog,
+            db_id,
+            &origin,
+            ReadConsistency::Eventual,
+            now_ms(),
+        ) {
             MirrorReadOutcome::ServeLocally => {}
             MirrorReadOutcome::Reject { message, .. } => {
                 panic!("Eventual should serve locally, got reject: {message}")
@@ -226,7 +237,7 @@ mod tests {
             .unwrap();
 
         let bound = ReadConsistency::BoundedStaleness(Duration::from_secs(10));
-        match check_mirror_read_consistency(&catalog, db_id, &origin, bound) {
+        match check_mirror_read_consistency(&catalog, db_id, &origin, bound, now_ms()) {
             MirrorReadOutcome::ServeLocally => {}
             MirrorReadOutcome::Reject { message, .. } => {
                 panic!("Fresh mirror should serve locally, got reject: {message}")
@@ -255,7 +266,7 @@ mod tests {
 
         // Bound = 5 seconds; actual lag ≈ 60 seconds.
         let bound = ReadConsistency::BoundedStaleness(Duration::from_secs(5));
-        match check_mirror_read_consistency(&catalog, db_id, &origin, bound) {
+        match check_mirror_read_consistency(&catalog, db_id, &origin, bound, now_ms()) {
             MirrorReadOutcome::Reject {
                 sqlstate_code,
                 message,
@@ -282,7 +293,7 @@ mod tests {
             ReadConsistency::Eventual,
             ReadConsistency::BoundedStaleness(Duration::from_secs(1)),
         ] {
-            match check_mirror_read_consistency(&catalog, db_id, &origin, consistency) {
+            match check_mirror_read_consistency(&catalog, db_id, &origin, consistency, now_ms()) {
                 MirrorReadOutcome::ServeLocally => {}
                 MirrorReadOutcome::Reject { message, .. } => {
                     panic!("Promoted mirror should always serve locally, got: {message}")

nodedb/src/control/server/pgwire/handler/dispatch.rs

@@ -159,11 +159,16 @@ impl NodeDbPgHandler {
             // Consistency defaults to Strong: mirrors are not the source leader,
             // so reads are rejected unless the session has explicitly opted into
             // BoundedStaleness or Eventual.
+            let now_ms = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or(std::time::Duration::ZERO)
+                .as_millis() as u64;
             let outcome = check_mirror_read_consistency(
                 catalog,
                 task.database_id,
                 origin,
                 ReadConsistency::Strong,
+                now_ms,
             );
             if let MirrorReadOutcome::Reject { message, .. } = outcome {
                 return Err(crate::Error::StaleReadNotLeader {

nodedb/tests/mirror/test_eventual_read_serves_local.rs

@@ -14,7 +14,7 @@ use nodedb::control::server::pgwire::ddl::database::{
 };
 use nodedb::types::ReadConsistency;
 
-use super::helpers::{TEST_SOURCE_CLUSTER, inject_lag_record_for_id, open_tmp_catalog};
+use super::helpers::{TEST_SOURCE_CLUSTER, inject_lag_record_for_id, now_ms, open_tmp_catalog};
 
 #[test]
 fn mirror_eventual_read_serves_local() {
@@ -35,7 +35,13 @@ fn mirror_eventual_read_serves_local() {
     };
 
     // Eventual always serves locally regardless of lag.
-    match check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Eventual) {
+    match check_mirror_read_consistency(
+        &catalog,
+        db_id,
+        &origin,
+        ReadConsistency::Eventual,
+        now_ms(),
+    ) {
         MirrorReadOutcome::ServeLocally => {}
         MirrorReadOutcome::Reject { message, .. } => {
             panic!("Eventual read must always serve locally, got reject: {message}");
@@ -57,7 +63,13 @@ fn mirror_eventual_serves_even_when_disconnected() {
         status: MirrorStatus::Disconnected,
     };
 
-    match check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Eventual) {
+    match check_mirror_read_consistency(
+        &catalog,
+        db_id,
+        &origin,
+        ReadConsistency::Eventual,
+        now_ms(),
+    ) {
         MirrorReadOutcome::ServeLocally => {}
         MirrorReadOutcome::Reject { message, .. } => {
             panic!("Eventual read on Disconnected mirror must still serve locally: {message}");

nodedb/tests/mirror/test_full_lifecycle.rs

@@ -124,6 +124,7 @@ fn mirror_full_lifecycle() {
         db_id,
         origin,
         ReadConsistency::BoundedStaleness(Duration::from_secs(5)),
+        now_ms(),
     ) {
         MirrorReadOutcome::ServeLocally => {}
         MirrorReadOutcome::Reject { message, .. } => {
@@ -159,7 +160,7 @@ fn mirror_full_lifecycle() {
         ReadConsistency::Eventual,
         ReadConsistency::BoundedStaleness(Duration::from_millis(1)),
     ] {
-        match check_mirror_read_consistency(&catalog, db_id, origin, consistency) {
+        match check_mirror_read_consistency(&catalog, db_id, origin, consistency, now_ms()) {
             MirrorReadOutcome::ServeLocally => {}
             MirrorReadOutcome::Reject { message, .. } => {
                 panic!("Promoted mirror must serve all reads locally: {message}");

nodedb/tests/mirror/test_simulated_latency.rs

@@ -16,7 +16,7 @@
 
 use std::time::Duration;
 
-use nodedb_types::{DatabaseId, Lsn, MirrorMode, MirrorOrigin, MirrorStatus};
+use nodedb_types::{DatabaseId, Lsn, MirrorLagRecord, MirrorMode, MirrorOrigin, MirrorStatus};
 use tempfile::TempDir;
 
 use nodedb::control::server::pgwire::ddl::database::{
@@ -51,6 +51,7 @@ fn mirror_simulated_cross_region_latency() {
         db_id,
         &origin,
         ReadConsistency::BoundedStaleness(Duration::from_millis(1_000)),
+        now_ms(),
     ) {
         MirrorReadOutcome::ServeLocally => {}
         MirrorReadOutcome::Reject { message, .. } => {
@@ -64,6 +65,7 @@ fn mirror_simulated_cross_region_latency() {
         db_id,
         &origin,
         ReadConsistency::BoundedStaleness(Duration::from_millis(100)),
+        now_ms(),
     ) {
         MirrorReadOutcome::Reject { sqlstate_code, .. } => {
             assert_eq!(
@@ -93,9 +95,21 @@ fn mirror_simulated_latency_at_threshold_boundary() {
     let catalog = open_tmp_catalog(&dir);
     let db_id = DatabaseId::new(5002);
 
-    // Simulate RTT exactly at the BoundedStaleness window: 199 ms lag,
-    // 200 ms bound — must pass.
-    inject_lag_record_for_id(&catalog, db_id, 199, 1);
+    // Fix `now` and derive `last_apply_ms` from it so the computed lag is
+    // exactly 199 ms — one millisecond inside the 200 ms bound — regardless of
+    // how long lag injection takes. Passing the same `now` into the check keeps
+    // this boundary deterministic; reading the wall clock inside the check would
+    // add real elapsed time to the lag and flake the assertion on a slow runner.
+    let now = now_ms();
+    catalog
+        .put_mirror_lag(
+            db_id,
+            &MirrorLagRecord {
+                last_applied_lsn: Lsn::new(1),
+                last_apply_ms: now.saturating_sub(199),
+            },
+        )
+        .expect("inject lag record");
 
     let origin = MirrorOrigin {
         source_cluster: TEST_SOURCE_CLUSTER.to_string(),
@@ -110,6 +124,7 @@ fn mirror_simulated_latency_at_threshold_boundary() {
         db_id,
         &origin,
         ReadConsistency::BoundedStaleness(Duration::from_millis(200)),
+        now,
     ) {
         MirrorReadOutcome::ServeLocally => {}
         MirrorReadOutcome::Reject { message, .. } => {

nodedb/tests/mirror/test_strong_read_rejected.rs

@@ -14,7 +14,7 @@ use nodedb::control::server::pgwire::ddl::database::{
 };
 use nodedb::types::ReadConsistency;
 
-use super::helpers::{TEST_SOURCE_CLUSTER, open_tmp_catalog};
+use super::helpers::{TEST_SOURCE_CLUSTER, now_ms, open_tmp_catalog};
 
 fn following_origin() -> MirrorOrigin {
     MirrorOrigin {
@@ -33,7 +33,8 @@ fn mirror_strong_read_rejected() {
     let db_id = DatabaseId::new(1001);
     let origin = following_origin();
 
-    let outcome = check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong);
+    let outcome =
+        check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong, now_ms());
 
     match outcome {
         MirrorReadOutcome::Reject {
@@ -69,7 +70,8 @@ fn mirror_strong_read_rejected_degraded() {
         status: MirrorStatus::Degraded { lag_ms: 8_000 },
     };
 
-    let outcome = check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong);
+    let outcome =
+        check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong, now_ms());
 
     match outcome {
         MirrorReadOutcome::Reject { sqlstate_code, .. } => {
@@ -97,7 +99,8 @@ fn mirror_strong_read_rejected_disconnected() {
         status: MirrorStatus::Disconnected,
     };
 
-    let outcome = check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong);
+    let outcome =
+        check_mirror_read_consistency(&catalog, db_id, &origin, ReadConsistency::Strong, now_ms());
 
     match outcome {
         MirrorReadOutcome::Reject { sqlstate_code, .. } => {