[#109] Serialize WebSocket message dispatch per socket to prevent out-of-order processing

[vharseko]

Fixes #109. Follow-up to #107/#108.

Problem

OpenICFServerAdapter.onMessage handed every incoming WebSocket message to a shared unbounded cached thread pool with no per-socket ordering, so two messages delivered in order on one socket could be processed out of order. #108 papered over the most visible instance (an operation response overtaking the handshake response on a fresh socket) with a 500 ms awaitHandshake poll.

Change

• WebSocketConnectionHolder.executeSerially: per-socket serial dispatch queue draining onto the shared pool — at most one in-flight task per socket, messages processed strictly in arrival order, different sockets still run concurrently. Lives on the holder, so its lifecycle is the socket's. Arrival ordering relies on the container invoking per-connection callbacks sequentially (both Jetty and Grizzly do) — now stated in the javadoc.
• OpenICFServerAdapter: dispatches via the serial queue; OperationRequest execution is re-submitted to the shared pool after ordered classification, so long-running operations keep their concurrency over one socket and CancelOpRequest stays timely. The awaitHandshake poll and its helpers are removed.
• Jetty SinglePrincipal: outgoing frames went through a 10-thread pool, so concurrent blocking sends could reach the wire in any order (a handshake response could be overtaken before ever reaching the network). Jetty's RemoteEndpoint itself is thread-safe — the defect was reordering, not thread-unsafety. Replaced with a single named daemon sender thread. SinglePrincipal is cached per principal name by OpenICFWebSocketCreator and serves every connection of that name, so the executor is a per-principal resource: it is shut down in doClose() (the principal's lifecycle end), not in the per-connection onWebSocketClose — and the pool being replaced accordingly leaked 10 threads per principal, not per connection.

The responseQueue re-sorting in BatchApiOpImpl and the ResultBuffer sequence handling stay: RemoteConnectionGroup.trySendMessage round-robins across a group's sockets, so cross-socket ordering is never guaranteed at the protocol level.

Review fixes (thanks @maximthomas)

• sendExecutor.shutdown() moved from onWebSocketClose (per-connection) to doClose() (per-principal): with the cached principal, the first disconnect used to kill the executor and permanently disable sending for every later connection of that name.
• Sender thread now comes from Utils.newThreadFactory (named, daemon) — nothing closes cached principals in the servlet lifecycle, so a non-daemon thread would outlive servlet destroy().
• Corrected the RemoteEndpoint thread-safety claim and the leak scope in the description and code comments.
• executeSerially javadoc states the sequential-callback assumption; the test comment now credits visibility to the release/acquire chain through dispatchScheduled, not to non-overlap alone.
• Added the RejectedExecutionException-path test and the reconnect repro test; added the two @Overrides flagged by CodeQL.

Tests

• WebSocketConnectionHolderTest: 1000-task submission order and no-overlap check, cross-socket independence (a blocked socket does not stall others), queue survival after a throwing task, and a rejected dispatch not wedging the queue.
• ReconnectSendTest (jetty): a send after close + reconnect on the same cached SinglePrincipal must still reach the wire.
• connector-framework-server 29/29, connector-server-grizzly 32/32, connector-server-jetty 31/31; OpenICFWebSocketTest run three extra times — stable.

[maximthomas]

The inbound half (executeSerially + OpenICFServerAdapter) looks correct — I walked the CAS interleavings and found no defect, and the tests are well-targeted. One blocker in the Jetty half.

Blocker: sendExecutor.shutdown() bricks the server after the first disconnect

OpenICFWebSocketCreator caches one SinglePrincipal per principal name and reuses it for every connection of that name (OpenICFWebSocketCreator.java:129-136). With the default anonymous authenticator that's a single instance for the whole server.

sendExecutor is a field of that instance, but onWebSocketClose is a per-connection callback. So:

1. Client #1 disconnects → sendExecutor.shutdown()
2. Client #2 connects → same cached principal, dead executor
3. Every send → RejectedExecutionException → silently masked by the new catch as ConnectorIOException("Socket is not connected.")

Sending is permanently disabled for all subsequent clients until restart.

Verified by running connect → close → reconnect against the actual code:

Tree	Result
master abc8e325	PASS
this PR 3ddeae15	FAIL — ConnectorIOException: Socket is not connected.
this PR + fix below	PASS (jetty suite 31/31)

No existing test in connector-server-jetty/src/test closes and reconnects, which is why CI is green. Repro test attached below.

Fix — shut down at the principal's real lifecycle end instead of per connection:

// onWebSocketClose: drop the sendExecutor.shutdown() line
protected void doClose() {
    sendExecutor.shutdown();
}

ConnectionPrincipal.close() calls doClose() under an isRunning CAS, so it fires exactly once when the principal actually goes away.

Two corrections to the description

• Jetty's RemoteEndpoint is not non-thread-safe. In 11.0.25 sendBlocking uses a per-call FutureCallback into a thread-safe FIFO FrameFlusher. Concurrent sends were unordered, never unsafe. The reordering argument alone justifies the change — the fix is right, the stated reason isn't.
• The old pool leaked 10 threads per principal, not per connection (same caching as above), so the leak being fixed is much smaller than described.

Minor

• executeSerially relies on the container delivering onWebSocketBinary sequentially per connection. True for Jetty and Grizzly, but load-bearing and unstated — worth a line in the javadoc.
• The test comment says the unsynchronized ArrayList is safe because tasks never overlap. Non-overlap alone doesn't give visibility; it works via the release/acquire chain through dispatchScheduled. Worth correcting so a future refactor doesn't silently break the premise.
• Nothing covers the RejectedExecutionException path in scheduleDispatch — the trickiest state in the new code, and a wedged queue if the flag reset were ever dropped. Cheap test: reject once, then submit with a live executor and assert both tasks drain in order.
• Executors.newSingleThreadExecutor() gives an unnamed non-daemon pool-N-thread-1; Utils.newThreadFactory is already on the classpath here.
• Pre-existing, not from this PR: CancelOpRequest can still arrive before the operation registers itself (registration happens in LocalRequest's constructor, on the pool thread). Worth a follow-up.

Repro test

package org.forgerock.openicf.framework.server.jetty;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.eclipse.jetty.websocket.api.RemoteEndpoint;
import org.eclipse.jetty.websocket.api.Session;
import org.forgerock.openicf.framework.remote.rpc.OperationMessageListener;
import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionGroup;
import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionHolder;
import org.testng.Assert;
import org.testng.annotations.Test;

/**
 * Does a SinglePrincipal that has seen one connection closed still send on a
 * subsequent connection? OpenICFWebSocketCreator caches SinglePrincipal per
 * principal name, so the same instance serves every connection of that name.
 */
public class ReconnectSendTest {

    private static final AtomicInteger SENT = new AtomicInteger();

    private static Session newSession() {
        final RemoteEndpoint remote = (RemoteEndpoint) Proxy.newProxyInstance(
                ReconnectSendTest.class.getClassLoader(),
                new Class<?>[] { RemoteEndpoint.class },
                new InvocationHandler() {
                    public Object invoke(Object p, Method m, Object[] a) {
                        if ("sendBytes".equals(m.getName())) {
                            SENT.incrementAndGet();
                        }
                        return null;
                    }
                });
        return (Session) Proxy.newProxyInstance(
                ReconnectSendTest.class.getClassLoader(),
                new Class<?>[] { Session.class },
                new InvocationHandler() {
                    public Object invoke(Object p, Method m, Object[] a) {
                        if ("isOpen".equals(m.getName())) {
                            return Boolean.TRUE;
                        }
                        if ("getRemote".equals(m.getName())) {
                            return remote;
                        }
                        return null;
                    }
                });
    }

    @Test(timeOut = 30000)
    public void testSendWorksAfterReconnectOnCachedPrincipal() throws Exception {
        final AtomicReference<WebSocketConnectionHolder> holder =
                new AtomicReference<WebSocketConnectionHolder>();

        OperationMessageListener capturing = (OperationMessageListener) Proxy.newProxyInstance(
                ReconnectSendTest.class.getClassLoader(),
                new Class<?>[] { OperationMessageListener.class },
                new InvocationHandler() {
                    public Object invoke(Object p, Method m, Object[] a) {
                        if ("onConnect".equals(m.getName())) {
                            holder.set((WebSocketConnectionHolder) a[0]);
                        }
                        return null;
                    }
                });

        SinglePrincipal principal = new SinglePrincipal("anonymous", capturing, null,
                new ConcurrentHashMap<String, WebSocketConnectionGroup>());

        // --- connection #1 ---
        principal.onWebSocketConnect(newSession());
        Future<?> first = holder.get().sendBytes(new byte[] { 1, 2, 3 });
        first.get(5, TimeUnit.SECONDS);
        Assert.assertEquals(SENT.get(), 1, "first connection should have sent");

        principal.onWebSocketClose(1000, "client went away");

        // --- connection #2 on the SAME cached principal instance ---
        principal.onWebSocketConnect(newSession());
        Future<?> second = holder.get().sendBytes(new byte[] { 4, 5, 6 });
        second.get(5, TimeUnit.SECONDS);

        Assert.assertEquals(SENT.get(), 2,
                "send after reconnect must reach the wire");
    }
}