Add RM_RdbLoad and RM_RdbSave module API functions (#11852)

Add `RM_RdbLoad()` and `RM_RdbSave()` to load/save RDB files from the module API. 

In our use case, we have our clustering implementation as a module. As part of this
implementation, the module needs to trigger RDB save operation at specific points.
Also, this module delivers RDB files to other nodes (not using Redis' replication).
When a node receives an RDB file, it should be able to load the RDB. Currently,
there is no module API to save/load RDB files. 


This PR adds four new APIs:
```c
RedisModuleRdbStream *RM_RdbStreamCreateFromFile(const char *filename);
void RM_RdbStreamFree(RedisModuleRdbStream *stream);

int RM_RdbLoad(RedisModuleCtx *ctx, RedisModuleRdbStream *stream, int flags);
int RM_RdbSave(RedisModuleCtx *ctx, RedisModuleRdbStream *stream, int flags);
```

The first step is to create a `RedisModuleRdbStream` object. This PR provides a function to
create RedisModuleRdbStream from the filename. (You can load/save RDB with the filename).
In the future, this API can be extended if needed: 
e.g., `RM_RdbStreamCreateFromFd()`, `RM_RdbStreamCreateFromSocket()` to save/load
RDB from an `fd` or a `socket`. 


Usage:
```c
/* Save RDB */
RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile("example.rdb");
RedisModule_RdbSave(ctx, stream, 0);
RedisModule_RdbStreamFree(stream);

/* Load RDB */
RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile("example.rdb");
RedisModule_RdbLoad(ctx, stream, 0);
RedisModule_RdbStreamFree(stream);
```

3 files changed, 364 insertions, 1 deletions

diff --git a/tests/modules/Makefile b/tests/modules/Makefile
index a1f5b074b..d63c8548d 100644
--- a/tests/modules/Makefile
+++ b/tests/modules/Makefile
@@ -61,7 +61,8 @@ TEST_MODULES = \
     publish.so \
     usercall.so \
     postnotifications.so \
-    moduleauthtwo.so
+    moduleauthtwo.so \
+    rdbloadsave.so
 
 .PHONY: all
 
diff --git a/tests/modules/rdbloadsave.c b/tests/modules/rdbloadsave.c
new file mode 100644
index 000000000..687269a5a
--- /dev/null
+++ b/tests/modules/rdbloadsave.c
@@ -0,0 +1,162 @@
+#include "redismodule.h"
+
+#include <stdlib.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <memory.h>
+#include <errno.h>
+
+/* Sanity tests to verify inputs and return values. */
+int sanity(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    RedisModuleRdbStream *s = RedisModule_RdbStreamCreateFromFile("dbnew.rdb");
+
+    /* NULL stream should fail. */
+    if (RedisModule_RdbLoad(ctx, NULL, 0) == REDISMODULE_OK || errno != EINVAL) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Invalid flags should fail. */
+    if (RedisModule_RdbLoad(ctx, s, 188) == REDISMODULE_OK || errno != EINVAL) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Missing file should fail. */
+    if (RedisModule_RdbLoad(ctx, s, 0) == REDISMODULE_OK || errno != ENOENT) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Save RDB file. */
+    if (RedisModule_RdbSave(ctx, s, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Load the saved RDB file. */
+    if (RedisModule_RdbLoad(ctx, s, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+
+ out:
+    RedisModule_RdbStreamFree(s);
+    return REDISMODULE_OK;
+}
+
+int cmd_rdbsave(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    if (RedisModule_RdbSave(ctx, stream, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+
+out:
+    RedisModule_RdbStreamFree(stream);
+    return REDISMODULE_OK;
+}
+
+/* Fork before calling RM_RdbSave(). */
+int cmd_rdbsave_fork(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    int fork_child_pid = RedisModule_Fork(NULL, NULL);
+    if (fork_child_pid < 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        return REDISMODULE_OK;
+    } else if (fork_child_pid > 0) {
+        /* parent */
+        RedisModule_ReplyWithSimpleString(ctx, "OK");
+        return REDISMODULE_OK;
+    }
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    int ret = 0;
+    if (RedisModule_RdbSave(ctx, stream, 0) != REDISMODULE_OK) {
+        ret = errno;
+    }
+    RedisModule_RdbStreamFree(stream);
+
+    RedisModule_ExitFromChild(ret);
+    return REDISMODULE_OK;
+}
+
+int cmd_rdbload(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    if (RedisModule_RdbLoad(ctx, stream, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_RdbStreamFree(stream);
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        return REDISMODULE_OK;
+    }
+
+    RedisModule_RdbStreamFree(stream);
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+    return REDISMODULE_OK;
+}
+
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    if (RedisModule_Init(ctx, "rdbloadsave", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.sanity", sanity, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbsave", cmd_rdbsave, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbsave_fork", cmd_rdbsave_fork, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbload", cmd_rdbload, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    return REDISMODULE_OK;
+}
diff --git a/tests/unit/moduleapi/rdbloadsave.tcl b/tests/unit/moduleapi/rdbloadsave.tcl
new file mode 100644
index 000000000..9319c9385
--- /dev/null
+++ b/tests/unit/moduleapi/rdbloadsave.tcl
@@ -0,0 +1,200 @@
+set testmodule [file normalize tests/modules/rdbloadsave.so]
+
+start_server {tags {"modules"}} {
+    r module load $testmodule
+
+    test "Module rdbloadsave sanity" {
+        r test.sanity
+
+        # Try to load non-existing file
+        assert_error {*No such file or directory*} {r test.rdbload sanity.rdb}
+
+        r set x 1
+        assert_equal OK [r test.rdbsave sanity.rdb]
+
+        r flushdb
+        assert_equal OK [r test.rdbload sanity.rdb]
+        assert_equal 1 [r get x]
+    }
+
+    test "Module rdbloadsave test with pipelining" {
+        r config set save ""
+        r config set loading-process-events-interval-bytes 1024
+        r config set key-load-delay 50
+        r flushdb
+
+        populate 3000 a 1024
+        r set x 111
+        assert_equal [r dbsize] 3001
+
+        assert_equal OK [r test.rdbsave blabla.rdb]
+        r flushdb
+        assert_equal [r dbsize] 0
+
+        # Send commands with pipeline. First command will call RM_RdbLoad() in
+        # the command callback. While loading RDB, Redis can go to networking to
+        # reply -LOADING. By sending commands in pipeline, we verify it doesn't
+        # cause a problem.
+        # e.g. Redis won't try to process next message of the current client
+        # while it is in the command callback for that client   .
+        set rd1 [redis_deferring_client]
+        $rd1 test.rdbload blabla.rdb
+
+        wait_for_condition 50 100 {
+            [s loading] eq 1
+        } else {
+            fail "Redis did not start loading or loaded RDB too fast"
+        }
+
+        $rd1 get x
+        $rd1 dbsize
+
+        assert_equal OK [$rd1 read]
+        assert_equal 111 [$rd1 read]
+        assert_equal 3001 [$rd1 read]
+        r flushdb
+        r config set key-load-delay 0
+    }
+
+    test "Module rdbloadsave with aof" {
+        r config set save ""
+
+        # Enable the AOF
+        r config set appendonly yes
+        r config set auto-aof-rewrite-percentage 0 ; # Disable auto-rewrite.
+        waitForBgrewriteaof r
+
+        r set k v1
+        assert_equal OK [r test.rdbsave aoftest.rdb]
+
+        r set k v2
+        r config set rdb-key-save-delay 10000000
+        r bgrewriteaof
+
+        # RM_RdbLoad() should kill aof fork
+        assert_equal OK [r test.rdbload aoftest.rdb]
+
+        wait_for_condition 50 100 {
+            [string match {*Killing*AOF*child*} [exec tail -20 < [srv 0 stdout]]]
+        } else {
+            fail "Can't find 'Killing AOF child' in recent log lines"
+        }
+
+        # Verify the value in the loaded rdb
+        assert_equal v1 [r get k]
+
+        r flushdb
+        r config set rdb-key-save-delay 0
+        r config set appendonly no
+    }
+
+    test "Module rdbloadsave with bgsave" {
+        r flushdb
+        r config set save ""
+
+        r set k v1
+        assert_equal OK [r test.rdbsave bgsave.rdb]
+
+        r set k v2
+        r config set rdb-key-save-delay 500000
+        r bgsave
+
+        # RM_RdbLoad() should kill RDB fork
+        assert_equal OK [r test.rdbload bgsave.rdb]
+
+        wait_for_condition 10 1000 {
+            [string match {*Background*saving*terminated*} [exec tail -20 < [srv 0 stdout]]]
+        } else {
+            fail "Can't find 'Background saving terminated' in recent log lines"
+        }
+
+        assert_equal v1 [r get k]
+        r flushall
+        waitForBgsave r
+        r config set rdb-key-save-delay 0
+    }
+
+    test "Module rdbloadsave calls rdbsave in a module fork" {
+        r flushdb
+        r config set save ""
+        r config set rdb-key-save-delay 500000
+
+        r set k v1
+
+        # Module will call RM_Fork() before calling RM_RdbSave()
+        assert_equal OK [r test.rdbsave_fork rdbfork.rdb]
+        assert_equal [s module_fork_in_progress] 1
+
+        wait_for_condition 10 1000 {
+            [status r module_fork_in_progress] == "0"
+        } else {
+            fail "Module fork didn't finish"
+        }
+
+        r set k v2
+        assert_equal OK [r test.rdbload rdbfork.rdb]
+        assert_equal v1 [r get k]
+
+        r config set rdb-key-save-delay 0
+    }
+
+    test "Unload the module - rdbloadsave" {
+        assert_equal {OK} [r module unload rdbloadsave]
+    }
+
+    tags {repl} {
+        test {Module rdbloadsave on master and replica} {
+            start_server [list overrides [list loadmodule "$testmodule"]] {
+                set replica [srv 0 client]
+                set replica_host [srv 0 host]
+                set replica_port [srv 0 port]
+                start_server [list overrides [list loadmodule "$testmodule"]] {
+                    set master [srv 0 client]
+                    set master_host [srv 0 host]
+                    set master_port [srv 0 port]
+
+                    $master set x 10000
+
+                    # Start the replication process...
+                    $replica replicaof $master_host $master_port
+
+                    wait_for_condition 100 100 {
+                        [status $master sync_full] == 1
+                    } else {
+                        fail "Master <-> Replica didn't start the full sync"
+                    }
+
+                    # RM_RdbSave() is allowed on replicas
+                    assert_equal OK [$replica test.rdbsave rep.rdb]
+
+                    # RM_RdbLoad() is not allowed on replicas
+                    assert_error {*supported*} {$replica test.rdbload rep.rdb}
+
+                    assert_equal OK [$master test.rdbsave master.rdb]
+                    $master set x 20000
+
+                    wait_for_condition 100 100 {
+                        [$replica get x] == 20000
+                    } else {
+                        fail "Replica didn't get the update"
+                    }
+
+                    # Loading RDB on master will drop replicas
+                    assert_equal OK [$master test.rdbload master.rdb]
+
+                    wait_for_condition 100 100 {
+                        [status $master sync_full] == 2
+                    } else {
+                        fail "Master <-> Replica didn't start the full sync"
+                    }
+
+                    wait_for_condition 100 100 {
+                        [$replica get x] == 10000
+                    } else {
+                        fail "Replica didn't get the update"
+                    }
+                }
+            }
+        }
+    }
+}