I am trying to grow my skills a bit and exploring the Registry module. I have a custom registry set up as follows:

defmodule MyReg do 
  def start_link() do 
    Registry.start_link(keys: :unique, name: MyReg)
    Registry.register(MyReg, :default_key, MapSet.new())
  end

I want to iterate over a collection and store each item in the MapSet accessed by the :default_key but also use each item as another key with it’s own value.

# in the same module
def register(item, value) do 
  case Registry.register(MyReg, item, [value]) do 
    {:ok, _} ->
       # the item was not previously registered so add it to the collection of keys
       Registry.update_value(MyReg, :default_key, fn keys -> MapSet.put(keys, item) end)
    _ ->
      # if the item has already been registered, update it's values
       Registry.update_value(MyReg, item, fn vals -> [value | vals] end)
  end
end

When I run this manually from within iex -S mix it seems to work great. But when I try to run it as follows it fails at the update_value step.

MyReg.start_link()

collection
|> Task.async_stream(fn {item, value} -> 
        MyReg.register(item, value)
end)
|> Enum.to_list()
# [ok: :error, ok: :error, ...]
> Registry.lookup(MyReg, :default_key)
# MapSet.new()

I believe that the problem is that somehow the processes spawned by Task.async_stream do not have access to the registry created with MyReg.start_link, but I don’t know why.

After enumerating over the collection I can look at the contents of the registry and see that the initial start_link call was successful but that it was never modified after that.