#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::future::Future;
    use std::sync::Arc;
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

    use etcd_client::Client;
    use pd_proto_grpc::admin::v1::{
        admin_client::AdminClient, admin_server::AdminServer, transfer_primary_result,
    };
    use pd_proto_tso::{
        FindGroupByKeyspaceIdRequest, GetMinTsRequest, KeyspaceGroup, RequestHeader, TsoRequest,
    };
    use pd_service_tso::etcd_store::EtcdTimestampStore;
    use pd_service_tso::keypath::{
        election_path, expected_primary_path, membership_path, timestamp_path,
    };
    use pd_service_tso::persistence::StoreError;
    use pd_service_tso::{
        AdminConfigState, AdminGrpcService, AllocatorConfig, InMemoryEtcdStore,
        KeyspaceGroupManager, TimestampStore, TsoRuntime, TsoRuntimeConfig, TsoService,
    };
    use pd_test_util::env::{EnvVarRestore, async_env_lock, remove_env};
    use tokio::sync::{Barrier, Mutex, RwLock};
    use tokio_stream::wrappers::TcpListenerStream;
    use tonic::Request;
    use tonic::transport::{Channel, Server};

    fn group(id: u32, keyspaces: Vec<u32>) -> KeyspaceGroup {
        KeyspaceGroup {
            id,
            user_kind: "basic".to_string(),
            split_state: None,
            merge_state: None,
            members: Vec::new(),
            keyspaces,
        }
    }

    async fn start_admin_service_with_active_config(
        active_config: HashMap<String, String>,
    ) -> (String, Arc<RwLock<TsoService>>) {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager
            .register_group(group(0, vec![0]))
            .expect("register default");
        manager
            .campaign_group_primary(0, now_ms())
            .expect("campaign default");

        let service = Arc::new(RwLock::new(TsoService::new(1, manager)));
        let active_epoch = if active_config.is_empty() { 0 } else { 1 };
        let admin_config = Arc::new(Mutex::new(AdminConfigState {
            active: active_config,
            history: Vec::new(),
            active_epoch,
        }));
        let admin = AdminGrpcService::new(Arc::clone(&service), admin_config);

        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(AdminServer::new(admin))
                .serve_with_incoming(stream)
                .await
                .expect("serve admin");
        });

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

    async fn start_admin_service() -> (String, Arc<RwLock<TsoService>>) {
        start_admin_service_with_active_config(HashMap::new()).await
    }

    async fn start_admin_service_with_extension_profile() -> (String, Arc<RwLock<TsoService>>) {
        start_admin_service_with_active_config(HashMap::from([(
            "compatibility-profile".to_string(),
            "extension".to_string(),
        )]))
        .await
    }

    fn real_etcd_endpoints_from_env() -> Vec<String> {
        let raw = std::env::var("PD_REAL_ETCD_ENDPOINTS")
            .unwrap_or_else(|_| "http://127.0.0.1:2379".to_string());
        let endpoints: Vec<String> = raw
            .split(',')
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
            .collect();
        assert!(
            !endpoints.is_empty(),
            "real etcd integration tests require at least one endpoint; set PD_REAL_ETCD_ENDPOINTS or ensure localhost:2379 is available"
        );
        endpoints
    }

    #[test]
    fn real_etcd_endpoint_defaults_to_localhost_when_env_unset() {
        let _guard = async_env_lock().blocking_lock();
        let _restore = EnvVarRestore::capture("PD_REAL_ETCD_ENDPOINTS");
        remove_env("PD_REAL_ETCD_ENDPOINTS");
        let endpoints = real_etcd_endpoints_from_env();
        assert_eq!(endpoints, vec!["http://127.0.0.1:2379".to_string()]);
    }

    async fn wait_until<F, Fut>(
        label: &str,
        timeout: Duration,
        poll_interval: Duration,
        mut predicate: F,
    ) where
        F: FnMut() -> Fut,
        Fut: Future<Output = bool>,
    {
        let deadline = Instant::now() + timeout;
        loop {
            if predicate().await {
                return;
            }
            if Instant::now() >= deadline {
                panic!("timeout while waiting for {label}");
            }
            tokio::time::sleep(poll_interval).await;
        }
    }

    #[test]
    fn integration_tso_find_and_get_min_ts_flow() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager.register_group(group(0, vec![0, 10])).expect("g0");
        manager.register_group(group(1, vec![11])).expect("g1");
        manager
            .campaign_group_primary(0, 10_000)
            .expect("campaign g0");
        manager
            .campaign_group_primary(1, 10_000)
            .expect("campaign g1");

        let service = TsoService::new(1, manager);

        // Warm both groups for first-request initialization.
        service
            .tso(
                10_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect("warm g0");

        let tso_resp = service
            .tso(
                10_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 11,
                        keyspace_group_id: 0,
                        ..RequestHeader::default()
                    },
                    count: 4,
                    dc_location: String::new(),
                },
            )
            .expect("route + tso");
        assert_eq!(tso_resp.header.keyspace_group_id, 1);
        assert_eq!(tso_resp.count, 4);

        let find_resp = service.find_group_by_keyspace_id(
            10_000,
            FindGroupByKeyspaceIdRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_group_id: 0,
                    ..RequestHeader::default()
                },
                keyspace_id: 11,
                mod_revision: 0,
            },
        );
        assert!(find_resp.header.error.is_none());
        assert_eq!(find_resp.header.keyspace_group_id, 1);

        let min_resp = service.get_min_ts(
            10_000,
            GetMinTsRequest {
                header: RequestHeader {
                    cluster_id: 1,
                    keyspace_group_id: 99,
                    ..RequestHeader::default()
                },
                dc_location: String::new(),
            },
        );
        assert!(min_resp.header.error.is_none());
        assert_eq!(min_resp.header.keyspace_group_id, 99);
        assert_eq!(min_resp.keyspace_groups_total, 2);
        assert_eq!(min_resp.keyspace_groups_serving, 2);
    }

    #[test]
    fn integration_split_merge_safety_flow() {
        let store = Arc::new(InMemoryEtcdStore::new());
        let mut manager = KeyspaceGroupManager::new(1, store, AllocatorConfig::default(), "node");
        manager.register_group(group(1, vec![11, 12])).expect("g1");
        manager
            .campaign_group_primary(1, 10_000)
            .expect("campaign g1");

        let service = TsoService::new(1, manager);

        // Warm source and move its timeline forward.
        service
            .tso(
                10_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 12,
                        keyspace_group_id: 1,
                        ..RequestHeader::default()
                    },
                    count: 5,
                    dc_location: String::new(),
                },
            )
            .expect("warm source");

        service
            .manager_mut()
            .split_keyspace_group(1, 2, vec![12])
            .expect("split");
        service
            .manager_mut()
            .campaign_group_primary(2, 10_000)
            .expect("campaign target");

        let split_resp = service
            .tso(
                10_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 12,
                        keyspace_group_id: 2,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect("split target tso");
        assert_eq!(split_resp.header.keyspace_group_id, 2);
        assert!(
            service
                .manager()
                .group(2)
                .expect("group2")
                .split_state
                .is_none()
        );

        service
            .manager_mut()
            .merge_keyspace_groups(1, vec![2])
            .expect("merge");
        service
            .manager_mut()
            .finish_merge_keyspace_group(1, 10_000)
            .expect("finish merge");

        let merged_resp = service
            .tso(
                10_000,
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 12,
                        keyspace_group_id: 2,
                        ..RequestHeader::default()
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
            .expect("merged routing");
        assert_eq!(merged_resp.header.keyspace_group_id, 1);

        service
            .manager_mut()
            .cleanup_deleted_group_timestamps()
            .expect("cleanup deleted groups");
        assert!(service.manager().deleted_groups().is_empty());
    }

    #[tokio::test]
    async fn integration_admin_validate_config_strict_guardrail() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .validate_config(Request::new(
                pd_proto_grpc::admin::v1::ValidateConfigRequest {
                    request_id: "req-1".to_string(),
                    proposed_config: std::collections::HashMap::from([
                        ("compatibility-profile".to_string(), "strict".to_string()),
                        (
                            "enable-async-tso-window-prefetch".to_string(),
                            "true".to_string(),
                        ),
                    ]),
                },
            ))
            .await
            .expect("validate")
            .into_inner();

        assert!(!response.accepted);
        assert!(
            response
                .violations
                .iter()
                .any(|v| v.code == "strict_profile_guardrail_violation")
        );
    }

    #[tokio::test]
    async fn integration_admin_get_group_state_exposes_prefetch_and_capacity_diagnostics() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .get_group_state(Request::new(
                pd_proto_grpc::admin::v1::GetGroupStateRequest { group_id: 0 },
            ))
            .await
            .expect("get group state")
            .into_inner();
        let state = response.state.expect("state");

        assert!(state.is_primary);
        assert_eq!(state.actor_state, "Serving");
        assert!(state.in_memory_timestamp.is_some());
        assert!(state.last_persisted_timestamp.is_some());
        assert!(state.prefetch_status.is_some());
        assert!(state.lease_expiration_time.is_some());
        assert!(state.mailbox_capacity > 0);
        assert_eq!(state.mailbox_size, state.mailbox_depth);
    }

    #[tokio::test]
    async fn integration_admin_drain_flips_service_state() {
        let (endpoint, service) = start_admin_service().await;
        {
            let guard = service.write().await;
            guard
                .manager_mut()
                .upsert_group(KeyspaceGroup {
                    id: 0,
                    user_kind: "basic".to_string(),
                    split_state: None,
                    merge_state: None,
                    members: vec![
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node:0".to_string(),
                            is_primary: true,
                        },
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node-peer:0".to_string(),
                            is_primary: false,
                        },
                    ],
                    keyspaces: vec![0],
                })
                .expect("set transfer candidates for drain");
        }
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .drain(Request::new(pd_proto_grpc::admin::v1::DrainRequest {
                request_id: "drain-1".to_string(),
                timeout_ms: 10_000,
            }))
            .await
            .expect("drain")
            .into_inner();
        assert_eq!(
            response.state,
            pd_proto_grpc::admin::v1::drain_response::State::Drained as i32
        );

        let guard = service.read().await;
        assert!(guard.is_draining());
    }

    #[tokio::test]
    async fn integration_admin_drain_concurrent_replay_returns_identical_result() {
        let (endpoint, service) = start_admin_service().await;
        {
            let guard = service.write().await;
            guard
                .manager_mut()
                .upsert_group(KeyspaceGroup {
                    id: 0,
                    user_kind: "basic".to_string(),
                    split_state: None,
                    merge_state: None,
                    members: vec![
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node:0".to_string(),
                            is_primary: true,
                        },
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node-peer:0".to_string(),
                            is_primary: false,
                        },
                    ],
                    keyspaces: vec![0],
                })
                .expect("set transfer candidates for concurrent drain replay");
        }

        let request_id = format!("drain-concurrent-{}", now_ms());
        let barrier = Arc::new(Barrier::new(3));
        let spawn_drain = |barrier: Arc<Barrier>| {
            let endpoint = endpoint.clone();
            let request_id = request_id.clone();
            tokio::spawn(async move {
                let channel = Channel::from_shared(endpoint)
                    .expect("channel")
                    .connect()
                    .await
                    .expect("connect");
                let mut client = AdminClient::new(channel);
                barrier.wait().await;
                client
                    .drain(Request::new(pd_proto_grpc::admin::v1::DrainRequest {
                        request_id,
                        timeout_ms: 10_000,
                    }))
                    .await
                    .expect("drain")
                    .into_inner()
            })
        };

        let first = spawn_drain(Arc::clone(&barrier));
        let second = spawn_drain(Arc::clone(&barrier));
        barrier.wait().await;

        let first_response = first.await.expect("join first");
        let second_response = second.await.expect("join second");
        assert_eq!(
            first_response, second_response,
            "concurrent replay must observe one idempotent drain result"
        );
        assert_eq!(
            first_response.state,
            pd_proto_grpc::admin::v1::drain_response::State::Drained as i32
        );
    }

    #[tokio::test]
    async fn integration_admin_drain_without_transfer_candidate_times_out() {
        let (endpoint, service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .drain(Request::new(pd_proto_grpc::admin::v1::DrainRequest {
                request_id: "drain-timeout-no-target".to_string(),
                timeout_ms: 100,
            }))
            .await
            .expect("drain")
            .into_inner();
        assert_eq!(
            response.state,
            pd_proto_grpc::admin::v1::drain_response::State::Timeout as i32
        );
        assert!(
            response.remaining_group_ids.contains(&0),
            "drain timeout should report group without transfer target"
        );

        let guard = service.read().await;
        assert!(
            !guard.is_draining(),
            "timeout should clear draining ingress gate"
        );
    }

    #[tokio::test]
    async fn integration_admin_apply_config_updates_epoch_and_history() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);
        let request_id = format!("apply-1-{}", now_ms());

        let apply = client
            .apply_config(Request::new(pd_proto_grpc::admin::v1::ApplyConfigRequest {
                request_id,
                proposed_config: HashMap::from([(
                    "tso-save-interval-ms".to_string(),
                    "3000".to_string(),
                )]),
                reason: "integration-test".to_string(),
            }))
            .await
            .expect("apply config")
            .into_inner();
        assert!(apply.accepted);
        assert_eq!(apply.active_epoch, 1);
        assert_eq!(
            apply
                .active_config
                .get("tso-save-interval-ms")
                .expect("active key"),
            "3000"
        );

        let history = client
            .config_history_diff(Request::new(
                pd_proto_grpc::admin::v1::ConfigHistoryDiffRequest {
                    diff_with_active: HashMap::new(),
                    limit: 10,
                },
            ))
            .await
            .expect("history diff")
            .into_inner();
        assert_eq!(history.history.len(), 1);
        assert_eq!(history.history[0].epoch, 1);
        assert_eq!(
            history.history[0]
                .config
                .get("tso-save-interval-ms")
                .expect("history key"),
            "3000"
        );

        let guard = _service.read().await;
        let manager = guard.manager();
        assert_eq!(manager.allocator_config().oracle.save_interval_ms, 3_000);
    }

    #[tokio::test]
    async fn integration_admin_apply_config_replay_is_idempotent() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);
        let request_id = format!("apply-idempotent-1-{}", now_ms());

        let mut first = Request::new(pd_proto_grpc::admin::v1::ApplyConfigRequest {
            request_id: request_id.clone(),
            proposed_config: HashMap::from([(
                "tso-save-interval-ms".to_string(),
                "3000".to_string(),
            )]),
            reason: "integration-test".to_string(),
        });
        first
            .metadata_mut()
            .insert("x-caller-id", "caller-a".parse().expect("caller metadata"));
        let first_response = client
            .apply_config(first)
            .await
            .expect("first apply")
            .into_inner();
        assert!(first_response.accepted);
        assert_eq!(first_response.active_epoch, 1);

        let mut replay = Request::new(pd_proto_grpc::admin::v1::ApplyConfigRequest {
            request_id,
            proposed_config: HashMap::from([(
                "tso-save-interval-ms".to_string(),
                "9999".to_string(),
            )]),
            reason: "integration-test-replay".to_string(),
        });
        replay
            .metadata_mut()
            .insert("x-caller-id", "caller-a".parse().expect("caller metadata"));
        let replay_response = client
            .apply_config(replay)
            .await
            .expect("replay apply")
            .into_inner();
        assert!(replay_response.accepted);
        assert_eq!(replay_response.active_epoch, first_response.active_epoch);
        assert_eq!(replay_response.active_config, first_response.active_config);

        let history = client
            .config_history_diff(Request::new(
                pd_proto_grpc::admin::v1::ConfigHistoryDiffRequest {
                    diff_with_active: HashMap::new(),
                    limit: 10,
                },
            ))
            .await
            .expect("history diff")
            .into_inner();
        assert_eq!(
            history.history.len(),
            1,
            "replayed request must not append new config history entry"
        );
    }

    #[tokio::test]
    async fn integration_admin_apply_config_updates_runtime_request_limits() {
        let (endpoint, service) = start_admin_service_with_extension_profile().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);
        let request_id = format!("apply-runtime-limit-1-{}", now_ms());

        let apply = client
            .apply_config(Request::new(pd_proto_grpc::admin::v1::ApplyConfigRequest {
                request_id,
                proposed_config: HashMap::from([(
                    "max-tso-count-per-request".to_string(),
                    "2".to_string(),
                )]),
                reason: "integration-test".to_string(),
            }))
            .await
            .expect("apply config")
            .into_inner();
        assert!(apply.accepted);
        assert_eq!(apply.apply_mode, "hot-reload");

        let guard = service.read().await;
        let err = guard
            .tso(
                now_ms(),
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        sender_id: 0,
                    },
                    count: 3,
                    dc_location: String::new(),
                },
            )
            .expect_err("count over runtime limit should fail");
        assert_eq!(err.code, pd_service_tso::GrpcCode::InvalidArgument);
    }

    #[tokio::test]
    async fn integration_admin_apply_config_rejects_restart_required_changes() {
        let (endpoint, service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let apply = client
            .apply_config(Request::new(pd_proto_grpc::admin::v1::ApplyConfigRequest {
                request_id: "apply-restart-required-1".to_string(),
                proposed_config: HashMap::from([(
                    "compatibility-profile".to_string(),
                    "extension".to_string(),
                )]),
                reason: "integration-test".to_string(),
            }))
            .await
            .expect("apply config")
            .into_inner();
        assert!(!apply.accepted);
        assert_eq!(apply.apply_mode, "restart-required");
        assert!(
            apply
                .violations
                .iter()
                .any(|item| item.code == "restart_required_config_change")
        );
        assert_eq!(apply.active_epoch, 0);

        let guard = service.read().await;
        let response = guard
            .tso(
                now_ms(),
                TsoRequest {
                    header: RequestHeader {
                        cluster_id: 1,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        sender_id: 0,
                    },
                    count: 3,
                    dc_location: String::new(),
                },
            )
            .expect("restart-required rejection must not mutate runtime behavior");
        assert_eq!(response.count, 3);
    }

    #[tokio::test]
    async fn integration_admin_transfer_primary_empty_target_without_candidate_fails() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .transfer_primary(Request::new(
                pd_proto_grpc::admin::v1::TransferPrimaryRequest {
                    request_id: "transfer-empty-target".to_string(),
                    group_ids: vec![0],
                    target_member_id: String::new(),
                },
            ))
            .await
            .expect("transfer")
            .into_inner();
        assert_eq!(response.results.len(), 1);
        assert_eq!(
            response.results[0].status,
            pd_proto_grpc::admin::v1::transfer_primary_result::Status::Failed as i32
        );
    }

    #[tokio::test]
    async fn integration_admin_transfer_primary_concurrent_replay_returns_identical_result() {
        let (endpoint, service) = start_admin_service().await;
        {
            let guard = service.write().await;
            guard
                .manager_mut()
                .upsert_group(KeyspaceGroup {
                    id: 0,
                    user_kind: "basic".to_string(),
                    split_state: None,
                    merge_state: None,
                    members: vec![
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node:0".to_string(),
                            is_primary: true,
                        },
                        pd_proto_tso::KeyspaceGroupMember {
                            address: "node-peer:0".to_string(),
                            is_primary: false,
                        },
                    ],
                    keyspaces: vec![0],
                })
                .expect("set transfer candidates for concurrent replay");
        }

        let request_id = format!("transfer-concurrent-{}", now_ms());
        let barrier = Arc::new(Barrier::new(3));
        let spawn_transfer = |barrier: Arc<Barrier>| {
            let endpoint = endpoint.clone();
            let request_id = request_id.clone();
            tokio::spawn(async move {
                let channel = Channel::from_shared(endpoint)
                    .expect("channel")
                    .connect()
                    .await
                    .expect("connect");
                let mut client = AdminClient::new(channel);
                barrier.wait().await;
                client
                    .transfer_primary(Request::new(
                        pd_proto_grpc::admin::v1::TransferPrimaryRequest {
                            request_id,
                            group_ids: vec![0],
                            target_member_id: "node-peer:0".to_string(),
                        },
                    ))
                    .await
                    .expect("transfer")
                    .into_inner()
            })
        };

        let first = spawn_transfer(Arc::clone(&barrier));
        let second = spawn_transfer(Arc::clone(&barrier));
        barrier.wait().await;

        let first_response = first.await.expect("join first");
        let second_response = second.await.expect("join second");
        assert_eq!(
            first_response, second_response,
            "concurrent replay must observe one idempotent transfer result"
        );
        assert_eq!(first_response.results.len(), 1);
        assert_eq!(
            first_response.results[0].status,
            transfer_primary_result::Status::Success as i32
        );
    }

    #[tokio::test]
    async fn integration_admin_set_log_level_applies_runtime_change() {
        let (endpoint, _service) = start_admin_service().await;
        let channel = Channel::from_shared(endpoint)
            .expect("channel")
            .connect()
            .await
            .expect("connect");
        let mut client = AdminClient::new(channel);

        let response = client
            .set_log_level(Request::new(pd_proto_grpc::admin::v1::SetLogLevelRequest {
                request_id: "set-log-level-1".to_string(),
                target: "all".to_string(),
                level: "debug".to_string(),
            }))
            .await
            .expect("set log level")
            .into_inner();
        assert!(response.success);
        assert!(response.message.contains("debug"));
    }

    #[tokio::test]
    async fn integration_admin_set_log_level_concurrent_replay_returns_identical_result() {
        let (endpoint, _service) = start_admin_service().await;
        let request_id = format!("set-log-level-concurrent-{}", now_ms());
        let barrier = Arc::new(Barrier::new(3));
        let spawn_set_log = |barrier: Arc<Barrier>| {
            let endpoint = endpoint.clone();
            let request_id = request_id.clone();
            tokio::spawn(async move {
                let channel = Channel::from_shared(endpoint)
                    .expect("channel")
                    .connect()
                    .await
                    .expect("connect");
                let mut client = AdminClient::new(channel);
                barrier.wait().await;
                client
                    .set_log_level(Request::new(pd_proto_grpc::admin::v1::SetLogLevelRequest {
                        request_id,
                        target: "all".to_string(),
                        level: "debug".to_string(),
                    }))
                    .await
                    .expect("set log level")
                    .into_inner()
            })
        };

        let first = spawn_set_log(Arc::clone(&barrier));
        let second = spawn_set_log(Arc::clone(&barrier));
        barrier.wait().await;

        let first_response = first.await.expect("join first");
        let second_response = second.await.expect("join second");
        assert_eq!(
            first_response, second_response,
            "concurrent replay must observe one idempotent set_log_level result"
        );
        assert!(first_response.success);
    }

    #[tokio::test]
    async fn integration_real_etcd_timestamp_store_fencing_round_trip() {
        let endpoints = real_etcd_endpoints_from_env();

        let cluster_id = now_ms() as u64;
        let group_id = 7_u32;
        let leader_key = election_path(cluster_id, group_id);
        let timestamp_key = timestamp_path(cluster_id, group_id);
        let leader_value = format!("node-{cluster_id}-{group_id}");
        let stale_leader_value = format!("node-stale-{cluster_id}-{group_id}");

        let mut etcd = Client::connect(endpoints.clone(), None)
            .await
            .expect("connect etcd");
        etcd.delete(leader_key.clone(), None)
            .await
            .expect("delete leader key");
        etcd.delete(timestamp_key, None)
            .await
            .expect("delete timestamp key");
        etcd.put(leader_key.clone(), leader_value.clone(), None)
            .await
            .expect("seed leader key");

        let store = EtcdTimestampStore::connect(endpoints).expect("connect timestamp store");
        store
            .save_timestamp_if_leader(cluster_id, group_id, 100, &leader_key, &leader_value)
            .expect("save timestamp");
        assert_eq!(
            store
                .load_timestamp(cluster_id, group_id)
                .expect("load timestamp"),
            Some(100)
        );

        let non_monotonic = store
            .save_timestamp_if_leader(cluster_id, group_id, 100, &leader_key, &leader_value)
            .expect_err("non-monotonic save must fail");
        assert!(matches!(non_monotonic, StoreError::NonMonotonic { .. }));

        etcd.put(leader_key.clone(), stale_leader_value, None)
            .await
            .expect("replace leader value");
        let stale = store
            .save_timestamp_if_leader(cluster_id, group_id, 200, &leader_key, &leader_value)
            .expect_err("stale leader must fail CAS");
        assert!(matches!(stale, StoreError::NotLeader));

        etcd.put(leader_key.clone(), leader_value.clone(), None)
            .await
            .expect("restore leader value");
        store
            .save_timestamp_if_leader(cluster_id, group_id, 200, &leader_key, &leader_value)
            .expect("save after leader restore");
        assert_eq!(
            store
                .load_timestamp(cluster_id, group_id)
                .expect("load after update"),
            Some(200)
        );

        store
            .delete_timestamp(cluster_id, group_id)
            .expect("delete timestamp");
        assert!(
            store
                .load_timestamp(cluster_id, group_id)
                .expect("load deleted timestamp")
                .is_none()
        );

        etcd.delete(leader_key, None)
            .await
            .expect("cleanup leader key");
    }

    #[tokio::test]
    async fn integration_real_etcd_runtime_campaign_lease_and_expected_primary_handoff() {
        let endpoints = real_etcd_endpoints_from_env();

        let cluster_id = now_ms() as u64 + 10_000;
        let group_id = 0_u32;
        let lease_ttl_secs = 5_i64;
        let planned_transfer_p99_budget = Duration::from_secs(5);
        let unplanned_failover_p99_budget =
            Duration::from_secs((lease_ttl_secs.saturating_mul(2)).max(5) as u64);
        let leader_key = election_path(cluster_id, group_id);
        let expected_primary_key = expected_primary_path(cluster_id, group_id);
        let timestamp_key = timestamp_path(cluster_id, group_id);
        let membership_key = membership_path(cluster_id, group_id);
        let local_member_value = "node:0".to_string();

        let mut etcd = Client::connect(endpoints.clone(), None)
            .await
            .expect("connect etcd");
        for key in [
            &leader_key,
            &expected_primary_key,
            &timestamp_key,
            &membership_key,
        ] {
            etcd.delete((*key).to_string(), None)
                .await
                .expect("cleanup etcd key");
        }
        etcd.put(
            membership_key.clone(),
            r#"{"user-kind":"basic","members":[{"address":"node:0"}],"keyspaces":[0]}"#,
            None,
        )
        .await
        .expect("seed membership group0");

        let store: Arc<dyn TimestampStore> = Arc::new(
            EtcdTimestampStore::connect(endpoints.clone()).expect("connect timestamp store"),
        );
        let mut manager =
            KeyspaceGroupManager::new(cluster_id, store, AllocatorConfig::default(), "node");
        manager
            .register_group(group(0, vec![0]))
            .expect("register default group");
        let service = Arc::new(RwLock::new(TsoService::new(cluster_id, manager)));
        let _runtime = TsoRuntime::start(
            Arc::clone(&service),
            TsoRuntimeConfig {
                cluster_id,
                etcd_endpoints: endpoints.clone(),
                campaign_interval: Duration::from_millis(50),
                keepalive_interval: Duration::from_millis(50),
                watch_retry_backoff: Duration::from_millis(50),
                lease_ttl_secs,
                ..TsoRuntimeConfig::default()
            },
        );

        let initial_campaign_started = Instant::now();
        wait_until(
            "group0 primary campaign",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    manager
                        .group_debug_state(group_id, now_ms())
                        .map(|state| state.is_primary && state.lease_id > 0)
                        .unwrap_or(false)
                }
            },
        )
        .await;
        let initial_campaign_elapsed = initial_campaign_started.elapsed();
        assert!(
            initial_campaign_elapsed <= unplanned_failover_p99_budget,
            "initial campaign exceeded unplanned-failover p99 budget {:?}: {:?}",
            unplanned_failover_p99_budget,
            initial_campaign_elapsed
        );

        let tso_ok = {
            let guard = service.read().await;
            guard.tso(
                now_ms(),
                TsoRequest {
                    header: RequestHeader {
                        cluster_id,
                        keyspace_id: 0,
                        keyspace_group_id: 0,
                        sender_id: 0,
                    },
                    count: 1,
                    dc_location: String::new(),
                },
            )
        };
        assert!(tso_ok.is_ok(), "runtime-campaigned group must serve tso");

        let expected_primary_stepdown_started = Instant::now();
        etcd.put(expected_primary_key.clone(), "remote:0", None)
            .await
            .expect("set remote expected primary");
        wait_until(
            "group0 expected-primary mismatch step-down",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    manager
                        .group_runtime_snapshot(group_id, now_ms())
                        .map(|snapshot| !snapshot.is_primary)
                        .unwrap_or(false)
                }
            },
        )
        .await;
        let expected_primary_stepdown_elapsed = expected_primary_stepdown_started.elapsed();
        assert!(
            expected_primary_stepdown_elapsed <= planned_transfer_p99_budget,
            "expected-primary stepdown exceeded planned-transfer p99 budget {:?}: {:?}",
            planned_transfer_p99_budget,
            expected_primary_stepdown_elapsed
        );

        let expected_primary_recampaign_started = Instant::now();
        etcd.put(expected_primary_key.clone(), local_member_value, None)
            .await
            .expect("restore expected primary to local member");
        wait_until(
            "group0 expected-primary local recampaign",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    manager
                        .group_runtime_snapshot(group_id, now_ms())
                        .map(|snapshot| snapshot.is_primary)
                        .unwrap_or(false)
                }
            },
        )
        .await;
        let expected_primary_recampaign_elapsed = expected_primary_recampaign_started.elapsed();
        assert!(
            expected_primary_recampaign_elapsed <= planned_transfer_p99_budget,
            "expected-primary recampaign exceeded planned-transfer p99 budget {:?}: {:?}",
            planned_transfer_p99_budget,
            expected_primary_recampaign_elapsed
        );

        let leader_value = etcd
            .get(leader_key.clone(), None)
            .await
            .expect("read leader key");
        assert_eq!(leader_value.kvs().len(), 1);
        assert_eq!(leader_value.kvs()[0].value(), b"node:0");

        for key in [
            &leader_key,
            &expected_primary_key,
            &timestamp_key,
            &membership_key,
        ] {
            etcd.delete((*key).to_string(), None)
                .await
                .expect("cleanup etcd key");
        }
    }

    #[tokio::test]
    async fn integration_real_etcd_runtime_membership_merge_finalize_flow() {
        let endpoints = real_etcd_endpoints_from_env();

        let cluster_id = now_ms() as u64 + 20_000;
        let group0_id = 0_u32;
        let group1_id = 1_u32;
        let membership0_key = membership_path(cluster_id, group0_id);
        let membership1_key = membership_path(cluster_id, group1_id);
        let leader0_key = election_path(cluster_id, group0_id);
        let expected_primary0_key = expected_primary_path(cluster_id, group0_id);

        let mut etcd = Client::connect(endpoints.clone(), None)
            .await
            .expect("connect etcd");
        for key in [
            &membership0_key,
            &membership1_key,
            &leader0_key,
            &expected_primary0_key,
        ] {
            etcd.delete((*key).to_string(), None)
                .await
                .expect("cleanup etcd key");
        }
        etcd.put(
            membership0_key.clone(),
            r#"{"user-kind":"basic","members":[{"address":"node:0"}],"keyspaces":[0]}"#,
            None,
        )
        .await
        .expect("seed membership group0");

        let store: Arc<dyn TimestampStore> = Arc::new(
            EtcdTimestampStore::connect(endpoints.clone()).expect("connect timestamp store"),
        );
        let mut manager =
            KeyspaceGroupManager::new(cluster_id, store, AllocatorConfig::default(), "node");
        manager
            .register_group(group(0, vec![0]))
            .expect("register default group");
        let service = Arc::new(RwLock::new(TsoService::new(cluster_id, manager)));
        let _runtime = TsoRuntime::start(
            Arc::clone(&service),
            TsoRuntimeConfig {
                cluster_id,
                etcd_endpoints: endpoints.clone(),
                campaign_interval: Duration::from_millis(50),
                keepalive_interval: Duration::from_millis(50),
                watch_retry_backoff: Duration::from_millis(50),
                lease_ttl_secs: 5,
                ..TsoRuntimeConfig::default()
            },
        );

        wait_until(
            "group0 primary before merge-flow watch updates",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    manager
                        .group_runtime_snapshot(group0_id, now_ms())
                        .map(|snapshot| snapshot.is_primary)
                        .unwrap_or(false)
                }
            },
        )
        .await;

        etcd.put(
            membership1_key.clone(),
            r#"{"user-kind":"basic","members":[{"address":"node:1"}],"keyspaces":[11]}"#,
            None,
        )
        .await
        .expect("insert membership group1");
        wait_until(
            "group1 membership visible",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    guard.manager().group(group1_id).is_some()
                }
            },
        )
        .await;

        etcd.put(
            membership0_key.clone(),
            r#"{"user-kind":"basic","merge-state":{"merge-list":[1]},"members":[{"address":"node:0"}],"keyspaces":[0,11]}"#,
            None,
        )
        .await
        .expect("set merge state on group0");
        etcd.delete(membership1_key.clone(), None)
            .await
            .expect("delete merge source group1 membership");

        wait_until(
            "group0 merge finalization after source removal",
            Duration::from_secs(10),
            Duration::from_millis(50),
            || {
                let service = Arc::clone(&service);
                async move {
                    let guard = service.read().await;
                    let manager = guard.manager();
                    let group0 = manager.group(group0_id);
                    let group1 = manager.group(group1_id);
                    let deleted = manager.deleted_groups();
                    group1.is_none()
                        && deleted.is_empty()
                        && group0.as_ref().is_some_and(|group| {
                            group.merge_state.is_none() && group.keyspaces.contains(&11)
                        })
                }
            },
        )
        .await;

        for key in [
            &membership0_key,
            &membership1_key,
            &leader0_key,
            &expected_primary0_key,
        ] {
            etcd.delete((*key).to_string(), None)
                .await
                .expect("cleanup etcd key");
        }
    }

    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
    }
}