From 0f8914275a67877d15305b79e29d6c336aaaef1e Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 18 Sep 2026 16:28:45 +0300 Subject: [PATCH] [#125] Do not resend the initial CONNECTOR_INFO request when shutdown() cancelled it The failure branch attached to the initial 'CONNECTOR_INFO' request in WebSocketConnectionGroup.handshakeComplete() resubmitted the request on any failure, cancellation included. shutdown() cancels every pending remote request with getPromise().cancel(true); PromiseImpl runs the listeners on the cancelling thread and routes CANCELLED to the onException function, so the retry ran synchronously inside shutdown()'s loop - before principalIsShuttingDown() removed the sockets - and was sent over the closing connection and registered in the group being torn down. When the initial request was the only pending entry, the ConcurrentSkipListMap iterator had already advanced past the end and the retry stayed registered and pending in the dead group. The retry is now skipped when the first request's promise is cancelled. The promise state is checked rather than the exception type because RemoteOperationRequest.createCancellationException returns a failed remote cancel's exception as-is. Fixes #125 --- .../remote/rpc/WebSocketConnectionGroup.java | 10 ++ .../WebSocketConnectionGroupShutdownTest.java | 141 ++++++++++++++++++ 2 files changed, 151 insertions(+) create mode 100644 OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroupShutdownTest.java 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 b5010290..7a9f169c 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 @@ -152,6 +152,16 @@ public Boolean apply(Boolean value) throws RuntimeException { request.getPromise().then(success, new Function() { @Override public Boolean apply(RuntimeException e) throws RuntimeException { + if (request.getPromise().isCancelled()) { + // Only shutdown() cancels this request, and the + // listener runs on its thread before the sockets are + // removed: a retry would be sent over the closing + // connection and registered in the group being torn + // down. The state is checked rather than the + // exception type because createCancellationException + // returns the failure of the remote cancel as-is. + return receivedConnectorInfo; + } logger.ok("Resending initial 'CONNECTOR_INFO' request", e); ControlMessageRequest retry = trySubmitRequest(connectorInfoRequestFactory()); if (null != retry) { diff --git a/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroupShutdownTest.java b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroupShutdownTest.java new file mode 100644 index 00000000..648cb1f8 --- /dev/null +++ b/OpenICF-java-framework/connector-framework-server/src/test/java/org/forgerock/openicf/framework/remote/rpc/WebSocketConnectionGroupShutdownTest.java @@ -0,0 +1,141 @@ +/* + * 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.security.Principal; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; + +import org.forgerock.openicf.common.protobuf.RPCMessages.HandshakeMessage; +import org.forgerock.openicf.common.protobuf.RPCMessages.RemoteMessage; +import org.testng.Assert; +import org.testng.annotations.Test; + +import com.google.protobuf.InvalidProtocolBufferException; + +/** + * Tests that {@link WebSocketConnectionGroup#shutdown()} does not resend the + * initial 'CONNECTOR_INFO' request it has just cancelled: the retry attached + * in {@link WebSocketConnectionGroup#handshakeComplete()} is meant for a + * failure of the first request, not for the group's own shutdown. + */ +public class WebSocketConnectionGroupShutdownTest { + + private static final String SESSION_ID = "session"; + + /** Exposes the request registry the shutdown loop iterates over. */ + private static final class TestGroup extends WebSocketConnectionGroup { + + TestGroup() { + super(SESSION_ID); + } + + int pendingRemoteRequests() { + return remoteRequests.size(); + } + } + + /** Records every frame the group writes to the socket. */ + private static final class RecordingHolder extends WebSocketConnectionHolder { + + final List sent = new CopyOnWriteArrayList(); + volatile RemoteOperationContext context; + + protected void handshake(HandshakeMessage message) { + } + + protected void tryClose() { + } + + public boolean isOperational() { + return true; + } + + public RemoteOperationContext getRemoteConnectionContext() { + return context; + } + + public Future sendBytes(byte[] data) { + try { + sent.add(RemoteMessage.parseFrom(data)); + } catch (InvalidProtocolBufferException e) { + throw new IllegalArgumentException(e); + } + 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 { + } + + int controlRequests() { + int count = 0; + for (RemoteMessage message : sent) { + if (message.hasRequest() && message.getRequest().hasControlRequest()) { + count++; + } + } + return count; + } + + int cancelRequests() { + int count = 0; + for (RemoteMessage message : sent) { + if (message.hasRequest() && message.getRequest().hasCancelOpRequest()) { + count++; + } + } + return count; + } + } + + @Test(timeOut = 30000) + public void testShutdownDoesNotResendCancelledInitialConnectorInfoRequest() { + final Principal principal = new Principal() { + public String getName() { + return "client"; + } + }; + final TestGroup group = new TestGroup(); + final RecordingHolder holder = new RecordingHolder(); + + holder.context = group.handshake(principal, holder, + HandshakeMessage.newBuilder().setSessionId(SESSION_ID).build()); + group.handshakeComplete(); + + Assert.assertEquals(holder.controlRequests(), 1, "initial 'CONNECTOR_INFO' request"); + Assert.assertEquals(group.pendingRemoteRequests(), 1, "initial request is pending"); + + // The last principal leaves: shutdown() cancels the pending initial + // request, which must not schedule its retry. + group.principalIsShuttingDown(principal); + + Assert.assertEquals(holder.cancelRequests(), 1, + "the pending initial request is cancelled on the remote side"); + Assert.assertEquals(holder.controlRequests(), 1, + "initial 'CONNECTOR_INFO' request was resent while shutting down"); + Assert.assertEquals(group.pendingRemoteRequests(), 0, + "a request stayed registered in the shut-down group"); + } +}