It is not recommended to await a long-running task inside an OTP behaviour such as GenServer. Instead, you should match on the message coming from a task inside your GenServer.handle_info/2callback. For more information on the format of the message, see the documentation for async/1.
In this case, the WS stores to DynamoDB, and publishes to Kafka. Those are fast operations in normal circumstances.
Could you folks help me understand what could happen in the Task.await/1 call in the original code?
The WS is not linked to the tasks because of the async_nolink call, but it still monitors them (the documentation of async_nolink does not say otherwise). Therefore, it receives exit signals from the task as messages. Channels have a handle_info callback, and it is not clear to me if Task.await/1 is correct in a Phoenix Channel at all because it enters a receive block to consume messages directly from the process mailbox. What do you think?
On the other hand, Task.await/1 emits exit signals from within that receive block, whose origin is the WS process itself. How is the supervisor of the WS going to react to those signals? Is this why it has been said in the thread that Task.await/1 would still need a rescue if I want the WS to not exit at all due to failures in those operations?
Kernel.raise/1 is an Elixir style exception - exit/1 is the original Erlang style process exception which can be caught within the process but will terminate that process if it is not caught.
Therefore, it receives exit signals from the task as messages.
Signals relate to process links (and convert to :EXIT messages when trapped) - :DOWN messages relate to monitors - two entirely separate mechanisms.
it is not clear to me if Task.await/1 is correct in a Phoenix Channel at all because it enters a receive block to consume messages directly from the process mailbox. What do you think?
That is the fundamental issue with await - it blocks the process outside of the OTP code. But keep in mind that even calls with GenServer.call/3 to another process are blocking. So there are situations where it can be OK - it’s more of an issue of awareness of blocking behaviour.
On the other hand, Task.await/1 emits exit signals from within that receive block
Your channel/topic process is executing await/2 and therefore exit/1 - so it’s not a signal yet - it can still be stopped with a try..catch.
How is the supervisor of the WS going to react to those signals?
If uncaught the channel/topic process will be terminated. I’m not sure how the supervisor is configured, so I can’t predict its behaviour.
And if I’m reading this right the supervisor is running with defaultmax_restarts: (3) and :max_seconds (5) i.e. if there are more than 3 restarts within 5 seconds the supervisor will terminate.
So it looks like intensely crashing channel/topic processes will in fact terminate the pool supervisor easily.
So I would go with the current approach, where the channel is blocked. This means that actions on the client won’t receive a reply until the publishing is done - but other than that it is generally fine. I assume the publishing to the message bus is super fast, so that’s how I would roll honestly.
Or I would create a GenServer on every new channel with the purpose of keeping ordering. Every time you have to publish, you just cast the GenServer to do it for you. The GenServer would be started in its own supervisor (a DynamicSupervisor) and it would monitor the channel. Once the channel is DOWN, it exits.
To compare, I started to write an alternative with GenServer as you described. I had also a registry to be able to cast by name during the lifetime of the channel, and the solution based on Task.Supervisor.async_stream_nolink/6 wins in simplicity.
The cost is less throghtput, but as a first implementation of this service I prefer to keep it simple first, and introduce the GenServer if usage shows it is justified. On the other side, the synchronous approach would allow us to send an ACK status back to the client (not a requirement right now, though).
