On our server, we have some remote resources that we want to fetch and periodically update in the background. To update these resources we need to make http requests to third-party services that have rate limits. This means we want to lock down the number of concurrent requests and also pause requests when the third party sends a 429 response. We’ve had a few thoughts about how to do this, but so far each has its pitfalls.

On a high level, the layout of our system looks like this…

• TheThing - stores info in the state about the resource that we can save to the database etc
• NetworkResource - a single-threaded http request queue that when presented with a 429 waits and retries the request

When the TheThing decides to update, it needs to make multiple calls to NetworkResource to re-fetch and save the result back into the database. Depending on the local state and remote state, not all network calls may need to be fired, so for example if the timestamp of the first remote resource is the same as the local one, subsequent calls may not need to be made.

We’ve played around with a couple of different methods, but none feel ideal and it feels like we’re swimming against the flow a little bit. Some of it might be down to our C++ & JavaScript background where “async good, sync bad”. I’ve tried to cut the code examples down to the bare bones just so you get the gist…

Method 1: Use callbacks

In the first method, we’ve been using multiple GenServers and writing queues to manage the amount of requests. When we make a request to the NetworkResource we pass a function definition to callback on completion. This has the advantage that none of the processes become locked, but makes the code quite complex and spaghetti-ish, especially as it grows and you have multiple resource calls and callbacks.

defmodule Test.TheThing do
  use GenServer

  def update(id) do
    GenServer.call(__MODULE__, {:update})
  end

  def callback_fetch_resource(response) do
    GenServer.call(__MODULE__, {:callback_fetch_resource, response})
  end

  def handle_call({:update}, from, state) do
    next_state = %{ action: "wait_fetch_resource" }
    Test.NetworkResource.fetch(&callback_fetch_resource/1)
    {:reply, :ok, next_state}
  end

  def handle_call({:callback_fetch_resource, response}, from, state) do
    # Do some stuff, fire off the next request using the same callback pattern
    # ...
    next_state = %{ action: "wait_fetch_other_resource" }
    {:reply, :ok, next_state}
  end
end

defmodule Test.NetworkResource do
  def init(_) do
    Process.send_after(self(), :do_next, 1000)
    {:ok, %{ requests: [] }}
  end

  def fetch(callback) do
    GenServer.call(__MODULE__, {:fetch, callback})
  end

  def handle_call({:fetch, callback}, from, state) do
    next_state = %{ requests: state.requests ++ [:fetch, callback]}
    {:reply, :ok, next_state}
  end

  def handle_info({:do_next}, state) do
    # Check if anything is running, if not create a new `Task` to run the actual HTTP request
    # and on completion run callback.(response) to reply to the calling GenServer
    Process.send_after(self(), :do_next, 1000)
    {:noreply, state}
  end
end

Method 2: Have GenServers block

In the second method, we’ve tried to make more use of the Task module and have the NetworkResource GenServer block when it’s busy. When we have multiple requests that we want to make all at once, we can also use Task.async_stream to send off a batch and wait for the reply. This gives the advantage of being able to write more linear-looking code, but then the NetworkResource GenServer becomes unresponsive while it’s waiting for network stuff to happen. Depending on how busy the network resource is, we may need a large timeout when making the call.

TheThing GenServer now just manages the local state and database storage and we’ve introduced a TheThingUpdater which manages fetching the resources and providing updates back to multiple instances of TheThing

defmodule Test.TheThing do
  use GenServer
end

defmodule Test.TheThingUpdater do
  use GenServer

  def update(id) do
    GenServer.call(__MODULE__, {:update})
  end

  def handle_call({:update}, from, state) do
    task = Task.async(fn ->
      response = Test.NetworkResource.fetch()

      # Do some stuff, fire off the next request
      # ... response2 = Test.NetworkResource.fetch2()
      # ... response3 = Test.NetworkResource.fetch3()

      {:ok, :ok}
    end)

    next_state = %{ running: state.running ++ [task.ref] }
    {:reply, :ok, next_state}
  end

  def handle_info(msg, state) do
    next_state = case msg do
      {sender, {:ok, value}} ->
        next_running = Enum.filter(state.running, fn task -> task != sender end)
        %{state | running: next_running}
      _ -> state
    end

    {:noreply, next_state}
  end
end

defmodule Test.NetworkResource do
  def fetch(callback) do
    GenServer.call(__MODULE__, {:fetch}, :infinity) # Replace infinity with something more reasonable like an hour
  end

  def handle_call({:fetch, callback}, from, state) do
    # Make the http request using HTTPoison etc and handle 429 errors
    # ...

    {:reply, http_response_body, state}
  end
end

Obviously, our actual use case is more complex than this example, it has multiple requests and a couple of different rate-limited resources. We’re just after some guidance on some of the recommended patterns for doing stuff like this and some pitfalls that we might fall into before ploughing more effort into what we’re doing. Thanks!