Fix handling of rejected setup errors (#1117)

Closes gh-1092

Author: rstoyanchev

rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -463,8 +463,14 @@ private Mono<Void> acceptSetup(
           return interceptors
               .initSocketAcceptor(acceptor)
               .accept(setupPayload, wrappedRSocketRequester)
-              .doOnError(
-                  err -> serverSetup.sendError(wrappedDuplexConnection, rejectedSetupError(err)))
+              .onErrorResume(
+                  err ->
+                      Mono.fromRunnable(
+                              () ->
+                                  serverSetup.sendError(
+                                      wrappedDuplexConnection, rejectedSetupError(err)))
+                          .then(wrappedDuplexConnection.onClose())
+                          .then(Mono.error(err)))
               .doOnNext(
                   rSocketHandler -> {
                     RSocket wrappedRSocketHandler = interceptors.initResponder(rSocketHandler);

rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java

@@ -287,6 +287,9 @@ public void setKeepAliveSupport(KeepAliveSupport keepAliveSupport) {
 
   @Override
   public void dispose() {
+    if (logger.isDebugEnabled()) {
+      logger.debug("Side[server]|Session[{}]. Disposing session", session);
+    }
     Operators.terminate(S, this);
     resumableConnection.dispose();
   }

rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,11 +23,14 @@
 import io.rsocket.Closeable;
 import io.rsocket.FrameAssert;
 import io.rsocket.RSocket;
+import io.rsocket.exceptions.RejectedSetupException;
 import io.rsocket.frame.FrameType;
 import io.rsocket.frame.KeepAliveFrameCodec;
 import io.rsocket.frame.RequestResponseFrameCodec;
+import io.rsocket.frame.SetupFrameCodec;
 import io.rsocket.test.util.TestDuplexConnection;
 import io.rsocket.test.util.TestServerTransport;
+import io.rsocket.util.EmptyPayload;
 import java.time.Duration;
 import java.util.Random;
 import org.assertj.core.api.Assertions;
@@ -164,4 +167,35 @@ public void unexpectedFramesBeforeSetup() {
     server.dispose();
     transport.alloc().assertHasNoLeaks();
   }
+
+  @Test
+  public void ensuresErrorFrameDeliveredPriorConnectionDisposal() {
+    TestServerTransport transport = new TestServerTransport();
+    Closeable server =
+        RSocketServer.create()
+            .acceptor(
+                (setup, sendingSocket) -> Mono.error(new RejectedSetupException("ACCESS_DENIED")))
+            .bind(transport)
+            .block();
+
+    TestDuplexConnection connection = transport.connect();
+    connection.addToReceivedBuffer(
+        SetupFrameCodec.encode(
+            ByteBufAllocator.DEFAULT,
+            false,
+            0,
+            1,
+            Unpooled.EMPTY_BUFFER,
+            "metadata_type",
+            "data_type",
+            EmptyPayload.INSTANCE));
+
+    StepVerifier.create(connection.onClose()).expectComplete().verify(Duration.ofSeconds(30));
+    FrameAssert.assertThat(connection.pollFrame())
+        .hasStreamIdZero()
+        .hasData("ACCESS_DENIED")
+        .hasNoLeaks();
+    server.dispose();
+    transport.alloc().assertHasNoLeaks();
+  }
 }