From 583957091964ad3204b5cecd547d7ad5121c7b13 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 13 Sep 2026 19:15:21 +0200 Subject: [PATCH] test: cover duringSuspendedShard and awaitQueuedCall --- test/excache/store_consistency_test.exs | 114 ++++++++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/test/excache/store_consistency_test.exs b/test/excache/store_consistency_test.exs index e9cbfa4..6431a39 100644 --- a/test/excache/store_consistency_test.exs +++ b/test/excache/store_consistency_test.exs @@ -157,3 +157,117 @@ defmodule Excache.StoreConsistencyTest do assert {3, conn} = Client.command(conn, ["SADD", key, "a", "b", "c"]) + {taken, conn} = Client.command(conn, ["SPOP", key, "3"]) + assert Enum.sort(taken) == ["a", "b", "c"] + assert {0, conn} = Client.command(conn, ["EXISTS", key]) + + log = flushedLog() + refute log =~ "SPOP" + + conn = replayFromScratch(conn) + + assert {0, _conn} = Client.command(conn, ["EXISTS", key]) + end + + test "popping from a missing set replays as a no-op", %{conn: conn} do + assert {nil, conn} = Client.command(conn, ["SPOP", "cons:pop:ghost"]) + assert {{:simple, "OK"}, conn} = Client.command(conn, ["SET", "cons:pop:marker", "v"]) + + flushedLog() + conn = replayFromScratch(conn) + + assert {0, conn} = Client.command(conn, ["EXISTS", "cons:pop:ghost"]) + assert {"v", _conn} = Client.command(conn, ["GET", "cons:pop:marker"]) + end + end + + describe "OBJECT IDLETIME must report the tracked idle time" do + test "a key left alone for five seconds reports five", %{conn: conn} do + assert :ok = Config.set("maxmemory-policy", "allkeys-lru") + + key = "cons:idle:tracked" + assert :ok = Store.set(key, "v", []) + Tracker.touch(Tracker, key, System.system_time(:millisecond) - 5_000) + + assert {5, _conn} = Client.command(conn, ["OBJECT", "IDLETIME", key]) + end + + test "a key without access metadata reports zero rather than an error", %{conn: conn} do + key = "cons:idle:untracked" + assert :ok = Store.set(key, "v", []) + + assert {0, _conn} = Client.command(conn, ["OBJECT", "IDLETIME", key]) + end + + test "a missing key is still refused", %{conn: conn} do + assert {{:error, "ERR no such key"}, _conn} = + Client.command(conn, ["OBJECT", "IDLETIME", "cons:idle:ghost"]) + end + end + + defp duringSuspendedShard(shardKey, work) do + shard = shardPid(shardKey) + parent = self() + + :sys.suspend(shard) + + worker = + Task.async(fn -> + send(parent, :entered) + work.() + end) + + queued = + try do + assert_receive :entered, 100 + waiting = awaitQueuedCall(shard, worker.pid) + Process.sleep(@holdMs) + waiting + after + :sys.resume(shard) + end + + assert queued, "the worker never reached the suspended shard, the expiry window was not open" + + Task.await(worker) + end + + defp awaitQueuedCall(shard, workerPid) do + Enum.any?(1..500, fn _attempt -> + if callPending?(shard, workerPid) do + true + else + Process.sleep(1) + false + end + end) + end + + defp callPending?(shard, workerPid) do + {:messages, pending} = Process.info(shard, :messages) + Enum.any?(pending, &match?({:"$gen_call", {^workerPid, _ref}, _request}, &1)) + end + + defp flushedLog do + Aof.flush() + {:ok, log} = File.read(Application.get_env(:excache, :aof_path)) + log + end + + defp replayFromScratch(conn) do + Store.flushdb() + Aof.replay(Application.get_env(:excache, :aof_path)) + conn + end + + defp shardIndex(key), do: rem(:erlang.phash2(key), Store.shard_count()) + + defp shardPid(key), do: Process.whereis(Store.shard_name(shardIndex(key))) + + defp keyInOtherShard(prefix, reference) do + Enum.find_value(1..1024, fn suffix -> + candidate = prefix <> Integer.to_string(suffix) + if shardIndex(candidate) != shardIndex(reference), do: candidate + end) + end +end