I’d consider using Task.async_stream/2 or Flow for this first part. I believe it will do a better job on managing throughput.

I assume this isn’t your real code because here you are using the map returned from handle_item/1 as a map key.

You could consider using an :ets table for the accumulator since that allows concurrent write access. However concurrency here isn’t a magic bullet since you are already running this code pithing a concurrent environment (the original 10_000 items).

This looks like you are exploding an rrule into individual event times (is that what CockTail.occurrences/1 does?). I think the key question here is whether you need to materialise all the times up front or whether there is an opportunity to lazily evaluate the individual times when you need them. The calendar_recurrence might help with this.

I think switching the Flow would be your best bet since it is designed for exactly this kind of use case. Something like (not tested):

items()
|> Flow.from_enumerable(max_demand: :erlang.system_info(:schedulers_online))
|> Flow.map(&handle_item/1))
|> Flow.partition()
|> Flow.reduce(fn -> %{} end, fn time, acc -> Map.put(acc, time, element.value) end)
|> Enum.to_list()

10k items? This should barely blip for 0.5 secs IMO. If you can put some example GitHub project that can produce the items so we can check a few algorithms, we should be able to help further.

I suspect the time is taken materialising the time events (the code says about 8,000 of them for each of the 10,000 items). Avoiding materialising these time events would, I think, be the better path.

+1 for leveraging Flow. It’s literally designed to solve tasks like that one.