#[cfg(test)]
mod tests {
    use std::fs;
    use std::pin::Pin;
    use std::sync::Arc;
    use std::sync::OnceLock;
    use std::time::{Duration, SystemTime, UNIX_EPOCH};

    use pd_proto_grpc::pdpb;
    use pd_proto_grpc::tsopb;
    use pd_service_tso::{
        AllocatorConfig, InMemoryEtcdStore, KeyspaceGroupManager, LegacyForwarder,
        LegacyPdGrpcService, TsoService,
        grpc::{
            LEGACY_FORWARD_FAILPOINT_METADATA_KEY, LEGACY_FORWARD_FAILPOINT_RECV_TIMEOUT_VALUE,
            LEGACY_FORWARD_FAILPOINT_SEND_TIMEOUT_VALUE, TsoGrpcService,
        },
    };
    use pd_test_util::env::{EnvVarRestore, async_env_lock, remove_env, set_env};
    use regex::Regex;
    use tokio::sync::Barrier;
    use tokio::sync::RwLock;
    use tokio_stream::Stream;
    use tokio_stream::wrappers::TcpListenerStream;
    use tonic::transport::{Channel, Server};
    use tonic::{Code, Request, Response, Status, Streaming};

    const NEXTGEN_DEFAULT_BOOTSTRAP_KEYSPACE_ID: u32 = 16_777_214;
    const LEGACY_CLIENT_MANIFEST_PATH: &str = "legacy-client-version-manifest.yaml";
    const TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN: &str = "tso count should be positive";
    const LEGACY_PROXY_MAX_CONCURRENT_TOKEN: &str =
        "max count of concurrent tso proxy routines exceeded";
    const LEGACY_PROXY_RECV_TIMEOUT_TOKEN: &str =
        "tso proxy timeout when receiving from client; stream closed by server";
    const LEGACY_FORWARD_OVERFLOW_TOKEN: &str = "merged tso request count overflow";
    const LEGACY_PROXY_IDLE_TIMEOUT_TOKEN: &str = "TSOProxyStreamIdleTimeout";
    const LEGACY_PROXY_NOT_FOUND_TOKEN: &str = "not found tso address";
    const LEGACY_PROXY_FORWARD_TIMEOUT_TOKEN: &str = "forward tso request timeout";
    const DEFAULT_CLUSTER_ID: u64 = 1;

    struct ReleaseOnDrop(Option<tokio::sync::oneshot::Sender<()>>);

    impl ReleaseOnDrop {
        fn new(sender: tokio::sync::oneshot::Sender<()>) -> Self {
            Self(Some(sender))
        }
    }

    impl Drop for ReleaseOnDrop {
        fn drop(&mut self) {
            if let Some(sender) = self.0.take() {
                let _ = sender.send(());
            }
        }
    }

    #[derive(Debug, Clone, PartialEq, Eq)]
    struct LegacyClientManifestEntry {
        version: String,
        endpoint_env: String,
    }

    fn default_group() -> pd_proto_tso::KeyspaceGroup {
        pd_proto_tso::KeyspaceGroup {
            id: 0,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: Vec::new(),
            keyspaces: vec![0, NEXTGEN_DEFAULT_BOOTSTRAP_KEYSPACE_ID],
        }
    }

    fn secondary_group() -> pd_proto_tso::KeyspaceGroup {
        pd_proto_tso::KeyspaceGroup {
            id: 1,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: Vec::new(),
            keyspaces: vec![1],
        }
    }

    fn default_group_with_members(member_urls: &[&str]) -> pd_proto_tso::KeyspaceGroup {
        pd_proto_tso::KeyspaceGroup {
            id: 0,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: member_urls
                .iter()
                .map(|url| pd_proto_tso::KeyspaceGroupMember {
                    address: (*url).to_string(),
                    is_primary: false,
                })
                .collect(),
            keyspaces: vec![0, NEXTGEN_DEFAULT_BOOTSTRAP_KEYSPACE_ID],
        }
    }

    fn parse_manifest_value(line: &str, key: &str) -> Option<String> {
        let mut normalized = line.trim();
        if let Some(stripped) = normalized.strip_prefix("- ") {
            normalized = stripped.trim();
        }
        let (parsed_key, parsed_value) = normalized.split_once(':')?;
        if parsed_key.trim() != key {
            return None;
        }
        let raw_value = parsed_value.trim();
        if raw_value.is_empty() {
            return None;
        }
        Some(raw_value.trim_matches('"').to_string())
    }

    fn load_legacy_client_manifest() -> Result<Vec<LegacyClientManifestEntry>, String> {
        let manifest_path = format!(
            "{}/{}",
            env!("CARGO_MANIFEST_DIR"),
            LEGACY_CLIENT_MANIFEST_PATH
        );
        let content = fs::read_to_string(&manifest_path)
            .map_err(|err| format!("read legacy client manifest failed: {err}"))?;
        let mut entries = Vec::new();
        let mut pending_version: Option<String> = None;
        let mut pending_endpoint_env: Option<String> = None;

        for raw_line in content.lines() {
            let line = raw_line.trim();
            if line.is_empty() || line.starts_with('#') || line == "legacy_clients:" {
                continue;
            }
            if let Some(version) = parse_manifest_value(line, "version") {
                if pending_version.is_some() || pending_endpoint_env.is_some() {
                    let version = pending_version.take().ok_or_else(|| {
                        "legacy client manifest entry missing `version`".to_string()
                    })?;
                    let endpoint_env = pending_endpoint_env.take().ok_or_else(|| {
                        format!("legacy client manifest entry `{version}` missing `endpoint_env`")
                    })?;
                    entries.push(LegacyClientManifestEntry {
                        version,
                        endpoint_env,
                    });
                }
                pending_version = Some(version);
                continue;
            }
            if let Some(endpoint_env) = parse_manifest_value(line, "endpoint_env") {
                pending_endpoint_env = Some(endpoint_env);
            }
        }

        if pending_version.is_some() || pending_endpoint_env.is_some() {
            let version = pending_version
                .take()
                .ok_or_else(|| "legacy client manifest entry missing `version`".to_string())?;
            let endpoint_env = pending_endpoint_env.take().ok_or_else(|| {
                format!("legacy client manifest entry `{version}` missing `endpoint_env`")
            })?;
            entries.push(LegacyClientManifestEntry {
                version,
                endpoint_env,
            });
        }

        if entries.is_empty() {
            return Err("legacy client manifest is empty".to_string());
        }
        Ok(entries)
    }

    fn allow_missing_go_endpoint() -> bool {
        std::env::var("PD_ALLOW_MISSING_GO_TSO_ENDPOINT")
            .ok()
            .as_deref()
            == Some("1")
    }

    fn resolve_release_gate_go_endpoints(
        manifest: &[LegacyClientManifestEntry],
        allow_missing: bool,
    ) -> Vec<(String, String)> {
        let mut endpoints = Vec::new();
        let mut missing = Vec::new();
        for entry in manifest {
            match std::env::var(&entry.endpoint_env) {
                Ok(value) if !value.trim().is_empty() => {
                    endpoints.push((entry.version.clone(), value.trim().to_string()));
                }
                _ => missing.push(entry.endpoint_env.clone()),
            }
        }
        if !missing.is_empty() && !allow_missing {
            panic!(
                "missing required Go endpoint env(s): {}; set PD_ALLOW_MISSING_GO_TSO_ENDPOINT=1 to skip locally",
                missing.join(", ")
            );
        }
        endpoints
    }

    async fn assert_direct_vs_legacy_differential_once() {
        let (endpoint_direct, _service_direct) = start_service().await;
        let (endpoint_legacy, _service_legacy) = start_service().await;

        // Success path parity.
        let direct_ok = direct_tso_once(&endpoint_direct, 1, 5)
            .await
            .expect("direct success");
        let legacy_ok = legacy_tso_once(&endpoint_legacy, 1, 5)
            .await
            .expect("legacy success");
        assert_eq!(direct_ok.count, legacy_ok.count);
        assert_eq!(
            direct_ok.timestamp.as_ref().expect("direct ts").logical,
            legacy_ok.timestamp.as_ref().expect("legacy ts").logical,
        );

        // Cluster-mismatch transport parity.
        let direct_cluster_mismatch = direct_tso_once(&endpoint_direct, 999, 1)
            .await
            .expect_err("direct mismatch must fail");
        let legacy_cluster_mismatch = legacy_tso_once(&endpoint_legacy, 999, 1)
            .await
            .expect_err("legacy mismatch must fail");
        assert_eq!(direct_cluster_mismatch.code(), Code::FailedPrecondition);
        assert_eq!(legacy_cluster_mismatch.code(), Code::FailedPrecondition);
        assert_eq!(
            direct_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(DEFAULT_CLUSTER_ID, 999)
        );
        assert_eq!(
            legacy_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(DEFAULT_CLUSTER_ID, 999)
        );

        // Token-level count-validation parity.
        let direct_bad_count = direct_tso_once(&endpoint_direct, 1, 0)
            .await
            .expect_err("direct bad count must fail");
        let legacy_bad_count = legacy_tso_once(&endpoint_legacy, 1, 0)
            .await
            .expect_err("legacy bad count must fail");
        assert_eq!(direct_bad_count.code(), Code::Unknown);
        assert_eq!(legacy_bad_count.code(), Code::Unknown);
        assert_eq!(
            direct_bad_count.message(),
            TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN
        );
        assert_eq!(
            legacy_bad_count.message(),
            TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN
        );
    }

    fn expected_cluster_id_from_mismatch(message: &str) -> Option<u64> {
        static PATTERN: OnceLock<Regex> = OnceLock::new();
        let captures = PATTERN
            .get_or_init(|| {
                Regex::new(r"^mismatch cluster id, need (\d+) but got (\d+)$")
                    .expect("valid mismatch-cluster-id regex")
            })
            .captures(message)?;
        let expected = captures.get(1)?.as_str().parse().ok()?;
        let _got: u64 = captures.get(2)?.as_str().parse().ok()?;
        Some(expected)
    }

    fn pd_transport_cluster_mismatch_message(expected: u64, got: u64) -> String {
        format!("mismatch cluster id, need {expected} but got {got}")
    }

    async fn cluster_id_from_discovery_endpoint(endpoint: &str) -> Option<u64> {
        if let Ok(members) = legacy_pd_get_members_once(endpoint).await
            && let Some(cluster_id) = members
                .header
                .as_ref()
                .map(|header| header.cluster_id)
                .filter(|cluster_id| *cluster_id != 0)
        {
            return Some(cluster_id);
        }

        if let Ok(cluster) = legacy_pd_get_cluster_info_once(endpoint).await
            && let Some(cluster_id) = cluster
                .header
                .as_ref()
                .map(|header| header.cluster_id)
                .filter(|cluster_id| *cluster_id != 0)
        {
            return Some(cluster_id);
        }

        None
    }

    async fn cluster_id_from_tso_endpoint(endpoint: &str) -> u64 {
        if let Some(cluster_id) = cluster_id_from_discovery_endpoint(endpoint).await {
            return cluster_id;
        }

        let mut observations = Vec::new();
        for probe_cluster_id in [
            DEFAULT_CLUSTER_ID.saturating_sub(1),
            DEFAULT_CLUSTER_ID,
            DEFAULT_CLUSTER_ID.saturating_add(1),
            u64::MAX,
        ] {
            match direct_tso_once(endpoint, probe_cluster_id, 1).await {
                Ok(response) => {
                    let response_cluster_id = response
                        .header
                        .as_ref()
                        .map(|header| header.cluster_id)
                        .unwrap_or_default();
                    observations.push(format!(
                        "probe cluster_id={probe_cluster_id} succeeded with response cluster_id={response_cluster_id}"
                    ));
                    if response_cluster_id != 0 {
                        return response_cluster_id;
                    }
                }
                Err(status) => {
                    observations.push(format!(
                        "probe cluster_id={probe_cluster_id} failed with {status}"
                    ));
                    if let Some(cluster_id) = expected_cluster_id_from_mismatch(status.message()) {
                        return cluster_id;
                    }
                }
            }
        }

        panic!(
            "extract cluster id from {endpoint} failed; discovery APIs and mismatch probes did not yield a cluster id: {}",
            observations.join("; ")
        )
    }

    async fn assert_go_vs_rust_differential_once(
        go_endpoint: &str,
        rust_endpoint: &str,
        cluster_id: u64,
    ) {
        let mismatch_cluster_id = if cluster_id == DEFAULT_CLUSTER_ID {
            DEFAULT_CLUSTER_ID.saturating_add(1)
        } else {
            DEFAULT_CLUSTER_ID
        };
        // Success path: both endpoints must return successful responses with matching count.
        let go_ok = direct_tso_once(go_endpoint, cluster_id, 5)
            .await
            .expect("go success");
        let rust_ok = direct_tso_once(rust_endpoint, cluster_id, 5)
            .await
            .expect("rust success");
        assert_eq!(go_ok.count, rust_ok.count);
        assert!(go_ok.timestamp.is_some());
        assert!(rust_ok.timestamp.is_some());

        // Cluster mismatch status/token parity.
        let go_cluster_mismatch = direct_tso_once(go_endpoint, mismatch_cluster_id, 1)
            .await
            .expect_err("go mismatch must fail");
        let rust_cluster_mismatch = direct_tso_once(rust_endpoint, mismatch_cluster_id, 1)
            .await
            .expect_err("rust mismatch must fail");
        assert_eq!(go_cluster_mismatch.code(), rust_cluster_mismatch.code());
        assert_eq!(
            go_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(cluster_id, mismatch_cluster_id)
        );
        assert_eq!(
            rust_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(cluster_id, mismatch_cluster_id)
        );

        // Count validation status/token parity.
        let go_bad_count = direct_tso_once(go_endpoint, cluster_id, 0)
            .await
            .expect_err("go bad count must fail");
        let rust_bad_count = direct_tso_once(rust_endpoint, cluster_id, 0)
            .await
            .expect_err("rust bad count must fail");
        assert_eq!(go_bad_count.code(), rust_bad_count.code());
        assert!(
            go_bad_count
                .message()
                .contains(TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN)
        );
        assert!(
            rust_bad_count
                .message()
                .contains(TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN)
        );

        // Legacy-forwarding differential coverage.
        let go_legacy_ok = legacy_tso_once(go_endpoint, cluster_id, 3)
            .await
            .expect("go legacy success");
        let rust_legacy_ok = legacy_tso_once(rust_endpoint, cluster_id, 3)
            .await
            .expect("rust legacy success");
        assert_eq!(go_legacy_ok.count, rust_legacy_ok.count);
        assert!(go_legacy_ok.timestamp.is_some());
        assert!(rust_legacy_ok.timestamp.is_some());

        let go_legacy_cluster_mismatch = legacy_tso_once(go_endpoint, mismatch_cluster_id, 1)
            .await
            .expect_err("go legacy mismatch must fail");
        let rust_legacy_cluster_mismatch = legacy_tso_once(rust_endpoint, mismatch_cluster_id, 1)
            .await
            .expect_err("rust legacy mismatch must fail");
        assert_eq!(
            go_legacy_cluster_mismatch.code(),
            rust_legacy_cluster_mismatch.code()
        );
        assert_eq!(
            go_legacy_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(cluster_id, mismatch_cluster_id)
        );
        assert_eq!(
            rust_legacy_cluster_mismatch.message(),
            pd_transport_cluster_mismatch_message(cluster_id, mismatch_cluster_id)
        );

        let go_legacy_bad_count = legacy_tso_once(go_endpoint, cluster_id, 0)
            .await
            .expect_err("go legacy bad count must fail");
        let rust_legacy_bad_count = legacy_tso_once(rust_endpoint, cluster_id, 0)
            .await
            .expect_err("rust legacy bad count must fail");
        assert_eq!(go_legacy_bad_count.code(), rust_legacy_bad_count.code());
        assert!(
            go_legacy_bad_count
                .message()
                .contains(TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN)
        );
        assert!(
            rust_legacy_bad_count
                .message()
                .contains(TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN)
        );
    }

    fn new_core_service() -> TsoService {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(
            DEFAULT_CLUSTER_ID,
            store,
            AllocatorConfig::default(),
            "node",
        );
        manager
            .register_group(default_group())
            .expect("register group");
        manager
            .register_group(secondary_group())
            .expect("register secondary group");
        manager
            .campaign_group_primary(0, now_ms())
            .expect("campaign primary");
        manager
            .campaign_group_primary(1, now_ms())
            .expect("campaign secondary primary");
        TsoService::new(DEFAULT_CLUSTER_ID, manager)
    }

    async fn start_service() -> (String, Arc<RwLock<TsoService>>) {
        start_service_with_cluster_id(DEFAULT_CLUSTER_ID).await
    }

    async fn start_service_with_cluster_id(cluster_id: u64) -> (String, Arc<RwLock<TsoService>>) {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager =
            KeyspaceGroupManager::new(cluster_id, store, AllocatorConfig::default(), "node");
        manager
            .register_group(default_group())
            .expect("register group");
        manager
            .register_group(secondary_group())
            .expect("register secondary group");
        manager
            .campaign_group_primary(0, now_ms())
            .expect("campaign primary");
        manager
            .campaign_group_primary(1, now_ms())
            .expect("campaign secondary primary");

        let service = Arc::new(RwLock::new(TsoService::new(cluster_id, manager)));
        let grpc = TsoGrpcService::new(Arc::clone(&service));
        let legacy_pd = LegacyPdGrpcService::new(Arc::clone(&service));

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind listener");
        let addr = listener.local_addr().expect("local addr");
        let stream = TcpListenerStream::new(listener);

        tokio::spawn(async move {
            Server::builder()
                .add_service(pdpb::pd_server::PdServer::new(legacy_pd))
                .add_service(tsopb::tso_server::TsoServer::new(grpc))
                .serve_with_incoming(stream)
                .await
                .expect("serve");
        });

        (format!("http://{addr}"), service)
    }

    async fn start_proxy_service_without_primary() -> (String, Arc<RwLock<TsoService>>) {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(default_group())
            .expect("register group");
        manager
            .register_group(secondary_group())
            .expect("register secondary group");

        let service = Arc::new(RwLock::new(TsoService::new(1, manager)));
        let grpc = TsoGrpcService::new(Arc::clone(&service));
        let legacy_pd = LegacyPdGrpcService::new(Arc::clone(&service));

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind listener");
        let addr = listener.local_addr().expect("local addr");
        let stream = TcpListenerStream::new(listener);

        tokio::spawn(async move {
            Server::builder()
                .add_service(pdpb::pd_server::PdServer::new(legacy_pd))
                .add_service(tsopb::tso_server::TsoServer::new(grpc))
                .serve_with_incoming(stream)
                .await
                .expect("serve");
        });

        (format!("http://{addr}"), service)
    }

    async fn start_service_with_runtime_members(
        member_urls: &[&str],
    ) -> (String, Arc<RwLock<TsoService>>) {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(default_group_with_members(member_urls))
            .expect("register group");
        manager
            .register_group(secondary_group())
            .expect("register secondary group");
        manager
            .campaign_group_primary(0, now_ms())
            .expect("campaign primary");
        manager
            .campaign_group_primary(1, now_ms())
            .expect("campaign secondary primary");

        let service = Arc::new(RwLock::new(TsoService::new(1, manager)));
        let grpc = TsoGrpcService::new(Arc::clone(&service));
        let legacy_pd = LegacyPdGrpcService::new(Arc::clone(&service));

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind listener");
        let addr = listener.local_addr().expect("local addr");
        let stream = TcpListenerStream::new(listener);

        tokio::spawn(async move {
            Server::builder()
                .add_service(pdpb::pd_server::PdServer::new(legacy_pd))
                .add_service(tsopb::tso_server::TsoServer::new(grpc))
                .serve_with_incoming(stream)
                .await
                .expect("serve");
        });

        (format!("http://{addr}"), service)
    }

    async fn configure_legacy_proxy_runtime(
        service: &Arc<RwLock<TsoService>>,
        max_concurrent: Option<i32>,
        recv_timeout: Option<Duration>,
        idle_timeout: Option<Duration>,
    ) {
        let guard = service.read().await;
        let mut runtime = guard.runtime_config();
        if let Some(max_concurrent) = max_concurrent {
            runtime.max_concurrent_tso_proxy_streamings = max_concurrent;
        }
        if let Some(recv_timeout) = recv_timeout {
            runtime.tso_proxy_recv_from_client_timeout = recv_timeout;
        }
        if let Some(idle_timeout) = idle_timeout {
            runtime.tso_proxy_stream_idle_timeout = idle_timeout;
        }
        guard.apply_runtime_config(runtime);
    }

    async fn start_blackhole_endpoint() -> String {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind blackhole listener");
        let addr = listener.local_addr().expect("blackhole local addr");
        tokio::spawn(async move {
            while let Ok((socket, _peer)) = listener.accept().await {
                tokio::spawn(async move {
                    let _socket = socket;
                    std::future::pending::<()>().await;
                });
            }
        });
        format!("http://{addr}")
    }

    #[derive(Clone)]
    struct MockGetMinTsService {
        response: Result<tsopb::GetMinTsResponse, Status>,
    }

    type MockTsoStream = Pin<Box<dyn Stream<Item = Result<tsopb::TsoResponse, Status>> + Send>>;

    #[tonic::async_trait]
    impl tsopb::tso_server::Tso for MockGetMinTsService {
        type TsoStream = MockTsoStream;

        async fn tso(
            &self,
            _request: Request<Streaming<tsopb::TsoRequest>>,
        ) -> Result<Response<Self::TsoStream>, Status> {
            Err(Status::unimplemented("mock tso stream not implemented"))
        }

        async fn find_group_by_keyspace_id(
            &self,
            _request: Request<tsopb::FindGroupByKeyspaceIdRequest>,
        ) -> Result<Response<tsopb::FindGroupByKeyspaceIdResponse>, Status> {
            Err(Status::unimplemented("mock find-group not implemented"))
        }

        async fn get_min_ts(
            &self,
            _request: Request<tsopb::GetMinTsRequest>,
        ) -> Result<Response<tsopb::GetMinTsResponse>, Status> {
            match &self.response {
                Ok(response) => Ok(Response::new(response.clone())),
                Err(status) => Err(Status::new(status.code(), status.message().to_string())),
            }
        }
    }

    async fn start_mock_get_min_ts_service(
        response: Result<tsopb::GetMinTsResponse, Status>,
    ) -> String {
        let mock = MockGetMinTsService { response };
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind listener");
        let addr = listener.local_addr().expect("local addr");
        let stream = TcpListenerStream::new(listener);

        tokio::spawn(async move {
            Server::builder()
                .add_service(tsopb::tso_server::TsoServer::new(mock))
                .serve_with_incoming(stream)
                .await
                .expect("serve mock");
        });
        format!("http://{addr}")
    }

    async fn direct_tso_once(
        endpoint: &str,
        cluster_id: u64,
        count: u32,
    ) -> Result<tsopb::TsoResponse, Status> {
        direct_tso_once_with_route(endpoint, cluster_id, count, 0, 0).await
    }

    async fn direct_tso_once_with_route(
        endpoint: &str,
        cluster_id: u64,
        count: u32,
        keyspace_id: u32,
        keyspace_group_id: u32,
    ) -> Result<tsopb::TsoResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = tsopb::tso_client::TsoClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(tsopb::TsoRequest {
            header: Some(tsopb::RequestHeader {
                cluster_id,
                keyspace_id,
                keyspace_group_id,
                sender_id: 0,
            }),
            count,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut stream = client
            .tso(Request::new(tokio_stream::wrappers::ReceiverStream::new(
                rx,
            )))
            .await
            .map_err(|status| Status::new(status.code(), status.message().to_string()))?
            .into_inner();
        stream
            .message()
            .await?
            .ok_or_else(|| Status::unknown("empty tso response"))
    }

    async fn direct_find_group_by_keyspace_id_once(
        endpoint: &str,
        cluster_id: u64,
        keyspace_id: u32,
        keyspace_group_id: u32,
    ) -> Result<tsopb::FindGroupByKeyspaceIdResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = tsopb::tso_client::TsoClient::new(channel);
        client
            .find_group_by_keyspace_id(Request::new(tsopb::FindGroupByKeyspaceIdRequest {
                header: Some(tsopb::RequestHeader {
                    cluster_id,
                    sender_id: 0,
                    keyspace_id: 0,
                    keyspace_group_id,
                }),
                keyspace_id,
                mod_revision: 0,
            }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn legacy_tso_once(
        endpoint: &str,
        cluster_id: u64,
        count: u32,
    ) -> Result<pdpb::TsoResponse, Status> {
        let mut forwarder = LegacyForwarder::connect(endpoint.to_string()).await?;
        forwarder
            .forward_tso(pdpb::TsoRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id,
                    sender_id: 42,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
                count,
                dc_location: String::new(),
            })
            .await
    }

    async fn legacy_pd_tso_once(
        endpoint: &str,
        cluster_id: u64,
        count: u32,
    ) -> Result<pdpb::TsoResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut stream = client
            .tso(Request::new(tokio_stream::wrappers::ReceiverStream::new(
                rx,
            )))
            .await
            .map_err(|status| Status::new(status.code(), status.message().to_string()))?
            .into_inner();
        stream
            .message()
            .await
            .map_err(|status| Status::new(status.code(), status.message().to_string()))?
            .ok_or_else(|| Status::unknown("empty pdpb tso response"))
    }

    async fn legacy_pd_get_min_ts_once(
        endpoint: &str,
        cluster_id: u64,
    ) -> Result<pdpb::GetMinTsResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);
        client
            .get_min_ts(Request::new(pdpb::GetMinTsRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id,
                    sender_id: 42,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
            }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn legacy_pd_get_cluster_info_once(
        endpoint: &str,
    ) -> Result<pdpb::GetClusterInfoResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);
        client
            .get_cluster_info(Request::new(pdpb::GetClusterInfoRequest { header: None }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn legacy_pd_get_members_once(
        endpoint: &str,
    ) -> Result<pdpb::GetMembersResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);
        client
            .get_members(Request::new(pdpb::GetMembersRequest { header: None }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn legacy_pd_sync_max_ts_once(
        endpoint: &str,
        cluster_id: u64,
    ) -> Result<pdpb::SyncMaxTsResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);
        client
            .sync_max_ts(Request::new(pdpb::SyncMaxTsRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id,
                    sender_id: 42,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
                max_ts: Some(pdpb::Timestamp::default()),
                skip_check: false,
            }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn legacy_pd_get_dc_location_info_once(
        endpoint: &str,
        cluster_id: u64,
    ) -> Result<pdpb::GetDcLocationInfoResponse, Status> {
        let channel = Channel::from_shared(endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);
        client
            .get_dc_location_info(Request::new(pdpb::GetDcLocationInfoRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id,
                    sender_id: 42,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
                dc_location: String::new(),
            }))
            .await
            .map(|resp| resp.into_inner())
    }

    async fn forwarded_legacy_tso_once_with_barrier(
        proxy_endpoint: &str,
        forwarded_host: &str,
        count: u32,
        barrier: Arc<Barrier>,
    ) -> Result<pdpb::TsoResponse, Status> {
        let channel = Channel::from_shared(proxy_endpoint.to_string())
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            forwarded_host.parse().expect("forwarded host metadata"),
        );
        let mut stream = client.tso(request).await?.into_inner();

        barrier.wait().await;
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        stream
            .message()
            .await
            .map_err(|status| Status::new(status.code(), status.message().to_string()))?
            .ok_or_else(|| Status::unknown("empty pdpb tso response"))
    }

    #[tokio::test]
    async fn strict_profile_tsopb_stream_surface_matches_core() {
        let (endpoint, _service) = start_service().await;
        let core = new_core_service();

        let transport_resp = direct_tso_once(&endpoint, 1, 3).await.expect("response");
        let core_resp = core
            .tso(
                now_ms(),
                pd_proto_tso::TsoRequest {
                    header: pd_proto_tso::RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        sender_id: 0,
                    },
                    count: 3,
                    dc_location: String::new(),
                },
            )
            .expect("core response");

        assert_eq!(transport_resp.count, core_resp.count);
        assert_eq!(
            transport_resp.timestamp.expect("timestamp").logical,
            core_resp.timestamp.logical
        );
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarder_preserves_count_timestamp_and_nil_error_mapping() {
        let (endpoint, _service) = start_service().await;

        let response = legacy_tso_once(&endpoint, 1, 4).await.expect("forward");
        assert_eq!(response.count, 4);
        assert!(response.timestamp.is_some());
        assert!(
            response.header.expect("header").error.is_none(),
            "strict parity should keep nil error on success"
        );
    }

    #[tokio::test]
    async fn strict_profile_stream_count_zero_token_parity_direct_and_legacy() {
        let (endpoint, _service) = start_service().await;

        let direct_err = direct_tso_once(&endpoint, 1, 0)
            .await
            .expect_err("direct count=0 must fail");
        assert_eq!(direct_err.code(), Code::Unknown);
        assert_eq!(direct_err.message(), TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN);

        let legacy_err = legacy_tso_once(&endpoint, 1, 0)
            .await
            .expect_err("legacy count=0 must fail");
        assert_eq!(legacy_err.code(), Code::Unknown);
        assert_eq!(legacy_err.message(), TSO_COUNT_SHOULD_BE_POSITIVE_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_get_min_ts_unary_surface_is_handled_response() {
        let (endpoint, _service) = start_service().await;
        let mut forwarder = LegacyForwarder::connect(endpoint).await.expect("forwarder");

        let response = forwarder
            .forward_get_min_ts(pdpb::GetMinTsRequest {
                header: Some(pdpb::RequestHeader {
                    cluster_id: 1,
                    sender_id: 1,
                    caller_id: String::new(),
                    caller_component: String::new(),
                }),
            })
            .await
            .expect("get min ts");
        assert!(response.timestamp.is_some());
        assert_eq!(response.header.expect("header").cluster_id, 1);
    }

    #[tokio::test]
    async fn strict_profile_pdpb_server_path_supports_tso_and_get_min_ts() {
        let (endpoint, _service) = start_service().await;

        let tso = legacy_pd_tso_once(&endpoint, 1, 4).await.expect("pdpb tso");
        assert_eq!(tso.count, 4);
        assert!(tso.timestamp.is_some());
        assert!(tso.header.expect("header").error.is_none());

        let min_ts = legacy_pd_get_min_ts_once(&endpoint, 1)
            .await
            .expect("pdpb get min ts");
        assert!(min_ts.timestamp.is_some());
        assert!(min_ts.header.is_some());
    }

    #[tokio::test]
    async fn strict_profile_pdpb_get_min_ts_cluster_mismatch_returns_failed_precondition() {
        let (endpoint, _service) = start_service().await;

        let err = legacy_pd_get_min_ts_once(&endpoint, 999)
            .await
            .expect_err("cluster mismatch must fail via transport status");
        assert_eq!(err.code(), Code::FailedPrecondition);
        assert_eq!(
            err.message(),
            pd_transport_cluster_mismatch_message(DEFAULT_CLUSTER_ID, 999)
        );
    }

    #[tokio::test]
    async fn strict_profile_pdpb_get_min_ts_fanout_aggregates_minimum_timestamp() {
        let backend_a = start_mock_get_min_ts_service(Ok(tsopb::GetMinTsResponse {
            header: Some(tsopb::ResponseHeader {
                cluster_id: 1,
                error: None,
                keyspace_id: 0,
                keyspace_group_id: 0,
            }),
            timestamp: Some(pdpb::Timestamp {
                physical: 30,
                logical: 1,
                suffix_bits: 0,
            }),
            keyspace_groups_serving: 1,
            keyspace_groups_total: 2,
        }))
        .await;
        let backend_b = start_mock_get_min_ts_service(Ok(tsopb::GetMinTsResponse {
            header: Some(tsopb::ResponseHeader {
                cluster_id: 1,
                error: None,
                keyspace_id: 0,
                keyspace_group_id: 0,
            }),
            timestamp: Some(pdpb::Timestamp {
                physical: 20,
                logical: 5,
                suffix_bits: 0,
            }),
            keyspace_groups_serving: 1,
            keyspace_groups_total: 2,
        }))
        .await;

        let member_urls = [backend_a.as_str(), backend_b.as_str()];
        let (endpoint, _service) = start_service_with_runtime_members(&member_urls).await;
        let response = legacy_pd_get_min_ts_once(&endpoint, 1)
            .await
            .expect("fan-out get_min_ts");
        assert!(response.header.as_ref().expect("header").error.is_none());
        let timestamp = response.timestamp.expect("timestamp");
        assert_eq!(timestamp.physical, 20);
        assert_eq!(timestamp.logical, 5);
    }

    #[tokio::test]
    async fn strict_profile_pdpb_get_min_ts_fanout_validation_failure_returns_handled_error() {
        let backend_a = start_mock_get_min_ts_service(Ok(tsopb::GetMinTsResponse {
            header: Some(tsopb::ResponseHeader {
                cluster_id: 1,
                error: None,
                keyspace_id: 0,
                keyspace_group_id: 0,
            }),
            timestamp: Some(pdpb::Timestamp {
                physical: 100,
                logical: 1,
                suffix_bits: 0,
            }),
            keyspace_groups_serving: 1,
            keyspace_groups_total: 2,
        }))
        .await;
        let backend_b = start_mock_get_min_ts_service(Ok(tsopb::GetMinTsResponse {
            header: Some(tsopb::ResponseHeader {
                cluster_id: 1,
                error: None,
                keyspace_id: 0,
                keyspace_group_id: 0,
            }),
            timestamp: Some(pdpb::Timestamp {
                physical: 90,
                logical: 1,
                suffix_bits: 0,
            }),
            keyspace_groups_serving: 1,
            keyspace_groups_total: 3,
        }))
        .await;

        let member_urls = [backend_a.as_str(), backend_b.as_str()];
        let (endpoint, _service) = start_service_with_runtime_members(&member_urls).await;
        let response = legacy_pd_get_min_ts_once(&endpoint, 1)
            .await
            .expect("fan-out get_min_ts");
        let header = response.header.expect("header");
        assert!(
            header.error.is_some(),
            "validation mismatch must be surfaced"
        );
        assert!(
            response.timestamp.is_some(),
            "handled path must keep timestamp object"
        );
    }

    #[tokio::test]
    async fn strict_profile_find_group_primary_flag_uses_address_equivalence() {
        let member_urls = ["node-a:1234", "http://node"];
        let (endpoint, _service) = start_service_with_runtime_members(&member_urls).await;

        let response = direct_find_group_by_keyspace_id_once(&endpoint, 1, 0, 0)
            .await
            .expect("find group");
        let header = response.header.as_ref().expect("header");
        assert!(
            header.error.is_none(),
            "find-group success must keep nil header.error"
        );
        let group = response.keyspace_group.as_ref().expect("group");
        assert_eq!(group.members.len(), 2);
        assert!(!group.members[0].is_primary);
        assert!(group.members[1].is_primary);
    }

    #[tokio::test]
    async fn strict_profile_pdpb_discovery_surface_reports_api_mode() {
        let (endpoint, _service) = start_service().await;

        let cluster = legacy_pd_get_cluster_info_once(&endpoint)
            .await
            .expect("get cluster info");
        assert!(cluster.header.is_some());
        assert!(cluster.header.as_ref().expect("header").error.is_none());
        assert_eq!(
            cluster.service_modes,
            vec![pdpb::ServiceMode::ApiSvcMode as i32]
        );

        let members = legacy_pd_get_members_once(&endpoint)
            .await
            .expect("get members");
        assert!(members.header.is_some());
        assert!(members.header.as_ref().expect("header").error.is_none());
        if members.members.is_empty() {
            assert!(
                members.leader.is_none(),
                "unresolved discovery should not synthesize a fallback leader"
            );
            assert!(
                members.etcd_leader.is_none(),
                "unresolved discovery should not synthesize a fallback etcd leader"
            );
        }
    }

    #[tokio::test]
    async fn strict_profile_pdpb_deprecated_stubs_return_compatibility_envelopes() {
        let (endpoint, _service) = start_service().await;

        let sync_max = legacy_pd_sync_max_ts_once(&endpoint, 1)
            .await
            .expect("sync max ts");
        assert!(sync_max.header.is_some());
        assert!(sync_max.header.as_ref().expect("header").error.is_none());
        assert!(
            sync_max.max_local_ts.is_some(),
            "compat stub should return default-zero timestamp object"
        );
        assert!(
            sync_max.synced_dcs.is_empty(),
            "compat stub should preserve default-empty synced_dcs"
        );

        let dc_info = legacy_pd_get_dc_location_info_once(&endpoint, 1)
            .await
            .expect("get dc location info");
        assert!(dc_info.header.is_some());
        assert!(dc_info.header.as_ref().expect("header").error.is_none());
        assert_eq!(dc_info.suffix, 0);
        assert!(
            dc_info.max_ts.is_some(),
            "compat stub should return default-zero max_ts timestamp object"
        );
    }

    // Release-gate differential harness entrypoint.
    #[tokio::test]
    async fn strict_profile_release_gate_direct_vs_legacy_differential() {
        let _env_guard = async_env_lock().lock().await;
        let [
            _bootstrap_mode_restore,
            _mode_restore,
            _bootstrap_keyspace_restore,
        ] = EnvVarRestore::capture_many([
            "PD_TSO_BOOTSTRAP_MODE",
            "PD_TSO_MODE",
            "PD_TSO_BOOTSTRAP_KEYSPACE_ID",
        ]);
        for mode in ["classic", "nextgen"] {
            set_env("PD_TSO_BOOTSTRAP_MODE", mode);
            remove_env("PD_TSO_MODE");
            remove_env("PD_TSO_BOOTSTRAP_KEYSPACE_ID");
            assert_direct_vs_legacy_differential_once().await;
        }
    }

    #[test]
    fn strict_profile_release_gate_cluster_id_probe_parses_pd_transport_format_only() {
        assert_eq!(
            expected_cluster_id_from_mismatch("cluster id mismatch: expected 42, got 0"),
            None
        );
        assert_eq!(
            expected_cluster_id_from_mismatch("mismatch cluster id, need 84 but got 0"),
            Some(84)
        );
        assert_eq!(
            expected_cluster_id_from_mismatch("not a cluster mismatch"),
            None
        );
    }

    #[tokio::test]
    async fn strict_profile_release_gate_cluster_id_probe_prefers_discovery_header() {
        let cluster_id = 42;
        let (endpoint, _service) = start_service_with_cluster_id(cluster_id).await;
        assert_eq!(cluster_id_from_tso_endpoint(&endpoint).await, cluster_id);
    }

    #[tokio::test]
    async fn strict_profile_draining_rejection_includes_retry_after_hint() {
        let (endpoint, service) = start_service().await;
        {
            let guard = service.write().await;
            guard.set_draining(true);
        }

        let err = direct_tso_once(&endpoint, 1, 1)
            .await
            .expect_err("draining should reject request");
        assert_eq!(err.code(), Code::Unavailable);
        let retry_after = err
            .metadata()
            .get("retry-after-ms")
            .and_then(|value| value.to_str().ok());
        assert_eq!(retry_after, Some("100"));
    }

    #[tokio::test]
    async fn strict_profile_legacy_stream_honors_pd_forwarded_host_metadata() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 2,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );
        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream")
            .into_inner();
        let response = stream
            .message()
            .await
            .expect("stream message")
            .expect("forwarded response");
        assert_eq!(response.count, 2);
        assert!(response.timestamp.is_some());
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_rejects_overload_with_resource_exhausted() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, proxy_service) = start_proxy_service_without_primary().await;
        configure_legacy_proxy_runtime(&proxy_service, Some(0), None, None).await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );
        let status = match client.tso(request).await {
            Ok(response) => response
                .into_inner()
                .message()
                .await
                .expect_err("stream should fail for overload"),
            Err(status) => status,
        };
        assert_eq!(status.code(), Code::ResourceExhausted);
        assert_eq!(status.message(), LEGACY_PROXY_MAX_CONCURRENT_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_recv_timeout_returns_deadline_exceeded() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, proxy_service) = start_proxy_service_without_primary().await;
        configure_legacy_proxy_runtime(&proxy_service, None, Some(Duration::from_millis(30)), None)
            .await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (_tx, rx) = tokio::sync::mpsc::channel(1);
        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("receive timeout should close forwarded stream");
        assert_eq!(err.code(), Code::DeadlineExceeded);
        assert_eq!(err.message(), LEGACY_PROXY_RECV_TIMEOUT_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_recv_eof_returns_unknown() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        drop(tx);
        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );

        let status = match client.tso(request).await {
            Ok(response) => response
                .into_inner()
                .message()
                .await
                .expect_err("recv EOF should fail forwarded stream"),
            Err(status) => status,
        };
        assert_eq!(status.code(), Code::Unknown);
        assert!(status.message().contains("EOF"));
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_merged_count_overflow_returns_unknown() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let mut last_errors = Vec::new();
        for _attempt in 0..16 {
            let barrier = Arc::new(Barrier::new(2));
            let first = forwarded_legacy_tso_once_with_barrier(
                &proxy_endpoint,
                &target_endpoint,
                u32::MAX,
                Arc::clone(&barrier),
            );
            let second = forwarded_legacy_tso_once_with_barrier(
                &proxy_endpoint,
                &target_endpoint,
                u32::MAX,
                barrier,
            );

            let (first, second) = tokio::join!(first, second);
            for result in [first, second] {
                if let Err(status) = result {
                    if status.code() == Code::Unknown
                        && status.message() == LEGACY_FORWARD_OVERFLOW_TOKEN
                    {
                        return;
                    }
                    last_errors.push(format!("{}:{}", status.code(), status.message()));
                }
            }
        }

        panic!(
            "overflow token not observed after retries; statuses: {}",
            last_errors.join(" | ")
        );
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_idle_close_returns_unknown() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, proxy_service) = start_proxy_service_without_primary().await;
        configure_legacy_proxy_runtime(&proxy_service, None, None, Some(Duration::from_millis(20)))
            .await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(4);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send first request");

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );
        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();

        let first = stream
            .message()
            .await
            .expect("first response status")
            .expect("first response payload");
        assert_eq!(first.count, 1);

        tokio::time::sleep(Duration::from_millis(120)).await;

        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send second request");
        drop(tx);

        let err = stream
            .message()
            .await
            .expect_err("idle close should be surfaced on next request");
        assert_eq!(err.code(), Code::Unknown);
        assert_eq!(err.message(), LEGACY_PROXY_IDLE_TIMEOUT_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_invalid_target_returns_not_found() {
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            "http://".parse().expect("invalid forwarded host metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("invalid target must fail with not found");
        assert_eq!(err.code(), Code::NotFound);
        assert_eq!(err.message(), LEGACY_PROXY_NOT_FOUND_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_dispatch_timeout_returns_deadline_exceeded() {
        let blackhole_endpoint = start_blackhole_endpoint().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            blackhole_endpoint.parse().expect("forwarded host metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("dispatch timeout should fail forwarded stream");
        assert_eq!(err.code(), Code::DeadlineExceeded);
        assert_eq!(err.message(), LEGACY_PROXY_FORWARD_TIMEOUT_TOKEN);
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_dispatch_timeout_fails_current_batch() {
        let blackhole_endpoint = start_blackhole_endpoint().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;
        let barrier = Arc::new(Barrier::new(2));

        let first = forwarded_legacy_tso_once_with_barrier(
            &proxy_endpoint,
            &blackhole_endpoint,
            1,
            Arc::clone(&barrier),
        );
        let second = forwarded_legacy_tso_once_with_barrier(
            &proxy_endpoint,
            &blackhole_endpoint,
            1,
            barrier,
        );

        let (first, second) = tokio::join!(first, second);
        for result in [first, second] {
            let err = result.expect_err("blackhole dispatch should fail current batch");
            assert_eq!(err.code(), Code::DeadlineExceeded);
            assert_eq!(err.message(), LEGACY_PROXY_FORWARD_TIMEOUT_TOKEN);
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn strict_profile_legacy_forwarded_stream_watchdog_timeout_returns_canceled() {
        let (target_endpoint, target_service) = start_service().await;
        let (lock_held_tx, lock_held_rx) = tokio::sync::oneshot::channel();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
        let release_guard = ReleaseOnDrop::new(release_tx);
        let lock_holder = tokio::task::spawn_blocking(move || {
            let target_guard = target_service.blocking_read();
            let _blocked_target_manager = target_guard.manager_mut();
            let _ = lock_held_tx.send(());
            let _ = release_rx.blocking_recv();
        });
        lock_held_rx.await.expect("manager lock held");
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("watchdog timeout should cancel forwarded request");
        assert_eq!(err.code(), Code::Cancelled);
        assert_eq!(err.message(), "context canceled");
        drop(release_guard);
        lock_holder.await.expect("lock holder finished");
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_send_failpoint_returns_canceled() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );
        request.metadata_mut().insert(
            LEGACY_FORWARD_FAILPOINT_METADATA_KEY,
            LEGACY_FORWARD_FAILPOINT_SEND_TIMEOUT_VALUE
                .parse()
                .expect("failpoint metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("send failpoint should surface watchdog cancellation");
        assert_eq!(err.code(), Code::Cancelled);
        assert_eq!(err.message(), "context canceled");
    }

    #[tokio::test]
    async fn strict_profile_legacy_forwarded_stream_recv_failpoint_returns_canceled() {
        let (target_endpoint, _target_service) = start_service().await;
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            target_endpoint.parse().expect("forwarded host metadata"),
        );
        request.metadata_mut().insert(
            LEGACY_FORWARD_FAILPOINT_METADATA_KEY,
            LEGACY_FORWARD_FAILPOINT_RECV_TIMEOUT_VALUE
                .parse()
                .expect("failpoint metadata"),
        );

        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("recv failpoint should surface watchdog cancellation");
        assert_eq!(err.code(), Code::Cancelled);
        assert_eq!(err.message(), "context canceled");
    }

    #[tokio::test]
    async fn strict_profile_forwarded_host_matching_is_exact_string() {
        let _env_guard = async_env_lock().lock().await;
        let _advertise_restore = EnvVarRestore::capture("PD_TSO_ADVERTISE_LISTEN_ADDR");
        set_env("PD_TSO_ADVERTISE_LISTEN_ADDR", "http://127.0.0.1:32111");
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            "127.0.0.1:32111".parse().expect("forwarded host metadata"),
        );
        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("exact-string mismatch should be treated as forwarded");
        assert_eq!(err.code(), Code::Unavailable);
    }

    #[tokio::test]
    async fn strict_profile_forwarded_host_runtime_members_do_not_imply_local_identity() {
        let _env_guard = async_env_lock().lock().await;
        let [_advertise_restore, _listen_restore] =
            EnvVarRestore::capture_many(["PD_TSO_ADVERTISE_LISTEN_ADDR", "PD_TSO_LISTEN_ADDR"]);
        remove_env("PD_TSO_ADVERTISE_LISTEN_ADDR");
        remove_env("PD_TSO_LISTEN_ADDR");

        let member_urls = vec!["http://198.51.100.10:3379", "http://198.51.100.11:3379"];
        let (endpoint, _service) = start_service_with_runtime_members(&member_urls).await;

        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            member_urls[0].parse().expect("forwarded host metadata"),
        );
        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("runtime membership alone must not classify forwarded host as local");
        assert_eq!(err.code(), Code::Unavailable);
    }

    #[tokio::test]
    async fn strict_profile_forwarded_host_legacy_discovery_urls_do_not_imply_local_identity() {
        let _env_guard = async_env_lock().lock().await;
        let [_legacy_restore, _advertise_restore, _listen_restore] = EnvVarRestore::capture_many([
            "PD_TSO_LEGACY_DISCOVERY_URLS",
            "PD_TSO_ADVERTISE_LISTEN_ADDR",
            "PD_TSO_LISTEN_ADDR",
        ]);
        remove_env("PD_TSO_ADVERTISE_LISTEN_ADDR");
        remove_env("PD_TSO_LISTEN_ADDR");
        set_env(
            "PD_TSO_LEGACY_DISCOVERY_URLS",
            "http://198.51.100.20:3379,http://198.51.100.21:3379",
        );
        let (proxy_endpoint, _proxy_service) = start_proxy_service_without_primary().await;

        let channel = Channel::from_shared(proxy_endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = pdpb::pd_client::PdClient::new(channel);

        let (tx, rx) = tokio::sync::mpsc::channel(1);
        tx.send(pdpb::TsoRequest {
            header: Some(pdpb::RequestHeader {
                cluster_id: 1,
                sender_id: 42,
                caller_id: String::new(),
                caller_component: String::new(),
            }),
            count: 1,
            dc_location: String::new(),
        })
        .await
        .expect("send request");
        drop(tx);

        let mut request = Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
        request.metadata_mut().insert(
            "pd-forwarded-host",
            "http://198.51.100.20:3379"
                .parse()
                .expect("forwarded host metadata"),
        );
        let mut stream = client
            .tso(request)
            .await
            .expect("forwarded stream open")
            .into_inner();
        let err = stream
            .message()
            .await
            .expect_err("legacy discovery URL list must not be used for local identity matching");
        assert_eq!(err.code(), Code::Unavailable);
    }

    #[tokio::test]
    async fn strict_profile_discovery_uses_runtime_member_topology() {
        let member_urls = vec!["http://10.10.0.1:3379", "10.10.0.2:3379"];
        let (endpoint, _service) = start_service_with_runtime_members(&member_urls).await;

        let cluster = legacy_pd_get_cluster_info_once(&endpoint)
            .await
            .expect("cluster info");
        assert_eq!(
            cluster.tso_urls,
            member_urls
                .iter()
                .map(|value| (*value).to_string())
                .collect::<Vec<_>>()
        );

        let members = legacy_pd_get_members_once(&endpoint)
            .await
            .expect("members");
        let returned = members
            .members
            .iter()
            .flat_map(|member| member.client_urls.clone())
            .collect::<Vec<_>>();
        assert!(returned.iter().any(|url| url == "http://10.10.0.1:3379"));
        assert!(returned.iter().any(|url| url == "10.10.0.2:3379"));
    }

    #[tokio::test]
    async fn strict_profile_parallel_group_requests_succeed() {
        let (endpoint, _service) = start_service().await;

        let (g0, g1) = tokio::join!(
            direct_tso_once_with_route(&endpoint, 1, 1, 0, 0),
            direct_tso_once_with_route(&endpoint, 1, 1, 1, 1)
        );

        let g0 = g0.expect("group0 response");
        let g1 = g1.expect("group1 response");
        assert_eq!(
            g0.header.as_ref().expect("group0 header").keyspace_group_id,
            0
        );
        assert_eq!(
            g1.header.as_ref().expect("group1 header").keyspace_group_id,
            1
        );
    }

    #[tokio::test]
    async fn strict_profile_release_gate_go_vs_rust_differential() {
        let _env_guard = async_env_lock().lock().await;
        let manifest = load_legacy_client_manifest().unwrap_or_else(|err| panic!("{err}"));
        let allow_missing = allow_missing_go_endpoint();
        let go_endpoints = resolve_release_gate_go_endpoints(&manifest, allow_missing);
        if go_endpoints.is_empty() {
            if allow_missing {
                return;
            }
            panic!(
                "no Go endpoints resolved from {}; set PD_ALLOW_MISSING_GO_TSO_ENDPOINT=1 to skip locally",
                LEGACY_CLIENT_MANIFEST_PATH
            );
        }
        for (version, go_endpoint) in go_endpoints {
            let cluster_id = cluster_id_from_tso_endpoint(&go_endpoint).await;
            let (rust_endpoint, _service_rust) = start_service_with_cluster_id(cluster_id).await;
            assert_go_vs_rust_differential_once(&go_endpoint, &rust_endpoint, cluster_id).await;
            let _ = version;
        }
    }

    fn now_ms() -> i64 {
        let duration = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default();
        (duration.as_millis().min(i64::MAX as u128)) as i64
    }
}