WebFlux may send incomplete response if the session persistence fails

[evgenyvsmirnov]

Components and interactions

Please have a look at the following test. It contains:

• WebSessionController - a server which replies to /just/{timestamp} HTTP requests and has an exception handler which responds the 500 answer.
• WebSessionTestWebSessionManager - sets a concurrent sessions limit to 0 for the sake of testing.
• WebSessionTest - a test client which sends /just/12345 HTTP request and waits for the response at most 5 seconds.

WebSessionController:

package com.example;

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.INTERNAL_SERVER_ERROR;

@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);
    }

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

    /*
    No difference how the handler is implemented or does it exist at all.

    @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(500).body(e.getMessage());
    }
    */
}

WebSessionTestWebSessionManager:

package com.example;

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);
    }
}

WebSessionTest:

package com.example;

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})
@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 after 5 seconds.");
        assertEquals(500, response.statusCode());
    }
}

pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>com.example.springweb</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.target>11</maven.compiler.target>
        <maven.compiler.source>11</maven.compiler.source>
    </properties>

    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <compilerArgs combine.children="append">
                        <arg>-implicit:none</arg>
                    </compilerArgs>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-jar-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifestFile combine.self="override" />
                    </archive>
                </configuration>
            </plugin>
        </plugins>
    </build>


    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <version>5.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-test-autoconfigure</artifactId>
            <version>2.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
            <version>2.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-beans</artifactId>
            <version>5.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
            <version>2.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>5.2.2.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.hamcrest</groupId>
            <artifactId>hamcrest-core</artifactId>
            <version>1.3</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>5.5.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

Expected result

Client gets 4xx/5xx response or at least ends up with the closed connection caused by exceeded concurrent sessions limit number.

Actual result

Client hangs.

Additional information

If I suspend the test with the IDE debugger (it blocks the test thread only) and send a request via curl I see the following output:

curl -v http://localhost:53857/just/1234567
*   Trying ::1...
* TCP_NODELAY set
* Connected to localhost (::1) port 53857 (#0)
> GET /just/1234567 HTTP/1.1
> Host: localhost:53857
> User-Agent: curl/7.64.1
> Accept: */*
>
< HTTP/1.1 200 OK
< Cache-Control: no-store
< Content-Type: text/plain;charset=UTF-8
< Content-Length: 7
<

Here "Content-Length: 7" is the length of 200 response (the contents of the URI's last segment) which WebSessionController sends, so the status line and the headers have been sent but the body hasn't. It looks like if the server had sent the status line and the headers with an exception occurred afterwards which resulted in the inability to send error response.

Environment:

• springframework: 5.2.2.RELEASE,
• springboot: 2.2.2.RELEASE,
• Java version: 11

[rstoyanchev]

I can confirm there is an issue. The session persistence failure happens at an inconvenient time, after the response is committed and the content-length set. That means error handling can't change the response status and the client hangs expecting content due to the content-length header.

[rstoyanchev]

This should be fixed in the latest 5.2.3 snapshots. I've verified with code similar to the snippets above.

[evgenyvsmirnov]

@rstoyanchev is it possible to customize an error response which is generated as a result of a before commit chain failure? Now if session.save fails (because the session store exceeds the predefined size) user invariably receives 500 response. For this particular situation I would prefer 503, but it seems WebExceptionHandlers are never activated.
I came up with a workaround:

...
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).

Your expertise would be valued. Thanks!

[rstoyanchev]

@evgenyvsmirnov from what I can see a WebExceptionHandler can handle the exception but when HttpWebHandlerAdapter calls setComplete() at the end, that fails to update the underlying response because the failed pre-commit action fails yet again.

I think we can introduce an additional state to remember that a pre-commit action failed. Then if the error is handled, on the second attempt to commit, if the response contains an error code and we know a pr-commit action failed, we can proceed straight to updating the response.

In addition we could also have InMemoryWebSessionStore use a ResponseStatusException to produce a 503 by default rather than an IllegalStateException.

Can you open a separate issue for this?

[evgenyvsmirnov]

@rstoyanchev , thank you for a swift reply. Have created the issue: #25753