Repository: whitfin/cachex (elixir)

Issue:
Fetch can allow multiple executions of `fallback` function 
First off - thanks, as always, for maintaining this library!

I had a question / bug(?) report regarding `fetch/4`

I rely heavily on `fetch/4` when using cachex to leverage the fact that the `fallback` function will only be called once per key per ttl and recently found out that the reason for a very infrequent test failure was that the `fallback` function was in fact being called multiple times due to what I assume is a race condition

The general flow for `fetch` is as follows

```mermaid
stateDiagram-v2
    state cache_hit_state <<fork>>
    state fetch_state <<fork>>
    state task_notification <<join>>
    state response_state <<join>>

    [*] --> Cachex.fetch/4
    Cachex.fetch/4 --> check_cache
    check_cache --> cache_hit_state
    cache_hit_state --> cache_hit
    cache_hit_state --> cache_miss

    cache_miss --> dispatch_to_currier
    dispatch_to_currier --> fetch_state
    fetch_state --> task_executing
    fetch_state --> no_task_executing

    state Currier {
    task_executing --> queue_caller_await_notification
    no_task_executing --> execute_task
    no_task_executing --> queue_caller_await_notification
    queue_caller_await_notification --> task_notification
    update_cache --> task_notification
    task_notification --> notification
    notification --> reply_to_callers
    }

    state external_task {
        execute_task --> fallback
        fallback --> update_cache
    }


    reply_to_callers --> response_state
    cache_hit --> response_state

    response_state --> return_value
 ```

I believe the issue I am seeing is related to the fact that the initial check cache / queue a message with the currier is not an "atomic" operation and can interleave in a way in which we would execute the `fallback` function a second time.

Imagine a situation where the Currier has dispatched a task to generate the value for `key_1`
A new client (Client_3) now calls `fetch/4` for the value of `key_1`
- The cache value is checked, and is still empty
BEFORE we can dispatch a message to the Currier, the existing task finishes executing the fallback function, updates the cache and sends its notification to the Currier
The Currier sends responses to all currently enqueued clients for `key_1`
The Currier then gets the `dispatch` message from `Client_3` which triggers a new task to be executed

Here is an attempt to diagram that same workflow

```mermaid
sequenceDiagram
    box cache
    participant Cache
    end
    box Clients
    participant Client_1
    participant Client_2
    participant Client_3
    end
    box Currier
    participant Currier
    participant Task
    end

    Client_1->>Cache: check key_1
    Cache->>Client_1: no value
    Client_1->>Currier: fetch key_1
    Currier->>Task: dispatch
    Currier->>Currier: record Client_1->key_1
    Client_2->>Cache: check key_1
    Cache->>Client_2: no value
    Client_2->>Currier: fetch key_1
    Currier->>Currier: record Client_2->key_1
    Task->>Task: execute_fallback
    Client_3->>Cache: check key_1
    Cache->>Client_3: no value
    Task->>Task: update_cache
    Client_3->>Currier: fetch key_1
    Task->>Currier: notification of result
    Currier->>Client_1: result
    Currier->>Client_2: result
    Client_3->>Currier: fetch key_1
    Currier->>Task: dispatch

 ```

* I hope these mermaids make some sense - If you need more context I can try and update them to better match my thinking

I was wondering if you agree that this behavior is undesirable and if you do agree wanted to see what your thoughts were on a potential resolution.  My first thoughts was that we could check the cache again from the Currier before executing the task to generate the value - this would add some additional overhead, but only for the single message that would execute the task


Pull request description:
Ensure fetch fallback is called once per key per ttl
Resolves #347 

This is the same logic as mentioned in the issue, just cleaned up a bit - let me know if you would like to see any changes!

Patch:
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


Test patch:
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