From 646e91612cdc056a0509ef2a610363589c3ef359 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Mon, 20 Jul 2026 19:23:00 +0300 Subject: [PATCH 1/2] Deliver CancelOpRequest arriving before the operation registers (#112) A CancelOpRequest processed before its OperationRequest registered in localRequests removed nothing and was silently dropped: the operation ran to completion and a subscription operation kept streaming events forever. Park such cancels in the connection group and apply them at registration. Registration is now an explicit step after construction, so a concurrent cancel can no longer observe a partially constructed processor, and the subscription/batch processors hand their Subscription over under a lock, closing it instead of throwing NPE when the cancel wins the race. --- .../openicf/common/rpc/LocalRequest.java | 28 +++- .../common/rpc/RemoteConnectionGroup.java | 50 +++++++ .../rpc/LocalRequestRegistrationTest.java | 136 ++++++++++++++++++ .../common/rpc/RequestDistributorTest.java | 6 +- .../common/rpc/impl/TestLocalRequest.java | 6 + .../framework/async/impl/BatchApiOpImpl.java | 43 +++++- .../ConnectorEventSubscriptionApiOpImpl.java | 35 ++++- .../impl/SyncEventSubscriptionApiOpImpl.java | 35 ++++- .../remote/OpenICFServerAdapter.java | 77 +++++----- 9 files changed, 370 insertions(+), 46 deletions(-) create mode 100644 OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java diff --git a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java index 59dd6851..1e257f46 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java @@ -26,6 +26,8 @@ package org.forgerock.openicf.common.rpc; +import java.util.concurrent.atomic.AtomicBoolean; + import org.forgerock.util.promise.ExceptionHandler; import org.forgerock.util.promise.ResultHandler; @@ -43,10 +45,27 @@ public abstract class LocalRequest> localRequests = new ConcurrentSkipListMap>(); + /** + * Cancel messages that arrived before the addressed operation registered + * itself in {@link #localRequests} are parked here (request id to expiry + * timestamp) and applied when {@link #receiveRequest} registers the + * request. Request ids are never reused within a group, so a parked + * cancel can only ever match the operation it was sent for; entries whose + * operation never arrives are dropped after {@link #PENDING_CANCEL_TTL_MS}. + */ + private final ConcurrentMap pendingCancels = + new ConcurrentHashMap(); + + private static final long PENDING_CANCEL_TTL_MS = 60L * 1000L; + protected final CopyOnWriteArrayList> webSockets = new CopyOnWriteArrayList>(); @@ -192,14 +208,30 @@ public , V, E extends Exception> R trySub return remoteRequest; } + /** + * Registers the given request so responses and cancel messages can be + * dispatched to it. + * + * @return the registered request, or {@code null} when a cancel for its + * request id arrived before registration - the request is + * cancelled and must not be executed. + */ public > R receiveRequest( final R localRequest) { + purgeExpiredPendingCancels(); LocalRequest tmp = localRequests.putIfAbsent(localRequest.getRequestId(), localRequest); if (null != tmp && !tmp.equals(localRequest)) { throw new IllegalStateException("Request has been registered with id: " + localRequest.getRequestId()); } + // Registration and receiveRequestCancel write the two maps in + // opposite order, so whichever call runs second is guaranteed to see + // the other's entry and deliver the cancel. + if (null != pendingCancels.remove(localRequest.getRequestId())) { + localRequest.cancel(); + return null; + } return localRequest; } @@ -216,13 +248,31 @@ public > R receive } public LocalRequest receiveRequestCancel(long messageId) { + purgeExpiredPendingCancels(); + // Park first, then look up: the operation may be racing through + // registration on another thread, and this order guarantees that at + // least one side observes the other (see receiveRequest). + pendingCancels.put(messageId, System.currentTimeMillis() + PENDING_CANCEL_TTL_MS); LocalRequest tmp = localRequests.remove(messageId); if (null != tmp) { + pendingCancels.remove(messageId); tmp.cancel(); } return tmp; } + private void purgeExpiredPendingCancels() { + if (pendingCancels.isEmpty()) { + return; + } + final long now = System.currentTimeMillis(); + for (Map.Entry entry : pendingCancels.entrySet()) { + if (entry.getValue() < now) { + pendingCancels.remove(entry.getKey(), entry.getValue()); + } + } + } + // -- Pair of methods to Cancel pending Request End -- // -- Pair of methods to Communicate pending Request Start -- diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java new file mode 100644 index 00000000..e2ed24a2 --- /dev/null +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java @@ -0,0 +1,136 @@ +/* + * 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.common.rpc; + +import java.util.concurrent.Future; + +import org.forgerock.openicf.common.rpc.impl.TestConnectionContext; +import org.forgerock.openicf.common.rpc.impl.TestConnectionGroup; +import org.forgerock.openicf.common.rpc.impl.TestLocalRequest; +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * Tests the explicit {@link LocalRequest#register()} step and the handling of + * a {@code CancelOpRequest} that is processed before the addressed operation + * has registered itself (OpenIdentityPlatform/OpenICF#112). + */ +public class LocalRequestRegistrationTest, H, TestConnectionContext>> { + + /** + * Minimal connection holder: the tests never transmit anything, they only + * need {@link #getRemoteConnectionContext()} to reach the group. + */ + private class StubConnectionHolder implements + RemoteConnectionHolder, H, TestConnectionContext> { + + private final TestConnectionGroup group; + + private StubConnectionHolder(TestConnectionGroup group) { + this.group = group; + } + + public TestConnectionContext getRemoteConnectionContext() { + return group.getRemoteConnectionContext(); + } + + public Future sendBytes(byte[] data) { + throw new UnsupportedOperationException(); + } + + public Future sendString(String data) { + throw new UnsupportedOperationException(); + } + + public void sendPing(byte[] applicationData) throws Exception { + throw new UnsupportedOperationException(); + } + + public void sendPong(byte[] applicationData) throws Exception { + throw new UnsupportedOperationException(); + } + + public void close() { + } + } + + @SuppressWarnings("unchecked") + private H newSocket(TestConnectionGroup group) { + return (H) new StubConnectionHolder(group); + } + + @Test + public void testConstructorDoesNotRegister() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + new TestLocalRequest(1L, newSocket(group)); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testRegisterThenCancel() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + + Assert.assertTrue(request.register()); + Assert.assertTrue(group.getLocalRequests().contains(1L)); + + Assert.assertSame(group.receiveRequestCancel(1L), request); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testCancelBeforeRegistrationIsParked() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + + // The cancel is processed before the operation registered itself. + Assert.assertNull(group.receiveRequestCancel(1L)); + + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + Assert.assertFalse(request.register()); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + @Test + public void testParkedCancelDoesNotAffectOtherRequests() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + + Assert.assertNull(group.receiveRequestCancel(1L)); + + TestLocalRequest other = new TestLocalRequest(2L, newSocket(group)); + Assert.assertTrue(other.register()); + Assert.assertFalse(other.isCancelled()); + Assert.assertTrue(group.getLocalRequests().contains(2L)); + } + + @Test + public void testCancelIsDeliveredOnce() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + final int[] cancelCount = new int[1]; + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)) { + public boolean tryCancel() { + cancelCount[0]++; + return super.tryCancel(); + } + }; + + Assert.assertTrue(request.register()); + Assert.assertTrue(request.cancel()); + Assert.assertFalse(request.cancel()); + Assert.assertEquals(cancelCount[0], 1); + } +} diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java index 3b31cfc2..fba02bc0 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RequestDistributorTest.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.common.rpc; @@ -73,7 +75,9 @@ public void onMessage(H socket, String message) { if (obj.request >= 0) { TestLocalRequest request = new TestLocalRequest(obj.messageId, socket); try { - getConnectionGroup().receiveRequest(request).execute(obj.request); + if (request.register()) { + request.execute(obj.request); + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); // request.handleException(e); diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java index 732d6072..b0d82b81 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.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.common.rpc.impl; @@ -111,6 +113,10 @@ public boolean tryCancel() { return true; } + public boolean isCancelled() { + return cancelled; + } + public void execute(int operation) throws InterruptedException { switch (operation) { case 0: { diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java index 7003b810..d3aeddfb 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/BatchApiOpImpl.java @@ -465,6 +465,7 @@ private static class InternalLocalOperationProcessor extends AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; private final AtomicBoolean commandChannelComplete = new AtomicBoolean(false); private final AtomicInteger resultChannelComplete = new AtomicInteger(0); @@ -553,6 +554,7 @@ public void onNext(BatchResult result) { } }; + Subscription batchSubscription = null; if (!requestMessage.getQuery()) { final List tasks = new ArrayList(); @@ -588,7 +590,7 @@ public void onNext(BatchResult result) { } try { - subscription = connectorFacade.executeBatch(tasks, observer, options); + batchSubscription = connectorFacade.executeBatch(tasks, observer, options); } catch (Throwable t) { tryHandleError(new ConnectorException(t)); return BatchOpResult.newBuilder().build(); @@ -599,16 +601,20 @@ public void onNext(BatchResult result) { for (int i = 0; i < requestMessage.getBatchTokenCount(); i++) { queryToken.addToken(requestMessage.getBatchToken(i)); } - subscription = connectorFacade.queryBatch(queryToken, observer, options); + batchSubscription = connectorFacade.queryBatch(queryToken, observer, options); } catch (Throwable t) { tryHandleError(new ConnectorException(t)); return BatchOpResult.newBuilder().build(); } } + if (null != batchSubscription) { + attachSubscription(batchSubscription); + } + BatchOpResult.Builder result = BatchOpResult.newBuilder(); - if (subscription != null && subscription.getReturnValue() != null) { - BatchToken returnedToken = (BatchToken) subscription.getReturnValue(); + if (batchSubscription != null && batchSubscription.getReturnValue() != null) { + BatchToken returnedToken = (BatchToken) batchSubscription.getReturnValue(); for (String token : returnedToken.getTokens()) { result.addBatchToken(token); } @@ -623,8 +629,35 @@ public void onNext(BatchResult result) { return result.build(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java index eb4c3006..0b731297 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/ConnectorEventSubscriptionApiOpImpl.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.async.impl; @@ -227,6 +229,7 @@ private static class InternalLocalOperationProcessor extends AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; protected InternalLocalOperationProcessor(long requestId, WebSocketConnectionHolder socket, ConnectorEventSubscriptionOpRequest message) { @@ -265,7 +268,7 @@ protected ConnectorEventSubscriptionOpResponse executeOperation( operationOptions = MessagesUtil.deserializeLegacy(requestMessage.getOptions()); } - subscription = + final Subscription result = connectorFacade.subscribe(objectClass, token, new Observer() { public void onCompleted() { @@ -294,11 +297,39 @@ public void onNext(ConnectorObject syncDelta) { }, operationOptions); + attachSubscription(result); return ConnectorEventSubscriptionOpResponse.getDefaultInstance(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } diff --git a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java index 149cb207..8cd51ec6 100644 --- a/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.java +++ b/OpenICF-java-framework/connector-framework-server/src/main/java/org/forgerock/openicf/framework/async/impl/SyncEventSubscriptionApiOpImpl.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.async.impl; @@ -219,6 +221,7 @@ private static class InternalLocalOperationProcessor AbstractLocalOperationProcessor { private Subscription subscription = null; + private boolean cancelled = false; protected InternalLocalOperationProcessor(long requestId, WebSocketConnectionHolder socket, SyncEventSubscriptionOpRequest message) { @@ -258,7 +261,7 @@ protected SyncEventSubscriptionOpResponse executeOperation( operationOptions = MessagesUtil.deserializeLegacy(requestMessage.getOptions()); } - subscription = connectorFacade.subscribe(objectClass, token, new Observer() { + final Subscription result = connectorFacade.subscribe(objectClass, token, new Observer() { public void onCompleted() { handleResult(SyncEventSubscriptionOpResponse.newBuilder().setCompleted( @@ -283,11 +286,39 @@ public void onNext(SyncDelta syncDelta) { }, operationOptions); + attachSubscription(result); return SyncEventSubscriptionOpResponse.getDefaultInstance(); } + /** + * The subscription is created only after this request is registered, + * so a cancel can be delivered before the field is assigned; the + * hand-over happens under the lock so exactly one side closes the + * subscription. + */ + private void attachSubscription(final Subscription result) { + final boolean closeNow; + synchronized (this) { + closeNow = cancelled; + if (!closeNow) { + subscription = result; + } + } + if (closeNow) { + result.close(); + } + } + protected boolean tryCancel() { - subscription.close(); + final Subscription current; + synchronized (this) { + cancelled = true; + current = subscription; + subscription = null; + } + if (null != current) { + current.close(); + } return super.tryCancel(); } } 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..a868b67e 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 @@ -37,6 +37,7 @@ import org.forgerock.openicf.common.protobuf.RPCMessages.RemoteMessage; import org.forgerock.openicf.framework.ConnectorFramework; import org.forgerock.openicf.framework.async.AsyncConnectorInfoManager; +import org.forgerock.openicf.framework.async.impl.AbstractLocalOperationProcessor; import org.forgerock.openicf.framework.async.impl.AuthenticationAsyncApiOpImpl; import org.forgerock.openicf.framework.async.impl.BatchApiOpImpl; import org.forgerock.openicf.framework.async.impl.ConnectorEventSubscriptionApiOpImpl; @@ -423,56 +424,55 @@ public void processOperationRequest(final WebSocketConnectionHolder socket, ConnectorFacade connectorFacade = newInstance(socket, info, connectorFacadeKey); + AbstractLocalOperationProcessor processor = null; if (message.hasBatchOpRequest()) { - BatchApiOpImpl.createProcessor(messageId, socket, - message.getBatchOpRequest()).execute(connectorFacade); + processor = BatchApiOpImpl.createProcessor(messageId, socket, + message.getBatchOpRequest()); } else if (message.hasAuthenticateOpRequest()) { - AuthenticationAsyncApiOpImpl.createProcessor(messageId, socket, - message.getAuthenticateOpRequest()).execute(connectorFacade); + processor = AuthenticationAsyncApiOpImpl.createProcessor(messageId, socket, + message.getAuthenticateOpRequest()); } else if (message.hasCreateOpRequest()) { - CreateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getCreateOpRequest()).execute(connectorFacade); + processor = CreateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getCreateOpRequest()); } else if (message.hasConnectorEventSubscriptionOpRequest()) { - ConnectorEventSubscriptionApiOpImpl.createProcessor(messageId, socket, - message.getConnectorEventSubscriptionOpRequest()).execute( - connectorFacade); + processor = ConnectorEventSubscriptionApiOpImpl.createProcessor(messageId, + socket, message.getConnectorEventSubscriptionOpRequest()); } else if (message.hasDeleteOpRequest()) { - DeleteAsyncApiOpImpl.createProcessor(messageId, socket, - message.getDeleteOpRequest()).execute(connectorFacade); + processor = DeleteAsyncApiOpImpl.createProcessor(messageId, socket, + message.getDeleteOpRequest()); } else if (message.hasGetOpRequest()) { - GetAsyncApiOpImpl.createProcessor(messageId, socket, - message.getGetOpRequest()).execute(connectorFacade); + processor = GetAsyncApiOpImpl.createProcessor(messageId, socket, + message.getGetOpRequest()); } else if (message.hasResolveUsernameOpRequest()) { - ResolveUsernameAsyncApiOpImpl.createProcessor(messageId, socket, - message.getResolveUsernameOpRequest()).execute(connectorFacade); + processor = ResolveUsernameAsyncApiOpImpl.createProcessor(messageId, + socket, message.getResolveUsernameOpRequest()); } else if (message.hasSchemaOpRequest()) { - SchemaAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSchemaOpRequest()).execute(connectorFacade); + processor = SchemaAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSchemaOpRequest()); } else if (message.hasScriptOnConnectorOpRequest()) { - ScriptOnConnectorAsyncApiOpImpl.createProcessor(messageId, socket, - message.getScriptOnConnectorOpRequest()).execute(connectorFacade); + processor = ScriptOnConnectorAsyncApiOpImpl.createProcessor(messageId, + socket, message.getScriptOnConnectorOpRequest()); } else if (message.hasScriptOnResourceOpRequest()) { - ScriptOnResourceAsyncApiOpImpl.createProcessor(messageId, socket, - message.getScriptOnResourceOpRequest()).execute(connectorFacade); + processor = ScriptOnResourceAsyncApiOpImpl.createProcessor(messageId, + socket, message.getScriptOnResourceOpRequest()); } else if (message.hasSearchOpRequest()) { - SearchAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSearchOpRequest()).execute(connectorFacade); + processor = SearchAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSearchOpRequest()); } else if (message.hasSyncOpRequest()) { - SyncAsyncApiOpImpl.createProcessor(messageId, socket, - message.getSyncOpRequest()).execute(connectorFacade); + processor = SyncAsyncApiOpImpl.createProcessor(messageId, socket, + message.getSyncOpRequest()); } else if (message.hasSyncEventSubscriptionOpRequest()) { - SyncEventSubscriptionApiOpImpl.createProcessor(messageId, socket, - message.getSyncEventSubscriptionOpRequest()).execute( - connectorFacade); + processor = SyncEventSubscriptionApiOpImpl.createProcessor(messageId, + socket, message.getSyncEventSubscriptionOpRequest()); } else if (message.hasTestOpRequest()) { - TestAsyncApiOpImpl.createProcessor(messageId, socket, - message.getTestOpRequest()).execute(connectorFacade); + processor = TestAsyncApiOpImpl.createProcessor(messageId, socket, + message.getTestOpRequest()); } else if (message.hasUpdateOpRequest()) { - UpdateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getUpdateOpRequest()).execute(connectorFacade); + processor = UpdateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getUpdateOpRequest()); } else if (message.hasValidateOpRequest()) { - ValidateAsyncApiOpImpl.createProcessor(messageId, socket, - message.getValidateOpRequest()).execute(connectorFacade); + processor = ValidateAsyncApiOpImpl.createProcessor(messageId, socket, + message.getValidateOpRequest()); } else { socket.getRemoteConnectionContext().getRemoteConnectionGroup() .trySendMessage( @@ -480,6 +480,15 @@ public void processOperationRequest(final WebSocketConnectionHolder socket, new ConnectorException("Unknown OperationRequest")) .build()); } + + // Registration is separate from construction so a + // concurrently processed CancelOpRequest can never observe + // a partially constructed processor; register() returns + // false when a pending cancel already consumed the + // request, in which case the operation must not start. + if (null != processor && processor.register()) { + processor.execute(connectorFacade); + } } catch (Throwable t) { logger.ok(t, "Failed handle OperationRequest {0}", messageId); socket.getRemoteConnectionContext().getRemoteConnectionGroup().trySendMessage( From 73542c9fed0b72cb41128aa696950fef8892a5aa Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Tue, 21 Jul 2026 10:46:20 +0300 Subject: [PATCH 2/2] Address review: register() must observe a directly delivered cancel A CancelOpRequest processed entirely between the registration's putIfAbsent and its pending-cancel check cancelled the request directly, yet register() still returned true and the operation was executed. Decide from the request's own cancelled state instead of from winning the pendingCancels removal, and never leave a request reported dead in localRequests. Covered by a deterministic replay test (fails on the previous logic) and a 1000-round two-thread stress test racing register() against receiveRequestCancel(). --- .../openicf/common/rpc/LocalRequest.java | 16 ++++ .../common/rpc/RemoteConnectionGroup.java | 16 +++- .../rpc/LocalRequestRegistrationTest.java | 75 +++++++++++++++++++ .../common/rpc/impl/TestLocalRequest.java | 4 - 4 files changed, 106 insertions(+), 5 deletions(-) diff --git a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java index 1e257f46..2b2521cb 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/LocalRequest.java @@ -47,6 +47,13 @@ public abstract class LocalRequest pendingCancels = new ConcurrentHashMap(); @@ -227,9 +231,19 @@ public > R receive } // Registration and receiveRequestCancel write the two maps in // opposite order, so whichever call runs second is guaranteed to see - // the other's entry and deliver the cancel. + // the other's entry and deliver the cancel. The verdict must come + // from the request's own state rather than from winning the park + // removal: receiveRequestCancel may find the request in localRequests + // right after the putIfAbsent above, consume its own park and cancel + // the request directly - this thread then observes no parked entry. if (null != pendingCancels.remove(localRequest.getRequestId())) { localRequest.cancel(); + } + if (localRequest.isCancelled()) { + // cancel() removes the registration, but only when the cancelling + // side found it - make sure a request reported dead is never left + // registered. + localRequests.remove(localRequest.getRequestId(), localRequest); return null; } return localRequest; diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java index e2ed24a2..e477782b 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java @@ -15,7 +15,12 @@ */ package org.forgerock.openicf.common.rpc; +import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; import org.forgerock.openicf.common.rpc.impl.TestConnectionContext; import org.forgerock.openicf.common.rpc.impl.TestConnectionGroup; @@ -133,4 +138,74 @@ public boolean tryCancel() { Assert.assertFalse(request.cancel()); Assert.assertEquals(cancelCount[0], 1); } + + /** + * Replays the interleaving where the whole {@code receiveRequestCancel} + * runs between the registration's {@code putIfAbsent} and its pending + * cancel check: the re-registration below stands in for the registering + * thread resuming after the cancel was delivered directly. The verdict + * must come from the request's own cancelled state, so register() must + * still report the request dead. + */ + @Test + public void testRegisterObservesDirectlyDeliveredCancel() throws Exception { + TestConnectionGroup group = new TestConnectionGroup("test"); + TestLocalRequest request = new TestLocalRequest(1L, newSocket(group)); + + Assert.assertTrue(request.register()); + Assert.assertSame(group.receiveRequestCancel(1L), request); + + Assert.assertFalse(request.register()); + Assert.assertTrue(request.isCancelled()); + Assert.assertTrue(group.getLocalRequests().isEmpty()); + } + + /** + * Races {@link LocalRequest#register()} against + * {@link TestConnectionGroup#receiveRequestCancel(long)}. Whatever the + * interleaving, the cancel must be delivered exactly once and the request + * must end up unregistered. + */ + @Test + public void testConcurrentRegisterAndCancel() throws Exception { + final TestConnectionGroup group = new TestConnectionGroup("test"); + final ExecutorService executor = Executors.newFixedThreadPool(2); + try { + for (long requestId = 1; requestId <= 1000; requestId++) { + final long id = requestId; + final AtomicInteger cancelCount = new AtomicInteger(0); + final TestLocalRequest request = + new TestLocalRequest(id, newSocket(group)) { + public boolean tryCancel() { + cancelCount.incrementAndGet(); + return super.tryCancel(); + } + }; + final CyclicBarrier barrier = new CyclicBarrier(2); + Future registered = executor.submit(new Callable() { + public Boolean call() throws Exception { + barrier.await(); + return request.register(); + } + }); + Future cancelled = executor.submit(new Callable() { + public Object call() throws Exception { + barrier.await(); + return group.receiveRequestCancel(id); + } + }); + cancelled.get(); + Boolean live = registered.get(); + + Assert.assertTrue(request.isCancelled(), "round " + id + + ": cancel was lost"); + Assert.assertEquals(cancelCount.get(), 1, "round " + id + + ": tryCancel() delivered more than once"); + Assert.assertFalse(group.getLocalRequests().contains(id), "round " + id + + ": cancelled request left registered (register() == " + live + ")"); + } + } finally { + executor.shutdownNow(); + } + } } diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java index b0d82b81..6df8f967 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/impl/TestLocalRequest.java @@ -113,10 +113,6 @@ public boolean tryCancel() { return true; } - public boolean isCancelled() { - return cancelled; - } - public void execute(int operation) throws InterruptedException { switch (operation) { case 0: {