Background jobs with Oban, part 2: live updates and smarter retries

This post picks up where part one left off. In it we'll push job results to a LiveView with Phoenix PubSub, mount the Oban Web dashboard to watch our queues, and teach our worker which failures are worth retrying.

You can watch the screencast above, or read the full walkthrough below. Unfurl's source is available on GitHub if you'd like to follow along.

In part one we took an app called Unfurl. You paste in a URL, and it fetches the web page, pulls out the title, description, and preview image, and displays them.

That fetch used to happen right inside the LiveView's save event handler, which meant one slow website froze the page for our user.

So we moved it into an Oban job. Now when a link is saved, the LiveView just writes the row and queues up the work to happen in the background, and the page comes right back.

But we left one thing unfinished. When that job finishes, it updates the row in the database, and the page has no idea. We still have to refresh to see our preview.

In this post we'll finish the job: live updates with no refresh, a dashboard for watching our queues, and a worker that knows which failures are worth retrying.

Live updates with PubSub

Let's start with the page. Since we're using Phoenix, we can broadcast events with Phoenix PubSub and update our links in real time. No browser refresh required.

We'll go back to our Links context module and add a new function, subscribe, that our LiveView can use to subscribe to link events.

Inside it we'll call Phoenix.PubSub.subscribe to subscribe to a @topic, which we'll set above in a module attribute. Let's call our topic "links".

@topic "links"

def subscribe do
  Phoenix.PubSub.subscribe(Unfurl.PubSub, @topic)
end

Then we'll need another helper function to announce changes.

For that we'll create a new private function called broadcast that takes two arguments. The first is the {:ok, link} tuple returned by the database call, which we bind to result. The second is the event.

We pattern match on :ok here because a successful database call means a change actually happened, and that's something we need to broadcast.

Inside it we'll call Phoenix.PubSub.broadcast to send the event and the link to our topic, and then we'll return the result so callers get back exactly what they got before.

Let's also add a second broadcast clause that matches on :error. If the database change didn't happen, there's no event to broadcast, so we just return the result.

defp broadcast({:ok, %Link{} = link} = result, event) do
  Phoenix.PubSub.broadcast(Unfurl.PubSub, @topic, {event, link})
  result
end

defp broadcast({:error, _changeset} = result, _event), do: result

We want to broadcast whenever a link changes, so let's update each function that modifies a link to pipe into broadcast, along with an event describing how the link changed.

• In create_link, we'll pipe the result of Repo.insert into broadcast with an event we'll call :link_created
• In apply_preview, we'll pipe the update into broadcast with :link_updated
• In mark_failed, we'll do the same, also with :link_updated
• And in delete_link, we'll pipe the result of Repo.delete into broadcast with :link_deleted

def create_link(attrs) do
  %Link{}
  |> Link.changeset(attrs)
  |> Repo.insert()
  |> broadcast(:link_created)
end

def apply_preview(%Link{} = link, attrs) do
  fetched_at = DateTime.utc_now(:second)
  link
  |> update_link(Map.merge(attrs, %{status: "ready", fetched_at: fetched_at}))
  |> broadcast(:link_updated)
end

def mark_failed(%Link{} = link) do
  link
  |> update_link(%{status: "failed"})
  |> broadcast(:link_updated)
end

def delete_link(%Link{} = link) do
  link
  |> Repo.delete()
  |> broadcast(:link_deleted)
end

Because these broadcasts live in the context, it doesn't matter who makes the change. The LiveView creates and deletes links, and the Oban job applies previews, but every change goes out on the same topic.

Subscribing from the LiveView

Now we need to update our LinkLive LiveView. In mount, we'll call Links.subscribe, but only if connected? returns true.

We do this because mount runs twice. The first run happens during the initial HTTP request, which renders the page once and exits. That process never handles messages, so our guard skips stateful work like subscribing during that throwaway render. The second run happens once the browser connects over the WebSocket, and that's the process that sticks around.

def mount(_params, _session, socket) do
  if connected?(socket), do: Links.subscribe()

  {:ok,
   socket
   |> assign(:form, to_form(Links.change_link(%Link{})))
   |> stream(:links, Links.list_links())}
end

Handling the events

Now we need to handle our messages. Let's add three handle_info callbacks, one to pattern match on each of our events: :link_created, :link_updated, and :link_deleted.

The links on the page are rendered from a LiveView stream, so we can use the stream functions to update them.

When a link is created, we'll call stream_insert, passing in the socket, the name of our stream, :links, and the link. We want new previews to appear at the top of the list, so we'll set at: 0.

When a link is updated, we'll call stream_insert again. If the link's preview is already on the page, LiveView updates it in place instead of moving it, which is exactly what we want. We'll also pass update_only: true, so if the link isn't in the stream, it won't be inserted.

And when a link is deleted, we'll call stream_delete to remove its preview from the stream.

@impl true
def handle_info({:link_created, link}, socket) do
  {:noreply, stream_insert(socket, :links, link, at: 0)}
end
def handle_info({:link_updated, link}, socket) do
  {:noreply, stream_insert(socket, :links, link, update_only: true)}
end
def handle_info({:link_deleted, link}, socket) do
  {:noreply, stream_delete(socket, :links, link)}
end

Now that every change to a link flows through the same broadcast, the LiveView doesn't need to touch its stream when it handles a user event. So let's go up to our delete handler and remove the stream_delete call. The :link_deleted broadcast takes care of it.

def handle_event("delete", %{"id" => id}, socket) do
  link = Links.get_link!(id)
  {:ok, _link} = Links.delete_link(link)
  {:noreply, put_flash(socket, :info, "Removed #{link.url}")}
end

A nice side effect: because every connected LiveView is subscribed to the same topic, a link saved in one browser tab shows up in every other open tab too.

With that, let's go back to the browser. Now if we save a new link, its card appears as pending, and a moment later, with no refresh, the title and description fill in and the badge flips to ready.

The Oban Web dashboard

In part one we added oban_web as a dependency, but we haven't used it yet.

Oban Web is a LiveView dashboard for watching and managing your queues and jobs. Let's configure our app to use it.

We'll open our router.ex, and down in the dev_routes section, we'll import Oban.Web.Router. Then inside the "/dev" scope we'll add oban_dashboard with the path we want to reach it at. We'll use "/oban".

One note: if you use the dashboard in production, be sure to put it behind authentication. We won't worry about that here, since this route only exists in development.

if Application.compile_env(:unfurl, :dev_routes) do
  import Phoenix.LiveDashboard.Router
  import Oban.Web.Router

  scope "/dev" do
    pipe_through :browser

    live_dashboard "/dashboard", metrics: UnfurlWeb.Telemetry
    oban_dashboard "/oban"
    forward "/mailbox", Plug.Swoosh.MailboxPreview
  end
end

Then we just need to open /dev/oban in our browser.

And there's our Oban dashboard. In the sidebar we can filter jobs by state, and below that we can see the previews queue we configured in part one.

From here on we'll use this to see what our jobs are doing.

When a fetch fails

Right now, every failed fetch is handled the same way. Whether it's a 404, a timeout, or a 500, we mark the link as failed and never try again.

But what if a site is briefly down, or slow, or rate limiting us? Those might work if we just tried again later.

Remember that Preview.fetch holds on to the status code when a request fails, returning {:error, {:http_status, status}}. It also passes retry: false to Req, turning off Req's own retries, so Oban gets to decide what happens. With a little pattern matching, we can stop swallowing every error and let Oban handle each kind of failure appropriately.

Our perform function can return any of the values in Oban's result/0 type. So far we've used :ok and {:cancel, reason}. Let's add {:snooze, period} and {:error, reason} to the mix.

A 404 won't fix itself

First we'll pattern match on the error reason {:http_status, 404}. A missing page will still be missing on the next attempt, so we'll mark the link failed with our existing Links.mark_failed call, and then return {:cancel, :not_found}. That stops the job immediately and marks it as cancelled.

A 429 means slow down

Next we'll match on {:http_status, 429}, which means we're being rate limited. Here we'll return {:snooze, {1, :minute}}. Oban reschedules the job a minute out, and the snooze doesn't count against the max_attempts we set for this worker.

Everything else gets retried

For anything else, like a 500 or a timeout, we'll match on {:error, reason} and return it as is. Because we set max_attempts: 5, Oban will retry up to four more times after the first failure, backing off a little longer each time, and then discard the job.

We still want to mark the link as failed, but only on the last attempt. So let's add a private last_attempt? function that takes the Oban.Job struct, and mark the link as failed only when it returns true.

In last_attempt? we'll pattern match on the Oban.Job struct to get its attempt and max_attempts values, and return true when attempt is greater than or equal to max_attempts.

Since we need the job now, we'll update unfurl to accept it as a second parameter. Then in perform, we'll bind the job and pass it into unfurl.

@impl Oban.Worker
def perform(%Oban.Job{args: %{"id" => id}} = job) do
  case Links.fetch_link(id) do
    {:ok, link} -> unfurl(link, job)
    :error -> {:cancel, :link_deleted}
  end
end

defp unfurl(%Link{} = link, %Oban.Job{} = job) do
  case Preview.fetch(link.url) do
    {:ok, attrs} ->
      {:ok, _link} = Links.apply_preview(link, attrs)
      :ok

    {:error, {:http_status, 404}} ->
      {:ok, _link} = Links.mark_failed(link)
      {:cancel, :not_found}

    {:error, {:http_status, 429}} ->
      {:snooze, {1, :minute}}

    {:error, reason} ->
      if last_attempt?(job), do: Links.mark_failed(link)

      {:error, reason}
  end
end

defp last_attempt?(%Oban.Job{attempt: attempt, max_attempts: max_attempts}) do
  attempt >= max_attempts
end

Now a link stays pending while Oban retries it, and it's only marked as failed if the final attempt fails.

Trying it out

Let's go back to Unfurl and add a few links. To make this easy to see, I have a second app running on port 4010 with routes that return different status codes.

First, http://localhost:4010/gone, which returns a 404. It fails right away, and if we open the Oban dashboard, we can see the job went straight to cancelled.

Next, http://localhost:4010/busy, which always returns a 429. On the page it shows as pending. In the dashboard, the job is scheduled, and it hasn't burned through any attempts, because snoozing doesn't spend one.

Then http://localhost:4010/broken, which returns a plain 500. In the dashboard the job is retryable, and Oban has scheduled another attempt. Meanwhile the card on the page stays pending. It won't flip until the job is out of attempts.

Finally, http://localhost:4010/flaky, which fails twice and then succeeds. The card is pending, and in the dashboard it's retryable, right alongside our /broken job.

A few seconds later, Oban retries it and it succeeds. Back on the page, the preview fills in with no refresh. In the dashboard it's no longer retryable, and the completed count went up by one.

And once our /broken link runs out of attempts, its card flips to failed, and in the dashboard its job moves to discarded after five of five attempts.

Wrapping up

We started with a LiveView that fetched each URL inside its event handler, so one slow site froze the page.

Then we moved that fetch into an Oban job, used Phoenix PubSub to update the page with no refresh, and added a dashboard to watch it all happen. Now failures that might recover get retried automatically, failures that won't get cancelled right away, and our user never waits on any of it.

Unfurl's source is available on GitHub.

That's all for now. Thanks for watching, and happy deploying.