fix(server): monitor lua at script execution (#767)

Signed-off-by: Boaz Sade <boaz@dragonflydb.io>

src/server/main_service.cc

@@ -644,6 +644,7 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx)
 
   uint64_t start_usec = ProactorBase::GetMonotonicTimeNs(), end_usec;
 
+  DispatchMonitorIfNeeded(cid->opt_mask() & CO::ADMIN, dfly_cntx, args);
   // Create command transaction
   intrusive_ptr<Transaction> dist_trans;
 
@@ -695,8 +696,6 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx)
     dfly_cntx->reply_builder()->CloseConnection();
   }
 
-  DispatchMonitorIfNeeded(cid->opt_mask() & CO::ADMIN, dfly_cntx, args);
-
   end_usec = ProactorBase::GetMonotonicTimeNs();
 
   request_latency_usec.IncBy(cmd_str, (end_usec - start_usec) / 1000);

tests/dragonfly/connection_test.py

@@ -4,14 +4,62 @@ import asyncio
 import aioredis
 import async_timeout
 
+async def run_monitor_eval(monitor, expected):
+    async with monitor as mon:
+        count = 0
+        max = len(expected)
+        while count < max:
+            try:
+                async with async_timeout.timeout(1):
+                    response = await mon.next_command()
+                    if "select" not in response["command"].lower():
+                        cmd = expected[count]
+                        if cmd not in response["command"]:
+                            print(f"command {response['command']} != {cmd}")
+                            return False
+                        else:
+                            count = count + 1
+            except Exception as e:
+                print(f"failed to monitor: {e}")
+                return False
+    return True
+
+'''
+Test issue https://github.com/dragonflydb/dragonfly/issues/756
+Monitor command do not return when we have lua script issue
+'''
+@pytest.mark.asyncio
+async def test_monitor_command_lua(async_pool):
+    expected = ["EVAL return redis", "GET bar", "EVAL return redis", "SET foo2"]
+
+    conn = aioredis.Redis(connection_pool=async_pool)
+    monitor = conn.monitor()
+
+    cmd1 = aioredis.Redis(connection_pool=async_pool)
+    future = asyncio.create_task(run_monitor_eval(monitor=monitor, expected=expected))
+    await asyncio.sleep(1)
+    try:
+        res = await cmd1.eval(r'return redis.call("GET", "bar")', 0)
+        assert False    # this will return an error
+    except Exception as e:
+        assert "script tried accessing undeclared key" in str(e)
+    try:
+        res = await cmd1.eval(r'return redis.call("SET", KEYS[1], ARGV[1])', 1, 'foo2', 'bar2')
+    except Exception as e:
+        print(f"EVAL error: {e}")
+        assert False
+    await asyncio.sleep(0.1)
+    await future
+    status = future.result()
+    assert status
+
+
 '''
 Test the monitor command.
 Open connection which is used for monitoring
 Then send on other connection commands to dragonfly instance
 Make sure that we are getting the commands in the monitor context
 '''
-
-
 @pytest.mark.asyncio
 async def test_monitor_command(async_pool):
     def generate(max):
@@ -39,9 +87,8 @@ async def process_cmd(monitor, key, value):
                 if "select" not in response["command"].lower():
                     success = verify_response(response, key, value)
                     if not success:
-                        print(f"failed to verfiy message {response} for {key}/{value}")
+                        print(f"failed to verify message {response} for {key}/{value}")
                         return False, f"failed on the verification of the message {response} at {key}: {value}"
-                        #await asyncio.sleep(0.01)
                     else:
                         return True, None
         except asyncio.TimeoutError: