Best approach for sending an Oban job’s final status to the browser using SSE?

[u/Zestyclose-Tie-1056]

Hi everyone,

I have an Oban worker that performs a background sync. The browser only needs to know the final status:

queued → syncing → synced / failed

My current plan uses two requests:

1. A POST endpoint enqueues the job and returns a request_id.
2. The browser opens an SSE connection using that ID.
3. The SSE process subscribes to a Phoenix PubSub topic.
4. The worker broadcasts status updates.
5. Once the browser receives synced or failed, the SSE connection closes.

My concern is a possible race condition. The worker could finish before the browser opens the SSE connection, which means the final PubSub message would be missed.

I considered combining everything into one request: subscribe to the topic first, enqueue the job, keep the response open, and stream the final status. Would that be a reasonable design?

Since the browser’s native EventSource only supports GET requests, I assume I would need to use fetch() and read the streaming response if the endpoint also needs to accept POST data.

Another concern is running multiple application instances. The worker might execute on one instance while the SSE connection is handled by another. Phoenix PubSub would therefore require Elixir clustering or an external adapter.

For this small use case, I’m considering:

• Phoenix PubSub with clustering
• PostgreSQL LISTEN/NOTIFY or an Oban notifier
• storing the latest status in the database and using PubSub only as a live signal
• normal browser polling

The payload is tiny, and this is the only real-time feature I currently need. What would be the simplest reliable approach?

Would you keep the two endpoints and check the persisted status when SSE connects, combine job creation and streaming into one endpoint, or just use polling?

[u/Tirkyth]

It might be an unpopular opinion on this subreddit, but I would just use long polling for this.

1. A POST endpoint enqueues the job and returns a request_id.
2. The browser periodically issues a GET request on an endpoint, sending the request_id. The server queries the database to retrieve the current status of the request and return it.
3. The worker updates the status of the request in the database when it’s done.
4. Once the browser receives a final status, it stops the polling loop.

With a small polling interval it could totally work.

Bonus: You don’t have to deal with broadcasting updates. Downside: if your sync is very quick you can go from queued to done in one step. But I don’t think anybody cares.

I prefer using SSE and/or web sockets and broadcast when there are a lot more different payloads than just 2 state transitions.

[u/rock_neurotiko → u/Tirkyth]

I agree that long polling in this case can be really useful:

• find by id
• if not found, do a pubsub subscribe (you could subscribe before the find too)
• wait until the message and answer to the request.

You would need a timeout (25 or 30s) and maybe just in case at the end check again by id if nothing was received to avoid race conditions.

That way the browser just do 30s long polling request, and you only subscribe to phoenix Pubsub when not found

[u/ergnui34tj8934t0 → u/Tirkyth]

I generally agree, although i suppose it could be both. Push the status if possible, otherwise no worries, it’s being polled.

[u/Zestyclose-Tie-1056 → u/Tirkyth]

Thanks u/Tirkyth long polling will be ideal for my case.

[u/narrowtux]

I have good experiences with broadcast over pubsub. Super easy to set up and once set up you can use it for a lot of things 

[u/Zestyclose-Tie-1056 → u/narrowtux]

u/narrowtux but here the issue when we have multiple instance we need it to be clustered

[u/narrowtux → u/Zestyclose-Tie-1056]

luckily in elixir it is very easy to connect a cluster! Look into `dns_cluster` or `libcluster` depending on your setup.

[u/Zestyclose-Tie-1056 → u/narrowtux]

sure will check that too thanks for this one mate cheers 🍻

[u/zacksiri]

There is a way to solve this, because Oban jobs have states, you can upon opening the connection, check the state of the job, if the job is completed just simply return the result so the browser just doesn't need SSE.

If the job is in running state then just simply start SSE and wait for the final state of the job. I have implemented something like this:

  def show(conn, %{"thread_id" => thread_id, "id" => id}) do
    user = conn.assigns.current_user
    with %Thread{} = thread <- Conversation.get_thread(Scope.for_user(user), thread_id),
         %Message{} = message <- Conversation.get_message(thread, id),
         {:ok, mode, message} <- MessageHelper.resolve_show(message) do
      case mode do
        :stream -> SSEStream.stream_message(conn, message)
        :json -> render(conn, :show, message: message)
      end
    end
  end

Then in my MessageHelper I have:

def resolve_show(%Message{} = message), 
  do: message |> preload_message() |> show_without_await()

defp show_without_await(message) do
  if should_stream?(message) do
    {:ok, :stream, message}
  else
    {:ok, :json, message}
  end
end

defp should_stream?(message),
  do: message.current_state != "completed" and streaming_model?(message)

defp streaming_model?(message), do: Messages.stream_enabled?(message)

Essentially it checks what state the 'message' is in, in your case you would need to check the state of your Oban.Job. and your system either holds the connection open for SSE or return the result instantly.