recordServerConnectionInactive never invoked after WebSocket upgrade

[joebyneil]

Hey 👋

We noticed that when using WebSockets with our own metrics recorder, the total connection count is accurate, but the active connection count seems to climb forever. It looks like our override of recordServerConnectionInactive is not invoked in these cases.

Expected Behavior

recordServerConnectionInactive should be invoked when a WebSocket based connection ends, in addition to recordServerConnectionClosed.

When no clients are connected, there should have been an equal amount of calls to recordServerConnectionActive and recordServerConnectionInactive.

Actual Behavior

Only recordServerConnectionClosed is recorded, causing any counts of Active connections to become inaccurate.

Steps to Reproduce

I have included a reproducer in the dropdown below:

package com.example.demo;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.netty.http.HttpProtocol;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.WebsocketClientSpec;
import reactor.netty.http.server.HttpServer;
import reactor.netty.http.server.HttpServerMetricsRecorder;
import reactor.netty.http.server.WebsocketServerSpec;

import java.net.SocketAddress;
import java.time.Duration;
import java.util.concurrent.atomic.LongAdder;

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

public class MyTest {
  @Test
  public void testWebSocketConnectionCounting() {
    var metrics = new MetricsRecorder();
    var websocketServerSpec = WebsocketServerSpec.builder().build();
    var server = HttpServer.create()
        .protocol(HttpProtocol.HTTP11)
        .metrics(true, () -> metrics)
        .handle((req, res) ->
            res.sendWebsocket((in, out) -> out.sendString(Mono.just("test")), websocketServerSpec))
        .bindNow();

    var webSocketClientSpec = WebsocketClientSpec.builder().build();
    HttpClient.create()
        .remoteAddress(server::address)
        .websocket(webSocketClientSpec)
        .handle((in, out) -> out.sendClose()).then().block();

    assertThat(metrics.totalConnectionsCounter.intValue()).isEqualTo(0);
    assertThat(metrics.activeConnectionCounter.intValue()).isEqualTo(0); // Throws
  }

  // HTTP works as expected
  @Test
  public void testWebSocketConnectionCountingHttp() {
    var metrics = new MetricsRecorder();
    var server = HttpServer.create()
        .protocol(HttpProtocol.HTTP11)
        .metrics(true, () -> metrics)
        .handle((req, res) -> res.send())
        .bindNow();

    HttpClient.create()
        .remoteAddress(server::address)
        .get().response().block();

    assertThat(metrics.totalConnectionsCounter.intValue()).isEqualTo(0);
    assertThat(metrics.activeConnectionCounter.intValue()).isEqualTo(0);
  }

  public class MetricsRecorder implements HttpServerMetricsRecorder {
    public final LongAdder activeConnectionCounter = new LongAdder();
    public final LongAdder totalConnectionsCounter = new LongAdder();

    public MetricsRecorder() {
    }

    @Override
    public void recordServerConnectionOpened(SocketAddress localAddress) {
      totalConnectionsCounter.increment();
    }

    @Override
    public void recordServerConnectionClosed(SocketAddress localAddress) {
      totalConnectionsCounter.decrement();
    }

    @Override
    public void recordServerConnectionActive(SocketAddress localAddress) {
      activeConnectionCounter.increment();
    }

    @Override
    public void recordServerConnectionInactive(SocketAddress localAddress) {
      activeConnectionCounter.decrement();
    }
    // Everything below this is noop for the example, but we do use them also
    @Override
    public void recordDataReceived(SocketAddress remoteAddress, long bytes) {
    }

    @Override
    public void recordDataSent(SocketAddress remoteAddress, long bytes) {
    }

    @Override
    public void incrementErrorsCount(SocketAddress remoteAddress) {
    }

    @Override
    public void recordTlsHandshakeTime(SocketAddress remoteAddress, Duration time, String status) {
    }

    @Override
    public void recordConnectTime(SocketAddress remoteAddress, Duration time, String status) {
    }

    @Override
    public void recordResolveAddressTime(SocketAddress remoteAddress, Duration time, String status) {
    }

    @Override
    public void recordDataReceivedTime(String uri, String method, Duration time) {
    }

    @Override
    public void recordDataSentTime(String uri, String method, String status, Duration time) {
    }

    @Override
    public void recordResponseTime(String uri, String method, String status, Duration time) {
    }

    @Override
    public void recordStreamOpened(SocketAddress localAddress) {
    }

    @Override
    public void recordStreamClosed(SocketAddress localAddress) {
    }

    @Override
    public void recordDataReceived(SocketAddress remoteAddress, String uri, long bytes) {
    }

    @Override
    public void recordDataSent(SocketAddress remoteAddress, String uri, long bytes) {
    }

    @Override
    public void incrementErrorsCount(SocketAddress remoteAddress, String uri) {
    }
  }
}

Possible Solution

We believe that this might be because the metrics handler is removed in WebSocketServerOperations.java.

reactor-netty/reactor-netty-http/src/main/java/reactor/netty/http/server/WebsocketServerOperations.java

Line 97 in dd0c42d

removeHandler(NettyPipeline.HttpMetricsHandler);

Thinking that if we expect this not to record active/inactive for every WebSocket message, perhaps we should record that the connection is inactive at the completion of the upgrade. Otherwise, closing when the WebSocket connection ends would also work.

Your Environment

reactor-netty: 1.1.18

• JVM version (java -version):
  openjdk version "17.0.6" 2023-01-17 LTS
  OpenJDK Runtime Environment Zulu17.40+19-CA (build 17.0.6+10-LTS)
  OpenJDK 64-Bit Server VM Zulu17.40+19-CA (build 17.0.6+10-LTS, mixed mode, sharing)
• OS and version (eg. uname -a):
  Darwin Joebys-Laptop.local 23.4.0 Darwin Kernel Version 23.4.0: Fri Mar 15 00:10:42 PDT 2024; root:xnu-10063.101.17~1/RELEASE_ARM64_T6000 arm64

[violetagg]

@joebyneil Thanks for the report and the reproducible example. I confirm that this is an issue and it will be fixed for the next release.