From 2e3c8efe2e9982d84ea2280c3387be05fc03cc22 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 13 Sep 2026 23:43:04 +0200 Subject: [PATCH] test: cover Excache.WritePathTest --- test/excache/write_path_test.exs | 159 +++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 test/excache/write_path_test.exs diff --git a/test/excache/write_path_test.exs b/test/excache/write_path_test.exs new file mode 100644 index 0000000..f55eab2 --- /dev/null +++ b/test/excache/write_path_test.exs @@ -0,0 +1,159 @@ +defmodule Excache.WritePathTest do + use Excache.Test.ServerCase, async: false + + alias Excache.Aof + alias Excache.Blocking + alias Excache.Command + alias Excache.Config + alias Excache.Eviction.Runner + alias Excache.Eviction.Tracker + alias Excache.Store + + @payload String.duplicate("x", 1000) + @oom {:error, "OOM command not allowed when used memory > 'maxmemory'"} + + describe "renaming a key onto itself keeps it" do + test "the store answers OK and the value survives" do + assert :ok = Store.set("self:string", "payload", []) + assert {:ok, "OK"} = Store.rename("self:string", "self:string") + assert {:ok, "payload"} = Store.get("self:string") + end + + test "a missing key renamed onto itself is still refused" do + assert {:error, :nokey} = Store.rename("self:ghost", "self:ghost") + end + + test "the remaining ttl is untouched" do + assert :ok = Store.set("self:ttl", "payload", px: 60_000) + assert {:ok, "OK"} = Store.rename("self:ttl", "self:ttl") + + remaining = Store.ttl("self:ttl", :seconds) + assert remaining > 0 and remaining <= 60 + end + + test "a list survives the round trip over the wire", %{conn: conn} do + assert {2, conn} = Client.command(conn, ["RPUSH", "self:list", "a", "b"]) + assert {{:simple, "OK"}, conn} = Client.command(conn, ["RENAME", "self:list", "self:list"]) + assert {["a", "b"], conn} = Client.command(conn, ["LRANGE", "self:list", "0", "-1"]) + + conn = replayFromScratch(conn) + + assert {["a", "b"], _conn} = Client.command(conn, ["LRANGE", "self:list", "0", "-1"]) + end + + test "RENAMENX onto itself answers zero and keeps the key", %{conn: conn} do + assert {{:simple, "OK"}, conn} = Client.command(conn, ["SET", "self:nx", "v"]) + assert {0, conn} = Client.command(conn, ["RENAMENX", "self:nx", "self:nx"]) + assert {"v", _conn} = Client.command(conn, ["GET", "self:nx"]) + end + end + + describe "blocking moves obey the out of memory gate" do + setup do + Application.put_env(:excache, :eviction_recheck, 0) + Config.reset() + Tracker.clear(Tracker) + Runner.resetStats() + + on_exit(fn -> + Config.reset() + Application.delete_env(:excache, :eviction_recheck) + end) + + :ok + end + + test "BRPOPLPUSH is refused once maxmemory is crossed", %{conn: conn} do + conn = fillOverLimit(conn, "oom:brpoplpush") + + assert {@oom, conn} = + Client.command(conn, ["BRPOPLPUSH", "oom:brpoplpush", "oom:moved", "0"]) + + assert {1, conn} = Client.command(conn, ["LLEN", "oom:brpoplpush"]) + assert {0, _conn} = Client.command(conn, ["EXISTS", "oom:moved"]) + end + + test "BLMOVE is refused once maxmemory is crossed", %{conn: conn} do + conn = fillOverLimit(conn, "oom:blmove") + + assert {@oom, conn} = + Client.command(conn, ["BLMOVE", "oom:blmove", "oom:moved", "LEFT", "RIGHT", "0"]) + + assert {1, conn} = Client.command(conn, ["LLEN", "oom:blmove"]) + assert {0, _conn} = Client.command(conn, ["EXISTS", "oom:moved"]) + end + + test "lifting the limit lets the same move through", %{conn: conn} do + conn = fillOverLimit(conn, "oom:lifted") + + assert {@oom, conn} = + Client.command(conn, ["BLMOVE", "oom:lifted", "oom:dst", "LEFT", "RIGHT", "0"]) + + {{:simple, "OK"}, conn} = Client.command(conn, ["CONFIG", "SET", "maxmemory", "0"]) + + assert {"value", conn} = + Client.command(conn, ["BLMOVE", "oom:lifted", "oom:dst", "LEFT", "RIGHT", "0"]) + + assert {["value"], _conn} = Client.command(conn, ["LRANGE", "oom:dst", "0", "-1"]) + end + end + + describe "blocking writes reach the shared log funnel" do + test "a blocked BLMOVE announces the push it performed", %{conn: conn, port: port} do + subscriber = subscribeTo(port, "__keyevent@0__:rpush") + + waiter = blockingTask(port, ["BLMOVE", "funnel:src", "funnel:dst", "LEFT", "RIGHT", "0"]) + awaitWaiters("funnel:src", 1) + + assert {1, _conn} = Client.command(conn, ["LPUSH", "funnel:src", "item"]) + assert Task.await(waiter) == "item" + + assert {["message", "__keyevent@0__:rpush", "funnel:dst"], _subscriber} = + Client.read(subscriber) + + Client.close(subscriber) + end + + test "an immediately served BLMOVE announces the push as well", %{conn: conn, port: port} do + assert {1, conn} = Client.command(conn, ["RPUSH", "inline:src", "item"]) + + subscriber = subscribeTo(port, "__keyevent@0__:rpush") + + assert {"item", _conn} = + Client.command(conn, ["BLMOVE", "inline:src", "inline:dst", "LEFT", "RIGHT", "0"]) + + assert {["message", "__keyevent@0__:rpush", "inline:dst"], _subscriber} = + Client.read(subscriber) + + Client.close(subscriber) + end + + test "a blocked BLMOVE replays into the very same lists", %{conn: conn, port: port} do + waiter = blockingTask(port, ["BLMOVE", "replay:src", "replay:dst", "LEFT", "RIGHT", "0"]) + awaitWaiters("replay:src", 1) + + assert {2, conn} = Client.command(conn, ["RPUSH", "replay:src", "moved", "stays"]) + assert Task.await(waiter) == "moved" + + {source, conn} = Client.command(conn, ["LRANGE", "replay:src", "0", "-1"]) + {destination, conn} = Client.command(conn, ["LRANGE", "replay:dst", "0", "-1"]) + + log = flushedLog() + refute log =~ "BLMOVE" + + conn = replayFromScratch(conn) + + assert {^source, conn} = Client.command(conn, ["LRANGE", "replay:src", "0", "-1"]) + assert {^destination, _conn} = Client.command(conn, ["LRANGE", "replay:dst", "0", "-1"]) + end + + test "the pop of a woken BLPOP is logged after the push that woke it", %{ + conn: conn, + port: port + } do + waiter = blockingTask(port, ["BLPOP", "order:queue", "0"]) + awaitWaiters("order:queue", 1) + + assert {1, conn} = Client.command(conn, ["LPUSH", "order:queue", "payload"]) + assert Task.await(waiter) == ["order:queue", "payload"] +