chore: connection pipeline cache grows without shrinking

Signed-off-by: kostas <[email protected]>

Author: kostasrim

tests/dragonfly/connection_test.py

@@ -1051,3 +1051,60 @@ async def test_hiredis(df_factory):
     server.start()
     client = base_redis.Redis(port=server.port, protocol=3, cache_config=CacheConfig())
     client.ping()
+
+
+@dfly_args({"proactor_threads": 1})
+async def test_pipeline_cache_size(df_factory):
+    server = df_factory.create(proactor_threads=1)
+    server.start()
+
+    # Start 1 client.
+    good_client = server.client()
+
+    await good_client.execute_command("set foo bar")
+
+    bad_actor_client = server.client()
+
+    info = await bad_actor_client.info()
+
+    # Cache is empty.
+    assert info["pipeline_cache_bytes"] == 0
+    assert info["dispatch_queue_bytes"] == 0
+
+    async def push_pipeline(bad_actor_client):
+        # Fill cache.
+        p = bad_actor_client.pipeline(transaction=True)
+        for i in range(1):
+            p.lpush(str(i), "V")
+        await p.execute()
+
+    await push_pipeline(bad_actor_client)
+    info = await good_client.info()
+
+    old_pipeline_cache_bytes = info["pipeline_cache_bytes"]
+    assert old_pipeline_cache_bytes > 0
+    assert info["dispatch_queue_bytes"] == 0
+
+    # Whoops, total pipeline_cache_bytes haven't changed. If a workload aggregates a bunch
+    # pipeline_cache_bytes because it recycled too many messages, they won't gradually be released
+    # if one command (one connection out of `n` connections) dispatches async. Only 1 command out of
+    # n connections must be dispatched async and the pipeline won't gradually be relesed.
+    for i in range(30):
+        await push_pipeline(bad_actor_client)
+        await good_client.execute_command(f"set foo{i} bar")
+
+    info = await good_client.info()
+
+    # Pipeline cache bytes remained constant :(
+    assert old_pipeline_cache_bytes == info["pipeline_cache_bytes"]
+    assert info["dispatch_queue_bytes"] == 0
+
+    # Now drain it
+    for i in range(30):
+        await good_client.execute_command(f"set foo{i} bar")
+
+    info = await good_client.info()
+
+    # Drained
+    assert info["pipeline_cache_bytes"] == 0
+    assert info["dispatch_queue_bytes"] == 0