STOMP ERROR frame is lost when another write is in progress on the session

[dkrizan]

Affects: 7.0.9, 7.1.0-M1

When an inbound frame fails, for example a ChannelInterceptor throws on SUBSCRIBE, the client should get an ERROR frame before the connection closes. If another thread is writing to the same session at that moment, the client gets only the close.

Cause

1. StompSubProtocolHandler.sendErrorMessage calls session.sendMessage(...) and then session.close(CloseStatus.PROTOCOL_ERROR) in finally.
2. If another thread holds the flush lock of ConcurrentWebSocketSessionDecorator, sendMessage only adds the frame to the buffer.
3. close() sets closeInProgress, so the other thread drops the buffered ERROR frame.

Clients that decide what to do based on the ERROR message, for example stop reconnecting on Unauthenticated, see a plain disconnect instead. Possibly related to #17911.

How to reproduce

A small Spring Boot 4.1.1 project (code below) with a simple broker and an interceptor that rejects every SUBSCRIBE:

• errorFrameIsDelivered passes.
• errorFrameIsDeliveredWhileAnotherWriteIsInProgress fails every time. A test-only decorator keeps the write of CONNECTED in progress until the SUBSCRIBE is rejected, like a slow client would. The client gets ConnectionLostException and no ERROR frame.

Possible fix

Maybe close() could first write what is already in the buffer, at least for PROTOCOL_ERROR. Or sendErrorMessage could close only after the frame is really written.

build.gradle

plugins {
    id 'java'
    id 'org.springframework.boot' version '4.1.1'
    id 'io.spring.dependency-management' version '1.1.7'
}

java {
    sourceCompatibility = JavaVersion.VERSION_17
}

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-websocket'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

test {
    useJUnitPlatform()
    testLogging { events 'passed', 'failed'; exceptionFormat 'full' }
}

src/main/java/repro/ReproApplication.java

package repro;

import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

@SpringBootApplication
public class ReproApplication {

    @Configuration
    @EnableWebSocketMessageBroker
    static class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

        @Override
        public void registerStompEndpoints(StompEndpointRegistry registry) {
            registry.addEndpoint("/ws");
        }

        @Override
        public void configureMessageBroker(MessageBrokerRegistry registry) {
            registry.enableSimpleBroker("/topic");
        }

        /** Rejects every SUBSCRIBE the documented way: by throwing from an inbound interceptor. */
        @Override
        public void configureClientInboundChannel(ChannelRegistration registration) {
            registration.interceptors(new ChannelInterceptor() {
                @Override
                public Message<?> preSend(Message<?> message, MessageChannel channel) {
                    StompHeaderAccessor accessor =
                            MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                    if (accessor != null && StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
                        throw new MessagingException("Unauthenticated");
                    }
                    return message;
                }
            });
        }
    }
}

src/test/java/repro/ErrorFrameDeliveryTest.java

package repro;

import java.lang.reflect.Type;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.jupiter.api.Test;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.converter.SimpleMessageConverter;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
import org.springframework.web.socket.handler.WebSocketHandlerDecorator;
import org.springframework.web.socket.handler.WebSocketSessionDecorator;
import org.springframework.web.socket.messaging.WebSocketStompClient;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(classes = ReproApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Import(ErrorFrameDeliveryTest.SlowOutboundWriteConfig.class)
class ErrorFrameDeliveryTest {

    /** When true, the server's write of CONNECTED is still in progress while SUBSCRIBE is rejected. */
    static volatile boolean slowOutboundWrite;

    @Value("${local.server.port}")
    int port;

    @Test
    void errorFrameIsDelivered() throws Exception {
        slowOutboundWrite = false;
        assertThat(subscribeAndAwaitErrorFrame()).isEqualTo("Unauthenticated");
    }

    @Test
    void errorFrameIsDeliveredWhileAnotherWriteIsInProgress() throws Exception {
        slowOutboundWrite = true;
        assertThat(subscribeAndAwaitErrorFrame()).isEqualTo("Unauthenticated");
    }

    private String subscribeAndAwaitErrorFrame() throws Exception {
        CompletableFuture<String> errorMessage = new CompletableFuture<>();
        StompSessionHandlerAdapter handler = new StompSessionHandlerAdapter() {
            @Override
            public Type getPayloadType(StompHeaders headers) {
                return byte[].class;
            }

            @Override
            public void handleFrame(StompHeaders headers, Object payload) {
                errorMessage.complete(headers.getFirst("message"));
            }

            @Override
            public void handleTransportError(StompSession session, Throwable exception) {
                errorMessage.complete("<no ERROR frame, transport closed: " + exception + ">");
            }
        };

        WebSocketStompClient client = new WebSocketStompClient(new StandardWebSocketClient());
        client.setMessageConverter(new SimpleMessageConverter());
        try {
            StompSession session = client.connectAsync("ws://localhost:" + port + "/ws", handler)
                    .get(5, TimeUnit.SECONDS);
            session.subscribe("/topic/anything", handler);
            return errorMessage.get(10, TimeUnit.SECONDS);
        }
        finally {
            client.stop();
        }
    }

    /**
     * Test-only session decorator that keeps the outbound write of CONNECTED "in progress" (as a slow
     * client or network would) until the rejected SUBSCRIBE has arrived and the resulting close() has
     * been requested. It sits under ConcurrentWebSocketSessionDecorator, so the flush lock stays held.
     */
    @TestConfiguration
    static class SlowOutboundWriteConfig implements WebSocketMessageBrokerConfigurer {

        private static final Map<String, CountDownLatch[]> latches = new ConcurrentHashMap<>();

        private static CountDownLatch[] latchesFor(WebSocketSession session) {
            return latches.computeIfAbsent(session.getId(), id -> new CountDownLatch[] {new CountDownLatch(1), new CountDownLatch(1)});
        }

        @Override
        public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
            registration.addDecoratorFactory(handler -> new WebSocketHandlerDecorator(handler) {
                @Override
                public void afterConnectionEstablished(WebSocketSession session) throws Exception {
                    super.afterConnectionEstablished(new WebSocketSessionDecorator(session) {
                        @Override
                        public void sendMessage(WebSocketMessage<?> message) throws java.io.IOException {
                            super.sendMessage(message);
                            if (slowOutboundWrite && message.getPayload().toString().startsWith("CONNECTED")) {
                                awaitQuietly(latchesFor(this)[0], 5000);   // SUBSCRIBE received
                                awaitQuietly(latchesFor(this)[1], 1000);   // close() requested
                            }
                        }

                        @Override
                        public void close(CloseStatus status) throws java.io.IOException {
                            latchesFor(this)[1].countDown();
                            super.close(status);
                        }
                    });
                }

                @Override
                public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
                    if (message.getPayload().toString().startsWith("SUBSCRIBE")) {
                        latchesFor(session)[0].countDown();
                    }
                    super.handleMessage(session, message);
                }
            });
        }

        private static void awaitQuietly(CountDownLatch latch, long millis) {
            try {
                latch.await(millis, TimeUnit.MILLISECONDS);
            }
            catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

[lzcGeek]

Independent confirmation on 7.0.9, plus one finding that I think decides the approach.

Reproduction

I rebuilt the reproduction from the description on Spring Boot 4.1.1 / Spring Framework 7.0.9 (JDK 17) and it fails deterministically:

expected: "Unauthenticated"
 but was: "<no ERROR frame, transport closed:
           org.springframework.messaging.simp.stomp.ConnectionLostException: Connection closed>"

errorFrameIsDelivered passes; errorFrameIsDeliveredWhileAnotherWriteIsInProgress fails every run.

I also reduced it to a seconds-long deterministic check against the decorator alone, without a broker or a servlet container - a stub WebSocketSession plus two latches that keep thread A inside the delegate's sendMessage until B has called close():

delegate received    : [CONNECTED]
ERROR delivered      : false
close() blocked for  : 2 ms

So the loss is entirely inside ConcurrentWebSocketSessionDecorator: B buffers the ERROR frame because flushLock is held, close() sets closeInProgress = true, and when A finally drains, tryFlushMessageBuffer() breaks immediately on shouldNotSend() (limitExceeded || closeInProgress) and discards the frame.

The finding: a tryLock-style drain cannot fix this

My first attempt added a close()-time drain that reused the existing tryFlushMessageBuffer() logic. It does not work: that method uses flushLock.tryLock(), and in exactly this race the lock is still held by the in-flight write, so the drain is a no-op and the ERROR frame is still lost. I verified this - the reproduction still reported ERROR delivered: false.

Waiting for the lock does fix it:

variant	delegate receives	ERROR delivered	close() blocks
as-is	[CONNECTED]	no	2 ms
drain on close() with tryLock	[CONNECTED]	no	2 ms
drain on close() after acquiring the lock	[CONNECTED, ERROR frame: Unauthenticated]	yes	~10 s in my probe (see below)

Why I did not open a PR for this

Making close() wait for the lock means a slow or stalled write can hold the close for an unbounded time, which cuts against the reason this decorator enforces a send-time limit in the first place. My probe shows close() blocking for the full duration of the in-flight write. A bounded wait changes behaviour too, and the right trade-off is yours to pick, so please tell me which you want:

1. drain on close() with an unbounded lock(),
2. drain with a bounded wait (e.g. sendTimeLimit), then close regardless,
3. or the other direction entirely - keep the decorator as-is and make StompSubProtocolHandler.sendErrorMessage only close once the ERROR frame has actually been written (serialize on the send path, or wait for a write-completion signal).

There is a second semantic question either way: a drain before closeInProgress is set opens a window in which a concurrent sendMessage enqueues into the same buffer and gets sent along with the ERROR frame. I assume that is acceptable, but it should be a decision rather than a side effect.

For context, #3940 was closed without merging, so I would rather agree on the shape before spending your review time on a patch.