Resume with cleanupStoreOnKeepAlive(): out-of-order KEEPALIVE positions close the session ("Given Remote Implied Position is behind the last received Remote Implied Position") #1134

With Resume#cleanupStoreOnKeepAlive() enabled, a short stall of the keepalive tick thread can make one side send a KEEPALIVE whose Last Received Position is lower than the one in a KEEPALIVE it sent just before. The receiving side treats that as fatal: InMemoryResumableFramesStore#releaseFrames throws, and the session is closed with a connection error. The transport is still healthy at that point, so resumption cannot recover the session, and every stream on it fails.

We hit this in a nightly test environment between two JVMs. It is rare, because it needs the tick thread to be descheduled for a few tens of milliseconds at the moment a keepalive is sent.

Expected behavior

A KEEPALIVE carrying a lower position than one already received is stale information. It should be ignored, or at most logged, and should not close the session. The spec describes the keepalive position as information the peer MAY use to trim a retransmit buffer. Since implied positions only grow in each direction, a lower value can only be out of date.

Actual behavior

The session is closed:

ConnectionErrorException (0x101): Given Remote Implied Position is behind the last received Remote Implied Position

Cause

Each side sends KEEPALIVE frames from two threads, and each thread reads the implied position before enqueueing the frame, with nothing ordering the two:

• Interval tick: ClientKeepAliveSupport#onIntervalTick runs on Schedulers.parallel() (KeepAliveSupport.java#L176-L180). The server runs this tick too, using the interval from SETUP (RSocketServer.java#L452).
• Reply: KeepAliveSupport#receive answers the peer's KEEPALIVE on the receiving thread (KeepAliveSupport.java#L88-L102).

If the tick thread is descheduled after reading position p1 and before enqueueing, the reply reads p2 > p1 and is enqueued first. The wire then carries p2 followed by p1. The receiving store throws on p1 (InMemoryResumableFramesStore.java#L139-L143), and ServerRSocketSession#onImpliedPosition turns that into sendErrorAndClose (ServerRSocketSession.java#L252-L260). ClientRSocketSession#onImpliedPosition does the same in the other direction.

Both sides start their keepalive intervals at connection setup, so their KEEPALIVEs go out nearly together, and a stall of roughly one round trip is enough. The server-initiated keepalives reported in #1128 are what give the client its second sending thread here.

Steps to reproduce

Dependencies: io.rsocket:rsocket-core:1.1.5 and io.rsocket:rsocket-transport-netty:1.1.5. Run main.

The reproducer stalls the client's first interval tick for 300 ms inside the frame allocation, which sits between reading the position and enqueueing the frame. It uses a client-only ByteBufAllocator for this. It also replaces Schedulers.parallel() with a pool. Otherwise, on a machine with few cores, the server's tick can land on the same pinned worker as the stalled client tick, be held back with it, and the race never happens.

It prints Session closed: ConnectionErrorException (0x101): Given Remote Implied Position is behind the last received Remote Implied Position on every run, including with -XX:ActiveProcessorCount=1. With Thread.sleep(0) it prints Session survived the stall.

import io.netty.buffer.AbstractByteBufAllocator;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelOption;
import io.rsocket.Payload;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketConnector;
import io.rsocket.core.RSocketServer;
import io.rsocket.core.Resume;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.netty.tcp.TcpClient;

import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

public class KeepAlivePositionRace {

    public static void main(String[] args) {
        // Both sides' keepalive ticks run on Schedulers.parallel(), whose workers are pinned per task. Use a pool
        // instead, so the stalled client tick cannot also hold back the server's tick on a machine with few cores.
        Schedulers.setFactory(new Schedulers.Factory() {
            @Override
            public Scheduler newParallel(int parallelism, ThreadFactory threadFactory) {
                return Schedulers.fromExecutorService(Executors.newScheduledThreadPool(8, threadFactory));
            }
        });

        var server = RSocketServer.create(SocketAcceptor.forRequestChannel(Flux::from))
                .resume(new Resume().cleanupStoreOnKeepAlive())
                .bindNow(TcpServerTransport.create("localhost", 0));

        var client = RSocketConnector.create()
                .keepAlive(Duration.ofMillis(100), Duration.ofSeconds(30))
                .resume(new Resume().cleanupStoreOnKeepAlive())
                .connect(TcpClientTransport.create(TcpClient.create()
                        .host("localhost").port(server.address().getPort())
                        .option(ChannelOption.ALLOCATOR, new StallFirstKeepAliveTick())))
                .block();

        try {
            client.requestChannel(Flux.interval(Duration.ofMillis(5)).onBackpressureDrop()
                            .map(i -> DefaultPayload.create("ping-" + i)))
                    .doOnNext(Payload::release)
                    .take(Duration.ofSeconds(3))
                    .blockLast();
            System.out.println("Session survived the stall.");
        } catch (RuntimeException e) {
            System.out.println("Session closed: " + e);
        }
        System.exit(0);
    }

    /**
     * ClientKeepAliveSupport#onIntervalTick reads the implied position, then allocates the KEEPALIVE frame, then
     * enqueues it. Sleeping in that allocation stands in for the tick thread being descheduled in between; meanwhile
     * the client's reply to the server's KEEPALIVE goes out with a higher position.
     */
    static final class StallFirstKeepAliveTick extends AbstractByteBufAllocator {
        private final AtomicBoolean stalled = new AtomicBoolean();

        @Override
        protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
            stallIfFirstKeepAliveTick();
            return Unpooled.buffer(initialCapacity, maxCapacity);
        }

        @Override
        protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
            stallIfFirstKeepAliveTick();
            return Unpooled.directBuffer(initialCapacity, maxCapacity);
        }

        @Override
        public boolean isDirectBufferPooled() {
            return false;
        }

        private void stallIfFirstKeepAliveTick() {
            if (this.stalled.get() || !StackWalker.getInstance().walk(frames ->
                    frames.anyMatch(f -> f.getMethodName().equals("onIntervalTick")))) {
                return;
            }
            if (this.stalled.compareAndSet(false, true)) {
                try {
                    Thread.sleep(300);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

Possible solution

Either of these would fix it:

1. Tolerate stale positions: have InMemoryResumableFramesStore#releaseFrames ignore a position lower than the last one received instead of throwing. This is the workaround we use now, by subclassing the store:

   @Override
   public synchronized void releaseFrames(long remoteImpliedPos) {
       if (remoteImpliedPos < this.highestRemoteImpliedPosition) {
           return;
       }
       this.highestRemoteImpliedPosition = remoteImpliedPos;
       super.releaseFrames(remoteImpliedPos);
   }

2. Keep the sender in order: read the implied position when the KEEPALIVE frame is written, not before it is enqueued, so the two sending threads cannot reorder positions.

Turning cleanupStoreOnKeepAlive() off avoids the exception. The store then only releases frames on resume, though, so on a long-lived session it fills up, drops its oldest frames, and a later resume fails.

Your environment

• RSocket version(s) used: 1.1.5 (master has the same KeepAliveSupport and InMemoryResumableFramesStore)
• Other relevant libraries versions: reactor-core 3.4.41 and netty 4.1.117 for the reproducer; also seen with reactor-core 3.8 / netty 4.2 through Spring Boot 4
• Platform: the reproducer was run on macOS 26 (arm64); an equivalent test fails the same way on Linux x86_64 and Windows CI runners; first seen with a client on Windows Server
• JVM version: 25 (reproducer compiled for release 17)