Activity poll can hang during shutdown when ShutdownWorker returns UNIMPLEMENTED #1297

#[tokio::test]
async fn graceful_activity_poll_shutdown_handles_unimplemented_shutdown_worker() {
    let activity_poll_started = Arc::new(Notify::new());
    let activity_poll_started_clone = activity_poll_started.clone();
    let shutdown_worker_called = Arc::new(Notify::new());
    let shutdown_worker_called_clone = shutdown_worker_called.clone();

    let mut mock_client = mock_manual_worker_client();
    mock_client
        .expect_describe_namespace()
        .times(1)
        .returning(|| {
            async {
                Ok(DescribeNamespaceResponse {
                    namespace_info: Some(NamespaceInfo {
                        capabilities: Some(Capabilities {
                            worker_poll_complete_on_shutdown: true,
                            ..Default::default()
                        }),
                        ..Default::default()
                    }),
                    ..Default::default()
                })
            }
            .boxed()
        });
    mock_client
        .expect_shutdown_worker()
        .times(1)
        .returning(move |_, _, _, _| {
            let shutdown_worker_called = shutdown_worker_called_clone.clone();
            async move {
                shutdown_worker_called.notify_one();
                Err(tonic::Status::unimplemented(
                    "ShutdownWorker disabled by server",
                ))
            }
            .boxed()
        });
    mock_client
        .expect_poll_activity_task()
        .times(1)
        .returning(move |_, _| {
            let activity_poll_started = activity_poll_started_clone.clone();
            async move {
                activity_poll_started.notify_one();
                future::pending::<Result<PollActivityTaskQueueResponse, tonic::Status>>().await
            }
            .boxed()
        });
    mock_client
        .expect_record_activity_heartbeat()
        .returning(|_, _| async { Ok(RecordActivityTaskHeartbeatResponse::default()) }.boxed());

    let mut cfg = test_worker_cfg()
        .activity_task_poller_behavior(PollerBehavior::SimpleMaximum(1_usize))
        .build()
        .unwrap();
    cfg.task_types = WorkerTaskTypes::activity_only();
    let worker = Worker::new_test(cfg, mock_client);
    worker.validate().await.unwrap();

    let poll_fut = async { worker.poll_activity_task().await };
    let shutdown_fut = async {
        activity_poll_started.notified().await;
        worker.initiate_shutdown();
        shutdown_worker_called.notified().await;
    };

    let (poll_result, _) = tokio::time::timeout(Duration::from_millis(500), async {
        tokio::join!(poll_fut, shutdown_fut)
    })
    .await
    .expect("activity poll remained pending after shutdown_worker returned UNIMPLEMENTED");

    assert_matches!(poll_result.unwrap_err(), PollError::ShutDown);
}

use std::{future, sync::Arc, time::Duration};
use temporalio_common::protos::temporal::api::{
    namespace::v1::{NamespaceInfo, namespace_info::Capabilities},
    workflowservice::v1::{
        DescribeNamespaceResponse, PollActivityTaskQueueResponse,
        RecordActivityTaskHeartbeatResponse,
    },
};
use temporalio_common::worker::WorkerTaskTypes;
use tokio::sync::Notify;