SocketAcceptorInterceptor receive an unexpected payload object rather than receive a setup payload object

We create a client and server with reconnection feature. The server register a interceptor to receive the setup payload and then deserialized for authentication.

Client have another daemon thread called Thread A and send data by rsocket periodically. The sended payload data we call it payload-A.

After server is stopped and we can see the priodically sended data is failed as what we expected. After the server is started again, client begin to reconnect to the server. We can see the socket acceptor receive the setup payload data, but it is not the setup data we generated by method getAuthInfo() method. The server sidecar acceptor receive an unexpected normal payload data which is very strange.

Related server and client code:

// client 
        requesterMono = rsocketRequesterBuilder.setupRoute(SETUP_ROUTE)
            .setupData(new ObjectMapper().writeValueAsString(getAuthInfo()))
            .dataMimeType(MimeTypeUtils.parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString()))
            .rsocketConnector(connector -> connector.acceptor(responder).fragment(MTU_BYTE_SIZE)
                .keepAlive(Duration.ofSeconds(KEEP_ALIVE_INTERVAL_SEC), Duration.ofSeconds(KEEP_ALIVE_MAX_TIME_SEC))
                .reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(RSOCKET_RETRY_INTERVAL_SECONDS))
                    .doAfterRetry(retrySignal -> log.warn("RSocket client is reconnecting to get the newest connection...."))))
            .connect(TcpClientTransport.create(TcpClient.create()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, TCP_CONN_TIMEOUT)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .host(consoleDomain)
                .port(SERVER_PORT)
                .secure(ssl -> ssl.sslContext(sslContext))));

// register a socket acceptor interceptor to receive the authentication setup payload object in the server side
    protected SocketAcceptorInterceptor genAcceptorAuthInterceptor() {
        SocketAcceptorInterceptor interceptor = acceptor -> (setup, sendingSocket) -> {
            try {
                log.info("receive an task connection try to init.");
                LocalTaskAuthInfo authInfo = new ObjectMapper().readValue(setup.getDataUtf8(), LocalTaskAuthInfo.class);

              //ignore some non-related code...
                return acceptor.accept(setup, sendingSocket);
            } catch (Exception e) {
                String errMsg = "try to accept a connection error.msg:" + ExceptionUtils.getRootCauseMessage(e);
                log.warn(errMsg, e);
                exceptionService.saveException(new SidecarRuntimeException(e));
                return Mono.error(new RSocketAuthException(errMsg));
            }
        };
        return interceptor;
    }



//server
 RSocketServer.create(routeHandler.responder()).fragment(RSocketConnConstans.MTU_BYTE_SIZE)
            .interceptors(registry -> registry.forSocketAcceptor(interceptor))
            .bind(TcpServerTransport.create(TcpServer.create()
                .option(EpollChannelOption.SO_KEEPALIVE, true)
                .option(EpollChannelOption.TCP_NODELAY, true)
                .secure(ssl -> ssl.sslContext(sslContext))
                .port(Integer.valueOf(consoleConfig.getRsocketConsolePort()))
                .doOnConnection(
                    t -> log.info("New sidecar CONNECTED. Pay attention that this is maybe a resumed connection..."))))
            .subscribe();

Expected Behavior

Receive setup payload in the socket acceptor

Actual Behavior

Receive a normal payload data.

Steps to Reproduce

Unfortuncately that we can't reproduce it stablely. The issue occur sometimes after server is restarted. Usually after the server is restarted everything goes well.

Possible Solution

May be this is a parallel bug. It occur now and then. I know this issue may ba a little difficult. Is there any suggestion that I can take to set the breakpoint. This issue's key point is to find out why the normal paryload data replace the setup payload data and then cause the deserialization problem.

Your Environment

RSocket version(s) used: rsocket-core-1.0.2 and springboot-starter-rsocket 2.3.4.RELEASE
Platform (eg. JVM version (javar -version) or Node version (node --version)): OpenJDK 1.8
OS and version (eg uname -a): Mac OS 10.15.4

@KaimingWan

@KaimingWan it looks like you are observing an already fixed bug (please see #842).

Can you please check if the same is reproducing with 1.0.3 or 1.1.0?

@OlegDokuka

@OlegDokuka Thanks, I will try to upgrade our rsocket version to check whether this issue will happen again. However, I still have some confusion. I checked the issue #842 and find its problem is the cnf count is 0 when reconnecting. His case seems can be reproduced always. My case have some difference because the issue happen now and then. Can you tell me the bug's mechanism and in what situation it will happen? I want to ensure that my issue is caused by the related bug.

@OlegDokuka

@OlegDokuka I updateed my issue. I checked that my rsocket-core version is 1.0.2 and the mentioned bug is fixed. So I think my probleam is not caused by mentioned bug in #842 . Can you give any idea that where I can set the breakpoint to capture the moment when setupPayload is not an object instance of ConnectionSetupPayload. I can set the breakpoint to check why during reconnection my server will receive a non-setup payload contents.

@OlegDokuka

@OlegDokuka To get more information. Would you mind telling me where to set the breakpoint when reconnection haapen and the setup payload is set? I now set the breakpoint in RSocketConnector class's method setupPayload and I find I can't capture any case that I receive a ConnectionSetupPayload.

@OlegDokuka

@OlegDokuka Now I set breakpoint in RSocketServer's acceptSetup method. I will give more info if I encounter this issue again.

@KaimingWan

@KaimingWan is it still an issue for you? If so, can you try to reproduce that issue with a pure rsocket setup (I mean your example uses spring-messaging rsocket, however, to minimize the possible error surface, I need to make sure that you observe the same problem with vanilla rsocet-java)

@OlegDokuka

@OlegDokuka Sorry for the late response. The issue happens now and then. Recently I meet the issue again and I put forward the issue here: spring-projects/spring-framework#26626 . May be a little more detailed than this issue I created before. I will try to write an example to show what is going on. But the issue may not happen every time, the example may not reproduce what I met. However, another issue always happen is that the first stream frame log that carry setup payload info is missing when client connect to server. As I expected, the first stream should carry the setup payload and been sent to SETUP_ROUTE( route name is "conn-setup") in my case, but I didn't see any logs. Later I will give an example here.

@OlegDokuka

@OlegDokuka I write a simple example but it seems everything goes well. The first frame carry the setup payload is print out normally. Where to set the break point if I want to debug my real world code. BTW, my test example is as follow:

package com.clougence.test.rsocket;

import static com.clougence.clouddm.comm.constants.rsocket.RSocketConnConstants.MTU_BYTE_SIZE;

import java.time.Duration;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.util.pattern.PathPatternRouteMatcher;
import com.clougence.clouddm.base.metadata.auth.ConnAuthDTO;
import com.clougence.clouddm.comm.constants.rsocket.RSocketConnConstants;
import com.clougence.clouddm.comm.util.HostUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelOption;
import io.netty.channel.epoll.EpollChannelOption;
import io.rsocket.SocketAcceptor;
import io.rsocket.metadata.WellKnownMimeType;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.TcpServerTransport;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.netty.tcp.TcpClient;
import reactor.netty.tcp.TcpServer;
import reactor.test.StepVerifier;
import reactor.util.retry.Retry;

/**
 * @author wanshao create time is 2021/3/2
 **/
@SpringBootTest
@Slf4j
public class TestSetupPayload {

    private static Mono<RSocketRequester> requesterMono;

    private final static int              SERVER_PORT                    = 5678;

    private final static int              TCP_CONN_TIMEOUT               = 3000;
    private final static int              RSOCKET_RETRY_INTERVAL_SECONDS = 5;
    private final static int              KEEP_ALIVE_INTERVAL_SEC        = 5;
    private final static int              KEEP_ALIVE_MAX_TIME_SEC        = 30;

    @SneakyThrows
    @BeforeAll
    public static void setupOnce() {
        RSocketStrategies rSocketStrategies = RSocketStrategies.builder()
            .encoders(encoders -> encoders.add(new Jackson2JsonEncoder()))
            .decoders(decoders -> decoders.add(new Jackson2JsonDecoder()))
            .routeMatcher(new PathPatternRouteMatcher())
            .build();
        RSocketRequester.Builder rsocketRequesterBuilder = RSocketRequester.builder().rsocketStrategies(rSocketStrategies);
        ConnAuthDTO connAuthDTO = ConnAuthDTO.builder()
            //
            .ak("myak")
            .sk("mysk")
            .ip(HostUtil.getHostIp())
            .wsn("wsn")
            .sendAuthTimeMs(System.currentTimeMillis())
            .build();

        SocketAcceptor responder = RSocketMessageHandler.responder(rSocketStrategies, new TestSetupPayload());

        // init server
        io.rsocket.core.RSocketServer.create()
            .interceptors(registry -> registry.forSocketAcceptor(new TestSocketAcceptor()))
            .acceptor(responder)
            .fragment(RSocketConnConstants.MTU_BYTE_SIZE)
            .bind(TcpServerTransport.create(TcpServer.create()
                .option(EpollChannelOption.SO_KEEPALIVE, true)
                .option(EpollChannelOption.TCP_NODELAY, true)
                .port(SERVER_PORT)
                .doOnConnection(t -> log.info("NEW WORKER CONNECTED...."))))
            .subscribe();

        // init client

        requesterMono = rsocketRequesterBuilder.setupRoute("conn-setup")
            .setupData(new ObjectMapper().writeValueAsString(connAuthDTO))
            .dataMimeType(MimeTypeUtils.parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString()))
            .rsocketConnector(connector -> connector.acceptor(responder)
                .fragment(MTU_BYTE_SIZE)
                .keepAlive(Duration.ofSeconds(KEEP_ALIVE_INTERVAL_SEC), Duration.ofSeconds(KEEP_ALIVE_MAX_TIME_SEC))
                .reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(RSOCKET_RETRY_INTERVAL_SECONDS))
                    .doAfterRetry(retrySignal -> log.warn("RSocket client is reconnecting to get the newest connection...."))))
            .connect(TcpClientTransport.create(TcpClient.create()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, TCP_CONN_TIMEOUT)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .host("localhost")
                .port(SERVER_PORT)));

    }

    @Test
    public void testRequestGetsResponse() {
        // Send a request message
        Mono<String> result = requesterMono.block().route("default-handler").data("hello").retrieveMono(String.class);

        // Verify that the response message contains the expected data
        StepVerifier.create(result).expectNext("success").verifyComplete();
    }

    @MessageMapping("default-handler")
    Mono<String> requestResponse(final String msg) {
        log.info("Received request-response request: {}", msg);
        return Mono.just("success");
    }

}

test acceptor is :

package com.clougence.test.rsocket;

import org.apache.commons.lang3.exception.ExceptionUtils;
import com.clougence.clouddm.base.metadata.auth.ConnAuthDTO;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.rsocket.SocketAcceptor;
import io.rsocket.plugins.SocketAcceptorInterceptor;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

/**
 * @author wanshao create time is 2021/3/2
 **/
@Slf4j
public class TestSocketAcceptor implements SocketAcceptorInterceptor {

    public TestSocketAcceptor(){
    }

    @Override
    public SocketAcceptor apply(SocketAcceptor socketAcceptor) {
        return (setup, sendingSocket) -> {
            try {
                // Here will occur issue
                ConnAuthDTO authInfo = new ObjectMapper().readValue(setup.getDataUtf8(), ConnAuthDTO.class);

                return socketAcceptor.accept(setup, sendingSocket);
            } catch (Exception e) {
                String errMsg = "try to accept a connection error.msg:" + ExceptionUtils.getRootCauseMessage(e);
                log.warn(errMsg, e);
                return Mono.error(new com.clougence.clouddm.base.metadata.rpc.rsocket.RSocketAuthException(errMsg));
            }
        };
    }

}

The log is ：

@KaimingWan

@KaimingWan we had a chat with @rstoyanchev and it can be that this is spring-messaging issue and the fact that the given DTO can be modified (at least that is what I understood from your comments on the linked issue) which means the spring codecs which coverts given object to ByteBuf may provide a corrupted one.

Thus, not sure it is rsocket-java issue

Ensure setupPayload is actually copied

@OlegDokuka

@OlegDokuka Thanks for your information. In my case, not only the given setup payload is truncated and it is also be replaced by another dto's value. It seems more like an issue caused by concurrency. In my real world code, I try to ensure there is no other send request when rsocket is reconnecting. You said it may caused by spring-messaging's process. Are there any advice about where to set the breakpoint? Another clus is clear that in my real world code, every time I restart the client I can find the first stream is missing.

Is keep alive frame have relation with this issue? I debug my code and find the first stream can be printed out during my step by step debug.

I debug my code and especially pay attention to the code from spring-messaging. I find after connected, the spring-rsocket only return a requester object and will not send setup payload to the remote

DefaultPayload#create properly copies ByteBuf content

@KaimingWan

@KaimingWan after some more looking, we found the problem. The fix is available in snapshots. Not sure if you're able to confirm the fix with 1.0.4-SNAPSHOT via https://oss.jfrog.org/artifactory/oss-snapshot-local -- that would be very helpful, but either way this fix should help to address the issue.

@rstoyanchev

@rstoyanchev Thanks for your work. Because the issue occur in our real world code, so I can't use the snapshot directly in the production environment. However, I will try to use the snapshot in my develeopment environment and see whether the issue will happen again. Are there any plan about when 1.0.4 release? BTW, it the following commit fix this issue? Would you mind telling me what is the key point of this fix? I still do not know in what situation the payload will be corrupted and why after receive the corrupted payload, rsocket will keep retrying.

@KaimingWan

@KaimingWan The issue was in the DefaultPayload.create(ByteBuf, ByteBuf) method, which internally did only ByteBuf#nioBuffer which returned access to shared memory without copying it. That said, when given ByteBuf is released, the shared NIO ByteBuffer can be easily corrupted. That is basically what you have seen. With the #993, we ensured that content is really copied so no corruption should occur

@OlegDokuka

@OlegDokuka Thanks for your explanation. I find 1.0.4 has been released yesterday and we have changed to this version. Respect for you guys's efficient work~

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Repository navigation

SocketAcceptorInterceptor receive an unexpected payload object rather than receive a setup payload object #970

Expected Behavior

Actual Behavior

Steps to Reproduce

Possible Solution

Your Environment

1 remaining item

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Sunbelt Computer Software

PL/B Language Development and Support

SocketAcceptorInterceptor receive an unexpected payload object rather than receive a setup payload object #970

Description

Expected Behavior

Actual Behavior

Steps to Reproduce

Possible Solution

Your Environment

Activity

OlegDokuka commented on Dec 30, 2020

KaimingWan commented on Dec 31, 2020

KaimingWan commented on Dec 31, 2020

KaimingWan commented on Jan 4, 2021

KaimingWan commented on Jan 4, 2021

OlegDokuka commented on Jan 20, 2021

KaimingWan commented on Mar 2, 2021

KaimingWan commented on Mar 2, 2021

OlegDokuka commented on Mar 2, 2021

1 remaining item

KaimingWan commented on Mar 3, 2021

KaimingWan commented on Mar 3, 2021

KaimingWan commented on Mar 3, 2021

rstoyanchev commented on Mar 3, 2021

KaimingWan commented on Mar 4, 2021

OlegDokuka commented on Mar 4, 2021

KaimingWan commented on Mar 5, 2021

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions