Fix module commandresult event cleanup during unsubscribe and module unload (valkey-io#3545)

This follows up on the commandresult API work and fixes cleanup around
unsubscribe and module unload.

The main issue was that command-result event listeners could leave stale
state behind. On unload, we removed the listeners themselves but didn’t
fully update the fast-path listener counters. Separately, unsubscribing
with a NULL callback could behave badly if the listener wasn’t present
anymore. In practice, that meant later commands could still walk into
command-result event handling after the module was supposed to be
cleaned up.

Failed in Daily as well yesterday:
https://github.com/valkey-io/valkey/actions/runs/24753491944/job/72421581610#step:10:852
Related Failures:
valkey-io#2936 (comment)

---------

Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>

Author: sarthakaggarwal97

src/module.c

@@ -12718,6 +12718,8 @@ int VM_SubscribeToServerEvent(ValkeyModuleCtx *ctx, ValkeyModuleEvent event, Val
         return VALKEYMODULE_OK;
     }
 
+    if (callback == NULL) return VALKEYMODULE_OK;
+
     /* No event found, we need to add a new one. */
     el = zmalloc(sizeof(*el));
     el->module = ctx->module;
@@ -12877,11 +12879,19 @@ void moduleUnsubscribeAllServerEvents(ValkeyModule *module) {
     ValkeyModuleEventListener *el;
     listIter li;
     listNode *ln;
-    listRewind(ValkeyModule_EventListeners, &li);
 
+    listRewind(ValkeyModule_EventListeners, &li);
     while ((ln = listNext(&li))) {
         el = ln->value;
         if (el->module == module) {
+            if (el->event.id == VALKEYMODULE_EVENT_COMMAND_RESULT_SUCCESS)
+                commandResultSuccessListeners--;
+            else if (el->event.id == VALKEYMODULE_EVENT_COMMAND_RESULT_FAILURE)
+                commandResultFailureListeners--;
+            else if (el->event.id == VALKEYMODULE_EVENT_COMMAND_RESULT_REJECTED)
+                commandResultRejectedListeners--;
+            else if (el->event.id == VALKEYMODULE_EVENT_COMMAND_RESULT_ACL_REJECTED)
+                commandResultACLRejectedListeners--;
             listDelNode(ValkeyModule_EventListeners, ln);
             zfree(el);
         }

tests/modules/commandresult.c

@@ -62,6 +62,14 @@ static int log_count = 0;
 #define MODE_ACL_REJECTED 0x8
 static int subscription_mode = 0;
 
+static void ResetState(void) {
+  memset(&stats, 0, sizeof(stats));
+  memset(result_log, 0, sizeof(result_log));
+  log_head = 0;
+  log_count = 0;
+  subscription_mode = 0;
+}
+
 /* Add entry to circular log */
 void LogResult(const char *cmd_name, int status, uint64_t subevent,
                long long duration, long long dirty,
@@ -243,30 +251,23 @@ int CmdResultRegister_ValkeyCommand(ValkeyModuleCtx *ctx,
 int CmdResultUnsubscribe_ValkeyCommand(ValkeyModuleCtx *ctx,
                                        ValkeyModuleString **argv, int argc) {
   VALKEYMODULE_NOT_USED(argv);
+  int had_subscription = (subscription_mode != 0);
 
   if (argc != 1) {
     return ValkeyModule_WrongArity(ctx);
   }
 
-  if (subscription_mode == 0) {
+  ValkeyModule_SubscribeToServerEvent(ctx, ValkeyModuleEvent_CommandResultSuccess, NULL);
+  ValkeyModule_SubscribeToServerEvent(ctx, ValkeyModuleEvent_CommandResultFailure, NULL);
+  ValkeyModule_SubscribeToServerEvent(ctx, ValkeyModuleEvent_CommandResultRejected, NULL);
+  ValkeyModule_SubscribeToServerEvent(ctx, ValkeyModuleEvent_CommandResultACLRejected, NULL);
+  subscription_mode = 0;
+
+  if (!had_subscription) {
     return ValkeyModule_ReplyWithError(
         ctx, "ERR not subscribed to command result events");
   }
 
-  if (subscription_mode & MODE_SUCCESS)
-    ValkeyModule_SubscribeToServerEvent(
-        ctx, ValkeyModuleEvent_CommandResultSuccess, NULL);
-  if (subscription_mode & MODE_FAILURE)
-    ValkeyModule_SubscribeToServerEvent(
-        ctx, ValkeyModuleEvent_CommandResultFailure, NULL);
-  if (subscription_mode & MODE_REJECTED)
-    ValkeyModule_SubscribeToServerEvent(
-        ctx, ValkeyModuleEvent_CommandResultRejected, NULL);
-  if (subscription_mode & MODE_ACL_REJECTED)
-    ValkeyModule_SubscribeToServerEvent(
-        ctx, ValkeyModuleEvent_CommandResultACLRejected, NULL);
-
-  subscription_mode = 0;
   return ValkeyModule_ReplyWithSimpleString(ctx, "OK");
 }
 
@@ -440,6 +441,8 @@ int ValkeyModule_OnLoad(ValkeyModuleCtx *ctx, ValkeyModuleString **argv,
   VALKEYMODULE_NOT_USED(argv);
   VALKEYMODULE_NOT_USED(argc);
 
+  ResetState();
+
   if (ValkeyModule_Init(ctx, "commandresult", 1, VALKEYMODULE_APIVER_1) ==
       VALKEYMODULE_ERR) {
     return VALKEYMODULE_ERR;

tests/unit/moduleapi/commandresult.tcl

@@ -71,11 +71,9 @@ start_server {tags {"modules"}} {
         cleanup_callback
         r cmdresult.register all
 
-        r cmdresult.success
-        r ping
+        r eval {local sum = 0; for i = 1, 100000 do sum = sum + i end; return sum} 0
 
         set stats [r cmdresult.stats]
-        # Duration should be > 0 microseconds
         assert {[dict get $stats total_duration_us] > 0}
 
         r cmdresult.unsubscribe
@@ -267,11 +265,15 @@ start_server {tags {"modules"}} {
         set stats [r cmdresult.stats]
         assert {[dict get $stats total_callbacks] >= 2}
 
-        # Unload module while subscription is still active
+        # Unload module while subscription is still active.
         assert_equal {OK} [r module unload commandresult]
 
-        # Reload module for remaining tests
+        # Unsubscribing after reload has no matching listener. It must not add
+        # NULL callbacks that would be invoked by later command result events.
         r module load $testmodule
+        catch {r cmdresult.unsubscribe} err
+        assert_match {*not subscribed*} $err
+        assert_equal {PONG} [r ping]
     }
 
     test {Module commandresult - Multiple callbacks from different operations} {
@@ -627,6 +629,9 @@ start_server {tags {"modules"}} {
 
     test {Module commandresult - rejected: command not allowed in Pub/Sub context (PUBSUB)} {
         cleanup_callback
+        if {$::force_resp3} {
+            skip "RESP3 Pub/Sub clients may issue arbitrary commands"
+        }
         r cmdresult.register rejected
 
         set rd [valkey_deferring_client]