From 32e38d5e86f223a1073329a296fc853c14566699 Mon Sep 17 00:00:00 2001
From: Konstantin Knizhnik <knizhnik@neon.tech>
Date: Wed, 17 Jan 2024 12:35:59 +0200
Subject: [PATCH] Limit max replication lag to prevent blocking GC

---
 libs/safekeeper_api/src/models.rs |  1 +
 pageserver/src/tenant/timeline.rs | 10 ++++++----
 safekeeper/src/send_wal.rs        | 14 +++++++++++---
 3 files changed, 18 insertions(+), 7 deletions(-)

diff --git a/libs/safekeeper_api/src/models.rs b/libs/safekeeper_api/src/models.rs
index d7e8dc5f9e..17f78e65d5 100644
--- a/libs/safekeeper_api/src/models.rs
+++ b/libs/safekeeper_api/src/models.rs
@@ -50,6 +50,7 @@ pub struct SkTimelineInfo {
     pub safekeeper_connstr: Option<String>,
     #[serde(default)]
     pub http_connstr: Option<String>,
+    // Minimum of all active RO replicas flush LSN
     #[serde(default = "lsn_invalid")]
     pub standby_flush_lsn: Lsn,
 }
diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs
index cc2ef9a5d0..2336836612 100644
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -4201,11 +4201,13 @@ impl Timeline {
             (horizon_cutoff, pitr_cutoff, retain_lsns)
         };
 
-        let mut new_gc_cutoff = Lsn::min(horizon_cutoff, pitr_cutoff);
+        let new_gc_cutoff = Lsn::min(horizon_cutoff, pitr_cutoff);
         let standby_flush_lsn = self.standby_flush_lsn.load();
-        if standby_flush_lsn != Lsn::INVALID && standby_flush_lsn < new_gc_cutoff {
-            new_gc_cutoff = standby_flush_lsn;
-        }
+        let new_gc_cutoff = if standby_flush_lsn != Lsn::INVALID {
+            Lsn::min(standby_flush_lsn, new_gc_cutoff)
+        } else {
+            new_gc_cutoff
+        };
 
         let res = self
             .gc_timeline(horizon_cutoff, pitr_cutoff, retain_lsns, new_gc_cutoff)
diff --git a/safekeeper/src/send_wal.rs b/safekeeper/src/send_wal.rs
index 33bd0fd8f6..3778a40b2f 100644
--- a/safekeeper/src/send_wal.rs
+++ b/safekeeper/src/send_wal.rs
@@ -37,6 +37,8 @@ const STANDBY_STATUS_UPDATE_TAG_BYTE: u8 = b'r';
 // neon extension of replication protocol
 const NEON_STATUS_UPDATE_TAG_BYTE: u8 = b'z';
 
+const MAX_STANDBY_LAG: u64 = 1024 * 1024 * 1024; // 1Gb
+
 type FullTransactionId = u64;
 
 /// Hot standby feedback received from replica
@@ -268,21 +270,27 @@ impl WalSendersShared {
                     agg.ts = min(agg.ts, hs_feedback.ts);
                 }
                 let reply = standby_feedback.reply;
-                if reply.write_lsn != Lsn::INVALID {
+                if reply.write_lsn != Lsn::INVALID
+                    && self.agg_ps_feedback.last_received_lsn < reply.write_lsn + MAX_STANDBY_LAG
+                {
                     if reply_agg.write_lsn != Lsn::INVALID {
                         reply_agg.write_lsn = Lsn::min(reply_agg.write_lsn, reply.write_lsn);
                     } else {
                         reply_agg.write_lsn = reply.write_lsn;
                     }
                 }
-                if reply.flush_lsn != Lsn::INVALID {
+                if reply.flush_lsn != Lsn::INVALID
+                    && self.agg_ps_feedback.last_received_lsn < reply.flush_lsn + MAX_STANDBY_LAG
+                {
                     if reply_agg.flush_lsn != Lsn::INVALID {
                         reply_agg.flush_lsn = Lsn::min(reply_agg.flush_lsn, reply.flush_lsn);
                     } else {
                         reply_agg.flush_lsn = reply.flush_lsn;
                     }
                 }
-                if reply.apply_lsn != Lsn::INVALID {
+                if reply.apply_lsn != Lsn::INVALID
+                    && self.agg_ps_feedback.last_received_lsn < reply.apply_lsn + MAX_STANDBY_LAG
+                {
                     if reply_agg.apply_lsn != Lsn::INVALID {
                         reply_agg.apply_lsn = Lsn::min(reply_agg.apply_lsn, reply.apply_lsn);
                     } else {