Fix operators showing "Running" state after workflow completion

[shengquan-ni]

This PR fixes a bug where the frontend incorrectly shows an operator as still running, even though the backend execution has completed.

Timeline for this bug

This issue was originally identified and fixed in #2411. However, it re-emerged after we migrated our RPC layer to gRPC in #2950. A minor mistake during the migration caused the originally chained future to be overridden by an EmptyResponse as the return value.

Root Cause

The core issue is that the controllerInitiatedQueryStats call returns immediately instead of waiting for the worker's response. Previously, we relied on this call to collect worker stats after execution, and used the updated stats to infer workflow completion. Because the call now returns immediately, the stats are never updated.

Additionally, our current design infers region completion from port completion and workflow completion from region completion. This decouples workflow execution state from the actual state of the workers. As a result, a workflow may be marked as complete even though the frontend still shows an operator as running, when in fact, the worker has already finished execution.

Example Scenario

Here’s how the issue manifests when the last operator finishes:

1. A port finishes, triggering a handler that marks the port as completed.
2. A QueryStats control message is sent to the worker that reported port completion (response pending).
3. We check if all ports in the region are complete—if so, the region is marked complete.
4. Separately, we receive a worker execution completion event and send another QueryStats message (response also pending).
5. Since all ports are marked complete, the workflow is marked complete and the entire execution is terminated.
6. Both pending QueryStats messages are lost because the worker is already shut down.

The Fix

The solution is straightforward: correctly chain and return the future so that the control message awaits the worker’s response before proceeding.

How the fix is verified

The bug is not always reproducible because sometimes the execution is killed after updating the states. So I looked at the WebSocket event sequence. In the correct behavior, setting the workflow state to complete should be after all the stats updates.

Before the fix:

After the fix:

[bobbai00]

Impressive PR description! LGTM!