Avoid losing shutdown RPC handle

[yuandrew]

What was changed

• Make shutdown initiation and RPC-handle publication atomic.
• Distinguish between an RPC that has not started and one whose join handle was already taken.
• Ensure concurrent shutdown() callers all wait for the same in-flight ShutdownWorker RPC.
• Serialize repeated shutdown calls because downstream shutdown routines consume one-shot handles.
• Add regression tests for concurrent shutdown and cancellation of the caller that took the RPC handle.

Why?

Graceful poll shutdown requires the ShutdownWorker RPC to complete before Core waits for local polls to drain.

Before, concurrent initiate_shutdown() and shutdown() calls could expose the shutdown state before the RPC handle was stored. Concurrent shutdown() calls could also race to take the handle, allowing one caller to begin draining polls without waiting for the RPC.

If the caller holding the join handle was cancelled, dropping the Tokio JoinHandle detached the RPC task, leaving subsequent callers without a handle to await. A shared completion signal now lets those callers wait for the original RPC.

Without this ordering, shutdown can stall until server poll timeouts and can recreate the graceful-poll shutdown deadlock this RPC is intended to prevent.

Original report: #1255 (comment)

Checklist

1. Closes

2. How was this tested:

wrote a few tests that verify that concurrent and replacement shutdown callers wait for the same single in-flight ShutdownWorker RPC, even when the caller owning its join handle is cancelled.

3. Any docs updates needed?

---

Note

Medium Risk
Changes ordering and synchronization on the core worker shutdown path that graceful poll shutdown depends on; risk is mitigated by focused regression tests but incorrect behavior could still stall shutdown until poll timeouts.

Overview
Fixes races where concurrent shutdown() / initiate_shutdown() could start poll draining before the ShutdownWorker RPC finished, or lose the RPC entirely when the task that held its JoinHandle was cancelled.

Worker shutdown replaces the single optional RPC join handle with ShutdownRpcState (not started vs started, handle may already be taken), a CancellationToken that fires when the RPC task completes (via drop guard), and a shutdown_complete flag so repeat shutdown() calls are safe and serialized. Callers that do not get the handle still block on that completion signal before draining pollers.

Tests add blocked-RPC scenarios for concurrent shutdown callers and for aborting the first shutdown() waiter; one activity shutdown test now asserts PollError::ShutDown and awaits the shutdown future instead of only draining the poller.

^{Reviewed by Cursor Bugbot for commit 684ced8. Bugbot is set up for automated code reviews on this repo. Configure here.}

[jmaeagle99]

Should this short circuit be move up just under the lock acquisition so that other callers do not do the same checks and logging?