Assuming that you want several clients each making the same request to get the same response, it should be very simple to roll it yourself. The general idea would be along these lines:

%% "Cache process"
receive_loop(State) ->
    receive
         {request, Pid, Ref, Request} -> 
              receive_loop(make_request(Pid, Ref, Request, State));
         {database_response, DbRef, Result} ->
              receive_loop(handle_response(DbRef, Result, State))
    end.

make_request(Pid, Ref, Request, {Cache, In, Out}) ->
    case {Cache, In} of
        {#{ Request := Cached }, #{}} ->
            %% Cache hit, reply with what we have.
            Pid ! {reply, Ref, Cached},
            {Cache, In, Out};
        {_, #{ Request := Others }} ->
            %% This request has already been made but we haven't received a
            %% reply yet. Tell the previously made request that we want the reply,
            %% too.
            {Cache, In#{ Request => [{Pid, Ref} | Others] }, Out};
        {#{}, #{}} ->
            %% Not in cache, no outstanding request: make a new request.
            %%
            %% It is very important that this request is made asynchronously,
            %% otherwise everyone will have to wait for each request to finish
            %% which makes the whole thing pointless.
            DbRef = call_your_database_asynchronously(self(), Request),
            {Cache,
             In#{ Request => [{Pid, Ref}] },
             Out#{ DbRef => Request } }
    end.

handle_response(DbRef, Result, {Cache, In, Out}) ->
    #{ DbRef := Request } = Out,
    #{ Request := Clients } = In,
    %% Respond to all clients that made the same request, then
    %% cache the result and remove this outstanding request
    %% from the state.
    %%
    %% If you only want the "single flight" functionality, just skip
    %% the cache parts in this example.
    [Pid ! {reply, ClientRef, Result} || {Pid, ClientRef} <- Clients],
    {Cache#{ Request => Result },
     maps:remove(Request, In),
     maps:remove(DbRef, Out)}.

(Apologies for putting it in Erlang, hopefully you’ll get the gist of it though .)

Cachex has got you covered as @kokolegorille mentioned.

Here’s a demo that shows you that the expensive work is done only once. You can paste this in e.g. dont_repeat_work.exs and just do elixir dont_repeat_work.exs after.

There:

Mix.install([{:cachex, "~> 3.6"}])

Cachex.start_link(name: :our_cache)

defmodule Worker do
  def expensive_work(id) do
    Cachex.fetch(:our_cache, "expensive_data", fn _key ->
      IO.puts(
        "#{inspect(NaiveDateTime.utc_now())}: doing expensive work on behalf of worker #{inspect(id)}"
      )

      :timer.sleep(1000)
      {:commit, "EXPENSIVE_VALUE"}
    end)
  end

  def use_expensive_work(id) do
    IO.puts("#{inspect(NaiveDateTime.utc_now())}: requesting access from worker #{inspect(id)}")
    value = expensive_work(id)
    IO.puts("#{inspect(NaiveDateTime.utc_now())}: worker #{id} received value: #{inspect(value)}")
  end
end

1..3
|> Task.async_stream(fn id ->
  Worker.use_expensive_work(id)
end)
|> Stream.run()

On my machine this returned:

~N[2024-04-09 11:31:13.743650]: requesting access from worker 1
~N[2024-04-09 11:31:13.743645]: requesting access from worker 2
~N[2024-04-09 11:31:13.743642]: requesting access from worker 3
~N[2024-04-09 11:31:13.746957]: doing expensive work on behalf of worker 3
~N[2024-04-09 11:31:14.748989]: worker 2 received value: {:ok, "EXPENSIVE_VALUE"}
~N[2024-04-09 11:31:14.748980]: worker 3 received value: {:commit, "EXPENSIVE_VALUE"}
~N[2024-04-09 11:31:14.749017]: worker 1 received value: {:ok, "EXPENSIVE_VALUE"}

Notice the {:commit, ...} tuple. That means the work has been done only once and the other 2 {:ok, ...} tuples mean a cache hit. The important part however is that this does NOT lead to all 3 workers concurrently doing the expensive work and all 3 writing to the cache at (nearly) the same time. Cachex took special care to prevent this.

There is the Cachex library…

As mentionned by the previous answers, it is simple to implement

in cache? → cache
not in cache? → query → cache the result

Maybe this diagram will make it clearer:

image556×320 4.05 KB

You cannot have a race condition in this situation because cache process will always process messages sequentially, this is how processes work in elixir.

Keep in mind that this is a simplistic approach that has a inherent problem that the mailbox of cache process can be overflowed. The approach you will see in real-world will be by having cache process replaced with a pool of processes that would share the same ETS table (for example, or other medium you want to cache in).

omg, TIL about Cachex.fetch. I don’t know how many versions of this I handrolled over the years.

https://hexdocs.pm/cachex/Cachex.html#fetch/4