Fix and improve module error reply statistics (#10278)

This PR handles several aspects
1. Calls to RM_ReplyWithError from thread safe contexts don't violate thread safety.
2. Errors returning from RM_Call to the module aren't counted in the statistics (they
  might be handled silently by the module)
3. When a module propagates a reply it got from RM_Call to it's client, then the error
  statistics are counted.

This is done by:
1. When appending an error reply to the output buffer, we avoid updating the global
  error statistics, instead we cache that error in a deferred list in the client struct.
2. When creating a RedisModuleCallReply object, the deferred error list is moved from
  the client into that object.
3. when a module calls RM_ReplyWithCallReply we copy the deferred replies to the dest
  client (if that's a real client, then that's when the error statistics are updated to the server)

Note about RM_ReplyWithCallReply: if the original reply had an array with errors, and the module
replied with just a portion of the original reply, and not the entire reply, the errors are currently not
propagated and the errors stats will not get propagated.

Fix #10180

1 files changed, 13 insertions, 2 deletions

diff --git a/src/call_reply.c b/src/call_reply.c
index 7aa79d089..3694db55e 100644
--- a/src/call_reply.c
+++ b/src/call_reply.c
@@ -60,7 +60,7 @@ struct CallReply {
         double d;        /* Reply value for double reply. */
         struct CallReply *array; /* Array of sub-reply elements. used for set, array, map, and attribute */
     } val;
-
+    list *deferred_error_list;   /* list of errors in sds form or NULL */
     struct CallReply *attribute; /* attribute reply, NULL if not exists */
 };
 
@@ -237,6 +237,8 @@ void freeCallReply(CallReply *rep) {
         freeCallReplyInternal(rep);
     }
     sdsfree(rep->original_proto);
+    if (rep->deferred_error_list)
+        listRelease(rep->deferred_error_list);
     zfree(rep);
 }
 
@@ -488,6 +490,11 @@ int callReplyIsResp3(CallReply *rep) {
     return rep->flags & REPLY_FLAG_RESP3;
 }
 
+/* Returns a list of errors in sds form, or NULL. */
+list *callReplyDeferredErrorList(CallReply *rep) {
+    return rep->deferred_error_list;
+}
+
 /* Create a new CallReply struct from the reply blob.
  *
  * The function will own the reply blob, so it must not be used or freed by
@@ -496,6 +503,9 @@ int callReplyIsResp3(CallReply *rep) {
  * The reply blob will be freed when the returned CallReply struct is later
  * freed using freeCallReply().
  *
+ * The deferred_error_list is an optional list of errors that are present
+ * in the reply blob, if given, this function will take ownership on it.
+ *
  * The private_data is optional and can later be accessed using
  * callReplyGetPrivateData().
  *
@@ -504,7 +514,7 @@ int callReplyIsResp3(CallReply *rep) {
  * DESIGNED TO HANDLE USER INPUT and using it to parse invalid replies is
  * unsafe.
  */
-CallReply *callReplyCreate(sds reply, void *private_data) {
+CallReply *callReplyCreate(sds reply, list *deferred_error_list, void *private_data) {
     CallReply *res = zmalloc(sizeof(*res));
     res->flags = REPLY_FLAG_ROOT;
     res->original_proto = reply;
@@ -512,5 +522,6 @@ CallReply *callReplyCreate(sds reply, void *private_data) {
     res->proto_len = sdslen(reply);
     res->private_data = private_data;
     res->attribute = NULL;
+    res->deferred_error_list = deferred_error_list;
     return res;
 }