diff --git a/lib/cachex/services/courier.ex b/lib/cachex/services/courier.ex index bcad3fa..c4e9af4 100644 --- a/lib/cachex/services/courier.ex +++ b/lib/cachex/services/courier.ex @@ -18,6 +18,7 @@ defmodule Cachex.Services.Courier do # add some aliases alias Cachex.Actions + alias Cachex.Actions.Get alias Cachex.Actions.Put alias Cachex.ExecutionError @@ -65,45 +66,50 @@ 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}, caller, {cache, tasks}) do - references = - case Map.get(tasks, key) do - {pid, listeners} -> - {pid, [caller | listeners]} - - nil -> - parent = self() - - worker = - spawn_link(fn -> - result = - try do - task.() - rescue - e -> - { - :error, - %ExecutionError{ - message: Exception.message(e), - stack: stack_compat() ++ stack + def handle_call({:dispatch, key, task, stack}, caller, {cache, tasks} = state) do + case Map.get(tasks, key) do + {pid, listeners} -> + {:noreply, {cache, Map.put(tasks, key, {pid, [caller | listeners]})}} + + nil -> + case Get.execute(cache, key, []) do + {:ok, nil} -> + parent = self() + + worker = + spawn_link(fn -> + result = + try do + task.() + rescue + e -> + { + :error, + %ExecutionError{ + message: Exception.message(e), + stack: stack_compat() ++ stack + } } - } - end + end - formatted = Actions.format_fetch_value(result) - normalized = Actions.normalize_commit(formatted) + formatted = Actions.format_fetch_value(result) + normalized = Actions.normalize_commit(formatted) - with {:commit, val, options} <- normalized do - Put.execute(cache, key, val, [const(:notify_false) | options]) - end + with {:commit, val, options} <- normalized do + Put.execute(cache, key, val, [ + const(:notify_false) | options + ]) + end - send(parent, {:notify, key, formatted}) - end) + send(parent, {:notify, key, formatted}) + end) - {worker, [caller]} - end + {:noreply, {cache, Map.put(tasks, key, {worker, [caller]})}} - {:noreply, {cache, Map.put(tasks, key, references)}} + {:ok, _value} = res -> + {:reply, res, state} + end + end end @doc false diff --git a/test/cachex/actions/fetch_test.exs b/test/cachex/actions/fetch_test.exs index c655ef9..067fba8 100644 --- a/test/cachex/actions/fetch_test.exs +++ b/test/cachex/actions/fetch_test.exs @@ -180,4 +180,26 @@ defmodule Cachex.Actions.FetchTest do assert(get1 == {:ok, "1"}) assert(get2 == {:ok, "2"}) end + + # This test ensures that the fallback is executed just once per key, per ttl + test "fetching will only call fallback once per key" do + cache = Helper.create_cache() + agent = start_supervised!({Agent, fn -> %{} end}) + + for test_index <- 1..100 do + test_key = "test_key_#{test_index}" + + 1..(System.schedulers_online() * 2) + |> Task.async_stream(fn _ -> + Cachex.fetch(cache, test_key, fn -> + Agent.update(agent, fn state -> + Map.update(state, test_key, 1, &(&1 + 1)) + end) + end) + end) + |> Stream.run() + + assert 1 == Agent.get(agent, &Map.get(&1, test_key)) + end + end end