MOD-14124 Handle shard error replies in internal command execution (#85)

* Handle shard error replies in internal command execution

When a shard replies with a Redis error (e.g. NOPERM) to an internal
command, the framework previously either silently disconnected
(cluster.c rejected non-ARRAY replies) or let the error reach
application reply parsers, causing assertion failures.

Changes:
- Allow REDIS_REPLY_ERROR through MR_OnDataResponseArrived to
  MR_SetInternalCommandResults instead of disconnecting (cluster.c)
- Add MR_SetShardError() for top-level shard errors: captures the
  error in e->errors, marks all steps done with NULL result
  placeholders, and advances the nodesDone counter so the execution
  completes normally
- Handle per-element errors: if an individual internal command reply
  is an error, capture it in e->errors instead of calling replyParser
- Add NULL guard to MR_RecordFree for safe cleanup of NULL placeholders

Errors propagate through the existing e->errors mechanism, which
application-level done callbacks already consume via
check_and_reply_on_error().

* ## Summary
- Homebrew's LLVM `libclang` on macOS causes `bindgen` to generate an opaque
  `Record` struct (with only `_address: u8`) instead of the actual definition
  containing `recordType: *mut MRRecordType`, failing the Rust compilation.
- Added a post-processing step in `build.rs` that detects when bindgen produces
  an opaque `Record` and patches the generated bindings with the correct field.
- Bumped `bindgen` dependency from `0.70` to `0.72`.

## Test plan
- [ ] macOS CI pipeline passes (`make run_tests`)
- [ ] Linux CI pipeline still passes (no change in behavior when Record is generated correctly)

* libmr error remove

* remove dead code

* fix remarks

* remove null check

Author: TalBarYakar

Cargo.toml

@@ -15,7 +15,7 @@ libc = "0.2"
 linkme = "0.3"
 
 [build-dependencies]
-bindgen = "0.70"
+bindgen = "0.72"
 
 [lib]
 crate-type = ["rlib"]

src/mr.c

@@ -1180,17 +1180,25 @@ void MR_SetInternalCommandResults(unsigned short nodeIndex, redisReply* reply, E
         ++mrCtx.stats.nMissedExecutions;
         return;
     }
+
     RedisModule_Assert(reply->type == REDIS_REPLY_ARRAY && reply->elements > 0);
     RedisModule_Assert(array_len(e->steps) == reply->elements);
     size_t nodesDone = 0;
     for (size_t i = 0; i < reply->elements; i++) {
         Step *s = e->steps + i;
-        s->internalCommand.reply = reply->element[i]; // First set the reply to be parsed
-        Record *record = s->internalCommand.replyParser(s->internalCommand.reply);  // Then parse it
-        s->internalCommand.reply = NULL;  // And keep things tidy
+        s->internalCommand.reply = reply->element[i];
+
+        // Per-element error: one internal command failed (e.g. NOPERM on a key).
+        // Capture the error instead of calling the replyParser.
+        if (s->internalCommand.reply->type == REDIS_REPLY_ERROR) {
+            e->errors = array_append(e->errors, MR_ErrorRecordCreate(s->internalCommand.reply->str));
+            s->internalCommand.reply = NULL;
+        } else {
+            Record *record = s->internalCommand.replyParser(s->internalCommand.reply);
+            s->internalCommand.reply = NULL;
+            e->results = array_append(e->results, record);
+        }
 
-        e->results = array_append(e->results, record);
-        // All steps should return the same number of done nodes because we update all steps of a single node in one go
         if (nodesDone == 0)
             nodesDone = MR_PerformStepDoneOp(e, i);
         else if (nodesDone != MR_PerformStepDoneOp(e, i))
@@ -1199,7 +1207,6 @@ void MR_SetInternalCommandResults(unsigned short nodeIndex, redisReply* reply, E
     }
 
     if (nodesDone == MR_ClusterGetSize()) {
-        // All nodes (shards), including myself, have answered
         MR_ExecutionAddTask(e, MR_RunExecution, NULL);
     }
 }

src/redismodule.h

@@ -1379,6 +1379,7 @@ REDISMODULE_API size_t (*RedisModule_MallocSizeDict)(RedisModuleDict* dict) REDI
 REDISMODULE_API RedisModuleUser * (*RedisModule_CreateModuleUser)(const char *name) REDISMODULE_ATTR;
 REDISMODULE_API void (*RedisModule_FreeModuleUser)(RedisModuleUser *user) REDISMODULE_ATTR;
 REDISMODULE_API void (*RedisModule_SetContextUser)(RedisModuleCtx *ctx, const RedisModuleUser *user) REDISMODULE_ATTR;
+REDISMODULE_API const RedisModuleUser * (*RedisModule_GetContextUser)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_SetModuleUserACL)(RedisModuleUser *user, const char* acl) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_SetModuleUserACLString)(RedisModuleCtx * ctx, RedisModuleUser *user, const char* acl, RedisModuleString **error) REDISMODULE_ATTR;
 REDISMODULE_API RedisModuleString * (*RedisModule_GetModuleUserACLString)(RedisModuleUser *user) REDISMODULE_ATTR;
@@ -1778,6 +1779,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(CreateModuleUser);
     REDISMODULE_GET_API(FreeModuleUser);
     REDISMODULE_GET_API(SetContextUser);
+    REDISMODULE_GET_API(GetContextUser);
     REDISMODULE_GET_API(SetModuleUserACL);
     REDISMODULE_GET_API(SetModuleUserACLString);
     REDISMODULE_GET_API(GetModuleUserACLString);