[9/10]SSE Reliability: Three Rounds of Fixing Event Loss in JAX-RS SseEventSink

Series: Building a Modular Assessment Engine (9/10)

Server-Sent Events (SSE) is the backbone of the streaming pipeline. When it works, it's elegant — events flow from server to client in real-time. When it breaks, events silently vanish and the user sees a loading spinner forever. This post covers the three rounds of debugging that made SSE reliable.

The Setup

The backend uses JAX-RS (Jakarta REST) with SseEventSink for SSE streaming:

@POST
@Produces(MediaType.SERVER_SENT_EVENTS)
@Consumes(MediaType.APPLICATION_JSON)
public void chat(@Context Sse sse, @Context SseEventSink eventSink,
                 @Context HttpServletRequest http, AgentRequest request) {
    AgentClient client = ExpertBuilder.create()
            .clientId(appId)
            .eventSink(eventSink)
            .sse(sse)
            .build();

    // Plan or execute...
    // Events sent via client.sendToClient()
}

Events are sent through AgentClient.sendToClient():

public void sendToClient(String input, MediaType mediaType) {
    if (!this.hasEventSink()) return;
    OutboundSseEvent sseEvent = sse.newEventBuilder()
            .data(input)
            .mediaType(mediaType)
            .build();
    eventSink.send(sseEvent);  // ← THE PROBLEM IS HERE
}

Round 1: The Missing await

The Bug

eventSink.send() returns CompletionStage<?> — it's asynchronous. The I/O write happens on a different thread. But the original code discarded the return value:

eventSink.send(sseEvent);  // Fire and forget
// ... immediately continues to next line
// ... eventually calls client.close()

The close() call would execute before the I/O write completed. The connection would close, and the last few events (including the critical done event) would be lost.

The client would sit there waiting for a done event that was never actually sent over the wire.

The Fix (Attempt 1)

Add .toCompletableFuture().join() to wait for the write to complete:

eventSink.send(sseEvent).toCompletableFuture().join();

This worked... until it didn't.

Round 2: The Concurrent Write Crash

The Bug

With join() in place, a new problem emerged: UT010035: Stream in async mode was not ready for IO operation.

This happened because two threads were calling send() simultaneously:

• The main execution thread sending task events
• A heartbeat thread sending periodic heartbeats every 10 seconds

JAX-RS SseEventSink is not thread-safe for concurrent writes. Two simultaneous send() calls corrupt the internal buffer state.

The Fix (Attempt 2)

Add a synchronized block:

public void sendToClient(String input, MediaType mediaType) {
    if (!this.hasEventSink()) return;
    synchronized(this) {
        if (!this.hasEventSink()) return;
        eventSink.send(sseEvent).toCompletableFuture().join();
    }
}

This fixed the concurrent write crash... but introduced a new problem on JDK 21+.

Round 3: The Virtual Thread Pinning

The Bug

The system runs on JDK 21 with virtual threads. Virtual threads are lightweight, user-mode threads that should never block the carrier thread. But synchronized blocks pin the virtual thread to its carrier thread — the carrier can't be freed for other virtual threads while one is blocked inside a synchronized block.

With synchronized(this) + join():

1. The heartbeat thread acquires the lock, starts an I/O write, calls join() which blocks
2. The carrier thread is now pinned — it can't run other virtual threads
3. The main execution thread tries to send an event, blocks on the synchronized lock
4. All virtual threads on this carrier are now stuck
5. Under load, the entire thread pool exhausts

The application would freeze after 3-4 concurrent users.

The Final Fix: ReentrantLock + get(timeout)

private final ReentrantLock writeLock = new ReentrantLock();

public void sendToClient(String input, MediaType mediaType) {
    if (!this.hasEventSink()) return;
    writeLock.lock();
    try {
        // Double-check: connection may have closed while waiting for lock
        if (!this.hasEventSink()) return;

        OutboundSseEvent sseEvent = sse.newEventBuilder()
                .data(input)
                .mediaType(mediaType)
                .build();

        // Wait for I/O completion with timeout
        eventSink.send(sseEvent).toCompletableFuture()
                .get(10, TimeUnit.SECONDS);

        lastSendTimeMs = System.currentTimeMillis();
    } catch (TimeoutException e) {
        // I/O didn't complete in 10s — connection is likely dead
        this.close();
    } catch (Exception e) {
        // Unwrap CompletionException to get real cause
        Throwable cause = (e.getCause() != null) ? e.getCause() : e;
        if (cause instanceof IllegalStateException) {
            // Connection already closed
            this.close();
        } else {
            FLog.e(TAG, "SSE send failed", cause);
            this.close();
        }
    } finally {
        writeLock.unlock();
    }
}

Why ReentrantLock Instead of synchronized

ReentrantLock doesn't pin virtual threads. When a virtual thread blocks on ReentrantLock.lock(), the JVM can unmount it from the carrier thread, freeing the carrier for other work. With synchronized, the JVM must keep the virtual thread mounted — this is the "pinning" problem.

Why get(10, SECONDS) Instead of join()

join() blocks indefinitely. If the network connection is half-open (TCP alive but SSE broken), join() waits forever, holding the lock. No other thread can send events. The heartbeat thread can't even send a heartbeat to check if the connection is alive.

get(10, TimeUnit.SECONDS) adds a timeout. If I/O doesn't complete in 10 seconds, something is wrong. The TimeoutException handler calls close(), releasing the lock and cleaning up the connection.

Why Double-Check hasEventSink()

Between acquiring the lock and sending, another thread might have closed the connection:

Thread A: writeLock.lock() ← acquired
Thread B: close() called, eventSink = null
Thread A: eventSink.send() ← NPE!

The double-check prevents this:

Thread A: writeLock.lock() ← acquired
Thread A: if (!hasEventSink()) return ← connection closed by Thread B, exit safely

The Heartbeat Problem

Heartbeats are sent every 10 seconds to keep the SSE connection alive. Without heartbeats, proxies and load balancers close idle connections after 30-60 seconds.

But heartbeats introduce a new problem: the heartbeat thread and the main thread both call sendToClient(). With the ReentrantLock, this is safe — one waits for the other. But the heartbeat thread must not block for too long:

// Heartbeat thread
if (System.currentTimeMillis() - lastSendTimeMs > 10_000) {
    sendToClient("{\"event\":\"heartbeat\"}", MediaType.APPLICATION_JSON_TYPE);
}

If the main thread is mid-write and the heartbeat thread tries to send, it waits on the lock. Once the main thread finishes, the heartbeat sends. Total delay: the duration of one I/O write (usually <1 second). Acceptable.

But what if the main thread's I/O is stuck? The heartbeat thread waits for the lock, and get(10, SECONDS) is the main thread's timeout. After 10 seconds, the main thread calls close(), the lock releases, and the heartbeat thread's sendToClient() sees hasEventSink() == false and exits. Clean.

The JSON Serialization Bug

One more SSE bug, unrelated to threading but worth mentioning: the ConvertUtil.toJson() method had a quirk where it didn't properly escape strings containing special characters.

When a task description contained a quote or newline, the JSON payload would be malformed:

{"event":"task.start","data":{"taskId":"t1","title":"Assess "stress" levels"}}
                                                              ↑ broken JSON

The client's JSON.parse() would throw, and the event would be silently dropped. The user would see a task that never starts.

The fix: ensure ConvertUtil.toJson() properly escapes all string values:

// Before (bug): raw string insertion
sb.append("\"title\":\"").append(task.getTitle()).append("\"");

// After (fixed): proper JSON string escaping
sb.append("\"title\":\"").append(jsonEscape(task.getTitle())).append("\"");

The Nginx Configuration

Behind all the Java code, Nginx needs specific configuration for SSE:

location /api/v1/assess/ {
    proxy_pass http://backend;
    proxy_read_timeout 300s;      # 5 minutes for long-running AI calls
    proxy_buffering off;          # SSE needs unbuffered transfer
    proxy_cache off;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
}

proxy_buffering off is critical. With buffering on, Nginx collects events in a buffer and sends them in batches. The client sees nothing for 30 seconds, then gets everything at once. SSE is supposed to be real-time.

proxy_read_timeout 300s prevents Nginx from closing the connection during long AI generation. Default is 60s, which is too short for a 30-second planning phase.

Lessons Learned

1. SseEventSink.send() is async. Always await it. Discarding the CompletionStage means events are lost. Use .get(timeout) not .join().

2. Virtual threads need ReentrantLock, not synchronized. synchronized pins virtual threads to carrier threads, killing the scalability benefit. This is a JDK 21+ specific issue.

3. Always timeout. Network connections fail in mysterious ways. A timeout is the difference between a recoverable error and a permanently stuck thread.

4. Double-check after acquiring a lock. State can change between lock acquisition and use. The double-check pattern prevents NPEs on closed connections.

5. proxy_buffering off is mandatory for SSE. Without it, events arrive in batches, not in real-time. The UX difference is dramatic.

---

Next: [PDF Rendering War — SVG, emoji, and the iText NPE]