Fix broken protocol in MISCONF error, RM_Yield bugs, RM_Call(EVAL) OOM check bug, and new RM_Call checks. (#10786)

* Fix broken protocol when redis can't persist to RDB (general commands, not
  modules), excessive newline. regression of #10372 (7.0 RC3)
* Fix broken protocol when Redis can't persist to AOF (modules and
  scripts), missing newline.
* Fix bug in OOM check of EVAL scripts called from RM_Call.
  set the cached OOM state for scripts before executing module commands too,
  so that it can serve scripts that are executed by modules.
  i.e. in the past EVAL executed by RM_Call could have either falsely
  fail or falsely succeeded because of a wrong cached OOM state flag.
* Fix bugs with RM_Yield:
  1. SHUTDOWN should only accept the NOSAVE mode
  2. Avoid eviction during yield command processing.
  3. Avoid processing master client commands while yielding from another client
* Add new two more checks to RM_Call script mode.
  1. READONLY You can't write against a read only replica
  2. MASTERDOWN Link with MASTER is down and `replica-serve-stale-data` is set to `no`
* Add new RM_Call flag to let redis automatically refuse `deny-oom` commands
  while over the memory limit. 
* Add tests to cover various errors from Scripts, Modules, Modules
  calling scripts, and Modules calling commands in script mode.

Add tests:
* Looks like the MISCONF error was completely uncovered by the tests,
  add tests for it, including from scripts, and modules
* Add tests for NOREPLICAS from scripts
* Add tests for the various errors in module RM_Call, including RM_Call that
  calls EVAL, and RM_call in "eval mode". that includes:
  NOREPLICAS, READONLY, MASTERDOWN, MISCONF

12 files changed, 416 insertions, 29 deletions

diff --git a/src/call_reply.c b/src/call_reply.c
index 759cd792a..7c74a58df 100644
--- a/src/call_reply.c
+++ b/src/call_reply.c
@@ -528,7 +528,10 @@ CallReply *callReplyCreate(sds reply, list *deferred_error_list, void *private_d
 
 /* Create a new CallReply struct from the reply blob representing an error message.
  * Automatically creating deferred_error_list and set a copy of the reply in it.
- * Refer to callReplyCreate for detailed explanation. */
+ * Refer to callReplyCreate for detailed explanation.
+ * Reply string can come in one of two forms:
+ * 1. A protocol reply starting with "-CODE" and ending with "\r\n"
+ * 2. A plain string, in which case this function adds the protocol header and footer. */
 CallReply *callReplyCreateError(sds reply, void *private_data) {
     sds err_buff = reply;
     if (err_buff[0] != '-') {
diff --git a/src/db.c b/src/db.c
index eaf45e9e3..99212bac4 100644
--- a/src/db.c
+++ b/src/db.c
@@ -1084,7 +1084,7 @@ void shutdownCommand(client *c) {
         return;
     }
 
-    if (!(flags & SHUTDOWN_NOSAVE) && scriptIsTimedout()) {
+    if (!(flags & SHUTDOWN_NOSAVE) && isInsideYieldingLongCommand()) {
         /* Script timed out. Shutdown allowed only with the NOSAVE flag. See
          * also processCommand where these errors are returned. */
         if (server.busy_module_yield_flags && server.busy_module_yield_reply) {
diff --git a/src/evict.c b/src/evict.c
index a5821a463..316edf606 100644
--- a/src/evict.c
+++ b/src/evict.c
@@ -481,7 +481,7 @@ void startEvictionTimeProc(void) {
 static int isSafeToPerformEvictions(void) {
     /* - There must be no script in timeout condition.
      * - Nor we are loading data right now.  */
-    if (scriptIsTimedout() || server.loading) return 0;
+    if (isInsideYieldingLongCommand() || server.loading) return 0;
 
     /* By default replicas should ignore maxmemory
      * and just be masters exact copies. */
diff --git a/src/module.c b/src/module.c
index 3e5d06b6e..fd76d96c2 100644
--- a/src/module.c
+++ b/src/module.c
@@ -356,6 +356,7 @@ typedef struct RedisModuleServerInfoData {
 #define REDISMODULE_ARGV_SCRIPT_MODE (1<<6)
 #define REDISMODULE_ARGV_NO_WRITES (1<<7)
 #define REDISMODULE_ARGV_CALL_REPLIES_AS_ERRORS (1<<8)
+#define REDISMODULE_ARGV_RESPECT_DENY_OOM (1<<9)
 
 /* Determine whether Redis should signalModifiedKey implicitly.
  * In case 'ctx' has no 'module' member (and therefore no module->options),
@@ -5625,6 +5626,8 @@ robj **moduleCreateArgvFromUserFormat(const char *cmdname, const char *fmt, int
             if (flags) (*flags) |= REDISMODULE_ARGV_SCRIPT_MODE;
         } else if (*p == 'W') {
             if (flags) (*flags) |= REDISMODULE_ARGV_NO_WRITES;
+        } else if (*p == 'M') {
+            if (flags) (*flags) |= REDISMODULE_ARGV_RESPECT_DENY_OOM;
         } else if (*p == 'E') {
             if (flags) (*flags) |= REDISMODULE_ARGV_CALL_REPLIES_AS_ERRORS;
         } else {
@@ -5673,6 +5676,7 @@ fmterr:
  *              not enough good replicas (as configured with `min-replicas-to-write`)
  *              or when the server is unable to persist to the disk.
  *     * `W` -- Do not allow to run any write command (flagged with the `write` flag).
+ *     * `M` -- Do not allow `deny-oom` flagged commands when over the memory limit.
  *     * `E` -- Return error as RedisModuleCallReply. If there is an error before
  *              invoking the command, the error is returned using errno mechanism.
  *              This flag allows to get the error also as an error CallReply with
@@ -5691,7 +5695,7 @@ fmterr:
  * * ENETDOWN: operation in Cluster instance when cluster is down.
  * * ENOTSUP: No ACL user for the specified module context
  * * EACCES: Command cannot be executed, according to ACL rules
- * * ENOSPC: Write command is not allowed
+ * * ENOSPC: Write or deny-oom command is not allowed
  * * ESPIPE: Command not allowed on script mode
  *
  * Example code fragment:
@@ -5783,8 +5787,21 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
         }
     }
 
-    if (cmd->flags & CMD_WRITE) {
-        if (flags & REDISMODULE_ARGV_NO_WRITES) {
+    if (flags & REDISMODULE_ARGV_RESPECT_DENY_OOM) {
+        if (cmd->flags & CMD_DENYOOM) {
+            if (server.pre_command_oom_state) {
+                errno = ENOSPC;
+                if (error_as_call_replies) {
+                    sds msg = sdsdup(shared.oomerr->ptr);
+                    reply = callReplyCreateError(msg, ctx);
+                }
+                goto cleanup;
+            }
+        }
+    }
+
+    if (flags & REDISMODULE_ARGV_NO_WRITES) {
+        if (cmd->flags & CMD_WRITE) {
             errno = ENOSPC;
             if (error_as_call_replies) {
                 sds msg = sdscatfmt(sdsempty(), "Write command '%S' was "
@@ -5793,14 +5810,17 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
             }
             goto cleanup;
         }
+    }
 
-        if (flags & REDISMODULE_ARGV_SCRIPT_MODE) {
+    /* Script mode tests */
+    if (flags & REDISMODULE_ARGV_SCRIPT_MODE) {
+        if (cmd->flags & CMD_WRITE) {
             /* on script mode, if a command is a write command,
              * We will not run it if we encounter disk error
              * or we do not have enough replicas */
 
             if (!checkGoodReplicasStatus()) {
-                errno = ENOSPC;
+                errno = ESPIPE;
                 if (error_as_call_replies) {
                     sds msg = sdsdup(shared.noreplicaserr->ptr);
                     reply = callReplyCreateError(msg, ctx);
@@ -5812,7 +5832,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
             int obey_client = mustObeyClient(server.current_client);
 
             if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) {
-                errno = ENOSPC;
+                errno = ESPIPE;
                 if (error_as_call_replies) {
                     sds msg = writeCommandsGetDiskErrorMessage(deny_write_type);
                     reply = callReplyCreateError(msg, ctx);
@@ -5820,6 +5840,24 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
                 goto cleanup;
             }
 
+            if (server.masterhost && server.repl_slave_ro && !obey_client) {
+                errno = ESPIPE;
+                if (error_as_call_replies) {
+                    sds msg = sdsdup(shared.roslaveerr->ptr);
+                    reply = callReplyCreateError(msg, ctx);
+                }
+                goto cleanup;
+            }
+        }
+
+        if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
+            server.repl_serve_stale_data == 0 && !(cmd->flags & CMD_STALE)) {
+            errno = ESPIPE;
+            if (error_as_call_replies) {
+                sds msg = sdsdup(shared.masterdownerr->ptr);
+                reply = callReplyCreateError(msg, ctx);
+            }
+            goto cleanup;
         }
     }
 
diff --git a/src/networking.c b/src/networking.c
index c1d6aa4bd..bf39c6582 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -2505,7 +2505,7 @@ int processInputBuffer(client *c) {
          * condition on the slave. We want just to accumulate the replication
          * stream (instead of replying -BUSY like we do with other clients) and
          * later resume the processing. */
-        if (scriptIsTimedout() && c->flags & CLIENT_MASTER) break;
+        if (isInsideYieldingLongCommand() && c->flags & CLIENT_MASTER) break;
 
         /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
          * written to the client. Make sure to not let the reply grow after
diff --git a/src/script.c b/src/script.c
index 07d6387cc..4d9cb6b1f 100644
--- a/src/script.c
+++ b/src/script.c
@@ -174,7 +174,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
 
         /* Check OOM state. the no-writes flag imply allow-oom. we tested it
          * after the no-write error, so no need to mention it in the error reply. */
-        if (server.script_oom && server.maxmemory &&
+        if (server.pre_command_oom_state && server.maxmemory &&
             !(script_flags & (SCRIPT_FLAG_ALLOW_OOM|SCRIPT_FLAG_NO_WRITES)))
         {
             addReplyError(caller, "-OOM allow-oom flag is not set on the script, "
@@ -389,8 +389,8 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
 
     if (server.maxmemory &&                            /* Maxmemory is actually enabled. */
         !mustObeyClient(run_ctx->original_client) &&   /* Don't care about mem for replicas or AOF. */
-        !(run_ctx->flags & SCRIPT_WRITE_DIRTY) &&        /* Script had no side effects so far. */
-        server.script_oom &&                           /* Detected OOM when script start. */
+        !(run_ctx->flags & SCRIPT_WRITE_DIRTY) &&      /* Script had no side effects so far. */
+        server.pre_command_oom_state &&                /* Detected OOM when script start. */
         (run_ctx->c->cmd->flags & CMD_DENYOOM))
     {
         *err = sdsdup(shared.oomerr->ptr);
diff --git a/src/server.c b/src/server.c
index 231ec5648..e87e4ece5 100644
--- a/src/server.c
+++ b/src/server.c
@@ -639,6 +639,11 @@ int isMutuallyExclusiveChildType(int type) {
     return type == CHILD_TYPE_RDB || type == CHILD_TYPE_AOF || type == CHILD_TYPE_MODULE;
 }
 
+/* Returns true when we're inside a long command that yielded to the event loop. */
+int isInsideYieldingLongCommand() {
+    return scriptIsTimedout() || server.busy_module_yield_flags;
+}
+
 /* Return true if this instance has persistence completely turned off:
  * both RDB and AOF are disabled. */
 int allPersistenceDisabled(void) {
@@ -3728,7 +3733,7 @@ int processCommand(client *c) {
      * the event loop since there is a busy Lua script running in timeout
      * condition, to avoid mixing the propagation of scripts with the
      * propagation of DELs due to eviction. */
-    if (server.maxmemory && !scriptIsTimedout()) {
+    if (server.maxmemory && !isInsideYieldingLongCommand()) {
         int out_of_memory = (performEvictions() == EVICT_FAIL);
 
         /* performEvictions may evict keys, so we need flush pending tracking
@@ -3760,18 +3765,12 @@ int processCommand(client *c) {
             return C_OK;
         }
 
-        /* Save out_of_memory result at script start, otherwise if we check OOM
-         * until first write within script, memory used by lua stack and
-         * arguments might interfere. */
-        if (c->cmd->proc == evalCommand ||
-            c->cmd->proc == evalRoCommand ||
-            c->cmd->proc == evalShaCommand ||
-            c->cmd->proc == evalShaRoCommand ||
-            c->cmd->proc == fcallCommand ||
-            c->cmd->proc == fcallroCommand)
-        {
-            server.script_oom = out_of_memory;
-        }
+        /* Save out_of_memory result at command start, otherwise if we check OOM
+         * in the first write within script, memory used by lua stack and
+         * arguments might interfere. We need to save it for EXEC and module
+         * calls too, since these can call EVAL, but avoid saving it during an
+         * interrupted / yielding busy script / module. */
+        server.pre_command_oom_state = out_of_memory;
     }
 
     /* Make sure to use a reasonable amount of memory for client side
@@ -3799,6 +3798,8 @@ int processCommand(client *c) {
             }
         } else {
             sds err = writeCommandsGetDiskErrorMessage(deny_write_type);
+            /* remove the newline since rejectCommandSds adds it. */
+            sdssubstr(err, 0, sdslen(err)-2);
             rejectCommandSds(c, err);
             return C_OK;
         }
@@ -3871,7 +3872,7 @@ int processCommand(client *c) {
      * the MULTI plus a few initial commands refused, then the timeout
      * condition resolves, and the bottom-half of the transaction gets
      * executed, see Github PR #7022. */
-    if ((scriptIsTimedout() || server.busy_module_yield_flags) && !(c->cmd->flags & CMD_ALLOW_BUSY)) {
+    if (isInsideYieldingLongCommand() && !(c->cmd->flags & CMD_ALLOW_BUSY)) {
         if (server.busy_module_yield_flags && server.busy_module_yield_reply) {
             rejectCommandFormat(c, "-BUSY %s", server.busy_module_yield_reply);
         } else if (server.busy_module_yield_flags) {
@@ -4238,7 +4239,7 @@ sds writeCommandsGetDiskErrorMessage(int error_code) {
         ret = sdsdup(shared.bgsaveerr->ptr);
     } else {
         ret = sdscatfmt(sdsempty(),
-                "-MISCONF Errors writing to the AOF file: %s",
+                "-MISCONF Errors writing to the AOF file: %s\r\n",
                 strerror(server.aof_last_write_errno));
     }
     return ret;
diff --git a/src/server.h b/src/server.h
index 8ce2edff5..81fd9af19 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1889,7 +1889,7 @@ struct redisServer {
     /* Scripting */
     client *script_caller;       /* The client running script right now, or NULL */
     mstime_t busy_reply_threshold;  /* Script / module timeout in milliseconds */
-    int script_oom;                    /* OOM detected when script start */
+    int pre_command_oom_state;         /* OOM before command (script?) was started */
     int script_disable_deny_script;    /* Allow running commands marked "no-script" inside a script. */
     /* Lazy free */
     int lazyfree_lazy_eviction;
@@ -3201,6 +3201,7 @@ void sha1hex(char *digest, char *script, size_t len);
 unsigned long evalMemory();
 dict* evalScriptsDict();
 unsigned long evalScriptsMemory();
+int isInsideYieldingLongCommand();
 
 typedef struct luaScript {
     uint64_t flags;
diff --git a/tests/integration/rdb.tcl b/tests/integration/rdb.tcl
index 326a17350..104d372e1 100644
--- a/tests/integration/rdb.tcl
+++ b/tests/integration/rdb.tcl
@@ -366,4 +366,49 @@ start_server [list overrides [list "dir" $server_path "dbfilename" "scriptbackup
     }
 }
 
+start_server {} {
+    test "failed bgsave prevents writes" {
+        r config set rdb-key-save-delay 10000000
+        populate 1000
+        r set x x
+        r bgsave
+        set pid1 [get_child_pid 0]
+        catch {exec kill -9 $pid1}
+        waitForBgsave r
+
+        # make sure a read command succeeds
+        assert_equal [r get x] x
+
+        # make sure a write command fails
+        assert_error {MISCONF *} {r set x y}
+
+        # repeate with script
+        assert_error {MISCONF *} {r eval {
+            return redis.call('set','x',1)
+            } 1 x
+        }
+        assert_equal {x} [r eval {
+            return redis.call('get','x')
+            } 1 x
+        ]
+
+        # again with script using shebang
+        assert_error {MISCONF *} {r eval {#!lua
+            return redis.call('set','x',1)
+            } 1 x
+        }
+        assert_equal {x} [r eval {#!lua flags=no-writes
+            return redis.call('get','x')
+            } 1 x
+        ]
+
+        r config set rdb-key-save-delay 0
+        r bgsave
+        waitForBgsave r
+
+        # server is writable again
+        r set x y
+    } {OK}
+}
+
 } ;# tags
diff --git a/tests/modules/misc.c b/tests/modules/misc.c
index db9c090d4..da6ee9f9e 100644
--- a/tests/modules/misc.c
+++ b/tests/modules/misc.c
@@ -310,6 +310,50 @@ int test_monotonic_time(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     return REDISMODULE_OK;
 }
 
+/* wrapper for RM_Call */
+int test_rm_call(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
+    if(argc < 2){
+        return RedisModule_WrongArity(ctx);
+    }
+
+    const char* cmd = RedisModule_StringPtrLen(argv[1], NULL);
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, cmd, "Ev", argv + 2, argc - 2);
+    if(!rep){
+        RedisModule_ReplyWithError(ctx, "NULL reply returned");
+    }else{
+        RedisModule_ReplyWithCallReply(ctx, rep);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    return REDISMODULE_OK;
+}
+
+/* wrapper for RM_Call with flags */
+int test_rm_call_flags(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
+    if(argc < 3){
+        return RedisModule_WrongArity(ctx);
+    }
+
+    /* Append Ev to the provided flags. */
+    RedisModuleString *flags = RedisModule_CreateStringFromString(ctx, argv[1]);
+    RedisModule_StringAppendBuffer(ctx, flags, "Ev", 2);
+
+    const char* flg = RedisModule_StringPtrLen(flags, NULL);
+    const char* cmd = RedisModule_StringPtrLen(argv[2], NULL);
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, cmd, flg, argv + 3, argc - 3);
+    if(!rep){
+        RedisModule_ReplyWithError(ctx, "NULL reply returned");
+    }else{
+        RedisModule_ReplyWithCallReply(ctx, rep);
+        RedisModule_FreeCallReply(rep);
+    }
+    RedisModule_FreeString(ctx, flags);
+
+    return REDISMODULE_OK;
+}
+
 int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     REDISMODULE_NOT_USED(argv);
     REDISMODULE_NOT_USED(argc);
@@ -351,6 +395,10 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
         return REDISMODULE_ERR;
     if (RedisModule_CreateCommand(ctx,"test.monotonic_time", test_monotonic_time,"",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
+    if (RedisModule_CreateCommand(ctx, "test.rm_call", test_rm_call,"allow-stale", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+    if (RedisModule_CreateCommand(ctx, "test.rm_call_flags", test_rm_call_flags,"allow-stale", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
 
     return REDISMODULE_OK;
 }
diff --git a/tests/unit/moduleapi/misc.tcl b/tests/unit/moduleapi/misc.tcl
index bc00b37d2..492974c67 100644
--- a/tests/unit/moduleapi/misc.tcl
+++ b/tests/unit/moduleapi/misc.tcl
@@ -133,6 +133,223 @@ start_server {tags {"modules"}} {
         assert { [r test.monotonic_time] >= $x }
     }
 
+    test {rm_call OOM} {
+        r config set maxmemory 1
+        r config set maxmemory-policy volatile-lru
+
+        # sanity test plain call
+        assert_equal {OK} [
+            r test.rm_call set x 1
+        ]
+
+        # add the M flag
+        assert_error {OOM *} {
+            r test.rm_call_flags M set x 1
+
+        }
+
+        # test a non deny-oom command
+        assert_equal {1} [
+            r test.rm_call_flags M get x
+        ]
+
+        r config set maxmemory 0
+    } {OK} {needs:config-maxmemory}
+
+    test {rm_call write flag} {
+        # add the W flag
+        assert_error {ERR Write command 'set' was called while write is not allowed.} {
+            r test.rm_call_flags W set x 1
+        }
+
+        # test a non deny-oom command
+        r test.rm_call_flags W get x
+    } {1}
+
+    test {rm_call EVAL} {
+        r test.rm_call eval {
+            redis.call('set','x',1)
+            return 1
+        } 1 x
+
+        assert_error {ERR Write commands are not allowed from read-only scripts.*} {
+            r test.rm_call eval {#!lua flags=no-writes
+                redis.call('set','x',1)
+                return 1
+            } 1 x
+        }
+    }
+
+    test {rm_call EVAL - OOM} {
+        r config set maxmemory 1
+
+        assert_error {OOM command not allowed when used memory > 'maxmemory'. script*} {
+            r test.rm_call eval {
+                redis.call('set','x',1)
+                return 1
+            } 1 x
+        }
+
+        r test.rm_call eval {#!lua flags=no-writes
+            redis.call('get','x')
+            return 2
+        } 1 x
+
+        assert_error {OOM allow-oom flag is not set on the script,*} {
+            r test.rm_call eval {#!lua
+                redis.call('get','x')
+                return 3
+            } 1 x
+        }
+
+        r test.rm_call eval {
+            redis.call('get','x')
+            return 4
+        } 1 x
+
+        r config set maxmemory 0
+    } {OK} {needs:config-maxmemory}
+
+    test "not enough good replicas" {
+        r set x "some value"
+        r config set min-replicas-to-write 1
+
+        # rm_call in script mode
+        assert_error {NOREPLICAS *} {r test.rm_call_flags S set x s}
+
+        assert_equal [
+            r test.rm_call eval {#!lua flags=no-writes
+                return redis.call('get','x')
+            } 1 x
+        ] "some value"
+
+        assert_equal [
+            r test.rm_call eval {
+                return redis.call('get','x')
+            } 1 x
+        ] "some value"
+
+        assert_error {NOREPLICAS *} {
+            r test.rm_call eval {#!lua
+                return redis.call('get','x')
+            } 1 x
+        }
+
+        assert_error {NOREPLICAS *} {
+            r test.rm_call eval {
+                return redis.call('set','x', 1)
+            } 1 x
+        }
+
+        r config set min-replicas-to-write 0
+    }
+
+    test {rm_call EVAL - read-only replica} {
+        r replicaof 127.0.0.1 1
+
+        # rm_call in script mode
+        assert_error {READONLY *} {r test.rm_call_flags S set x 1}
+
+        assert_error {READONLY You can't write against a read only replica. script*} {
+            r test.rm_call eval {
+                redis.call('set','x',1)
+                return 1
+            } 1 x
+        }
+
+        r test.rm_call eval {#!lua flags=no-writes
+            redis.call('get','x')
+            return 2
+        } 1 x
+
+        assert_error {ERR Can not run script with write flag on readonly replica} {
+            r test.rm_call eval {#!lua
+                redis.call('get','x')
+                return 3
+            } 1 x
+        }
+
+        r test.rm_call eval {
+            redis.call('get','x')
+            return 4
+        } 1 x
+
+        r replicaof no one
+    } {OK} {needs:config-maxmemory}
+
+    test {rm_call EVAL - stale replica} {
+        r replicaof 127.0.0.1 1
+        r config set replica-serve-stale-data no
+
+        # rm_call in script mode
+        assert_error {MASTERDOWN *} {
+            r test.rm_call_flags S get x
+        }
+
+        assert_error {MASTERDOWN *} {
+            r test.rm_call eval {#!lua flags=no-writes
+                redis.call('get','x')
+                return 2
+            } 1 x
+        }
+
+        assert_error {MASTERDOWN *} {
+            r test.rm_call eval {
+                redis.call('get','x')
+                return 4
+            } 1 x
+        }
+
+        r replicaof no one
+        r config set replica-serve-stale-data yes
+    } {OK} {needs:config-maxmemory}
+
+    test "rm_call EVAL - failed bgsave prevents writes" {
+        r config set rdb-key-save-delay 10000000
+        populate 1000
+        r set x x
+        r bgsave
+        set pid1 [get_child_pid 0]
+        catch {exec kill -9 $pid1}
+        waitForBgsave r
+
+        # make sure a read command succeeds
+        assert_equal [r get x] x
+
+        # make sure a write command fails
+        assert_error {MISCONF *} {r set x y}
+
+        # rm_call in script mode
+        assert_error {MISCONF *} {r test.rm_call_flags S set x 1}
+
+        # repeate with script
+        assert_error {MISCONF *} {r test.rm_call eval {
+            return redis.call('set','x',1)
+            } 1 x
+        }
+        assert_equal {x} [r test.rm_call eval {
+            return redis.call('get','x')
+            } 1 x
+        ]
+
+        # again with script using shebang
+        assert_error {MISCONF *} {r test.rm_call eval {#!lua
+            return redis.call('set','x',1)
+            } 1 x
+        }
+        assert_equal {x} [r test.rm_call eval {#!lua flags=no-writes
+            return redis.call('get','x')
+            } 1 x
+        ]
+
+        r config set rdb-key-save-delay 0
+        r bgsave
+        waitForBgsave r
+
+        # server is writable again
+        r set x y
+    } {OK}
+
     test "Unload the module - misc" {
         assert_equal {OK} [r module unload misc]
     }
diff --git a/tests/unit/scripting.tcl b/tests/unit/scripting.tcl
index f85fc8484..ea7b17054 100644
--- a/tests/unit/scripting.tcl
+++ b/tests/unit/scripting.tcl
@@ -1494,7 +1494,41 @@ start_server {tags {"scripting"}} {
                     return redis.call('get','x')
                 } 1 x
             }
+
+            # sanity check on protocol after error reply
+            assert_equal [r ping] PONG
+        }
+    }
+
+    test "not enough good replicas" {
+        r set x "some value"
+        r config set min-replicas-to-write 1
+
+        assert_equal [
+            r eval {#!lua flags=no-writes
+                return redis.call('get','x')
+            } 1 x
+        ] "some value"
+
+        assert_equal [
+            r eval {
+                return redis.call('get','x')
+            } 1 x
+        ] "some value"
+
+        assert_error {NOREPLICAS *} {
+            r eval {#!lua
+                return redis.call('get','x')
+            } 1 x
         }
+
+        assert_error {NOREPLICAS *} {
+            r eval {
+                return redis.call('set','x', 1)
+            } 1 x
+        }
+
+        r config set min-replicas-to-write 0
     }
 
     test "allow-stale shebang flag" {