diff --git a/README.md b/README.md index d79713ce8..f2b9727bb 100644 --- a/README.md +++ b/README.md @@ -2149,6 +2149,27 @@ 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. Pause other +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. + ## Contributing Please share bug reports and feature requests through [GitHub issues](https://github.com/openai/openai-java/issues). 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..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 @@ -3,7 +3,9 @@ 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.net.ProtocolException import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.concurrent.TimeUnit @@ -108,7 +110,7 @@ internal object BoundedWebSocket { ) { chain.call().cancel() response.close() - throw IOException("WebSocket extensions are disabled") + throw ProtocolException("WebSocket extensions are disabled") } response } @@ -197,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) @@ -214,32 +217,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" + ) } } @@ -258,13 +260,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 1032b48a1..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 @@ -6,13 +6,19 @@ 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.errors.OpenAIIoException 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.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 @@ -22,6 +28,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 @@ -63,6 +70,818 @@ 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) + + @Test + fun failureBeforeFlushRejectionRecovers() = failureDuringWrite(flush = true, rejected = true) + + @Test + fun failureBeforeAcceptedSendRemainsTerminal() = + failureDuringWrite(flush = false, rejected = false) + + @Test + 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 = + 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 send(text: String) { + sent.add(text) + } + + override fun close() {} + } + 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 -> + 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() + val sent = LinkedBlockingQueue() + val rejection = WebSocketWriteNotAttempted.Closed("closed before enqueue") + val attempts = java.util.concurrent.atomic.AtomicInteger() + OkHttpClient.builder().build().use { http -> + val transport = + 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 + } + sent.add(text) + } + + override fun close() {} + } + if (flush && attempt == 2) opening.thenApply { connection } + else CompletableFuture.completedFuture(connection) + } + val configuration = + ClientOptions.builder().httpClient(transport).apiKey("fake-key").build() + val settings = + RealtimeWebSocketOptions.builder() + .reconnect(true) + .reconnectDelayMillis(1) + .maxPendingBytes(1024) + .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) + if (!fromCallback) + assertThat(recovered.poll(5, TimeUnit.SECONDS)).isSameAs(connection) + } + if (!rejected) { + connection.send(command()) + assertThatThrownBy { waiting.get(5, TimeUnit.SECONDS) } + .hasRootCauseMessage("peer reset") + assertThat(attempts.get()).isEqualTo(1) + assertThat(sent).hasSize(1) + return + } + 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()) + 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 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 = + fakeWebSocketTransport(http) { listener -> + listening.complete(listener) + 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) + + @Test + fun serializationStartedBeforeRecoveryStillWaitsForExplicitFlush() = + serializationWaitsForExplicitFlush(startDuringRecovery = false) + + private fun serializationWaitsForExplicitFlush(startDuringRecovery: Boolean) { + 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 -> + 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( + RealtimeClientEvent.ofInputAudioBufferAppend( + InputAudioBufferAppendEvent.builder() + .audio("pending-audio") + .build() + ) + ) + } + 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() + 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 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() + } + } + + @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() + 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, observedHttp) + .toBuilder() + .timeout(com.openai.core.Timeout.builder().write(writeTimeout).build()) + .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 requestOptions = + RequestOptions.builder() + .timeout( + com.openai.core.Timeout.builder() + .connect(Duration.ofSeconds(2)) + .build() + ) + .build() + val connection = + if (async) + RealtimeConnection.connectAsync(configuration, settings, requestOptions) + .get(5, TimeUnit.SECONDS) + else RealtimeConnection.connect(configuration, settings, requestOptions) + 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()) + 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(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") + 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 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 = + 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() + 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 -> + 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-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..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 @@ -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,18 @@ class ResponsesWebSocketTest { .asText() ) .isEqualTo("response.completed") + connection.close() + 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/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/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 8239cf37f..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 @@ -9,18 +9,28 @@ 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 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.net.ProtocolException +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 +import java.util.concurrent.TimeoutException +import javax.net.ssl.SSLException /** * Managed GA Realtime connection. Consume typed events synchronously or asynchronously, with one @@ -49,6 +59,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 @@ -57,8 +72,23 @@ private constructor( private var failure: Throwable? = null private var closed = false private var sending = false + private var write: Write? = null + 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 + private val ownerLifetime = if (options.reconnect) CompletableFuture() else null companion object { + private val recoveryTimer = + ScheduledThreadPoolExecutor(1) { task -> + Thread(task, "openai-realtime-recovery").apply { isDaemon = true } + } + .apply { removeOnCancelPolicy = true } + @JvmStatic @JvmOverloads @MustBeClosed @@ -87,109 +117,272 @@ private constructor( val connection = RealtimeConnection(clientOptions, options, requestOptions) val result = CompletableFuture() result.whenComplete { _, _ -> if (result.isCancelled) connection.close() } - try { - clientOptions.requireWebSocketTransport() + 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 + } + + 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 (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 { - connection.opening = pending - false + opening = null + false to + if (closed || failure != null) + failure ?: CancellationException("Realtime connection closed") + else { + socket = active + 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 (stale) { + try { + active?.close() + } catch (closing: Throwable) { + if (problem != null && closing !== problem) problem.addSuppressed(closing) + } finally { + problem?.let { completeFailure(result, it, blocking) } + } + return@whenComplete + } + 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, callbackEpoch = epoch) } - 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) } - return result + } catch (error: Exception) { + disconnect(epoch, error, canRetry(error), result, blocking) } + } - private fun await(future: CompletableFuture): T = - awaitWebSocketFuture(future, "Realtime") + 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)) { + if ( + cause is SSLException || + cause is CancellationException || + cause is ProtocolException + ) + return false + if (cause === root && (cause is CompletionException || cause is OpenAIIoException)) + root = cause.cause ?: cause + cause = cause.cause + } + 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) && + 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++ + val old = Pair(socket, opening) + socket = null + opening = null + val delay = + (options.reconnectDelayMillis * (1L shl minOf(attempts - 1, 14))) + .coerceAtMost(8000) + retryTimer = + recoveryTimer.schedule( + { CompletableFuture.runAsync { 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,29 +419,161 @@ 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 (epoch, admittedDuringRecovery) = + synchronized(lock) { + 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 + } + var notAttempted = false try { val text = clientOptions.jsonMapper.writeValueAsString(event) - require(utf8Size(text) <= options.maxMessageBytes) { - "Realtime command exceeds 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" } - socket ?: throw IllegalStateException("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( + "Realtime recovery queue is full; command was not sent" + ) + pending.addLast(text) + pendingBytes += size + return + } + val active = + socket + ?: throw WebSocketWriteNotAttempted.Closed("Realtime is not connected") + 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) { sending = false } + finishWrite(notAttempted) { sending = false } + } + } + + /** 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) { + requireActiveSocket(message = "Realtime cannot flush now") + if (flushing || sending || write != null) + throw WebSocketWriteNotAttempted.Busy("Realtime cannot flush now") + flushing = true + } + var notAttempted = false + try { + while (true) { + val next = + synchronized(lock) { + val active = requireActiveSocket() + if (pending.isEmpty()) return + Pair(active, pending.first) + } + awaitWritable(next.first) + synchronized(lock) { + requireActiveSocket(next.first) + pending.removeFirst() + pendingBytes -= utf8Size(next.second) + write = Write() + } + try { + next.first.send(next.second) + } catch (error: Throwable) { + if (error is WebSocketWriteNotAttempted) + synchronized(lock) { + pending.addFirst(next.second) + pendingBytes += utf8Size(next.second) + notAttempted = true + } + else fail(error) + throw error + } + finishWrite() + } + } finally { + 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 } + } + } + + // 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 = + (requestOptions.timeout?.assign(clientOptions.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) { + synchronized(lock) { requireActiveSocket(active) } + if (!timeout.isZero) + throw OpenAIIoException("Realtime WebSocket operation failed", error) + } } } @@ -279,7 +604,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 +631,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,15 +649,26 @@ 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) } } - 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++ + retryTimer?.cancel(false) + retryTimer = null val unassigned = waiter?.takeUnless { it.claimed || it.cancellationReserved || it.isDone } waiter = null @@ -340,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. 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(), )