XREADGROUP: Unblock client if stream is deleted (#10306)

Deleting a stream while a client is blocked XREADGROUP should unblock the client.

The idea is that if a client is blocked via XREADGROUP is different from
any other blocking type in the sense that it depends on the existence of both
the key and the group. Even if the key is deleted and then revived with XADD
it won't help any clients blocked on XREADGROUP because the group no longer
exist, so they would fail with -NOGROUP anyway.
The conclusion is that it's better to unblock these clients (with error) upon
the deletion of the key, rather than waiting for the first XADD. 

Other changes:
1. Slightly optimize all `serveClientsBlockedOn*` functions by checking `server.blocked_clients_by_type`
2. All `serveClientsBlockedOn*` functions now use a list iterator rather than looking at `listFirst`, relying
  on `unblockClient` to delete the head of the list. Before this commit, only `serveClientsBlockedOnStreams`
  used to work like that.
3. bugfix: CLIENT UNBLOCK ERROR should work even if the command doesn't have a timeout_callback
  (only relevant to module commands)

1 files changed, 10 insertions, 3 deletions

diff --git a/tests/modules/blockonkeys.c b/tests/modules/blockonkeys.c
index 4568b9fa8..1aa576489 100644
--- a/tests/modules/blockonkeys.c
+++ b/tests/modules/blockonkeys.c
@@ -135,22 +135,29 @@ int bpop_timeout_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int arg
     return RedisModule_ReplyWithSimpleString(ctx, "Request timedout");
 }
 
-/* FSL.BPOP <key> <timeout> - Block clients until list has two or more elements.
+/* FSL.BPOP <key> <timeout> [NO_TO_CB]- Block clients until list has two or more elements.
  * When that happens, unblock client and pop the last two elements (from the right). */
 int fsl_bpop(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
-    if (argc != 3)
+    if (argc < 3)
         return RedisModule_WrongArity(ctx);
 
     long long timeout;
     if (RedisModule_StringToLongLong(argv[2],&timeout) != REDISMODULE_OK || timeout < 0)
         return RedisModule_ReplyWithError(ctx,"ERR invalid timeout");
 
+    int to_cb = 1;
+    if (argc == 4) {
+        if (strcasecmp("NO_TO_CB", RedisModule_StringPtrLen(argv[3], NULL)))
+            return RedisModule_ReplyWithError(ctx,"ERR invalid argument");
+        to_cb = 0;
+    }
+
     fsl_t *fsl;
     if (!get_fsl(ctx, argv[1], REDISMODULE_READ, 0, &fsl, 1))
         return REDISMODULE_OK;
 
     if (!fsl) {
-        RedisModule_BlockClientOnKeys(ctx, bpop_reply_callback, bpop_timeout_callback,
+        RedisModule_BlockClientOnKeys(ctx, bpop_reply_callback, to_cb ? bpop_timeout_callback : NULL,
                                       NULL, timeout, &argv[1], 1, NULL);
     } else {
         RedisModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);