fix: prevent sender control future race

[lokidundun]

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?

• Nope
• Dependencies (add/update license info)
• Modify configurations
• The public API
• Other affects (typed here)

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.

[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]

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]

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]

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]

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]

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]

Blocking: yes. Summary: A data-send failure can still be forgotten before FINISH is registered, allowing the step to succeed after silently dropping a message; the new timeout regression also has a minor timing-flake window. Evidence: six independent exact-head review lanes; static MessageSendManager/QueuedMessageSender interleaving analysis; targeted QueuedMessageSenderTest passed 9/9; git diff --check passed; visible exact-head checks are green.

[imbajin]

Blocking: yes. Summary: A control-completion race can silently discard transport failures, and service cleanup errors can still be hidden after successful execution. Evidence: six independent exact-head review lanes; targeted QueuedMessageSenderTest passed 12/12 on JDK 11; git diff --check passed; visible exact-head checks are green.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

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

• The local variable transoprtServerPort has a spelling typo. Keeping the typo in new code makes it harder to search/understand and propagates the misspelling.

        int transoprtServerPort = 8998;

[imbajin]

Blocking: yes. Summary: The new public control-message constructor can produce a future that is silently dropped by the existing queued-message send path. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 13/13 on JDK 11; git diff --check passed; visible exact-head checks are green.

[imbajin]

Blocking: yes. Summary: The partial WorkerService initialization cleanup remains covered by an existing review thread, and the new transport-exception tests do not verify per-connection isolation. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 13/13 on JDK 11; git diff --check passed; visible exact-head checks are green.

[imbajin]

🧹 Both ControlFutureClient instances inherit the same localhost:8080 connectionId from MockTransportClient. Because transportExceptionCaught() notifies every channel whose ID matches, injecting a worker-1 failure also marks worker 2 failed, but these tests never exercise worker 2 afterward; an implementation that broadcasts failures across workers would therefore still pass. Please give each test client a distinct ConnectionId and assert that START/FINISH on the unaffected worker still succeeds after the injected exception.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (3)

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

• This changes the behavior for conflicting consecutive control messages from throwing immediately (previous newFuture() threw a ComputerException) to returning a CompletableFuture already completed exceptionally (via setControlFuture). That’s a semantic change for callers that may not inspect or await the returned future. Consider either (a) keeping the immediate exception behavior for conflicts, or (b) documenting this explicitly on send(int, MessageType) to make it clear that errors may be reported only via the returned future.

    public CompletableFuture<Void> send(int workerId, MessageType type)
                                        throws InterruptedException {
        WorkerChannel channel = this.channels[channelId(workerId)];
        CompletableFuture<Void> future = new CompletableFuture<>();
        if (!channel.setControlFuture(future)) {
            return future;
        }
        /*
         * Control message just need message type is enough,
         * partitionId = -1 and buffer = null represents a meaningless value
         */
        try {
            channel.queue.put(new QueuedMessage(-1, type, null, future));
        } catch (InterruptedException e) {
            channel.completeControlFuture(future, e);
            throw e;
        }
        return future;
    }

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

• The error message "The origin future must be null" is hard to interpret, especially now that it is surfaced via an exceptionally completed future instead of being thrown. A clearer message like “Another control message is already in-flight for this worker” (and ideally including worker/channel context) would make debugging much easier.

            ComputerException e = new ComputerException(
                                  "The origin future must be null");
            future.completeExceptionally(e);
            return false;

computer/computer-test/src/main/java/org/apache/hugegraph/computer/core/sender/QueuedMessageSenderTest.java:364

• Many of these new concurrency tests rely on hard-coded 1-second waits/gets, which can be flaky under slower CI or high load. Consider centralizing time budgets into a named constant (similar to TEST_THREAD_JOIN_TIMEOUT in the integration test) and using a slightly larger default to improve test stability.

    private static boolean await(CountDownLatch latch) throws InterruptedException {
        return latch.await(1, TimeUnit.SECONDS);
    }