Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Open
rocketraman opened this issue May 6, 2025 · 0 comments
Open

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

rocketraman opened this issue May 6, 2025 · 0 comments

Comments

@rocketraman
Copy link

rocketraman commented May 6, 2025 •
edited
Loading

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:

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
@rocketraman rocketraman changed the title Auth failures at SETUP are not raised to client code properly Rejected SETUP failures are not raised to client code properly May 6, 2025
@rocketraman rocketraman changed the title Rejected SETUP failures are not raised to client code properly Rejected SETUP failures are not raised to client code properly for reconnectable sockets May 7, 2025
@rocketraman rocketraman changed the title Rejected SETUP failures are not raised to client code properly for reconnectable sockets Reconnectable client sockets: Rejected SETUP failures are not raised to client code properly and hundreds of attempts are made May 7, 2025
@rocketraman rocketraman changed the title Reconnectable client sockets: Rejected SETUP failures are not raised to client code properly and hundreds of attempts are made Client sockets: Rejected SETUP failures are not raised to client code properly and sometimes hundreds of attempts are made May 7, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

No branches or pull requests
1 participant
@rocketraman