Return the ring buffer to its original size if we temporarily resized it

Former-commit-id: a12ce4a0d105bf7d6ccff95f7dc0044c4676b0a7

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;
 }

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);
+}

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) {

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);