[#112] Deliver CancelOpRequest arriving before the operation registers

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,34 @@ public abstract class LocalRequest<V, E extends Exception, G extends RemoteConne
 
     private final P remoteConnectionContext;
 
+    private final AtomicBoolean cancelled = new AtomicBoolean(Boolean.FALSE);
+
+    /**
+     * Creates the request without registering it. Historically the
+     * constructor registered {@code this} in the
+     * {@link RemoteConnectionGroup}; subclasses relying on that must now call
+     * {@link #register()} after construction, or the request will never
+     * receive responses or cancel messages.
+     */
     protected LocalRequest(final long requestId, final H socket) {
         this.requestId = requestId;
         remoteConnectionContext = socket.getRemoteConnectionContext();
-        remoteConnectionContext.getRemoteConnectionGroup().receiveRequest(this);
+    }
+
+    /**
+     * Registers this request in the
+     * {@link RemoteConnectionGroup} so responses and cancel messages can be
+     * dispatched to it. Must be called once the request is fully constructed
+     * and before it is executed: registration publishes this instance to
+     * other threads, and a registration racing with the constructor would let
+     * a concurrent cancel observe a partially initialized subclass.
+     *
+     * @return {@code true} when the request is registered and live,
+     *         {@code false} when a cancel for this request id was already
+     *         received - the request is cancelled and must not be executed.
+     */
+    public final boolean register() {
+        return null != remoteConnectionContext.getRemoteConnectionGroup().receiveRequest(this);
     }
 
     /**
@@ -76,9 +102,23 @@ public P getRemoteConnectionContext() {
         return remoteConnectionContext;
     }
 
+    /**
+     * Returns {@code true} once a {@link #cancel()} has been delivered to
+     * this request, whether directly or as a pending cancel applied during
+     * {@link #register()}.
+     */
+    public final boolean isCancelled() {
+        return cancelled.get();
+    }
+
     public final boolean cancel() {
         remoteConnectionContext.getRemoteConnectionGroup().removeRequest(getRequestId());
-        return tryCancel();
+        // A cancel may be delivered twice when a direct cancel races with a
+        // pending cancel applied during register() - deliver tryCancel() once.
+        if (cancelled.compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
+            return tryCancel();
+        }
+        return false;
     }
 
     public final void handleResult(final V result) {

OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/RemoteConnectionGroup.java

@@ -27,6 +27,9 @@
 package org.forgerock.openicf.common.rpc;
 
 import java.util.ArrayList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -60,6 +63,23 @@ public abstract class RemoteConnectionGroup<G extends RemoteConnectionGroup<G, H
     protected final ConcurrentNavigableMap<Long, LocalRequest<?, ?, G, H, P>> localRequests =
             new ConcurrentSkipListMap<Long, LocalRequest<?, ?, G, H, P>>();
 
+    /**
+     * 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}.
+     * The park in {@link #receiveRequestCancel} is unconditional, so a cancel
+     * that loses the race with completion (the operation already produced its
+     * result and unregistered) also leaves an entry behind for the full TTL;
+     * such entries are harmless and are removed by the next purge.
+     */
+    private final ConcurrentMap<Long, Long> pendingCancels =
+            new ConcurrentHashMap<Long, Long>();
+
+    private static final long PENDING_CANCEL_TTL_MS = 60L * 1000L;
+
     protected final CopyOnWriteArrayList<Pair<String, H>> webSockets =
             new CopyOnWriteArrayList<Pair<String, H>>();
 
@@ -192,14 +212,40 @@ public <R extends RemoteRequest<V, E, G, H, P>, 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 <V, E extends Exception, R extends LocalRequest<V, E, G, H, P>> R receiveRequest(
             final R localRequest) {
+        purgeExpiredPendingCancels();
         LocalRequest<?, ?, G, H, P> 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. 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;
     }
 
@@ -216,13 +262,31 @@ public <V, E extends Exception, R extends LocalRequest<V, E, G, H, P>> R receive
     }
 
     public LocalRequest<?, ?, G, H, P> 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<?, ?, G, H, P> 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<Long, Long> 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 --

OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/LocalRequestRegistrationTest.java

@@ -0,0 +1,211 @@
+/*
+ * 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.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;
+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 extends RemoteConnectionHolder<TestConnectionGroup<H>, H, TestConnectionContext<H>>> {
+
+    /**
+     * Minimal connection holder: the tests never transmit anything, they only
+     * need {@link #getRemoteConnectionContext()} to reach the group.
+     */
+    private class StubConnectionHolder implements
+            RemoteConnectionHolder<TestConnectionGroup<H>, H, TestConnectionContext<H>> {
+
+        private final TestConnectionGroup<H> group;
+
+        private StubConnectionHolder(TestConnectionGroup<H> group) {
+            this.group = group;
+        }
+
+        public TestConnectionContext<H> 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<H> group) {
+        return (H) new StubConnectionHolder(group);
+    }
+
+    @Test
+    public void testConstructorDoesNotRegister() throws Exception {
+        TestConnectionGroup<H> group = new TestConnectionGroup<H>("test");
+        new TestLocalRequest<H>(1L, newSocket(group));
+        Assert.assertTrue(group.getLocalRequests().isEmpty());
+    }
+
+    @Test
+    public void testRegisterThenCancel() throws Exception {
+        TestConnectionGroup<H> group = new TestConnectionGroup<H>("test");
+        TestLocalRequest<H> request = new TestLocalRequest<H>(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<H> group = new TestConnectionGroup<H>("test");
+
+        // The cancel is processed before the operation registered itself.
+        Assert.assertNull(group.receiveRequestCancel(1L));
+
+        TestLocalRequest<H> request = new TestLocalRequest<H>(1L, newSocket(group));
+        Assert.assertFalse(request.register());
+        Assert.assertTrue(request.isCancelled());
+        Assert.assertTrue(group.getLocalRequests().isEmpty());
+    }
+
+    @Test
+    public void testParkedCancelDoesNotAffectOtherRequests() throws Exception {
+        TestConnectionGroup<H> group = new TestConnectionGroup<H>("test");
+
+        Assert.assertNull(group.receiveRequestCancel(1L));
+
+        TestLocalRequest<H> other = new TestLocalRequest<H>(2L, newSocket(group));
+        Assert.assertTrue(other.register());
+        Assert.assertFalse(other.isCancelled());
+        Assert.assertTrue(group.getLocalRequests().contains(2L));
+    }
+
+    @Test
+    public void testCancelIsDeliveredOnce() throws Exception {
+        TestConnectionGroup<H> group = new TestConnectionGroup<H>("test");
+        final int[] cancelCount = new int[1];
+        TestLocalRequest<H> request = new TestLocalRequest<H>(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);
+    }
+
+    /**
+     * 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<H> group = new TestConnectionGroup<H>("test");
+        TestLocalRequest<H> request = new TestLocalRequest<H>(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<H> group = new TestConnectionGroup<H>("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<H> request =
+                        new TestLocalRequest<H>(id, newSocket(group)) {
+                            public boolean tryCancel() {
+                                cancelCount.incrementAndGet();
+                                return super.tryCancel();
+                            }
+                        };
+                final CyclicBarrier barrier = new CyclicBarrier(2);
+                Future<Boolean> registered = executor.submit(new Callable<Boolean>() {
+                    public Boolean call() throws Exception {
+                        barrier.await();
+                        return request.register();
+                    }
+                });
+                Future<Object> cancelled = executor.submit(new Callable<Object>() {
+                    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();
+        }
+    }
+}