From f0397c750595b679a27d073ef622033a9d599e8e Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 17 Jul 2026 19:13:03 +0300 Subject: [PATCH 1/2] Fix WebSocket client wedging that made OpenICFWebSocketTest flaky (#102) - ClientRemoteConnectorInfoManager: always release the connection permit in onClosed - the inverted guard leaked the last permit of a single-connection client, leaving it unable to reconnect forever; self-heal a permit lost by a connect attempt that failed before the Grizzly Connection was created - WebSocketConnectionGroup/Holder: send the initial CONNECTOR_INFO request only after the holder has stored its RemoteOperationContext, so an early response can no longer be dropped with an NPE; guard trySubmitRequest returning null - OpenICFWebSocketTest: replace the unreachable 5-minute poll loop with real 20-second polling attempts - grizzly/jetty poms: raise forkedProcessTimeoutInSeconds to 900 so a single hung test fails itself instead of killing the whole fork --- .../ClientRemoteConnectorInfoManager.java | 28 ++++++-- .../remote/rpc/WebSocketConnectionGroup.java | 68 +++++++++++++------ .../remote/rpc/WebSocketConnectionHolder.java | 10 +++ .../connector-server-grizzly/pom.xml | 4 +- .../server/OpenICFWebSocketTest.java | 21 ++++-- .../connector-server-jetty/pom.xml | 4 +- 6 files changed, 98 insertions(+), 37 deletions(-) diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java index abb785683..749c64013 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework.client; @@ -172,12 +174,13 @@ protected void preConfigure(final Connection conn) { conn.addCloseListener(new CloseListener() { public void onClosed(Closeable closeable, ICloseType type) throws IOException { logger.ok("DEBUG = Connection Closed {0}", type); - - if (availableConnectionPermitCount.get()<1) { - logger.ok("Destroy onClosed. Remaining permits: {0}", availableConnectionPermitCount.get()); - return; - } - + + // Always give back the permit held by this connection. + // Skipping the release when no permit is available + // leaks the last permit of a single-connection client + // and leaves it unable to reconnect forever. + // tryReleaseConnectionPermit() is capped at + // permittedConnectionCount so it can not over-credit. tryReleaseConnectionPermit(); if (!socket.connectPromise.isDone()) { @@ -203,6 +206,17 @@ public void run() { ws.close(); } } + // Self-heal: a connect attempt that failed before Grizzly + // created the Connection has no close listener to release its + // permit. If there is no live connection, no free permit and + // no recent connect attempt, restore the lost permit so the + // client can reconnect instead of staying wedged forever. + if (isRunning.get() && privateConnections.isEmpty() && !hasFreeConnectionPermit() + && System.currentTimeMillis() - lastConnectAttempt.get() > 30000) { + logger.ok("Restoring connection permit lost by a failed connect attempt - {0}", + getName()); + tryReleaseConnectionPermit(); + } while (isRunning.get() && null != connect(false)) { logger.ok("New connection is created - {0}. Remaining permits:{1}", getName(), availableConnectionPermitCount.get()); @@ -322,6 +336,7 @@ public boolean isOperational() { // ---- private final AtomicLong lastConnectithenOnException = new AtomicLong(0L); + private final AtomicLong lastConnectAttempt = new AtomicLong(0L); public Promise connect() { return connect(true); @@ -367,6 +382,7 @@ public void updated(Connection result) { // Successfully acquired one permit if (index > -1) { if (isRunning.get()) { + lastConnectAttempt.set(System.currentTimeMillis()); if (connectionInfo.getLocalAddress() != null) { connectionHandler.connect(remoteAddress, new InetSocketAddress(connectionInfo .getLocalAddress(), 0), callback); diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java index c6722b16b..2bb87872f 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.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.remote.rpc; @@ -90,6 +91,7 @@ public void onClosed(final WebSocketConnectionHolder connection) { }; private boolean receivedConnectorInfo = false; + private final AtomicBoolean initialConnectorInfoRequest = new AtomicBoolean(false); private final RemoteConnectorInfoManager delegate = new RemoteConnectorInfoManager(); public WebSocketConnectionGroup(final String remoteSessionId) { @@ -112,31 +114,57 @@ public RemoteOperationContext handshake(final Principal connectionPrincipal, webSocketConnection.listeners.add(closeListener); principals.add(connectionPrincipal.getName()); if (webSockets.indexOf(entry) == 0) { - ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); - requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); - final Function success = new Function() { - @Override - public Boolean apply(Boolean value) throws RuntimeException { - receivedConnectorInfo = value; - return receivedConnectorInfo; - } - }; - - trySubmitRequest(requestFactory).getPromise().then(success, new Function() { - @Override - public Boolean apply(RuntimeException e) throws RuntimeException { - logger.ok("Resending initial 'CONNECTOR_INFO' request", e); - ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); - requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); - trySubmitRequest(requestFactory).getPromise().then(success); - return receivedConnectorInfo; - } - }); + // The initial 'CONNECTOR_INFO' request is deferred until + // handshakeComplete() - sending it from here races with the + // response: the holder assigns its RemoteOperationContext only + // after this method returns, so an early response would be + // dropped with a NullPointerException. + initialConnectorInfoRequest.set(true); } } return operationContext; } + /** + * Invoked by {@link WebSocketConnectionHolder#receiveHandshake} after the + * holder has stored the {@link RemoteOperationContext} returned by + * {@link #handshake(Principal, WebSocketConnectionHolder, HandshakeMessage)}, + * so responses to the requests sent below can always be dispatched. + */ + public void handshakeComplete() { + if (initialConnectorInfoRequest.compareAndSet(true, false)) { + final Function success = new Function() { + @Override + public Boolean apply(Boolean value) throws RuntimeException { + receivedConnectorInfo = value; + return receivedConnectorInfo; + } + }; + + ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); + requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); + ControlMessageRequest request = trySubmitRequest(requestFactory); + if (null == request) { + // No operational connection yet - checkIsActive() retries + // while receivedConnectorInfo is false. + return; + } + request.getPromise().then(success, new Function() { + @Override + public Boolean apply(RuntimeException e) throws RuntimeException { + logger.ok("Resending initial 'CONNECTOR_INFO' request", e); + ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); + requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); + ControlMessageRequest retry = trySubmitRequest(requestFactory); + if (null != retry) { + retry.getPromise().then(success); + } + return receivedConnectorInfo; + } + }); + } + } + public void principalIsShuttingDown(final Principal connectionPrincipal) { final String name = connectionPrincipal.getName(); if (principals.remove(name)) { 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 edbf7a3db..3a6785541 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 @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework.remote.rpc; @@ -45,6 +47,14 @@ public abstract class WebSocketConnectionHolder public boolean receiveHandshake(HandshakeMessage message) { if (null == getRemoteConnectionContext()) { handshake(message); + final RemoteOperationContext context = getRemoteConnectionContext(); + if (null != context) { + // The context is visible only from this point on, so the + // initial connector info exchange is started here instead of + // inside handshake(message) - a response arriving before the + // context is assigned would otherwise be dropped. + context.getRemoteConnectionGroup().handshakeComplete(); + } } return isHandHooked(); } diff --git a/OpenICF-java-framework/connector-server-grizzly/pom.xml b/OpenICF-java-framework/connector-server-grizzly/pom.xml index d28073e25..bb2dbf954 100644 --- a/OpenICF-java-framework/connector-server-grizzly/pom.xml +++ b/OpenICF-java-framework/connector-server-grizzly/pom.xml @@ -22,7 +22,7 @@ your own identifying information: "Portions Copyrighted [year] [name of copyright owner]" - Portions Copyrighted 2018-2025 3A Systems, LLC + Portions Copyrighted 2018-2026 3A Systems, LLC --> @@ -160,7 +160,7 @@ **/AsyncJavaPlainConnectorInfoManagerTest.java kill - 300 + 900 diff --git a/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/OpenICFWebSocketTest.java b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/OpenICFWebSocketTest.java index 52177f28f..2b15bbfc0 100644 --- a/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/OpenICFWebSocketTest.java +++ b/OpenICF-java-framework/connector-server-grizzly/src/test/java/org/forgerock/openicf/framework/server/OpenICFWebSocketTest.java @@ -20,6 +20,8 @@ * with the fields enclosed by brackets [] replaced by * your own identifying information: * "Portions Copyrighted [year] [name of copyright owner]" + * + * Portions Copyrighted 2026 3A Systems, LLC */ package org.forgerock.openicf.framework.server; @@ -37,6 +39,7 @@ import java.util.Arrays; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.forgerock.openicf.framework.ConnectorFramework; import org.forgerock.openicf.framework.ConnectorFrameworkFactory; @@ -204,13 +207,17 @@ public void testRemoteConnectorInfoManager() throws Exception { Assert.assertNotNull(manager); ConnectorInfo c = null; - for (int i = 0; (c = - manager.findConnectorInfoAsync(TEST_CONNECTOR_KEY).getOrThrowUninterruptibly(5, - TimeUnit.MINUTES)) == null - && i < 5; i++) { - // Wait until the connection is established and the connector - // info are transferred. - Thread.sleep(20000); + for (int i = 0; null == c && i < 5; i++) { + try { + // Wait until the connection is established and the + // connector info are transferred. Keep each attempt well + // below the Surefire fork timeout so a hang fails this + // test instead of killing the whole forked JVM. + c = manager.findConnectorInfoAsync(TEST_CONNECTOR_KEY) + .getOrThrowUninterruptibly(20, TimeUnit.SECONDS); + } catch (TimeoutException e) { + Reporter.log("Connector info not received yet, retrying", true); + } } Assert.assertNotNull(c); for (ConnectorInfo ci : manager.getConnectorInfos()) { diff --git a/OpenICF-java-framework/connector-server-jetty/pom.xml b/OpenICF-java-framework/connector-server-jetty/pom.xml index a0600682e..5f80d8f26 100644 --- a/OpenICF-java-framework/connector-server-jetty/pom.xml +++ b/OpenICF-java-framework/connector-server-jetty/pom.xml @@ -22,7 +22,7 @@ your own identifying information: "Portions Copyrighted [year] [name of copyright owner]" - Portions Copyrighted 2018-2025 3A Systems, LLC + Portions Copyrighted 2018-2026 3A Systems, LLC --> @@ -132,7 +132,7 @@ false kill 1 - 300 + 900 -Dorg.eclipse.jetty.LEVEL=DEBUG From 0fd90820f0e0c1f0592cebf69397a31995646909 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Sun, 19 Jul 2026 08:53:52 +0300 Subject: [PATCH 2/2] Address review: guard self-heal against over-crediting an in-flight connection - Heal only when the last connect attempt never created a Connection (lastConnectionCreated < lastConnectAttempt): if preConfigure ran, the close listener is registered and will release the permit itself, so healing would over-credit while the connection is pending its handshake. - Make receivedConnectorInfo volatile: written from transport callbacks, read by the scheduler in checkIsActive(). - Dedupe the CONNECTOR_INFO request factory into a helper. --- .../ClientRemoteConnectorInfoManager.java | 24 ++++++++++++++---- .../remote/rpc/WebSocketConnectionGroup.java | 25 ++++++++++--------- 2 files changed, 32 insertions(+), 17 deletions(-) diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java index 749c64013..598afff86 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/client/ClientRemoteConnectorInfoManager.java @@ -196,6 +196,12 @@ public void onClosed(Closeable closeable, ICloseType type) throws IOException { } } }); + + // Stamped after the close listener is registered: once this + // timestamp is >= lastConnectAttempt, the permit held by the + // last attempt is guaranteed to be released on close, so the + // heartbeat self-heal must not restore it. + lastConnectionCreated.set(System.currentTimeMillis()); } }; @@ -207,12 +213,19 @@ public void run() { } } // Self-heal: a connect attempt that failed before Grizzly - // created the Connection has no close listener to release its - // permit. If there is no live connection, no free permit and - // no recent connect attempt, restore the lost permit so the - // client can reconnect instead of staying wedged forever. + // created the Connection (openSocketChannel or + // obtainNIOConnection threw, e.g. fd exhaustion) has no close + // listener to release its permit. If there is no live + // connection, no free permit, no recent connect attempt and + // the last attempt never created a Connection + // (lastConnectionCreated < lastConnectAttempt), restore the + // lost permit so the client can reconnect instead of staying + // wedged forever. When a Connection exists its close listener + // releases the permit, so healing then would over-credit + // while the connection is still pending its handshake. if (isRunning.get() && privateConnections.isEmpty() && !hasFreeConnectionPermit() - && System.currentTimeMillis() - lastConnectAttempt.get() > 30000) { + && System.currentTimeMillis() - lastConnectAttempt.get() > 30000 + && lastConnectionCreated.get() < lastConnectAttempt.get()) { logger.ok("Restoring connection permit lost by a failed connect attempt - {0}", getName()); tryReleaseConnectionPermit(); @@ -337,6 +350,7 @@ public boolean isOperational() { private final AtomicLong lastConnectithenOnException = new AtomicLong(0L); private final AtomicLong lastConnectAttempt = new AtomicLong(0L); + private final AtomicLong lastConnectionCreated = new AtomicLong(0L); public Promise connect() { return connect(true); diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java index 2bb87872f..b5010290c 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroup.java @@ -90,7 +90,9 @@ public void onClosed(final WebSocketConnectionHolder connection) { } }; - private boolean receivedConnectorInfo = false; + // Written from transport callback threads, read by the scheduler in + // checkIsActive(). + private volatile boolean receivedConnectorInfo = false; private final AtomicBoolean initialConnectorInfoRequest = new AtomicBoolean(false); private final RemoteConnectorInfoManager delegate = new RemoteConnectorInfoManager(); @@ -141,9 +143,7 @@ public Boolean apply(Boolean value) throws RuntimeException { } }; - ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); - requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); - ControlMessageRequest request = trySubmitRequest(requestFactory); + ControlMessageRequest request = trySubmitRequest(connectorInfoRequestFactory()); if (null == request) { // No operational connection yet - checkIsActive() retries // while receivedConnectorInfo is false. @@ -153,9 +153,7 @@ public Boolean apply(Boolean value) throws RuntimeException { @Override public Boolean apply(RuntimeException e) throws RuntimeException { logger.ok("Resending initial 'CONNECTOR_INFO' request", e); - ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); - requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); - ControlMessageRequest retry = trySubmitRequest(requestFactory); + ControlMessageRequest retry = trySubmitRequest(connectorInfoRequestFactory()); if (null != retry) { retry.getPromise().then(success); } @@ -165,6 +163,12 @@ public Boolean apply(RuntimeException e) throws RuntimeException { } } + private static ControlMessageRequestFactory connectorInfoRequestFactory() { + final ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); + requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); + return requestFactory; + } + public void principalIsShuttingDown(final Principal connectionPrincipal) { final String name = connectionPrincipal.getName(); if (principals.remove(name)) { @@ -350,11 +354,8 @@ public boolean checkIsActive() { } if (operational) { - ControlMessageRequestFactory requestFactory = new ControlMessageRequestFactory(); - if (!receivedConnectorInfo) { - requestFactory.infoLevels.add(InfoLevel.CONNECTOR_INFO); - } - trySubmitRequest(requestFactory); + trySubmitRequest(receivedConnectorInfo ? new ControlMessageRequestFactory() + : connectorInfoRequestFactory()); } } return operational || !remoteRequests.isEmpty() || !localRequests.isEmpty();