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

• Associate each queued START/FINISH message with its own control future.
• Keep the sender executor alive when synchronous transport or runtime failures occur.
• Preserve data-send failures so queued or later FINISH futures fail instead of timing out or succeeding silently.
• Prevent stale control callbacks and CAS races from clearing or completing a newer control future.
• Reject control messages submitted through the data-message API and invalid message types submitted through the control API.
• Use distinct connection IDs in multi-worker regression tests.
• Reduce integration-test BSP timeouts from 24 hours to 30 seconds.
• Register services before initialization and safely clean up partially initialized services.
• Preserve the original failure and attach cleanup failures as suppressed exceptions.
• Keep the existing slow/busy-client integration coverage while removing redundant lifecycle-test infrastructure.

sequenceDiagram
    autonumber
    participant Master
    participant ServiceThread as ServiceThread(init)
    participant MainThread as MainThread(close)

    ServiceThread->>Master: workerInitDone()
    Note over Master: master 把该 worker 计入名单，<br/>开始等待它的 workerCloseDone()
    ServiceThread->>ServiceThread: registered = true
    ServiceThread->>ServiceThread: connectToWorkers() 失败!
    ServiceThread-->>ServiceThread: init() 抛出异常 (inited 仍为 false)
    MainThread->>MainThread: close() 被调用
    MainThread->>MainThread: OLD: if (this.inited) → false → 跳过 workerCloseDone()
    Master->>Master: 等 workerCloseDone()，最长 24h
    Note over Master: ⚠️ 死锁 / 超时

Loading

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 7 out of 7 changed files in this pull request and generated no new comments.

Suppressed comments (1)

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

• The master options builder sets rpcServerPort twice (8611 and then 0). This is redundant and makes it harder to tell which port is actually used; convertToMap() will silently let the later value win. Please remove the fixed port and keep only the ephemeral port (0).

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

[imbajin]

Blocking: yes. Summary: Partial worker initialization can now omit the BSP close signal, and the integration tests have reverted to unbounded 24-hour BSP waits on failure. Evidence: six independent exact-head review lanes; QueuedMessageSenderTest passed 15/15 on JDK 11; git diff --check passed; visible exact-head checks are green.

[imbajin]

Blocking: yes. Summary: Two test changes weaken coverage of sender backpressure and executor shutdown behavior. Evidence: six independent exact-head review lanes; targeted QueuedMessageSenderTest passed 15/15 on JDK 11; git diff --check passed; visible exact-head checks are green.

[imbajin]

Blocking: yes. Summary: The partial-initialization regression test bypasses the production registration transition, so it cannot catch the lifecycle bug returning. Evidence: five completed independent exact-head review lanes; one additional lane ended without a result; QueuedMessageSenderTest passed 15/15 on JDK 11; git diff --check passed; visible exact-head checks are green.

[imbajin]

Blocking: no. Summary: The sender control-future and partial worker-initialization paths are consistent at this head, with no actionable findings after independent review. Evidence: six exact-head review lanes; QueuedMessageSenderTest 15/15 and WorkerServiceTest#testInitFailsAfterRegistration 1/1 passed on JDK 11; git diff --check passed; all visible exact-head checks are green. Score: 8.6/10.

[bitflicker64]

Blocking: no. Summary: The control-future rework is sound and its 15 unit regressions pass locally at this head, but two gaps remain on the failure paths the PR targets: partially initialized worker managers are still only half closed because Managers.closeAll() aborts at the first failing manager, and a fatal send-executor error leaves already-queued control futures unresolved so callers still wait the full 10s/80s transport timeout and get a misleading Timeout(...) instead of the recorded cause. Evidence: mvn install -DskipTests then java -cp <computer-test classpath> org.junit.runner.JUnitCore org.apache.hugegraph.computer.core.sender.QueuedMessageSenderTest -> OK (15 tests); WorkerServiceTest#testInitFailsAfterRegistration -> RESULT run=1 failures=0; two standalone probes compiled against 8492fd4 reproduce the two gaps (outputs quoted inline).

[bitflicker64]

Blocking: no. Summary: The latest commit closes the queued-control-future and partial-cleanup gaps from the previous round, but two small failure-path gaps remain: an unexpected interrupt on an idle send-executor still exits without recording a fatal error, and WorkerInputManager still throws an NPE when closed after a partial init. Evidence: JDK 11 test-compile of computer-test at 5189d79; QueuedMessageSenderTest OK (17 tests); DataServerManagerTest#testCloseManagersAfterPartialInitFailure and WorkerServiceTest#testInitFailsAfterRegistration pass; two standalone probes against this head reproduce both gaps (outputs inline); all checks on this head are green.

[codecov]

Codecov Report

❌ Patch coverage is 86.48649% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 85.32%. Comparing base (e0b484a) to head (afabe1d).
⚠️ Report is 58 commits behind head on master.

Files with missing lines	Patch %	Lines
...raph/computer/core/sender/QueuedMessageSender.java	85.81%	13 Missing and 8 partials ⚠️
...ache/hugegraph/computer/core/manager/Managers.java	81.81%	1 Missing and 1 partial ⚠️
.../hugegraph/computer/core/master/MasterService.java	0.00%	0 Missing and 1 partial ⚠️
.../hugegraph/computer/core/worker/WorkerService.java	90.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #357      +/-   ##
============================================
+ Coverage     85.03%   85.32%   +0.29%     
- Complexity     3296     3401     +105     
============================================
  Files           349      362      +13     
  Lines         12485    12839     +354     
  Branches       1130     1186      +56     
============================================
+ Hits          10616    10955     +339     
- Misses         1329     1352      +23     
+ Partials        540      532       -8     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.