feat: pgdog.min_lsn read-your-writes replica routing floor

.schema/pgdog.schema.json

@@ -71,6 +71,7 @@
         "lsn_check_delay": 9223372036854775807,
         "lsn_check_interval": 5000,
         "lsn_check_timeout": 5000,
+        "min_lsn_primary_fallback": false,
         "min_pool_size": 1,
         "mirror_exposure": 1.0,
         "mirror_queue": 128,
@@ -892,6 +893,11 @@
           "default": 5000,
           "minimum": 0
         },
+        "min_lsn_primary_fallback": {
+          "description": "When a read sets `pgdog.min_lsn` and no replica has replayed that far, fall\nback to the primary instead of erroring.\n\n_Default:_ `false`\n\nhttps://docs.pgdog.dev/configuration/pgdog.toml/general/#min_lsn_primary_fallback",
+          "type": "boolean",
+          "default": false
+        },
         "min_pool_size": {
           "description": "Default minimum number of connections per database pool to keep open at all times.\n\n_Default:_ `1`\n\nhttps://docs.pgdog.dev/configuration/pgdog.toml/general/#min_pool_size",
           "type": "integer",

integration/load_balancer/pgx/min_lsn_test.go

@@ -0,0 +1,96 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func primaryConn(t *testing.T) *pgx.Conn {
+	conn, err := pgx.Connect(context.Background(), "postgres://postgres:postgres@127.0.0.1:45000/postgres?sslmode=disable")
+	require.NoError(t, err)
+	return conn
+}
+
+// `postgres` uses exclude_primary, so reads normally go to replicas. With
+// pgdog.min_lsn set, reads must go only to a replica that has replayed past it,
+// falling back to the primary when none has.
+func pgdogConn(t *testing.T) *pgx.Conn {
+	conn, err := pgx.Connect(context.Background(), "postgres://postgres:postgres@127.0.0.1:6432/postgres?sslmode=disable")
+	require.NoError(t, err)
+	return conn
+}
+
+func replicaTotal(table string) int64 {
+	var total int64
+	for _, call := range LoadStatsForReplicas(table) {
+		total += call.Calls
+	}
+	return total
+}
+
+func TestMinLsnRoutesReads(t *testing.T) {
+	p := primaryConn(t)
+	_, err := p.Exec(context.Background(),
+		"CREATE TABLE IF NOT EXISTS lb_minlsn_ok (id BIGINT); CREATE TABLE IF NOT EXISTS lb_minlsn_far (id BIGINT)")
+	require.NoError(t, err)
+	p.Close(context.Background())
+	defer func() {
+		p := primaryConn(t)
+		p.Exec(context.Background(), "DROP TABLE IF EXISTS lb_minlsn_ok; DROP TABLE IF EXISTS lb_minlsn_far")
+		p.Close(context.Background())
+	}()
+
+	// Let the replicas catch up and the LSN monitor (lsn_check_interval=1s) populate
+	// each pool's replay LSN, so lsn_stats.valid() is true.
+	time.Sleep(3 * time.Second)
+
+	const reads = 20
+
+	// Case A: min_lsn already satisfied -> reads served by replicas, not primary.
+	ResetStats()
+	a := pgdogConn(t)
+	_, err = a.Exec(context.Background(), "SET pgdog.min_lsn = '0/0'")
+	require.NoError(t, err)
+	for i := range reads {
+		_, err := a.Exec(context.Background(), "SELECT * FROM lb_minlsn_ok WHERE id = $1", int64(i))
+		assert.NoError(t, err)
+	}
+	a.Close(context.Background())
+
+	okReplicas := replicaTotal("lb_minlsn_ok")
+	okPrimary := LoadStatsForPrimary("lb_minlsn_ok").Calls
+	t.Logf("min_lsn=0/0: replicas=%d primary=%d", okReplicas, okPrimary)
+	assert.Equal(t, int64(reads), okReplicas, "satisfied min_lsn must be served by replicas")
+	assert.Equal(t, int64(0), okPrimary, "satisfied min_lsn must not touch the primary")
+
+	// Case B: min_lsn far in the future -> no replica caught up. Default behavior
+	// (min_lsn_primary_fallback off) is to error, not serve stale or hit the primary.
+	ResetStats()
+	b := pgdogConn(t)
+	_, err = b.Exec(context.Background(), "SET pgdog.min_lsn = '7FFFFFFF/FFFFFFFF'")
+	require.NoError(t, err)
+	_, readErr := b.Exec(context.Background(), "SELECT * FROM lb_minlsn_far WHERE id = $1", int64(1))
+	b.Close(context.Background())
+
+	farReplicas := replicaTotal("lb_minlsn_far")
+	farPrimary := LoadStatsForPrimary("lb_minlsn_far").Calls
+	t.Logf("min_lsn=max: err=%v replicas=%d primary=%d", readErr, farReplicas, farPrimary)
+	assert.Error(t, readErr, "unmet min_lsn must error, not serve a stale read")
+	// Pin the wire message: clients (and the glow relation_writer gate) match on
+	// this exact text to recognize the rejection and defer, so it must not drift.
+	if readErr != nil {
+		assert.Contains(t, readErr.Error(), "no replica caught up to the requested min_lsn",
+			"unmet min_lsn must surface PgDog's NoReplicaCaughtUp message to the client")
+		// The message also carries a catch-up ETA (eta ~Ns) so the client can size
+		// its retry/defer to the real deficit instead of a fixed backoff.
+		assert.Regexp(t, `eta ~\d+s`, readErr.Error(),
+			"NoReplicaCaughtUp must report the catch-up ETA")
+	}
+	assert.Equal(t, int64(0), farReplicas, "unmet min_lsn must not be served by a behind replica")
+	assert.Equal(t, int64(0), farPrimary, "unmet min_lsn must not hit the primary when fallback is off")
+}

pgdog-config/src/general.rs

@@ -221,6 +221,15 @@ pub struct General {
     #[serde(default)]
     pub read_write_split: ReadWriteSplit,
 
+    /// When a read sets `pgdog.min_lsn` and no replica has replayed that far, fall
+    /// back to the primary instead of erroring.
+    ///
+    /// _Default:_ `false`
+    ///
+    /// https://docs.pgdog.dev/configuration/pgdog.toml/general/#min_lsn_primary_fallback
+    #[serde(default)]
+    pub min_lsn_primary_fallback: bool,
+
     /// Path to the TLS certificate PgDog will use to setup TLS connections with clients.
     ///
     /// https://docs.pgdog.dev/configuration/pgdog.toml/general/#tls_certificate
@@ -829,6 +838,7 @@ impl Default for General {
             load_balancing_strategy: Self::load_balancing_strategy(),
             read_write_strategy: Self::read_write_strategy(),
             read_write_split: Self::read_write_split(),
+            min_lsn_primary_fallback: false,
             tls_certificate: Self::tls_certificate(),
             tls_private_key: Self::tls_private_key(),
             tls_client_required: bool::default(),

pgdog/src/backend/error.rs

@@ -160,10 +160,20 @@ impl Error {
             Error::Pool(PoolError::CheckoutTimeout) => true,
             Error::Pool(PoolError::AllReplicasDown) => true,
             Error::Pool(PoolError::Banned) => true,
+            Error::Pool(PoolError::NoReplicaCaughtUp { .. }) => true,
             _ => false,
         }
     }
 
+    /// Expected read-your-writes backpressure: the `pgdog.min_lsn` floor isn't
+    /// met yet. The client receives SQLSTATE 58000 and defers/retries, so this
+    /// is normal control flow, not a server fault; callers log it at debug and
+    /// do not count it as a pool error.
+    pub fn is_no_replica_caught_up(&self) -> bool {
+        use crate::backend::pool::Error as PoolError;
+        matches!(self, Error::Pool(PoolError::NoReplicaCaughtUp { .. }))
+    }
+
     /// Transient network/pool fault worth retrying.
     pub fn is_retryable(&self) -> bool {
         match self {

pgdog/src/backend/pool/error.rs

@@ -76,6 +76,13 @@ pub enum Error {
 
     #[error("replica lag")]
     ReplicaLag,
+
+    // `eta_seconds`: how long until the soonest replica reaches `min_lsn`,
+    // derived from its replication lag in time; `0` means not estimable. Clients
+    // size their read-your-writes defer from it. The message text is part of the
+    // wire contract (clients match on it), so keep it stable.
+    #[error("no replica caught up to the requested min_lsn (eta ~{eta_seconds}s)")]
+    NoReplicaCaughtUp { eta_seconds: i64 },
 }
 
 impl Error {
@@ -98,6 +105,10 @@ impl Error {
                 | Self::UntrackedConnCheckin(_)
                 // Deliberate shutdown.
                 | Self::FastShutdown
+                // Read-your-writes backpressure: no replica has reached the
+                // requested min_lsn. An immediate retry can't help; the client
+                // must wait out the reported ETA.
+                | Self::NoReplicaCaughtUp { .. }
         )
     }
 }
@@ -133,5 +144,6 @@ mod tests {
         assert!(!Error::PubSubDisabled.is_retryable());
         assert!(!Error::FastShutdown.is_retryable());
         assert!(!Error::NoShard(0).is_retryable());
+        assert!(!Error::NoReplicaCaughtUp { eta_seconds: 5 }.is_retryable());
     }
 }

pgdog/src/backend/pool/lb/mod.rs

@@ -29,6 +29,10 @@ pub use ban::UnbanReason;
 use monitor::*;
 pub use target_health::*;
 
+/// A real catch-up estimate is floored at 1s; `0` is reserved for "not
+/// estimable", which tells the client to fall back to its own default backoff.
+const MIN_ETA_SECONDS: i64 = 1;
+
 #[cfg(test)]
 mod test;
 
@@ -86,6 +90,9 @@ pub struct LoadBalancer {
     pub(super) role_detection: Arc<Notify>,
     /// Read/write split.
     pub(super) rw_split: ReadWriteSplit,
+    /// Fall back to the primary when no replica satisfies a read's `min_lsn`
+    /// (instead of erroring).
+    pub(super) min_lsn_primary_fallback: bool,
 }
 
 impl LoadBalancer {
@@ -126,9 +133,16 @@ impl LoadBalancer {
             maintenance: Arc::new(Notify::new()),
             role_detection: Arc::new(Notify::new()),
             rw_split,
+            min_lsn_primary_fallback: false,
         }
     }
 
+    /// Fall back to the primary when no replica satisfies a read's `min_lsn`.
+    pub fn with_min_lsn_primary_fallback(mut self, fallback: bool) -> Self {
+        self.min_lsn_primary_fallback = fallback;
+        self
+    }
+
     /// Get the primary pool, if configured.
     pub fn primary(&self) -> Option<&Pool> {
         self.primary_target().map(|target| &target.pool)
@@ -370,6 +384,41 @@ impl LoadBalancer {
             return Err(Error::AllReplicasDown);
         }
 
+        // Read-your-writes floor: keep only replicas that have replayed at least
+        // `min_lsn` (replay LSN is monotonic and the cached value is <= actual, so
+        // this never serves a stale read). If none qualify, fall back to the
+        // primary when `min_lsn_primary_fallback` is set, else error with an ETA.
+        if let Some(min_lsn) = request.min_lsn {
+            let caught_up: Vec<&Target> = candidates
+                .iter()
+                .copied()
+                .filter(|target| {
+                    let stats = target.pool.lsn_stats();
+                    stats.valid() && stats.lsn.lsn >= min_lsn
+                })
+                .collect();
+
+            if !caught_up.is_empty() {
+                candidates = caught_up;
+            } else if let Some(primary) = self
+                .min_lsn_primary_fallback
+                .then(|| self.primary_target())
+                .flatten()
+            {
+                candidates = vec![primary];
+            } else {
+                // ETA until the soonest replica reaches the floor, so the client
+                // can size its defer; `0` = not estimable (no valid LSN stats).
+                let eta_seconds = candidates
+                    .iter()
+                    .filter_map(|target| target.pool.lsn_stats().eta_to(min_lsn))
+                    .min()
+                    .map(|eta| (eta.as_secs() as i64).max(MIN_ETA_SECONDS))
+                    .unwrap_or(0);
+                return Err(Error::NoReplicaCaughtUp { eta_seconds });
+            }
+        }
+
         match self.lb_strategy {
             Random => candidates.shuffle(&mut rand::rng()),
             RoundRobin => {

pgdog/src/backend/pool/lb/test.rs

@@ -2031,3 +2031,23 @@ async fn test_params_returns_all_replicas_down_when_empty() {
         "params() should return AllReplicasDown when no targets exist"
     );
 }
+
+#[tokio::test]
+async fn test_min_lsn_errors_when_no_replica_caught_up() {
+    // Two replicas with default (invalid, lsn 0) stats: nothing is caught up.
+    let replicas = LoadBalancer::new(
+        &None,
+        &[
+            create_test_pool_config("127.0.0.1", 5432),
+            create_test_pool_config("localhost", 5432),
+        ],
+        LoadBalancingStrategy::Random,
+        ReadWriteSplit::IncludePrimary,
+    );
+
+    // Fallback is off by default, so an unmet min_lsn must error before any checkout.
+    let request = Request::default().with_min_lsn(Some(100));
+    let result = replicas.get(&request).await;
+
+    assert!(matches!(result, Err(Error::NoReplicaCaughtUp { .. })));
+}