Client sockets: Rejected SETUP failures are not raised to client code properly and sometimes hundreds of attempts are made

## Expected Behavior
I am sending auth data in the SETUP frame metadata. I have tried this using both a reconnectable socket as well as a regular one.

The auth is rejected by the server. I would expect that the client on the call to create the `RSocket` i.e. `client.rSocket(url)` will raise the exception which allows me to see that the setup was rejected, e.g. `RSocketError.Setup.Rejected`.

When using the `rsc` tool, it can be seen that the server is returning this data:

```
2025-05-06 08:26:05.616 DEBUG 3033455 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : receiving -> 
Frame => Stream ID: 0 Type: ERROR Flags: 0b0 Length: 45
Data:

RejectedSetupException (0x3): Jwt expired at 2025-05-05T16:28:40Z
        at io.rsocket.exceptions.Exceptions.from(Exceptions.java:62)
        at io.rsocket.core.RSocketRequester.lambda$tryTerminateOnZeroError$4(RSocketRequester.java:313)
…
```

## Actual Behavior
The call to `client.rSocket` completes without any error. In some cases (though I have not been able to reproduce this consistently) hundreds of setup requests are made to the server. Subsequent calls do fail, but not in a useful way. There is a race condition. In most cases (especially when not using the debugger), the following error is received upon the first request to the `RSocket`:

```
kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed
	at kotlinx.coroutines.channels.BufferedChannel.getSendException(BufferedChannel.kt:1757)
	at kotlinx.coroutines.channels.BufferedChannel.onClosedSend(BufferedChannel.kt:141)
	at kotlinx.coroutines.channels.BufferedChannel.send$suspendImpl(BufferedChannel.kt:126)
	at kotlinx.coroutines.channels.BufferedChannel.send(BufferedChannel.kt)
	at io.rsocket.kotlin.transport.internal.PrioritizationFrameQueue.enqueueFrame(PrioritizationFrameQueue.kt:45)
	at io.rsocket.kotlin.transport.ktor.websocket.internal.KtorWebSocketConnection.sendFrame(KtorWebSocketConnection.kt:67)
	at io.rsocket.kotlin.connection.SequentialConnection$Outbound.sendFrame(SequentialConnection.kt:142)
	at io.rsocket.kotlin.operation.OperationOutbound.sendFrame(OperationOutbound.kt:38)
	at io.rsocket.kotlin.operation.OperationOutbound.sendCancel(OperationOutbound.kt:45)
	at io.rsocket.kotlin.operation.RequesterRequestResponseOperation$execute$2.invokeSuspend(RequesterRequestResponseOperation.kt:38)
	at io.rsocket.kotlin.operation.RequesterRequestResponseOperation$execute$2.invoke(RequesterRequestResponseOperation.kt)
	at io.rsocket.kotlin.operation.RequesterRequestResponseOperation$execute$2.invoke(RequesterRequestResponseOperation.kt)
	at kotlinx.coroutines.intrinsics.UndispatchedKt.startUndispatchedOrReturn(Undispatched.kt:61)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.withContext(Builders.common.kt:163)
	at kotlinx.coroutines.BuildersKt.withContext(Unknown Source)
	at io.rsocket.kotlin.internal.io.ContextKt.nonCancellable(Context.kt:27)
	at io.rsocket.kotlin.operation.RequesterRequestResponseOperation.execute(RequesterRequestResponseOperation.kt:38)
	at io.rsocket.kotlin.operation.RequesterRequestResponseOperation$execute$1.invokeSuspend(RequesterRequestResponseOperation.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl$CoroutineOwner.<init>(DebugProbesImpl.kt:531)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.createOwner(DebugProbesImpl.kt:510)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.probeCoroutineCreated$kotlinx_coroutines_core(DebugProbesImpl.kt:497)
	at kotlin.coroutines.jvm.internal.DebugProbesKt.probeCoroutineCreated(DebugProbes.kt:7)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:161)
	at kotlin.coroutines.ContinuationKt.startCoroutine(Continuation.kt:129)
	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:89)
	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:123)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:52)
	at kotlinx.coroutines.BuildersKt.launch(Unknown Source)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch$default(Builders.common.kt:43)
	at kotlinx.coroutines.BuildersKt.launch$default(Unknown Source)
	at io.rsocket.kotlin.connection.SequentialConnection.launchRequest(SequentialConnection.kt:64)
	at io.rsocket.kotlin.connection.RequesterRSocket.requestResponse(RequesterRSocket.kt:58)
	at io.rsocket.kotlin.core.ReconnectableRSocket.requestResponse(ReconnectableRSocket.kt:110)
	at mycode.cli.RsocketClientKt.main(RsocketClient.kt:99)
	at mycode.cli.RsocketClientKt$main$1.invokeSuspend(RsocketClient.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at kotlinx.coroutines.DispatchedTaskKt.resume(DispatchedTask.kt:175)
	at kotlinx.coroutines.DispatchedTaskKt.dispatch(DispatchedTask.kt:164)
	at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:466)
	at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl(CancellableContinuationImpl.kt:500)
	at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$default(CancellableContinuationImpl.kt:489)
	at kotlinx.coroutines.CancellableContinuationImpl.resumeWith(CancellableContinuationImpl.kt:364)
	at kotlinx.coroutines.flow.StateFlowSlot.makePending(StateFlow.kt:286)
	at kotlinx.coroutines.flow.StateFlowImpl.updateState(StateFlow.kt:349)
	at kotlinx.coroutines.flow.StateFlowImpl.setValue(StateFlow.kt:316)
	at kotlinx.coroutines.flow.StateFlowImpl.emit(StateFlow.kt:373)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:113)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:82)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:113)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:82)
	at io.rsocket.kotlin.core.ReconnectableRSocketKt$connectWithReconnect$$inlined$transform$1$1.emit(Emitters.kt:39)
	at kotlinx.coroutines.flow.FlowKt__ErrorsKt$catchImpl$2.emit(Errors.kt:154)
	at kotlinx.coroutines.flow.FlowKt__ErrorsKt$catchImpl$2.emit(Errors.kt:154)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollectorKt$emitFun$1.invoke(SafeCollector.kt:11)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:113)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:82)
	at io.rsocket.kotlin.core.ReconnectableRSocketKt$connectWithReconnect$state$1.invokeSuspend(ReconnectableRSocket.kt:41)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:104)
	at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:111)
	at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:99)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:811)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:715)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:702)

```

But if a `delay` is inserted after the call to `client.rSocket` and before the call to the `rSocket`, then the interaction gets this exception, which does actually show the desired error as a cause of a `JobCancellationException`:

```
kotlinx.coroutines.JobCancellationException: Parent job is Cancelling; job=JobImpl{Cancelling}@31198e3
Caused by: io.rsocket.kotlin.RSocketError$Setup$Rejected: Jwt expired at 2025-05-05T16:28:40Z
	at io.rsocket.kotlin.RSocketErrorKt.RSocketError(RSocketError.kt:61)
	at io.rsocket.kotlin.frame.ErrorFrameKt.readError(ErrorFrame.kt:47)
	at io.rsocket.kotlin.frame.FrameKt.readFrame(Frame.kt:70)
	at io.rsocket.kotlin.frame.FrameCodec.decodeFrame(FrameCodec.kt:25)
	at io.rsocket.kotlin.connection.SequentialConnection.handleConnection(SequentialConnection.kt:50)
	at io.rsocket.kotlin.connection.SequentialConnection$handleConnection$1.invokeSuspend(SequentialConnection.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl$CoroutineOwner.<init>(DebugProbesImpl.kt:531)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.createOwner(DebugProbesImpl.kt:510)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.probeCoroutineCreated$kotlinx_coroutines_core(DebugProbesImpl.kt:497)
	at kotlin.coroutines.jvm.internal.DebugProbesKt.probeCoroutineCreated(DebugProbes.kt:7)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:161)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:26)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable$default(Cancellable.kt:21)
	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:88)
	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:123)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:52)
	at kotlinx.coroutines.BuildersKt.launch(Unknown Source)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch$default(Builders.common.kt:43)
	at kotlinx.coroutines.BuildersKt.launch$default(Unknown Source)
	at io.rsocket.kotlin.connection.ConnectionInitializer.initialize(ConnectionInitializer.kt:98)
	at io.rsocket.kotlin.connection.ConnectionInitializer.access$initialize(ConnectionInitializer.kt:27)
	at io.rsocket.kotlin.connection.ConnectionInitializer$asyncInitializer$1.invokeSuspend(ConnectionInitializer.kt:121)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl$CoroutineOwner.<init>(DebugProbesImpl.kt:531)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.createOwner(DebugProbesImpl.kt:510)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.probeCoroutineCreated$kotlinx_coroutines_core(DebugProbesImpl.kt:497)
	at kotlin.coroutines.jvm.internal.DebugProbesKt.probeCoroutineCreated(DebugProbes.kt:7)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:161)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:26)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable$default(Cancellable.kt:21)
	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:88)
	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:123)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.async(Builders.common.kt:87)
	at kotlinx.coroutines.BuildersKt.async(Unknown Source)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.async$default(Builders.common.kt:78)
	at kotlinx.coroutines.BuildersKt.async$default(Unknown Source)
	at io.rsocket.kotlin.connection.ConnectionInitializer.asyncInitializer(ConnectionInitializer.kt:119)
	at io.rsocket.kotlin.connection.ConnectionInitializer.runInitializer(ConnectionInitializer.kt:129)
	at io.rsocket.kotlin.core.RSocketConnector.connectOnce(RSocketConnector.kt:61)
	at io.rsocket.kotlin.core.RSocketConnector.access$connectOnce(RSocketConnector.kt:27)
	at io.rsocket.kotlin.core.RSocketConnector$connectOnce$1.invokeSuspend(RSocketConnector.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl$CoroutineOwner.<init>(DebugProbesImpl.kt:531)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.createOwner(DebugProbesImpl.kt:510)
	at kotlinx.coroutines.debug.internal.DebugProbesImpl.probeCoroutineCreated$kotlinx_coroutines_core(DebugProbesImpl.kt:497)
	at kotlin.coroutines.jvm.internal.DebugProbesKt.probeCoroutineCreated(DebugProbes.kt:7)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:161)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:26)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable$default(Cancellable.kt:21)
	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:88)
	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:123)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:52)
	at kotlinx.coroutines.BuildersKt.launch(Unknown Source)
	at kotlinx.coroutines.flow.FlowKt__ShareKt.launchSharing$FlowKt__ShareKt(Share.kt:205)
	at kotlinx.coroutines.flow.FlowKt__ShareKt.stateIn(Share.kt:310)
	at kotlinx.coroutines.flow.FlowKt.stateIn(Unknown Source)
	at io.rsocket.kotlin.core.ReconnectableRSocketKt.connectWithReconnect(ReconnectableRSocket.kt:62)
	at io.rsocket.kotlin.core.RSocketConnector.connect(RSocketConnector.kt:52)
	at io.rsocket.kotlin.ktor.client.RSocketSupportKt.connectRSocket(RSocketSupport.kt:57)
	at io.rsocket.kotlin.ktor.client.BuildersKt.rSocket(Builders.kt:26)
	at io.rsocket.kotlin.ktor.client.BuildersKt.rSocket(Builders.kt:32)
	at io.rsocket.kotlin.ktor.client.BuildersKt.rSocket$default(Builders.kt:28)
	at mycode.cli.RsocketClientKt.main(RsocketClient.kt:91)
	at mycode.cli.RsocketClientKt$main$3.invoke(RsocketClient.kt)
	at mycode.cli.RsocketClientKt$main$3.invoke(RsocketClient.kt)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt$createCoroutineUnintercepted$$inlined$createCoroutineFromSuspendFunction$IntrinsicsKt__IntrinsicsJvmKt$1.invokeSuspend(IntrinsicsJvm.kt:223)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at kotlin.coroutines.ContinuationKt.startCoroutine(Continuation.kt:115)
	at kotlin.coroutines.jvm.internal.RunSuspendKt.runSuspend(RunSuspend.kt:19)
	at mycode.cli.RsocketClientKt.main(RsocketClient.kt)
```

It does not seem to matter whether reconnectable sockets are enabled or not.

Potentially related issues:

* https://github.com/rsocket/rsocket-kotlin/issues/148
* https://github.com/rsocket/rsocket-kotlin/issues/226

## Your Environment



* RSocket version(s) used: 0.20.0
* Other relevant libraries versions (eg. `netty`, ...): Netty 4.1.119
* Platform (eg. JVM version (`javar -version`) or Node version (`node --version`)):
```
openjdk version "17.0.13" 2024-10-15
OpenJDK Runtime Environment Temurin-17.0.13+11 (build 17.0.13+11)
OpenJDK 64-Bit Server VM Temurin-17.0.13+11 (build 17.0.13+11, mixed mode, sharing)
```

* OS and version (eg `uname -a`):
```
Linux edison 6.14.3-300.fc42.x86_64 #1 SMP PREEMPT_DYNAMIC Sun Apr 20 16:08:39 UTC 2025 x86_64 GNU/Linux
```

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Client sockets: Rejected SETUP failures are not raised to client code properly and sometimes hundreds of attempts are made #312

Expected Behavior

Actual Behavior

Your Environment

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Client sockets: Rejected SETUP failures are not raised to client code properly and sometimes hundreds of attempts are made #312

Description

Expected Behavior

Actual Behavior

Your Environment

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions