Pass null-terminated node ID for VM_RegisterClusterMessageReceiver and add test coverage (#1708)

* Pass `sender_id` as `NULL` terminated string as part of
`ValkeyModuleClusterMessageReceiver` for ease of usage by the module(s).

* Implement test coverage described in #1656 and ensures that nodes in
the cluster module properly acknowledge a "DING" message by sending a
"DONG" response.

Closes #1656.

---------

Signed-off-by: Nikhil Manglore <[email protected]>

Author: Nikhil-Manglore

src/cluster_legacy.c

@@ -3045,7 +3045,10 @@ static void clusterProcessModulePacket(clusterMsgModule *module_data, clusterNod
     uint32_t len = ntohl(module_data->len);
     uint8_t type = module_data->type;
     unsigned char *payload = module_data->bulk_data;
-    moduleCallClusterReceivers(sender->name, module_id, type, payload, len);
+
+    sds sender_name = sdsnewlen(sender->name, CLUSTER_NAMELEN);
+    moduleCallClusterReceivers(sender_name, module_id, type, payload, len);
+    sdsfree(sender_name);
 }
 
 static void clusterProcessLightPacket(clusterNode *sender, clusterLink *link, uint16_t type) {

src/module.c

@@ -8950,7 +8950,13 @@ void moduleCallClusterReceivers(const char *sender_id,
  * was already a registered callback, this will replace the callback function
  * with the one provided, otherwise if the callback is set to NULL and there
  * is already a callback for this function, the callback is unregistered
- * (so this API call is also used in order to delete the receiver). */
+ * (so this API call is also used in order to delete the receiver).
+ *
+ * When a message of this type is received, the registered callback function
+ * will be invoked with details, including the 40-byte node ID of the sender.
+ *
+ * In Valkey 8.1 and later, the node ID is null-terminated. Prior to 8.1, it was
+ * not null-terminated */
 void VM_RegisterClusterMessageReceiver(ValkeyModuleCtx *ctx,
                                        uint8_t type,
                                        ValkeyModuleClusterMessageReceiver callback) {

tests/modules/cluster.c

@@ -46,6 +46,15 @@ int PingallCommand(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int argc) {
     return ValkeyModule_ReplyWithSimpleString(ctx, "OK");
 }
 
+void DingReceiver(ValkeyModuleCtx *ctx, const char *sender_id, uint8_t type, const unsigned char *payload, uint32_t len) {
+    ValkeyModule_Log(ctx, "notice", "DING (type %d) RECEIVED from %.*s: '%.*s'", type, VALKEYMODULE_NODE_ID_LEN, sender_id, (int)len, payload);
+    ValkeyModule_SendClusterMessage(ctx, sender_id, MSGTYPE_DONG, "Message Received!", 17);
+}
+
+void DongReceiver(ValkeyModuleCtx *ctx, const char *sender_id, uint8_t type, const unsigned char *payload, uint32_t len) {
+    ValkeyModule_Log(ctx, "notice", "DONG (type %d) RECEIVED from %.*s: '%.*s'", type, VALKEYMODULE_NODE_ID_LEN, sender_id, (int)len, payload);
+}
+
 int ValkeyModule_OnLoad(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int argc) {
     VALKEYMODULE_NOT_USED(argv);
     VALKEYMODULE_NOT_USED(argc);
@@ -63,5 +72,8 @@ int ValkeyModule_OnLoad(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int arg
     if (ValkeyModule_CreateCommand(ctx, "test.cluster_shards", test_cluster_shards, "", 0, 0, 0) == VALKEYMODULE_ERR)
         return VALKEYMODULE_ERR;
 
+    /* Register our handlers for different message types. */
+    ValkeyModule_RegisterClusterMessageReceiver(ctx, MSGTYPE_DING, DingReceiver);
+    ValkeyModule_RegisterClusterMessageReceiver(ctx, MSGTYPE_DONG, DongReceiver);
     return VALKEYMODULE_OK;
 }

tests/unit/moduleapi/cluster.tcl

@@ -22,6 +22,25 @@ start_cluster 3 0 [list config_lines $modules] {
             fail "node 2 or node 3 didn't receive cluster module message"
         }
     }
+
+    $node1 CONFIG RESETSTAT
+    $node2 CONFIG RESETSTAT
+    $node3 CONFIG RESETSTAT
+
+    test "Cluster module message DING/DONG acknowledgment" {
+        assert_equal OK [$node1 test.pingall]
+        wait_for_condition 50 100 {
+            [CI 0 cluster_stats_messages_module_received] eq 2 &&
+            [CI 1 cluster_stats_messages_module_received] eq 1 &&
+            [CI 2 cluster_stats_messages_module_received] eq 1
+        } else {
+            fail "node 2 or node 3 didn't receive DING messages or node 1 didn't receive DONG message"
+        }
+
+        verify_log_message -1 "*DING (type 1) RECEIVED*Hey*" 0
+        verify_log_message -2 "*DING (type 1) RECEIVED*Hey*" 0
+        assert_equal 2 [count_log_message 0 "* <cluster> DONG (type 2) RECEIVED*"]
+    }
 }
 
 set testmodule_nokey [file normalize tests/modules/blockonbackground.so]