rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-10-29 23:07:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Limit max replication lag to prevent blocking GC

Browse Source
This commit is contained in:
Konstantin Knizhnik
2024-02-07 08:54:11 +02:00
parent 8cfc5ade57
commit 32e38d5e86
3 changed files with 18 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/safekeeper_api/src/models.rs
+1
View File
@@ -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,
}
pageserver/src/tenant/timeline.rs
+6 -4
View File
@@ -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)
safekeeper/src/send_wal.rs
+11 -3
View File
@@ -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 {