From bc65c6a66939c03ab12ac02a40f521e8749d6ec7 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 18 Sep 2026 14:55:39 +0300 Subject: [PATCH] [#124] Create the RemoteRequest promise at construction so an unsent request can be cancelled RemoteRequest was registered in RemoteConnectionGroup.remoteRequests before its promise existed: the promise was created inside the send function and reset to null when a send failed. Cancelling such a request - from WebSocketConnectionGroup.shutdown(), submitRequestCancel() or cancel() - threw NullPointerException from getPromise().cancel(...). On the client it escaped ClientRemoteConnectorInfoManager.doClose() after isRunning had been flipped, so the group, the private WebSocket connections and the close listeners (ConnectionManager's registry entry) were never cleaned up. The promise is now created in the constructor and getPromise() never returns null. The send function does not send a request whose promise is already done, so a request cancelled before it was sent hands its caller the cancelled promise instead of an answer that never arrives; a failed send keeps the same promise for the next connection. tryCancel(true) notifies the remote side only for a delivered message, and a cancel that races the send in progress is delivered after the request rather than ahead of it. Fixes #124 --- .../openicf/common/rpc/RemoteRequest.java | 169 +++++---- .../RemoteRequestCancelBeforeSendTest.java | 331 ++++++++++++++++++ 2 files changed, 433 insertions(+), 67 deletions(-) create mode 100644 OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RemoteRequestCancelBeforeSendTest.java diff --git a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/RemoteRequest.java b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/RemoteRequest.java index d12f762b..a63842a1 100644 --- a/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/RemoteRequest.java +++ b/OpenICF-java-framework/connector-framework-rpc/src/main/java/org/forgerock/openicf/common/rpc/RemoteRequest.java @@ -27,6 +27,7 @@ package org.forgerock.openicf.common.rpc; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; import org.forgerock.util.Function; @@ -49,15 +50,53 @@ public abstract class RemoteRequest completionCallback; - private Long requestTime = null; - private PromiseImpl promise = null; + // The request is registered in RemoteConnectionGroup#remoteRequests + // before it is sent, so it can be cancelled from another thread while + // the send is still in progress. The promise therefore exists from + // construction on, and requestTime records whether the message has left + // (read by tryCancel on the cancelling thread). + private volatile Long requestTime = null; + private final PromiseImpl promise; private final ReentrantLock lock = new ReentrantLock(); + // tryCancel(true) sets this before it reads requestTime; the send + // function sets requestTime before it reads this. Whichever runs second + // sees the other's write, so a cancel racing with the send never leaves + // the remote side uninformed - and remoteCancelSent keeps it to one + // cancel message when both do. + private volatile boolean remoteCancelRequested = false; + private final AtomicBoolean remoteCancelSent = new AtomicBoolean(false); + public RemoteRequest(P context, long requestId, RemoteRequestFactory.CompletionCallback completionCallback) { this.context = context; this.requestId = requestId; this.completionCallback = completionCallback; + this.promise = new PromiseImpl() { + + protected E tryCancel(boolean mayInterruptIfRunning) { + if (mayInterruptIfRunning) { + remoteCancelRequested = true; + // Nothing to cancel remotely while the message has not + // been delivered: the send function does not send a + // cancelled request. + if (isSent()) { + try { + notifyRemoteCancelOnce(); + } catch (final Throwable t) { + return createCancellationException(t); + } + } + } + return createCancellationException(null); + } + + }; + this.promise.thenOnResultOrException(new Runnable() { + public void run() { + RemoteRequest.this.completionCallback.complete(RemoteRequest.this); + } + }); } /** @@ -89,6 +128,13 @@ public Long getRequestTime() { return requestTime; } + /** + * Returns the promise of this request. It exists from construction on, + * so a request that is registered but not yet sent can be cancelled; the + * result arrives only once the message has been sent and answered. + * + * @return the promise, never {@code null}. + */ public Promise getPromise() { return promise; } @@ -109,79 +155,68 @@ protected boolean cancel() { return promise.cancel(false); } + private boolean isSent() { + return null != requestTime; + } + + private void notifyRemoteCancelOnce() { + if (remoteCancelSent.compareAndSet(false, true)) { + tryCancelRemote(context, requestId); + } + } + public Function, Exception> getSendFunction() { - final Promise resultPromise = promise; - if (null == resultPromise) { - final MessageElement message = createMessageElement(context, requestId); - if (message == null || !(message.isString() || message.isByte())) { - throw new IllegalStateException("RemoteRequest has empty message"); - } + if (isSent()) { return new Function, Exception>() { - public Promise apply(H remoteConnectionHolder) throws Exception { - if (null == promise) { - // Single thread should process it so it should not - // return false - if (lock.tryLock(1, TimeUnit.MINUTES)) { - try { - if (null == promise) { - - promise = new PromiseImpl() { - - protected E tryCancel(boolean mayInterruptIfRunning) { - if (mayInterruptIfRunning) { - try { - tryCancelRemote(context, requestId); - } catch (final Throwable t) { - return createCancellationException(t); - } - } - return createCancellationException(null); - } - - }; - - promise.thenOnResultOrException(new Runnable() { - public void run() { - completionCallback.complete(RemoteRequest.this); - } - }); - - try { - if (message.isByte()) { - remoteConnectionHolder.sendBytes(message.byteMessage) - .get(); - } else if (message.isString()) { - remoteConnectionHolder - .sendString(message.stringMessage).get(); - } - } catch (final Exception e) { - promise = null; - throw e; - } catch (final Throwable t) { - promise = null; - throw new Exception(t); - } - // Message has been delivered - Report - // success - requestTime = System.currentTimeMillis(); - } - } finally { - lock.unlock(); - } - } - } + public Promise apply(H value) throws Exception { return promise; } }; - } else { - return new Function, Exception>() { + } + final MessageElement message = createMessageElement(context, requestId); + if (message == null || !(message.isString() || message.isByte())) { + throw new IllegalStateException("RemoteRequest has empty message"); + } + return new Function, Exception>() { - public Promise apply(H value) throws Exception { - return resultPromise; + public Promise apply(H remoteConnectionHolder) throws Exception { + // A request cancelled before it was sent stays unsent: the + // caller gets the cancelled promise back instead of waiting + // for an answer that can never arrive. + if (isSent() || promise.isDone()) { + return promise; } - }; - } + // Single thread should process it so it should not + // return false + if (!lock.tryLock(1, TimeUnit.MINUTES)) { + throw new IllegalStateException("RemoteRequest " + requestId + + " is still being sent by another thread"); + } + try { + if (!isSent() && !promise.isDone()) { + // A failed send propagates to the group, which + // retries on its next connection with this same + // promise. + if (message.isByte()) { + remoteConnectionHolder.sendBytes(message.byteMessage).get(); + } else if (message.isString()) { + remoteConnectionHolder.sendString(message.stringMessage).get(); + } + // Message has been delivered - Report success + requestTime = System.currentTimeMillis(); + if (remoteCancelRequested) { + // Cancelled while the message was on its way: + // tryCancel saw it unsent. + notifyRemoteCancelOnce(); + } + } + } finally { + lock.unlock(); + } + return promise; + } + }; } // --- inner Classes diff --git a/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RemoteRequestCancelBeforeSendTest.java b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RemoteRequestCancelBeforeSendTest.java new file mode 100644 index 00000000..211a60be --- /dev/null +++ b/OpenICF-java-framework/connector-framework-rpc/src/test/java/org/forgerock/openicf/common/rpc/RemoteRequestCancelBeforeSendTest.java @@ -0,0 +1,331 @@ +/* + * 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.io.IOException; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.forgerock.openicf.common.rpc.impl.TestConnectionContext; +import org.forgerock.openicf.common.rpc.impl.TestConnectionGroup; +import org.forgerock.openicf.common.rpc.impl.TestMessage; +import org.forgerock.openicf.common.rpc.impl.TestRemoteRequest; +import org.forgerock.util.promise.Promises; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * A {@link RemoteRequest} is registered in + * {@link RemoteConnectionGroup#trySubmitRequest} before its message is sent, + * so it can be cancelled - by {@link RemoteConnectionGroup#submitRequestCancel} + * or by a group shutting down - while the send is still in progress. These + * tests pin down what happens to such a request. + */ +public class RemoteRequestCancelBeforeSendTest { + + private static final long TIMEOUT_SECONDS = 10; + + private TestConnectionGroup group; + private ExecutorService executor; + + @BeforeMethod + public void setUp() { + group = new TestConnectionGroup("client"); + executor = Executors.newSingleThreadExecutor(); + } + + @AfterMethod + public void tearDown() { + executor.shutdownNow(); + } + + @Test + public void promiseExistsWhileRequestIsRegisteredButNotSent() throws Exception { + RecordingHolder holder = new RecordingHolder(group, false); + group.addConnection(holder); + BlockingRequestFactory factory = new BlockingRequestFactory(); + + Future> submitted = submit(factory); + try { + factory.awaitBlockedBeforeSend(); + + Assert.assertEquals(group.getRemoteRequests().size(), 1); + Assert.assertNotNull(factory.request.get().getPromise(), + "registered request must expose its promise before the message is sent"); + } finally { + factory.releaseSend(); + submitted.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + } + + @Test + public void cancelBeforeSendCancelsPromiseWithoutSendingAnything() throws Exception { + RecordingHolder holder = new RecordingHolder(group, false); + group.addConnection(holder); + BlockingRequestFactory factory = new BlockingRequestFactory(); + + Future> submitted = submit(factory); + factory.awaitBlockedBeforeSend(); + long requestId = group.getRemoteRequests().iterator().next(); + + RemoteRequest cancelled = group.submitRequestCancel(requestId); + + Assert.assertSame(cancelled, factory.request.get()); + Assert.assertTrue(cancelled.getPromise().isCancelled()); + + factory.releaseSend(); + TestRemoteRequest request = submitted.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + + Assert.assertSame(request, cancelled, + "the caller gets its request back so it can observe the cancellation"); + Assert.assertTrue(request.getPromise().isCancelled()); + Assert.assertEquals(holder.sent.size(), 0, "nothing goes on the wire: " + holder.sent); + Assert.assertTrue(group.getRemoteRequests().isEmpty()); + } + + @Test + public void groupShutdownBeforeSendCancelsPromiseWithoutSendingAnything() throws Exception { + RecordingHolder holder = new RecordingHolder(group, false); + group.addConnection(holder); + BlockingRequestFactory factory = new BlockingRequestFactory(); + + Future> submitted = submit(factory); + factory.awaitBlockedBeforeSend(); + + // Same loop as WebSocketConnectionGroup.shutdown(): cancel(true) on + // every registered remote request. + group.close(); + + factory.releaseSend(); + TestRemoteRequest request = submitted.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + + Assert.assertTrue(request.getPromise().isCancelled()); + Assert.assertEquals(holder.sent.size(), 0, "nothing goes on the wire: " + holder.sent); + Assert.assertTrue(group.getRemoteRequests().isEmpty()); + } + + @Test + public void cancelDuringSendNotifiesRemoteAfterTheRequestIsDelivered() throws Exception { + RecordingHolder holder = new RecordingHolder(group, false); + holder.blockFirstSend(); + group.addConnection(holder); + BlockingRequestFactory factory = new BlockingRequestFactory(); + factory.releaseSend(); + + Future> submitted = submit(factory); + holder.awaitBlockedInSend(); + + // The request is on its way: cancel(true) must not race the cancel + // message ahead of the request itself. + Assert.assertTrue(factory.request.get().getPromise().cancel(true)); + + holder.releaseSend(); + TestRemoteRequest request = submitted.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + + Assert.assertTrue(request.getPromise().isCancelled()); + Assert.assertEquals(holder.sent.size(), 2, "request then cancel: " + holder.sent); + Assert.assertEquals(read(holder.sent.get(0)).request, 0); + Assert.assertEquals(read(holder.sent.get(1)).cancel, Boolean.TRUE); + Assert.assertTrue(group.getRemoteRequests().isEmpty()); + } + + // Behaviour that must survive the change: the cases below pass before it. + + @Test + public void cancelAfterSendNotifiesRemote() throws Exception { + RecordingHolder holder = new RecordingHolder(group, false); + group.addConnection(holder); + BlockingRequestFactory factory = new BlockingRequestFactory(); + factory.releaseSend(); + + TestRemoteRequest request = group.trySubmitRequest(factory); + Assert.assertNotNull(request); + Assert.assertEquals(holder.sent.size(), 1); + + Assert.assertSame(group.submitRequestCancel(request.getRequestId()), request); + + Assert.assertTrue(request.getPromise().isCancelled()); + Assert.assertEquals(holder.sent.size(), 2, "request then cancel: " + holder.sent); + Assert.assertEquals(read(holder.sent.get(1)).cancel, Boolean.TRUE); + Assert.assertTrue(group.getRemoteRequests().isEmpty()); + } + + @Test + public void sendFailsOverToTheNextConnectionAndCompletesNormally() throws Exception { + RecordingHolder broken = new RecordingHolder(group, true); + RecordingHolder working = new RecordingHolder(group, false); + group.addConnection(broken); + group.addConnection(working); + BlockingRequestFactory factory = new BlockingRequestFactory(); + factory.releaseSend(); + + TestRemoteRequest request = group.trySubmitRequest(factory); + Assert.assertNotNull(request); + + Assert.assertEquals(broken.attempted.size(), 1); + Assert.assertEquals(broken.sent.size(), 0); + Assert.assertEquals(working.sent.size(), 1); + Assert.assertFalse(request.getPromise().isDone()); + Assert.assertTrue(group.getRemoteRequests().contains(request.getRequestId())); + + TestMessage response = new TestMessage(); + response.response = "OK"; + group.receiveRequestResponse(working, request.getRequestId(), response); + + Assert.assertEquals(request.getPromise().getOrThrow(TIMEOUT_SECONDS, TimeUnit.SECONDS), "OK"); + Assert.assertTrue(group.getRemoteRequests().isEmpty()); + } + + // ---- helpers --------------------------------------------------------- + + private Future> submit(final BlockingRequestFactory factory) { + return executor.submit(() -> group.trySubmitRequest(factory)); + } + + private TestMessage read(String message) { + return group.getRemoteConnectionContext().read(message); + } + + /** + * Creates requests that block in {@code createMessageElement} - after + * registration, before anything is sent - until {@link #releaseSend()}. + */ + private static final class BlockingRequestFactory + implements + RemoteRequestFactory, String, Exception, TestConnectionGroup, RecordingHolder, TestConnectionContext> { + + final AtomicReference> request = new AtomicReference<>(); + private final CountDownLatch blocked = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + public TestRemoteRequest createRemoteRequest( + TestConnectionContext context, + long requestId, + CompletionCallback, RecordingHolder, TestConnectionContext> completionCallback) { + TestRemoteRequest created = + new TestRemoteRequest(context, requestId, completionCallback) { + protected TestMessage getTestMessage() { + blocked.countDown(); + try { + if (!release.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("send was never released"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + TestMessage message = new TestMessage(); + message.request = 0; + return message; + } + + protected void handle(RecordingHolder sourceConnection, + TestRemoteRequest request, TestMessage message) { + } + }; + request.set(created); + return created; + } + + void awaitBlockedBeforeSend() throws InterruptedException { + Assert.assertTrue(blocked.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), + "request did not reach createMessageElement"); + } + + void releaseSend() { + release.countDown(); + } + } + + /** Records every message handed to it; completes sends immediately. */ + private static final class RecordingHolder + implements + RemoteConnectionHolder, RecordingHolder, TestConnectionContext> { + + final List attempted = new CopyOnWriteArrayList<>(); + final List sent = new CopyOnWriteArrayList<>(); + private final TestConnectionContext context; + private final boolean failing; + private CountDownLatch inSend; + private CountDownLatch releaseSend; + + RecordingHolder(TestConnectionGroup group, boolean failing) { + this.context = group.getRemoteConnectionContext(); + this.failing = failing; + } + + /** The first send blocks until {@link #releaseSend()}. */ + void blockFirstSend() { + inSend = new CountDownLatch(1); + releaseSend = new CountDownLatch(1); + } + + void awaitBlockedInSend() throws InterruptedException { + Assert.assertTrue(inSend.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), + "request did not reach sendString"); + } + + void releaseSend() { + releaseSend.countDown(); + } + + public TestConnectionContext getRemoteConnectionContext() { + return context; + } + + public Future sendBytes(byte[] data) { + throw new UnsupportedOperationException("string messages only"); + } + + public Future sendString(String data) { + attempted.add(data); + if (failing) { + return Promises.newExceptionPromise(new IOException("send failed")); + } + if (null != inSend && inSend.getCount() > 0) { + inSend.countDown(); + try { + if (!releaseSend.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + return Promises.newExceptionPromise(new IOException("send was never released")); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return Promises.newExceptionPromise(new IOException(e)); + } + } + sent.add(data); + return Promises.newResultPromise(null); + } + + public void sendPing(byte[] applicationData) { + } + + public void sendPong(byte[] applicationData) { + } + + public void close() { + } + } +}