excache/test/excache/write_path_test.exs

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"]