Fixed tracking of command duration for multi/eval/module/wait (#11970)

In #11012, we changed the way command durations were computed to handle the same command being executed multiple times. This commit fixes some misses from that commit.

* Wait commands were not correctly reporting their duration if the timeout was reached.
* Multi/scripts/and modules with RM_Call were not properly resetting the duration between inner calls, leading to them reporting cumulative duration.
* When a blocked client is freed, the call and duration are always discarded.

This commit also adds an assert if the duration is not properly reset, potentially indicating that a report to call statistics was missed. The assert potentially be removed in the future, as it's mainly intended to detect misses in tests.

src/blocked.c

@@ -114,6 +114,7 @@ void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_err
         updateCommandLatencyHistogram(&(c->lastcmd->latency_histogram), total_cmd_duration*1000);
     /* Log the command into the Slow log if needed. */
     slowlogPushCurrentCommand(c, c->lastcmd, total_cmd_duration);
+    c->duration = 0;
     /* Log the reply duration event. */
     latencyAddSampleIfNeeded("command-unblocking",reply_us/1000);
 }

src/module.c

@@ -647,6 +647,7 @@ void moduleReleaseTempClient(client *c) {
     clearClientConnectionState(c);
     listEmpty(c->reply);
     c->reply_bytes = 0;
+    c->duration = 0;
     resetClient(c);
     c->bufpos = 0;
     c->flags = CLIENT_MODULE;

src/networking.c

@@ -1583,6 +1583,8 @@ void freeClient(client *c) {
     c->querybuf = NULL;
 
     /* Deallocate structures used to block on blocking ops. */
+    /* If there is any in-flight command, we don't record their duration. */
+    c->duration = 0;
     if (c->flags & CLIENT_BLOCKED) unblockClient(c, 1);
     dictRelease(c->bstate.keys);
 
@@ -2039,8 +2041,10 @@ void resetClient(client *c) {
     c->multibulklen = 0;
     c->bulklen = -1;
     c->slot = -1;
-    c->duration = 0;
     c->flags &= ~CLIENT_EXECUTING_COMMAND;
+
+    /* Make sure the duration has been recorded to some command. */
+    serverAssert(c->duration == 0);
 #ifdef LOG_REQ_RES
     reqresReset(c, 1);
 #endif

src/replication.c

@@ -3606,6 +3606,7 @@ void unblockClientWaitingReplicas(client *c) {
     listNode *ln = listSearchKey(server.clients_waiting_acks,c);
     serverAssert(ln != NULL);
     listDelNode(server.clients_waiting_acks,ln);
+    updateStatsOnUnblock(c, 0, 0, 0);
 }
 
 /* Check if there are clients blocked in WAIT or WAITAOF that can be unblocked

src/server.c

@@ -3596,6 +3596,12 @@ void call(client *c, int flags) {
             updateCommandLatencyHistogram(&(real_cmd->latency_histogram), c->duration*1000);
     }
 
+    /* The duration needs to be reset after each call except for a blocked command,
+     * which is expected to record and reset the duration after unblocking. */
+    if (!(c->flags & CLIENT_BLOCKED)) {
+        c->duration = 0;
+    }
+
     /* Propagate the command into the AOF and replication link.
      * We never propagate EXEC explicitly, it will be implicitly
      * propagated if needed (see propagatePendingCommands).