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

OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java

@@ -12,6 +12,7 @@
  * information: "Portions copyright [year] [name of copyright owner]".
  *
  * Copyright 2015-2016 ForgeRock AS.
+ * Portions Copyrighted 2026 3A Systems, LLC
  */
 package org.forgerock.openicf.framework;
 
@@ -22,6 +23,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
@@ -307,6 +309,10 @@ public void executeMessage(Runnable processMessage) {
         messageExecutor.execute(processMessage);
     }
 
+    public Executor getMessageExecutor() {
+        return messageExecutor;
+    }
+
     // ------ LocalConnectorFramework Implementation Start ------
 
     private final ConcurrentMap<ClassLoader, AsyncLocalConnectorInfoManager> localConnectorInfoManagerCache =

OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java

@@ -128,7 +128,12 @@ public void onMessage(WebSocketConnectionHolder socket, String data) {
     public void onMessage(final WebSocketConnectionHolder socket, final byte[] bytes) {
         logger.ok("{0} onMessage({1}:bytes)", loggerName(), bytes.length);
 
-        connectorFramework.executeMessage(new Runnable() {
+        // Dispatch messages of one socket in arrival order: an operation
+        // response must not overtake the handshake response on a freshly
+        // opened connection (the pre-handshake branches below would drop it).
+        // Operation requests are handed off to the shared pool inside
+        // processMessage, so operations of one socket still run concurrently.
+        socket.executeSerially(connectorFramework.getMessageExecutor(), new Runnable() {
             public void run() {
                 processMessage(socket, bytes);
             }
@@ -137,14 +142,11 @@ public void run() {
 
     public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes){
         try {
-            RemoteMessage message = RemoteMessage.parseFrom(bytes);
+            final RemoteMessage message = RemoteMessage.parseFrom(bytes);
 
             if (logger.isOk()) {
                 logger.ok("{0} onMessage({1})", loggerName(), message.toString());
             }
-            if (!isHandshakeMessage(message) && !isErrorResponse(message)) {
-                awaitHandshake(socket, message.getMessageId());
-            }
             if (message.hasRequest()) {
                 if (message.getRequest().hasHandshakeMessage()) {
                     if (isClient()) {
@@ -155,9 +157,18 @@ public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes)
                     }
                 } else if (socket.isHandHooked()) {
                     if (message.getRequest().hasOperationRequest()) {
-                        processOperationRequest(socket, message.getMessageId(), message
-                                .getRequest().getOperationRequest());
-
+                        // Execute on the shared pool: operations may run for a
+                        // long time and several of them multiplex over one
+                        // socket. Keeping them out of the serial dispatch
+                        // queue preserves their concurrency and keeps later
+                        // messages (e.g. CancelOpRequest) timely.
+                        connectorFramework.executeMessage(new Runnable() {
+                            @Override
+                            public void run() {
+                                processOperationRequest(socket, message.getMessageId(), message
+                                        .getRequest().getOperationRequest());
+                            }
+                        });
                     } else if (message.getRequest().hasCancelOpRequest()) {
                         processCancelOpRequest(socket, message.getMessageId(), message.getRequest()
                                 .getCancelOpRequest());
@@ -204,47 +215,6 @@ public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes)
         }
     }
 
-    private static boolean isHandshakeMessage(final RemoteMessage message) {
-        return (message.hasRequest() && message.getRequest().hasHandshakeMessage())
-                || (message.hasResponse() && message.getResponse().hasHandshakeMessage());
-    }
-
-    /**
-     * Error responses are handled before the isHandHooked() guards below, so
-     * they never need to wait for the handshake.
-     */
-    private static boolean isErrorResponse(final RemoteMessage message) {
-        return message.hasResponse() && message.getResponse().hasError();
-    }
-
-    /**
-     * Messages are processed on a thread pool, so a message received on a
-     * freshly opened connection can overtake the handshake that arrived just
-     * before it and would be dropped by the pre-handshake branches below. The
-     * handshake is being processed concurrently and normally lands within
-     * milliseconds, so briefly wait for it instead of discarding the message.
-     */
-    private static final long HANDSHAKE_WAIT_MS = 500;
-
-    private void awaitHandshake(final WebSocketConnectionHolder socket, long messageId) {
-        if (socket.isHandHooked()) {
-            return;
-        }
-        final long deadline = System.currentTimeMillis() + HANDSHAKE_WAIT_MS;
-        while (!socket.isHandHooked() && System.currentTimeMillis() < deadline) {
-            try {
-                Thread.sleep(10);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                return;
-            }
-        }
-        if (!socket.isHandHooked()) {
-            logger.warn("{0} handshake still pending after wait, message('{1}') via socket:{2} ",
-                    loggerName(), messageId, socket.hashCode());
-        }
-    }
-
     protected void handleRemoteMessage(final WebSocketConnectionHolder socket,
             final RemoteMessage message) {
         logger.warn("{0} received unknown message('{1}') via socket:{2} ", loggerName(), message.getMessageId(),

OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java

@@ -29,10 +29,13 @@
 import java.io.Closeable;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage;
 import org.forgerock.openicf.common.rpc.RemoteConnectionHolder;
 import org.forgerock.openicf.framework.CloseListener;
+import org.identityconnectors.common.logging.Log;
 
 import com.google.protobuf.MessageLite;
 
@@ -41,9 +44,72 @@ public abstract class WebSocketConnectionHolder
         Closeable,
         RemoteConnectionHolder<WebSocketConnectionGroup, WebSocketConnectionHolder, RemoteOperationContext> {
 
+    private static final Log logger = Log.getLog(WebSocketConnectionHolder.class);
+
     protected final Queue<CloseListener<WebSocketConnectionHolder>> listeners =
             new ConcurrentLinkedQueue<CloseListener<WebSocketConnectionHolder>>();
 
+    private final Queue<Runnable> dispatchQueue = new ConcurrentLinkedQueue<Runnable>();
+    private final AtomicBoolean dispatchScheduled = new AtomicBoolean(false);
+
+    /**
+     * Runs {@code task} on {@code executor} after every task previously
+     * submitted to this holder has completed, preserving the submission order.
+     * At most one task of this holder is in flight at any time, so messages of
+     * one socket are processed in arrival order while different sockets still
+     * run concurrently on the shared executor.
+     * <p>
+     * Tasks must not block waiting for a later message of the same socket:
+     * that message cannot be dispatched until the current task returns.
+     * <p>
+     * Arrival order is preserved only because the WebSocket container invokes
+     * the per-connection message callbacks sequentially (both Jetty and
+     * Grizzly do): this method keeps the order of its own calls, so
+     * concurrent callers would enqueue in an undefined order.
+     */
+    public void executeSerially(final Executor executor, final Runnable task) {
+        dispatchQueue.add(task);
+        scheduleDispatch(executor);
+    }
+
+    private void scheduleDispatch(final Executor executor) {
+        if (dispatchScheduled.compareAndSet(false, true)) {
+            try {
+                executor.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        drainDispatchQueue(executor);
+                    }
+                });
+            } catch (RuntimeException e) {
+                // Typically RejectedExecutionException on shutdown - allow a
+                // later submission to try again instead of wedging the queue.
+                dispatchScheduled.set(false);
+                throw e;
+            }
+        }
+    }
+
+    private void drainDispatchQueue(final Executor executor) {
+        try {
+            Runnable task;
+            while ((task = dispatchQueue.poll()) != null) {
+                try {
+                    task.run();
+                } catch (Throwable t) {
+                    logger.warn(t, "Failed to process a dispatched message");
+                }
+            }
+        } finally {
+            dispatchScheduled.set(false);
+            // A task may have been added after poll() returned null but before
+            // the flag was cleared; whoever wins the CAS reschedules the drain.
+            if (!dispatchQueue.isEmpty()) {
+                scheduleDispatch(executor);
+            }
+        }
+    }
+
     public boolean receiveHandshake(HandshakeMessage message) {
         if (null == getRemoteConnectionContext()) {
             handshake(message);