fix: prevent sender control future race by lokidundun · Pull Request #357 · apache/hugegraph-computer

lokidundun · 2026-07-25T10:53:03Z

Purpose of the PR

close [Task] Investigate slow integrate-test stalls in message sender barrier #350

Main Changes

Fixes a race condition in QueuedMessageSender where consecutive control messages could fail with The origin future must be null.
The sender now clears the in-flight control future before completing it, allowing the next control message to be sent safely. Transport failures also complete the affected future exceptionally.

Tests

Added regression tests for consecutive control messages and transport failures.
Added fail-fast timeout handling to sender integration tests.
Verified with QueuedMessageSenderTest and IntegrateTestSuite.

mvn test -P integrate-test -pl computer-test -am \
  -Dtest=QueuedMessageSenderTest,IntegrateTestSuite \
  -DfailIfNoTests=false -Djacoco.skip=true

Verifying these changes

Trivial rework / code cleanup without any test coverage. (No Need)
Already covered by existing tests, such as (please modify tests here).
Need tests and can be verified as follows.

Does this PR potentially affect the following parts?

Documentation Status

Doc - TODO
Doc - Done
Doc - No Need

Copilot

Pull request overview

This PR addresses a control-message future race in the computer-core message-sending pipeline (specifically QueuedMessageSender) that could surface as The origin future must be null, and also tightens computer-test integration behavior to fail faster instead of stalling for long BSP timeouts.

Changes:

Refactors control-message handling in QueuedMessageSender so each START/FINISH message carries its own CompletableFuture, and the in-flight control future is cleared before completion.
Adds unit regressions for consecutive control messages and for transport failures completing the control future exceptionally.
Updates sender integration tests to apply shorter BSP wait timeouts and to use a bounded waitForServices() helper.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java	Adds CI-oriented BSP/service timeouts and a fail-fast wait helper for master/worker futures
computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java	Adds regressions for control-future sequencing and transport exception completion
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java	Refactors control-message future lifecycle and transport-exception handling
computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessage.java	Extends queued message model to optionally carry a control future

Comments suppressed due to low confidence (2)

computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSender.java:273

Same issue as sendStartMessage(): if setControlFuture() throws ComputerException, it will currently bubble out of sendFinishMessage() and can terminate the send-executor thread. Since the future is already completed exceptionally in setControlFuture(), catch ComputerException here and return to avoid killing the sender thread.

        public void sendFinishMessage(CompletableFuture<Void> future)
                                      throws TransportException {
            this.setControlFuture(future);
            try {
                this.client.finishSessionAsync().whenComplete((r, e) -> {

computer/computer-test/src/main/java/org/apache/hugegraph/computer/suite/integrate/SenderIntegrateTest.java:122

The master options set withRpcServerPort() twice (8611 then 0). The first value is immediately overridden and can be misleading when debugging port binding issues; it’s clearer to keep only the effective port setting.

                                          .withRpcServerHost("127.0.0.1")
                                          .withRpcServerPort(8611)
                                          .withRpcServerPort(0)

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+        public void sendStartMessage(CompletableFuture<Void> future)
+                                     throws TransportException {
+            this.setControlFuture(future);
+            try {
+                this.client.startSessionAsync().whenComplete((r, e) -> {


imbajin

Blocking: yes. Summary: An existing review thread covers the sender-executor termination risk, and the new fail-fast service wait can race past lifecycle cleanup. Evidence: exact-head static inspection by six independent lanes; git diff --check passed; visible exact-head checks are green.

imbajin · 2026-07-25T16:07:56Z

+        for (CompletableFuture<Void> future : futures) {
+            future.whenComplete((r, e) -> {
+                if (e != null) {
+                    result.completeExceptionally(e);


⚠️ Completing result on the first failure lets the caller enter cleanup before the other service threads have initialized and published their service references. A thread can add its service after closeWorker()/closeMaster() has already observed null or finished iterating workerServices; the multi-worker threads also do not own their services with try-with-resources. That can leave Netty/etcd resources running into later tests. Please cancel and join every spawned thread and ensure each thread closes any service it initializes in a finally, including services published after the first failure.

imbajin

Blocking: yes. Summary: Synchronous runtime failures can still strand an in-flight control future, and the fail-fast cleanup/tests leave concurrency gaps. Evidence: six independent exact-head review lanes; targeted QueuedMessageSenderTest passed; git diff --check passed; visible exact-head checks are green.

imbajin · 2026-07-26T12:22:36Z

+                return;
+            }
+            try {
+                this.client.startSessionAsync().whenComplete((r, e) -> {


‼️ A synchronous unchecked exception from this call leaves controlFutureRef pointing at future. ClientSession.startAsync() can throw IllegalArgumentException from its state checks, and its send function can propagate unchecked failures; neither reaches this TransportException catch, so the returned future stays incomplete and Sender.run() also exits. Please clear and complete the control future for synchronous runtime failures here and in finishSessionAsync(), then add a client-stub regression that throws synchronously and verifies both the returned future and the executor's intended state.

imbajin · 2026-07-26T12:22:36Z

+
+    private static void closeServicesAndJoin(ServiceLifecycle lifecycle,
+                                             List<Thread> threads) {
+        lifecycle.closeAll();


⚠️ closeAll() runs every closer sequentially before any service thread is interrupted, but registration order is concurrent. If the master closer runs before a still-active worker closer, MasterService.close() waits for workerCloseDone that the queued worker closer has not had a chance to send, so this fail-fast path can block for the full five-minute BSP timeout; one thrown closer also skips the remaining cleanup. Please encode worker-before-master shutdown (or close services concurrently), ensure one close failure cannot prevent the rest, and cover a master-first registration with one failed and one active worker.

imbajin · 2026-07-26T12:22:36Z

+
+            CompletableFuture<Void> finishFuture = sender.send(1,
+                                                                MessageType.FINISH);
+            allowCompletion.countDown();


⚠️ This releases the blocked completion dependent before confirming that FINISH was dispatched. An implementation that still clears the control reference only after completion dependents return can therefore process FINISH after this countdown and pass the test. Please await client.awaitFinish() before releasing allowCompletion (while keeping the finally release) so the test actually proves the next control message proceeds while the prior dependent is still blocked.

imbajin · 2026-07-26T12:22:36Z

+            sender.transportExceptionCaught(cause, client.connectionId());
+            assertFutureFailedWith(startFuture, cause);
+
+            client.completeStart();


⚠️ This completes the stale START callback immediately after the transport exception, without first installing a next-generation FINISH future. It therefore does not protect the identity-sensitive CAS behavior in completeControlFuture(); an unconditional clear could still pass while later stranding FINISH. Please send and await FINISH first, then complete the stale START callback, and finally complete/assert FINISH so the test locks in cross-generation isolation.

imbajin

Blocking: yes. Summary: Two control-message failure windows can still strand work or hide a connection failure, and initialization failures can bypass the new cleanup path; latest-head Computer CI is also failing. Evidence: exact-head static interleaving analysis; local Maven validation stopped before the target test because the available JDK cannot compile computer-k8s; GitHub Actions run 30206151982 reports three unit-test failures.

imbajin · 2026-07-26T21:03:45Z

+        }
+
+        public void transportExceptionCaught(TransportException cause) {
+            CompletableFuture<Void> future = this.controlFutureRef.getAndSet(null);


‼️ send() now only places the control future in the queue (lines 88-93), while controlFutureRef stays null until the sender thread dispatches that entry at lines 254/279. A connection callback in that window therefore makes this getAndSet(null) discard the original failure; FINISH may later succeed, or the caller may wait until timeout without the real cause, even though preceding data was lost. Please reserve the future before enqueue while retaining clear-before-complete ordering, or explicitly fail queued as well as in-flight controls, and add a regression that injects the exception after send() returns but before the session method is called.

imbajin · 2026-07-26T21:03:46Z

+                });
+            } catch (TransportException e) {
+                this.completeControlFuture(future, e);
+                throw e;


‼️ After completing only this control future, rethrowing the declared TransportException reaches Sender.run(), which converts it to ComputerException and permanently terminates the sole send executor. A control already queued for another worker can then remain unresolved until its timeout; the new synchronous-failure test covers only RuntimeException, not this checked path. Please keep the executor alive after completing the affected future, or transition it to an explicit failed state that completes every queued/in-flight control future, and add a two-client regression for a synchronous TransportException here and in the FINISH path.

imbajin · 2026-07-26T21:03:46Z

-                service.execute();
-                masterFuture.complete(null);
+            try {
+                MasterService service = initMaster(args);


⚠️ initMaster()/initWorker() allocates and initializes the service before it is registered with ServiceLifecycle. If init() throws after creating BSP or network resources, neither lifecycle registration nor the inner finally runs, so the new fail-fast cleanup still cannot close that partially initialized service and CI can hang or leak resources. Please register a safely closeable service before initialization, or close it inside the helper on initialization failure, and add an init-failure cleanup test.

imbajin

Blocking: yes. Summary: Two fail-fast cleanup paths can still extend CI stalls and hide the original service failure. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 7/7 on JDK 11; git diff --check passed; visible exact-head checks were green.

imbajin · 2026-07-27T06:40:31Z

+        Throwable failure = null;
+        for (Thread thread : threads) {
+            try {
+                thread.join(SERVICE_WAIT_TIMEOUT);


⚠️ join(SERVICE_WAIT_TIMEOUT) applies the full 310-second budget separately to every thread. testMultiWorkers() can therefore spend 3 × 310 seconds on workers and another 310 seconds on the master—over 20 minutes—before cleanup finally fails, which defeats the new fail-fast behavior. Please keep the fix local: compute one absolute deadline in closeServicesAndJoin(), pass only the remaining milliseconds to the existing join helper, and add one short-timeout multi-thread regression. This does not need a new lifecycle abstraction or concurrent shutdown model.

imbajin · 2026-07-27T06:40:31Z

        } finally {
-            workerServiceRef.get().close();
-            masterServiceRef.get().close();
+            closeServicesAndJoin(lifecycle, Arrays.asList(workerThread),


⚠️ If waitForServices() throws and closeServicesAndJoin() also throws from this finally, Java replaces the original service failure with the cleanup exception; the same pattern appears in the other integration cases. That hides the root cause precisely when teardown also fails. Please keep the change narrow: preserve the primary exception at these existing call sites and attach any cleanup failure with addSuppressed(), plus one focused regression where both operations fail. No new exception hierarchy or lifecycle state is needed.

imbajin

Blocking: yes. Summary: A synchronous data-path failure can still terminate the sole sender and strand queued control futures, while two new cleanup tests can leak non-daemon threads on an early assertion failure. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 7/7 on JDK 11; git diff --check passed; visible exact-head checks are green.

imbajin · 2026-07-27T22:36:21Z

+            }
        }

        public boolean sendDataMessage(QueuedMessage message)


‼️ A synchronous failure on this data path still terminates the sole send executor without resolving control work. TransportClient.send() declares TransportException and the underlying session/send function can also throw unchecked state failures; either escapes the loop, while a FINISH future already reserved behind that data message remains queued and incomplete. Please transition the sender to an explicit failed state that exceptionally completes every queued/in-flight control future (or otherwise keep the executor alive while propagating the failure), and add regressions for both synchronous TransportException and RuntimeException with a queued FINISH.

imbajin · 2026-07-27T22:36:21Z

+        });
+        lifecycle.registerWorker(() -> closed.set(true));
+        thread.start();
+        Assert.assertTrue(started.await(1, TimeUnit.SECONDS));


⚠️ These cleanup regressions can leak the very non-daemon threads they create when a startup assertion fails. This thread is started before the one-second await, but cleanup is not in a finally; the worker/master test below has the same gap before lines 271-272. Under scheduler delay, the failed assertion leaves a thread sleeping indefinitely and can hang or contaminate the remaining suite. Please put unconditional interrupt plus bounded join cleanup in finally blocks around both tests.

imbajin

Blocking: yes. Summary: An uncaught master Error can leave the service future pending until the five-minute timeout; two previously reported sender and cleanup risks also remain covered by existing threads. Evidence: six independent exact-head review lanes, static control-flow and interleaving analysis, QueuedMessageSenderTest passed 9/9, and visible exact-head checks are green.

imbajin · 2026-07-28T10:39:58Z

+                } finally {
+                    closeMaster(service);
+                }
            } catch (Exception e) {


⚠️ initializeService() explicitly rethrows Error after cleanup (lines 602-608), but this master thread only catches Exception. An Error from initMaster(), execute(), or closeMaster() therefore terminates the daemon thread without completing masterFuture; waitForServices() reports only a generic timeout after 310 seconds. Please catch Throwable here, as the other service threads do, complete masterFuture exceptionally with the original cause, and add a focused regression where the master task throws an Error and fails immediately.

fix: prevent sender control future race

abca23b

dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. bug Something isn't working labels Jul 25, 2026

fix ci

1f28ea1

imbajin requested a review from Copilot July 25, 2026 13:52

Copilot started reviewing on behalf of imbajin July 25, 2026 13:57 View session

Copilot AI reviewed Jul 25, 2026

View reviewed changes

imbajin reviewed Jul 25, 2026

View reviewed changes

fix: harden sender control lifecycle

6672de9

dosubot Bot added size:XL This PR changes 500-999 lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Jul 26, 2026

imbajin reviewed Jul 26, 2026

View reviewed changes

Fix sender control future and service cleanup races

3d8441a

imbajin reviewed Jul 26, 2026

View reviewed changes

lokidundun added 2 commits July 27, 2026 09:24

fix ci

187a83f

delete irrelavant test

ff76512

dosubot Bot added size:XXL This PR changes 1000+ lines, ignoring generated files. and removed size:XL This PR changes 500-999 lines, ignoring generated files. labels Jul 27, 2026

imbajin reviewed Jul 27, 2026

View reviewed changes

chagne the exception

61bbf71

imbajin reviewed Jul 27, 2026

View reviewed changes

fix: handle sender data path failures

ca692c6

imbajin reviewed Jul 28, 2026

View reviewed changes

fix: complete master future on errors

99e825d

Uh oh!

Conversation

lokidundun commented Jul 25, 2026

Purpose of the PR

Main Changes

Tests

Verifying these changes

Does this PR potentially affect the following parts?

Documentation Status

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

imbajin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants