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..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 @@ -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()) { @@ -193,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()); } }; @@ -203,6 +212,24 @@ public void run() { ws.close(); } } + // Self-heal: a connect attempt that failed before Grizzly + // 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 + && lastConnectionCreated.get() < lastConnectAttempt.get()) { + 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 +349,8 @@ 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); @@ -367,6 +396,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..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 @@ -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; @@ -89,7 +90,10 @@ 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(); public WebSocketConnectionGroup(final String remoteSessionId) { @@ -112,31 +116,59 @@ 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; + } + }; + + ControlMessageRequest request = trySubmitRequest(connectorInfoRequestFactory()); + 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); + ControlMessageRequest retry = trySubmitRequest(connectorInfoRequestFactory()); + if (null != retry) { + retry.getPromise().then(success); + } + return receivedConnectorInfo; + } + }); + } + } + + 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)) { @@ -322,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(); 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