// file: crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs
// version: 3

#[derive(Clone, Copy)]
enum FixtureMode {
    RoundTrip,
    ClientHalfClose,
    HostileClose,
    Flood,
    RemoteStatus,
    Malformed,
    Oversized,
    Idle,
    ReconnectReplay,
    ReplayGap,
    ReconnectExhausted,
}

#[derive(Clone)]
struct FixtureGeyser {
    mode: FixtureMode,
    half_close_seen: std::sync::Arc<std::sync::atomic::AtomicBool>,
    subscribe_calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}

#[allow(clippy::implicit_return)] // tonic::async_trait generates async wrapper tails outside the authored fixture bodies.
#[tonic::async_trait]
impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
    type SubscribeStream = super::MpscStream<std::result::Result<yellowstone_grpc_proto::geyser::SubscribeUpdate, tonic::Status>>;

    async fn subscribe(
        &self,
        request: tonic::Request<tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeRequest>>,
    ) -> std::result::Result<tonic::Response<Self::SubscribeStream>, tonic::Status> {
        if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) {
            return std::result::Result::Err(error);
        }
        let subscribe_call = self.subscribe_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst).saturating_add(1);
        if matches!(self.mode, FixtureMode::ReconnectExhausted) && subscribe_call > 1 {
            return std::result::Result::Err(tonic::Status::unavailable("GRPC-RECONNECT-SECRET-CANARY"));
        }
        let mut inbound = request.into_inner();
        let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(64);
        let mode = self.mode;
        let half_close_seen = self.half_close_seen.clone();
        tokio::spawn(async move {
            let initial = inbound.message().await;
            let initial = match initial {
                std::result::Result::Ok(std::option::Option::Some(value)) => value,
                _ => return,
            };
            match mode {
                FixtureMode::RoundTrip => {
                    assert!(initial.slots.contains_key("initial"));
                    if outbound_tx.send(std::result::Result::Ok(ping_update())).await.is_err() {
                        return;
                    }
                    let ping = inbound.message().await;
                    let ping = match ping {
                        std::result::Result::Ok(std::option::Option::Some(value)) => value,
                        _ => return,
                    };
                    assert_eq!(ping.ping.map(|value| value.id), std::option::Option::Some(1));
                    assert!(ping.accounts.is_empty());
                    assert!(ping.slots.is_empty());
                    assert!(ping.transactions.is_empty());
                    assert!(ping.transactions_status.is_empty());
                    assert!(ping.blocks.is_empty());
                    assert!(ping.blocks_meta.is_empty());
                    assert!(ping.entry.is_empty());
                    if outbound_tx.send(std::result::Result::Ok(pong_update(1))).await.is_err() {
                        return;
                    }
                    let mutation = inbound.message().await;
                    let mutation = match mutation {
                        std::result::Result::Ok(std::option::Option::Some(value)) => value,
                        _ => return,
                    };
                    assert!(mutation.accounts.contains_key("mutated"));
                    let _ = outbound_tx.send(std::result::Result::Ok(slot_update(901))).await;
                },
                FixtureMode::ClientHalfClose => {
                    let half_close = inbound.message().await;
                    if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) {
                        half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst);
                    }
                },
                FixtureMode::HostileClose => {
                    let half_close = inbound.message().await;
                    if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) {
                        half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst);
                    }
                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                    drop(outbound_tx);
                },
                FixtureMode::Flood => {
                    for slot in 0_u64..32_u64 {
                        if outbound_tx.send(std::result::Result::Ok(slot_update(1_000 + slot))).await.is_err() {
                            return;
                        }
                    }
                },
                FixtureMode::RemoteStatus => {
                    let _ = outbound_tx.send(std::result::Result::Err(tonic::Status::permission_denied("GRPC-STREAM-REMOTE-SECRET-CANARY"))).await;
                },
                FixtureMode::Malformed => {
                    let _ = outbound_tx
                        .send(std::result::Result::Ok(yellowstone_grpc_proto::geyser::SubscribeUpdate {
                            filters: std::vec::Vec::new(),
                            update_oneof: std::option::Option::None,
                            created_at: std::option::Option::None,
                        }))
                        .await;
                },
                FixtureMode::Oversized => {
                    let _ = outbound_tx.send(std::result::Result::Ok(slot_update(77))).await;
                },
                FixtureMode::Idle => {
                    let _ = inbound.message().await;
                },
                FixtureMode::ReconnectReplay => {
                    if subscribe_call == 1 {
                        assert_eq!(initial.from_slot, std::option::Option::None);
                        let _ = outbound_tx.send(std::result::Result::Ok(slot_update(500))).await;
                    } else {
                        assert_eq!(subscribe_call, 2);
                        assert_eq!(initial.from_slot, std::option::Option::Some(500));
                        if outbound_tx.send(std::result::Result::Ok(slot_update(500))).await.is_err() {
                            return;
                        }
                        if outbound_tx.send(std::result::Result::Ok(slot_update(501))).await.is_err() {
                            return;
                        }
                        let half_close = inbound.message().await;
                        if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) {
                            half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst);
                        }
                    }
                },
                FixtureMode::ReplayGap => {
                    if subscribe_call == 1 {
                        assert_eq!(initial.from_slot, std::option::Option::None);
                        let _ = outbound_tx.send(std::result::Result::Ok(slot_update(500))).await;
                    } else {
                        assert_eq!(subscribe_call, 2);
                        assert_eq!(initial.from_slot, std::option::Option::Some(505));
                        if outbound_tx.send(std::result::Result::Ok(slot_update(505))).await.is_err() {
                            return;
                        }
                        let half_close = inbound.message().await;
                        if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) {
                            half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst);
                        }
                    }
                },
                FixtureMode::ReconnectExhausted => {
                    assert_eq!(subscribe_call, 1);
                    let _ = outbound_tx.send(std::result::Result::Ok(slot_update(700))).await;
                },
            }
        });
        return std::result::Result::Ok(tonic::Response::new(super::MpscStream::new(outbound_rx)));
    }

    type SubscribeDeshredStream = futures_util::stream::Empty<std::result::Result<yellowstone_grpc_proto::geyser::SubscribeUpdateDeshred, tonic::Status>>;

    async fn subscribe_deshred(
        &self,
        _request: tonic::Request<tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeDeshredRequest>>,
    ) -> std::result::Result<tonic::Response<Self::SubscribeDeshredStream>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("deshred is outside KSP 0.2.9"));
    }

    async fn subscribe_replay_info(
        &self,
        request: tonic::Request<yellowstone_grpc_proto::geyser::SubscribeReplayInfoRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse>, tonic::Status> {
        if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) {
            return std::result::Result::Err(error);
        }
        let first_available = match self.mode {
            FixtureMode::ReconnectReplay | FixtureMode::ReconnectExhausted => std::option::Option::Some(400),
            FixtureMode::ReplayGap => std::option::Option::Some(505),
            _ => return std::result::Result::Err(tonic::Status::unimplemented("replay info is outside this fixture mode")),
        };
        return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse { first_available }));
    }

    async fn ping(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::PingRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::PongResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }

    async fn get_latest_blockhash(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::GetLatestBlockhashRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::GetLatestBlockhashResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }

    async fn get_block_height(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::GetBlockHeightRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::GetBlockHeightResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }

    async fn get_slot(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::GetSlotRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::GetSlotResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }

    async fn is_blockhash_valid(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::IsBlockhashValidRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::IsBlockhashValidResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }

    async fn get_version(
        &self,
        _request: tonic::Request<yellowstone_grpc_proto::geyser::GetVersionRequest>,
    ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::GetVersionResponse>, tonic::Status> {
        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
    }
}

struct FixtureServer {
    endpoint_url: std::string::String,
    half_close_seen: std::sync::Arc<std::sync::atomic::AtomicBool>,
    subscribe_calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    shutdown: std::option::Option<tokio::sync::oneshot::Sender<()>>,
    task: tokio::task::JoinHandle<()>,
}

impl FixtureServer {
    async fn start(mode: FixtureMode) -> Self {
        let bind_address: std::net::SocketAddr = "127.0.0.1:0".parse().expect("fixture bind address must parse");
        let incoming = tonic::transport::server::TcpIncoming::bind(bind_address).expect("fixture gRPC listener must bind");
        let local_address = incoming.local_addr().expect("fixture gRPC listener must expose local address");
        let (shutdown, shutdown_receiver) = tokio::sync::oneshot::channel();
        let half_close_seen = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let subscribe_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let service_half_close_seen = half_close_seen.clone();
        let service_subscribe_calls = subscribe_calls.clone();
        let task = tokio::spawn(async move {
            let service = yellowstone_grpc_proto::geyser::geyser_server::GeyserServer::new(FixtureGeyser {
                mode,
                half_close_seen: service_half_close_seen,
                subscribe_calls: service_subscribe_calls,
            });
            let result = tonic::transport::Server::builder()
                .serve_with_incoming_shutdown(service, incoming, async move {
                    let _ = shutdown_receiver.await;
                })
                .await;
            assert!(result.is_ok());
        });
        return Self {
            endpoint_url: format!("http://{local_address}"),
            half_close_seen,
            subscribe_calls,
            shutdown: std::option::Option::Some(shutdown),
            task,
        };
    }

    async fn stop(mut self) {
        if let std::option::Option::Some(shutdown) = self.shutdown.take() {
            let _ = shutdown.send(());
        }
        let result = self.task.await;
        assert!(result.is_ok());
    }
}

fn verify_fixture_metadata(metadata: &tonic::metadata::MetadataMap) -> std::result::Result<(), tonic::Status> {
    let public = metadata.get("x-ksp-public").and_then(|value| return value.to_str().ok());
    let secret = metadata.get("x-ksp-token").and_then(|value| return value.to_str().ok());
    if public != std::option::Option::Some("fixture-public") || secret != std::option::Option::Some("GRPC-STREAM-SECRET-CANARY") {
        return std::result::Result::Err(tonic::Status::unauthenticated("fixture metadata mismatch"));
    }
    return std::result::Result::Ok(());
}

fn fixture_settings(
    url: &str,
    request_capacity: usize,
    update_capacity: usize,
    max_inbound_message_size_bytes: usize,
    max_outbound_message_size_bytes: usize,
) -> crate::YellowstoneGrpcEndpointSettings {
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    return fixture_settings_with_reconnect(
        url,
        request_capacity,
        update_capacity,
        max_inbound_message_size_bytes,
        max_outbound_message_size_bytes,
        crate::YellowstoneGrpcReconnectSettings::new(0, defaults.reconnect().initial_backoff(), defaults.reconnect().max_backoff()),
    );
}

fn fixture_settings_with_reconnect(
    url: &str,
    request_capacity: usize,
    update_capacity: usize,
    max_inbound_message_size_bytes: usize,
    max_outbound_message_size_bytes: usize,
    reconnect: crate::YellowstoneGrpcReconnectSettings,
) -> crate::YellowstoneGrpcEndpointSettings {
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let session = crate::YellowstoneGrpcSessionSettings::new(
        defaults.connect_timeout(),
        defaults.unary_timeout(),
        std::time::Duration::from_millis(250),
        reconnect,
        request_capacity,
        update_capacity,
        max_inbound_message_size_bytes,
        max_outbound_message_size_bytes,
    );
    let metadata = std::vec![
        crate::YellowstoneGrpcMetadataEntry::public("x-ksp-public", "fixture-public").expect("fixture public metadata must be valid"),
        crate::YellowstoneGrpcMetadataEntry::secret("x-ksp-token", "GRPC-STREAM-SECRET-CANARY").expect("fixture secret metadata must be valid"),
    ];
    return crate::YellowstoneGrpcEndpointSettings::new(
        "fixture-stream",
        true,
        crate::YellowstoneGrpcProviderName::new("fixture-provider"),
        crate::YellowstoneGrpcClusterName::new("devnet"),
        crate::YellowstoneGrpcEndpointUrl::parse(url).expect("fixture URL must parse"),
        session,
    )
    .with_metadata(metadata)
    .expect("fixture metadata settings must validate");
}

fn initial_request() -> crate::YellowstoneSubscribeRequest {
    let mut request = crate::YellowstoneSubscribeRequest::new();
    request
        .insert_slot_filter(
            crate::YellowstoneSubscribeFilterName::new("initial").expect("fixture filter name must validate"),
            crate::YellowstoneSubscribeSlotFilter::new(),
        )
        .expect("fixture slot filter must insert");
    return request;
}

fn mutated_request() -> crate::YellowstoneSubscribeRequest {
    let mut request = crate::YellowstoneSubscribeRequest::new();
    request
        .insert_account_filter(
            crate::YellowstoneSubscribeFilterName::new("mutated").expect("fixture filter name must validate"),
            crate::YellowstoneSubscribeAccountFilter::new(),
        )
        .expect("fixture account filter must insert");
    return request;
}

fn ping_update() -> yellowstone_grpc_proto::geyser::SubscribeUpdate {
    return yellowstone_grpc_proto::geyser::SubscribeUpdate {
        filters: std::vec::Vec::new(),
        update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Ping(
            yellowstone_grpc_proto::geyser::SubscribeUpdatePing {},
        )),
        created_at: std::option::Option::None,
    };
}

fn pong_update(id: i32) -> yellowstone_grpc_proto::geyser::SubscribeUpdate {
    return yellowstone_grpc_proto::geyser::SubscribeUpdate {
        filters: std::vec::Vec::new(),
        update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Pong(
            yellowstone_grpc_proto::geyser::SubscribeUpdatePong { id },
        )),
        created_at: std::option::Option::None,
    };
}

fn slot_update(slot: u64) -> yellowstone_grpc_proto::geyser::SubscribeUpdate {
    return yellowstone_grpc_proto::geyser::SubscribeUpdate {
        filters: std::vec!["initial".to_owned()],
        update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Slot(
            yellowstone_grpc_proto::geyser::SubscribeUpdateSlot {
                slot,
                parent: std::option::Option::Some(slot.saturating_sub(1)),
                status: yellowstone_grpc_proto::geyser::SlotStatus::SlotProcessed as i32,
                dead_error: std::option::Option::None,
            },
        )),
        created_at: std::option::Option::None,
    };
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_bidi_round_trip_mutates_request_replies_to_ping_and_observes_server_half_close() {
    let server = FixtureServer::start(FixtureMode::RoundTrip).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    assert_eq!(session.state(), crate::YellowstoneGrpcSubscribeState::Active);
    let ping = session.next_update().await.expect("Ping update must decode").expect("Ping update must be present");
    assert!(matches!(ping, crate::YellowstoneSubscribeUpdate::Ping(_)));
    let pong = session.next_update().await.expect("Pong update must decode").expect("Pong update must be present");
    match pong {
        crate::YellowstoneSubscribeUpdate::Pong(value) => assert_eq!(value.id(), 1),
        _ => panic!("fixture must return Pong"),
    }
    session.try_update(&mutated_request()).expect("request mutation must enter bounded queue");
    let slot = session.next_update().await.expect("slot update must decode").expect("slot update must be present");
    match slot {
        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 901),
        _ => panic!("fixture must return Slot"),
    }
    assert!(session.next_update().await.expect("server half-close must be normal").is_none());
    assert_eq!(session.state(), crate::YellowstoneGrpcSubscribeState::Closed);
    session.close().await.expect("already half-closed fixture must close cleanly");
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_explicit_close_half_closes_request_stream_before_deadline() {
    let server = FixtureServer::start(FixtureMode::ClientHalfClose).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    session.close().await.expect("graceful half-close must complete");
    assert!(server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst));
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_hostile_server_shutdown_is_bounded_by_close_timeout() {
    let server = FixtureServer::start(FixtureMode::HostileClose).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let error = session.close().await.expect_err("hostile server must hit the bounded graceful close deadline");
    assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT);
    assert!(server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst));
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_session_drop_best_effort_half_closes_request_stream() {
    let server = FixtureServer::start(FixtureMode::ClientHalfClose).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    drop(session);
    let observed = tokio::time::timeout(std::time::Duration::from_secs(1), async {
        loop {
            if server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst) {
                return true;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("fixture must observe dropped-session half-close before timeout");
    assert!(observed);
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_slow_receiver_overflow_is_terminal_and_observable() {
    let server = FixtureServer::start(FixtureMode::Flood).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 2, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    assert!(session.next_update().await.expect("first queued update must remain readable").is_some());
    assert!(session.next_update().await.expect("second queued update must remain readable").is_some());
    let error = session.next_update().await.expect_err("overflow must become terminal after bounded queued updates drain");
    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW);
    let rendered = format!("{error:?} {session:?}");
    assert!(!rendered.contains("GRPC-STREAM-SECRET-CANARY"));
    assert!(!rendered.contains(server.endpoint_url.as_str()));
    let _ = session.close().await;
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_stream_remote_status_is_safe_and_terminal() {
    let server = FixtureServer::start(FixtureMode::RemoteStatus).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let error = session.next_update().await.expect_err("remote Status must fail stream");
    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_STATUS);
    let rendered = format!("{error:?} {session:?}");
    assert!(!rendered.contains("GRPC-STREAM-REMOTE-SECRET-CANARY"));
    assert!(!rendered.contains("GRPC-STREAM-SECRET-CANARY"));
    assert!(!rendered.contains(server.endpoint_url.as_str()));
    let _ = session.close().await;
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_malformed_update_is_rejected_without_raw_wire_escape() {
    let server = FixtureServer::start(FixtureMode::Malformed).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let error = session.next_update().await.expect_err("missing update oneof must be rejected");
    assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RESPONSE);
    let _ = session.close().await;
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_inbound_message_limit_is_enforced_by_tonic_stream_decoder() {
    let server = FixtureServer::start(FixtureMode::Oversized).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, 1, defaults.max_outbound_message_size_bytes());
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let error = session.next_update().await.expect_err("message above configured inbound bound must fail decoding");
    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_STATUS);
    let rendered = format!("{error:?} {session:?}");
    assert!(!rendered.contains(server.endpoint_url.as_str()));
    let _ = session.close().await;
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_outbound_mutation_size_is_rejected_before_queue_dispatch() {
    let server = FixtureServer::start(FixtureMode::Idle).await;
    let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, 64 * 1024 * 1024, 1);
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let session = channel
        .open_standard_subscribe(crate::YellowstoneSubscribeRequest::new())
        .await
        .expect("empty initial request must fit one-byte fixture bound");
    let error = session.try_update(&mutated_request()).expect_err("encoded mutation above configured bound must be rejected before dispatch");
    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW);
    session.close().await.expect("idle fixture must observe client half-close");
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_reconnect_replays_from_last_observed_slot_and_counts_duplicate_identity() {
    let server = FixtureServer::start(FixtureMode::ReconnectReplay).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings_with_reconnect(
        server.endpoint_url.as_str(),
        8,
        8,
        defaults.max_inbound_message_size_bytes(),
        defaults.max_outbound_message_size_bytes(),
        crate::YellowstoneGrpcReconnectSettings::new(3, std::time::Duration::from_millis(5), std::time::Duration::from_millis(20)),
    );
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let first = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
    match first {
        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 500),
        _ => panic!("fixture must return first Slot"),
    }
    let duplicate = session.next_update().await.expect("replayed duplicate must decode").expect("replayed duplicate must be present");
    match duplicate {
        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 500),
        _ => panic!("fixture must return replayed Slot"),
    }
    let resumed = session.next_update().await.expect("resumed slot must decode").expect("resumed slot must be present");
    match resumed {
        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 501),
        _ => panic!("fixture must return resumed Slot"),
    }
    let snapshot = session.snapshot();
    assert_eq!(snapshot.state(), crate::YellowstoneGrpcSubscribeState::Active);
    assert_eq!(snapshot.reconnect_count(), 1);
    assert_eq!(snapshot.replay_attempt_count(), 1);
    assert_eq!(snapshot.continuity_gap_count(), 0);
    assert_eq!(snapshot.duplicate_update_count(), 1);
    assert_eq!(snapshot.last_requested_from_slot(), std::option::Option::Some(500));
    assert_eq!(snapshot.last_observed_slot(), std::option::Option::Some(501));
    session.close().await.expect("reconnected stream must close cleanly");
    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 2);
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_replay_info_proves_and_clamps_retention_gap_without_lossless_claim() {
    let server = FixtureServer::start(FixtureMode::ReplayGap).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings_with_reconnect(
        server.endpoint_url.as_str(),
        8,
        8,
        defaults.max_inbound_message_size_bytes(),
        defaults.max_outbound_message_size_bytes(),
        crate::YellowstoneGrpcReconnectSettings::new(3, std::time::Duration::from_millis(5), std::time::Duration::from_millis(20)),
    );
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
    let resumed = session.next_update().await.expect("clamped replay must decode").expect("clamped replay must be present");
    match resumed {
        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 505),
        _ => panic!("fixture must return clamped Slot"),
    }
    let snapshot = session.snapshot();
    assert_eq!(snapshot.reconnect_count(), 1);
    assert_eq!(snapshot.replay_attempt_count(), 1);
    assert_eq!(snapshot.continuity_gap_count(), 1);
    assert_eq!(snapshot.duplicate_update_count(), 0);
    assert_eq!(snapshot.last_requested_from_slot(), std::option::Option::Some(505));
    assert_eq!(snapshot.last_observed_slot(), std::option::Option::Some(505));
    session.close().await.expect("gap fixture must close cleanly");
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_reconnect_budget_exhaustion_is_terminal_and_safe() {
    let server = FixtureServer::start(FixtureMode::ReconnectExhausted).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings_with_reconnect(
        server.endpoint_url.as_str(),
        8,
        8,
        defaults.max_inbound_message_size_bytes(),
        defaults.max_outbound_message_size_bytes(),
        crate::YellowstoneGrpcReconnectSettings::new(2, std::time::Duration::from_millis(5), std::time::Duration::from_millis(10)),
    );
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
    let error = session.next_update().await.expect_err("failed reconnect attempts must exhaust the bounded budget");
    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_CHANNEL_FAILED);
    let snapshot = session.snapshot();
    assert_eq!(snapshot.state(), crate::YellowstoneGrpcSubscribeState::Failed);
    assert_eq!(snapshot.reconnect_count(), 0);
    assert_eq!(snapshot.replay_attempt_count(), 2);
    assert_eq!(snapshot.terminal_error_code(), std::option::Option::Some(crate::ERROR_CODE_GRPC_CHANNEL_FAILED));
    let rendered = format!("{error:?} {session:?}");
    assert!(!rendered.contains("GRPC-RECONNECT-SECRET-CANARY"));
    assert!(!rendered.contains(server.endpoint_url.as_str()));
    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 3);
    let _ = session.close().await;
    server.stop().await;
}

#[tokio::test(flavor = "current_thread")]
async fn yellowstone_shutdown_interrupts_reconnect_backoff_and_mutation_is_rejected_during_reconnect() {
    let server = FixtureServer::start(FixtureMode::ReconnectReplay).await;
    let defaults = crate::YellowstoneGrpcSessionSettings::default();
    let settings = fixture_settings_with_reconnect(
        server.endpoint_url.as_str(),
        8,
        8,
        defaults.max_inbound_message_size_bytes(),
        defaults.max_outbound_message_size_bytes(),
        crate::YellowstoneGrpcReconnectSettings::new(3, std::time::Duration::from_millis(500), std::time::Duration::from_millis(500)),
    );
    let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect");
    let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open");
    let _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
    tokio::time::timeout(std::time::Duration::from_secs(1), async {
        loop {
            if session.state() == crate::YellowstoneGrpcSubscribeState::Reconnecting {
                return;
            }
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("fixture must enter reconnecting state");
    let mutation_error = session.try_update(&mutated_request()).expect_err("request mutation must be rejected during reconnect");
    assert_eq!(mutation_error.code(), crate::ERROR_CODE_GRPC_SESSION_CLOSED);
    session.close().await.expect("shutdown must interrupt reconnect backoff without opening another stream");
    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 1);
    server.stop().await;
}