From a08beff540eacb29e511b85b9caa1ed19578d2e9 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 9 Sep 2026 16:40:13 +0200 Subject: [PATCH] test: cover Excache.BlockingTest --- test/excache/blocking_test.exs | 154 +++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 test/excache/blocking_test.exs diff --git a/test/excache/blocking_test.exs b/test/excache/blocking_test.exs new file mode 100644 index 0000000..a0ca761 --- /dev/null +++ b/test/excache/blocking_test.exs @@ -0,0 +1,154 @@ +defmodule Excache.BlockingTest do + use Excache.Test.ServerCase, async: false + + alias Excache.Blocking + + describe "immediate path" do + test "BLPOP takes the head when the list already holds data", %{conn: conn} do + assert {2, conn} = Client.command(conn, ["RPUSH", "jobs", "first", "second"]) + assert {["jobs", "first"], conn} = Client.command(conn, ["BLPOP", "jobs", "0"]) + assert {["second"], conn} = Client.command(conn, ["LRANGE", "jobs", "0", "-1"]) + assert {["jobs", "second"], conn} = Client.command(conn, ["BLPOP", "jobs", "0"]) + assert {0, _conn} = Client.command(conn, ["EXISTS", "jobs"]) + end + + test "BRPOP takes the tail", %{conn: conn} do + assert {3, conn} = Client.command(conn, ["RPUSH", "tail", "a", "b", "c"]) + assert {["tail", "c"], _conn} = Client.command(conn, ["BRPOP", "tail", "0"]) + end + + test "BLPOP serves the first key given that has data", %{conn: conn} do + assert {1, conn} = Client.command(conn, ["RPUSH", "second", "later"]) + assert {1, conn} = Client.command(conn, ["RPUSH", "first", "sooner"]) + + assert {["first", "sooner"], conn} = + Client.command(conn, ["BLPOP", "first", "second", "0"]) + + assert {["second", "later"], _conn} = + Client.command(conn, ["BLPOP", "first", "second", "0"]) + end + + test "BLPOP on a key holding another type answers WRONGTYPE without blocking", %{conn: conn} do + assert {{:simple, "OK"}, conn} = Client.command(conn, ["SET", "notalist", "v"]) + + assert {{:error, "WRONGTYPE" <> _}, _conn} = + Client.command(conn, ["BLPOP", "notalist", "0"]) + end + + test "argument errors", %{conn: conn} do + assert {{:error, "ERR timeout is negative"}, conn} = + Client.command(conn, ["BLPOP", "k", "-1"]) + + assert {{:error, "ERR timeout is not a float or out of range"}, conn} = + Client.command(conn, ["BLPOP", "k", "later"]) + + assert {{:error, "ERR wrong number of arguments for 'blpop' command"}, conn} = + Client.command(conn, ["BLPOP", "k"]) + + assert {{:error, "ERR syntax error"}, _conn} = + Client.command(conn, ["BLMOVE", "a", "b", "LEFT", "SIDEWAYS", "0"]) + end + end + + describe "blocking path" do + test "BLPOP waits until another connection pushes", %{conn: conn, port: port} do + waiter = blockingTask(port, ["BLPOP", "queue", "0"]) + awaitWaiters("queue", 1) + + assert {1, conn} = Client.command(conn, ["LPUSH", "queue", "payload"]) + assert Task.await(waiter) == ["queue", "payload"] + assert {0, _conn} = Client.command(conn, ["EXISTS", "queue"]) + end + + test "BRPOP waits on the tail side", %{conn: conn, port: port} do + waiter = blockingTask(port, ["BRPOP", "queue", "0"]) + awaitWaiters("queue", 1) + + assert {2, _conn} = Client.command(conn, ["RPUSH", "queue", "old", "new"]) + assert Task.await(waiter) == ["queue", "new"] + end + + test "a waiter registered on several keys wakes on whichever key receives data", %{ + conn: conn, + port: port + } do + waiter = blockingTask(port, ["BLPOP", "left", "right", "0"]) + awaitWaiters("right", 1) + + assert {1, _conn} = Client.command(conn, ["LPUSH", "right", "hello"]) + assert Task.await(waiter) == ["right", "hello"] + end + + test "a timed out BLPOP answers with a nil array on the wire", %{port: port} do + socket = rawConnect(port) + :ok = :gen_tcp.send(socket, encodeCommand(["BLPOP", "never", "0.15"])) + + {elapsed, reply} = :timer.tc(fn -> :gen_tcp.recv(socket, 0, 2000) end) + + assert reply == {:ok, "*-1\r\n"} + assert elapsed > 100_000 + assert elapsed < 1_500_000 + + :gen_tcp.close(socket) + end + + test "a blocked connection still answers commands sent after it is unblocked", %{ + conn: conn, + port: port + } do + pipelined = Client.connect(port) + + pipelined = + Client.send_raw(pipelined, [ + encodeCommand(["BLPOP", "pipeq", "0"]), + encodeCommand(["PING"]) + ]) + + awaitWaiters("pipeq", 1) + assert :gen_tcp.recv(pipelined.socket, 0, 100) == {:error, :timeout} + + assert {1, _conn} = Client.command(conn, ["LPUSH", "pipeq", "v"]) + + assert {["pipeq", "v"], pipelined} = Client.read(pipelined) + assert {{:simple, "PONG"}, pipelined} = Client.read(pipelined) + + Client.close(pipelined) + end + end + + describe "delivery accounting" do + test "three waiters are served in the order they blocked", %{conn: conn, port: port} do + first = blockingTask(port, ["BLPOP", "fifo", "0"]) + awaitWaiters("fifo", 1) + second = blockingTask(port, ["BLPOP", "fifo", "0"]) + awaitWaiters("fifo", 2) + third = blockingTask(port, ["BLPOP", "fifo", "0"]) + awaitWaiters("fifo", 3) + + assert {1, conn} = Client.command(conn, ["LPUSH", "fifo", "a"]) + assert Task.await(first) == ["fifo", "a"] + + assert {1, conn} = Client.command(conn, ["LPUSH", "fifo", "b"]) + assert Task.await(second) == ["fifo", "b"] + + assert {1, _conn} = Client.command(conn, ["LPUSH", "fifo", "c"]) + assert Task.await(third) == ["fifo", "c"] + end + + test "one pushed element wakes exactly one of two waiters", %{conn: conn, port: port} do + first = blockingTask(port, ["BLPOP", "single", "0"]) + awaitWaiters("single", 1) + second = blockingTask(port, ["BLPOP", "single", "0"]) + awaitWaiters("single", 2) + + assert {1, conn} = Client.command(conn, ["LPUSH", "single", "only"]) + assert Task.await(first) == ["single", "only"] + + assert Task.yield(second, 150) == nil + assert Blocking.waitingCount("single") == 1 + assert {0, conn} = Client.command(conn, ["EXISTS", "single"]) + + assert {1, _conn} = Client.command(conn, ["LPUSH", "single", "later"]) + assert Task.await(second) == ["single", "later"] + end +