From 3ddeae1554f14555ccbd1e7ddf173fadab60cb45 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 20 Jul 2026 15:41:12 +0300 Subject: [PATCH 1/2] Serialize WebSocket message dispatch per socket (#109) Messages of one socket were handed to a shared unbounded pool with no ordering, so a message could be processed before the handshake that arrived just ahead of it. Replace the interim awaitHandshake wait with a per-socket serial dispatch queue draining onto the shared pool: messages of one socket are processed in arrival order, different sockets still run concurrently. OperationRequest execution is re-submitted to the pool so long-running operations keep their concurrency and CancelOpRequest stays timely. Also serialize outgoing frames in the Jetty connector server: sends went through a 10-thread pool per socket, which could reorder frames on the wire and leaked the pool on close. --- .../openicf/framework/ConnectorFramework.java | 6 + .../remote/OpenICFServerAdapter.java | 67 ++----- .../remote/rpc/WebSocketConnectionHolder.java | 60 ++++++ .../rpc/WebSocketConnectionHolderTest.java | 176 ++++++++++++++++++ .../server/jetty/SinglePrincipal.java | 50 +++-- 5 files changed, 294 insertions(+), 65 deletions(-) create mode 100644 OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java index 1878cbf2..5186edf3 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/ConnectorFramework.java +++ b/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 localConnectorInfoManagerCache = diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java index 9c3d9521..23b1dabd 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java +++ b/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,17 @@ 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() { + public void run() { + processOperationRequest(socket, message.getMessageId(), message + .getRequest().getOperationRequest()); + } + }); } else if (message.getRequest().hasCancelOpRequest()) { processCancelOpRequest(socket, message.getMessageId(), message.getRequest() .getCancelOpRequest()); @@ -204,47 +214,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(), diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java index 3a678554..5991337d 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java +++ b/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,66 @@ public abstract class WebSocketConnectionHolder Closeable, RemoteConnectionHolder { + private static final Log logger = Log.getLog(WebSocketConnectionHolder.class); + protected final Queue> listeners = new ConcurrentLinkedQueue>(); + private final Queue dispatchQueue = new ConcurrentLinkedQueue(); + 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. + *

+ * Tasks must not block waiting for a later message of the same socket: + * that message cannot be dispatched until the current task returns. + */ + 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() { + 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); diff --git a/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java new file mode 100644 index 00000000..e938f468 --- /dev/null +++ b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java @@ -0,0 +1,176 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.forgerock.openicf.framework.remote.rpc; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.Test; + +/** + * Tests the per-socket serial dispatch queue of + * {@link WebSocketConnectionHolder#executeSerially}: tasks of one holder run + * in submission order without overlapping, tasks of different holders still + * run concurrently, and a failing task does not stall the queue. + */ +public class WebSocketConnectionHolderTest { + + private final ExecutorService pool = Executors.newFixedThreadPool(8); + + @AfterClass + public void shutdownPool() { + pool.shutdownNow(); + } + + private static WebSocketConnectionHolder newHolder() { + return new WebSocketConnectionHolder() { + + protected void handshake(HandshakeMessage message) { + } + + protected void tryClose() { + } + + public boolean isOperational() { + return true; + } + + public RemoteOperationContext getRemoteConnectionContext() { + return null; + } + + public Future sendBytes(byte[] data) { + return CompletableFuture.completedFuture(null); + } + + public Future sendString(String data) { + return CompletableFuture.completedFuture(null); + } + + public void sendPing(byte[] applicationData) throws Exception { + } + + public void sendPong(byte[] applicationData) throws Exception { + } + }; + } + + @Test(timeOut = 30000) + public void testTasksRunInSubmissionOrderWithoutOverlap() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final int taskCount = 1000; + final List executed = new ArrayList(taskCount); + final AtomicInteger inFlight = new AtomicInteger(0); + final AtomicInteger maxInFlight = new AtomicInteger(0); + final CountDownLatch done = new CountDownLatch(taskCount); + + for (int i = 0; i < taskCount; i++) { + final int id = i; + holder.executeSerially(pool, new Runnable() { + public void run() { + int current = inFlight.incrementAndGet(); + maxInFlight.accumulateAndGet(current, Math::max); + // executed is not synchronized on purpose: the serial + // queue must never run two tasks at the same time. + executed.add(id); + inFlight.decrementAndGet(); + done.countDown(); + } + }); + } + + Assert.assertTrue(done.await(20, TimeUnit.SECONDS), "tasks did not finish"); + Assert.assertEquals(maxInFlight.get(), 1, "tasks of one socket overlapped"); + Assert.assertEquals(executed.size(), taskCount); + for (int i = 0; i < taskCount; i++) { + Assert.assertEquals(executed.get(i).intValue(), i, "task order broken at " + i); + } + } + + @Test(timeOut = 30000) + public void testBlockedHolderDoesNotBlockOtherHolder() throws Exception { + final WebSocketConnectionHolder blocked = newHolder(); + final WebSocketConnectionHolder free = newHolder(); + final CountDownLatch blockedStarted = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CountDownLatch freeDone = new CountDownLatch(1); + final CountDownLatch blockedDone = new CountDownLatch(1); + + blocked.executeSerially(pool, new Runnable() { + public void run() { + blockedStarted.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }); + blocked.executeSerially(pool, new Runnable() { + public void run() { + blockedDone.countDown(); + } + }); + + Assert.assertTrue(blockedStarted.await(10, TimeUnit.SECONDS)); + + free.executeSerially(pool, new Runnable() { + public void run() { + freeDone.countDown(); + } + }); + + // The other socket keeps working while the first one is busy... + Assert.assertTrue(freeDone.await(10, TimeUnit.SECONDS), + "an unrelated socket was blocked by a busy one"); + // ...and the second task of the busy socket is still pending. + Assert.assertEquals(blockedDone.getCount(), 1, + "second task of the busy socket must wait for the first"); + + release.countDown(); + Assert.assertTrue(blockedDone.await(10, TimeUnit.SECONDS)); + } + + @Test(timeOut = 30000) + public void testFailingTaskDoesNotStallTheQueue() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final CountDownLatch survived = new CountDownLatch(1); + + holder.executeSerially(pool, new Runnable() { + public void run() { + throw new RuntimeException("boom"); + } + }); + holder.executeSerially(pool, new Runnable() { + public void run() { + survived.countDown(); + } + }); + + Assert.assertTrue(survived.await(10, TimeUnit.SECONDS), + "a failing task stalled the serial queue"); + } +} diff --git a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java index 79c968be..4abad4d4 100644 --- a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java +++ b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java @@ -44,6 +44,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; public class SinglePrincipal extends ConnectionPrincipal implements WebSocketPingPongListener, WebSocketListener, WebSocketFrameListener { @@ -102,6 +103,7 @@ public void onWebSocketClose(int statusCode, String reason) { return; } hasCloseBeenCalled = true; + sendExecutor.shutdown(); getConnectionPrincipal().getOperationMessageListener().onClose(adapter, statusCode, reason); } @@ -149,6 +151,12 @@ public void onWebSocketFrame(Frame frame) { // threads via getRemoteConnectionContext()/isHandHooked(). private volatile RemoteOperationContext context = null; + // Single thread per socket: frames must leave in submission order (the + // peer drops e.g. an operation response that overtakes the handshake + // response), and Jetty's RemoteEndpoint does not support concurrent + // blocking sends. Shut down in onWebSocketClose. + private final ExecutorService sendExecutor = Executors.newSingleThreadExecutor(); + private final WebSocketConnectionHolder adapter = new WebSocketConnectionHolder() { protected void handshake(RPCMessages.HandshakeMessage message) { @@ -163,17 +171,22 @@ public RemoteOperationContext getRemoteConnectionContext() { return context; } - private final ExecutorService executorService = Executors.newFixedThreadPool(10); - public Future sendBytes(byte[] data) { if (isOperational()) { - return executorService.submit(() -> { - try { - getSession().getRemote().sendBytes(ByteBuffer.wrap(data)); - } catch (IOException e) { - throw new RuntimeException(e); - } - }); + try { + return sendExecutor.submit(() -> { + try { + getSession().getRemote().sendBytes(ByteBuffer.wrap(data)); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } catch (RejectedExecutionException e) { + // Socket closed between the isOperational() check and the + // submit - the executor is already shut down. + return Promises.newExceptionPromise(new ConnectorIOException( + "Socket is not connected.")); + } } else { return Promises.newExceptionPromise(new ConnectorIOException( "Socket is not connected.")); @@ -182,13 +195,18 @@ public Future sendBytes(byte[] data) { public Future sendString(String data) { if (isOperational()) { - return executorService.submit(() -> { - try { - getSession().getRemote().sendString(data); - } catch (IOException e) { - throw new RuntimeException(e); - } - }); + try { + return sendExecutor.submit(() -> { + try { + getSession().getRemote().sendString(data); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } catch (RejectedExecutionException e) { + return Promises.newExceptionPromise(new ConnectorIOException( + "Socket is not connected.")); + } } else { return Promises.newExceptionPromise(new ConnectorIOException( "Socket is not connected.")); From b615be655a9b426df8ea8154ab61e800821a65ce Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 20 Jul 2026 18:18:17 +0300 Subject: [PATCH 2/2] Fix send executor lifecycle and address #110 review findings SinglePrincipal is cached per principal name and serves every connection of that name, but sendExecutor.shutdown() ran in the per-connection onWebSocketClose: the first disconnect killed the executor and every send on a later connection of the same principal failed as "Socket is not connected." until restart. Shut the executor down in doClose() instead and create its thread named and daemon via Utils.newThreadFactory, since the servlet lifecycle never closes cached principals. Also from the review: correct the RemoteEndpoint comment (it is thread-safe; the defect was frame reordering), document the sequential-callback assumption of executeSerially, credit the test's unsynchronized list to the release/acquire chain through dispatchScheduled, cover the RejectedExecutionException path, add a reconnect repro test, and add two @Override annotations flagged by CodeQL. --- .../remote/OpenICFServerAdapter.java | 1 + .../remote/rpc/WebSocketConnectionHolder.java | 6 + .../rpc/WebSocketConnectionHolderTest.java | 56 ++++++++- .../server/jetty/SinglePrincipal.java | 23 ++-- .../server/jetty/ReconnectSendTest.java | 108 ++++++++++++++++++ 5 files changed, 182 insertions(+), 12 deletions(-) create mode 100644 OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java index 23b1dabd..4a6924fd 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/OpenICFServerAdapter.java @@ -163,6 +163,7 @@ public void processMessage(final WebSocketConnectionHolder socket, byte[] bytes) // 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()); diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java index 5991337d..4ea78edd 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolder.java @@ -61,6 +61,11 @@ public abstract class WebSocketConnectionHolder *

* Tasks must not block waiting for a later message of the same socket: * that message cannot be dispatched until the current task returns. + *

+ * 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); @@ -71,6 +76,7 @@ private void scheduleDispatch(final Executor executor) { if (dispatchScheduled.compareAndSet(false, true)) { try { executor.execute(new Runnable() { + @Override public void run() { drainDispatchQueue(executor); } diff --git a/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java index e938f468..b5c6df4d 100644 --- a/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java +++ b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionHolderTest.java @@ -16,13 +16,17 @@ package org.forgerock.openicf.framework.remote.rpc; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage; @@ -34,7 +38,8 @@ * Tests the per-socket serial dispatch queue of * {@link WebSocketConnectionHolder#executeSerially}: tasks of one holder run * in submission order without overlapping, tasks of different holders still - * run concurrently, and a failing task does not stall the queue. + * run concurrently, a failing task does not stall the queue, and a rejected + * dispatch does not wedge it. */ public class WebSocketConnectionHolderTest { @@ -93,8 +98,12 @@ public void testTasksRunInSubmissionOrderWithoutOverlap() throws Exception { public void run() { int current = inFlight.incrementAndGet(); maxInFlight.accumulateAndGet(current, Math::max); - // executed is not synchronized on purpose: the serial - // queue must never run two tasks at the same time. + // executed is not synchronized on purpose: tasks never + // overlap, and each task's writes are published to the + // next one by the release/acquire chain through + // dispatchScheduled (set(false) after a drain, winning + // CAS before the next) - non-overlap alone would not + // guarantee visibility. executed.add(id); inFlight.decrementAndGet(); done.countDown(); @@ -173,4 +182,45 @@ public void run() { Assert.assertTrue(survived.await(10, TimeUnit.SECONDS), "a failing task stalled the serial queue"); } + + @Test(timeOut = 30000) + public void testRejectedDispatchDoesNotWedgeTheQueue() throws Exception { + final WebSocketConnectionHolder holder = newHolder(); + final AtomicBoolean rejectNext = new AtomicBoolean(true); + final Executor rejectingOnce = new Executor() { + public void execute(Runnable command) { + if (rejectNext.getAndSet(false)) { + throw new RejectedExecutionException("simulated shutdown"); + } + pool.execute(command); + } + }; + final List executed = new ArrayList(); + final CountDownLatch done = new CountDownLatch(2); + + try { + holder.executeSerially(rejectingOnce, recordingTask(1, executed, done)); + Assert.fail("the rejection must reach the caller"); + } catch (RejectedExecutionException expected) { + // The task stays queued; the dispatchScheduled flag must have + // been reset so a later submission can reschedule the drain. + } + + holder.executeSerially(rejectingOnce, recordingTask(2, executed, done)); + + Assert.assertTrue(done.await(10, TimeUnit.SECONDS), + "a rejected dispatch wedged the serial queue"); + Assert.assertEquals(executed, Arrays.asList(1, 2), + "both tasks must drain in submission order"); + } + + private static Runnable recordingTask(final int id, final List executed, + final CountDownLatch done) { + return new Runnable() { + public void run() { + executed.add(id); + done.countDown(); + } + }; + } } diff --git a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java index 4abad4d4..ab66a707 100644 --- a/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java +++ b/OpenICF-java-framework/connector-server-jetty/src/main/java/org/forgerock/openicf/framework/server/jetty/SinglePrincipal.java @@ -35,6 +35,7 @@ import org.forgerock.openicf.framework.remote.rpc.RemoteOperationContext; import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionGroup; import org.forgerock.openicf.framework.remote.rpc.WebSocketConnectionHolder; +import org.forgerock.util.Utils; import org.forgerock.util.promise.Promises; import org.identityconnectors.framework.common.exceptions.ConnectorIOException; @@ -74,7 +75,7 @@ public RemoteOperationContext handshake( } protected void doClose() { - + sendExecutor.shutdown(); } @@ -103,7 +104,6 @@ public void onWebSocketClose(int statusCode, String reason) { return; } hasCloseBeenCalled = true; - sendExecutor.shutdown(); getConnectionPrincipal().getOperationMessageListener().onClose(adapter, statusCode, reason); } @@ -151,11 +151,16 @@ public void onWebSocketFrame(Frame frame) { // threads via getRemoteConnectionContext()/isHandHooked(). private volatile RemoteOperationContext context = null; - // Single thread per socket: frames must leave in submission order (the - // peer drops e.g. an operation response that overtakes the handshake - // response), and Jetty's RemoteEndpoint does not support concurrent - // blocking sends. Shut down in onWebSocketClose. - private final ExecutorService sendExecutor = Executors.newSingleThreadExecutor(); + // Single send thread per principal: frames must leave in submission order + // (the peer drops e.g. an operation response that overtakes the handshake + // response). Jetty's RemoteEndpoint is thread-safe, but concurrent + // blocking sends may reach the wire in any order. This instance is cached + // by OpenICFWebSocketCreator and serves every connection of this + // principal name, so the executor must survive onWebSocketClose; it is + // shut down in doClose(). The thread is a daemon because cached + // principals are not closed on servlet destroy. + private final ExecutorService sendExecutor = Executors.newSingleThreadExecutor( + Utils.newThreadFactory(null, "OpenICF Jetty WebSocket Send %d", true)); private final WebSocketConnectionHolder adapter = new WebSocketConnectionHolder() { @@ -182,8 +187,8 @@ public Future sendBytes(byte[] data) { } }); } catch (RejectedExecutionException e) { - // Socket closed between the isOperational() check and the - // submit - the executor is already shut down. + // The principal was closed and doClose() shut the + // executor down. return Promises.newExceptionPromise(new ConnectorIOException( "Socket is not connected.")); } diff --git a/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java b/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java new file mode 100644 index 00000000..69769286 --- /dev/null +++ b/OpenICF-java-framework/connector-server-jetty/src/test/java/org/forgerock/openicf/framework/server/jetty/ReconnectSendTest.java @@ -0,0 +1,108 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +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 holder = + new AtomicReference(); + + 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()); + + // --- 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"); + } +}