From a8d8bc6caa39e7e18cf04fceac7dab57c0bfdc9a Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 9 Sep 2026 17:24:00 +0200 Subject: [PATCH] test: cover blocking_test --- test/excache/blocking_test.exs | 158 +++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/test/excache/blocking_test.exs b/test/excache/blocking_test.exs index a0ca761..4971614 100644 --- a/test/excache/blocking_test.exs +++ b/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"]) +