test: cover blocking_test

test/excache/blocking_test.exs

@@ -152,3 +152,161 @@ defmodule Excache.BlockingTest do
       assert Task.await(second) == ["single", "later"]
     end
 
+    test "a two element push wakes both waiters with one element each", %{
+      conn: conn,
+      port: port
+    } do
+      first = blockingTask(port, ["BLPOP", "pair", "0"])
+      awaitWaiters("pair", 1)
+      second = blockingTask(port, ["BLPOP", "pair", "0"])
+      awaitWaiters("pair", 2)
+
+      assert {2, conn} = Client.command(conn, ["LPUSH", "pair", "a", "b"])
+
+      assert Task.await(first) == ["pair", "b"]
+      assert Task.await(second) == ["pair", "a"]
+      assert {0, _conn} = Client.command(conn, ["EXISTS", "pair"])
+    end
+
+    test "a dropped connection leaves the element for the next client", %{
+      conn: conn,
+      port: port
+    } do
+      victim = Client.connect(port)
+      Client.send_raw(victim, encodeCommand(["BLPOP", "orphan", "0"]))
+      awaitWaiters("orphan", 1)
+
+      Client.close(victim)
+      awaitUntil(fn -> Blocking.waitingCount("orphan") == 0 end)
+
+      assert {1, conn} = Client.command(conn, ["LPUSH", "orphan", "kept"])
+      assert {["kept"], conn} = Client.command(conn, ["LRANGE", "orphan", "0", "-1"])
+      assert {["orphan", "kept"], _conn} = Client.command(conn, ["BLPOP", "orphan", "0"])
+    end
+  end
+
+  describe "isolation" do
+    test "a blocked client stalls neither other clients nor the expiry sweeper", %{
+      conn: conn,
+      port: port
+    } do
+      subscriber = Client.connect(port)
+
+      assert {["subscribe", "__keyevent@0__:expired", 1], subscriber} =
+               Client.command(subscriber, ["SUBSCRIBE", "__keyevent@0__:expired"])
+
+      waiter = blockingTask(port, ["BLPOP", "idle", "1"])
+      awaitWaiters("idle", 1)
+
+      assert {{:simple, "OK"}, conn} = Client.command(conn, ["SET", "other", "v"])
+      assert {"v", conn} = Client.command(conn, ["GET", "other"])
+      assert {3, conn} = Client.command(conn, ["RPUSH", "unrelated", "a", "b", "c"])
+      assert {["a", "b", "c"], conn} = Client.command(conn, ["LRANGE", "unrelated", "0", "-1"])
+
+      assert {{:simple, "OK"}, _conn} =
+               Client.command(conn, ["SET", "ephemeral", "v", "PX", "60"])
+
+      assert {["message", "__keyevent@0__:expired", "ephemeral"], _subscriber} =
+               Client.read(subscriber)
+
+      assert Task.await(waiter) == nil
+
+      Client.close(subscriber)
+    end
+
+    test "blocking commands inside MULTI answer at once instead of waiting", %{port: port} do
+      socket = rawConnect(port)
+
+      :ok =
+        :gen_tcp.send(socket, [
+          encodeCommand(["MULTI"]),
+          encodeCommand(["BLPOP", "txn", "0"]),
+          encodeCommand(["EXEC"])
+        ])
+
+      assert readBytes(socket, byte_size("+OK\r\n+QUEUED\r\n*1\r\n*-1\r\n")) ==
+               "+OK\r\n+QUEUED\r\n*1\r\n*-1\r\n"
+
+      :gen_tcp.close(socket)
+    end
+
+    test "a queued blocking command still pops available data on EXEC", %{conn: conn} do
+      assert {1, conn} = Client.command(conn, ["RPUSH", "txn", "ready"])
+      assert {{:simple, "OK"}, conn} = Client.command(conn, ["MULTI"])
+      assert {{:simple, "QUEUED"}, conn} = Client.command(conn, ["BLPOP", "txn", "0"])
+      assert {[["txn", "ready"]], conn} = Client.command(conn, ["EXEC"])
+      assert {0, _conn} = Client.command(conn, ["EXISTS", "txn"])
+    end
+
+    test "a push made inside a transaction wakes a waiter", %{conn: conn, port: port} do
+      waiter = blockingTask(port, ["BLPOP", "txnpush", "0"])
+      awaitWaiters("txnpush", 1)
+
+      assert {{:simple, "OK"}, conn} = Client.command(conn, ["MULTI"])
+      assert {{:simple, "QUEUED"}, conn} = Client.command(conn, ["LPUSH", "txnpush", "fromtxn"])
+      assert {[1], _conn} = Client.command(conn, ["EXEC"])
+
+      assert Task.await(waiter) == ["txnpush", "fromtxn"]
+    end
+  end
+
+  describe "persistence and notifications" do
+    test "a woken BLPOP is logged as LPOP and replays without blocking", %{
+      conn: conn,
+      port: port
+    } do
+      aofPath = Application.get_env(:excache, :aof_path)
+
+      waiter = blockingTask(port, ["BLPOP", "aofq", "0"])
+      awaitWaiters("aofq", 1)
+
+      assert {1, conn} = Client.command(conn, ["LPUSH", "aofq", "payload"])
+      assert Task.await(waiter) == ["aofq", "payload"]
+
+      Excache.Aof.flush()
+      log = File.read!(aofPath)
+
+      assert log =~ "LPOP"
+      refute log =~ "BLPOP"
+
+      Excache.Store.flushdb()
+      Excache.Aof.replay(aofPath)
+
+      assert {0, _conn} = Client.command(conn, ["EXISTS", "aofq"])
+    end
+
+    test "a woken BLPOP fires the same keyspace event as LPOP", %{conn: conn, port: port} do
+      subscriber = Client.connect(port)
+
+      assert {["subscribe", "__keyevent@0__:lpop", 1], subscriber} =
+               Client.command(subscriber, ["SUBSCRIBE", "__keyevent@0__:lpop"])
+
+      waiter = blockingTask(port, ["BLPOP", "notified", "0"])
+      awaitWaiters("notified", 1)
+
+      assert {1, _conn} = Client.command(conn, ["LPUSH", "notified", "v"])
+      assert Task.await(waiter) == ["notified", "v"]
+
+      assert {["message", "__keyevent@0__:lpop", "notified"], _subscriber} =
+               Client.read(subscriber)
+
+      Client.close(subscriber)
+    end
+  end
+
+  describe "BLMOVE and BRPOPLPUSH" do
+    test "BRPOPLPUSH moves an available element straight away", %{conn: conn} do
+      assert {2, conn} = Client.command(conn, ["RPUSH", "src", "a", "b"])
+      assert {"b", conn} = Client.command(conn, ["BRPOPLPUSH", "src", "dst", "0"])
+      assert {["b"], conn} = Client.command(conn, ["LRANGE", "dst", "0", "-1"])
+      assert {["a"], _conn} = Client.command(conn, ["LRANGE", "src", "0", "-1"])
+    end
+
+    test "BRPOPLPUSH waits for the source and times out with nil", %{conn: conn, port: port} do
+      waiter = blockingTask(port, ["BRPOPLPUSH", "src", "dst", "0"])
+      awaitWaiters("src", 1)
+
+      assert {1, conn} = Client.command(conn, ["RPUSH", "src", "moved"])
+      assert Task.await(waiter) == "moved"
+      assert {["moved"], conn} = Client.command(conn, ["LRANGE", "dst", "0", "-1"])
+