Repository: whitfin/cachex (elixir)

Issue:
Cache warmer throws error for async long running task
## Issue

Whenever there is a long running task invoked inside of the cachex warmer behaviour, the application fails to boot with the following error:

```
** (Mix) Could not start application cachex_bug: CachexBug.Application.start(:normal, []) returned an error: shutdown: failed to start child: Cachex
    ** (EXIT) an exception was raised:
        ** (FunctionClauseError) no function clause matching in Cachex.Actions.Warm.extract_name/1
            (cachex 4.0.3) lib/cachex/actions/warm.ex:65: Cachex.Actions.Warm.extract_name({%Task{mfa: {:erlang, :apply, 2}, owner: #PID<0.214.0>, pid: #PID<0.224.0>, ref: #Reference<0.0.27395.97194186.2661613573.38107>}, nil})
            (elixir 1.18.1) lib/enum.ex:1714: Enum."-map/2-lists^map/1-1-"/2
            (cachex 4.0.3) lib/cachex/actions/warm.ex:35: Cachex.Actions.Warm.execute/2
            (cachex 4.0.3) lib/cachex/router.ex:182: Cachex.Router.route_local/3
            (cachex 4.0.3) lib/cachex.ex:1441: Cachex.setup_warmers/1
            (cachex 4.0.3) lib/cachex.ex:279: Cachex.start_link/2
            (stdlib 6.2) supervisor.erl:959: :supervisor.do_start_child_i/3
            (stdlib 6.2) supervisor.erl:945: :supervisor.do_start_child/3
```

## Error reproduction

```elixir
defmodule CachexBug.AsyncTasks do
  require Logger

  def long_running_task() do
    [1, 2, 3, 4]
    |> Task.async_stream(
      fn num ->
        cache_key = "key_#{num}"
        Logger.info("generating cache for #{cache_key}")

        Process.sleep(6000)

        {cache_key, num}
      end,
      timeout: :infinity
    )
    |> Stream.map(fn {:ok, result} -> result end)
    |> Enum.to_list()
  end
end

defmodule CachexBug.CacheWarmer do
  use Cachex.Warmer

  def execute(_) do
    {:ok, CachexBug.AsyncTasks.long_running_task()}
  end
end

defmodule CachexBug.Application do
  @moduledoc false

  use Application

  import Cachex.Spec

  @cache_name :cachex_bug
  @cache_default_expiration_milliseconds :timer.hours(1)
  @cache_options [
    @cache_name,
    [
      expiration: expiration(default: @cache_default_expiration_milliseconds),
      warmers: [
        warmer(
          module: CachexBug.CacheWarmer,
          interval: @cache_default_expiration_milliseconds,
          required: true
        )
      ]
    ]
  ]

  @impl true
  def start(_type, _args) do
    children = [
      {Cachex, @cache_options}
    ]
    opts = [strategy: :one_for_one, name: CachexBug.Supervisor]
    Supervisor.start_link(children, opts)
  end
end
```

## Investigation

The problem seems to be coming from the `Cachex.Actions.Warm` module while invoking [`|> Task.yield_many()`](https://github.com/whitfin/cachex/blob/439169897924ea1820308ed0b8786b2a327fbfab/lib/cachex/actions/warm.ex#L34). It's called with the default timeout value, which returns `nil` for some cases if the invoked warmer is also calling a task but with a custom timeout value greater than the default value of `5000`. Maybe there is a configuration option to change it somewhere, I couldn't find it anywhere in the cachex docs or code or specs.

## Hacky fix
Since there is no override or configuration option, I'm using the following hack to override the entire `Cachex.Actions.Warm` module before application boot.

```elixir
defmodule MyApp.CachexActionsWarmPatch do
  use Supervisor

  def start_link(arg) do
    Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
  end

  def init(_args) do
    Code.eval_string("""
    defmodule Cachex.Actions.Warm do
      import Cachex.Spec

      def execute(cache(warmers: warmers), options) do
        only = Keyword.get(options, :only, nil)
        wait = Keyword.get(options, :wait, false)

        warmed =
          warmers
          |> Enum.filter(&filter_mod(&1, only))
          |> Enum.map(&spawn_call(&1, wait))
          |> Task.yield_many(:infinity) # added `:infinity` option to support nested long running tasks
          |> Enum.map(&extract_name/1)

        {:ok, warmed}
      end

      defp filter_mod(warmer(module: mod, name: name), only),
        do: only == nil or mod in only or name in only

      defp spawn_call(warmer(name: name) = warmer, wait) do
        Task.async(fn ->
          call_warmer(warmer, wait)
          name
        end)
      end

      defp call_warmer(warmer(name: name), true),
        do: GenServer.call(name, :cachex_warmer, :infinity)

      defp call_warmer(warmer(name: name), _),
        do: send(name, :cachex_warmer)

      defp extract_name({_, {:ok, name}}),
        do: name
    end
    """)

    Supervisor.init([], strategy: :one_for_one)
  end
end
```

```elixir
# Then add the following in application supervision tree before initializing cachex.
children = [
  ...
  MyApp.CachexActionsWarmPatch,
  {Cachex,  ...},
  ...
]
```

## Fix

I'm not the library author, but as a consumer I can propose either allowing the `timeout` value to be passed on to the `warmer` spec, or invoking the `Task.yield_many` with `:infinity` timeout value in all cases.

I can open a PR if there is a decision on which approach works best for all consumers of this library ✌ 

Pull request description:
Fix cache warmer for long running tasks
As discussed in [the issue](#400), this PR fixes #400 for cases where the cache warmer fails for long running tasks.


Patch:
diff --git a/lib/cachex/actions/warm.ex b/lib/cachex/actions/warm.ex
index 8651538..e798898 100644
--- a/lib/cachex/actions/warm.ex
+++ b/lib/cachex/actions/warm.ex
@@ -31,7 +31,7 @@ defmodule Cachex.Actions.Warm do
       warmers
       |> Enum.filter(&filter_mod(&1, only))
       |> Enum.map(&spawn_call(&1, wait))
-      |> Task.yield_many()
+      |> Task.yield_many(:infinity)
       |> Enum.map(&extract_name/1)
 
     {:ok, warmed}


Test patch:
diff --git a/test/cachex/warmer_test.exs b/test/cachex/warmer_test.exs
index 63bdd58..697a4cd 100644
--- a/test/cachex/warmer_test.exs
+++ b/test/cachex/warmer_test.exs
@@ -14,6 +14,50 @@ defmodule Cachex.WarmerTest do
     assert Cachex.get!(cache, 1) == 1
   end
 
+  test "warmers with long running tasks" do
+    # create a test warmer to pass to the cache
+    TestUtils.create_warmer(:long_running_warmer, fn _ ->
+      # exceed default timeout of `5000`
+      Process.sleep(5001)
+      {:ok, [{1, 1}]}
+    end)
+
+    # create a cache instance with a warmer
+    cache =
+      TestUtils.create_cache(warmers: [warmer(module: :long_running_warmer)])
+
+    # check that the key was warmed
+    assert Cachex.get!(cache, 1) == 1
+  end
+
+  test "warmers with long running async tasks" do
+    # create a test warmer to pass to the cache
+    TestUtils.create_warmer(:long_running_async_warmer, fn _ ->
+      {:ok,
+       [1, 2]
+       |> Task.async_stream(
+         fn num ->
+           # exceed default timeout of `5000`
+           Process.sleep(5001)
+           {num, num}
+         end,
+         timeout: :infinity
+       )
+       |> Stream.map(fn {:ok, result} -> result end)
+       |> Enum.to_list()}
+    end)
+
+    # create a cache instance with a warmer
+    cache =
+      TestUtils.create_cache(
+        warmers: [warmer(module: :long_running_async_warmer)]
+      )
+
+    # check that the keys were warmed
+    assert Cachex.get!(cache, 1) == 1
+    assert Cachex.get!(cache, 2) == 2
+  end
+
   test "warmers which set values with options" do
     # create a test warmer to pass to the cache
     TestUtils.create_warmer(:options_warmer, fn _ ->