Repository: whitfin/cachex
Pull request #411 (2025-04-20 21:38:54)

Update tests and utilities around $callers usage
This PR will just normalize some tests around $callers to use the common utilities.

- Refactored calls to avoid duplicated passing (caller exists in `handle_call/3`)
- Added `$callers` support to locksmith execution
- Fixed re-schedule in warming due to missing `$callers`
- Fixed the issue with Elixir 1.7 (where `$callers` was not automated in `Task`)




Patch:
diff --git a/lib/cachex/actions/warm.ex b/lib/cachex/actions/warm.ex
index 9e500c9..5833129 100644
--- a/lib/cachex/actions/warm.ex
+++ b/lib/cachex/actions/warm.ex
@@ -46,22 +46,26 @@ defmodule Cachex.Actions.Warm do
     do: only == nil or mod in only or name in only
 
   # Spawns a task to invoke the call to the remote warmer.
+  #
+  # We have to manually set $callers because we support Elixir v1.7 and this
+  # wasn't automated via the Task module at that point in time.
   defp spawn_call(warmer(name: name) = warmer, wait) do
+    callers = [self() | callers()]
+
     Task.async(fn ->
+      Process.put(:"$callers", callers)
       call_warmer(warmer, wait)
       name
     end)
   end
 
   # Invokes a warmer with blocking enabled.
-  defp call_warmer(warmer(name: name), true) do
-    GenServer.call(name, {:cachex_warmer, get_callers()}, :infinity)
-  end
+  defp call_warmer(warmer(name: name), true),
+    do: GenServer.call(name, {:cachex_warmer, callers()}, :infinity)
 
   # Invokes a warmer with blocking disabled.
-  defp call_warmer(warmer(name: name), _) do
-    send(name, {:cachex_warmer, get_callers()})
-  end
+  defp call_warmer(warmer(name: name), _),
+    do: send(name, {:cachex_warmer, callers()})
 
   # Converts a task result to a name reference.
   defp extract_name({_, {:ok, name}}),
diff --git a/lib/cachex/hook.ex b/lib/cachex/hook.ex
index 0a3963f..37df4c1 100644
--- a/lib/cachex/hook.ex
+++ b/lib/cachex/hook.ex
@@ -140,7 +140,7 @@ defmodule Cachex.Hook do
 
       @doc false
       def handle_info({:cachex_notify, {event, result, callers}}, state) do
-        put_callers(callers)
+        Process.put(:"$callers", callers)
 
         case timeout() do
           nil ->
diff --git a/lib/cachex/services/courier.ex b/lib/cachex/services/courier.ex
index 215020b..55be053 100644
--- a/lib/cachex/services/courier.ex
+++ b/lib/cachex/services/courier.ex
@@ -39,13 +39,8 @@ defmodule Cachex.Services.Courier do
   call which will wait until a result can be loaded.
   """
   @spec dispatch(Cachex.t(), any, (-> any)) :: any
-  def dispatch(cache() = cache, key, task) when is_function(task, 0) do
-    service_call(
-      cache,
-      :courier,
-      {:dispatch, key, task, local_stack(), get_callers()}
-    )
-  end
+  def dispatch(cache() = cache, key, task) when is_function(task, 0),
+    do: service_call(cache, :courier, {:dispatch, key, task, local_stack(), callers()})
 
   ####################
   # Server Callbacks #
@@ -69,14 +64,7 @@ defmodule Cachex.Services.Courier do
   # Due to the nature of the async behaviour, this call will return before
   # the task has been completed, and the :notify callback will receive the
   # results from the task after completion (regardless of outcome).
-  def handle_call(
-        {:dispatch, key, task, stack, callers},
-        caller,
-        {cache, tasks} = state
-      ) do
-    put_callers(callers)
-    callers = get_callers()
-
+  def handle_call({:dispatch, key, task, stack, callers}, caller, {cache, tasks} = state) do
     case Map.get(tasks, key) do
       {pid, listeners} ->
         {:noreply, {cache, Map.put(tasks, key, {pid, [caller | listeners]})}}
@@ -88,9 +76,10 @@ defmodule Cachex.Services.Courier do
 
             worker =
               spawn_link(fn ->
+                Process.put(:"$callers", [parent, elem(caller, 0) | callers])
+
                 result =
                   try do
-                    put_callers(callers)
                     task.()
                   rescue
                     e ->
diff --git a/lib/cachex/services/informant.ex b/lib/cachex/services/informant.ex
index 96ec241..fab2da5 100644
--- a/lib/cachex/services/informant.ex
+++ b/lib/cachex/services/informant.ex
@@ -67,7 +67,7 @@ defmodule Cachex.Services.Informant do
         # skip notifying service hooks
         if module.type() != :service do
           # define the base payload, regardless of type
-          payload = {:cachex_notify, {action, result, get_callers()}}
+          payload = {:cachex_notify, {action, result, [self() | callers()]}}
 
           # handle async vs. sync
           case module.async?() do
diff --git a/lib/cachex/services/locksmith/queue.ex b/lib/cachex/services/locksmith/queue.ex
index 375fc69..c1be3ea 100644
--- a/lib/cachex/services/locksmith/queue.ex
+++ b/lib/cachex/services/locksmith/queue.ex
@@ -31,16 +31,14 @@ defmodule Cachex.Services.Locksmith.Queue do
   """
   @spec execute(Cachex.t(), (-> any)) :: any
   def execute(cache() = cache, func) when is_function(func, 0),
-    do: service_call(cache, :locksmith, {:exec, func})
+    do: service_call(cache, :locksmith, {:exec, func, callers()})
 
   @doc """
   Executes a function in a transactional context.
   """
   @spec transaction(Cachex.t(), [any], (-> any)) :: any
-  def transaction(cache() = cache, keys, func)
-      when is_list(keys) and is_function(func, 0) do
-    service_call(cache, :locksmith, {:transaction, keys, func, get_callers()})
-  end
+  def transaction(cache() = cache, keys, func) when is_list(keys) and is_function(func, 0),
+    do: service_call(cache, :locksmith, {:transaction, keys, func, callers()})
 
   ####################
   # Server Callbacks #
@@ -63,8 +61,8 @@ defmodule Cachex.Services.Locksmith.Queue do
   #
   # Because locks are handled sequentially inside this process, this execution
   # can guarantee that there are no locks set on the table when it fires.
-  def handle_call({:exec, func}, _ctx, cache),
-    do: {:reply, safe_exec(func), cache}
+  def handle_call({:exec, func, callers}, {caller, _tag}, cache),
+    do: {:reply, safe_exec(func, [caller | callers]), cache}
 
   @doc false
   # Executes a function in a transactional context.
@@ -73,10 +71,9 @@ defmodule Cachex.Services.Locksmith.Queue do
   # locks after execution. The key here is that the locks set on a key will stop
   # other processes from writing them, and force them to queue their writes
   # inside this queue process instead.
-  def handle_call({:transaction, keys, func, callers}, _ctx, cache) do
-    put_callers(callers)
+  def handle_call({:transaction, keys, func, callers}, {caller, _tag}, cache) do
     true = lock(cache, keys)
-    val = safe_exec(func)
+    val = safe_exec(func, [caller | callers])
     true = unlock(cache, keys)
 
     {:reply, val, cache}
@@ -90,9 +87,13 @@ defmodule Cachex.Services.Locksmith.Queue do
   #
   # Any errors which occur are rescued and returned in an
   # `:error` tagged Tuple to avoid crashing the process.
-  defp safe_exec(fun) do
-    fun.()
-  rescue
-    e -> {:error, Exception.message(e)}
+  defp safe_exec(fun, chain) do
+    Process.put(:"$callers", chain)
+
+    try do
+      fun.()
+    rescue
+      e -> {:error, Exception.message(e)}
+    end
   end
 end
diff --git a/lib/cachex/spec.ex b/lib/cachex/spec.ex
index 4b3d3ae..45c608f 100644
--- a/lib/cachex/spec.ex
+++ b/lib/cachex/spec.ex
@@ -421,6 +421,12 @@ defmodule Cachex.Spec do
   # Utilities #
   #############
 
+  @doc false
+  # Retrieves the $callers for the current process.
+  @spec callers() :: [pid()]
+  defmacro callers(),
+    do: quote(do: Process.get(:"$callers", []))
+
   @doc false
   # Determines if a value is a negative integer.
   @spec is_negative_integer(integer) :: boolean
@@ -470,22 +476,4 @@ defmodule Cachex.Spec do
   @spec wrap(any, atom) :: {atom, any}
   defmacro wrap(value, tag) when is_atom(tag),
     do: quote(do: {unquote(tag), unquote(value)})
-
-  # extracts list of `$callers` from the process dicionary
-  # and adds the current process to the list
-  # the intended use is to save it in a variable
-  # and use it inside spawned code
-  # e.g.
-  #
-  #     callers = get_callers()
-  #     spawn(fn -> put_callers() ... end)
-  #
-  # `$callers` allows mocking libraries like `Mox`
-  # to set mocks in the test process and use them
-  # in other processes that have the test process in `$callers`
-  @doc false
-  def get_callers(), do: [self() | Process.get(:"$callers") || []]
-
-  @doc false
-  def put_callers(callers), do: Process.put(:"$callers", callers)
 end
diff --git a/lib/cachex/warmer.ex b/lib/cachex/warmer.ex
index 11fd4b4..395485a 100644
--- a/lib/cachex/warmer.ex
+++ b/lib/cachex/warmer.ex
@@ -90,11 +90,10 @@ defmodule Cachex.Warmer do
       # cache via `Cachex.put_many/3` if returns in a Tuple tagged with the
       # `:ok` atom. If `:ignore` is returned, nothing happens aside from
       # scheduling the next execution of the warming to occur on interval.
-      def handle_info(
-            {:cachex_warmer, callers},
-            {cache, state, interval, timer}
-          ) do
-        put_callers(callers)
+      def handle_info({:cachex_warmer, callers}, {cache, state, interval, timer}) do
+        # store the calling processes
+        Process.put(:"$callers", callers)
+
         # clean our any existing timers
         if timer, do: Process.cancel_timer(timer)
 
@@ -117,7 +116,7 @@ defmodule Cachex.Warmer do
         timer =
           case interval do
             nil -> nil
-            val -> :erlang.send_after(val, self(), :cachex_warmer)
+            val -> :erlang.send_after(val, self(), {:cachex_warmer, [self()]})
           end
 
         # pass the new state


Test patch:
diff --git a/test/cachex/actions/fetch_test.exs b/test/cachex/actions/fetch_test.exs
index 59ce00f..f75a390 100644
--- a/test/cachex/actions/fetch_test.exs
+++ b/test/cachex/actions/fetch_test.exs
@@ -203,20 +203,24 @@ defmodule Cachex.Actions.FetchTest do
     assert %{1 => 1000} == calls
   end
 
-  test "fallback function has test process in $callers" do
-    test_process = self()
-    callers_reference = make_ref()
-
+  # This test covers whether $callers is correctly propagated through to the
+  # fallback function to allow things like mocking, etc.
+  test "fetching functions have access to $callers" do
+    # create a test cache
     cache = TestUtils.create_cache()
+    cache = Services.Overseer.get(cache)
 
-    fallback_fun = fn ->
-      send(test_process, {callers_reference, Process.get(:"$callers")})
-      "value"
-    end
+    # process chain
+    parent = self()
+    courier = Services.locate(cache, Services.Courier)
 
-    {:commit, "value"} = Cachex.fetch(cache, "key", fallback_fun)
+    # trigger a fetch in another process
+    Cachex.fetch(cache, "key", fn ->
+      send(parent, Process.get(:"$callers"))
+      {:ignore, nil}
+    end)
 
-    assert_receive({^callers_reference, callers})
-    assert test_process in callers
+    # check callers are the Courier and us
+    assert_receive([^courier, ^parent])
   end
 end
diff --git a/test/cachex/actions/get_and_update_test.exs b/test/cachex/actions/get_and_update_test.exs
index 51f62b8..f8d76b0 100644
--- a/test/cachex/actions/get_and_update_test.exs
+++ b/test/cachex/actions/get_and_update_test.exs
@@ -138,4 +138,24 @@ defmodule Cachex.Actions.GetAndUpdateTest do
     assert_receive({^callers_reference, callers})
     assert test_process in callers
   end
+
+  # This test covers whether $callers is correctly propagated through to the
+  # update function to allow things like mocking, etc.
+  test "update functions have access to $callers" do
+    # create a test cache
+    cache = TestUtils.create_cache()
+    cache = Services.Overseer.get(cache)
+
+    # process chain
+    parent = self()
+
+    # trigger a fetch in another process
+    Cachex.get_and_update(cache, "key", fn _ ->
+      send(parent, Process.get(:"$callers"))
+      {:ignore, nil}
+    end)
+
+    # check callers are just the base process
+    assert_receive([^parent])
+  end
 end
diff --git a/test/cachex/hook_test.exs b/test/cachex/hook_test.exs
index 87df62e..fca2b5f 100644
--- a/test/cachex/hook_test.exs
+++ b/test/cachex/hook_test.exs
@@ -1,26 +1,6 @@
 defmodule Cachex.HookTest do
   use Cachex.Test.Case
 
-  defmodule Hook.Callers do
-    use Cachex.Hook
-    import Cachex.Spec
-
-    def async?, do: true
-    def actions, do: :all
-    def type, do: :pre
-
-    @doc """
-    Returns a hook definition for a custom execute hook.
-    """
-    def create(name \\ nil),
-      do: hook(module: __MODULE__, args: self(), name: name)
-
-    def handle_notify(_, _, proc) do
-      send(proc, Process.get(:"$callers"))
-      {:noreply, proc}
-    end
-  end
-
   setup_all do
     ForwardHook.bind(
       concat_hook_1: [type: :pre],
@@ -31,15 +11,6 @@ defmodule Cachex.HookTest do
     :ok
   end
 
-  test "$callers in hooks" do
-    test_process = self()
-    callers_hook = Hook.Callers.create()
-    cache = TestUtils.create_cache(hooks: [callers_hook])
-    Cachex.fetch(cache, "key1", fn _ -> "val1" end)
-    assert_receive hook_callers
-    assert test_process in hook_callers
-  end
-
   test "concatenating hooks in a cache" do
     # create a set of 3 hooks to test with
     hook1 = ForwardHook.create(:concat_hook_1)
@@ -101,4 +72,24 @@ defmodule Cachex.HookTest do
     # check that locating with the wrong type finds nothing
     assert Cachex.Hook.locate(cache1, :concat_hook_1, :post) == nil
   end
+
+  # This test covers whether $callers is correctly propagated through to hooks
+  # when triggered by a parent process. We validate this by sending the value
+  # back through to the test process to validate it contains the process identifier.
+  test "accessing $callers in hooks" do
+    # create a test cache and execution hook
+    cache = TestUtils.create_cache(hooks: [ExecuteHook.create()])
+
+    # find the hook (with the populated runtime process identifier)
+    cache(hooks: hooks(post: [hook])) = Services.Overseer.get(cache)
+
+    # notify and fetch callers in order to send them back to this parent process
+    Services.Informant.notify([hook], {:exec, fn -> Process.get(:"$callers") end}, nil)
+
+    # process chain
+    parent = self()
+
+    # check callers are just us
+    assert_receive([^parent])
+  end
 end
diff --git a/test/cachex/test/hook/execute.ex b/test/cachex/test/hook/execute.ex
index 4f084de..52a88df 100644
--- a/test/cachex/test/hook/execute.ex
+++ b/test/cachex/test/hook/execute.ex
@@ -58,18 +58,14 @@ defmodule Cachex.Test.Hook.Execute do
           @doc """
           Executes a received function and forwards to the state process.
           """
-          def handle_notify({_tag, fun}, _results, proc) do
-            handle_info(fun.(), proc)
-            {:ok, proc}
-          end
+          def handle_notify({_tag, fun}, _results, proc) when is_function(fun, 0),
+            do: handle_info(fun.(), proc) && {:ok, proc}
 
           @doc """
           Forwards received messages to the state process.
           """
-          def handle_info(msg, proc) do
-            send(proc, msg)
-            {:noreply, proc}
-          end
+          def handle_info(msg, proc),
+            do: send(proc, msg) && {:noreply, proc}
         end
       end
     end
diff --git a/test/cachex/warmer_test.exs b/test/cachex/warmer_test.exs
index 7516320..9e1dc8e 100644
--- a/test/cachex/warmer_test.exs
+++ b/test/cachex/warmer_test.exs
@@ -1,15 +1,6 @@
 defmodule Cachex.WarmerTest do
   use Cachex.Test.Case
 
-  defmodule Warmer.Callers do
-    use Cachex.Warmer
-
-    def execute(proc) do
-      send(proc, Process.get(:"$callers"))
-      {:ok, []}
-    end
-  end
-
   test "warmers which set basic values" do
     # create a test warmer to pass to the cache
     TestUtils.create_warmer(:basic_warmer, fn _ ->
@@ -152,21 +143,28 @@ defmodule Cachex.WarmerTest do
     assert_receive({{:put_many, [[{2, 2}], []]}, {:ok, true}})
   end
 
-  test "warmers populate $callers" do
-    test_process = self()
+  test "accessing $callers in warmers" do
+    # current process
+    parent = self()
+
+    # create a warmer to pass $callers
+    TestUtils.create_warmer(:callers_warmer, fn parent ->
+      send(parent, Process.get(:"$callers")) && {:ok, []}
+    end)
 
+    # create a test cache
     TestUtils.create_cache(
       warmers: [
         warmer(
-          module: Warmer.Callers,
-          interval: 15000,
+          module: :callers_warmer,
+          interval: :timer.seconds(60),
           required: true,
-          state: test_process
+          state: parent
         )
       ]
     )
 
-    assert_receive callers
-    assert test_process in callers
+    # check callers are just us
+    assert_receive([^parent])
   end
 end