Spring A beforeCommit chain failure invariably results in a 500 response

Please consider the following test (the entire test code is provided below): it is comprised of the WebSessionTest which sends a request to the WebSessionController, the latter responds with a 200 response, afterwards the web session is being saved (see the WebSessionTestWebSessionManager) which results in the response before commit chain failure owing to the InMemoryWebSessionStore’s size is 0 (is employed by the WebSessionTestWebSessionManager). WebExceptionHandlers are invoked (no matter how an exception handler is implemented it may be either an annotated method or a WebExceptionHandler implementation) notwithstanding the client receives a 500 response.

I would like to return a 503 response with an application specific representation. I came up with a workaround in the DefaultWebSessionManager's overloaded method:

exchange.getResponse()
        .beforeCommit(() -> Mono.defer(session::save)
                .onErrorResume(this::isMaxSessionsLimitReachedException, e -> {
exchange.getResponse().setStatusCode(HttpStatus.SERVICE_UNAVAILABLE);
                    exchange.getResponse().getHeaders().clear();
                    exchange.getResponse().getCookies().clear();

                    return Mono.empty();
                }));

however it doesn't give me a chance to provide a body otherwise the client hangs.

Environment: * Java 11, * Spring 5.2.8.RELEASE, * Spring Boot 2.3.3.RELEASE.

The code of the test:

WebSessionTest:

package com.example.session;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.test.context.junit.jupiter.SpringExtension;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;

@ExtendWith(SpringExtension.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
        classes = {
                WebSessionController.class,
                WebSessionTestWebSessionManager.class,
                WebSessionTestWebExceptionHandler.class
})
@EnableAutoConfiguration
public class WebSessionTest
{
    @LocalServerPort
    private int serverPort;

    @Test
    public void testJustReponse() throws ExecutionException, InterruptedException
    {
        long timestamp = System.currentTimeMillis();

        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:" + serverPort + "/just/" + timestamp))
                .build();
        CompletableFuture<HttpResponse<String>> future =
                client.sendAsync(request, BodyHandlers.ofString()).completeOnTimeout(null, 5, TimeUnit.SECONDS);
        HttpResponse<String> response = future.get();

        assertNotNull(response, "No response in 5 seconds.");
        assertEquals(503, response.statusCode());
    }
}

WebSessionController:

package com.example.session;

import java.util.concurrent.Executors;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.WebSession;

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import static org.springframework.http.HttpStatus.SERVICE_UNAVAILABLE;

@RestController
public class WebSessionController implements InitializingBean, DisposableBean
{
    private Scheduler scheduler;

    @Override
    public void afterPropertiesSet()
    {
        scheduler = Schedulers.fromExecutorService(Executors.newCachedThreadPool());
    }

    @Override
    public void destroy()
    {
        scheduler.dispose();
    }

    @GetMapping("/just/{timestamp}")
    public Mono<ResponseEntity<String>> just(@PathVariable String timestamp, WebSession session)
    {
        return Mono.fromCallable(() -> {
            session.getAttributes().putIfAbsent("test", timestamp);

            return ResponseEntity.status(HttpStatus.OK)
                    .header(HttpHeaders.CACHE_CONTROL, "no-store")
                    .body(timestamp);
        }).subscribeOn(scheduler);
    }

    /*
    No matter how the exception handler is implemented it may be either an annotated method or a WebExceptionHandler 
    implementation – the client invariably receives a 500 response.   

    @ResponseStatus(value= SERVICE_UNAVAILABLE, reason="Too many sessions")
    @ExceptionHandler
    public void tooManySessions(Exception e)
    {
    }

    @ExceptionHandler
    public Mono<ResponseEntity<String>> tooManySessions(Exception e)
    {
        return Mono.fromCallable(() -> ResponseEntity.status(500).body(e.getMessage())).subscribeOn(scheduler);
    }

    @ExceptionHandler
    public ResponseEntity<String> tooManySessions(Exception e)
    {
        return ResponseEntity.status(503).body("To many sessions: " + e.getMessage());
    }
    */
}

WebSessionTestWebSessionManager:

package com.example.session;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.web.server.session.DefaultWebSessionManager;
import org.springframework.web.server.session.InMemoryWebSessionStore;
import org.springframework.web.server.session.WebSessionManager;

@Component("webSessionManager")
public class WebSessionTestWebSessionManager extends DefaultWebSessionManager implements WebSessionManager,
        InitializingBean
{
    private final InMemoryWebSessionStore sessionStore = new InMemoryWebSessionStore();

    @Override
    public void afterPropertiesSet()
    {
        sessionStore.setMaxSessions(0);
        super.setSessionStore(sessionStore);
    }
}

WebSessionTestWebExceptionHandler:

package com.example.session;

import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebExceptionHandler;

import reactor.core.publisher.Mono;

@Component
public class WebSessionTestWebExceptionHandler implements WebExceptionHandler
{
    @Override
    public Mono<Void> handle(ServerWebExchange exchange, Throwable throwable)
    {
        exchange.getResponse().setStatusCode(HttpStatus.SERVICE_UNAVAILABLE);
        exchange.getResponse().getHeaders().clear();
        exchange.getResponse().getCookies().clear();
        return exchange.getResponse()
                .writeWith(Flux.just(exchange.getResponse()
                        .bufferFactory()
                        .wrap("Service unavailable".getBytes(StandardCharsets.UTF_8))));
    }
}

Comment From: rstoyanchev

My initial analysis in https://github.com/spring-projects/spring-framework/issues/24186#issuecomment-690187094 is that we need to refine the solution from #24186 in order to skip pre-commit actions, when they have previously failed, and at least allow applying the current error response state.