Sort out the mess around Lua error messages and error stats (#10329)

This PR fix 2 issues on Lua scripting:
* Server error reply statistics (some errors were counted twice).
* Error code and error strings returning from scripts (error code was missing / misplaced).

## Statistics
a Lua script user is considered part of the user application, a sophisticated transaction,
so we want to count an error even if handled silently by the script, but when it is
propagated outwards from the script we don't wanna count it twice. on the other hand,
if the script decides to throw an error on its own (using `redis.error_reply`), we wanna
count that too.
Besides, we do count the `calls` in command statistics for the commands the script calls,
we we should certainly also count `failed_calls`.
So when a simple `eval "return redis.call('set','x','y')" 0` fails, it should count the failed call
to both SET and EVAL, but the `errorstats` and `total_error_replies` should be counted only once.

The PR changes the error object that is raised on errors. Instead of raising a simple Lua
string, Redis will raise a Lua table in the following format:

```
{
    err='<error message (including error code)>',
    source='<User source file name>',
    line='<line where the error happned>',
    ignore_error_stats_update=true/false,
}
```

The `luaPushError` function was modified to construct the new error table as describe above.
The `luaRaiseError` was renamed to `luaError` and is now simply called `lua_error` to raise
the table on the top of the Lua stack as the error object.
The reason is that since its functionality is changed, in case some Redis branch / fork uses it,
it's better to have a compilation error than a bug.

The `source` and `line` fields are enriched by the error handler (if possible) and the
`ignore_error_stats_update` is optional and if its not present then the default value is `false`.
If `ignore_error_stats_update` is true, the error will not be counted on the error stats.

When parsing Redis call reply, each error is translated to a Lua table on the format describe
above and the `ignore_error_stats_update` field is set to `true` so we will not count errors
twice (we counted this error when we invoke the command).

The changes in this PR might have been considered as a breaking change for users that used
Lua `pcall` function. Before, the error was a string and now its a table. To keep backward
comparability the PR override the `pcall` implementation and extract the error message from
the error table and return it.

Example of the error stats update:

```
127.0.0.1:6379> lpush l 1
(integer) 2
127.0.0.1:6379> eval "return redis.call('get', 'l')" 0
(error) WRONGTYPE Operation against a key holding the wrong kind of value. script: e471b73f1ef44774987ab00bdf51f21fd9f7974a, on @user_script:1.

127.0.0.1:6379> info Errorstats
# Errorstats
errorstat_WRONGTYPE:count=1

127.0.0.1:6379> info commandstats
# Commandstats
cmdstat_eval:calls=1,usec=341,usec_per_call=341.00,rejected_calls=0,failed_calls=1
cmdstat_info:calls=1,usec=35,usec_per_call=35.00,rejected_calls=0,failed_calls=0
cmdstat_lpush:calls=1,usec=14,usec_per_call=14.00,rejected_calls=0,failed_calls=0
cmdstat_get:calls=1,usec=10,usec_per_call=10.00,rejected_calls=0,failed_calls=1
```

## error message
We can now construct the error message (sent as a reply to the user) from the error table,
so this solves issues where the error message was malformed and the error code appeared
in the middle of the error message:

```diff
127.0.0.1:6379> eval "return redis.call('set','x','y')" 0
-(error) ERR Error running script (call to 71e6319f97b0fe8bdfa1c5df3ce4489946dda479): @user_script:1: OOM command not allowed when used memory > 'maxmemory'.
+(error) OOM command not allowed when used memory > 'maxmemory' @user_script:1. Error running script (call to 71e6319f97b0fe8bdfa1c5df3ce4489946dda479)
```

```diff
127.0.0.1:6379> eval "redis.call('get', 'l')" 0
-(error) ERR Error running script (call to f_8a705cfb9fb09515bfe57ca2bd84a5caee2cbbd1): @user_script:1: WRONGTYPE Operation against a key holding the wrong kind of value
+(error) WRONGTYPE Operation against a key holding the wrong kind of value script: 8a705cfb9fb09515bfe57ca2bd84a5caee2cbbd1, on @user_script:1.
```

Notica that `redis.pcall` was not change:
```
127.0.0.1:6379> eval "return redis.pcall('get', 'l')" 0
(error) WRONGTYPE Operation against a key holding the wrong kind of value
```


## other notes
Notice that Some commands (like GEOADD) changes the cmd variable on the client stats so we
can not count on it to update the command stats. In order to be able to update those stats correctly
we needed to promote `realcmd` variable to be located on the client struct.

Tests was added and modified to verify the changes.

Related PR's: #10279, #10218, #10278, #10309

Co-authored-by: Oran Agra <oran@redislabs.com>

1 files changed, 195 insertions, 56 deletions

diff --git a/src/script_lua.c b/src/script_lua.c
index 82591d3fc..9a08a7e47 100644
--- a/src/script_lua.c
+++ b/src/script_lua.c
@@ -238,9 +238,12 @@ static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len,
          * to push elements to the stack. On failure, exit with panic. */
         serverPanic("lua stack limit reach when parsing redis.call reply");
     }
-    lua_newtable(lua);
-    lua_pushstring(lua,"err");
-    lua_pushlstring(lua,str,len);
+    sds err_msg = sdscatlen(sdsnew("-"), str, len);
+    luaPushErrorBuff(lua,err_msg);
+    /* push a field indicate to ignore updating the stats on this error
+     * because it was already updated when executing the command. */
+    lua_pushstring(lua,"ignore_error_stats_update");
+    lua_pushboolean(lua, true);
     lua_settable(lua,-3);
 }
 
@@ -428,46 +431,66 @@ static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto
 
 /* This function is used in order to push an error on the Lua stack in the
  * format used by redis.pcall to return errors, which is a lua table
- * with a single "err" field set to the error string. Note that this
- * table is never a valid reply by proper commands, since the returned
- * tables are otherwise always indexed by integers, never by strings. */
-void luaPushError(lua_State *lua, char *error) {
+ * with an "err" field set to the error string including the error code.
+ * Note that this table is never a valid reply by proper commands,
+ * since the returned tables are otherwise always indexed by integers, never by strings.
+ *
+ * The function takes ownership on the given err_buffer. */
+void luaPushErrorBuff(lua_State *lua, sds err_buffer) {
     sds msg;
+    sds error_code;
 
     /* If debugging is active and in step mode, log errors resulting from
      * Redis commands. */
     if (ldbIsEnabled()) {
-        ldbLog(sdscatprintf(sdsempty(),"<error> %s",error));
+        ldbLog(sdscatprintf(sdsempty(),"<error> %s",err_buffer));
     }
 
-    lua_newtable(lua);
-    lua_pushstring(lua,"err");
-
     /* There are two possible formats for the received `error` string:
      * 1) "-CODE msg": in this case we remove the leading '-' since we don't store it as part of the lua error format.
      * 2) "msg": in this case we prepend a generic 'ERR' code since all error statuses need some error code.
      * We support format (1) so this function can reuse the error messages used in other places in redis.
      * We support format (2) so it'll be easy to pass descriptive errors to this function without worrying about format.
      */
-    if (error[0] == '-')
-        msg = sdsnew(error+1);
-    else
-        msg = sdscatprintf(sdsempty(), "ERR %s", error);
+    if (err_buffer[0] == '-') {
+        /* derive error code from the message */
+        char *err_msg = strstr(err_buffer, " ");
+        if (!err_msg) {
+            msg = sdsnew(err_buffer+1);
+            error_code = sdsnew("ERR");
+        } else {
+            *err_msg = '\0';
+            msg = sdsnew(err_msg+1);
+            error_code = sdsnew(err_buffer + 1);
+        }
+        sdsfree(err_buffer);
+    } else {
+        msg = err_buffer;
+        error_code = sdsnew("ERR");
+    }
     /* Trim newline at end of string. If we reuse the ready-made Redis error objects (case 1 above) then we might
      * have a newline that needs to be trimmed. In any case the lua Redis error table shouldn't end with a newline. */
     msg = sdstrim(msg, "\r\n");
-    lua_pushstring(lua, msg);
-    sdsfree(msg);
+    sds final_msg = sdscatfmt(error_code, " %s", msg);
+
+    lua_newtable(lua);
+    lua_pushstring(lua,"err");
+    lua_pushstring(lua, final_msg);
     lua_settable(lua,-3);
+
+    sdsfree(msg);
+    sdsfree(final_msg);
+}
+
+void luaPushError(lua_State *lua, const char *error) {
+    luaPushErrorBuff(lua, sdsnew(error));
 }
 
 /* In case the error set into the Lua stack by luaPushError() was generated
  * by the non-error-trapping version of redis.pcall(), which is redis.call(),
  * this function will raise the Lua error so that the execution of the
  * script will be halted. */
-int luaRaiseError(lua_State *lua) {
-    lua_pushstring(lua,"err");
-    lua_gettable(lua,-2);
+int luaError(lua_State *lua) {
     return lua_error(lua);
 }
 
@@ -517,8 +540,15 @@ static void luaReplyToRedisReply(client *c, client* script_client, lua_State *lu
         lua_gettable(lua,-2);
         t = lua_type(lua,-1);
         if (t == LUA_TSTRING) {
-            addReplyErrorFormat(c,"-%s",lua_tostring(lua,-1));
-            lua_pop(lua,2);
+            lua_pop(lua, 1); /* pop the error message, we will use luaExtractErrorInformation to get error information */
+            errorInfo err_info = {0};
+            luaExtractErrorInformation(lua, &err_info);
+            addReplyErrorFormatEx(c,
+                                  err_info.ignore_err_stats_update? ERR_REPLY_FLAG_NO_STATS_UPDATE: 0,
+                                  "-%s",
+                                  err_info.msg);
+            luaErrorInformationDiscard(&err_info);
+            lua_pop(lua,1); /* pop the result table */
             return;
         }
         lua_pop(lua,1); /* Discard field name pushed before. */
@@ -719,7 +749,7 @@ static int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     scriptRunCtx* rctx = luaGetFromRegistry(lua, REGISTRY_RUN_CTX_NAME);
     if (!rctx) {
         luaPushError(lua, "redis.call/pcall can only be called inside a script invocation");
-        return luaRaiseError(lua);
+        return luaError(lua);
     }
     sds err = NULL;
     client* c = rctx->c;
@@ -728,7 +758,7 @@ static int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     int argc;
     robj **argv = luaArgsToRedisArgv(lua, &argc);
     if (argv == NULL) {
-        return raise_error ? luaRaiseError(lua) : 1;
+        return raise_error ? luaError(lua) : 1;
     }
 
     static int inuse = 0;   /* Recursive calls detection. */
@@ -767,6 +797,11 @@ static int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     if (err) {
         luaPushError(lua, err);
         sdsfree(err);
+        /* push a field indicate to ignore updating the stats on this error
+         * because it was already updated when executing the command. */
+        lua_pushstring(lua,"ignore_error_stats_update");
+        lua_pushboolean(lua, true);
+        lua_settable(lua,-3);
         goto cleanup;
     }
 
@@ -811,11 +846,40 @@ cleanup:
         /* If we are here we should have an error in the stack, in the
          * form of a table with an "err" field. Extract the string to
          * return the plain error. */
-        return luaRaiseError(lua);
+        return luaError(lua);
     }
     return 1;
 }
 
+/* Our implementation to lua pcall.
+ * We need this implementation for backward
+ * comparability with older Redis versions.
+ *
+ * On Redis 7, the error object is a table,
+ * compare to older version where the error
+ * object is a string. To keep backward
+ * comparability we catch the table object
+ * and just return the error message. */
+static int luaRedisPcall(lua_State *lua) {
+    int argc = lua_gettop(lua);
+    lua_pushboolean(lua, 1); /* result place holder */
+    lua_insert(lua, 1);
+    if (lua_pcall(lua, argc - 1, LUA_MULTRET, 0)) {
+        /* Error */
+        lua_remove(lua, 1); /* remove the result place holder, now we have room for at least one element */
+        if (lua_istable(lua, -1)) {
+            lua_getfield(lua, -1, "err");
+            if (lua_isstring(lua, -1)) {
+                lua_replace(lua, -2); /* replace the error message with the table */
+            }
+        }
+        lua_pushboolean(lua, 0); /* push result */
+        lua_insert(lua, 1);
+    }
+    return lua_gettop(lua);
+
+}
+
 /* redis.call() */
 static int luaRedisCallCommand(lua_State *lua) {
     return luaRedisGenericCommand(lua,1);
@@ -835,8 +899,8 @@ static int luaRedisSha1hexCommand(lua_State *lua) {
     char *s;
 
     if (argc != 1) {
-        lua_pushstring(lua, "wrong number of arguments");
-        return lua_error(lua);
+        luaPushError(lua, "wrong number of arguments");
+        return luaError(lua);
     }
 
     s = (char*)lua_tolstring(lua,1,&len);
@@ -867,7 +931,21 @@ static int luaRedisReturnSingleFieldTable(lua_State *lua, char *field) {
 
 /* redis.error_reply() */
 static int luaRedisErrorReplyCommand(lua_State *lua) {
-    return luaRedisReturnSingleFieldTable(lua,"err");
+    if (lua_gettop(lua) != 1 || lua_type(lua,-1) != LUA_TSTRING) {
+        luaPushError(lua, "wrong number or type of arguments");
+        return 1;
+    }
+
+    /* add '-' if not exists */
+    const char *err = lua_tostring(lua, -1);
+    sds err_buff = NULL;
+    if (err[0] != '-') {
+        err_buff = sdscatfmt(sdsempty(), "-%s", err);
+    } else {
+        err_buff = sdsnew(err);
+    }
+    luaPushErrorBuff(lua, err_buff);
+    return 1;
 }
 
 /* redis.status_reply() */
@@ -884,19 +962,19 @@ static int luaRedisSetReplCommand(lua_State *lua) {
 
     scriptRunCtx* rctx = luaGetFromRegistry(lua, REGISTRY_RUN_CTX_NAME);
     if (!rctx) {
-        lua_pushstring(lua, "redis.set_repl can only be called inside a script invocation");
-        return lua_error(lua);
+        luaPushError(lua, "redis.set_repl can only be called inside a script invocation");
+        return luaError(lua);
     }
 
     if (argc != 1) {
-         lua_pushstring(lua, "redis.set_repl() requires two arguments.");
-         return lua_error(lua);
+        luaPushError(lua, "redis.set_repl() requires two arguments.");
+         return luaError(lua);
     }
 
     flags = lua_tonumber(lua,-1);
     if ((flags & ~(PROPAGATE_AOF|PROPAGATE_REPL)) != 0) {
-        lua_pushstring(lua, "Invalid replication flags. Use REPL_AOF, REPL_REPLICA, REPL_ALL or REPL_NONE.");
-        return lua_error(lua);
+        luaPushError(lua, "Invalid replication flags. Use REPL_AOF, REPL_REPLICA, REPL_ALL or REPL_NONE.");
+        return luaError(lua);
     }
 
     scriptSetRepl(rctx, flags);
@@ -909,8 +987,8 @@ static int luaRedisSetReplCommand(lua_State *lua) {
 static int luaRedisAclCheckCmdPermissionsCommand(lua_State *lua) {
     scriptRunCtx* rctx = luaGetFromRegistry(lua, REGISTRY_RUN_CTX_NAME);
     if (!rctx) {
-        lua_pushstring(lua, "redis.acl_check_cmd can only be called inside a script invocation");
-        return lua_error(lua);
+        luaPushError(lua, "redis.acl_check_cmd can only be called inside a script invocation");
+        return luaError(lua);
     }
     int raise_error = 0;
 
@@ -918,12 +996,12 @@ static int luaRedisAclCheckCmdPermissionsCommand(lua_State *lua) {
     robj **argv = luaArgsToRedisArgv(lua, &argc);
 
     /* Require at least one argument */
-    if (argv == NULL) return lua_error(lua);
+    if (argv == NULL) return luaError(lua);
 
     /* Find command */
     struct redisCommand *cmd;
     if ((cmd = lookupCommand(argv, argc)) == NULL) {
-        lua_pushstring(lua, "Invalid command passed to redis.acl_check_cmd()");
+        luaPushError(lua, "Invalid command passed to redis.acl_check_cmd()");
         raise_error = 1;
     } else {
         int keyidxptr;
@@ -937,7 +1015,7 @@ static int luaRedisAclCheckCmdPermissionsCommand(lua_State *lua) {
     while (argc--) decrRefCount(argv[argc]);
     zfree(argv);
     if (raise_error)
-        return lua_error(lua);
+        return luaError(lua);
     else
         return 1;
 }
@@ -950,16 +1028,16 @@ static int luaLogCommand(lua_State *lua) {
     sds log;
 
     if (argc < 2) {
-        lua_pushstring(lua, "redis.log() requires two arguments or more.");
-        return lua_error(lua);
+        luaPushError(lua, "redis.log() requires two arguments or more.");
+        return luaError(lua);
     } else if (!lua_isnumber(lua,-argc)) {
-        lua_pushstring(lua, "First argument must be a number (log level).");
-        return lua_error(lua);
+        luaPushError(lua, "First argument must be a number (log level).");
+        return luaError(lua);
     }
     level = lua_tonumber(lua,-argc);
     if (level < LL_DEBUG || level > LL_WARNING) {
-        lua_pushstring(lua, "Invalid debug level.");
-        return lua_error(lua);
+        luaPushError(lua, "Invalid debug level.");
+        return luaError(lua);
     }
     if (level < server.verbosity) return 0;
 
@@ -984,20 +1062,20 @@ static int luaLogCommand(lua_State *lua) {
 static int luaSetResp(lua_State *lua) {
     scriptRunCtx* rctx = luaGetFromRegistry(lua, REGISTRY_RUN_CTX_NAME);
     if (!rctx) {
-        lua_pushstring(lua, "redis.setresp can only be called inside a script invocation");
-        return lua_error(lua);
+        luaPushError(lua, "redis.setresp can only be called inside a script invocation");
+        return luaError(lua);
     }
     int argc = lua_gettop(lua);
 
     if (argc != 1) {
-        lua_pushstring(lua, "redis.setresp() requires one argument.");
-        return lua_error(lua);
+        luaPushError(lua, "redis.setresp() requires one argument.");
+        return luaError(lua);
     }
 
     int resp = lua_tonumber(lua,-argc);
     if (resp != 2 && resp != 3) {
-        lua_pushstring(lua, "RESP version must be 2 or 3.");
-        return lua_error(lua);
+        luaPushError(lua, "RESP version must be 2 or 3.");
+        return luaError(lua);
     }
     scriptSetResp(rctx, resp);
     return 0;
@@ -1197,6 +1275,9 @@ void luaRegisterRedisAPI(lua_State* lua) {
     luaLoadLibraries(lua);
     luaRemoveUnsupportedFunctions(lua);
 
+    lua_pushcfunction(lua,luaRedisPcall);
+    lua_setglobal(lua, "pcall");
+
     /* Register the redis commands table and fields */
     lua_newtable(lua);
 
@@ -1357,11 +1438,50 @@ static void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
          */
         lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0);
 
-        lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
-        lua_error(lua);
+        luaPushError(lua,"Script killed by user with SCRIPT KILL...");
+        luaError(lua);
     }
 }
 
+void luaErrorInformationDiscard(errorInfo *err_info) {
+    if (err_info->msg) sdsfree(err_info->msg);
+    if (err_info->source) sdsfree(err_info->source);
+    if (err_info->line) sdsfree(err_info->line);
+}
+
+void luaExtractErrorInformation(lua_State *lua, errorInfo *err_info) {
+    if (lua_isstring(lua, -1)) {
+        err_info->msg = sdscatfmt(sdsempty(), "ERR %s", lua_tostring(lua, -1));
+        err_info->line = NULL;
+        err_info->source = NULL;
+        err_info->ignore_err_stats_update = 0;
+    }
+
+    lua_getfield(lua, -1, "err");
+    if (lua_isstring(lua, -1)) {
+        err_info->msg = sdsnew(lua_tostring(lua, -1));
+    }
+    lua_pop(lua, 1);
+
+    lua_getfield(lua, -1, "source");
+    if (lua_isstring(lua, -1)) {
+        err_info->source = sdsnew(lua_tostring(lua, -1));
+    }
+    lua_pop(lua, 1);
+
+    lua_getfield(lua, -1, "line");
+    if (lua_isstring(lua, -1)) {
+        err_info->line = sdsnew(lua_tostring(lua, -1));
+    }
+    lua_pop(lua, 1);
+
+    lua_getfield(lua, -1, "ignore_error_stats_update");
+    if (lua_isboolean(lua, -1)) {
+        err_info->ignore_err_stats_update = lua_toboolean(lua, -1);
+    }
+    lua_pop(lua, 1);
+}
+
 void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t nkeys, robj** args, size_t nargs, int debug_enabled) {
     client* c = run_ctx->original_client;
     int delhook = 0;
@@ -1419,9 +1539,28 @@ void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t
     }
 
     if (err) {
-        addReplyErrorFormat(c,"Error running script (call to %s): %s\n",
-                run_ctx->funcname, lua_tostring(lua,-1));
-        lua_pop(lua,1); /* Consume the Lua reply and remove error handler. */
+        /* Error object is a table of the following format:
+         * {err='<error msg>', source='<source file>', line=<line>}
+         * We can construct the error message from this information */
+        if (!lua_istable(lua, -1)) {
+            /* Should not happened, and we should considered assert it */
+            addReplyErrorFormat(c,"Error running script (call to %s)\n", run_ctx->funcname);
+        } else {
+            errorInfo err_info = {0};
+            sds final_msg = sdsempty();
+            luaExtractErrorInformation(lua, &err_info);
+            final_msg = sdscatfmt(final_msg, "-%s",
+                                  err_info.msg);
+            if (err_info.line && err_info.source) {
+                final_msg = sdscatfmt(final_msg, " script: %s, on %s:%s.",
+                                      run_ctx->funcname,
+                                      err_info.source,
+                                      err_info.line);
+            }
+            addReplyErrorSdsEx(c, final_msg, err_info.ignore_err_stats_update? ERR_REPLY_FLAG_NO_STATS_UPDATE : 0);
+            luaErrorInformationDiscard(&err_info);
+        }
+        lua_pop(lua,1); /* Consume the Lua error */
     } else {
         /* On success convert the Lua return value into Redis protocol, and
          * send it to * the client. */