From 4b9778c7be975b645faaf99424d406bd64cc0d9b Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:04:02 +0000 Subject: [PATCH 01/10] feat(realtime): add opt-in bounded transport recovery --- README.md | 21 + .../client/okhttp/RealtimeConnectionTest.kt | 296 +++++++++++++ .../com/openai/helpers/RealtimeConnection.kt | 416 ++++++++++++++---- .../helpers/RealtimeWebSocketOptions.kt | 41 ++ 4 files changed, 679 insertions(+), 95 deletions(-) diff --git a/README.md b/README.md index d79713ce8..fb1312ca1 100644 --- a/README.md +++ b/README.md @@ -2155,3 +2155,24 @@ Please share bug reports and feature requests through [GitHub issues](https://gi Pull requests are limited to repository collaborators; we do not accept pull requests from non-collaborators. See [CONTRIBUTING.md](./CONTRIBUTING.md) for the contribution policy and development guide. For security vulnerabilities, follow [SECURITY.md](SECURITY.md). + + +### Optional Realtime transport recovery + +GA Realtime model connections can opt into transport recovery with +`RealtimeWebSocketOptions.builder().reconnect(true).build()`. It is disabled by +default. A recovered connection keeps the same Java object, but opens a **new +server session**. Transcription and existing-call connections do not support this +option. Normal closure, authentication rejection, and TLS failures remain terminal. + +Recovery refreshes credentials for each attempt and stops after five additional +attempts over the connection's lifetime. `maxReconnectAttempts` and +`reconnectDelayMillis` configure this bound and the initial backoff; backoff is +capped at eight seconds. Closing the connection cancels recovery. + +Sends during recovery are rejected unless `maxPendingBytes` permits a bounded +queue. Only commands that have never been attempted can enter that queue. +Use `onReconnected` to restore application-owned session state, then explicitly +call `flushPending()`. `takePending()` instead returns the retained JSON commands +without sending them. The SDK never replays an uncertain write or restores server +state automatically. Send success is not a server acknowledgment. diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index 1032b48a1..47b892188 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -63,6 +63,302 @@ class RealtimeConnectionTest { .baseUrl(server.url("/v1?inherited=1").toString()) .build() + @Test + fun serializationStartedDuringRecoveryStillWaitsForExplicitFlush() { + MockWebServer().use { server -> + val first = Listener() + val second = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().withWebSocketUpgrade(second)) + val opening = CountDownLatch(1) + val allowOpen = CountDownLatch(1) + val encoding = CountDownLatch(1) + val allowEncode = CountDownLatch(1) + val recovered = CompletableFuture() + val calls = java.util.concurrent.atomic.AtomicInteger() + val customMapper = + jsonMapper() + .rebuild() + .addModule( + com.fasterxml.jackson.databind.module + .SimpleModule() + .addSerializer( + String::class.java, + object : com.fasterxml.jackson.databind.JsonSerializer() { + override fun serialize( + value: String, + generator: com.fasterxml.jackson.core.JsonGenerator, + provider: com.fasterxml.jackson.databind.SerializerProvider, + ) { + if (value == "pending-audio") { + encoding.countDown() + check(allowEncode.await(5, TimeUnit.SECONDS)) + } + generator.writeString(value) + } + }, + ) + ) + .build() + OkHttpClient.builder().build().use { http -> + val configuration = + options(server, http) + .toBuilder() + .jsonMapper(customMapper) + .credential( + com.openai.credential.BearerTokenCredential.create { + if (calls.incrementAndGet() == 2) { + opening.countDown() + check(allowOpen.await(5, TimeUnit.SECONDS)) + } + "fake-recovery" + } + ) + .build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .maxPendingBytes(1024) + .onReconnected { recovered.complete(it) } + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThat(opening.await(5, TimeUnit.SECONDS)).isTrue() + val sent = + CompletableFuture.runAsync { + connection.send( + RealtimeClientEvent.ofInputAudioBufferAppend( + InputAudioBufferAppendEvent.builder() + .audio("pending-audio") + .build() + ) + ) + } + try { + assertThat(encoding.await(5, TimeUnit.SECONDS)).isTrue() + allowOpen.countDown() + recovered.get(5, TimeUnit.SECONDS) + allowEncode.countDown() + sent.get(5, TimeUnit.SECONDS) + assertThat(second.messages).isEmpty() + connection.flushPending() + assertThat(second.messages.poll(5, TimeUnit.SECONDS)) + .contains("pending-audio") + } finally { + allowOpen.countDown() + allowEncode.countDown() + } + } + } + } + } + + @Test + fun recoveryRejectsInheritedSessionSelectorsBeforeOpening() { + MockWebServer().use { server -> + OkHttpClient.builder().build().use { http -> + for (query in + listOf("call_id=existing", "intent=transcription", "%63all_id=existing")) { + val configuration = + options(server, http) + .toBuilder() + .baseUrl(server.url("/v1?$query").toString()) + .build() + assertThatThrownBy { + RealtimeConnection.connect( + configuration, + RealtimeWebSocketOptions.builder().reconnect(true).build(), + ) + } + .hasMessageContaining("requires a model session") + } + assertThat(server.requestCount).isZero() + } + } + } + + @Test + fun synchronousRecoveryCallbackCanWaitForAnEvent() { + MockWebServer().use { server -> + val first = Listener() + val second = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().withWebSocketUpgrade(second)) + val callbackStarted = CountDownLatch(1) + val received = CompletableFuture() + OkHttpClient.builder().build().use { http -> + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .onReconnected { + callbackStarted.countDown() + received.complete(it.receive().asInputAudioBufferCleared().eventId()) + } + .build() + RealtimeConnection.connect(options(server, http), settings).use { + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThat(callbackStarted.await(5, TimeUnit.SECONDS)).isTrue() + second.socket + .get(5, TimeUnit.SECONDS) + .send("""{"type":"input_audio_buffer.cleared","event_id":"restored"}""") + assertThat(received.get(5, TimeUnit.SECONDS)).isEqualTo("restored") + } + } + } + } + + @Test + fun automaticRecoveryRefreshesCredentialsAndOnlyFlushesWhenAsked() { + for (async in listOf(false, true)) { + MockWebServer().use { server -> + val first = Listener() + val second = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().withWebSocketUpgrade(second)) + val refreshing = CountDownLatch(1) + val allowRefresh = CountDownLatch(1) + val calls = java.util.concurrent.atomic.AtomicInteger() + val recovered = CompletableFuture() + OkHttpClient.builder().build().use { http -> + val configuration = + options(server, http) + .toBuilder() + .credential( + com.openai.credential.BearerTokenCredential.create { + val attempt = calls.incrementAndGet() + if (attempt == 2) { + refreshing.countDown() + check(allowRefresh.await(5, TimeUnit.SECONDS)) + } + "fake-recovery-$attempt" + } + ) + .build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .maxReconnectAttempts(1) + .reconnectDelayMillis(1) + .maxPendingBytes(100) + .onReconnected { recovered.complete(it) } + .build() + val connection = + if (async) + RealtimeConnection.connectAsync(configuration, settings) + .get(5, TimeUnit.SECONDS) + else RealtimeConnection.connect(configuration, settings) + connection.use { + it.send(command()) + assertThat(first.messages.poll(5, TimeUnit.SECONDS)) + .contains("input_audio_buffer.clear") + val nextEvent = it.receiveAsync() + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThat(refreshing.await(5, TimeUnit.SECONDS)).isTrue() + it.send(command()) + allowRefresh.countDown() + assertThat(recovered.get(5, TimeUnit.SECONDS)).isSameAs(it) + assertThat(second.messages).isEmpty() + it.flushPending() + assertThat(second.messages.poll(5, TimeUnit.SECONDS)) + .contains("input_audio_buffer.clear") + assertThat(it.takePending()).isEmpty() + second.socket + .get(5, TimeUnit.SECONDS) + .send("""{"type":"input_audio_buffer.cleared","event_id":"fresh"}""") + assertThat( + nextEvent + .get(5, TimeUnit.SECONDS) + .asInputAudioBufferCleared() + .eventId() + ) + .isEqualTo("fresh") + val exhausted = it.receiveAsync() + second.socket.get().close(1012, "again") + assertThatThrownBy { exhausted.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage("GA Realtime WebSocket closed (code 1012)") + } + assertThat(calls.get()).isEqualTo(2) + assertThat(server.takeRequest(5, TimeUnit.SECONDS)!!.getHeader("Authorization")) + .isEqualTo("Bearer fake-recovery-1") + assertThat(server.takeRequest(5, TimeUnit.SECONDS)!!.getHeader("Authorization")) + .isEqualTo("Bearer fake-recovery-2") + assertThat(server.requestCount).isEqualTo(2) + } + } + } + } + + @Test + fun recoveryCanBeCanceledWhileRefreshingAndDoesNotAdmitUnboundedPendingWork() { + MockWebServer().use { server -> + val first = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + val refreshing = CountDownLatch(1) + val allowRefresh = CountDownLatch(1) + val calls = java.util.concurrent.atomic.AtomicInteger() + OkHttpClient.builder().build().use { http -> + val configuration = + options(server, http) + .toBuilder() + .credential( + com.openai.credential.BearerTokenCredential.create { + if (calls.incrementAndGet() > 1) { + refreshing.countDown() + check(allowRefresh.await(5, TimeUnit.SECONDS)) + } + "fake-recovery" + } + ) + .build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + val nextEvent = connection.receiveAsync() + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThat(refreshing.await(5, TimeUnit.SECONDS)).isTrue() + assertThatThrownBy { connection.send(command()) } + .hasMessageContaining("queue is full") + connection.close() + allowRefresh.countDown() + assertThatThrownBy { nextEvent.get(5, TimeUnit.SECONDS) } + .isInstanceOf(java.util.concurrent.CancellationException::class.java) + assertThat(connection.takePending()).isEmpty() + assertThat(server.requestCount).isEqualTo(1) + } + } + } + } + + @Test + fun recoveryDoesNotRetryAuthenticationRejection() { + MockWebServer().use { server -> + val first = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().setResponseCode(401)) + OkHttpClient.builder().build().use { http -> + RealtimeConnection.connect( + options(server, http), + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .build(), + ) + .use { connection -> + val pending = connection.receiveAsync() + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThatThrownBy { pending.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage("WebSocket handshake failed (HTTP 401)") + assertThat(server.requestCount).isEqualTo(2) + } + } + } + } + @Test fun blockingContinuationAfterHandshakeDoesNotBlockIncomingFrames() { MockWebServer().use { server -> diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 8239cf37f..3982bda5a 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -9,6 +9,7 @@ import com.openai.core.RequestOptions import com.openai.core.http.HttpMethod import com.openai.core.http.HttpRequest import com.openai.core.http.WebSocketClient +import com.openai.core.http.WebSocketHandshakeException import com.openai.core.http.WebSocketReadFuture import com.openai.core.http.WebSocketWriteNotAttempted import com.openai.core.http.awaitWebSocketFuture @@ -21,6 +22,10 @@ import java.io.IOException import java.util.ArrayDeque import java.util.concurrent.CancellationException import java.util.concurrent.CompletableFuture +import java.util.concurrent.ScheduledFuture +import java.util.concurrent.ScheduledThreadPoolExecutor +import java.util.concurrent.TimeUnit +import javax.net.ssl.SSLException /** * Managed GA Realtime connection. Consume typed events synchronously or asynchronously, with one @@ -57,8 +62,21 @@ private constructor( private var failure: Throwable? = null private var closed = false private var sending = false + private var generation = 0L + private var recovering = false + private var attempts = 0 + private var retryTimer: ScheduledFuture<*>? = null + private val pending = ArrayDeque() + private var pendingBytes = 0L + private var flushing = false companion object { + private val recoveryTimer = + ScheduledThreadPoolExecutor(1) { task -> + Thread(task, "openai-realtime-recovery").apply { isDaemon = true } + } + .apply { removeOnCancelPolicy = true } + @JvmStatic @JvmOverloads @MustBeClosed @@ -87,109 +105,246 @@ private constructor( val connection = RealtimeConnection(clientOptions, options, requestOptions) val result = CompletableFuture() result.whenComplete { _, _ -> if (result.isCancelled) connection.close() } - try { - clientOptions.requireWebSocketTransport() + connection.open(result, blocking) + return result + } + + private fun await(future: CompletableFuture): T = + awaitWebSocketFuture(future, "Realtime") + } + + private fun open(result: CompletableFuture, blocking: Boolean) { + val epoch = + synchronized(lock) { + if (closed || failure != null) return + retryTimer = null + ++generation + } + try { + clientOptions.requireWebSocketTransport() + require( + AzureUrlCategory.categorizeBaseUrl( + clientOptions.baseUrl(), + clientOptions.azureUrlPathMode, + ) != AzureUrlCategory.AZURE_LEGACY + ) { + "GA Realtime WebSockets require unified Azure routing" + } + val request = + HttpRequest.builder() + .method(HttpMethod.GET) + .baseUrl(clientOptions.baseUrl()) + .pathSegments( + com.openai.models.realtime.RealtimeWebSocketBinding.create().pathSegments + ) + .build() + .prepare( + clientOptions, + options, + com.openai.models.realtime.RealtimeWebSocketBinding.create().security, + ) + if (options.reconnect) { + val baseQueryKeys = + java.net + .URI(request.baseUrl) + .rawQuery + ?.split('&') + ?.map { java.net.URLDecoder.decode(it.substringBefore('='), "UTF-8") } + .orEmpty() require( - AzureUrlCategory.categorizeBaseUrl( - clientOptions.baseUrl(), - clientOptions.azureUrlPathMode, - ) != AzureUrlCategory.AZURE_LEGACY + "intent" !in baseQueryKeys && + "call_id" !in baseQueryKeys && + request.queryParams.values("intent").isEmpty() && + request.queryParams.values("call_id").isEmpty() ) { - "GA Realtime WebSockets require unified Azure routing" + "Automatic Realtime recovery requires a model session" } - val request = - HttpRequest.builder() - .method(HttpMethod.GET) - .baseUrl(clientOptions.baseUrl()) - .pathSegments( - com.openai.models.realtime.RealtimeWebSocketBinding.create() - .pathSegments - ) - .build() - .prepare( - clientOptions, - options, - com.openai.models.realtime.RealtimeWebSocketBinding.create().security, - ) - val pending = - clientOptions.connectWebSocket( - request, - requestOptions, - options.maxMessageBytes, - object : WebSocketClient.Listener { - override fun onMessage(text: String) = connection.accept(text) + } + if (synchronized(lock) { generation != epoch || closed || failure != null }) return + val pending = + clientOptions.connectWebSocket( + request, + requestOptions, + options.maxMessageBytes, + object : WebSocketClient.Listener { + override fun onMessage(text: String) = accept(text, epoch) - override fun onClosed(code: Int) = - connection.fail( - IOException("GA Realtime WebSocket closed (code $code)") - ) + override fun onClosed(code: Int) { + disconnect( + epoch, + IOException("GA Realtime WebSocket closed (code $code)"), + code in setOf(1006, 1011, 1012, 1013), + result, + blocking, + ) + } - override fun onFailure(error: Throwable) = connection.fail(error) - }, - ) - val shouldCancel = - synchronized(connection.lock) { - if (connection.closed || connection.failure != null) true + override fun onFailure(error: Throwable) { + disconnect(epoch, error, canRetry(error), result, blocking) + } + }, + ) + val shouldCancel = + synchronized(lock) { + if (closed || failure != null || generation != epoch) true + else { + this.opening = pending + false + } + } + if (shouldCancel) pending.cancel(true) + pending.whenComplete { active, error -> + if (synchronized(lock) { generation != epoch }) { + val problem = synchronized(lock) { failure } + try { + active?.close() + } catch (closing: Throwable) { + if (problem != null && closing !== problem) problem.addSuppressed(closing) + } finally { + problem?.let { completeFailure(result, it, blocking) } + } + return@whenComplete + } + if (error != null && disconnect(epoch, error, canRetry(error), result, blocking)) + return@whenComplete + val problem = + synchronized(lock) { + this.opening = null + if (error != null) error + else if (this.closed || this.failure != null) + this.failure ?: CancellationException("Realtime connection closed") else { - connection.opening = pending - false + this.socket = active + this.recovering = false + null } } - if (shouldCancel) pending.cancel(true) - pending.whenComplete { active, error -> - val problem = - synchronized(connection.lock) { - connection.opening = null - if (error != null) error - else if (connection.closed || connection.failure != null) - connection.failure - ?: CancellationException("Realtime connection closed") - else { - connection.socket = active - null + if (problem != null) { + try { + active?.close() + } catch (closing: Throwable) { + if (closing !== problem) problem.addSuppressed(closing) + } + this.fail(problem) + } + val complete: () -> Unit = completion@{ + if (result.isDone && synchronized(lock) { generation != epoch }) + return@completion + // The owner or socket can close while completion waits on the executor. + // Inspect state here, but keep user callbacks outside the lock. + val completionError = + problem + ?: synchronized(lock) { + when { + this.failure != null -> this.failure + this.closed -> + CancellationException("Realtime connection closed") + clientOptions.isWebSocketClosed() -> + CancellationException("Client is closed") + else -> null + } } - } - if (problem != null) { + if (completionError != null) { + this.fail(completionError) + result.completeExceptionally(completionError) + } else if (!result.isDone) { + if (!result.complete(this)) close() + } else if (result.isCancelled) close() + else { try { - active?.close() - } catch (closing: Throwable) { - if (closing !== problem) problem.addSuppressed(closing) + options.onReconnected?.accept(this) + } catch (callbackError: Exception) { + fail(callbackError) } - connection.fail(problem) } - val complete: () -> Unit = { - // The owner or socket can close while completion waits on the executor. - // Inspect state here, but keep user callbacks outside the lock. - val completionError = - problem - ?: synchronized(connection.lock) { - when { - connection.failure != null -> connection.failure - connection.closed -> - CancellationException("Realtime connection closed") - clientOptions.isWebSocketClosed() -> - CancellationException("Client is closed") - else -> null - } - } - if (completionError != null) { - connection.fail(completionError) - result.completeExceptionally(completionError) - } else if (!result.complete(connection)) connection.close() - } - // Blocking futures are private: a caller may already occupy the stream - // executor. Public async continuations must still leave the transport reader. - if (blocking) complete() else connection.dispatch(complete) } - } catch (error: Exception) { - connection.fail(error) - result.completeExceptionally(error) + // Blocking futures are private: a caller may already occupy the stream + // executor. Public async continuations must still leave the transport reader. + if (blocking && !result.isDone) complete() else this.dispatch(complete) + } + } catch (error: Exception) { + if (!disconnect(epoch, error, canRetry(error), result, blocking)) { + fail(error) + completeFailure(result, error, blocking) } - return result } + } - private fun await(future: CompletableFuture): T = - awaitWebSocketFuture(future, "Realtime") + private fun canRetry(error: Throwable): Boolean { + var cause: Throwable? = error + val seen = + java.util.Collections.newSetFromMap(java.util.IdentityHashMap()) + while (cause != null && seen.add(cause)) { + if (cause is SSLException || cause is CancellationException) return false + cause = cause.cause + } + val root = + if (error is java.util.concurrent.CompletionException && error.cause != null) + error.cause!! + else error + return root is IOException || + (root is WebSocketHandshakeException && + root.statusCode in setOf(408, 429, 500, 502, 503, 504)) + } + + private fun completeFailure( + result: CompletableFuture, + error: Throwable, + blocking: Boolean, + ) { + if (blocking) result.completeExceptionally(error) + else dispatch { result.completeExceptionally(error) } + } + + private fun disconnect( + epoch: Long, + error: Throwable, + retryable: Boolean, + result: CompletableFuture, + blocking: Boolean, + ): Boolean { + val retired = + synchronized(lock) { + if (generation != epoch) return true + if ( + !closed && + failure == null && + options.reconnect && + retryable && + (socket != null || recovering) && + !sending && + attempts < options.maxReconnectAttempts + ) { + generation++ + recovering = true + attempts++ + val old = Pair(socket, opening) + socket = null + opening = null + val delay = + (options.reconnectDelayMillis * (1L shl minOf(attempts - 1, 14))) + .coerceAtMost(8000) + retryTimer = + recoveryTimer.schedule( + { dispatch { open(result, blocking) } }, + delay, + TimeUnit.MILLISECONDS, + ) + old + } else null + } + if (retired != null) { + retired.second?.cancel(true) + try { + retired.first?.close() + } catch (_: Exception) { + /* Preserve the initiating error. */ + } + return true + } + fail(error) + completeFailure(result, error, blocking) + return false } /** Returns one event, even if the peer closes immediately after it. */ @@ -226,11 +381,16 @@ private constructor( * resends the command. */ fun send(event: RealtimeClientEvent) { - synchronized(lock) { - check(!closed && failure == null && socket != null) { "Realtime is not connected" } - check(!sending) { "A Realtime send is already pending; command was not sent" } - sending = true - } + val admittedDuringRecovery = + synchronized(lock) { + check(!closed && failure == null && (socket != null || recovering)) { + "Realtime is not connected" + } + check(!flushing) { "A Realtime flush is already pending; command was not sent" } + check(!sending) { "A Realtime send is already pending; command was not sent" } + sending = true + recovering + } try { val text = clientOptions.jsonMapper.writeValueAsString(event) require(utf8Size(text) <= options.maxMessageBytes) { @@ -239,6 +399,16 @@ private constructor( val active = synchronized(lock) { check(!closed && failure == null) { "Realtime is not connected" } + if (admittedDuringRecovery || recovering) { + val size = utf8Size(text) + if (size > options.maxPendingBytes - pendingBytes) + throw WebSocketWriteNotAttempted.Busy( + "Realtime recovery queue is full; command was not sent" + ) + pending.addLast(text) + pendingBytes += size + return + } socket ?: throw IllegalStateException("Realtime is not connected") } try { @@ -252,6 +422,58 @@ private constructor( } } + /** Remove retained, never-attempted JSON commands without sending them. */ + fun takePending(): List = + synchronized(lock) { + check(!flushing) { "A Realtime flush is pending" } + pending.toList().also { + pending.clear() + pendingBytes = 0 + } + } + + /** Send retained commands in order after the application restores session state. */ + fun flushPending() { + synchronized(lock) { + check( + !flushing && !sending && !recovering && !closed && failure == null && socket != null + ) { + "Realtime cannot flush now" + } + flushing = true + } + try { + while (true) { + val next = + synchronized(lock) { + check(!closed && failure == null && !recovering) { + "Realtime is not connected" + } + if (pending.isEmpty()) return + val text = pending.removeFirst() + pendingBytes -= utf8Size(text) + sending = true + Pair(socket!!, text) + } + try { + next.first.send(next.second) + } catch (error: Throwable) { + if (error is WebSocketWriteNotAttempted) + synchronized(lock) { + pending.addFirst(next.second) + pendingBytes += utf8Size(next.second) + } + else fail(error) + throw error + } finally { + synchronized(lock) { sending = false } + } + } + } finally { + synchronized(lock) { flushing = false } + } + } + /** Closes this socket and returns a new session. No input or session updates are replayed. */ @JvmOverloads @MustBeClosed @@ -279,7 +501,8 @@ private constructor( fail(CancellationException("Realtime connection closed")) } - private fun accept(text: String) { + private fun accept(text: String, epoch: Long) { + if (synchronized(lock) { epoch != generation }) return try { // Reject malformed envelopes independently of lenient or user-supplied model readers. clientOptions.jsonMapper.factory.createParser(text).use { parser -> @@ -305,7 +528,7 @@ private constructor( val bytes = utf8Size(text) val target = synchronized(lock) { - if (closed || failure != null) return + if (closed || failure != null || epoch != generation) return val available = waiter?.takeUnless { it.claimed || it.cancellationReserved || it.isDone } if (available == null) { @@ -323,7 +546,7 @@ private constructor( if (target.blocking) target.complete(event) else dispatch { target.complete(event) } } } catch (error: Exception) { - fail(error) + if (synchronized(lock) { epoch == generation }) fail(error) } } @@ -332,6 +555,9 @@ private constructor( synchronized(lock) { if (failure != null) return failure = error + generation++ + retryTimer?.cancel(false) + retryTimer = null val unassigned = waiter?.takeUnless { it.claimed || it.cancellationReserved || it.isDone } waiter = null diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeWebSocketOptions.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeWebSocketOptions.kt index 61a9f3563..8c3f8b22e 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeWebSocketOptions.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeWebSocketOptions.kt @@ -10,6 +10,11 @@ private constructor( @get:JvmName("maxMessageBytes") val maxMessageBytes: Int, @get:JvmName("maxQueuedBytes") val maxQueuedBytes: Long, @get:JvmName("maxQueuedEvents") val maxQueuedEvents: Int, + @get:JvmName("reconnect") val reconnect: Boolean, + @get:JvmName("maxReconnectAttempts") val maxReconnectAttempts: Int, + @get:JvmName("reconnectDelayMillis") val reconnectDelayMillis: Long, + @get:JvmName("maxPendingBytes") val maxPendingBytes: Long, + internal val onReconnected: java.util.function.Consumer?, private val headers: Headers, private val queryParams: QueryParams, ) : Params { @@ -27,6 +32,11 @@ private constructor( private var maxMessageBytes = Int.MAX_VALUE private var maxQueuedBytes = Long.MAX_VALUE private var maxQueuedEvents = Int.MAX_VALUE + private var reconnect = false + private var maxReconnectAttempts = 5 + private var reconnectDelayMillis = 500L + private var maxPendingBytes = 0L + private var onReconnected: java.util.function.Consumer? = null private val headers = Headers.builder() private val queryParams = QueryParams.builder() @@ -58,11 +68,42 @@ private constructor( maxQueuedEvents = value } + /** Enable bounded recovery of model sessions. Default: false. No command is replayed. */ + fun reconnect(value: Boolean) = apply { reconnect = value } + + /** Additional attempts over the logical connection's lifetime. Default: 5. */ + fun maxReconnectAttempts(value: Int) = apply { + require(value >= 0) + maxReconnectAttempts = value + } + + /** Initial backoff, doubled up to eight seconds. Default: 500 milliseconds. */ + fun reconnectDelayMillis(value: Long) = apply { + require(value in 0..8000) + reconnectDelayMillis = value + } + + /** Encoded bytes retained during recovery until explicitly flushed. Default: 0. */ + fun maxPendingBytes(value: Long) = apply { + require(value >= 0) + maxPendingBytes = value + } + + /** Restore application-owned session state before explicitly flushing pending commands. */ + fun onReconnected(value: java.util.function.Consumer) = apply { + onReconnected = value + } + fun build() = RealtimeWebSocketOptions( maxMessageBytes, maxQueuedBytes, maxQueuedEvents, + reconnect, + maxReconnectAttempts, + reconnectDelayMillis, + maxPendingBytes, + onReconnected, headers.build(), queryParams.build(), ) From 436916e1e188e30b57040240d9f9ba9d34b81183 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:25:15 +0000 Subject: [PATCH 02/10] refactor(realtime): simplify recovery failure handling and docs --- README.md | 41 +++++++++---------- .../com/openai/helpers/RealtimeConnection.kt | 9 ++-- 2 files changed, 23 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index fb1312ca1..60880c0f8 100644 --- a/README.md +++ b/README.md @@ -2149,30 +2149,29 @@ We take backwards-compatibility seriously and work hard to ensure you can rely o We are keen for your feedback; please open an [issue](https://www.github.com/openai/openai-java/issues) with questions, bugs, or suggestions. +## Reconnecting to Realtime + +Set `RealtimeWebSocketOptions.builder().reconnect(true).build()` to reconnect +after a temporary connection failure. This option is off by default. It does not +support transcription sessions or connections to an existing call. The SDK does +not retry normal closures, rejected credentials, or TLS failures. + +Reconnecting keeps the same Java object but starts a new server session. The SDK +refreshes credentials for each attempt and allows five reconnect attempts in +total by default. Use `maxReconnectAttempts` to change this limit and +`reconnectDelayMillis` to set the initial delay. Delays grow to eight seconds. +Closing the connection cancels pending retries. + +By default, the SDK rejects sends while reconnecting. Set `maxPendingBytes` to +hold unsent commands up to that size in bytes. Use `onReconnected` to restore any +session settings and conversation state your application needs. Then call +`flushPending()` to send the waiting commands. To remove them without sending, +call `takePending()` instead. The SDK does not resend commands it already tried +to send. A successful send does not confirm that the server received the command. + ## Contributing Please share bug reports and feature requests through [GitHub issues](https://github.com/openai/openai-java/issues). Pull requests are limited to repository collaborators; we do not accept pull requests from non-collaborators. See [CONTRIBUTING.md](./CONTRIBUTING.md) for the contribution policy and development guide. For security vulnerabilities, follow [SECURITY.md](SECURITY.md). - - -### Optional Realtime transport recovery - -GA Realtime model connections can opt into transport recovery with -`RealtimeWebSocketOptions.builder().reconnect(true).build()`. It is disabled by -default. A recovered connection keeps the same Java object, but opens a **new -server session**. Transcription and existing-call connections do not support this -option. Normal closure, authentication rejection, and TLS failures remain terminal. - -Recovery refreshes credentials for each attempt and stops after five additional -attempts over the connection's lifetime. `maxReconnectAttempts` and -`reconnectDelayMillis` configure this bound and the initial backoff; backoff is -capped at eight seconds. Closing the connection cancels recovery. - -Sends during recovery are rejected unless `maxPendingBytes` permits a bounded -queue. Only commands that have never been attempted can enter that queue. -Use `onReconnected` to restore application-owned session state, then explicitly -call `flushPending()`. `takePending()` instead returns the retained JSON commands -without sending them. The SDK never replays an uncertain write or restores server -state automatically. Send success is not a server acknowledgment. diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 3982bda5a..110ecda74 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -263,10 +263,7 @@ private constructor( if (blocking && !result.isDone) complete() else this.dispatch(complete) } } catch (error: Exception) { - if (!disconnect(epoch, error, canRetry(error), result, blocking)) { - fail(error) - completeFailure(result, error, blocking) - } + disconnect(epoch, error, canRetry(error), result, blocking) } } @@ -393,14 +390,14 @@ private constructor( } try { val text = clientOptions.jsonMapper.writeValueAsString(event) - require(utf8Size(text) <= options.maxMessageBytes) { + val size = utf8Size(text) + require(size <= options.maxMessageBytes) { "Realtime command exceeds maxMessageBytes" } val active = synchronized(lock) { check(!closed && failure == null) { "Realtime is not connected" } if (admittedDuringRecovery || recovering) { - val size = utf8Size(text) if (size > options.maxPendingBytes - pendingBytes) throw WebSocketWriteNotAttempted.Busy( "Realtime recovery queue is full; command was not sent" From 29eb86d65b8a78d593734b2ec9b63682b7b23812 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 04:30:41 +0000 Subject: [PATCH 03/10] style(realtime): apply repository Kotlin formatting --- .../src/main/kotlin/com/openai/helpers/RealtimeConnection.kt | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 110ecda74..10d82dee5 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -391,9 +391,7 @@ private constructor( try { val text = clientOptions.jsonMapper.writeValueAsString(event) val size = utf8Size(text) - require(size <= options.maxMessageBytes) { - "Realtime command exceeds maxMessageBytes" - } + require(size <= options.maxMessageBytes) { "Realtime command exceeds maxMessageBytes" } val active = synchronized(lock) { check(!closed && failure == null) { "Realtime is not connected" } From 2d0f603dbce92296d3d6204112a46726a09dc074 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 05:07:25 +0000 Subject: [PATCH 04/10] fix(realtime): retain unsent commands across recovery --- README.md | 3 +- .../openai/client/okhttp/BoundedWebSocket.kt | 44 +++++++------- .../client/okhttp/RealtimeConnectionTest.kt | 26 +++++++- .../client/okhttp/ResponsesWebSocketTest.kt | 6 ++ .../core/http/WebSocketWriteNotAttempted.kt | 3 + .../com/openai/helpers/RealtimeConnection.kt | 59 +++++++++++++++---- 6 files changed, 104 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 60880c0f8..31d60ffd7 100644 --- a/README.md +++ b/README.md @@ -2164,7 +2164,8 @@ Closing the connection cancels pending retries. By default, the SDK rejects sends while reconnecting. Set `maxPendingBytes` to hold unsent commands up to that size in bytes. Use `onReconnected` to restore any -session settings and conversation state your application needs. Then call +session settings and conversation state your application needs. Pause other +application sends while restoring state; the callback does not block them. Then call `flushPending()` to send the waiting commands. To remove them without sending, call `takePending()` instead. The SDK does not resend commands it already tried to send. A successful send does not confirm that the server received the command. diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt index d446517b3..9027183d1 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt @@ -3,6 +3,7 @@ package com.openai.client.okhttp import com.openai.core.RequestOptions import com.openai.core.http.WebSocketClient import com.openai.core.http.WebSocketHandshakeException +import com.openai.core.http.WebSocketWriteNotAttempted import java.io.IOException import java.time.Duration import java.util.concurrent.CompletableFuture @@ -214,32 +215,31 @@ internal object BoundedWebSocket { override fun send(text: String) { synchronized(this) { - check(!ended.get()) { "WebSocket is closed" } - require(text.isNotEmpty()) { - "WebSocket command must not be empty" - } + if (ended.get()) + throw WebSocketWriteNotAttempted.Closed( + "WebSocket is closed" + ) + if (text.isEmpty()) + throw WebSocketWriteNotAttempted.Message( + "WebSocket command must not be empty" + ) val size = text.utf8Size() if (size > maxMessageBytes) - throw com.openai.core.http - .WebSocketWriteNotAttempted - .Message( - "WebSocket command exceeds maxMessageBytes" - ) + throw WebSocketWriteNotAttempted.Message( + "WebSocket command exceeds maxMessageBytes" + ) if (size > MAX_SEND_BYTES) - throw com.openai.core.http - .WebSocketWriteNotAttempted - .Message( - "WebSocket command exceeds OkHttp's 16 MiB send limit" - ) + throw WebSocketWriteNotAttempted.Message( + "WebSocket command exceeds OkHttp's 16 MiB send limit" + ) if (webSocket.queueSize() != 0L) - throw com.openai.core.http - .WebSocketWriteNotAttempted - .Busy( - "WebSocket send buffer is full; command was not queued" - ) - check(webSocket.send(text)) { - "WebSocket is closed; command was not queued" - } + throw WebSocketWriteNotAttempted.Busy( + "WebSocket send buffer is full; command was not queued" + ) + if (!webSocket.send(text)) + throw WebSocketWriteNotAttempted.Closed( + "WebSocket is closed; command was not queued" + ) } } diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index 47b892188..e2edb8ed1 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -64,7 +64,14 @@ class RealtimeConnectionTest { .build() @Test - fun serializationStartedDuringRecoveryStillWaitsForExplicitFlush() { + fun serializationStartedDuringRecoveryStillWaitsForExplicitFlush() = + serializationWaitsForExplicitFlush(startDuringRecovery = true) + + @Test + fun serializationStartedBeforeRecoveryStillWaitsForExplicitFlush() = + serializationWaitsForExplicitFlush(startDuringRecovery = false) + + private fun serializationWaitsForExplicitFlush(startDuringRecovery: Boolean) { MockWebServer().use { server -> val first = Listener() val second = Listener() @@ -123,8 +130,11 @@ class RealtimeConnectionTest { .onReconnected { recovered.complete(it) } .build() RealtimeConnection.connect(configuration, settings).use { connection -> - first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") - assertThat(opening.await(5, TimeUnit.SECONDS)).isTrue() + val original = first.socket.get(5, TimeUnit.SECONDS) + if (startDuringRecovery) { + original.close(1012, "restart") + assertThat(opening.await(5, TimeUnit.SECONDS)).isTrue() + } val sent = CompletableFuture.runAsync { connection.send( @@ -137,6 +147,10 @@ class RealtimeConnectionTest { } try { assertThat(encoding.await(5, TimeUnit.SECONDS)).isTrue() + if (!startDuringRecovery) { + original.close(1012, "restart") + assertThat(opening.await(5, TimeUnit.SECONDS)).isTrue() + } allowOpen.countDown() recovered.get(5, TimeUnit.SECONDS) allowEncode.countDown() @@ -257,12 +271,18 @@ class RealtimeConnectionTest { first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") assertThat(refreshing.await(5, TimeUnit.SECONDS)).isTrue() it.send(command()) + it.send( + RealtimeClientEvent.ofInputAudioBufferClear( + InputAudioBufferClearEvent.builder().eventId("second").build() + ) + ) allowRefresh.countDown() assertThat(recovered.get(5, TimeUnit.SECONDS)).isSameAs(it) assertThat(second.messages).isEmpty() it.flushPending() assertThat(second.messages.poll(5, TimeUnit.SECONDS)) .contains("input_audio_buffer.clear") + assertThat(second.messages.poll(5, TimeUnit.SECONDS)).contains("second") assertThat(it.takePending()).isEmpty() second.socket .get(5, TimeUnit.SECONDS) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt index aa2b813f5..c4f84fc6c 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt @@ -2,6 +2,7 @@ package com.openai.client.okhttp import com.fasterxml.jackson.databind.JsonNode import com.openai.core.http.ResponseWebSocketOptions +import com.openai.core.http.WebSocketWriteNotAttempted import com.openai.core.jsonMapper import com.openai.models.responses.ResponsesClientEvent import java.io.DataInputStream @@ -1541,6 +1542,7 @@ class ResponsesWebSocketTest { .use { connection -> assertThatThrownBy { connection.send("") } .isInstanceOf(IllegalArgumentException::class.java) + .isInstanceOf(WebSocketWriteNotAttempted::class.java) .hasMessageContaining("empty") connection.send(command) assertThat( @@ -1549,6 +1551,10 @@ class ResponsesWebSocketTest { .asText() ) .isEqualTo("response.completed") + connection.close() + assertThatThrownBy { connection.send(command) } + .isInstanceOf(IllegalStateException::class.java) + .isInstanceOf(WebSocketWriteNotAttempted::class.java) } peer.await() } diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/WebSocketWriteNotAttempted.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/WebSocketWriteNotAttempted.kt index 29a895555..e59cd1d23 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/WebSocketWriteNotAttempted.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/WebSocketWriteNotAttempted.kt @@ -10,4 +10,7 @@ interface WebSocketWriteNotAttempted { /** The transport is busy. Retains the original IllegalStateException contract. */ class Busy(message: String) : IllegalStateException(message), WebSocketWriteNotAttempted + + /** The transport is closed. Retains the original IllegalStateException contract. */ + class Closed(message: String) : IllegalStateException(message), WebSocketWriteNotAttempted } diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 10d82dee5..9de41a61b 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -16,15 +16,18 @@ import com.openai.core.http.awaitWebSocketFuture import com.openai.core.http.dispatchWebSocketCallback import com.openai.core.http.utf8Size import com.openai.core.prepare +import com.openai.errors.OpenAIIoException import com.openai.models.realtime.RealtimeClientEvent import com.openai.models.realtime.RealtimeServerEvent import java.io.IOException +import java.time.Duration import java.util.ArrayDeque import java.util.concurrent.CancellationException import java.util.concurrent.CompletableFuture import java.util.concurrent.ScheduledFuture import java.util.concurrent.ScheduledThreadPoolExecutor import java.util.concurrent.TimeUnit +import java.util.concurrent.TimeoutException import javax.net.ssl.SSLException /** @@ -62,6 +65,7 @@ private constructor( private var failure: Throwable? = null private var closed = false private var sending = false + private var writeAttempted = false private var generation = 0L private var recovering = false private var attempts = 0 @@ -309,7 +313,7 @@ private constructor( options.reconnect && retryable && (socket != null || recovering) && - !sending && + !writeAttempted && attempts < options.maxReconnectAttempts ) { generation++ @@ -378,7 +382,7 @@ private constructor( * resends the command. */ fun send(event: RealtimeClientEvent) { - val admittedDuringRecovery = + val (epoch, admittedDuringRecovery) = synchronized(lock) { check(!closed && failure == null && (socket != null || recovering)) { "Realtime is not connected" @@ -386,7 +390,7 @@ private constructor( check(!flushing) { "A Realtime flush is already pending; command was not sent" } check(!sending) { "A Realtime send is already pending; command was not sent" } sending = true - recovering + generation to recovering } try { val text = clientOptions.jsonMapper.writeValueAsString(event) @@ -395,7 +399,7 @@ private constructor( val active = synchronized(lock) { check(!closed && failure == null) { "Realtime is not connected" } - if (admittedDuringRecovery || recovering) { + if (admittedDuringRecovery || generation != epoch) { if (size > options.maxPendingBytes - pendingBytes) throw WebSocketWriteNotAttempted.Busy( "Realtime recovery queue is full; command was not sent" @@ -404,7 +408,9 @@ private constructor( pendingBytes += size return } - socket ?: throw IllegalStateException("Realtime is not connected") + val active = socket ?: throw IllegalStateException("Realtime is not connected") + writeAttempted = true + active } try { active.send(text) @@ -413,7 +419,10 @@ private constructor( throw error } } finally { - synchronized(lock) { sending = false } + synchronized(lock) { + writeAttempted = false + sending = false + } } } @@ -445,11 +454,17 @@ private constructor( "Realtime is not connected" } if (pending.isEmpty()) return - val text = pending.removeFirst() - pendingBytes -= utf8Size(text) - sending = true - Pair(socket!!, text) + Pair(socket!!, pending.first) + } + awaitWritable(next.first) + synchronized(lock) { + check(!closed && failure == null && !recovering && socket === next.first) { + "Realtime is not connected" } + pending.removeFirst() + pendingBytes -= utf8Size(next.second) + writeAttempted = true + } try { next.first.send(next.second) } catch (error: Throwable) { @@ -461,7 +476,7 @@ private constructor( else fail(error) throw error } finally { - synchronized(lock) { sending = false } + synchronized(lock) { writeAttempted = false } } } } finally { @@ -469,6 +484,28 @@ private constructor( } } + private fun awaitWritable(active: WebSocketClient.Connection) { + if (active !is WebSocketClient.WritableConnection) return + val timeout = (requestOptions.timeout ?: clientOptions.timeout).write() + while (true) { + try { + active.awaitWritable(if (timeout.isZero) Duration.ofSeconds(1) else timeout) + return + } catch (error: InterruptedException) { + Thread.currentThread().interrupt() + throw OpenAIIoException("Realtime WebSocket operation interrupted", error) + } catch (error: TimeoutException) { + if (!timeout.isZero) + throw OpenAIIoException("Realtime WebSocket operation failed", error) + synchronized(lock) { + check(!closed && failure == null && !recovering && socket === active) { + "Realtime is not connected" + } + } + } + } + } + /** Closes this socket and returns a new session. No input or session updates are replayed. */ @JvmOverloads @MustBeClosed From 7e59829cf154fa097e23ec5287c7a45f2de04c3b Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 05:19:17 +0000 Subject: [PATCH 05/10] fix(realtime): isolate retries and inherit write timeouts --- .../client/okhttp/RealtimeConnectionTest.kt | 103 +++++++++++++----- .../com/openai/helpers/RealtimeConnection.kt | 5 +- 2 files changed, 78 insertions(+), 30 deletions(-) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index e2edb8ed1..b70dc3ca5 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -12,6 +12,7 @@ import com.openai.helpers.RealtimeWebSocketOptions import com.openai.models.realtime.InputAudioBufferAppendEvent import com.openai.models.realtime.InputAudioBufferClearEvent import com.openai.models.realtime.RealtimeClientEvent +import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors @@ -193,33 +194,47 @@ class RealtimeConnectionTest { } @Test - fun synchronousRecoveryCallbackCanWaitForAnEvent() { - MockWebServer().use { server -> - val first = Listener() - val second = Listener() - server.enqueue(MockResponse().withWebSocketUpgrade(first)) - server.enqueue(MockResponse().withWebSocketUpgrade(second)) - val callbackStarted = CountDownLatch(1) - val received = CompletableFuture() - OkHttpClient.builder().build().use { http -> - val settings = - RealtimeWebSocketOptions.builder() - .reconnect(true) - .reconnectDelayMillis(1) - .onReconnected { - callbackStarted.countDown() - received.complete(it.receive().asInputAudioBufferCleared().eventId()) - } - .build() - RealtimeConnection.connect(options(server, http), settings).use { - first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") - assertThat(callbackStarted.await(5, TimeUnit.SECONDS)).isTrue() - second.socket - .get(5, TimeUnit.SECONDS) - .send("""{"type":"input_audio_buffer.cleared","event_id":"restored"}""") - assertThat(received.get(5, TimeUnit.SECONDS)).isEqualTo("restored") + fun blockingRecoveryCallbackDoesNotBlockLaterRecovery() { + val executor = Executors.newSingleThreadExecutor() + try { + MockWebServer().use { server -> + val first = Listener() + val second = Listener() + val third = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().withWebSocketUpgrade(second)) + server.enqueue(MockResponse().withWebSocketUpgrade(third)) + val callbackStarted = CountDownLatch(1) + val received = CompletableFuture() + OkHttpClient.builder().build().use { http -> + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .onReconnected { + if (!received.isDone) { + callbackStarted.countDown() + received.complete( + it.receive().asInputAudioBufferCleared().eventId() + ) + } + } + .build() + val configuration = + options(server, http).toBuilder().streamHandlerExecutor(executor).build() + RealtimeConnection.connect(configuration, settings).use { + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThat(callbackStarted.await(5, TimeUnit.SECONDS)).isTrue() + second.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + third.socket + .get(5, TimeUnit.SECONDS) + .send("""{"type":"input_audio_buffer.cleared","event_id":"restored"}""") + assertThat(received.get(5, TimeUnit.SECONDS)).isEqualTo("restored") + } } } + } finally { + executor.shutdownNow() } } @@ -235,10 +250,33 @@ class RealtimeConnectionTest { val allowRefresh = CountDownLatch(1) val calls = java.util.concurrent.atomic.AtomicInteger() val recovered = CompletableFuture() + val writeTimeout = Duration.ofSeconds(3) + val observedTimeouts = mutableListOf() OkHttpClient.builder().build().use { http -> + val observedHttp = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture = + http + .connectWebSocket(request, options, maxMessageBytes, listener) + .thenApply { active -> + val writable = active as WebSocketClient.WritableConnection + object : WebSocketClient.WritableConnection by writable { + override fun awaitWritable(timeout: Duration) { + observedTimeouts.add(timeout) + writable.awaitWritable(timeout) + } + } + } + } val configuration = - options(server, http) + options(server, observedHttp) .toBuilder() + .timeout(com.openai.core.Timeout.builder().write(writeTimeout).build()) .credential( com.openai.credential.BearerTokenCredential.create { val attempt = calls.incrementAndGet() @@ -258,11 +296,19 @@ class RealtimeConnectionTest { .maxPendingBytes(100) .onReconnected { recovered.complete(it) } .build() + val requestOptions = + RequestOptions.builder() + .timeout( + com.openai.core.Timeout.builder() + .connect(Duration.ofSeconds(2)) + .build() + ) + .build() val connection = if (async) - RealtimeConnection.connectAsync(configuration, settings) + RealtimeConnection.connectAsync(configuration, settings, requestOptions) .get(5, TimeUnit.SECONDS) - else RealtimeConnection.connect(configuration, settings) + else RealtimeConnection.connect(configuration, settings, requestOptions) connection.use { it.send(command()) assertThat(first.messages.poll(5, TimeUnit.SECONDS)) @@ -280,6 +326,7 @@ class RealtimeConnectionTest { assertThat(recovered.get(5, TimeUnit.SECONDS)).isSameAs(it) assertThat(second.messages).isEmpty() it.flushPending() + assertThat(observedTimeouts).containsExactly(writeTimeout, writeTimeout) assertThat(second.messages.poll(5, TimeUnit.SECONDS)) .contains("input_audio_buffer.clear") assertThat(second.messages.poll(5, TimeUnit.SECONDS)).contains("second") diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 9de41a61b..363e19c31 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -327,7 +327,7 @@ private constructor( .coerceAtMost(8000) retryTimer = recoveryTimer.schedule( - { dispatch { open(result, blocking) } }, + { CompletableFuture.runAsync { open(result, blocking) } }, delay, TimeUnit.MILLISECONDS, ) @@ -486,7 +486,8 @@ private constructor( private fun awaitWritable(active: WebSocketClient.Connection) { if (active !is WebSocketClient.WritableConnection) return - val timeout = (requestOptions.timeout ?: clientOptions.timeout).write() + val timeout = + (requestOptions.timeout?.assign(clientOptions.timeout) ?: clientOptions.timeout).write() while (true) { try { active.awaitWritable(if (timeout.isZero) Duration.ofSeconds(1) else timeout) From 7111be867f897af20f552035c9bf4d34162adfdd Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 05:47:27 +0000 Subject: [PATCH 06/10] fix(realtime): resolve send admission before disconnect --- .../client/okhttp/RealtimeConnectionTest.kt | 90 +++++++++++++++++++ .../com/openai/helpers/RealtimeConnection.kt | 69 +++++++++++--- 2 files changed, 145 insertions(+), 14 deletions(-) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index b70dc3ca5..b52234c6a 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -6,12 +6,14 @@ import com.openai.core.http.HttpClient import com.openai.core.http.HttpRequest import com.openai.core.http.HttpResponse import com.openai.core.http.WebSocketClient +import com.openai.core.http.WebSocketWriteNotAttempted import com.openai.core.jsonMapper import com.openai.helpers.RealtimeConnection import com.openai.helpers.RealtimeWebSocketOptions import com.openai.models.realtime.InputAudioBufferAppendEvent import com.openai.models.realtime.InputAudioBufferClearEvent import com.openai.models.realtime.RealtimeClientEvent +import java.io.IOException import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.CountDownLatch @@ -64,6 +66,94 @@ class RealtimeConnectionTest { .baseUrl(server.url("/v1?inherited=1").toString()) .build() + @Test + fun failureBeforeSendRejectionRecovers() = failureDuringWrite(flush = false, rejected = true) + + @Test + fun failureBeforeFlushRejectionRecovers() = failureDuringWrite(flush = true, rejected = true) + + @Test + fun failureBeforeAcceptedSendRemainsTerminal() = + failureDuringWrite(flush = false, rejected = false) + + private fun failureDuringWrite(flush: Boolean, rejected: Boolean) { + val opening = CompletableFuture() + val opened = LinkedBlockingQueue() + val recovered = LinkedBlockingQueue() + val sent = LinkedBlockingQueue() + val rejection = WebSocketWriteNotAttempted.Closed("closed before enqueue") + val attempts = java.util.concurrent.atomic.AtomicInteger() + OkHttpClient.builder().build().use { http -> + val transport = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture { + val attempt = attempts.incrementAndGet() + opened.add(listener) + val connection = + object : WebSocketClient.Connection { + override fun send(text: String) { + if (attempt == if (flush) 2 else 1) { + listener.onFailure(IOException("peer reset")) + if (rejected) throw rejection + } + sent.add(text) + } + + override fun close() {} + } + if (flush && attempt == 2) return opening.thenApply { connection } + return CompletableFuture.completedFuture(connection) + } + } + val configuration = + ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .maxPendingBytes(1024) + .onReconnected { recovered.add(it) } + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + val first = opened.poll(5, TimeUnit.SECONDS)!! + if (flush) { + first.onFailure(IOException("initial disconnect")) + assertThat(opened.poll(5, TimeUnit.SECONDS)).isNotNull() + connection.send(command()) + opening.complete(Unit) + assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) + } + val waiting = connection.receiveAsync() + if (!rejected) { + connection.send(command()) + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage("peer reset") + assertThat(attempts.get()).isEqualTo(1) + assertThat(sent).hasSize(1) + return + } + assertThatThrownBy { + if (flush) connection.flushPending() else connection.send(command()) + } + .isSameAs(rejection) + assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) + assertThat(sent).isEmpty() + if (flush) connection.flushPending() else connection.send(command()) + assertThat(sent).hasSize(1) + opened + .poll(5, TimeUnit.SECONDS)!! + .onMessage("""{"type":"input_audio_buffer.cleared","event_id":"recovered"}""") + assertThat(waiting.get(5, TimeUnit.SECONDS).asInputAudioBufferCleared().eventId()) + .isEqualTo("recovered") + } + } + } + @Test fun serializationStartedDuringRecoveryStillWaitsForExplicitFlush() = serializationWaitsForExplicitFlush(startDuringRecovery = true) diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 363e19c31..246ec6f0a 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -57,6 +57,11 @@ private constructor( private class Item(val event: RealtimeServerEvent, val bytes: Long) + private class Write { + var notAttempted: Boolean? = null + var disconnected: (() -> Unit)? = null + } + private val queue = ArrayDeque() private var queuedBytes = 0L private var socket: WebSocketClient.Connection? = null @@ -65,7 +70,7 @@ private constructor( private var failure: Throwable? = null private var closed = false private var sending = false - private var writeAttempted = false + private var write: Write? = null private var generation = 0L private var recovering = false private var attempts = 0 @@ -313,9 +318,18 @@ private constructor( options.reconnect && retryable && (socket != null || recovering) && - !writeAttempted && attempts < options.maxReconnectAttempts ) { + val activeWrite = write + if (activeWrite != null && activeWrite.notAttempted == null) { + if (activeWrite.disconnected == null) + activeWrite.disconnected = { + disconnect(epoch, error, retryable, result, blocking) + } + return true + } + if (activeWrite?.notAttempted == false) return@synchronized null + write = null generation++ recovering = true attempts++ @@ -388,10 +402,13 @@ private constructor( "Realtime is not connected" } check(!flushing) { "A Realtime flush is already pending; command was not sent" } - check(!sending) { "A Realtime send is already pending; command was not sent" } + check(!sending && write == null) { + "A Realtime send is already pending; command was not sent" + } sending = true generation to recovering } + var notAttempted = false try { val text = clientOptions.jsonMapper.writeValueAsString(event) val size = utf8Size(text) @@ -409,20 +426,18 @@ private constructor( return } val active = socket ?: throw IllegalStateException("Realtime is not connected") - writeAttempted = true + write = Write() active } try { active.send(text) } catch (error: Throwable) { - if (error !is WebSocketWriteNotAttempted) fail(error) + notAttempted = error is WebSocketWriteNotAttempted + if (!notAttempted) fail(error) throw error } } finally { - synchronized(lock) { - writeAttempted = false - sending = false - } + finishWrite(notAttempted) { sending = false } } } @@ -440,12 +455,19 @@ private constructor( fun flushPending() { synchronized(lock) { check( - !flushing && !sending && !recovering && !closed && failure == null && socket != null + !flushing && + !sending && + write == null && + !recovering && + !closed && + failure == null && + socket != null ) { "Realtime cannot flush now" } flushing = true } + var notAttempted = false try { while (true) { val next = @@ -463,7 +485,7 @@ private constructor( } pending.removeFirst() pendingBytes -= utf8Size(next.second) - writeAttempted = true + write = Write() } try { next.first.send(next.second) @@ -472,15 +494,33 @@ private constructor( synchronized(lock) { pending.addFirst(next.second) pendingBytes += utf8Size(next.second) + notAttempted = true } else fail(error) throw error - } finally { - synchronized(lock) { writeAttempted = false } } + finishWrite() } } finally { - synchronized(lock) { flushing = false } + finishWrite(notAttempted) { flushing = false } + } + } + + private fun finishWrite(notAttempted: Boolean = false, release: () -> Unit = {}) { + var disconnected: (() -> Unit)? = null + val active = + synchronized(lock) { + val active = write + active?.notAttempted = notAttempted + disconnected = active?.disconnected + if (disconnected == null) write = null + release() + active + } + try { + disconnected?.invoke() + } finally { + synchronized(lock) { if (write === active) write = null } } } @@ -588,6 +628,7 @@ private constructor( synchronized(lock) { if (failure != null) return failure = error + write?.disconnected = null generation++ retryTimer?.cancel(false) retryTimer = null From 87bcafb7b79b901dfa969eb5e2c0490dcd909b7d Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:01:26 +0000 Subject: [PATCH 07/10] fix(realtime): preserve recovery across transport races --- .../openai/client/okhttp/BoundedWebSocket.kt | 11 +- .../client/okhttp/RealtimeConnectionTest.kt | 141 +++++++++++++++++- .../com/openai/helpers/RealtimeConnection.kt | 57 ++++--- 3 files changed, 179 insertions(+), 30 deletions(-) diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt index 9027183d1..6a8c98ac3 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt @@ -5,6 +5,7 @@ import com.openai.core.http.WebSocketClient import com.openai.core.http.WebSocketHandshakeException import com.openai.core.http.WebSocketWriteNotAttempted import java.io.IOException +import java.net.ProtocolException import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.TimeUnit @@ -109,7 +110,7 @@ internal object BoundedWebSocket { ) { chain.call().cancel() response.close() - throw IOException("WebSocket extensions are disabled") + throw ProtocolException("WebSocket extensions are disabled") } response } @@ -258,13 +259,17 @@ internal object BoundedWebSocket { synchronized(callbacks) { if (ended.get()) return if (text.utf8Size() > maxMessageBytes) { - fail(IOException("WebSocket message exceeds maxMessageBytes")) + fail( + ProtocolException( + "WebSocket message exceeds maxMessageBytes" + ) + ) } else listener.onMessage(text) } } override fun onMessage(webSocket: WebSocket, bytes: ByteString) { - fail(IOException("Unexpected binary WebSocket message")) + fail(ProtocolException("Unexpected binary WebSocket message")) } override fun onClosing(webSocket: WebSocket, code: Int, reason: String) { diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index b52234c6a..0104aac5e 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -14,6 +14,7 @@ import com.openai.models.realtime.InputAudioBufferAppendEvent import com.openai.models.realtime.InputAudioBufferClearEvent import com.openai.models.realtime.RealtimeClientEvent import java.io.IOException +import java.net.ProtocolException import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.CountDownLatch @@ -25,6 +26,7 @@ import okhttp3.WebSocket import okhttp3.WebSocketListener import okhttp3.mockwebserver.MockResponse import okhttp3.mockwebserver.MockWebServer +import okio.ByteString.Companion.encodeUtf8 import org.assertj.core.api.Assertions.assertThat import org.assertj.core.api.Assertions.assertThatThrownBy import org.junit.jupiter.api.Test @@ -76,7 +78,38 @@ class RealtimeConnectionTest { fun failureBeforeAcceptedSendRemainsTerminal() = failureDuringWrite(flush = false, rejected = false) - private fun failureDuringWrite(flush: Boolean, rejected: Boolean) { + @Test + fun rejectedFlushInRecoveryCallbackKeepsTheRetry() = + failureDuringWrite(flush = true, rejected = true, fromCallback = true) + + @Test + fun callbackWriteRejectionWithoutAnotherRecoveryIsTerminal() { + MockWebServer().use { server -> + val peer = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(peer)) + server.enqueue(MockResponse().withWebSocketUpgrade(Listener())) + val rejection = WebSocketWriteNotAttempted.Busy("application busy") + OkHttpClient.builder().build().use { http -> + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .onReconnected { throw rejection } + .build() + RealtimeConnection.connect(options(server, http), settings).use { connection -> + val waiting = connection.receiveAsync() + peer.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) }.hasCause(rejection) + } + } + } + } + + private fun failureDuringWrite( + flush: Boolean, + rejected: Boolean, + fromCallback: Boolean = false, + ) { val opening = CompletableFuture() val opened = LinkedBlockingQueue() val recovered = LinkedBlockingQueue() @@ -117,18 +150,22 @@ class RealtimeConnectionTest { .reconnect(true) .reconnectDelayMillis(1) .maxPendingBytes(1024) - .onReconnected { recovered.add(it) } + .onReconnected { + if (fromCallback && attempts.get() == 2) it.flushPending() + else recovered.add(it) + } .build() RealtimeConnection.connect(configuration, settings).use { connection -> val first = opened.poll(5, TimeUnit.SECONDS)!! + val waiting = connection.receiveAsync() if (flush) { first.onFailure(IOException("initial disconnect")) assertThat(opened.poll(5, TimeUnit.SECONDS)).isNotNull() connection.send(command()) opening.complete(Unit) - assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) + if (!fromCallback) + assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) } - val waiting = connection.receiveAsync() if (!rejected) { connection.send(command()) assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } @@ -137,10 +174,11 @@ class RealtimeConnectionTest { assertThat(sent).hasSize(1) return } - assertThatThrownBy { - if (flush) connection.flushPending() else connection.send(command()) - } - .isSameAs(rejection) + if (!fromCallback) + assertThatThrownBy { + if (flush) connection.flushPending() else connection.send(command()) + } + .isSameAs(rejection) assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) assertThat(sent).isEmpty() if (flush) connection.flushPending() else connection.send(command()) @@ -154,6 +192,93 @@ class RealtimeConnectionTest { } } + @Test + fun localProtocolFailuresDoNotReconnect() { + for (binary in listOf(false, true)) { + MockWebServer().use { server -> + val peer = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(peer)) + server.enqueue(MockResponse().setResponseCode(403)) + OkHttpClient.builder().build().use { http -> + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .maxMessageBytes(64) + .build() + RealtimeConnection.connect(options(server, http), settings).use { connection -> + val waiting = connection.receiveAsync() + val socket = peer.socket.get(5, TimeUnit.SECONDS) + if (binary) socket.send("binary".encodeUtf8()) + else socket.send("x".repeat(65)) + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } + .hasRootCauseInstanceOf(ProtocolException::class.java) + .hasRootCauseMessage( + if (binary) "Unexpected binary WebSocket message" + else "WebSocket message exceeds maxMessageBytes" + ) + assertThat(server.requestCount).isEqualTo(1) + } + } + } + } + } + + @Test + fun disabledExtensionsOnReplacementDoNotReconnectAgain() { + MockWebServer().use { server -> + val peer = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(peer)) + server.enqueue( + MockResponse() + .withWebSocketUpgrade(Listener()) + .setHeader("Sec-WebSocket-Extensions", "permessage-deflate") + ) + server.enqueue(MockResponse().setResponseCode(403)) + OkHttpClient.builder().build().use { http -> + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .build() + RealtimeConnection.connect(options(server, http), settings).use { connection -> + val waiting = connection.receiveAsync() + peer.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } + .hasRootCauseInstanceOf(ProtocolException::class.java) + .hasRootCauseMessage("WebSocket extensions are disabled") + assertThat(server.requestCount).isEqualTo(2) + } + } + } + } + + @Test + fun failureDuringOpeningCompletesThePublicFuture() { + val opening = CompletableFuture() + val listening = CompletableFuture() + OkHttpClient.builder().build().use { http -> + val transport = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture { + listening.complete(listener) + return opening + } + } + val configuration = + ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() + val connection = RealtimeConnection.connectAsync(configuration) + listening.get(5, TimeUnit.SECONDS).onMessage("[]") + assertThatThrownBy { connection.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage("Invalid Realtime event") + } + } + @Test fun serializationStartedDuringRecoveryStillWaitsForExplicitFlush() = serializationWaitsForExplicitFlush(startDuringRecovery = true) diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 246ec6f0a..67594c407 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -20,6 +20,7 @@ import com.openai.errors.OpenAIIoException import com.openai.models.realtime.RealtimeClientEvent import com.openai.models.realtime.RealtimeServerEvent import java.io.IOException +import java.net.ProtocolException import java.time.Duration import java.util.ArrayDeque import java.util.concurrent.CancellationException @@ -203,8 +204,28 @@ private constructor( } if (shouldCancel) pending.cancel(true) pending.whenComplete { active, error -> - if (synchronized(lock) { generation != epoch }) { - val problem = synchronized(lock) { failure } + if (error != null) { + if (disconnect(epoch, error, canRetry(error), result, blocking)) + synchronized(lock) { failure } + ?.let { completeFailure(result, it, blocking) } + return@whenComplete + } + val (stale, problem) = + synchronized(lock) { + if (generation != epoch) true to failure + else { + opening = null + false to + if (closed || failure != null) + failure ?: CancellationException("Realtime connection closed") + else { + socket = active + recovering = false + null + } + } + } + if (stale) { try { active?.close() } catch (closing: Throwable) { @@ -214,20 +235,6 @@ private constructor( } return@whenComplete } - if (error != null && disconnect(epoch, error, canRetry(error), result, blocking)) - return@whenComplete - val problem = - synchronized(lock) { - this.opening = null - if (error != null) error - else if (this.closed || this.failure != null) - this.failure ?: CancellationException("Realtime connection closed") - else { - this.socket = active - this.recovering = false - null - } - } if (problem != null) { try { active?.close() @@ -263,7 +270,7 @@ private constructor( try { options.onReconnected?.accept(this) } catch (callbackError: Exception) { - fail(callbackError) + fail(callbackError, callbackEpoch = epoch) } } } @@ -281,7 +288,12 @@ private constructor( val seen = java.util.Collections.newSetFromMap(java.util.IdentityHashMap()) while (cause != null && seen.add(cause)) { - if (cause is SSLException || cause is CancellationException) return false + if ( + cause is SSLException || + cause is CancellationException || + cause is ProtocolException + ) + return false cause = cause.cause } val root = @@ -623,10 +635,17 @@ private constructor( } } - private fun fail(error: Throwable) { + private fun fail(error: Throwable, callbackEpoch: Long? = null) { val detached = synchronized(lock) { if (failure != null) return + if ( + callbackEpoch != null && + error is WebSocketWriteNotAttempted && + generation != callbackEpoch && + !closed + ) + return failure = error write?.disconnected = null generation++ From 0790f760f5223982edadad7294e23fdd9b0815f7 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:18:16 +0000 Subject: [PATCH 08/10] fix(realtime): classify pre-write lifecycle rejections --- README.md | 4 +- .../openai/client/okhttp/BoundedWebSocket.kt | 7 +- .../client/okhttp/RealtimeConnectionTest.kt | 114 ++++++++++++++++++ .../client/okhttp/ResponsesWebSocketTest.kt | 8 ++ .../com/openai/helpers/RealtimeConnection.kt | 71 ++++++----- 5 files changed, 167 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 31d60ffd7..f2b9727bb 100644 --- a/README.md +++ b/README.md @@ -2165,8 +2165,8 @@ Closing the connection cancels pending retries. By default, the SDK rejects sends while reconnecting. Set `maxPendingBytes` to hold unsent commands up to that size in bytes. Use `onReconnected` to restore any session settings and conversation state your application needs. Pause other -application sends while restoring state; the callback does not block them. Then call -`flushPending()` to send the waiting commands. To remove them without sending, +application sends and wait for in-flight sends to finish before restoring state; +the callback does not block them. Then call `flushPending()` to send the waiting commands. To remove them without sending, call `takePending()` instead. The SDK does not resend commands it already tried to send. A successful send does not confirm that the server received the command. diff --git a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt index 6a8c98ac3..ae9e0ed08 100644 --- a/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt +++ b/openai-java-client-okhttp/src/main/kotlin/com/openai/client/okhttp/BoundedWebSocket.kt @@ -199,9 +199,10 @@ internal object BoundedWebSocket { } val deadline = System.nanoTime() + timeout.toNanos() while (true) { - check(!ended.get() && closeDeadline.get() == null) { - "WebSocket is closed or closing" - } + if (ended.get() || closeDeadline.get() != null) + throw WebSocketWriteNotAttempted.Closed( + "WebSocket is closed or closing" + ) if (webSocket.queueSize() == 0L) return val remaining = deadline - System.nanoTime() if (remaining <= 0L) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index 0104aac5e..28b45189d 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -82,6 +82,120 @@ class RealtimeConnectionTest { fun rejectedFlushInRecoveryCallbackKeepsTheRetry() = failureDuringWrite(flush = true, rejected = true, fromCallback = true) + private enum class CallbackFailure { + FLUSH_ENTRY, + CAPACITY_WAIT, + APPLICATION, + } + + @Test + fun staleCallbackFlushEntryPreservesRecovery() = staleCallback(CallbackFailure.FLUSH_ENTRY) + + @Test + fun staleCallbackCapacityWaitPreservesRecovery() = staleCallback(CallbackFailure.CAPACITY_WAIT) + + @Test + fun staleCallbackApplicationFailureRemainsTerminal() = + staleCallback(CallbackFailure.APPLICATION) + + private fun staleCallback(mode: CallbackFailure) { + val secondOpening = CompletableFuture() + val thirdOpening = CompletableFuture() + val opened = LinkedBlockingQueue() + val sent = LinkedBlockingQueue() + val callbackFinished = CountDownLatch(1) + val recovered = CompletableFuture() + val attempts = java.util.concurrent.atomic.AtomicInteger() + val callbacks = java.util.concurrent.atomic.AtomicInteger() + val applicationError = IllegalStateException("application callback failed") + OkHttpClient.builder().build().use { http -> + val transport = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture { + val attempt = attempts.incrementAndGet() + opened.add(listener) + val active = + object : WebSocketClient.WritableConnection { + override fun awaitWritable(timeout: Duration) { + if (attempt == 2 && mode == CallbackFailure.CAPACITY_WAIT) { + listener.onFailure(IOException("replacement failed")) + throw java.util.concurrent.TimeoutException("capacity wait") + } + } + + override fun send(text: String) { + sent.add(text) + } + + override fun close() {} + } + return when (attempt) { + 2 -> secondOpening.thenApply { active } + 3 -> thirdOpening.thenApply { active } + else -> CompletableFuture.completedFuture(active) + } + } + } + val configuration = + ClientOptions.builder() + .httpClient(transport) + .apiKey("fake-key") + .timeout(com.openai.core.Timeout.builder().write(Duration.ofSeconds(1)).build()) + .build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .maxPendingBytes(1024) + .onReconnected { + if (callbacks.incrementAndGet() == 1) { + try { + val listener = opened.poll(5, TimeUnit.SECONDS)!! + if (mode != CallbackFailure.CAPACITY_WAIT) + listener.onFailure(IOException("replacement failed")) + if (mode == CallbackFailure.APPLICATION) throw applicationError + it.flushPending() + } finally { + callbackFinished.countDown() + } + } else recovered.complete(it) + } + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + val waiting = connection.receiveAsync() + opened.poll(5, TimeUnit.SECONDS)!!.onFailure(IOException("initial disconnect")) + connection.send(command()) + secondOpening.complete(Unit) + assertThat(callbackFinished.await(5, TimeUnit.SECONDS)).isTrue() + thirdOpening.complete(Unit) + if (mode == CallbackFailure.APPLICATION) { + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } + .hasCause(applicationError) + assertThat(sent).isEmpty() + } else { + assertThat(recovered.get(5, TimeUnit.SECONDS)).isSameAs(connection) + assertThat(sent).isEmpty() + connection.flushPending() + assertThat(sent).hasSize(1) + opened + .poll(5, TimeUnit.SECONDS)!! + .onMessage( + """{"type":"input_audio_buffer.cleared","event_id":"restored"}""" + ) + assertThat( + waiting.get(5, TimeUnit.SECONDS).asInputAudioBufferCleared().eventId() + ) + .isEqualTo("restored") + } + } + } + } + @Test fun callbackWriteRejectionWithoutAnotherRecoveryIsTerminal() { MockWebServer().use { server -> diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt index c4f84fc6c..7ec7da6d7 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/ResponsesWebSocketTest.kt @@ -1555,6 +1555,14 @@ class ResponsesWebSocketTest { assertThatThrownBy { connection.send(command) } .isInstanceOf(IllegalStateException::class.java) .isInstanceOf(WebSocketWriteNotAttempted::class.java) + assertThatThrownBy { + (connection + as + com.openai.core.http.WebSocketClient.WritableConnection) + .awaitWritable(java.time.Duration.ofSeconds(1)) + } + .isInstanceOf(WebSocketWriteNotAttempted.Closed::class.java) + .hasMessage("WebSocket is closed or closing") } peer.await() } diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 67594c407..4dc1018d2 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -410,13 +410,16 @@ private constructor( fun send(event: RealtimeClientEvent) { val (epoch, admittedDuringRecovery) = synchronized(lock) { - check(!closed && failure == null && (socket != null || recovering)) { - "Realtime is not connected" - } - check(!flushing) { "A Realtime flush is already pending; command was not sent" } - check(!sending && write == null) { - "A Realtime send is already pending; command was not sent" - } + if (closed || failure != null || (socket == null && !recovering)) + throw WebSocketWriteNotAttempted.Closed("Realtime is not connected") + if (flushing) + throw WebSocketWriteNotAttempted.Busy( + "A Realtime flush is already pending; command was not sent" + ) + if (sending || write != null) + throw WebSocketWriteNotAttempted.Busy( + "A Realtime send is already pending; command was not sent" + ) sending = true generation to recovering } @@ -427,7 +430,8 @@ private constructor( require(size <= options.maxMessageBytes) { "Realtime command exceeds maxMessageBytes" } val active = synchronized(lock) { - check(!closed && failure == null) { "Realtime is not connected" } + if (closed || failure != null) + throw WebSocketWriteNotAttempted.Closed("Realtime is not connected") if (admittedDuringRecovery || generation != epoch) { if (size > options.maxPendingBytes - pendingBytes) throw WebSocketWriteNotAttempted.Busy( @@ -437,7 +441,9 @@ private constructor( pendingBytes += size return } - val active = socket ?: throw IllegalStateException("Realtime is not connected") + val active = + socket + ?: throw WebSocketWriteNotAttempted.Closed("Realtime is not connected") write = Write() active } @@ -466,17 +472,9 @@ private constructor( /** Send retained commands in order after the application restores session state. */ fun flushPending() { synchronized(lock) { - check( - !flushing && - !sending && - write == null && - !recovering && - !closed && - failure == null && - socket != null - ) { - "Realtime cannot flush now" - } + requireActiveSocket(message = "Realtime cannot flush now") + if (flushing || sending || write != null) + throw WebSocketWriteNotAttempted.Busy("Realtime cannot flush now") flushing = true } var notAttempted = false @@ -484,17 +482,13 @@ private constructor( while (true) { val next = synchronized(lock) { - check(!closed && failure == null && !recovering) { - "Realtime is not connected" - } + val active = requireActiveSocket() if (pending.isEmpty()) return - Pair(socket!!, pending.first) + Pair(active, pending.first) } awaitWritable(next.first) synchronized(lock) { - check(!closed && failure == null && !recovering && socket === next.first) { - "Realtime is not connected" - } + requireActiveSocket(next.first) pending.removeFirst() pendingBytes -= utf8Size(next.second) write = Write() @@ -536,6 +530,23 @@ private constructor( } } + // Caller holds lock. + private fun requireActiveSocket( + expected: WebSocketClient.Connection? = null, + message: String = "Realtime is not connected", + ): WebSocketClient.Connection { + val active = socket + if ( + closed || + failure != null || + recovering || + active == null || + (expected != null && active !== expected) + ) + throw WebSocketWriteNotAttempted.Closed(message) + return active + } + private fun awaitWritable(active: WebSocketClient.Connection) { if (active !is WebSocketClient.WritableConnection) return val timeout = @@ -548,13 +559,9 @@ private constructor( Thread.currentThread().interrupt() throw OpenAIIoException("Realtime WebSocket operation interrupted", error) } catch (error: TimeoutException) { + synchronized(lock) { requireActiveSocket(active) } if (!timeout.isZero) throw OpenAIIoException("Realtime WebSocket operation failed", error) - synchronized(lock) { - check(!closed && failure == null && !recovering && socket === active) { - "Realtime is not connected" - } - } } } } From ce59fed4b00223c3b0576b8238859d9dc686200b Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 06:39:57 +0000 Subject: [PATCH 09/10] fix(realtime): cancel recovery with client lifecycle --- .../client/okhttp/RealtimeConnectionTest.kt | 142 ++++++++++++++++++ .../kotlin/com/openai/core/ClientOptions.kt | 21 +-- .../com/openai/helpers/RealtimeConnection.kt | 20 ++- 3 files changed, 169 insertions(+), 14 deletions(-) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index 28b45189d..78cc3a2de 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -8,6 +8,7 @@ import com.openai.core.http.HttpResponse import com.openai.core.http.WebSocketClient import com.openai.core.http.WebSocketWriteNotAttempted import com.openai.core.jsonMapper +import com.openai.errors.OpenAIIoException import com.openai.helpers.RealtimeConnection import com.openai.helpers.RealtimeWebSocketOptions import com.openai.models.realtime.InputAudioBufferAppendEvent @@ -17,6 +18,7 @@ import java.io.IOException import java.net.ProtocolException import java.time.Duration import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionException import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.LinkedBlockingQueue @@ -730,6 +732,146 @@ class RealtimeConnectionTest { } } + @Test + fun closingOwnerDuringBackoffSettlesTheLogicalConnection() { + val openedListener = CompletableFuture() + val opens = java.util.concurrent.atomic.AtomicInteger() + val retired = java.util.concurrent.atomic.AtomicInteger() + OkHttpClient.builder().build().use { http -> + val transport = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture { + opens.incrementAndGet() + openedListener.complete(listener) + return CompletableFuture.completedFuture( + object : WebSocketClient.Connection { + override fun send(text: String) = error("no command expected") + + override fun close() { + retired.incrementAndGet() + } + } + ) + } + } + val configuration = + ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(8000) + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + val pending = connection.receiveAsync() + openedListener + .get(5, TimeUnit.SECONDS) + .onFailure(IOException("temporary disconnect")) + assertThat(retired.get()).isEqualTo(1) + configuration.close() + assertThatThrownBy { pending.get(2, TimeUnit.SECONDS) } + .isInstanceOf(java.util.concurrent.CancellationException::class.java) + .hasMessage("Client is closed") + assertThatThrownBy { + RealtimeConnection.connectAsync(configuration, settings) + .get(2, TimeUnit.SECONDS) + } + .isInstanceOf(java.util.concurrent.CancellationException::class.java) + .hasMessage("Client is closed") + assertThat(opens.get()).isEqualTo(1) + } + } + } + + @Test + fun wrappedCredentialTransportIoRetries() = + credentialRefreshFailure( + CompletionException( + OpenAIIoException("token exchange failed", IOException("transient read")) + ), + retryable = true, + ) + + @Test + fun credentialRefreshExclusionsRemainTerminal() { + for (error in + listOf( + OpenAIIoException( + "token exchange failed", + javax.net.ssl.SSLException("TLS failure"), + ), + OpenAIIoException("token exchange failed", ProtocolException("protocol failure")), + OpenAIIoException( + "token exchange failed", + java.util.concurrent.CancellationException("credential canceled"), + ), + IllegalStateException("application failure", IOException("nested I/O")), + )) credentialRefreshFailure(error, retryable = false) + } + + private fun credentialRefreshFailure(error: RuntimeException, retryable: Boolean) { + MockWebServer().use { server -> + val first = Listener() + val second = Listener() + server.enqueue(MockResponse().withWebSocketUpgrade(first)) + server.enqueue(MockResponse().withWebSocketUpgrade(second)) + val calls = java.util.concurrent.atomic.AtomicInteger() + OkHttpClient.builder().build().use { http -> + val configuration = + options(server, http) + .toBuilder() + .credential( + com.openai.credential.BearerTokenCredential.create { + val call = calls.incrementAndGet() + if (call == 2) throw error + "fake-recovery-$call" + } + ) + .build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .build() + RealtimeConnection.connect(configuration, settings).use { connection -> + val pending = connection.receiveAsync() + first.socket.get(5, TimeUnit.SECONDS).close(1012, "restart") + if (retryable) { + second.socket + .get(5, TimeUnit.SECONDS) + .send( + """{"type":"input_audio_buffer.cleared","event_id":"refreshed"}""" + ) + assertThat( + pending + .get(5, TimeUnit.SECONDS) + .asInputAudioBufferCleared() + .eventId() + ) + .isEqualTo("refreshed") + assertThat( + server.takeRequest(5, TimeUnit.SECONDS)!!.getHeader("Authorization") + ) + .isEqualTo("Bearer fake-recovery-1") + assertThat( + server.takeRequest(5, TimeUnit.SECONDS)!!.getHeader("Authorization") + ) + .isEqualTo("Bearer fake-recovery-3") + } else { + assertThatThrownBy { pending.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage(error.cause!!.message) + } + assertThat(calls.get()).isEqualTo(if (retryable) 3 else 2) + assertThat(server.requestCount).isEqualTo(if (retryable) 2 else 1) + } + } + } + } + @Test fun recoveryDoesNotRetryAuthenticationRejection() { MockWebServer().use { server -> diff --git a/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt b/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt index 3b14e6959..38f39af6c 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt @@ -155,13 +155,11 @@ private constructor( } } - // Derived client options share the transport owner, including openings awaiting authentication. + // Derived client options share the transport owner, including openings and recovering sessions. private class WebSocketLifecycle { val pending = java.util.concurrent.ConcurrentHashMap.newKeySet< - java.util.concurrent.CompletableFuture< - com.openai.core.http.WebSocketClient.Connection - > + java.util.concurrent.CompletableFuture<*> >() val closed = java.util.concurrent.atomic.AtomicBoolean() } @@ -192,6 +190,13 @@ private constructor( @JvmSynthetic internal fun isWebSocketClosed(): Boolean = webSocketLifecycle.closed.get() + @JvmSynthetic + internal fun trackWebSocketLifetime(lifetime: java.util.concurrent.CompletableFuture<*>) { + webSocketLifecycle.pending.add(lifetime) + lifetime.whenComplete { _, _ -> webSocketLifecycle.pending.remove(lifetime) } + if (webSocketLifecycle.closed.get()) lifetime.cancel(true) + } + @JvmSynthetic internal fun connectWebSocket( request: com.openai.core.http.HttpRequest, @@ -205,12 +210,8 @@ private constructor( java.util.concurrent.CompletableFuture< com.openai.core.http.WebSocketClient.Connection >() - webSocketLifecycle.pending.add(result) - result.whenComplete { _, _ -> webSocketLifecycle.pending.remove(result) } - if (webSocketLifecycle.closed.get()) { - result.cancel(true) - return result - } + trackWebSocketLifetime(result) + if (result.isCancelled) return result val authenticated = try { when { diff --git a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt index 4dc1018d2..55c555401 100644 --- a/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt +++ b/openai-java-core/src/main/kotlin/com/openai/helpers/RealtimeConnection.kt @@ -25,6 +25,7 @@ import java.time.Duration import java.util.ArrayDeque import java.util.concurrent.CancellationException import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionException import java.util.concurrent.ScheduledFuture import java.util.concurrent.ScheduledThreadPoolExecutor import java.util.concurrent.TimeUnit @@ -79,6 +80,7 @@ private constructor( private val pending = ArrayDeque() private var pendingBytes = 0L private var flushing = false + private val ownerLifetime = if (options.reconnect) CompletableFuture() else null companion object { private val recoveryTimer = @@ -115,6 +117,16 @@ private constructor( val connection = RealtimeConnection(clientOptions, options, requestOptions) val result = CompletableFuture() result.whenComplete { _, _ -> if (result.isCancelled) connection.close() } + connection.ownerLifetime?.let { lifetime -> + lifetime.whenComplete { _, _ -> + if (lifetime.isCancelled) { + val error = CancellationException("Client is closed") + connection.fail(error) + if (!result.isDone) connection.completeFailure(result, error, blocking) + } + } + clientOptions.trackWebSocketLifetime(lifetime) + } connection.open(result, blocking) return result } @@ -285,6 +297,7 @@ private constructor( private fun canRetry(error: Throwable): Boolean { var cause: Throwable? = error + var root = error val seen = java.util.Collections.newSetFromMap(java.util.IdentityHashMap()) while (cause != null && seen.add(cause)) { @@ -294,12 +307,10 @@ private constructor( cause is ProtocolException ) return false + if (cause === root && (cause is CompletionException || cause is OpenAIIoException)) + root = cause.cause ?: cause cause = cause.cause } - val root = - if (error is java.util.concurrent.CompletionException && error.cause != null) - error.cause!! - else error return root is IOException || (root is WebSocketHandshakeException && root.statusCode in setOf(408, 429, 500, 502, 503, 504)) @@ -666,6 +677,7 @@ private constructor( opening = null } } + ownerLifetime?.complete(null) detached.second?.cancel(true) // Complete the detached read before cleanup: injected close() may depend on the reader // having received its failure. Only private blocking futures complete on this thread. From 574268752cd152a7f584d62bb1ea4553f8c65857 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Sun, 18 Oct 2026 07:21:18 +0000 Subject: [PATCH 10/10] test(realtime): reuse fake WebSocket transport setup --- .../client/okhttp/RealtimeConnectionTest.kt | 135 ++++++++---------- 1 file changed, 60 insertions(+), 75 deletions(-) diff --git a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt index 78cc3a2de..96ea7d75f 100644 --- a/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt +++ b/openai-java-client-okhttp/src/test/kotlin/com/openai/client/okhttp/RealtimeConnectionTest.kt @@ -70,6 +70,19 @@ class RealtimeConnectionTest { .baseUrl(server.url("/v1?inherited=1").toString()) .build() + private fun fakeWebSocketTransport( + http: HttpClient, + open: (WebSocketClient.Listener) -> CompletableFuture, + ): HttpClient = + object : HttpClient by http, WebSocketClient { + override fun connectWebSocket( + request: HttpRequest, + options: RequestOptions, + maxMessageBytes: Int, + listener: WebSocketClient.Listener, + ): CompletableFuture = open(listener) + } + @Test fun failureBeforeSendRejectionRecovers() = failureDuringWrite(flush = false, rejected = true) @@ -112,35 +125,28 @@ class RealtimeConnectionTest { val applicationError = IllegalStateException("application callback failed") OkHttpClient.builder().build().use { http -> val transport = - object : HttpClient by http, WebSocketClient { - override fun connectWebSocket( - request: HttpRequest, - options: RequestOptions, - maxMessageBytes: Int, - listener: WebSocketClient.Listener, - ): CompletableFuture { - val attempt = attempts.incrementAndGet() - opened.add(listener) - val active = - object : WebSocketClient.WritableConnection { - override fun awaitWritable(timeout: Duration) { - if (attempt == 2 && mode == CallbackFailure.CAPACITY_WAIT) { - listener.onFailure(IOException("replacement failed")) - throw java.util.concurrent.TimeoutException("capacity wait") - } - } - - override fun send(text: String) { - sent.add(text) + fakeWebSocketTransport(http) { listener -> + val attempt = attempts.incrementAndGet() + opened.add(listener) + val active = + object : WebSocketClient.WritableConnection { + override fun awaitWritable(timeout: Duration) { + if (attempt == 2 && mode == CallbackFailure.CAPACITY_WAIT) { + listener.onFailure(IOException("replacement failed")) + throw java.util.concurrent.TimeoutException("capacity wait") } + } - override fun close() {} + override fun send(text: String) { + sent.add(text) } - return when (attempt) { - 2 -> secondOpening.thenApply { active } - 3 -> thirdOpening.thenApply { active } - else -> CompletableFuture.completedFuture(active) + + override fun close() {} } + when (attempt) { + 2 -> secondOpening.thenApply { active } + 3 -> thirdOpening.thenApply { active } + else -> CompletableFuture.completedFuture(active) } } val configuration = @@ -234,30 +240,23 @@ class RealtimeConnectionTest { val attempts = java.util.concurrent.atomic.AtomicInteger() OkHttpClient.builder().build().use { http -> val transport = - object : HttpClient by http, WebSocketClient { - override fun connectWebSocket( - request: HttpRequest, - options: RequestOptions, - maxMessageBytes: Int, - listener: WebSocketClient.Listener, - ): CompletableFuture { - val attempt = attempts.incrementAndGet() - opened.add(listener) - val connection = - object : WebSocketClient.Connection { - override fun send(text: String) { - if (attempt == if (flush) 2 else 1) { - listener.onFailure(IOException("peer reset")) - if (rejected) throw rejection - } - sent.add(text) + fakeWebSocketTransport(http) { listener -> + val attempt = attempts.incrementAndGet() + opened.add(listener) + val connection = + object : WebSocketClient.Connection { + override fun send(text: String) { + if (attempt == if (flush) 2 else 1) { + listener.onFailure(IOException("peer reset")) + if (rejected) throw rejection } - - override fun close() {} + sent.add(text) } - if (flush && attempt == 2) return opening.thenApply { connection } - return CompletableFuture.completedFuture(connection) - } + + override fun close() {} + } + if (flush && attempt == 2) opening.thenApply { connection } + else CompletableFuture.completedFuture(connection) } val configuration = ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() @@ -375,16 +374,9 @@ class RealtimeConnectionTest { val listening = CompletableFuture() OkHttpClient.builder().build().use { http -> val transport = - object : HttpClient by http, WebSocketClient { - override fun connectWebSocket( - request: HttpRequest, - options: RequestOptions, - maxMessageBytes: Int, - listener: WebSocketClient.Listener, - ): CompletableFuture { - listening.complete(listener) - return opening - } + fakeWebSocketTransport(http) { listener -> + listening.complete(listener) + opening } val configuration = ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() @@ -739,25 +731,18 @@ class RealtimeConnectionTest { val retired = java.util.concurrent.atomic.AtomicInteger() OkHttpClient.builder().build().use { http -> val transport = - object : HttpClient by http, WebSocketClient { - override fun connectWebSocket( - request: HttpRequest, - options: RequestOptions, - maxMessageBytes: Int, - listener: WebSocketClient.Listener, - ): CompletableFuture { - opens.incrementAndGet() - openedListener.complete(listener) - return CompletableFuture.completedFuture( - object : WebSocketClient.Connection { - override fun send(text: String) = error("no command expected") - - override fun close() { - retired.incrementAndGet() - } + fakeWebSocketTransport(http) { listener -> + opens.incrementAndGet() + openedListener.complete(listener) + CompletableFuture.completedFuture( + object : WebSocketClient.Connection { + override fun send(text: String) = error("no command expected") + + override fun close() { + retired.incrementAndGet() } - ) - } + } + ) } val configuration = ClientOptions.builder().httpClient(transport).apiKey("fake-key").build()