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.

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

[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);
    }

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

⚠️ Dropping the early return is the right fix, but the cleanup it unblocks stops at the first manager that fails. Managers.closeAll() (computer/computer-core/src/main/java/org/apache/hugegraph/computer/core/manager/Managers.java:69) calls manager.close(config) in a bare loop with no per-manager isolation, and on exactly this partial-init path DataServerManager.close(config) calls address() -> TransportConnectionManager.getServer(), which throws when the transport server was never bound.

Evidence — harness compiled against this head (registers DataServerManager holding a never-started TransportConnectionManager, plus one more manager behind it, then calls closeAll):

closeAll threw: java.lang.IllegalArgumentException: The TransportServer has not been initialized yet
later manager closed = false

In initManagers() the managers registered after DataServerManager are DataClientManager, SendSortManager, MessageSendManager, SnapshotManager and WorkerInputManager. So when initAll() fails before the data server binds (transport.server_port already in use is the obvious trigger), none of them are closed — and DataClientManager.close() is what calls sender.close(), i.e. the QueuedMessageSender send-executor this PR hardens is one of the threads left running. The surrounding catch (Exception e) swallows the IllegalArgumentException, so nothing in the log says cleanup stopped halfway.

Requested change: make Managers.closeAll() close every manager, wrapping each manager.close(config) in its own try/catch and rethrowing the first failure with the rest attached as suppressed (or, narrower, make DataServerManager.close() a no-op when the server never bound). Please add a regression that fails initAll() before DataServerManager binds and asserts the managers registered after it were still closed.

[bitflicker64]

⚠️ recordFatal() stores the error but leaves every control future the channels are already holding unresolved, so the new fail-fast path only takes effect on the next caller. When the executor dies here, an in-flight or queued START/FINISH is never completed, and MessageSendManager.sendControlMessageToWorkers() is already parked in future.get(timeout, MILLISECONDS) — the this.sender.checkFatal() calls added to startSend()/finishSend() run only after that wait returns, so they can neither shorten it nor surface the real cause.

Evidence — probe compiled against this head: one channel whose client always returns false from send() (so the executor parks in waitAnyClientNotBusy()), one queued data message plus a FINISH, then a spurious interrupt with closed == false:

executor state before interrupt = WAITING
ComputerException: Interrupted when waiting any client not busy
    at QueuedMessageSender.waitAnyClientNotBusy(QueuedMessageSender.java:256)
    at QueuedMessageSender$Sender.run(QueuedMessageSender.java:206)
executor alive after fatal = false
FINISH future done        = false
checkFatal()              = Send-executor encountered fatal error
FINISH still incomplete after 3004 ms

With defaults the caller therefore blocks transport.sync_request_timeout = 10s for START and 10_000 * transport.max_pending_requests(8) = 80s for FINISH (TransportConf.timeoutFinishSession()), then reports Timeout(80000ms) to wait for controlling message(FINISH) to finished rather than the fatal error that was recorded 80 seconds earlier. close() has the same gap: it sets closed = true and joins without resolving pending control futures.

Requested change: when the first fatal error is recorded (and on the close() path), walk channels and hand the error to each one — channel.failDataSend(error) already fails the registered control future and poisons the channel against later ones — so sendControlMessageToWorkers() returns immediately with the real cause. Please add a regression that terminates the send-executor with a FINISH already queued and asserts the future completes exceptionally with the recorded fatal error.

[bitflicker64]

🧹 This null check and the package-private WorkerService(Bsp4Worker) constructor at line 91 exist only so WorkerServiceTest#testInitFailsAfterRegistration can inject a mock; initManagers() was widened from private to package-private at line 320 for the same test. Two consequences worth avoiding: production init() now carries a test-only branch, and because bsp4Worker is never reset to null anywhere in this class, an instance whose init() threw — which leaves inited == false, so the E.checkArgument(!this.inited, ...) guard at line 102 still permits a retry — would re-init() against the already-closed Bsp4Worker from the failed attempt instead of building a fresh one.

Requested change: keep init() constructing the Bsp4Worker unconditionally through a package-private factory method (e.g. Bsp4Worker newBsp4Worker() { return new Bsp4Worker(this.config, this.workerInfo); }) and have the test override it — the test already subclasses WorkerService to override initManagers(), so no extra seam is needed and the production path keeps no test-only branch.