From 29ef9708fac47ff80e968b991377a91b5301b143 Mon Sep 17 00:00:00 2001
From: John Sully <john@csquare.ca>
Date: Sun, 18 Jul 2021 20:45:32 +0000
Subject: [PATCH] Return the ring buffer to its original size if we temporarily
 resized it

Former-commit-id: a12ce4a0d105bf7d6ccff95f7dc0044c4676b0a7
---
 src/config.cpp      |  1 +
 src/replication.cpp | 16 ++++++++++++++++
 src/server.cpp      |  2 ++
 src/server.h        |  3 +++
 4 files changed, 22 insertions(+)

diff --git a/src/config.cpp b/src/config.cpp
index a10cdbe12..6fc957485 100644
--- a/src/config.cpp
+++ b/src/config.cpp
@@ -2470,6 +2470,7 @@ static int updateReplBacklogSize(long long val, long long prev, const char **err
      * being able to tell when the size changes, so restore prev before calling it. */
     UNUSED(err);
     g_pserver->repl_backlog_size = prev;
+    g_pserver->repl_backlog_config_size = val;
     resizeReplicationBacklog(val);
     return 1;
 }
diff --git a/src/replication.cpp b/src/replication.cpp
index 7eabeef6b..c3b902db4 100644
--- a/src/replication.cpp
+++ b/src/replication.cpp
@@ -4321,6 +4321,8 @@ void replicationCron(void) {
     
     replicationStartPendingFork();
 
+    trimReplicationBacklog();
+
     /* Remove the RDB file used for replication if Redis is not running
      * with any persistence. */
     removeRDBUsedToSyncReplicas();
@@ -5066,3 +5068,17 @@ void updateFailoverStatus(void) {
             g_pserver->target_replica_port);
     }
 }
+
+// If we automatically grew the backlog we need to trim it back to
+//  the config setting when possible
+void trimReplicationBacklog() {
+    serverAssert(GlobalLocksAcquired());
+    serverAssert(g_pserver->repl_batch_offStart < 0);   // we shouldn't be in a batch
+    if (g_pserver->repl_backlog_size <= g_pserver->repl_backlog_config_size)
+        return; // We're already a good size
+    if (g_pserver->repl_lowest_off > 0 && (g_pserver->master_repl_offset - g_pserver->repl_lowest_off + 1) > g_pserver->repl_backlog_config_size)
+        return; // There is untransmitted data we can't truncate
+
+    serverLog(LL_NOTICE, "Reclaiming %lld replication backlog bytes", g_pserver->repl_backlog_size - g_pserver->repl_backlog_config_size);
+    resizeReplicationBacklog(g_pserver->repl_backlog_config_size);
+}
\ No newline at end of file
diff --git a/src/server.cpp b/src/server.cpp
index 3d3b07172..183440f3b 100644
--- a/src/server.cpp
+++ b/src/server.cpp
@@ -7084,6 +7084,8 @@ static void validateConfiguration()
         serverLog(LL_WARNING, "\tKeyDB will now exit.  Please update your configuration file.");
         exit(EXIT_FAILURE);
     }
+
+    g_pserver->repl_backlog_config_size = g_pserver->repl_backlog_size; // this is normally set in the update logic, but not on initial config
 }
 
 int iAmMaster(void) {
diff --git a/src/server.h b/src/server.h
index 32040fb72..5178e86ba 100644
--- a/src/server.h
+++ b/src/server.h
@@ -2358,6 +2358,7 @@ struct redisServer {
     int repl_ping_slave_period;     /* Master pings the replica every N seconds */
     char *repl_backlog;             /* Replication backlog for partial syncs */
     long long repl_backlog_size;    /* Backlog circular buffer size */
+    long long repl_backlog_config_size; /* The repl backlog may grow but we want to know what the user set it to */
     long long repl_backlog_histlen; /* Backlog actual data length */
     long long repl_backlog_idx;     /* Backlog circular buffer current offset,
                                        that is the next byte will'll write to.*/
@@ -3024,6 +3025,8 @@ void clearFailoverState(void);
 void updateFailoverStatus(void);
 void abortFailover(redisMaster *mi, const char *err);
 const char *getFailoverStateString();
+int canFeedReplicaReplBuffer(client *replica);
+void trimReplicationBacklog();
 
 /* Generic persistence functions */
 void startLoadingFile(FILE* fp, const char * filename, int rdbflags);